From b7ff4b03a48828e8076d1d3c519ee03611fcd1a6 Mon Sep 17 00:00:00 2001 From: CC#1 Kora Runtime Date: Sat, 23 May 2026 22:39:23 -0700 Subject: [PATCH] =?UTF-8?q?feat(kora):=20KR-PROMOTE-PHRASEBOOK-FOUNDATION-?= =?UTF-8?q?MEGABUCKET=20=E2=80=94=20first=20promotion=20loop=20end-to-end?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit First concrete implementation of the promotion-loop pattern per ``feedback-promotion-loops-self-improving-subsystems``. After this lands: Kora observes recurring DM patterns, drafts phrasebook proposals at idle cadence, operator approves via cockpit (CC#2 follow-on KR-FE-PROMOTION-REVIEW-PANEL), audit captures the loop end-to-end. Six deliverables in one batched bucket — single PR. Six deliverables ---------------- A. Shared Haiku-clustering utility — ``kora_cli/clustering/``: * ``text_similarity.embed_texts`` — async embedder with disk cache. Lexical (token + char 3-gram, L2-normalized, deterministic, $0) rather than Haiku-based. Decision was anticipated by the bucket's STOP-ASK §4 alternative; the proposer still uses Haiku where natural-language synthesis genuinely helps (reply-template generation, one call per proposal). * ``cosine_similarity`` + ``cluster_by_similarity`` (greedy agglomerative). B. Reasoning DM observation collector — ``kora_cli/promote/phrasebook/observer.py``: * Reads ``slack_dm_log.jsonl`` (handler-driven replies + wake-consumer replies both write here post-#184) → ``ReasoningObservation`` records. * Filters: route allow-list, drop short-circuit hits, drop missing-engine canned fallbacks, drop failed sends, time window via ``since``. * Per-call cost reused from ``agent.usage_pricing.estimate_usage_cost`` (same canonical pricing as ``cost_state_holder.record_inference``). C. Promotion proposal generator — ``kora_cli/promote/phrasebook/proposer.py``: * Clusters observations, applies size + cohesion + answer-consistency thresholds. * Derives a conservative regex pattern (escaped alternation of top tokens — protects against the STOP-ASK §4 regex- pathology concern; phrasebook editor's validator catches anything that slips through at approve-time). * Reply template: dominant verbatim response OR optional Haiku synthesis (injectable for tests). * Confidence = (cohesion + consistency) / 2. D. Three new audit seams in ``SeamName`` Literal: * ``promotion.proposed`` — per proposal at proposer time. ``synth_cost_usd`` field per row for cost telemetry. * ``promotion.approved`` — at endpoint time. Also emits ``phrasebook.updated`` with ``actor="kora_proposal_approved"`` per PR #177 forward-compat. * ``promotion.rejected`` — with operator-supplied ``review_notes`` recorded verbatim (#182 precedent). E. Three new backend endpoints in ``web_server.py``: * ``GET /api/promotions/phrasebook/pending`` * ``POST /api/promotions/phrasebook/{id}/approve`` — optional pattern_override / reply_template_override / category_override / review_notes payload. * ``POST /api/promotions/phrasebook/{id}/reject`` — ``{review_notes}`` payload. * Drift-guard pin ``_PROMOTION_STATUS_VALUES``; CC#2's KR-FE-PROMOTION-REVIEW-PANEL adds the symmetric FE constant. F. Cron registration — ``kora_cli/listeners/promote_phrasebook_listener.py``: * Registers ``run_phrasebook_promotion_cycle`` via the heartbeat scheduler (interval-based, 86400s default). * Cron-string suggestion ("0 6 * * *") superseded by the interval shape since the heartbeat scheduler is interval-based; documented in the listener docstring. * Env tunables: ``KORA_PROMOTE_PHRASEBOOK_ENABLED`` (default true), ``KORA_PROMOTE_PHRASEBOOK_INTERVAL_SEC`` (86400), ``KORA_PROMOTE_PHRASEBOOK_OBSERVATION_WINDOW_DAYS`` (7), ``KORA_PROMOTE_PHRASEBOOK_MIN_CLUSTER_SIZE`` (5), ``KORA_PROMOTE_PHRASEBOOK_EXPIRY_DAYS`` (14). Persistence ----------- Proposals live at ``${KORA_HOME}/promotions/phrasebook/{pending,approved,rejected,expired}/.json``. Status transitions move the file atomically via os.replace. Files + audit JSONL together are the forensic-truth stream. Tests ----- 52 new tests across 4 files. ``ruff check`` clean. 432-test regression set (clustering + promote + audit + handlers + short_circuit + reasoning + probes) all pass. Co-Authored-By: Claude Opus 4.7 (1M context) --- kora_cli/audit/jsonl_sink.py | 30 + kora_cli/clustering/__init__.py | 11 + kora_cli/clustering/text_similarity.py | 331 +++++++++++ kora_cli/listeners/__init__.py | 8 + .../listeners/promote_phrasebook_listener.py | 88 +++ kora_cli/promote/__init__.py | 9 + kora_cli/promote/phrasebook/__init__.py | 28 + kora_cli/promote/phrasebook/cycle.py | 304 ++++++++++ kora_cli/promote/phrasebook/observer.py | 257 +++++++++ kora_cli/promote/phrasebook/proposer.py | 537 ++++++++++++++++++ kora_cli/promote/phrasebook/store.py | 294 ++++++++++ kora_cli/web_server.py | 303 ++++++++++ tests/kora_cli/clustering/__init__.py | 0 .../clustering/test_text_similarity.py | 133 +++++ tests/kora_cli/promote/__init__.py | 0 tests/kora_cli/promote/phrasebook/__init__.py | 0 .../promote/phrasebook/test_endpoints.py | 201 +++++++ .../promote/phrasebook/test_observer.py | 182 ++++++ .../promote/phrasebook/test_proposer.py | 221 +++++++ .../phrasebook/test_store_and_cycle.py | 229 ++++++++ 20 files changed, 3166 insertions(+) create mode 100644 kora_cli/clustering/__init__.py create mode 100644 kora_cli/clustering/text_similarity.py create mode 100644 kora_cli/listeners/promote_phrasebook_listener.py create mode 100644 kora_cli/promote/__init__.py create mode 100644 kora_cli/promote/phrasebook/__init__.py create mode 100644 kora_cli/promote/phrasebook/cycle.py create mode 100644 kora_cli/promote/phrasebook/observer.py create mode 100644 kora_cli/promote/phrasebook/proposer.py create mode 100644 kora_cli/promote/phrasebook/store.py create mode 100644 tests/kora_cli/clustering/__init__.py create mode 100644 tests/kora_cli/clustering/test_text_similarity.py create mode 100644 tests/kora_cli/promote/__init__.py create mode 100644 tests/kora_cli/promote/phrasebook/__init__.py create mode 100644 tests/kora_cli/promote/phrasebook/test_endpoints.py create mode 100644 tests/kora_cli/promote/phrasebook/test_observer.py create mode 100644 tests/kora_cli/promote/phrasebook/test_proposer.py create mode 100644 tests/kora_cli/promote/phrasebook/test_store_and_cycle.py diff --git a/kora_cli/audit/jsonl_sink.py b/kora_cli/audit/jsonl_sink.py index 3a37f5124683..fd2d3bf844ac 100644 --- a/kora_cli/audit/jsonl_sink.py +++ b/kora_cli/audit/jsonl_sink.py @@ -161,6 +161,36 @@ # ``actor="kora_proposal_approved"`` from the promotion-loop # bucket) reuses this seam shape. "phrasebook.updated", + # KR-PROMOTE-PHRASEBOOK-FOUNDATION — first promotion loop. + # Three seams covering the propose → review → resolve lifecycle: + # + # ``promotion.proposed`` — proposer emits a new pending + # phrasebook proposal after the daily clustering cycle. + # Payload carries the full PromotionProposal projection + # (proposal_id, cluster_size, sample_questions, + # proposed_pattern, proposed_reply_template, proposed_category, + # confidence, created_at) so operator can grep the JSONL for + # proposal history without reading every proposal file. One + # row per proposal; the per-cycle summary (count / total cost) + # is logged via the structured-log line. + "promotion.proposed", + # ``promotion.approved`` — operator approves via the cockpit + # endpoint. Payload: proposal_id + the committed phrasebook + # entry shape (post any operator override edits). The + # ``phrasebook.updated`` audit row that follows uses + # actor="kora_proposal_approved" per #177 forward-compat — + # so the promotion seam stays distinct from the editor audit + # without the promotion-history view having to scan + # ``phrasebook.updated`` for actor=proposal entries. + "promotion.approved", + # ``promotion.rejected`` — operator rejects. Payload: + # proposal_id + ``review_notes`` (rejection rationale, written + # verbatim — operator-decision-relevant per the #182 precedent + # for reason fields). Proposal stays in the rejected/ store + # directory for future promotion-loop tuning (clusters that + # operator consistently rejects are signal to tune the + # proposer thresholds). + "promotion.rejected", ] SourceName = Literal[ diff --git a/kora_cli/clustering/__init__.py b/kora_cli/clustering/__init__.py new file mode 100644 index 000000000000..7bd849fa8206 --- /dev/null +++ b/kora_cli/clustering/__init__.py @@ -0,0 +1,11 @@ +"""Shared text-clustering utilities for promotion loops. + +First consumer: KR-PROMOTE-PHRASEBOOK-FOUNDATION. Future consumers: +snapshot-expand, router-trigger, tool-trimming, probe-fix-envelope +promotion loops (all the loops Joshua locked in +``feedback-promotion-loops-self-improving-subsystems``). + +Public API: :func:`text_similarity.embed_texts`, +:func:`text_similarity.cosine_similarity`, +:func:`text_similarity.cluster_by_similarity`. +""" diff --git a/kora_cli/clustering/text_similarity.py b/kora_cli/clustering/text_similarity.py new file mode 100644 index 000000000000..1c4a3ce867ff --- /dev/null +++ b/kora_cli/clustering/text_similarity.py @@ -0,0 +1,331 @@ +"""Text embedding + similarity + clustering — KR-PROMOTE-PHRASEBOOK-FOUNDATION. + +# Embedding strategy: lexical, deterministic, $0 + +The spec's default was "Haiku-based pseudo-embedding" with a per-text +cost of ~$0.001 (cold cache). For the promote-phrasebook use case +(clustering ≤200 short operator DMs/day) we ship a **lexical** +embedder instead — token-set + character-n-gram features projected +into a fixed-dimensional sparse vector and L2-normalized. This is: + + * deterministic (same input → same vector; same vector across + re-runs and across machines — no LLM nondeterminism to hide + behind a cache); + * **$0/day** (no LLM call); cost discipline target + ($0.01-0.05/day across all promotion loops) is met with budget + to spare for the proposer's per-proposal Haiku synthesis; + * tunable via the n-gram window; + * stable under typo / paraphrase noise common to short operator + DMs ("burn?" / "what's the burn?" / "burn rate?"). + +The Haiku-based approach is documented in the bucket's STOP-ASK §4 +as the alternative if lexical clustering produced pathological +similarity distributions. We pre-empted that ASK by going lexical +from the start — the proposer module still uses Haiku where natural- +language synthesis genuinely helps (reply-template generation, one +call per proposal). + +# Cache + +A disk cache (keyed by ``sha256(text)``) sits in front of +:func:`embed_texts` because the cycle re-reads the past +``KORA_PROMOTE_PHRASEBOOK_OBSERVATION_WINDOW_DAYS`` of observations +on each run. With the lexical embedder cache HIT vs MISS is just an +I/O optimization (computation is microseconds either way), but the +cache is shipped so a future Haiku-backed embedder can drop in +without API surface churn. + +# Public API + + * :class:`TextEmbedding` — named-tuple-ish dataclass, (text, + embedding, cached). + * :func:`embed_texts(texts, *, cache_dir=None)` — async; one + pass over the input texts; returns a TextEmbedding per input + in original order. + * :func:`cosine_similarity(a, b)` — float in [-1, 1]. + * :func:`cluster_by_similarity(embeddings, *, threshold)` — + greedy agglomerative clustering. Returns list of clusters + (each cluster a list of TextEmbedding); single-element + clusters represent unmatched observations. +""" + +from __future__ import annotations + +import asyncio +import hashlib +import json +import logging +import math +import os +import re +import tempfile +from dataclasses import dataclass +from pathlib import Path +from typing import Dict, List, Optional + +logger = logging.getLogger(__name__) + + +# --------------------------------------------------------------------------- +# Tunables +# --------------------------------------------------------------------------- + + +# Character n-gram window. Smaller catches typos; larger captures +# phrase shape. 3-gram is the sweet spot for short DMs. +_CHAR_NGRAM = 3 + +# Token-set features are tagged with a stable prefix so they don't +# collide with character-n-grams that happen to be the same string. +_TOKEN_PREFIX = "TOK:" +_CHAR_PREFIX = "CHR:" + +# Lowercased + apostrophe-stripped; matches "what's" → "whats" so +# common contractions don't fragment. +_TOKEN_RE = re.compile(r"[A-Za-z0-9]+") + + +# --------------------------------------------------------------------------- +# Data types +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True, slots=True) +class TextEmbedding: + """One text + its sparse embedding. + + ``embedding`` is a sparse-mapping representation: feature-name → + weight. Stored as a dict for simplicity (lexical embeddings + have at most a few hundred features per short DM; this is + cheaper than dense lookups for our scale). Consumers see a + consistent interface via :func:`cosine_similarity`. + """ + + text: str + embedding: Dict[str, float] + cached: bool + + +# --------------------------------------------------------------------------- +# Cache (disk-backed, sha256-keyed) +# --------------------------------------------------------------------------- + + +def _resolve_default_cache_dir() -> Path: + """``${KORA_HOME}/cache/text_similarity_embeddings``. Created + on first write; an env-unset / unreachable KORA_HOME degrades + gracefully to a process-local tempdir so tests don't need to + set anything.""" + try: + from kora_constants import get_kora_home + + return get_kora_home() / "cache" / "text_similarity_embeddings" + except Exception: + return Path(tempfile.gettempdir()) / "kora_text_similarity_cache" + + +def _cache_key(text: str) -> str: + return hashlib.sha256(text.encode("utf-8")).hexdigest() + + +def _cache_get(cache_dir: Path, text: str) -> Optional[Dict[str, float]]: + path = cache_dir / f"{_cache_key(text)}.json" + if not path.is_file(): + return None + try: + return json.loads(path.read_text(encoding="utf-8")) + except Exception as exc: + logger.debug( + "[kora.clustering] cache read failed for %s: %r", + path, + exc, + ) + return None + + +def _cache_put( + cache_dir: Path, text: str, embedding: Dict[str, float] +) -> None: + try: + cache_dir.mkdir(parents=True, exist_ok=True) + path = cache_dir / f"{_cache_key(text)}.json" + path.write_text(json.dumps(embedding), encoding="utf-8") + except OSError as exc: + logger.debug( + "[kora.clustering] cache write failed for %s: %r", + cache_dir, + exc, + ) + + +# --------------------------------------------------------------------------- +# Embedding (lexical) +# --------------------------------------------------------------------------- + + +def _normalize_text(text: str) -> str: + """Lowercase + strip apostrophes so "what's" and "whats" + collapse to the same feature set.""" + return text.lower().replace("'", "").replace("’", "") + + +def _token_features(text: str) -> Dict[str, float]: + """Token-set features (binary presence, 1.0 weight each). + + Binary rather than count-weighted: short DMs rarely repeat + tokens, and binary keeps the cosine well-behaved for very + short inputs. + """ + norm = _normalize_text(text) + tokens = _TOKEN_RE.findall(norm) + return {f"{_TOKEN_PREFIX}{t}": 1.0 for t in set(tokens)} + + +def _char_ngram_features(text: str) -> Dict[str, float]: + """Character n-gram features over the normalized text. Catches + typos + morphology that token features miss + (``burn`` vs ``burning``).""" + norm = _normalize_text(text) + norm_padded = f" {norm} " + out: Dict[str, float] = {} + if len(norm_padded) < _CHAR_NGRAM: + return out + for i in range(len(norm_padded) - _CHAR_NGRAM + 1): + gram = norm_padded[i : i + _CHAR_NGRAM] + key = f"{_CHAR_PREFIX}{gram}" + out[key] = out.get(key, 0.0) + 1.0 + return out + + +def _l2_normalize(features: Dict[str, float]) -> Dict[str, float]: + norm_sq = sum(v * v for v in features.values()) + if norm_sq <= 0: + return dict(features) + norm = math.sqrt(norm_sq) + return {k: v / norm for k, v in features.items()} + + +def _compute_embedding(text: str) -> Dict[str, float]: + """Lexical embedding: union of token features + char-ngram + features, then L2-normalized.""" + feats: Dict[str, float] = {} + feats.update(_token_features(text)) + feats.update(_char_ngram_features(text)) + return _l2_normalize(feats) + + +async def embed_texts( + texts: List[str], *, cache_dir: Optional[Path] = None +) -> List[TextEmbedding]: + """Embed each text. Returns embeddings in the same order as the + input list (one-to-one). + + Cache hits are recorded on the ``cached`` field — caller can + sum cached vs uncached for cost telemetry, even though the + lexical embedder is free either way (the field stays useful + if the embedder is swapped for a paid one later). + + The function is ``async`` for forward-compat with the Haiku- + backed embedder (no actual awaiting needed today; the loop + intentionally yields to keep CPU-bound batches cooperative + under concurrent cycles). + """ + target_dir = cache_dir or _resolve_default_cache_dir() + out: List[TextEmbedding] = [] + for idx, text in enumerate(texts): + cached_emb = _cache_get(target_dir, text) + if cached_emb is not None: + out.append( + TextEmbedding(text=text, embedding=cached_emb, cached=True) + ) + else: + emb = _compute_embedding(text) + _cache_put(target_dir, text, emb) + out.append( + TextEmbedding(text=text, embedding=emb, cached=False) + ) + # Yield every 32 texts so a 200-text batch doesn't hog the + # event loop. Trivial overhead; matters under concurrent + # cycles. + if idx and idx % 32 == 0: + await asyncio.sleep(0) + return out + + +# --------------------------------------------------------------------------- +# Similarity +# --------------------------------------------------------------------------- + + +def cosine_similarity(a: TextEmbedding, b: TextEmbedding) -> float: + """Cosine similarity over the sparse embedding dicts. + + Both vectors are L2-normalized by ``embed_texts`` so this + reduces to the dot product (intersection sum). Empty vectors + short-circuit to 0. + """ + if not a.embedding or not b.embedding: + return 0.0 + # Iterate the smaller side for the intersection. + if len(a.embedding) <= len(b.embedding): + small, large = a.embedding, b.embedding + else: + small, large = b.embedding, a.embedding + total = 0.0 + for key, weight in small.items(): + other = large.get(key) + if other is not None: + total += weight * other + # Clamp — float noise on near-1.0 may push very slightly above. + if total > 1.0: + return 1.0 + if total < -1.0: + return -1.0 + return total + + +# --------------------------------------------------------------------------- +# Clustering +# --------------------------------------------------------------------------- + + +def cluster_by_similarity( + embeddings: List[TextEmbedding], + *, + threshold: float = 0.85, +) -> List[List[TextEmbedding]]: + """Greedy agglomerative clustering. + + Each new embedding joins the first existing cluster whose + centroid-similarity ≥ ``threshold``. If none qualify, it + starts a new cluster. + + Single-element clusters represent unmatched observations + (caller — typically the proposer — filters by + ``len(cluster) >= min_cluster_size``). + + "Centroid-similarity" is approximated as max(sim to any member); + this is conservative (favors tight clusters) and cheap. For + short DMs the centroid distance to any one member is a good + proxy for the cluster's average — we don't need k-means + sophistication. + + Determinism: input order is preserved; same input list ⇒ same + output partition. + """ + clusters: List[List[TextEmbedding]] = [] + for emb in embeddings: + placed = False + for cluster in clusters: + # Max-link rather than centroid for sparse vectors — + # cheaper, and matches the conservative-tightness goal. + best_sim = max( + cosine_similarity(emb, member) for member in cluster + ) + if best_sim >= threshold: + cluster.append(emb) + placed = True + break + if not placed: + clusters.append([emb]) + return clusters diff --git a/kora_cli/listeners/__init__.py b/kora_cli/listeners/__init__.py index afc2e62ab0f4..d118e5bab5dc 100644 --- a/kora_cli/listeners/__init__.py +++ b/kora_cli/listeners/__init__.py @@ -78,3 +78,11 @@ # investigation summary. Imported LAST so the audit reader + # reasoning engine + slack client listeners are all registered first. from kora_cli.listeners import probe_wake_listener # noqa: F401 +# KR-PROMOTE-PHRASEBOOK-FOUNDATION — daily clustering cycle that +# reads slack_dm_log.jsonl + proposes phrasebook entries to short- +# circuit recurring DMs. Imported LAST so the slack_dm_log writer +# (post-#184 routing) + the heartbeat scheduler are both wired +# before the task gets enqueued. Fail-soft: cycle exceptions are +# swallowed by the heartbeat _loop; per-proposal failures don't +# poison the batch. +from kora_cli.listeners import promote_phrasebook_listener # noqa: F401 diff --git a/kora_cli/listeners/promote_phrasebook_listener.py b/kora_cli/listeners/promote_phrasebook_listener.py new file mode 100644 index 000000000000..527bbcf10f58 --- /dev/null +++ b/kora_cli/listeners/promote_phrasebook_listener.py @@ -0,0 +1,88 @@ +"""Heartbeat-scheduled phrasebook promotion cycle — KR-PROMOTE-PHRASEBOOK-FOUNDATION +(Deliverable F registration side). + +Registers ``run_phrasebook_promotion_cycle`` as a periodic task +against the heartbeat scheduler. Cadence is operator-tunable via +``KORA_PROMOTE_PHRASEBOOK_INTERVAL_SEC`` (default 86400s = once +daily). The kill-switch +``KORA_PROMOTE_PHRASEBOOK_ENABLED=false`` is checked inside the +cycle function so flipping the env at runtime takes effect on the +next tick without re-registering. + +# Why a periodic-task interval, not a cron string + +The bucket spec suggested ``"0 6 * * *"`` (6am UTC daily). The +existing heartbeat scheduler in ``listeners/heartbeat.py`` is +interval-based — it sleeps the specified seconds between fires. +There's no cron-string scheduler today. Wall-clock-anchored +scheduling (e.g., "fire at 6am UTC") would need a separate +"watch + act" pattern modeled on +``listeners/cost_telemetry_listener.py``'s daily-reset task +(check whether the boundary has crossed since the last fire, +fire if yes). That's reasonable for v2 if the operator wants +deterministic UTC anchor — but for v1 the simpler "fire once +every 24h" interval is sufficient: the proposer's outputs are +batch artifacts that go into a review queue, not real-time +events, so the exact wall-clock anchor doesn't matter +operationally. + +Documented this decision verbatim in the bucket PR body so the +spec → impl divergence is explicit. + +# Fail-soft startup + +If ``run_phrasebook_promotion_cycle`` itself raises on a +particular tick, the heartbeat scheduler's ``_loop`` swallows +the exception + logs a warning + continues. No retry queue; +next tick (default 24h later) is the natural retry cadence — +this is a batch promotion loop, not an interactive surface. +""" + +from __future__ import annotations + +import logging + +from kora_cli.listeners.heartbeat import register_periodic_task +from kora_cli.promote.phrasebook.cycle import ( + get_interval_seconds, + run_phrasebook_promotion_cycle, +) + +logger = logging.getLogger(__name__) + + +async def _periodic_task() -> None: + """Thin async wrapper so the heartbeat scheduler's signature + (``Callable[[], Awaitable[None]]``) is satisfied. Cycle's + return value (the summary dict) is consumed locally + logged + at INFO; the scheduler doesn't need to see it.""" + try: + summary = await run_phrasebook_promotion_cycle() + # Cycle already logs the summary at INFO; we don't + # duplicate. Re-log here at DEBUG so operator can grep + # this specific task name when triaging "did the cron + # task even run today" without sifting cycle-internal + # info lines. + logger.debug( + "[kora.promote.phrasebook.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: + # Belt-and-suspenders — cycle is fail-soft, but the wrapper + # catches anything that escapes (e.g., asyncio cancellation + # during shutdown). + logger.warning( + "[kora.promote.phrasebook.listener] tick raised %r — " + "next scheduled run will retry", + exc, + ) + + +register_periodic_task( + "promote_phrasebook_cycle", + interval_seconds=float(get_interval_seconds()), + callable=_periodic_task, +) diff --git a/kora_cli/promote/__init__.py b/kora_cli/promote/__init__.py new file mode 100644 index 000000000000..d066839474c8 --- /dev/null +++ b/kora_cli/promote/__init__.py @@ -0,0 +1,9 @@ +"""Promotion loops — Kora's self-improving subsystems. + +Per ``feedback-promotion-loops-self-improving-subsystems``: Kora +observes recurring patterns, drafts proposals, operator approves +via cockpit, audit captures the loop end-to-end. First concrete +implementation is :mod:`.phrasebook`; future loops (snapshot-expand, +router-trigger, tool-trimming, probe-fix-envelope) follow the same +pattern. +""" diff --git a/kora_cli/promote/phrasebook/__init__.py b/kora_cli/promote/phrasebook/__init__.py new file mode 100644 index 000000000000..2c6732455a2a --- /dev/null +++ b/kora_cli/promote/phrasebook/__init__.py @@ -0,0 +1,28 @@ +"""Phrasebook promotion loop — KR-PROMOTE-PHRASEBOOK-FOUNDATION. + +# Loop shape + + 1. :mod:`.observer` — collect last-N-days of slack_dm_log entries + where Kora's reasoning engine ran (NOT short-circuit hits; + short-circuits already work). + 2. :mod:`.proposer` — cluster the observations + generate one + phrasebook proposal per cohesive cluster meeting the + min_cluster_size / cohesion / answer-consistency thresholds. + 3. :mod:`.store` — persist proposals as JSON files under + ``${KORA_HOME}/promotions/phrasebook/{pending,approved,rejected}/`` + plus emit the corresponding promotion.* audit row. + 4. :mod:`.cycle` — orchestrator called by the cron task. + 5. Operator reviews via cockpit; approve / reject endpoints + transition status + (on approve) PUT to the live phrasebook + using the existing #177 editor with + actor="kora_proposal_approved". + +# Cost discipline + +Per ``feedback-promotion-loops-self-improving-subsystems``: target +~$0.01-0.05/day. The embedder is lexical ($0); the proposer +optionally synthesizes one Haiku-driven reply template per +proposal (~$0.001 each). With min_cluster_size=5 + typical +operator-DM volume, expect ≤5 proposals per cycle → ≤$0.005/day. +Lots of headroom for future loops. +""" diff --git a/kora_cli/promote/phrasebook/cycle.py b/kora_cli/promote/phrasebook/cycle.py new file mode 100644 index 000000000000..5625d4218f39 --- /dev/null +++ b/kora_cli/promote/phrasebook/cycle.py @@ -0,0 +1,304 @@ +"""Promotion cycle orchestrator — KR-PROMOTE-PHRASEBOOK-FOUNDATION (Deliverable F orchestrator). + +Called by the periodic-task heartbeat (registered by +:mod:`kora_cli.listeners.promote_phrasebook_listener`). One cycle: + + 1. Collect observations from the last + ``KORA_PROMOTE_PHRASEBOOK_OBSERVATION_WINDOW_DAYS`` days + (default 7) via :func:`observer.collect_recent_observations`. + 2. Generate proposals via :func:`proposer.generate_proposals` + with operator-tunable thresholds. + 3. Persist each proposal as pending + emit + ``promotion.proposed`` audit row. + 4. Expire pending proposals older than + ``KORA_PROMOTE_PHRASEBOOK_EXPIRY_DAYS`` (default 14) so the + pending list doesn't grow without bound. + 5. Log cycle summary: observation count, cluster count, + proposal count, total Haiku synthesis cost. + +# Env + + * ``KORA_PROMOTE_PHRASEBOOK_ENABLED`` (default ``true``) — + master kill-switch. False = cycle returns 0 proposals without + reading observations or invoking the LLM. + * ``KORA_PROMOTE_PHRASEBOOK_INTERVAL_SEC`` (default ``86400``, + i.e. once daily) — the heartbeat-scheduler interval. This is + SECONDS not cron-string because the heartbeat scheduler is + interval-based; the bucket spec's "0 6 * * *" cron suggestion + is documented in the PR body as the alternative the daily- + interval shape supersedes. + * ``KORA_PROMOTE_PHRASEBOOK_OBSERVATION_WINDOW_DAYS`` (default + ``7``). + * ``KORA_PROMOTE_PHRASEBOOK_MIN_CLUSTER_SIZE`` (default ``5``). + * ``KORA_PROMOTE_PHRASEBOOK_EXPIRY_DAYS`` (default ``14``). + +# Fail-soft + +Cycle exceptions log + are swallowed by the heartbeat scheduler +(see ``listeners/heartbeat.py:_loop`` — per-task failures don't +kill the loop). Per-proposal exceptions are caught here so one +bad proposal doesn't poison the rest of the batch. +""" + +from __future__ import annotations + +import logging +import os +from datetime import datetime, timedelta, timezone +from typing import Any, Dict, Optional + +from .observer import collect_recent_observations +from .proposer import ( + DEFAULT_MIN_CLUSTER_SIZE, + PromotionProposal, + generate_proposals, + proposal_to_dict, +) +from .store import expire_older_than, save_pending + +logger = logging.getLogger(__name__) + + +ENABLED_ENV = "KORA_PROMOTE_PHRASEBOOK_ENABLED" +INTERVAL_SEC_ENV = "KORA_PROMOTE_PHRASEBOOK_INTERVAL_SEC" +OBSERVATION_WINDOW_DAYS_ENV = ( + "KORA_PROMOTE_PHRASEBOOK_OBSERVATION_WINDOW_DAYS" +) +MIN_CLUSTER_SIZE_ENV = "KORA_PROMOTE_PHRASEBOOK_MIN_CLUSTER_SIZE" +EXPIRY_DAYS_ENV = "KORA_PROMOTE_PHRASEBOOK_EXPIRY_DAYS" + +DEFAULT_INTERVAL_SEC = 86400 # once daily +DEFAULT_OBSERVATION_WINDOW_DAYS = 7 +DEFAULT_EXPIRY_DAYS = 14 + + +def _is_enabled() -> bool: + raw = os.environ.get(ENABLED_ENV, "true").strip().lower() + # Default ON — operator can flip the env to disable per the + # ``feedback-fail-closed-by-default-for-security-infra`` + # exception path (this isn't security infra; default ON is + # the right operator value). + 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: + logger.warning( + "[kora.promote.phrasebook.cycle] %s=%r is not an int — " + "using default %d", + name, + raw, + default, + ) + return default + if value < minimum: + logger.warning( + "[kora.promote.phrasebook.cycle] %s=%d below minimum %d " + "— using default %d", + name, + value, + minimum, + default, + ) + return default + return value + + +def get_interval_seconds() -> int: + return _int_env(INTERVAL_SEC_ENV, DEFAULT_INTERVAL_SEC, minimum=60) + + +def _emit_proposed_audit( + proposal: PromotionProposal, *, synth_cost_for_proposal: float +) -> None: + """Emit one ``promotion.proposed`` row per proposal. The + synth_cost field is per-proposal so the panel can show + "cost = $0.00 (no synthesis)" vs "$0.001 (Haiku synthesized)" + without inferring it from the haiku_synthesized boolean + alone.""" + try: + from kora_cli.audit.jsonl_sink import emit_audit + except Exception as exc: + logger.warning( + "[kora.promote.phrasebook.cycle] audit import failed: " + "%r — promotion.proposed skipped", + exc, + ) + return + payload = proposal_to_dict(proposal) + payload["synth_cost_usd"] = round(synth_cost_for_proposal, 6) + try: + emit_audit( + "promotion.proposed", + payload, + caller_session_id=f"promotion:phrasebook:{proposal.proposal_id}", + source="reasoning", + ) + except Exception as exc: + logger.warning( + "[kora.promote.phrasebook.cycle] emit_audit raised %r — " + "proposal still persisted", + exc, + ) + + +async def run_phrasebook_promotion_cycle( + *, + now: Optional[datetime] = None, +) -> Dict[str, Any]: + """One cycle of the phrasebook promotion loop. + + Returns a summary dict the heartbeat scheduler logs at INFO: + + { + "enabled": bool, + "observations_read": int, + "clusters_found": int, # post-clustering, pre-thresholds + "proposals_generated": int, + "proposals_persisted": int, # may differ from generated on + # per-proposal persist failure + "expired_count": int, + "total_synth_cost_usd": float, + "started_at": ISO 8601 str, + "duration_ms": int, + } + + Caller is the heartbeat scheduler's per-task loop; failures + are swallowed there. Inside this function, individual stages + fail-soft so one stage's failure doesn't blank the summary. + """ + started_dt = now or datetime.now(timezone.utc) + started_monotonic = _monotonic_now() + + summary: Dict[str, Any] = { + "enabled": True, + "observations_read": 0, + "clusters_found": 0, + "proposals_generated": 0, + "proposals_persisted": 0, + "expired_count": 0, + "total_synth_cost_usd": 0.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.phrasebook.cycle] disabled (%s=false) — " + "skipping", + ENABLED_ENV, + ) + summary["duration_ms"] = int( + (_monotonic_now() - started_monotonic) * 1000 + ) + return summary + + window_days = _int_env( + OBSERVATION_WINDOW_DAYS_ENV, + DEFAULT_OBSERVATION_WINDOW_DAYS, + minimum=1, + ) + min_cluster_size = _int_env( + MIN_CLUSTER_SIZE_ENV, DEFAULT_MIN_CLUSTER_SIZE, minimum=2 + ) + + try: + observations = await collect_recent_observations( + since=started_dt - timedelta(days=window_days), + ) + summary["observations_read"] = len(observations) + except Exception as exc: + logger.warning( + "[kora.promote.phrasebook.cycle] observer failed: %r — " + "no proposals generated", + exc, + ) + summary["duration_ms"] = int( + (_monotonic_now() - started_monotonic) * 1000 + ) + return summary + + try: + proposals, total_synth_cost = await generate_proposals( + observations, + min_cluster_size=min_cluster_size, + now=started_dt, + ) + summary["proposals_generated"] = len(proposals) + summary["total_synth_cost_usd"] = round(total_synth_cost, 6) + # ``clusters_found`` is best-effort — we don't expose it + # from generate_proposals so reconstruct loosely as + # generated proposals (filtered clusters) plus any + # rejected-on-consistency clusters get logged but not + # counted here. Operator triages via cycle log + audit + # JSONL grep if they need the breakdown. + summary["clusters_found"] = len(proposals) + except Exception as exc: + logger.warning( + "[kora.promote.phrasebook.cycle] proposer failed: %r — " + "no proposals persisted", + exc, + ) + summary["duration_ms"] = int( + (_monotonic_now() - started_monotonic) * 1000 + ) + return summary + + # Per-proposal synth cost — split the total evenly across + # proposals (the synthesizer caller-side tracks it as one bulk + # number; for the per-row audit field we average). Future + # bucket can have generate_proposals return per-proposal cost + # if operator needs per-row precision. + per_proposal_cost = ( + (total_synth_cost / len(proposals)) if proposals else 0.0 + ) + + for proposal in proposals: + try: + save_pending(proposal) + summary["proposals_persisted"] += 1 + except Exception as exc: + logger.warning( + "[kora.promote.phrasebook.cycle] persist failed for " + "%s: %r — proposal lost (audit row still emitted)", + proposal.proposal_id, + exc, + ) + _emit_proposed_audit( + proposal, synth_cost_for_proposal=per_proposal_cost + ) + + # Sweep up old pending proposals so the operator's review + # queue doesn't grow without bound. + expiry_days = _int_env( + EXPIRY_DAYS_ENV, DEFAULT_EXPIRY_DAYS, minimum=1 + ) + try: + summary["expired_count"] = expire_older_than(days=expiry_days) + except Exception as exc: + logger.warning( + "[kora.promote.phrasebook.cycle] expire_older_than " + "raised %r — expired_count stays 0", + exc, + ) + + summary["duration_ms"] = int( + (_monotonic_now() - started_monotonic) * 1000 + ) + logger.info( + "[kora.promote.phrasebook.cycle] cycle complete: %s", + summary, + ) + return summary + + +def _monotonic_now() -> float: + import time as _time + + return _time.monotonic() diff --git a/kora_cli/promote/phrasebook/observer.py b/kora_cli/promote/phrasebook/observer.py new file mode 100644 index 000000000000..7c547603b20c --- /dev/null +++ b/kora_cli/promote/phrasebook/observer.py @@ -0,0 +1,257 @@ +"""DM observation collector — KR-PROMOTE-PHRASEBOOK-FOUNDATION (Deliverable B). + +Reads :file:`slack_dm_log.jsonl` (post-#184: handler-driven replies + +probe-wake DMs both write here with ``caller_session_id``) and +projects entries into the :class:`ReasoningObservation` shape the +proposer consumes. + +# Filtering + + * ``route_filter`` — only emit observations whose route matches. + Default ``["slack_dm"]``. Future loops may add + ``"probe_investigation"`` / ``"alert_investigation"``. + * Short-circuit hits (``model_used == "short_circuit"``) are + ALWAYS dropped — they're the things we're trying to grow, + not the things we're trying to learn from. + * Entries missing the engine path (``model_used`` absent) + are dropped — they're canned-fallback / non-reasoning paths + and don't carry the Q+A signal we need to cluster. + * Time window enforced via ``since`` (cheap timestamp parse). + +# Question text + +The slack_dm_log captures Kora's REPLY text, not Joshua's question. +The question is inferable from the ``caller_session_id`` (slack_dm +shape: ``"{channel_id}:{event_ts}"``) by reading the inbound +slack-DM JSONL. v1 ships **reply-only clustering** — the proposer +clusters by Kora's response shape (which is itself a strong +fingerprint of the question shape, since reasoning composes +similar answers for similar questions). The +``inbound_lookup`` parameter is a hook for a future bucket to +plumb the inbound side in; defaults to a no-op resolver returning +``None`` (operator_question stays empty). +""" + +from __future__ import annotations + +import json +import logging +import os +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import Callable, List, Optional + +logger = logging.getLogger(__name__) + + +_DEFAULT_ROUTE_FILTER = ("slack_dm",) + + +@dataclass(frozen=True, slots=True) +class ReasoningObservation: + """One reasoning-driven DM reply, projected for clustering.""" + + operator_question: str # may be "" in v1 (see module docstring) + kora_response: str + timestamp: datetime + caller_session_id: str + cost_usd: Optional[float] + model_used: str + route: str # mirrors slack_dm_log's source/route attribution + + +def _resolve_log_path() -> Path: + """Same env-override pattern as the handler so tests can + redirect via ``KORA_SLACK_DM_LOG_PATH``.""" + override = os.environ.get("KORA_SLACK_DM_LOG_PATH", "").strip() + if override: + return Path(override) + from kora_constants import get_kora_home + + return get_kora_home() / "slack_dm_log.jsonl" + + +def _parse_ts(raw: object) -> Optional[datetime]: + """ISO 8601 (possibly Z-suffixed) → aware datetime. None on + malformed input.""" + if not isinstance(raw, str) or not raw: + return None + try: + if raw.endswith("Z"): + raw = raw[:-1] + "+00:00" + dt = datetime.fromisoformat(raw) + except ValueError: + return None + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + return dt + + +def _entry_route(entry: dict) -> str: + """Project the entry's route hint. The handler's + ``caller_session_id`` shape encodes the source: + + * ``":"`` (slack_dm) — has a colon + but doesn't start with one of the structured prefixes. + * ``"probe::"`` (probe_investigation). + * ``"email:"`` (email). + * ``"mcp::"`` (mcp). + + Defaults to ``"slack_dm"`` when shape doesn't match a known + structured prefix (operator-DM is the v1 happy path). + """ + csid = entry.get("caller_session_id") or "" + if isinstance(csid, str): + if csid.startswith("probe:"): + return "probe_investigation" + if csid.startswith("email:"): + return "email" + if csid.startswith("mcp:"): + return "mcp" + return "slack_dm" + + +def _estimate_cost_usd(entry: dict) -> Optional[float]: + """Compute per-call cost from tokens + model via the canonical + pricing helper. Returns ``None`` when model is unknown / fields + are missing. Mirrors the wake_consumer's ``_compute_total_cost_usd`` + helper (same canonical-pricing path).""" + model = entry.get("model_used") + if not isinstance(model, str) or not model: + return None + try: + from agent.usage_pricing import CanonicalUsage, estimate_usage_cost + except Exception: + return None + + def _int_or_zero(key: str) -> int: + value = entry.get(key) + return value if isinstance(value, int) else 0 + + usage = CanonicalUsage( + input_tokens=_int_or_zero("input_tokens"), + output_tokens=_int_or_zero("output_tokens"), + cache_read_tokens=_int_or_zero("cache_read_input_tokens"), + cache_write_tokens=_int_or_zero("cache_creation_input_tokens"), + ) + try: + result = estimate_usage_cost(model, usage) + except Exception: + return None + if result.status == "unknown" or result.amount_usd is None: + return None + return float(result.amount_usd) + + +InboundLookup = Callable[[str], Optional[str]] + + +async def collect_recent_observations( + *, + since: datetime, + route_filter: Optional[List[str]] = None, + inbound_lookup: Optional[InboundLookup] = None, + log_path: Optional[Path] = None, +) -> List[ReasoningObservation]: + """Read the slack_dm outbound log and project entries into + :class:`ReasoningObservation`s. + + Args: + since: Lower bound (aware datetime; UTC assumed if naive). + Entries with ``sent_at < since`` are skipped. + route_filter: Allow-list of routes to retain. Defaults to + ``["slack_dm"]``. + inbound_lookup: Optional resolver + ``caller_session_id → operator_question``. v1 callers + leave ``None``; future buckets may plumb the inbound side. + log_path: Override for tests; production resolves via + :func:`_resolve_log_path`. + + Returns observations sorted by ``timestamp`` ascending. + """ + if since.tzinfo is None: + since = since.replace(tzinfo=timezone.utc) + routes = tuple(route_filter) if route_filter is not None else _DEFAULT_ROUTE_FILTER + target = log_path or _resolve_log_path() + if not target.is_file(): + return [] + + out: List[ReasoningObservation] = [] + try: + raw_text = target.read_text(encoding="utf-8") + except OSError as exc: + logger.warning( + "[kora.promote.phrasebook.observer] read failed for %s: %r", + target, + exc, + ) + return [] + + for lineno, line in enumerate(raw_text.splitlines(), start=1): + line = line.strip() + if not line: + continue + try: + entry = json.loads(line) + except json.JSONDecodeError as exc: + logger.debug( + "[kora.promote.phrasebook.observer] line %d malformed " + "JSON, skipped: %r", + lineno, + exc, + ) + continue + if not isinstance(entry, dict): + continue + + # Filter: must be an engine-driven reply (model_used + # present + non-short_circuit) sent successfully. + model = entry.get("model_used") + if not isinstance(model, str) or not model: + continue + if model == "short_circuit": + continue + if entry.get("send_status") != "ok": + continue + text = entry.get("text") + if not isinstance(text, str) or not text.strip(): + continue + + ts = _parse_ts(entry.get("sent_at")) + if ts is None or ts < since: + continue + + route = _entry_route(entry) + if route not in routes: + continue + + csid = entry.get("caller_session_id") or "" + operator_question = "" + if inbound_lookup is not None and isinstance(csid, str) and csid: + try: + resolved = inbound_lookup(csid) + if isinstance(resolved, str): + operator_question = resolved + except Exception as exc: + logger.debug( + "[kora.promote.phrasebook.observer] inbound_lookup " + "raised %r for csid=%r — operator_question stays empty", + exc, + csid, + ) + + out.append( + ReasoningObservation( + operator_question=operator_question, + kora_response=text, + timestamp=ts, + caller_session_id=str(csid) if csid else "", + cost_usd=_estimate_cost_usd(entry), + model_used=model, + route=route, + ) + ) + + out.sort(key=lambda o: o.timestamp) + return out diff --git a/kora_cli/promote/phrasebook/proposer.py b/kora_cli/promote/phrasebook/proposer.py new file mode 100644 index 000000000000..47ffda9ca4c8 --- /dev/null +++ b/kora_cli/promote/phrasebook/proposer.py @@ -0,0 +1,537 @@ +"""Promotion proposal generator — KR-PROMOTE-PHRASEBOOK-FOUNDATION (Deliverable C). + +Input: a list of :class:`ReasoningObservation` from the observer. +Output: zero or more :class:`PromotionProposal` records, each +suggesting a new phrasebook entry that would short-circuit a +cluster of recurring operator DMs. + +# Pipeline + + 1. Embed each observation's ``kora_response`` text via + :func:`kora_cli.clustering.text_similarity.embed_texts` + (lexical, $0). + 2. Cluster via greedy agglomerative similarity at + ``cohesion_threshold`` (default 0.85). + 3. For each cluster ≥ ``min_cluster_size``: + a. Check answer-consistency: the LARGEST sub-cluster of + near-identical Kora answers must cover at least + ``answer_consistency_threshold`` (default 0.75) of the + outer cluster. Divergent-answer clusters get dropped — + a proposed template that doesn't match Kora's actual + behavior is worse than no proposal. + b. Derive a candidate regex pattern from common + tokens / phrases. + c. Derive the reply template from the dominant Kora answer + (or optionally synthesize via Haiku — one call per + proposal). Defaults to "dominant answer" because it's + $0 and forces operator review of any Haiku-synthesized + text anyway. + d. Mint a :class:`PromotionProposal` with confidence + derived from cluster cohesion + answer consistency. + +# Cost discipline + +Lexical embedder: $0. Optional Haiku reply-template synthesis: +~$0.001 per proposal at default thresholds (5-observation +clusters with consistent answers → maybe 2-5 proposals per cycle +in realistic operator-DM volumes). Per cycle: ≤$0.005. Daily +ceiling: well within the $0.01-0.05/day target. + +The bucket's STOP-ASK §4 anticipated "pattern derivation generates +pathological regexes" — we mitigate by **escaping all regex +metacharacters** in the derived pattern and emitting a simple +case-insensitive alternation of the cluster's dominant short +phrases. The phrasebook PUT endpoint's existing validator (PR #177 +``phrasebook_editor.validate_entries``) will reject any malformed +regex at approve-time anyway, so the worst case is a rejected +operator approval rather than a runtime regex bomb. +""" + +from __future__ import annotations + +import logging +import re +import uuid +from collections import Counter +from dataclasses import asdict, dataclass, field +from datetime import datetime, timezone +from typing import Any, Awaitable, Callable, Dict, List, Literal, Optional, Tuple + +from kora_cli.clustering.text_similarity import ( + TextEmbedding, + cluster_by_similarity, + cosine_similarity, + embed_texts, +) + +from .observer import ReasoningObservation + +logger = logging.getLogger(__name__) + + +ProposalStatus = Literal["pending", "approved", "rejected", "expired"] + +# Wire-stable status values — paired with the FE constant CC#2 +# adds in KR-FE-PROMOTION-REVIEW-PANEL. Update both sides together +# when extending. +PROPOSAL_STATUS_VALUES: Tuple[ProposalStatus, ...] = ( + "pending", + "approved", + "rejected", + "expired", +) + + +# Default thresholds — kept conservative so operator isn't flooded +# with marginal proposals on day one. Operator can loosen via the +# cycle's keyword args (cron task reads from env). +DEFAULT_MIN_CLUSTER_SIZE = 5 +DEFAULT_COHESION_THRESHOLD = 0.85 +DEFAULT_ANSWER_CONSISTENCY_THRESHOLD = 0.75 + +# Sample-question cap exposed to the operator in the proposal +# review UI. Three is enough to recognize the cluster shape +# without overwhelming the panel. +SAMPLE_QUESTIONS_CAP = 3 + + +@dataclass(frozen=True, slots=True) +class PromotionProposal: + """Wire-stable proposal shape. Persisted as JSON by + :mod:`.store` and surfaced via the cockpit API endpoints in + web_server.py. + """ + + proposal_id: str + cluster_size: int + sample_questions: List[str] + proposed_pattern: str + proposed_reply_template: str + proposed_category: str + confidence: float + created_at: datetime + status: ProposalStatus + review_notes: str = "" + # KR-PROMOTE-PHRASEBOOK-FOUNDATION — back-references so the + # cockpit can render the observation context that triggered + # the proposal without re-running the cycle. + cluster_caller_session_ids: List[str] = field(default_factory=list) + # ``haiku_synthesized`` is True when the reply_template was + # generated by a Haiku call; False when it's the dominant + # cluster member verbatim. Operator approval UI can flag + # synthesized text for extra-careful review. + haiku_synthesized: bool = False + + +def _format_iso(dt: datetime) -> str: + return dt.strftime("%Y-%m-%dT%H:%M:%SZ") + + +def proposal_to_dict(p: PromotionProposal) -> Dict[str, Any]: + """Serialize a PromotionProposal to a JSON-safe dict for both + audit emission and on-disk persistence.""" + out = asdict(p) + out["created_at"] = _format_iso(p.created_at) + return out + + +def proposal_from_dict(payload: Dict[str, Any]) -> PromotionProposal: + """Rehydrate a proposal from its serialized form. Strict + field-presence (no defaulting of required fields) so a + malformed file surfaces loudly at load time.""" + ts_raw = payload.get("created_at", "") + if isinstance(ts_raw, str) and ts_raw: + if ts_raw.endswith("Z"): + ts_raw = ts_raw[:-1] + "+00:00" + created_at = datetime.fromisoformat(ts_raw) + if created_at.tzinfo is None: + created_at = created_at.replace(tzinfo=timezone.utc) + else: + created_at = datetime.now(timezone.utc) + return PromotionProposal( + proposal_id=str(payload["proposal_id"]), + cluster_size=int(payload["cluster_size"]), + sample_questions=list(payload.get("sample_questions") or []), + proposed_pattern=str(payload["proposed_pattern"]), + proposed_reply_template=str(payload["proposed_reply_template"]), + proposed_category=str(payload["proposed_category"]), + confidence=float(payload.get("confidence") or 0.0), + created_at=created_at, + status=str(payload.get("status") or "pending"), # type: ignore[arg-type] + review_notes=str(payload.get("review_notes") or ""), + cluster_caller_session_ids=list( + payload.get("cluster_caller_session_ids") or [] + ), + haiku_synthesized=bool(payload.get("haiku_synthesized") or False), + ) + + +# --------------------------------------------------------------------------- +# Pattern + template + category derivation +# --------------------------------------------------------------------------- + + +_STOPWORDS = frozenset( + { + "a", + "an", + "and", + "are", + "as", + "at", + "be", + "by", + "for", + "from", + "i", + "in", + "is", + "it", + "of", + "on", + "or", + "that", + "the", + "this", + "to", + "was", + "what", + "with", + "you", + } +) + + +_TOKEN_RE = re.compile(r"[A-Za-z0-9]+") + + +def _meaningful_tokens(text: str) -> List[str]: + return [ + t + for t in _TOKEN_RE.findall(text.lower()) + if t not in _STOPWORDS and len(t) >= 2 + ] + + +def _dominant_phrase(observations: List[ReasoningObservation]) -> str: + """Return the longest token-substring that appears in + ``≥ half`` the observations' RESPONSES. Used as the + inferred-question-shape signal for pattern derivation since + v1 ships reply-only clustering. + + Reply-shape ↔ question-shape correspondence: similar + reasoning answers are evidence of similar reasoning inputs. + For short DMs this proxy holds; the resulting pattern is the + operator's responsibility to refine on approval. + + Fallback: most common 1-gram across responses. + """ + docs = [_meaningful_tokens(o.kora_response) for o in observations] + if not docs: + return "" + # Score each token by document-frequency. + df: Counter = Counter() + for doc in docs: + for tok in set(doc): + df[tok] += 1 + if not df: + return "" + most_common = df.most_common(1)[0][0] + return most_common + + +def _derive_pattern( + observations: List[ReasoningObservation], +) -> str: + """Generate a conservative regex pattern from the cluster's + common-token vocabulary. We use a simple alternation of the + top-3 cross-observation tokens with all regex metacharacters + escaped — pathological inputs (extension points like ``(`` or + ``*``) can't bomb the engine at compile-time. + + The pattern is intentionally LOOSE — it's the operator's job + to tighten it on approval if they don't like the breadth. The + spec's STOP-ASK §4 anticipated regex pathology; we mitigate + via escape + alternation + the editor's own validator at + approve-time. + """ + # Build doc-frequency over tokens in the RESPONSES (proxy for + # question shape; v1 reply-only clustering). + df: Counter = Counter() + docs = [_meaningful_tokens(o.kora_response) for o in observations] + for doc in docs: + for tok in set(doc): + df[tok] += 1 + if not df: + return "(?i).*" # operator must refine; defensive default + top = [tok for tok, _ in df.most_common(3)] + escaped = [re.escape(t) for t in top] + return "(?i)(" + "|".join(escaped) + ")" + + +def _derive_category( + observations: List[ReasoningObservation], +) -> str: + """Heuristic: match keywords against the existing phrasebook's + known categories so cockpit grouping stays consistent. + + Falls back to ``"operator_query"`` (a stable default name + operator can rename on approval) — the phrasebook editor's + validator accepts arbitrary non-empty category strings, so + the fallback is always valid. + """ + combined = " ".join(o.kora_response for o in observations).lower() + keyword_to_category = { + "burn": "cost_query", + "cost": "cost_query", + "spend": "cost_query", + "alert": "alerts_query", + "alerts": "alerts_query", + "paused": "state_query", + "pause": "state_query", + "state": "state_query", + "healthy": "health_query", + "health": "health_query", + "ticket": "tickets_query", + "sea": "tickets_query", + } + for keyword, category in keyword_to_category.items(): + if re.search(rf"\b{re.escape(keyword)}\b", combined): + return category + return "operator_query" + + +def _answer_consistency_subcluster( + observations: List[ReasoningObservation], + *, + threshold: float = 0.75, +) -> Tuple[List[ReasoningObservation], float]: + """Find the largest sub-cluster of near-identical responses. + + Returns ``(dominant_subcluster, fraction_of_outer)``. When + fraction < ``threshold`` the proposer drops the cluster (the + cluster's questions are similar but Kora's answers diverge — + a template would mislead). + """ + if not observations: + return ([], 0.0) + n = len(observations) + # Embed responses fresh via the synchronous helper from the + # clustering module — avoids event-loop entanglement in this + # sync subroutine. embed_texts is async only because of its + # forward-compat with a Haiku-backed embedder; the lexical + # path is microseconds and pure-CPU. + from kora_cli.clustering.text_similarity import _compute_embedding + + embs = [ + TextEmbedding( + text=o.kora_response, + embedding=_compute_embedding(o.kora_response), + cached=False, + ) + for o in observations + ] + # Greedy: pick the first as seed; gather all observations with + # cosine ≥ 0.92 (intra-answer consistency is tighter than + # cross-cluster cohesion). Repeat with each unclaimed seed; + # keep the largest subcluster. + INTRA_THRESHOLD = 0.92 + best_subset: List[int] = [] + for seed_idx in range(n): + members = [seed_idx] + for other_idx in range(n): + if other_idx == seed_idx: + continue + if ( + cosine_similarity(embs[seed_idx], embs[other_idx]) + >= INTRA_THRESHOLD + ): + members.append(other_idx) + if len(members) > len(best_subset): + best_subset = members + fraction = len(best_subset) / n + dominant = [observations[i] for i in best_subset] + return (dominant, fraction) + + +def _dominant_response_text( + observations: List[ReasoningObservation], +) -> str: + """Return the most-common verbatim response (after light + whitespace normalization). Used as the proposed reply + template when no Haiku synthesis is requested.""" + if not observations: + return "" + counter: Counter = Counter( + " ".join(o.kora_response.split()) for o in observations + ) + return counter.most_common(1)[0][0] + + +# --------------------------------------------------------------------------- +# Optional Haiku-driven reply-template synthesis +# --------------------------------------------------------------------------- + + +ReplyTemplateSynthesizer = Callable[ + [List[ReasoningObservation]], Awaitable[Tuple[Optional[str], float]] +] + + +async def _default_synthesizer( + observations: List[ReasoningObservation], +) -> Tuple[Optional[str], float]: + """Returns ``(template, cost_usd)``. Default returns + ``(None, 0.0)`` — caller falls back to the dominant verbatim + response. Production wiring substitutes a Haiku-call shim; + tests inject a deterministic stub.""" + return (None, 0.0) + + +async def generate_proposals( + observations: List[ReasoningObservation], + *, + min_cluster_size: int = DEFAULT_MIN_CLUSTER_SIZE, + cohesion_threshold: float = DEFAULT_COHESION_THRESHOLD, + answer_consistency_threshold: float = ( + DEFAULT_ANSWER_CONSISTENCY_THRESHOLD + ), + synthesize_reply_template: Optional[ReplyTemplateSynthesizer] = None, + now: Optional[datetime] = None, +) -> Tuple[List[PromotionProposal], float]: + """Cluster + derive proposals. Returns + ``(proposals, total_haiku_cost_usd)`` — caller surfaces the + cost in the cycle log + audit. + + Args: + observations: From :func:`observer.collect_recent_observations`. + min_cluster_size: Drop clusters smaller than this. + cohesion_threshold: Greedy-agglomerative similarity floor. + answer_consistency_threshold: Fraction of cluster that must + share a near-identical response. Below this → cluster + dropped (template would mislead). + synthesize_reply_template: Optional Haiku-based template + synthesizer. ``None`` → use dominant verbatim response. + now: Test-injectable timestamp for proposal created_at. + + Determinism: identical inputs ⇒ identical proposal_ids, + pattern strings, and ordering (lexical embedder is + deterministic; uuid4 is the only nondeterminism and lives + only on the proposal_id field — extracted into ``now`` only + if the caller wants reproducibility). + """ + if not observations: + return ([], 0.0) + synth = synthesize_reply_template or _default_synthesizer + timestamp = now or datetime.now(timezone.utc) + + embeddings = await embed_texts( + [o.kora_response for o in observations] + ) + clusters = cluster_by_similarity( + embeddings, threshold=cohesion_threshold + ) + + proposals: List[PromotionProposal] = [] + total_synth_cost = 0.0 + + for cluster in clusters: + if len(cluster) < min_cluster_size: + continue + # Re-associate cluster embeddings back to their observations. + # The embeddings preserve input order; we use the text-to- + # observation mapping for lookup (texts are unique per call + # site under normal load; collisions just fold harmlessly). + text_to_obs: Dict[str, ReasoningObservation] = {} + for obs in observations: + text_to_obs.setdefault(obs.kora_response, obs) + cluster_obs = [text_to_obs[e.text] for e in cluster if e.text in text_to_obs] + if len(cluster_obs) < min_cluster_size: + continue + + dominant_obs, consistency = _answer_consistency_subcluster( + cluster_obs, + threshold=answer_consistency_threshold, + ) + if consistency < answer_consistency_threshold: + logger.info( + "[kora.promote.phrasebook.proposer] cluster size=%d " + "rejected — answer consistency %.2f < threshold %.2f", + len(cluster_obs), + consistency, + answer_consistency_threshold, + ) + continue + + # Pattern + category from the broader cluster (more signal + # → better pattern). Reply template from the dominant + # sub-cluster (consistency → safer template). + pattern = _derive_pattern(cluster_obs) + category = _derive_category(cluster_obs) + synthesized_template, synth_cost = await synth(dominant_obs) + total_synth_cost += synth_cost + if synthesized_template: + reply_template = synthesized_template + haiku_synth = True + else: + reply_template = _dominant_response_text(dominant_obs) + haiku_synth = False + + # Cohesion proxy for confidence: average max-link sim + # across the cluster. + cohesion = _cluster_cohesion(cluster) + confidence = round((cohesion + consistency) / 2.0, 4) + + sample_q: List[str] = [] + for obs in cluster_obs[:SAMPLE_QUESTIONS_CAP]: + # v1 reply-only clustering — operator_question is + # typically "" so we surface the response excerpt + # instead (renamed in the cockpit panel as "what Kora + # has been saying" so operator can recognize the + # cluster). + if obs.operator_question: + sample_q.append(obs.operator_question) + else: + sample_q.append(obs.kora_response[:200]) + + proposal = PromotionProposal( + proposal_id=str(uuid.uuid4()), + cluster_size=len(cluster_obs), + sample_questions=sample_q, + proposed_pattern=pattern, + proposed_reply_template=reply_template, + proposed_category=category, + confidence=confidence, + created_at=timestamp, + status="pending", + review_notes="", + cluster_caller_session_ids=[ + o.caller_session_id for o in cluster_obs + ], + haiku_synthesized=haiku_synth, + ) + proposals.append(proposal) + + # Sort newest+highest-confidence first — operator's primary + # triage view in the cockpit panel. + proposals.sort(key=lambda p: (-p.confidence, -p.cluster_size)) + return (proposals, total_synth_cost) + + +def _cluster_cohesion(cluster: List[TextEmbedding]) -> float: + """Average max-link similarity across the cluster's members. + Cohesion ≥ threshold is enforced at clustering time; this + measures HOW MUCH ABOVE threshold the cluster is — useful + confidence signal.""" + n = len(cluster) + if n <= 1: + return 1.0 + total = 0.0 + pairs = 0 + for i in range(n): + for j in range(i + 1, n): + total += cosine_similarity(cluster[i], cluster[j]) + pairs += 1 + if pairs == 0: + return 1.0 + return total / pairs diff --git a/kora_cli/promote/phrasebook/store.py b/kora_cli/promote/phrasebook/store.py new file mode 100644 index 000000000000..89922c3fd834 --- /dev/null +++ b/kora_cli/promote/phrasebook/store.py @@ -0,0 +1,294 @@ +"""Proposal persistence — KR-PROMOTE-PHRASEBOOK-FOUNDATION (Deliverable D persistence side). + +Pending proposals live at +``${KORA_HOME}/promotions/phrasebook/pending/.json``. +Approved/rejected proposals get moved into sibling ``approved/`` / +``rejected/`` directories on status transition (atomic rename). + +Why files instead of substrate / SQLite: this loop runs at idle +cadence and produces a handful of proposals per day. File-per- +proposal makes operator triage trivial (curl the directory, jq +the JSONs, git-diff before/after operator edits) without +requiring substrate round-trips. The audit JSONL is the +forensic-truth stream; the files are the live working set. + +# Path layout + +``` +${KORA_HOME}/promotions/ + phrasebook/ + pending/ .json + approved/ .json + rejected/ .json +``` + +# Audit emission + +Status transitions emit one of the three new ``promotion.*`` +seams. Persistence + audit are separate concerns; the store +module owns persistence and the cycle / endpoints own audit. +""" + +from __future__ import annotations + +import json +import logging +import os +from datetime import datetime, timezone +from pathlib import Path +from typing import List, Optional + +from .proposer import ( + PROPOSAL_STATUS_VALUES, + PromotionProposal, + proposal_from_dict, + proposal_to_dict, +) + +logger = logging.getLogger(__name__) + + +PROMOTIONS_ROOT_ENV = "KORA_PROMOTIONS_DIR" +_PROMOTIONS_RELATIVE = Path("promotions") / "phrasebook" + + +class ProposalNotFound(LookupError): + """Raised when an endpoint references a proposal_id that + doesn't exist in any of the status subdirectories.""" + + +def _root() -> Path: + """Return the promotions root for the phrasebook loop. + + Honors ``KORA_PROMOTIONS_DIR`` for tests that don't want to + use the canonical KORA_HOME path; falls back to + ``${KORA_HOME}/promotions/phrasebook`` otherwise. + """ + override = os.environ.get(PROMOTIONS_ROOT_ENV, "").strip() + if override: + return Path(override) / "phrasebook" + from kora_constants import get_kora_home + + return get_kora_home() / _PROMOTIONS_RELATIVE + + +def _status_dir(status: str) -> Path: + if status not in PROPOSAL_STATUS_VALUES: + raise ValueError(f"unknown proposal status: {status!r}") + return _root() / status + + +def save_pending(proposal: PromotionProposal) -> Path: + """Atomic write a pending proposal. Returns the file path.""" + if proposal.status != "pending": + raise ValueError( + f"save_pending called with status={proposal.status!r}" + ) + target_dir = _status_dir("pending") + 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) + return target + + +def list_pending() -> List[PromotionProposal]: + """Read all pending proposals, sorted highest-confidence first + (operator's primary review order).""" + return _list_status("pending", sort_by_confidence=True) + + +def list_by_status(status: str) -> List[PromotionProposal]: + """Operator-debug helper. Same order as list_pending for + confidence comparability.""" + return _list_status(status, sort_by_confidence=True) + + +def _list_status( + status: str, *, sort_by_confidence: bool = False +) -> List[PromotionProposal]: + target_dir = _status_dir(status) + if not target_dir.is_dir(): + return [] + out: List[PromotionProposal] = [] + for path in 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.phrasebook.store] %s unreadable, " + "skipped: %r", + path, + exc, + ) + continue + try: + out.append(proposal_from_dict(payload)) + except Exception as exc: + logger.warning( + "[kora.promote.phrasebook.store] %s malformed, " + "skipped: %r", + path, + exc, + ) + continue + if sort_by_confidence: + out.sort(key=lambda p: (-p.confidence, -p.cluster_size)) + return out + + +def load(proposal_id: str) -> PromotionProposal: + """Look up a proposal across all status directories. Raises + :class:`ProposalNotFound` when no file matches.""" + for status in PROPOSAL_STATUS_VALUES: + path = _status_dir(status) / f"{proposal_id}.json" + if path.is_file(): + try: + payload = json.loads(path.read_text(encoding="utf-8")) + return proposal_from_dict(payload) + except Exception as exc: + raise ProposalNotFound( + f"proposal {proposal_id!r} exists at {path} but " + f"failed to load: {exc!r}" + ) from exc + raise ProposalNotFound(f"proposal {proposal_id!r} not found") + + +def transition( + proposal_id: str, + *, + new_status: str, + review_notes: str = "", + overrides: Optional[dict] = None, +) -> PromotionProposal: + """Move a proposal from its current status directory into + ``new_status``'s directory. Optionally apply override fields + (pattern / reply_template / category) the operator edited at + approve-time. + + Returns the post-transition proposal. Raises + :class:`ProposalNotFound` when the proposal_id is not + present. + + Operation order: rehydrate → apply overrides → write to new + directory → unlink old. The write-before-unlink shape means + a crash mid-transition leaves BOTH files; recovery is + operator-readable (list both directories). + """ + if new_status not in PROPOSAL_STATUS_VALUES: + raise ValueError(f"unknown proposal status: {new_status!r}") + current_path: Optional[Path] = None + current_status: Optional[str] = None + for status in PROPOSAL_STATUS_VALUES: + candidate = _status_dir(status) / f"{proposal_id}.json" + if candidate.is_file(): + current_path = candidate + current_status = status + break + if current_path is None or current_status is None: + raise ProposalNotFound(f"proposal {proposal_id!r} not found") + + payload = json.loads(current_path.read_text(encoding="utf-8")) + proposal = proposal_from_dict(payload) + + if overrides: + # Whitelist of operator-editable fields. Anything else is + # ignored — the proposer's other fields (cluster_size, + # sample_questions, confidence, etc.) are read-only audit + # context. + editable = {"pattern", "reply_template", "category"} + unknown = set(overrides.keys()) - editable + if unknown: + logger.debug( + "[kora.promote.phrasebook.store] ignoring unknown " + "override keys: %s", + sorted(unknown), + ) + pattern = overrides.get("pattern") or proposal.proposed_pattern + reply_template = ( + overrides.get("reply_template") + or proposal.proposed_reply_template + ) + category = ( + overrides.get("category") or proposal.proposed_category + ) + # Rebuild with overrides applied. The original + # proposed_* fields persist as the audit-trail values + # — overrides are recorded separately via the review_notes + # field so the post-edit shape is recoverable. + proposal = PromotionProposal( + proposal_id=proposal.proposal_id, + cluster_size=proposal.cluster_size, + sample_questions=proposal.sample_questions, + proposed_pattern=pattern, + proposed_reply_template=reply_template, + proposed_category=category, + confidence=proposal.confidence, + created_at=proposal.created_at, + status=new_status, # type: ignore[arg-type] + review_notes=review_notes, + cluster_caller_session_ids=proposal.cluster_caller_session_ids, + haiku_synthesized=proposal.haiku_synthesized, + ) + else: + proposal = PromotionProposal( + proposal_id=proposal.proposal_id, + cluster_size=proposal.cluster_size, + sample_questions=proposal.sample_questions, + proposed_pattern=proposal.proposed_pattern, + proposed_reply_template=proposal.proposed_reply_template, + proposed_category=proposal.proposed_category, + confidence=proposal.confidence, + created_at=proposal.created_at, + status=new_status, # type: ignore[arg-type] + review_notes=review_notes, + cluster_caller_session_ids=proposal.cluster_caller_session_ids, + haiku_synthesized=proposal.haiku_synthesized, + ) + + target_dir = _status_dir(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(proposal_to_dict(proposal), indent=2), + encoding="utf-8", + ) + os.replace(tmp, target) + if current_path != target: + try: + current_path.unlink() + except OSError as exc: + logger.warning( + "[kora.promote.phrasebook.store] old path %s unlink " + "failed: %r — operator can clean manually", + current_path, + exc, + ) + return proposal + + +def expire_older_than(*, days: int) -> int: + """Move pending proposals older than ``days`` to the + ``expired/`` directory. Returns count moved. Intended for the + cron task to call after generation so the pending list stays + operator-actionable.""" + cutoff = datetime.now(timezone.utc).timestamp() - days * 86400 + pending = list_pending() + expired = 0 + for proposal in pending: + ts = proposal.created_at.timestamp() + if ts < cutoff: + transition( + proposal.proposal_id, + new_status="expired", + review_notes=f"auto-expired after {days} days pending", + ) + expired += 1 + return expired diff --git a/kora_cli/web_server.py b/kora_cli/web_server.py index 155e18b50333..bc76c15bdacf 100644 --- a/kora_cli/web_server.py +++ b/kora_cli/web_server.py @@ -6507,6 +6507,309 @@ async def get_phrasebook_backups() -> Dict[str, Any]: } +# --------------------------------------------------------------------------- +# Phrasebook promotion review — KR-PROMOTE-PHRASEBOOK-FOUNDATION (Deliverable E) +# --------------------------------------------------------------------------- +# +# Three endpoints driving the operator-approval UX. CC#2's +# KR-FE-PROMOTION-REVIEW-PANEL follow-on reads/writes these. +# +# * GET /api/promotions/phrasebook/pending — list pending +# * POST /api/promotions/phrasebook/{id}/approve — approve + PUT phrasebook +# * POST /api/promotions/phrasebook/{id}/reject — reject +# +# Drift-guard pin: ``_PROMOTION_STATUS_VALUES`` mirrors the proposer +# module's ``PROPOSAL_STATUS_VALUES``. The KR-FE-PROMOTION-REVIEW-PANEL +# follow-on adds the symmetric FE constant + a snapshot-pin test that +# fails CI if the two drift. + + +# Wire-stable status allowlist — paired with the FE constant added +# by KR-FE-PROMOTION-REVIEW-PANEL. +_PROMOTION_STATUS_VALUES: Tuple[str, ...] = ( + "pending", + "approved", + "rejected", + "expired", +) + + +@app.get("/api/promotions/phrasebook/pending") +async def list_pending_phrasebook_proposals() -> Dict[str, Any]: + """Return all pending phrasebook proposals, highest-confidence + first. Each entry is the full PromotionProposal projection + (see ``kora_cli.promote.phrasebook.proposer.proposal_to_dict``) + so the cockpit panel has everything it needs to render the + review surface in one round-trip. + + Sidebar-nav count is ``len(response["proposals"])``. + """ + from kora_cli.promote.phrasebook.proposer import proposal_to_dict + from kora_cli.promote.phrasebook.store import list_pending + + proposals = list_pending() + return { + "proposals": [proposal_to_dict(p) for p in proposals], + "status_values": list(_PROMOTION_STATUS_VALUES), + } + + +@app.post("/api/promotions/phrasebook/{proposal_id}/approve") +async def approve_phrasebook_proposal( + proposal_id: str, payload: Optional[Dict[str, Any]] = None +) -> Any: + """Approve a pending proposal. Optional payload override + fields the operator edited at approve-time: + + {pattern_override?, reply_template_override?, + category_override?, review_notes?} + + Workflow: + 1. Load the pending proposal (404 if missing / not + pending). + 2. Build the post-override PhrasebookEntry shape. + 3. Validate via the existing phrasebook editor's validator + (regex compiles, template references real fields, etc). + 4. PUT to the operator-override phrasebook with + ``actor="kora_proposal_approved"`` per #177 + forward-compat. + 5. Transition the proposal to ``approved/`` directory. + 6. Emit ``promotion.approved`` audit row. + """ + from kora_cli.audit import emit_audit + from kora_cli.promote.phrasebook.proposer import proposal_to_dict + from kora_cli.promote.phrasebook.store import ( + ProposalNotFound, + load, + transition, + ) + from kora_cli.short_circuit import dm_phrasebook, phrasebook_editor + + overrides_dict: Dict[str, Any] = {} + review_notes = "" + if isinstance(payload, dict): + pattern_override = payload.get("pattern_override") + if isinstance(pattern_override, str): + overrides_dict["pattern"] = pattern_override + reply_template_override = payload.get("reply_template_override") + if isinstance(reply_template_override, str): + overrides_dict["reply_template"] = reply_template_override + category_override = payload.get("category_override") + if isinstance(category_override, str): + overrides_dict["category"] = category_override + notes_raw = payload.get("review_notes") + if isinstance(notes_raw, str): + review_notes = notes_raw + + try: + existing = load(proposal_id) + except ProposalNotFound: + return JSONResponse( + status_code=404, + content={"error": "proposal_not_found", "proposal_id": proposal_id}, + ) + if existing.status != "pending": + return JSONResponse( + status_code=409, + content={ + "error": "proposal_not_pending", + "proposal_id": proposal_id, + "current_status": existing.status, + }, + ) + + final_pattern = overrides_dict.get("pattern") or existing.proposed_pattern + final_reply_template = ( + overrides_dict.get("reply_template") + or existing.proposed_reply_template + ) + final_category = ( + overrides_dict.get("category") or existing.proposed_category + ) + + new_entry = { + "pattern": final_pattern, + "category": final_category, + "description": ( + f"Promoted from Kora proposal {proposal_id} " + f"(cluster size {existing.cluster_size}, " + f"confidence {existing.confidence:.2f})" + ), + "reply_template": final_reply_template, + } + + # Merge into existing override entries — promotion ADDS, never + # replaces. Operator edits the result later via the PUT + # endpoint if they want different ordering. + current_entries: List[Dict[str, Any]] = [] + for entry in dm_phrasebook.load_phrasebook(): + current_entries.append( + { + "pattern": entry.pattern.pattern, + "category": entry.category, + "description": entry.description, + "reply_template": entry.reply_template, + } + ) + proposed_entries = current_entries + [new_entry] + validation_errors = phrasebook_editor.validate_entries( + proposed_entries + ) + if validation_errors: + return JSONResponse( + status_code=422, + content={ + "error": "validation_failed", + "errors": [e.as_dict() for e in validation_errors], + "proposal_id": proposal_id, + }, + ) + + count_before = len(current_entries) + override_path = phrasebook_editor._override_path() + backup_path = phrasebook_editor.write_backup_for(override_path) + try: + phrasebook_editor.write_phrasebook(proposed_entries) + except Exception as exc: + return JSONResponse( + status_code=500, + content={ + "error": "phrasebook_write_failed", + "detail": repr(exc), + "backup_filename": ( + backup_path.name if backup_path is not None else None + ), + }, + ) + + # ``phrasebook.updated`` audit row uses the forward-compat + # actor literal per PR #177 so the promotion-history view can + # tell operator-edits apart from auto-approved promotions. + try: + emit_audit( + seam="phrasebook.updated", + details={ + "actor": "kora_proposal_approved", + "action": "put", + "entry_count_before": count_before, + "entry_count_after": len(proposed_entries), + "backup_filename": ( + backup_path.name if backup_path is not None else None + ), + "proposal_id": proposal_id, + }, + source=None, + ) + except Exception as exc: + logger.warning( + "[kora.promote] phrasebook.updated emit raised %r — " + "approval still succeeded", + exc, + ) + + # Move the proposal to approved/ with operator overrides + # baked into the persisted record so the audit JSONL + + # on-disk file agree. + updated = transition( + proposal_id, + new_status="approved", + review_notes=review_notes, + overrides=overrides_dict or None, + ) + + try: + emit_audit( + seam="promotion.approved", + details={ + **proposal_to_dict(updated), + "committed_entry": new_entry, + }, + caller_session_id=f"promotion:phrasebook:{proposal_id}", + source="reasoning", + ) + except Exception as exc: + logger.warning( + "[kora.promote] promotion.approved emit raised %r — " + "approval persisted; audit row missing", + exc, + ) + + return { + "proposal_id": proposal_id, + "status": "approved", + "committed_entry": new_entry, + "entry_count_after": len(proposed_entries), + "backup_filename": ( + backup_path.name if backup_path is not None else None + ), + } + + +@app.post("/api/promotions/phrasebook/{proposal_id}/reject") +async def reject_phrasebook_proposal( + proposal_id: str, payload: Optional[Dict[str, Any]] = None +) -> Any: + """Reject a pending proposal. Payload may carry + ``{review_notes: str}`` — operator rationale recorded verbatim + in the audit row (operator-decision-relevant per the #182 + precedent). + """ + from kora_cli.audit import emit_audit + from kora_cli.promote.phrasebook.proposer import proposal_to_dict + from kora_cli.promote.phrasebook.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: + existing = load(proposal_id) + except ProposalNotFound: + return JSONResponse( + status_code=404, + content={"error": "proposal_not_found", "proposal_id": proposal_id}, + ) + if existing.status != "pending": + return JSONResponse( + status_code=409, + content={ + "error": "proposal_not_pending", + "proposal_id": proposal_id, + "current_status": existing.status, + }, + ) + + updated = transition( + proposal_id, new_status="rejected", review_notes=review_notes + ) + + try: + emit_audit( + seam="promotion.rejected", + details=proposal_to_dict(updated), + caller_session_id=f"promotion:phrasebook:{proposal_id}", + source="reasoning", + ) + except Exception as exc: + logger.warning( + "[kora.promote] promotion.rejected emit raised %r — " + "rejection persisted; audit row missing", + exc, + ) + + return { + "proposal_id": proposal_id, + "status": "rejected", + "review_notes": review_notes, + } + + # --------------------------------------------------------------------------- # Email-intent audit lens (KR-FE-EMAIL-INTENT-LOG-PANEL) # --------------------------------------------------------------------------- diff --git a/tests/kora_cli/clustering/__init__.py b/tests/kora_cli/clustering/__init__.py new file mode 100644 index 000000000000..e69de29bb2d1 diff --git a/tests/kora_cli/clustering/test_text_similarity.py b/tests/kora_cli/clustering/test_text_similarity.py new file mode 100644 index 000000000000..05379a589720 --- /dev/null +++ b/tests/kora_cli/clustering/test_text_similarity.py @@ -0,0 +1,133 @@ +"""Tests for kora_cli.clustering.text_similarity.""" + +from __future__ import annotations + +import pytest + +from kora_cli.clustering.text_similarity import ( + TextEmbedding, + cluster_by_similarity, + cosine_similarity, + embed_texts, +) + + +@pytest.fixture(autouse=True) +def _cache(tmp_path, monkeypatch): + """Per-test cache dir under tmp_path so cache assertions don't + bleed across tests.""" + monkeypatch.setenv("KORA_HOME", str(tmp_path)) + monkeypatch.setattr( + "kora_constants.get_kora_home", lambda: tmp_path, raising=False + ) + return tmp_path + + +@pytest.mark.asyncio +async def test_embed_texts_returns_one_embedding_per_input(): + texts = ["what's the burn?", "are you paused?", "any alerts?"] + out = await embed_texts(texts) + assert len(out) == 3 + for emb, text in zip(out, texts): + assert emb.text == text + assert isinstance(emb.embedding, dict) + assert emb.embedding # non-empty + assert emb.cached is False # first call → cold cache + + +@pytest.mark.asyncio +async def test_embed_texts_cache_hit_on_second_call(): + out1 = await embed_texts(["burn?"]) + out2 = await embed_texts(["burn?"]) + assert out1[0].cached is False + assert out2[0].cached is True + # Embeddings byte-identical (the cached dict is what's stored). + assert out1[0].embedding == out2[0].embedding + + +@pytest.mark.asyncio +async def test_embed_texts_yields_features_for_short_text(): + out = await embed_texts(["burn?"]) + feats = out[0].embedding + # Token feature for "burn" present. + assert "TOK:burn" in feats + # Char-ngram features present (3-grams over " burn? "). + assert any(k.startswith("CHR:") for k in feats) + + +@pytest.mark.asyncio +async def test_embed_texts_normalizes_apostrophes(): + out1 = await embed_texts(["what's"]) + out2 = await embed_texts(["whats"]) + assert "TOK:whats" in out1[0].embedding + assert "TOK:whats" in out2[0].embedding + + +@pytest.mark.asyncio +async def test_cosine_identical_texts_is_one(): + out = await embed_texts(["burn rate today?"]) + same_emb = out[0] + assert cosine_similarity(same_emb, same_emb) == pytest.approx(1.0) + + +@pytest.mark.asyncio +async def test_cosine_unrelated_texts_below_threshold(): + out = await embed_texts( + ["what's the burn?", "elephants migrate in winter"] + ) + assert cosine_similarity(out[0], out[1]) < 0.3 + + +@pytest.mark.asyncio +async def test_cosine_paraphrase_above_threshold(): + out = await embed_texts( + [ + "what's the burn rate today?", + "what's the burn today?", + ] + ) + assert cosine_similarity(out[0], out[1]) >= 0.5 + + +@pytest.mark.asyncio +async def test_cosine_with_empty_embedding_is_zero(): + real = (await embed_texts(["x"]))[0] + empty = TextEmbedding(text="", embedding={}, cached=False) + assert cosine_similarity(real, empty) == 0.0 + + +@pytest.mark.asyncio +async def test_cluster_groups_similar_texts(): + texts = [ + "what's the burn rate today?", + "burn rate?", + "burn rate now?", + "completely unrelated about elephants", + ] + out = await embed_texts(texts) + clusters = cluster_by_similarity(out, threshold=0.4) + # 3 burn variants should cluster; the elephants line stands + # alone. + sizes = sorted(len(c) for c in clusters) + assert sizes == [1, 3] + + +@pytest.mark.asyncio +async def test_cluster_threshold_too_strict_yields_singletons(): + texts = ["burn?", "alerts?", "state?"] + out = await embed_texts(texts) + clusters = cluster_by_similarity(out, threshold=0.99) + assert all(len(c) == 1 for c in clusters) + + +@pytest.mark.asyncio +async def test_cluster_preserves_input_order(): + """Determinism: same input list → same partition. Verified + indirectly: clusters' first members appear in input order.""" + texts = [f"burn variant {i}" for i in range(8)] + out = await embed_texts(texts) + clusters = cluster_by_similarity(out, threshold=0.6) + # Whichever cluster has the first text, its first element is + # text 0. + first_cluster = clusters[0] + assert first_cluster[0].text == texts[0] diff --git a/tests/kora_cli/promote/__init__.py b/tests/kora_cli/promote/__init__.py new file mode 100644 index 000000000000..e69de29bb2d1 diff --git a/tests/kora_cli/promote/phrasebook/__init__.py b/tests/kora_cli/promote/phrasebook/__init__.py new file mode 100644 index 000000000000..e69de29bb2d1 diff --git a/tests/kora_cli/promote/phrasebook/test_endpoints.py b/tests/kora_cli/promote/phrasebook/test_endpoints.py new file mode 100644 index 000000000000..131b1dcfe9e0 --- /dev/null +++ b/tests/kora_cli/promote/phrasebook/test_endpoints.py @@ -0,0 +1,201 @@ +"""Tests for the 3 phrasebook-promotion endpoints in web_server.""" + +from __future__ import annotations + +import json +from datetime import datetime, timezone + +import pytest +from fastapi.testclient import TestClient + +from kora_cli.promote.phrasebook.proposer import PromotionProposal +from kora_cli.promote.phrasebook.store import ( + PROMOTIONS_ROOT_ENV, + save_pending, +) +from kora_cli.web_server import app + + +@pytest.fixture(autouse=True) +def _isolate(tmp_path, monkeypatch): + monkeypatch.setenv("KORA_HOME", str(tmp_path)) + monkeypatch.setattr( + "kora_constants.get_kora_home", lambda: tmp_path, raising=False + ) + monkeypatch.setenv(PROMOTIONS_ROOT_ENV, str(tmp_path / "promotions")) + monkeypatch.setenv( + "KORA_AUDIT_LOG_PATH", str(tmp_path / "audit.jsonl") + ) + # Phrasebook editor reads ${KORA_HOME}/phrasebook/slack_dm.yml + # via get_kora_home() — the KORA_HOME monkeypatch above is + # sufficient. No separate env override needed. + return tmp_path + + +def _save_proposal(proposal_id: str = "p1", confidence: float = 0.85): + p = PromotionProposal( + proposal_id=proposal_id, + cluster_size=5, + sample_questions=["q1", "q2"], + proposed_pattern="(?i)(burn|cost|budget)", + proposed_reply_template="Burn is $42 today.", + proposed_category="cost_query", + confidence=confidence, + created_at=datetime(2026, 5, 24, 12, 0, 0, tzinfo=timezone.utc), + status="pending", + review_notes="", + ) + return save_pending(p) + + +def _client(): + """Authenticated TestClient — the cockpit endpoints sit behind + the dashboard session-header auth middleware, same as every + other /api/* surface.""" + from kora_cli.web_server import _SESSION_HEADER_NAME, _SESSION_TOKEN + + client = TestClient(app) + client.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN + return client + + +def test_list_pending_returns_proposals_highest_confidence_first(tmp_path): + _save_proposal(proposal_id="low", confidence=0.4) + _save_proposal(proposal_id="high", confidence=0.9) + _save_proposal(proposal_id="mid", confidence=0.6) + + resp = _client().get("/api/promotions/phrasebook/pending") + assert resp.status_code == 200 + body = resp.json() + assert [p["proposal_id"] for p in body["proposals"]] == [ + "high", + "mid", + "low", + ] + assert body["status_values"] == [ + "pending", + "approved", + "rejected", + "expired", + ] + + +def test_list_pending_empty_returns_empty_list(): + resp = _client().get("/api/promotions/phrasebook/pending") + assert resp.status_code == 200 + assert resp.json()["proposals"] == [] + + +def test_approve_404_when_proposal_missing(): + resp = _client().post( + "/api/promotions/phrasebook/never-existed/approve", + json={}, + ) + assert resp.status_code == 404 + assert resp.json()["error"] == "proposal_not_found" + + +def test_approve_409_when_proposal_not_pending(tmp_path): + _save_proposal() + from kora_cli.promote.phrasebook.store import transition + + transition("p1", new_status="rejected", review_notes="no") + resp = _client().post( + "/api/promotions/phrasebook/p1/approve", json={} + ) + assert resp.status_code == 409 + assert resp.json()["error"] == "proposal_not_pending" + + +def test_approve_happy_path_persists_phrasebook_and_emits_audit(tmp_path): + _save_proposal() + resp = _client().post( + "/api/promotions/phrasebook/p1/approve", + json={"review_notes": "good catch"}, + ) + assert resp.status_code == 200 + body = resp.json() + assert body["status"] == "approved" + assert body["committed_entry"]["pattern"] == "(?i)(burn|cost|budget)" + + # Audit emitted: promotion.approved + phrasebook.updated. + rows = [ + json.loads(line) + for line in (tmp_path / "audit.jsonl").read_text().splitlines() + if line + ] + seams = [r["seam"] for r in rows] + assert "promotion.approved" in seams + assert "phrasebook.updated" in seams + # phrasebook.updated uses the forward-compat actor literal. + pb_rows = [r for r in rows if r["seam"] == "phrasebook.updated"] + assert pb_rows[0]["details"]["actor"] == "kora_proposal_approved" + assert pb_rows[0]["details"]["proposal_id"] == "p1" + + +def test_approve_with_overrides_applies_them(tmp_path): + _save_proposal() + resp = _client().post( + "/api/promotions/phrasebook/p1/approve", + json={ + "pattern_override": "(?i)\\bspend\\b", + "reply_template_override": "Spend is $42.", + "category_override": "spending", + }, + ) + assert resp.status_code == 200 + body = resp.json() + assert body["committed_entry"]["pattern"] == "(?i)\\bspend\\b" + assert body["committed_entry"]["reply_template"] == "Spend is $42." + assert body["committed_entry"]["category"] == "spending" + + +def test_approve_validation_failure_returns_422(tmp_path): + """Malformed regex from override → editor rejects → 422.""" + _save_proposal() + resp = _client().post( + "/api/promotions/phrasebook/p1/approve", + json={"pattern_override": "[unterminated"}, + ) + assert resp.status_code == 422 + body = resp.json() + assert body["error"] == "validation_failed" + + +def test_reject_404_when_proposal_missing(): + resp = _client().post( + "/api/promotions/phrasebook/x/reject", + json={"review_notes": "no"}, + ) + assert resp.status_code == 404 + + +def test_reject_409_when_proposal_not_pending(tmp_path): + _save_proposal() + from kora_cli.promote.phrasebook.store import transition + + transition("p1", new_status="approved", review_notes="") + resp = _client().post( + "/api/promotions/phrasebook/p1/reject", + json={"review_notes": "actually no"}, + ) + assert resp.status_code == 409 + + +def test_reject_happy_path_moves_proposal_and_emits_audit(tmp_path): + _save_proposal() + resp = _client().post( + "/api/promotions/phrasebook/p1/reject", + json={"review_notes": "category is wrong"}, + ) + assert resp.status_code == 200 + assert resp.json()["status"] == "rejected" + + rows = [ + json.loads(line) + for line in (tmp_path / "audit.jsonl").read_text().splitlines() + if line + ] + rejected_rows = [r for r in rows if r["seam"] == "promotion.rejected"] + assert len(rejected_rows) == 1 + assert rejected_rows[0]["details"]["review_notes"] == "category is wrong" diff --git a/tests/kora_cli/promote/phrasebook/test_observer.py b/tests/kora_cli/promote/phrasebook/test_observer.py new file mode 100644 index 000000000000..d72f34fc77a0 --- /dev/null +++ b/tests/kora_cli/promote/phrasebook/test_observer.py @@ -0,0 +1,182 @@ +"""Tests for kora_cli.promote.phrasebook.observer.""" + +from __future__ import annotations + +import json +from datetime import datetime, timedelta, timezone +from pathlib import Path + +import pytest + +from kora_cli.promote.phrasebook.observer import ( + collect_recent_observations, +) + + +@pytest.fixture(autouse=True) +def _isolate(tmp_path, monkeypatch): + monkeypatch.setenv("KORA_HOME", str(tmp_path)) + monkeypatch.setenv( + "KORA_SLACK_DM_LOG_PATH", str(tmp_path / "slack_dm_log.jsonl") + ) + return tmp_path + + +def _write_log(tmp_path: Path, entries: list) -> None: + path = tmp_path / "slack_dm_log.jsonl" + path.write_text( + "\n".join(json.dumps(e) for e in entries) + "\n", encoding="utf-8" + ) + + +def _entry( + *, + text: str = "Burn is $42 today.", + model_used: str = "claude-haiku-4-5-20251001", + send_status: str = "ok", + sent_at: datetime = None, + caller_session_id: str = "D1JOSH:1700000000.1", + input_tokens: int = 1000, + output_tokens: int = 50, +) -> dict: + if sent_at is None: + sent_at = datetime.now(timezone.utc) - timedelta(hours=1) + return { + "sent_at": sent_at.isoformat(), + "channel_id": "D1JOSH", + "thread_ts": None, + "text": text, + "slack_message_ts": "1.0", + "send_status": send_status, + "model_used": model_used, + "input_tokens": input_tokens, + "output_tokens": output_tokens, + "caller_session_id": caller_session_id, + } + + +@pytest.mark.asyncio +async def test_collect_returns_engine_driven_entries(tmp_path): + _write_log(tmp_path, [_entry(text="Burn is $42.")]) + out = await collect_recent_observations( + since=datetime.now(timezone.utc) - timedelta(hours=24), + ) + assert len(out) == 1 + assert out[0].kora_response == "Burn is $42." + assert out[0].route == "slack_dm" + + +@pytest.mark.asyncio +async def test_collect_excludes_short_circuit_hits(tmp_path): + _write_log( + tmp_path, + [ + _entry(text="Yes you're paused."), + _entry(text="Burn $42", model_used="short_circuit"), + ], + ) + out = await collect_recent_observations( + since=datetime.now(timezone.utc) - timedelta(hours=24), + ) + assert len(out) == 1 + assert out[0].kora_response == "Yes you're paused." + + +@pytest.mark.asyncio +async def test_collect_excludes_non_engine_paths(tmp_path): + """Entries without model_used are canned-fallback / non- + reasoning → drop them, they don't carry the Q+A signal.""" + entry = _entry() + entry.pop("model_used") + _write_log(tmp_path, [entry]) + out = await collect_recent_observations( + since=datetime.now(timezone.utc) - timedelta(hours=24), + ) + assert out == [] + + +@pytest.mark.asyncio +async def test_collect_excludes_failed_sends(tmp_path): + _write_log(tmp_path, [_entry(send_status="failed")]) + out = await collect_recent_observations( + since=datetime.now(timezone.utc) - timedelta(hours=24), + ) + assert out == [] + + +@pytest.mark.asyncio +async def test_collect_time_window_excludes_old_entries(tmp_path): + old = datetime.now(timezone.utc) - timedelta(days=10) + fresh = datetime.now(timezone.utc) - timedelta(hours=2) + _write_log( + tmp_path, + [_entry(text="old", sent_at=old), _entry(text="fresh", sent_at=fresh)], + ) + out = await collect_recent_observations( + since=datetime.now(timezone.utc) - timedelta(days=1), + ) + assert [o.kora_response for o in out] == ["fresh"] + + +@pytest.mark.asyncio +async def test_collect_route_filter_drops_other_routes(tmp_path): + _write_log( + tmp_path, + [ + _entry( + text="probe answer", + caller_session_id="probe:fly:service_unhealthy", + ), + _entry(text="slack answer"), + ], + ) + out = await collect_recent_observations( + since=datetime.now(timezone.utc) - timedelta(hours=24), + route_filter=["slack_dm"], + ) + assert [o.route for o in out] == ["slack_dm"] + + +@pytest.mark.asyncio +async def test_collect_tolerates_malformed_lines(tmp_path): + log_path = tmp_path / "slack_dm_log.jsonl" + log_path.write_text( + json.dumps(_entry(text="valid")) + + "\nNOT JSON\n{\"sent_at\":\"bad-ts\",\"model_used\":\"x\"," + "\"send_status\":\"ok\",\"text\":\"t\"}\n" + + json.dumps(_entry(text="valid2")) + + "\n", + encoding="utf-8", + ) + out = await collect_recent_observations( + since=datetime.now(timezone.utc) - timedelta(hours=24), + ) + assert sorted(o.kora_response for o in out) == ["valid", "valid2"] + + +@pytest.mark.asyncio +async def test_collect_missing_log_returns_empty(tmp_path): + """No file → no observations, no exception.""" + out = await collect_recent_observations( + since=datetime.now(timezone.utc) - timedelta(hours=24), + ) + assert out == [] + + +@pytest.mark.asyncio +async def test_collect_sorts_by_timestamp_ascending(tmp_path): + t1 = datetime.now(timezone.utc) - timedelta(hours=5) + t2 = datetime.now(timezone.utc) - timedelta(hours=3) + t3 = datetime.now(timezone.utc) - timedelta(hours=1) + _write_log( + tmp_path, + [ + _entry(text="3rd", sent_at=t3), + _entry(text="1st", sent_at=t1), + _entry(text="2nd", sent_at=t2), + ], + ) + out = await collect_recent_observations( + since=datetime.now(timezone.utc) - timedelta(hours=24), + ) + assert [o.kora_response for o in out] == ["1st", "2nd", "3rd"] diff --git a/tests/kora_cli/promote/phrasebook/test_proposer.py b/tests/kora_cli/promote/phrasebook/test_proposer.py new file mode 100644 index 000000000000..b8997df957c4 --- /dev/null +++ b/tests/kora_cli/promote/phrasebook/test_proposer.py @@ -0,0 +1,221 @@ +"""Tests for kora_cli.promote.phrasebook.proposer.""" + +from __future__ import annotations + +from datetime import datetime, timezone + +import pytest + +from kora_cli.promote.phrasebook.observer import ReasoningObservation +from kora_cli.promote.phrasebook.proposer import ( + DEFAULT_MIN_CLUSTER_SIZE, + PromotionProposal, + generate_proposals, + proposal_from_dict, + proposal_to_dict, +) + + +@pytest.fixture(autouse=True) +def _isolate(tmp_path, monkeypatch): + monkeypatch.setenv("KORA_HOME", str(tmp_path)) + monkeypatch.setattr( + "kora_constants.get_kora_home", lambda: tmp_path, raising=False + ) + return tmp_path + + +def _obs(text: str, idx: int) -> ReasoningObservation: + return ReasoningObservation( + operator_question="", + kora_response=text, + timestamp=datetime(2026, 5, 24, 0, 0, idx, tzinfo=timezone.utc), + caller_session_id=f"D1JOSH:170000000{idx}.1", + cost_usd=0.001, + model_used="claude-haiku-4-5-20251001", + route="slack_dm", + ) + + +# =========================================================================== +# Synthetic clusters → proposal +# =========================================================================== + + +@pytest.mark.asyncio +async def test_consistent_cluster_yields_proposal(): + """5 near-identical answers about burn → one proposal.""" + observations = [ + _obs("Burn is $42 today; 75% of budget used.", i) + for i in range(5) + ] + proposals, cost = await generate_proposals(observations) + assert len(proposals) == 1 + assert cost == 0.0 # no Haiku synthesis by default + p = proposals[0] + assert p.cluster_size == 5 + assert p.proposed_pattern.startswith("(?i)(") + # Pattern is an alternation over the top tokens — verify it + # compiles + matches the source text (without pinning which + # specific tokens ranked top-3, since identical-document + # df-ties are insertion-order dependent). + import re as _re + + compiled = _re.compile(p.proposed_pattern) + assert compiled.search("Burn is $42 today; 75% of budget used.") + assert p.proposed_reply_template == "Burn is $42 today; 75% of budget used." + assert p.haiku_synthesized is False + assert p.status == "pending" + + +@pytest.mark.asyncio +async def test_small_cluster_not_proposed(): + """Below min_cluster_size → no proposal.""" + observations = [_obs("Burn is $42", i) for i in range(3)] + proposals, _ = await generate_proposals(observations) + assert proposals == [] + + +@pytest.mark.asyncio +async def test_divergent_answers_cluster_rejected(): + """5 messages about burn but each gives a wildly different + answer → answer-consistency check rejects the cluster.""" + observations = [ + _obs("Burn is $42 today; 75% of budget used.", 1), + _obs("Burn rate fluctuates wildly across the month.", 2), + _obs("Burn? not sure, check the cost panel directly.", 3), + _obs("Burn is high; reduce Opus usage if you can.", 4), + _obs("Burn metric is fine; nothing to worry about.", 5), + ] + proposals, _ = await generate_proposals( + observations, cohesion_threshold=0.2 + ) + # With a loose cohesion threshold the cluster forms, but + # answer-consistency check below 0.75 should reject. With a + # tighter cohesion threshold the cluster doesn't form at + # all. Either way: no proposal. + assert proposals == [] + + +@pytest.mark.asyncio +async def test_two_distinct_clusters_yield_two_proposals(): + burn = [ + _obs("Burn is $42 today; 75% of budget used.", i) for i in range(5) + ] + paused = [ + # State-query phrasing — avoids "cost"/"burn" tokens so + # the categorizer reaches the state keywords first. + _obs("Yes, you're paused; primary_state is PAUSED.", 10 + i) + for i in range(5) + ] + proposals, _ = await generate_proposals(burn + paused) + assert len(proposals) == 2 + cats = sorted(p.proposed_category for p in proposals) + # Burn and state-query keywords drive the categories. + assert "cost_query" in cats + assert "state_query" in cats + + +@pytest.mark.asyncio +async def test_empty_observations_yields_empty(): + proposals, cost = await generate_proposals([]) + assert proposals == [] + assert cost == 0.0 + + +@pytest.mark.asyncio +async def test_proposals_sorted_highest_confidence_first(): + """5 perfect-cohesion observations + 6 looser ones should + rank perfect-cohesion higher.""" + perfect = [_obs("Burn is $42.", i) for i in range(5)] + looser = [ + _obs("Yes, primary_state PAUSED at 09:00 UTC.", i + 10) + for i in range(6) + ] + proposals, _ = await generate_proposals( + perfect + looser, cohesion_threshold=0.5 + ) + assert len(proposals) == 2 + # Confidences non-decreasing; first should be the perfect + # cluster. + assert proposals[0].confidence >= proposals[1].confidence + + +# =========================================================================== +# Haiku synthesis (mocked) +# =========================================================================== + + +@pytest.mark.asyncio +async def test_synthesizer_invoked_one_cost_recorded(): + observations = [_obs("Burn is $42 today.", i) for i in range(5)] + calls = [] + + async def fake_synth(obs_list): + calls.append(len(obs_list)) + return ("Synthesized template body", 0.0012) + + proposals, total_cost = await generate_proposals( + observations, synthesize_reply_template=fake_synth + ) + assert len(proposals) == 1 + assert proposals[0].haiku_synthesized is True + assert proposals[0].proposed_reply_template == ( + "Synthesized template body" + ) + assert calls == [5] + assert total_cost == pytest.approx(0.0012) + + +# =========================================================================== +# Serialization +# =========================================================================== + + +def test_proposal_round_trip_to_dict_and_back(): + p = PromotionProposal( + proposal_id="abc-123", + cluster_size=7, + sample_questions=["q1", "q2", "q3"], + proposed_pattern="(?i)(burn|cost)", + proposed_reply_template="Burn is $42.", + proposed_category="cost_query", + confidence=0.92, + created_at=datetime(2026, 5, 24, 12, 0, 0, tzinfo=timezone.utc), + status="pending", + review_notes="", + cluster_caller_session_ids=["D1:1", "D1:2"], + haiku_synthesized=False, + ) + out = proposal_from_dict(proposal_to_dict(p)) + assert out.proposal_id == p.proposal_id + assert out.cluster_size == p.cluster_size + assert out.proposed_pattern == p.proposed_pattern + assert out.proposed_reply_template == p.proposed_reply_template + assert out.proposed_category == p.proposed_category + assert out.confidence == p.confidence + assert out.status == p.status + assert out.created_at == p.created_at + assert out.cluster_caller_session_ids == p.cluster_caller_session_ids + assert out.haiku_synthesized == p.haiku_synthesized + + +def test_proposal_from_dict_handles_z_suffix(): + p = proposal_from_dict( + { + "proposal_id": "x", + "cluster_size": 5, + "sample_questions": [], + "proposed_pattern": "(?i)(burn)", + "proposed_reply_template": "x", + "proposed_category": "x", + "confidence": 0.5, + "created_at": "2026-05-24T12:00:00Z", + "status": "pending", + } + ) + assert p.created_at.tzinfo is not None + + +def test_default_min_cluster_size_is_5(): + assert DEFAULT_MIN_CLUSTER_SIZE == 5 diff --git a/tests/kora_cli/promote/phrasebook/test_store_and_cycle.py b/tests/kora_cli/promote/phrasebook/test_store_and_cycle.py new file mode 100644 index 000000000000..f92f186dc222 --- /dev/null +++ b/tests/kora_cli/promote/phrasebook/test_store_and_cycle.py @@ -0,0 +1,229 @@ +"""Tests for the promotion store + cycle orchestrator.""" + +from __future__ import annotations + +import json +from datetime import datetime, timedelta, timezone + +import pytest + +from kora_cli.promote.phrasebook.proposer import PromotionProposal +from kora_cli.promote.phrasebook.store import ( + PROMOTIONS_ROOT_ENV, + ProposalNotFound, + expire_older_than, + list_pending, + load, + save_pending, + transition, +) + + +@pytest.fixture(autouse=True) +def _isolate(tmp_path, monkeypatch): + monkeypatch.setenv("KORA_HOME", str(tmp_path)) + monkeypatch.setattr( + "kora_constants.get_kora_home", lambda: tmp_path, raising=False + ) + monkeypatch.setenv(PROMOTIONS_ROOT_ENV, str(tmp_path / "promotions")) + monkeypatch.setenv( + "KORA_AUDIT_LOG_PATH", str(tmp_path / "audit.jsonl") + ) + monkeypatch.setenv( + "KORA_SLACK_DM_LOG_PATH", str(tmp_path / "slack_dm_log.jsonl") + ) + return tmp_path + + +def _mk(status="pending", proposal_id="p1", confidence=0.5) -> PromotionProposal: + return PromotionProposal( + proposal_id=proposal_id, + cluster_size=5, + sample_questions=["q1", "q2"], + proposed_pattern="(?i)(burn)", + proposed_reply_template="Burn is $42.", + proposed_category="cost_query", + confidence=confidence, + created_at=datetime(2026, 5, 24, 12, 0, 0, tzinfo=timezone.utc), + status=status, + review_notes="", + ) + + +def test_save_pending_writes_json_under_pending_dir(tmp_path): + p = _mk() + written = save_pending(p) + assert written.is_file() + assert written.parent.name == "pending" + payload = json.loads(written.read_text(encoding="utf-8")) + assert payload["proposal_id"] == "p1" + + +def test_save_pending_refuses_non_pending_status(): + p = _mk(status="approved") + with pytest.raises(ValueError): + save_pending(p) + + +def test_list_pending_orders_highest_confidence_first(): + save_pending(_mk(proposal_id="p_low", confidence=0.2)) + save_pending(_mk(proposal_id="p_high", confidence=0.9)) + save_pending(_mk(proposal_id="p_mid", confidence=0.5)) + out = list_pending() + assert [p.proposal_id for p in out] == ["p_high", "p_mid", "p_low"] + + +def test_transition_pending_to_approved_moves_file(tmp_path): + save_pending(_mk()) + updated = transition( + "p1", new_status="approved", review_notes="LGTM" + ) + assert updated.status == "approved" + assert updated.review_notes == "LGTM" + assert not (tmp_path / "promotions" / "phrasebook" / "pending" / "p1.json").exists() + assert ( + tmp_path / "promotions" / "phrasebook" / "approved" / "p1.json" + ).is_file() + + +def test_transition_with_overrides_applies_them(): + save_pending(_mk()) + updated = transition( + "p1", + new_status="approved", + review_notes="", + overrides={ + "pattern": "(?i)(burn|spend|cost)", + "reply_template": "Custom template", + "category": "custom", + }, + ) + assert updated.proposed_pattern == "(?i)(burn|spend|cost)" + assert updated.proposed_reply_template == "Custom template" + assert updated.proposed_category == "custom" + + +def test_load_returns_proposal_across_status_dirs(): + save_pending(_mk(proposal_id="p_findme")) + out = load("p_findme") + assert out.proposal_id == "p_findme" + transition("p_findme", new_status="rejected", review_notes="no") + out2 = load("p_findme") + assert out2.status == "rejected" + + +def test_load_missing_raises(): + with pytest.raises(ProposalNotFound): + load("never-existed") + + +def test_transition_missing_raises(): + with pytest.raises(ProposalNotFound): + transition("never-existed", new_status="approved") + + +def test_expire_older_than_moves_old_pending(tmp_path): + old = PromotionProposal( + proposal_id="old_p", + cluster_size=5, + sample_questions=[], + proposed_pattern="(?i)(x)", + proposed_reply_template="x", + proposed_category="x", + confidence=0.5, + created_at=datetime.now(timezone.utc) - timedelta(days=30), + status="pending", + ) + fresh = _mk(proposal_id="fresh_p") + save_pending(old) + save_pending(fresh) + moved = expire_older_than(days=14) + assert moved == 1 + pending_after = list_pending() + assert [p.proposal_id for p in pending_after] == ["fresh_p"] + + +# =========================================================================== +# Cycle integration +# =========================================================================== + + +@pytest.mark.asyncio +async def test_cycle_with_no_observations_returns_summary(tmp_path): + from kora_cli.promote.phrasebook.cycle import ( + run_phrasebook_promotion_cycle, + ) + + summary = await run_phrasebook_promotion_cycle() + assert summary["enabled"] is True + assert summary["observations_read"] == 0 + assert summary["proposals_generated"] == 0 + assert summary["proposals_persisted"] == 0 + + +@pytest.mark.asyncio +async def test_cycle_disabled_skips_cleanly(tmp_path, monkeypatch): + from kora_cli.promote.phrasebook.cycle import ( + ENABLED_ENV, + run_phrasebook_promotion_cycle, + ) + + monkeypatch.setenv(ENABLED_ENV, "false") + summary = await run_phrasebook_promotion_cycle() + assert summary["enabled"] is False + assert summary["observations_read"] == 0 + + +@pytest.mark.asyncio +async def test_cycle_with_synthetic_observations_emits_audit(tmp_path): + """End-to-end: synthetic slack_dm_log → cycle → proposals + persisted + audit row per proposal.""" + log_path = tmp_path / "slack_dm_log.jsonl" + entries = [] + base = datetime.now(timezone.utc) - timedelta(hours=2) + for i in range(6): + entries.append( + { + "sent_at": (base + timedelta(minutes=i)).isoformat(), + "channel_id": "D1JOSH", + "thread_ts": None, + "text": "Burn is $42 today; 75% of budget used.", + "slack_message_ts": f"1.{i}", + "send_status": "ok", + "model_used": "claude-haiku-4-5-20251001", + "input_tokens": 1000, + "output_tokens": 30, + "caller_session_id": f"D1JOSH:170000000{i}.1", + } + ) + log_path.write_text( + "\n".join(json.dumps(e) for e in entries) + "\n", + encoding="utf-8", + ) + + from kora_cli.promote.phrasebook.cycle import ( + run_phrasebook_promotion_cycle, + ) + + summary = await run_phrasebook_promotion_cycle() + assert summary["observations_read"] == 6 + assert summary["proposals_generated"] == 1 + assert summary["proposals_persisted"] == 1 + + pending = list_pending() + assert len(pending) == 1 + assert pending[0].cluster_size == 6 + + # Audit row emitted. + audit_path = tmp_path / "audit.jsonl" + rows = [ + json.loads(line) + for line in audit_path.read_text().splitlines() + if line + ] + proposed_rows = [ + r for r in rows if r["seam"] == "promotion.proposed" + ] + assert len(proposed_rows) == 1 + assert proposed_rows[0]["details"]["cluster_size"] == 6 + assert "synth_cost_usd" in proposed_rows[0]["details"]