scripts.mine_prompt_history

Reference › scripts › mine_prompt_history

scripts/mine_prompt_history.py is the quarterly prompt-history drift miner from plans/workflow-improvements-plan.md. It looks for instructions that users keep repeating in prompts and grades each recurring cluster against the existing rulings, so repeated corrections that no ruling covers surface as drift. A human runs it; tests/scripts/test_mine_prompt_history.py covers it. It uses orchestrata_ultimate.core.graph_factory and orchestrata_ultimate.knowledge_graph.canonical_schema.

Sources are ~/.kimi/user-history/*.jsonl (read_kimi_history), ~/.claude/history.jsonl (read_claude_history) and user-role ConversationTurn nodes in the graph (query_kg_user_turns, skipped with --no-kg, limited by --kg-limit). The work is three passes: filter_marker_prompts keeps prompts that hit a marker word, cluster_by_signature groups them by matched-marker signature, and grade_clusters ranks clusters by recurrence and grades each against the rulings loaded by load_rulings (parse_rulings_index, match_ruling, with a token overlap threshold from --min-overlap). Other options are --out, --rulings-dir, --kimi-dir and --claude-history. It writes a markdown report, by default reports/ruling-drift-<date>.md, and prints the prompt and cluster counts. Without the rulings canon, load_rulings returns an empty list.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

MinedPrompt (dataclass)

Decorators: dataclass(frozen=True) · Line 114

One marker-matching (or candidate) prompt from any source.

RulingRow (dataclass)

Decorators: dataclass(frozen=True) · Line 124

One parsed row of rulings/index.md.

ClusterGrade (dataclass)

Decorators: dataclass · Line 135

MiningResult (dataclass)

Decorators: dataclass · Line 147

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/mine_prompt_history.py

#!/usr/bin/env python3
"""Quarterly prompt-history drift miner (``plans/workflow-improvements-plan.md``
Phase 5, Step 5.1). Corrections repeat across sessions -- the same rulings
get re-issued verbally for months before they're mechanically enforced. This
script catches that pattern early: it mines conversation history for
ruling-shaped prompts, clusters them, and diffs the clusters against the
ratified rulings canon (``rulings/index.md``) to name what's already
covered and what is genuinely new.

**Read-only.** This script never writes a ruling, never touches the wiki,
never touches Neo4j except to read. Its only output is a markdown report,
``reports/ruling-drift-YYYY-MM-DD.md``. Filing a new ruling for an
UNTRACKED cluster is a human judgment call for whoever reads the report.

Inputs, in priority order:

1. **Primary: the KG's SPEC-021 conversation capture.** The Claude producer
   is deployed user-level and kimi/IRC has its own producer, so the
   operational KG already holds structured, multi-platform turns
   (``ConversationTurn`` nodes, Tier I --
   ``knowledge_graph/canonical_schema.py``). Mining it covers every agent
   tool uniformly without re-parsing each tool's raw format. Connects via
   the ``neo4j`` driver directly (not ``core.graph``/``knowledge_graph``
   client wrappers -- this script's only Neo4j dependency, by design, to
   stay clear of concurrently-owned client code).
2. **Fallback, pre-capture history only:** ``~/.kimi/user-history/*.jsonl``
   and ``~/.claude/history.jsonl``. These carry prompts only -- no
   correction/response context -- so findings from them are graded the same
   as KG findings but are inherently less certain evidence.

Pipeline (Passes 1-3 of the plan):

1. **Marker filter** -- case-insensitive substring match against a fixed
   set of ruling-shaped phrases (``never``, ``forbidden``, ``banned``,
   ``i commanded``, ``i want``, ``do not``, ``no mock``, ``no xdist``,
   ``hardcod``).
2. **Cluster** -- group marker-matched prompts by which markers matched
   (a normalized keyword signature), ranked by recurrence count.
3. **Diff vs. canon** -- for each cluster, token-overlap match against every
   ratified ruling's text; graded ``KNOWN+ENFORCED``, ``KNOWN+JUDGMENT-ONLY``,
   or ``UNTRACKED`` (candidate new ruling, reported with verbatim examples).

CLI:
    python scripts/mine_prompt_history.py
    python scripts/mine_prompt_history.py --no-kg --out /tmp/drift.md
"""

from __future__ import annotations

import argparse
import json
import os
import re
import sys
from collections.abc import Iterable, Iterator
from dataclasses import dataclass, field
from datetime import UTC, datetime
from pathlib import Path

from orchestrata_ultimate.core.graph_factory import resolve_kg_database, resolve_uri

REPO_ROOT = Path(__file__).resolve().parent.parent

#: Pass 1 markers (verbatim from the plan -- do not add/remove without
#: updating plans/workflow-improvements-plan.md Phase 5 Step 5.1).
MARKERS: tuple[str, ...] = (
    "never",
    "forbidden",
    "banned",
    "i commanded",
    "i want",
    "do not",
    "no mock",
    "no xdist",
    "hardcod",
)

#: A cluster must share at least this many significant tokens with a
#: ruling's own text to be graded KNOWN rather than UNTRACKED.
DEFAULT_MIN_OVERLAP = 2

#: Small curated stopword list -- just enough to keep token-overlap
#: matching from being swamped by connective English, not a general NLP
#: pipeline. Deliberately EXCLUDES "no", "not", "never", "don" -- this is a
#: corpus of prohibition-shaped rulings (RUL-001 is literally "NEVER run
#: pytest with xdist"), so negation words are the single most discriminative
#: signal, not noise; stripping them as generic-English filler silently
#: collapses ruling-matching accuracy.
_STOPWORDS = frozenset("""
    the a an and or but if then else for to of in on at by with from as is
    are was were be been being this that these those it its you your i me
    my we our do does did always should would could can will
    have has had into about over under again further once here there when
    where why how all any both each few more most other some such nor
    only own same so than too very just should now project user
    """.split())

_DEFAULT_RULINGS_DIR = str(Path.home() / "workspace" / "knowledge-wiki" / "rulings")
_DEFAULT_KIMI_DIR = Path.home() / ".kimi" / "user-history"
_DEFAULT_CLAUDE_HISTORY = Path.home() / ".claude" / "history.jsonl"

_RULING_ROW_RE = re.compile(
    r"^\|\s*\[(?P<id>RUL-\d+)\][^|]*\|\s*(?P<date>[^|]*)\|\s*(?P<ruling>.*?)\s*"
    r"\|\s*`?(?P<mechanism>[^`|]*)`?\s*\|\s*(?P<status>[^|]*)\|\s*$"
)


# ---------------------------------------------------------------------------
# Data model
# ---------------------------------------------------------------------------


@dataclass(frozen=True)
class MinedPrompt:
    """One marker-matching (or candidate) prompt from any source."""

    text: str
    source: str  # "kg" | "kimi-fallback" | "claude-fallback"
    origin: str  # provenance string (platform:project, filename, session id)
    ts: str | None = None


@dataclass(frozen=True)
class RulingRow:
    """One parsed row of ``rulings/index.md``."""

    id: str
    ruling: str
    mechanism: str
    status: str  # normalized: "enforced" | "judgment" | "unknown"
    tokens: frozenset[str]


@dataclass
class ClusterGrade:
    signature: tuple[str, ...]
    prompts: list[MinedPrompt]
    matched_ruling: RulingRow | None
    grade: str  # "KNOWN+ENFORCED" | "KNOWN+JUDGMENT-ONLY" | "UNTRACKED"

    @property
    def count(self) -> int:
        return len(self.prompts)


@dataclass
class MiningResult:
    date: str
    total_prompts: int
    by_source: dict[str, int]
    clusters: list[ClusterGrade] = field(default_factory=list)


# ---------------------------------------------------------------------------
# Pass 1/2: marker matching + tokenization
# ---------------------------------------------------------------------------


def matched_markers(text: str, markers: Iterable[str] = MARKERS) -> tuple[str, ...]:
    """The subset of ``markers`` present in ``text`` (case-insensitive),
    in ``markers``' own order -- this IS the cluster signature (Pass 2)."""
    lowered = text.lower()
    return tuple(m for m in markers if m in lowered)


_TOKEN_RE = re.compile(r"[a-z0-9][a-z0-9_-]{2,}")


def tokenize(text: str) -> set[str]:
    """Lowercased, punctuation-stripped, stopword-filtered token set."""
    return {w for w in _TOKEN_RE.findall(text.lower()) if w not in _STOPWORDS}


def filter_marker_prompts(
    prompts: Iterable[MinedPrompt], markers: Iterable[str] = MARKERS
) -> list[tuple[MinedPrompt, tuple[str, ...]]]:
    """Pass 1: keep only prompts that hit at least one marker, paired with
    which marker(s) hit."""
    out: list[tuple[MinedPrompt, tuple[str, ...]]] = []
    for p in prompts:
        hits = matched_markers(p.text, markers)
        if hits:
            out.append((p, hits))
    return out


def cluster_by_signature(
    matched: Iterable[tuple[MinedPrompt, tuple[str, ...]]],
) -> dict[tuple[str, ...], list[MinedPrompt]]:
    """Pass 2: group by matched-marker signature."""
    clusters: dict[tuple[str, ...], list[MinedPrompt]] = {}
    for prompt, signature in matched:
        clusters.setdefault(signature, []).append(prompt)
    return clusters


# ---------------------------------------------------------------------------
# Pass 3: diff vs. the rulings canon
# ---------------------------------------------------------------------------


def _normalize_status(raw: str) -> str:
    lowered = raw.strip().lower()
    if "enforced" in lowered:
        return "enforced"
    if "judgment" in lowered or "judgement" in lowered:
        return "judgment"
    return "unknown"


def parse_rulings_index(text: str) -> list[RulingRow]:
    """Parse the ``| [RUL-NNN](...) | date | ruling | mechanism | status |``
    markdown table rows from ``rulings/index.md``'s content. Tolerant: rows
    that don't match the expected shape are skipped, not fatal.
    """
    rows: list[RulingRow] = []
    for line in text.splitlines():
        m = _RULING_ROW_RE.match(line.strip())
        if not m:
            continue
        ruling_text = m.group("ruling")
        rows.append(
            RulingRow(
                id=m.group("id"),
                ruling=ruling_text,
                mechanism=m.group("mechanism").strip(),
                status=_normalize_status(m.group("status")),
                tokens=frozenset(tokenize(ruling_text) | tokenize(m.group("id"))),
            )
        )
    return rows


def load_rulings(rulings_dir: Path) -> list[RulingRow]:
    """``[]`` when the canon is unavailable (a machine without the wiki
    checkout) -- never a hard failure, per the read-path rule (Phase 0)."""
    index_path = rulings_dir / "index.md"
    if not index_path.is_file():
        return []
    try:
        text = index_path.read_text(encoding="utf-8", errors="replace")
    except OSError:
        return []
    return parse_rulings_index(text)


def match_ruling(
    cluster_tokens: set[str],
    rulings: list[RulingRow],
    *,
    min_overlap: int = DEFAULT_MIN_OVERLAP,
) -> RulingRow | None:
    """The ruling with the most token overlap, if it clears ``min_overlap``."""
    best: RulingRow | None = None
    best_score = 0
    for r in rulings:
        score = len(cluster_tokens & r.tokens)
        if score > best_score:
            best_score = score
            best = r
    return best if best_score >= min_overlap else None


def grade_clusters(
    clusters: dict[tuple[str, ...], list[MinedPrompt]],
    rulings: list[RulingRow],
    *,
    min_overlap: int = DEFAULT_MIN_OVERLAP,
) -> list[ClusterGrade]:
    """Pass 3: grade + rank (by recurrence count, descending) every cluster."""
    graded: list[ClusterGrade] = []
    for signature, prompts in clusters.items():
        cluster_tokens: set[str] = set()
        for p in prompts:
            cluster_tokens |= tokenize(p.text)
        ruling = match_ruling(cluster_tokens, rulings, min_overlap=min_overlap)
        if ruling is None:
            grade = "UNTRACKED"
        elif ruling.status == "enforced":
            grade = "KNOWN+ENFORCED"
        else:
            grade = "KNOWN+JUDGMENT-ONLY"
        graded.append(
            ClusterGrade(
                signature=signature, prompts=prompts, matched_ruling=ruling, grade=grade
            )
        )
    graded.sort(key=lambda c: c.count, reverse=True)
    return graded


# ---------------------------------------------------------------------------
# Sources
# ---------------------------------------------------------------------------


def read_kimi_history(dir_path: Path) -> Iterator[MinedPrompt]:
    """``~/.kimi/user-history/*.jsonl``: one ``{"content": "..."}`` per line."""
    if not dir_path.is_dir():
        return
    for jsonl_path in sorted(dir_path.glob("*.jsonl")):
        try:
            fh = jsonl_path.open(encoding="utf-8", errors="replace")
        except OSError:
            continue
        with fh:
            for line in fh:
                line = line.strip()
                if not line:
                    continue
                try:
                    obj = json.loads(line)
                except json.JSONDecodeError:
                    continue
                content = obj.get("content") if isinstance(obj, dict) else None
                if isinstance(content, str) and content.strip():
                    yield MinedPrompt(
                        text=content, source="kimi-fallback", origin=jsonl_path.name
                    )


def read_claude_history(path: Path) -> Iterator[MinedPrompt]:
    """``~/.claude/history.jsonl``: ``{"display": "...", "timestamp": ms,
    "project": "..."}`` per line."""
    if not path.is_file():
        return
    try:
        fh = path.open(encoding="utf-8", errors="replace")
    except OSError:
        return
    with fh:
        for line in fh:
            line = line.strip()
            if not line:
                continue
            try:
                obj = json.loads(line)
            except json.JSONDecodeError:
                continue
            if not isinstance(obj, dict):
                continue
            display = obj.get("display")
            if not isinstance(display, str) or not display.strip():
                continue
            ts_ms = obj.get("timestamp")
            ts = None
            if isinstance(ts_ms, (int, float)):
                try:
                    ts = datetime.fromtimestamp(ts_ms / 1000, tz=UTC).isoformat()
                except OSError, OverflowError, ValueError:
                    ts = None
            project = obj.get("project") if isinstance(obj.get("project"), str) else "?"
            yield MinedPrompt(
                text=display, source="claude-fallback", origin=str(project), ts=ts
            )


def query_kg_user_turns(
    uri: str,
    username: str,
    password: str,
    *,
    markers: Iterable[str] = MARKERS,
    limit: int = 5000,
) -> list[MinedPrompt]:
    """Marker-matching ``user``-role ``ConversationTurn`` content, straight
    from the graph (SPEC-021 Tier I). Filters server-side (``CONTAINS`` over the
    marker list) so a multi-thousand-turn graph never ships unfiltered
    content over the wire. Raises on connection failure -- the caller
    decides whether that's fatal (CLI: a warning, KG source degrades to 0
    prompts, fallback sources still run).
    """
    from neo4j import GraphDatabase

    from orchestrata_ultimate.knowledge_graph.canonical_schema import (
        LABEL_CONVERSATION_TURN,
    )

    query = (
        f"MATCH (t:{LABEL_CONVERSATION_TURN} {{role: 'user'}}) "
        "WHERE any(term IN $markers WHERE toLower(t.content) CONTAINS term) "
        "RETURN t.content AS content, t.timestamp AS ts, t.platform AS platform, "
        "t.project AS project "
        "ORDER BY t.timestamp DESC LIMIT $limit"
    )
    driver = GraphDatabase.driver(uri, auth=(username, password))
    try:
        with driver.session(database=resolve_kg_database()) as session:
            records = session.run(
                query, markers=[m.lower() for m in markers], limit=limit
            )
            out: list[MinedPrompt] = []
            for r in records:
                content = r.get("content")
                if not content:
                    continue
                platform = r.get("platform") or "?"
                project = r.get("project") or "?"
                out.append(
                    MinedPrompt(
                        text=content,
                        source="kg",
                        origin=f"{platform}:{project}",
                        ts=r.get("ts"),
                    )
                )
            return out
    finally:
        driver.close()


# ---------------------------------------------------------------------------
# Report rendering
# ---------------------------------------------------------------------------


def _example_lines(prompts: list[MinedPrompt], n: int = 3) -> list[str]:
    lines = []
    for p in prompts[:n]:
        snippet = p.text.strip().replace("\n", " ")
        if len(snippet) > 200:
            snippet = snippet[:200] + "..."
        lines.append(f'    - ({p.source}/{p.origin}) "{snippet}"')
    return lines


def render_report(result: MiningResult) -> str:
    lines: list[str] = [
        f"# Ruling drift report — {result.date}",
        "",
        (
            f"Mined {result.total_prompts} marker-matching prompt(s): "
            + ", ".join(f"{v} {k}" for k, v in sorted(result.by_source.items()))
            if result.by_source
            else f"Mined {result.total_prompts} marker-matching prompt(s)."
        ),
        "",
        f"Clustered into {len(result.clusters)} signature group(s), by matched "
        "marker(s) — ranked by recurrence count.",
        "",
    ]
    for grade_name in ("KNOWN+ENFORCED", "KNOWN+JUDGMENT-ONLY", "UNTRACKED"):
        group = [c for c in result.clusters if c.grade == grade_name]
        lines.append(f"## {grade_name}")
        lines.append("")
        if not group:
            lines.append("(none)")
            lines.append("")
            continue
        for cluster in group:
            sig = ", ".join(cluster.signature) or "(no marker — should not happen)"
            ruling_note = (
                f" — matches {cluster.matched_ruling.id}: {cluster.matched_ruling.ruling}"
                if cluster.matched_ruling
                else ""
            )
            lines.append(f"- `{sig}` × {cluster.count}{ruling_note}")
            lines.extend(_example_lines(cluster.prompts))
        lines.append("")
    lines.append(
        "UNTRACKED clusters are candidate new rulings — filing one (wiki page + "
        "gate/hook, or `enforcement: none (judgment)`) is a human judgment call; "
        "this report never files one automatically."
    )
    return "\n".join(lines) + "\n"


def write_report(path: Path, content: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(content, encoding="utf-8")


# ---------------------------------------------------------------------------
# Orchestration
# ---------------------------------------------------------------------------


def mine(
    *,
    rulings_dir: Path,
    kimi_dir: Path,
    claude_history_path: Path,
    use_kg: bool,
    kg_uri: str,
    kg_username: str,
    kg_password: str,
    kg_limit: int = 5000,
    min_overlap: int = DEFAULT_MIN_OVERLAP,
    date: str | None = None,
) -> MiningResult:
    """Run the full pipeline (sources -> Pass 1 -> Pass 2 -> Pass 3)."""
    all_prompts: list[MinedPrompt] = []
    by_source: dict[str, int] = {}

    if use_kg:
        try:
            kg_prompts = query_kg_user_turns(kg_uri, kg_username, kg_password)
        except Exception as exc:  # noqa: BLE001 — a KG outage degrades, never aborts
            print(f"[mine_prompt_history] ⚠ KG query failed: {exc}", file=sys.stderr)
            kg_prompts = []
        all_prompts.extend(kg_prompts)

    all_prompts.extend(read_kimi_history(kimi_dir))
    all_prompts.extend(read_claude_history(claude_history_path))

    for p in all_prompts:
        by_source[p.source] = by_source.get(p.source, 0) + 1

    matched = filter_marker_prompts(all_prompts)
    clusters = cluster_by_signature(matched)
    rulings = load_rulings(rulings_dir)
    graded = grade_clusters(clusters, rulings, min_overlap=min_overlap)

    return MiningResult(
        date=date or datetime.now(tz=UTC).strftime("%Y-%m-%d"),
        total_prompts=len(matched),
        by_source=by_source,
        clusters=graded,
    )


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--out", type=Path, default=None, help="Report output path.")
    parser.add_argument(
        "--rulings-dir",
        type=Path,
        default=Path(os.environ.get("RULINGS_DIR", _DEFAULT_RULINGS_DIR)),
    )
    parser.add_argument("--kimi-dir", type=Path, default=_DEFAULT_KIMI_DIR)
    parser.add_argument("--claude-history", type=Path, default=_DEFAULT_CLAUDE_HISTORY)
    parser.add_argument(
        "--no-kg", action="store_true", help="Skip the Neo4j KG source entirely."
    )
    parser.add_argument("--kg-limit", type=int, default=5000)
    parser.add_argument("--min-overlap", type=int, default=DEFAULT_MIN_OVERLAP)
    args = parser.parse_args(argv)

    result = mine(
        rulings_dir=args.rulings_dir,
        kimi_dir=args.kimi_dir,
        claude_history_path=args.claude_history,
        use_kg=not args.no_kg,
        # KG_ROUTING_URI/KG_URI via resolve_uri(); the NEO4J_* keys were retired
        # with the Neo4j containers on 2026-07-27. The old kg_password default
        # was the literal "password", which was not the credential of any live
        # instance — it only ever produced an auth failure that looked like an
        # outage. Every remaining endpoint negotiates no auth.
        kg_uri=resolve_uri(),
        kg_username="",
        kg_password="",
        kg_limit=args.kg_limit,
        min_overlap=args.min_overlap,
    )
    report = render_report(result)
    out_path = args.out or (REPO_ROOT / "reports" / f"ruling-drift-{result.date}.md")
    write_report(out_path, report)
    print(
        f"[mine_prompt_history] wrote {out_path} ({result.total_prompts} prompt(s), "
        f"{len(result.clusters)} cluster(s))"
    )
    return 0


if __name__ == "__main__":
    sys.exit(main())