scripts.history_intel

Reference › scripts › history_intel

scripts/history_intel.py adds the time dimension to the Prolog code knowledge base: for a symbol that disappeared, it answers what was lost, when, how, and why, or states that no rationale was recorded. A human or agent runs it; nothing imports it. See also trace_intel and code_intel.

The subcommands are extract, report, what, when, how, why, no-recorded-rationale and explain. extract regenerates history_facts.pl from the evidence JSON corpus, a curated list of hand-verified loss events (CURATED_LOSS_EVENTS) and the docs/adr/*.md files. The fact builders are _bulk_facts, _curated_facts, _commit_facts (commit date and message for every commit sha involved, read with _git) and _rationale_source_facts. _commit_names_symbol decides whether a message names the symbol as a whole word, _pl_atom quotes strings as Prolog atoms, and _normalize_req_id canonicalises requirement ids. The query subcommands take a bare symbol name, an optional --commit to disambiguate, and common options added by _add_common_args; report gives corpus-wide counts and explain prints a proof tree for the why of one loss. For queries main calls ensure_kb (honouring --no-refresh), dispatches through _QUERY_DISPATCH, prints the result (--json for machine output) and exits 1 when the result contains an error.

At a glance

Class diagram

Dependency diagram

No internal imports and no importers.

API

Constants

Classes

CuratedLoss (dataclass)

Decorators: dataclass(frozen=True) · Line 122

One hand-verified (Symbol, Commit) loss event, plus its mechanism.

Functions

Source

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

#!/usr/bin/env python3
"""scripts/history_intel.py -- the TIME dimension over the Prolog code KB.

Companion to ``scripts/trace_intel.py``/``scripts/code_intel.py``, same
shape (a fact generator over ``.cache/`` + a thin argparse CLI over a
one-shot ``swipl`` query), but answering a question those two cannot: not
"what exists now" but "what USED to exist, and what happened to it" --
WHAT disappeared, WHEN (commit + date), HOW (dropped / renamed / moved /
retyped), and WHY (a real recorded rationale, a document that discusses the
loss without rationalizing it, or -- a first-class, positive answer, never
a query failure -- nothing at all).

CORRECTED SCOPE (2026-08-19): this tool's original design brief cited a
worked example -- ``postman.types.MessageEnvelope`` allegedly losing
``sender``/``recipient``/``timestamp``/``correlation_id`` fields in commit
``2883703a`` -- that turned out to be FALSE on direct git verification
(``git show 2883703a^:src/orchestrata_ultimate/postman/types.py`` shows the
canonical class never had those fields; they lived only on an independent,
zero-importer twin class in the file that commit deleted). See
``src/orchestrata_ultimate/prolog/history.pl``'s own module docstring for
the full account and ``no_loss_on_record/1`` for the predicate this forced.
The lesson is now load-bearing in this file's own design: every fact this
module generates is derived from git or from a document's own literal text
(commit sha / symbol name actually present, checked with ``re`` word
boundaries) -- never inferred from what would be a plausible-sounding story.

Facts (regenerated by the ``extract`` subcommand into
``.cache/history_intel/history_facts.pl``), two sources, composed but never
duplicating either:

  - ``plans/feature-loss-invalid-citations.json`` -- the T1 feature-loss
    citation audit's 176 validated records (a deleted symbol some code or
    test had cited against the WRONG requirement). Bulk source for
    ``deleted_in/2``, ``deleted_from_file/3``, ``citation_mismatch/4``,
    ``mismatch_evidence/4``. ``deleted_symbol`` there is a BARE name and
    ``file`` a path -- NOT dotted qualnames (an earlier agent on this
    project lost a day assuming otherwise; see the module's own header
    comment in that JSON file's producer).
  - ``CURATED_LOSS_EVENTS`` below -- a small, hand-verified set (four
    symbols) demonstrating each of the four HOW mechanisms
    (dropped/renamed/moved -- ``retyped`` has no real, git-verified
    instance found within this task's budget; it is schema-complete and
    unit-tested against a synthetic fixture, never asserted here as real
    history -- see ``tests/scripts/test_history_intel.py`` and this
    module's final report). Every entry cites the exact ``git show``/
    ``git log`` invocation that verified it, in its own comment.

For both sources, WHY facts (``rationale_source/4``) are derived
mechanically: a commit's rationale for a symbol is recorded only when the
commit's own message text names that symbol (regex word-boundary,
case-sensitive). This is deliberately conservative -- a commit message
that explains why a FILE was safe to delete (e.g. "0 production importers")
does not, by itself, count as a rationale for every symbol that file
happened to define; only a message that names the symbol directly does.
See ``_commit_names_symbol``.

ADR facts (``adr_status/2``, ``adr_title/2``, ``adr_date/2``,
``adr_reconstructed/1``, ``discussed_in/4``, plus ``rationale_source/4``
where SourceType=adr) come from ``docs/adr/*.md``: an ADR "discusses" a
(Symbol, Commit) pair when its text contains BOTH the symbol name
(word-boundary) and a hex token that is a prefix of the commit's full sha
(``_adr_mentions_commit``). A RECONSTRUCTED ADR (its own Status line or
body says so -- exactly ADR-0038's self-description) never promotes to
``rationale_source`` -- a document written weeks after the fact reconstructing
intent is real evidence that the loss was later AUDITED, not evidence of
what motivated it at the time; see ``history.pl``'s ``rationale_status/3``
for how the query layer keeps these two kinds of evidence apart.

Usage:
    python scripts/history_intel.py extract
    python scripts/history_intel.py report [--json]
    python scripts/history_intel.py what SYMBOL [--commit SHA] [--json]
    python scripts/history_intel.py when SYMBOL [--commit SHA] [--json]
    python scripts/history_intel.py how SYMBOL [--commit SHA] [--json]
    python scripts/history_intel.py why SYMBOL [--commit SHA] [--json]
    python scripts/history_intel.py no-recorded-rationale SYMBOL [--commit SHA] [--json]
    python scripts/history_intel.py explain SYMBOL [--commit SHA] [--json]

Every query subcommand auto-runs ``extract`` when ``history_facts.pl`` is
missing or older than its sources (the evidence JSON, this script, or any
``docs/adr/*.md``) -- pass ``--no-refresh`` to skip that and query whatever
is already on disk.
"""

from __future__ import annotations

import argparse
import contextlib
import json
import os
import re
import subprocess
import sys
import tempfile
from collections.abc import Iterable
from dataclasses import dataclass
from pathlib import Path
from typing import Any

REPO_ROOT = Path(__file__).resolve().parent.parent
EVIDENCE_JSON = REPO_ROOT / "plans" / "feature-loss-invalid-citations.json"
ADR_DIR = REPO_ROOT / "docs" / "adr"
PROLOG_DIR = REPO_ROOT / "src" / "orchestrata_ultimate" / "prolog"
HISTORY_PL = PROLOG_DIR / "history.pl"
EXPLAIN_PL = PROLOG_DIR / "explain.pl"

SCRATCH_DIR = REPO_ROOT / ".cache" / "history_intel"
HISTORY_FACTS = SCRATCH_DIR / "history_facts.pl"
TRACEABILITY_FACTS = REPO_ROOT / ".cache" / "trace_intel" / "traceability_facts.pl"

_PROLOG_TIMEOUT_SECONDS = 60

_HEX_TOKEN_RE = re.compile(r"\b[0-9a-f]{7,40}\b")


# ---------------------------------------------------------------------------
# Curated, git-verified worked examples (see module docstring)
# ---------------------------------------------------------------------------


@dataclass(frozen=True)
class CuratedLoss:
    """One hand-verified (Symbol, Commit) loss event, plus its mechanism.

    ``verification`` is the exact command(s) run to confirm this entry --
    kept here so a reviewer (or a future agent) can re-run it, not asked to
    trust prose."""

    symbol: str
    kind: str  # class / function / method
    commit: str  # full 40-char sha
    file: str  # path the symbol was deleted from
    mechanism: str  # dropped / renamed / moved
    renamed_to: str | None
    moved_to: str | None
    verification: str


CURATED_LOSS_EVENTS: tuple[CuratedLoss, ...] = (
    # DROPPED, no recorded rationale: a real, checked-in-once, never-touched-
    # again class deleted by an automated "checkpoint" commit whose message
    # names a completely unrelated file and gives no reason at all.
    CuratedLoss(
        symbol="CXPMixin",
        kind="class",
        commit="412e5cabf50cc8ecd63b5cdab1d503fec870d862",
        file="src/orchestrata_ultimate/agents/base_agent.py",
        mechanism="dropped",
        renamed_to=None,
        moved_to=None,
        verification=(
            "git show 412e5cab -- src/orchestrata_ultimate/agents/base_agent.py "
            "(deleted file mode, class CXPMixin at old line ~19); "
            "git show -s --format=%B 412e5cab (message: 'Pre-edit checkpoint: "
            "test_governance.py' -- names neither CXPMixin nor base_agent.py)"
        ),
    ),
    # DROPPED, WITH a recorded (commit-message) rationale -- the contrast
    # case: grpc.aio's transport-specific server wrapper, retired outright
    # (no successor) when cxpt became the sole CXP transport.
    CuratedLoss(
        symbol="CXPServer",
        kind="class",
        commit="5c7fb9f3a35562918305525833569291bf12255f",
        file="src/orchestrata_ultimate/cxp/grpc_server.py",
        mechanism="dropped",
        renamed_to=None,
        moved_to=None,
        verification=(
            "git show 5c7fb9f3^:src/orchestrata_ultimate/cxp/grpc_server.py "
            "| grep -n 'class CXPServer' (present pre-commit); "
            "git show 5c7fb9f3:src/orchestrata_ultimate/cxp/servicer.py "
            "| grep -n 'class CXPServer' (absent post-commit, unlike "
            "CXPServicer -- see below); commit message names grpc.aio's "
            "retirement and SPEC-023's GIL-off requirement directly"
        ),
    ),
    # MOVED: a verified 1:1 class-and-method migration to a new module, the
    # commit's own words: "carving the protocol core out was a move, not a
    # rewrite."
    CuratedLoss(
        symbol="CXPServicer",
        kind="class",
        commit="5c7fb9f3a35562918305525833569291bf12255f",
        file="src/orchestrata_ultimate/cxp/grpc_server.py",
        mechanism="moved",
        renamed_to=None,
        moved_to="orchestrata_ultimate.cxp.servicer.CXPServicer",
        verification=(
            "git show 5c7fb9f3^:.../cxp/grpc_server.py and "
            "git show 5c7fb9f3:.../cxp/servicer.py both grepped for "
            "'^class \\|def ' -- identical class name, identical method set "
            "(__init__, stats, enqueue_reply, send_packet, send_packet_stream, "
            "health_check, get_status, receive_packet, subscribe, acknowledge, "
            "_matches_subscription, _proto_to_dict, _dict_to_proto), near-"
            "identical line numbers"
        ),
    ),
    # RENAMED: extensively self-documented in the commit's own message,
    # including a filed AST-toolkit defect the rename exposed.
    CuratedLoss(
        symbol="DirectGRPCClient",
        kind="class",
        commit="11ebfc2e74215fbfdb22c2ff3977fd27ae9c1726",
        file="src/orchestrata_ultimate/cxp/direct_cxp_adapter.py",
        mechanism="renamed",
        renamed_to="DirectCXPClient",
        moved_to=None,
        verification=(
            "git show -s --format=%B 11ebfc2e (message: 'DirectGRPCClient -> "
            "DirectCXPClient; de-grpc the prod unit files ... Renamed across "
            "13 files, 1 def + 30 references'); grep -n '^class DirectCXPClient' "
            "src/orchestrata_ultimate/cxp/direct_cxp_adapter.py confirms the "
            "successor exists at HEAD"
        ),
    ),
)


def _pl_atom(value: str) -> str:
    """Quote a Python string as a Prolog atom (backslash + single-quote
    escaping), and flatten embedded newlines so every fact stays one
    greppable line -- extends ``scripts/trace_intel.py``'s ``_pl_atom``
    (which only doubles quotes) with backslash-escaping: real commit
    messages and evidence text in this corpus contain literal backslashes
    (e.g. a quoted ``grep 'context\\.'``), and SWI's quoted-atom reader
    treats a bare ``\\`` as the start of an escape sequence -- ``\\.`` is
    not a recognized one and is a hard syntax error at consult time, not a
    warning. Backslash escaped FIRST so quote-doubling below never doubles
    a backslash this step just inserted."""
    flat = " ".join(value.split())
    escaped = flat.replace("\\", "\\\\").replace("'", "''")
    return "'" + escaped + "'"


# Canonical story-id grammar (docs/req-id-grammar.md, ratified 2026-08-19):
# NEW citations use the padded, hyphenated `US-<idx>-<seq>` form. The legacy
# unpadded/unhyphenated `US<idx>-<seq>` form (e.g. `US1-001`, `US007-013`)
# stays independently grammar-valid for EXISTING corpus citations -- that
# doc is explicit that the existing corpus is NOT mass-rewritten ("bulk
# requirement-citation rewrites are exactly how this corpus got corrupted
# before"). This module writes brand-new facts, not corpus text, so its own
# output normalizes to canonical on the way out -- the evidence JSON's
# `req_id` field (11 legacy-form ids as of this writing) is read as-is and
# only the FACT ARGUMENT this module emits is reshaped; mismatch_evidence/4's
# free-text quotes of the audit's own prose are left untouched (rewriting a
# quote would misquote the source, the same reason commit_message/2 is never
# rewritten either).
_LEGACY_STORY_ID_RE = re.compile(r"-US(\d{1,3})-")


def _normalize_req_id(req_id: str) -> str:
    """``REQ-002-US-001-001`` -> ``REQ-002-US-001-001``; already-canonical or
    non-story ids pass through unchanged (idempotent: the regex only matches
    ``US`` directly followed by digits, never ``US-``)."""
    return _LEGACY_STORY_ID_RE.sub(lambda m: f"-US-{int(m.group(1)):03d}-", req_id)


# ---------------------------------------------------------------------------
# git helpers
# ---------------------------------------------------------------------------


def _git(*args: str) -> str:
    """Run ``git <args>`` from REPO_ROOT and return stripped stdout, raising
    on failure (a commit sha this module was told to use that git cannot
    resolve is itself a finding worth a loud error, not a silent gap)."""
    proc = subprocess.run(
        ["git", *args],
        cwd=REPO_ROOT,
        capture_output=True,
        text=True,
        timeout=30,
    )
    if proc.returncode != 0:
        raise RuntimeError(f"git {' '.join(args)} failed: {proc.stderr.strip()}")
    return proc.stdout.strip()


def _commit_date(commit: str) -> str:
    """ISO-8601 committer date for ``commit`` (the historical WHEN)."""
    return _git("show", "-s", "--format=%cI", commit)


def _commit_message(commit: str) -> str:
    """Full commit message (subject + body) for ``commit``."""
    return _git("show", "-s", "--format=%B", commit)


def _commit_names_symbol(message: str, symbol: str) -> bool:
    """True iff ``message`` names ``symbol`` as a whole word -- the
    conservative WHY-evidence gate this module's docstring documents."""
    return re.search(rf"\b{re.escape(symbol)}\b", message) is not None


# ---------------------------------------------------------------------------
# extract -- bulk evidence JSON -> loss/citation facts
# ---------------------------------------------------------------------------

_KIND_MARKERS: tuple[tuple[str, str], ...] = (
    ("dataclass", "class"),
    ("@dataclass", "class"),
    (" class ", "class"),
    ("Protocol", "class"),
    ("enum", "enum"),
    ("Enum", "enum"),
    ("method", "method"),
    ("function", "function"),
)


def _best_effort_kind(symbol_did: str) -> str | None:
    """A symbol's kind (class/method/function/enum) from the evidence
    corpus's free-text ``symbol_did`` field, ONLY when an explicit keyword
    is present -- never guessed from the bare name's capitalization alone
    (a guess is not evidence)."""
    for marker, kind in _KIND_MARKERS:
        if marker in symbol_did:
            return kind
    return None


def _load_evidence_records(path: Path = EVIDENCE_JSON) -> list[dict[str, Any]]:
    with open(path, encoding="utf-8") as fh:
        data = json.load(fh)
    records: list[dict[str, Any]] = data.get("records", [])
    return records


def _bulk_facts(
    records: list[dict[str, Any]],
    curated_keys: frozenset[tuple[str, str]] = frozenset(),
) -> tuple[list[str], set[str]]:
    """Facts from the T1 evidence corpus: ``deleted_in/2``,
    ``deleted_from_file/3``, ``deletion_mechanism/3`` (always ``dropped`` --
    see module docstring), ``citation_mismatch/4``, ``mismatch_evidence/4``,
    plus best-effort ``symbol_kind/2``. ``curated_keys`` is the
    (symbol, commit) set already covered by ``CURATED_LOSS_EVENTS`` --
    ``CXPMixin``/``412e5cab`` genuinely appears in BOTH sources (the T1
    audit independently flagged it as a wrong-spec citation, this module's
    curated set independently picked it as a no-rationale worked example);
    the loss-event/kind/mechanism facts are skipped here to avoid emitting
    the identical ground fact twice, while citation_mismatch/mismatch_evidence
    -- data ONLY the T1 corpus has -- are still emitted. Returns
    (facts, commit_shas)."""
    facts: list[str] = []
    commits: set[str] = set()
    seen_loss: set[tuple[str, str]] = set(curated_keys)
    seen_kind: set[str] = {sym for sym, _ in curated_keys}
    for rec in records:
        symbol = rec["deleted_symbol"]
        commit = rec["deleting_commit"]
        file_ = rec["file"]
        req_id = _normalize_req_id(rec["req_id"])
        mismatch_class = rec["mismatch_class"]
        evidence_raw = rec.get("methods", "")
        # A handful of records carry `methods` as a list of investigation
        # steps rather than one prose string; flatten either shape the same
        # way (real content either way, never fabricated).
        evidence = (
            "; ".join(evidence_raw) if isinstance(evidence_raw, list) else evidence_raw
        )
        commits.add(commit)

        kind = _best_effort_kind(rec.get("symbol_did", ""))
        if kind is not None and symbol not in seen_kind:
            seen_kind.add(symbol)
            facts.append(f"symbol_kind({_pl_atom(symbol)}, {kind}).")

        key = (symbol, commit)
        if key not in seen_loss:
            seen_loss.add(key)
            facts.append(f"deleted_in({_pl_atom(symbol)}, {_pl_atom(commit)}).")
            facts.append(
                f"deleted_from_file({_pl_atom(symbol)}, {_pl_atom(commit)}, "
                f"{_pl_atom(file_)})."
            )
            facts.append(
                f"deletion_mechanism({_pl_atom(symbol)}, {_pl_atom(commit)}, dropped)."
            )

        facts.append(
            f"citation_mismatch({_pl_atom(symbol)}, {_pl_atom(commit)}, "
            f"{_pl_atom(req_id)}, {mismatch_class.replace('-', '_').lower()})."
        )
        if evidence:
            facts.append(
                f"mismatch_evidence({_pl_atom(symbol)}, {_pl_atom(commit)}, "
                f"{_pl_atom(req_id)}, {_pl_atom(evidence)})."
            )
    return facts, commits


def _curated_facts() -> tuple[list[str], set[str]]:
    """Facts from ``CURATED_LOSS_EVENTS`` -- the four hand-verified worked
    examples covering dropped/renamed/moved (see module docstring for why
    ``retyped`` is absent here)."""
    facts: list[str] = []
    commits: set[str] = set()
    for ev in CURATED_LOSS_EVENTS:
        commits.add(ev.commit)
        facts.append(f"symbol_kind({_pl_atom(ev.symbol)}, {ev.kind}).")
        facts.append(f"deleted_in({_pl_atom(ev.symbol)}, {_pl_atom(ev.commit)}).")
        facts.append(
            f"deleted_from_file({_pl_atom(ev.symbol)}, {_pl_atom(ev.commit)}, "
            f"{_pl_atom(ev.file)})."
        )
        facts.append(
            f"deletion_mechanism({_pl_atom(ev.symbol)}, {_pl_atom(ev.commit)}, "
            f"{ev.mechanism})."
        )
        if ev.renamed_to is not None:
            facts.append(
                f"renamed_to({_pl_atom(ev.symbol)}, {_pl_atom(ev.commit)}, "
                f"{_pl_atom(ev.renamed_to)})."
            )
        if ev.moved_to is not None:
            facts.append(
                f"moved_to({_pl_atom(ev.symbol)}, {_pl_atom(ev.commit)}, "
                f"{_pl_atom(ev.moved_to)})."
            )
    return facts, commits


def _commit_facts(commits: Iterable[str]) -> list[str]:
    """``commit_date/2`` + ``commit_message/2`` for every commit sha in
    ``commits``, plus the mechanical ``rationale_source(..., commit_message,
    Commit)`` fact for every (symbol, commit) loss event the message names
    (see ``_commit_names_symbol``). Requires the loss facts to already be
    computed by the caller so it can check symbol membership; simplest to
    just recompute the name-check here against the SAME two sources."""
    facts: list[str] = []
    for commit in sorted(commits):
        date = _commit_date(commit)
        message = _commit_message(commit)
        facts.append(f"commit_date({_pl_atom(commit)}, {_pl_atom(date)}).")
        facts.append(f"commit_message({_pl_atom(commit)}, {_pl_atom(message)}).")
    return facts


def _rationale_source_facts(
    symbol_commit_pairs: Iterable[tuple[str, str]],
    commit_messages: dict[str, str],
) -> list[str]:
    """``rationale_source(Symbol, Commit, commit_message, Commit)`` for
    every (Symbol, Commit) pair whose commit message literally names
    Symbol -- the conservative WHY-evidence gate (module docstring)."""
    facts: list[str] = []
    seen: set[tuple[str, str]] = set()
    for symbol, commit in symbol_commit_pairs:
        key = (symbol, commit)
        if key in seen:
            continue
        message = commit_messages.get(commit, "")
        if message and _commit_names_symbol(message, symbol):
            seen.add(key)
            facts.append(
                f"rationale_source({_pl_atom(symbol)}, {_pl_atom(commit)}, "
                f"commit_message, {_pl_atom(commit)})."
            )
    return facts


# ---------------------------------------------------------------------------
# extract -- ADR corpus scan
# ---------------------------------------------------------------------------

_ADR_ID_RE = re.compile(r"^(\d{4})-")
_ADR_TITLE_RE = re.compile(r"^#\s*ADR-\d{4}:\s*(.+)$", re.MULTILINE)
_ADR_STATUS_RE = re.compile(r"^-\s*\*\*Status\*\*:\s*(.+)$", re.MULTILINE)
_ADR_DATE_RE = re.compile(r"^-\s*\*\*Date\*\*:\s*(\d{4}-\d{2}-\d{2})", re.MULTILINE)


def _adr_files() -> list[Path]:
    if not ADR_DIR.is_dir():
        return []
    return sorted(p for p in ADR_DIR.glob("*.md") if p.stem.lower() != "readme")


def _adr_id(path: Path) -> str | None:
    match = _ADR_ID_RE.match(path.stem)
    if not match:
        return None
    return f"ADR-{match.group(1)}"


def _adr_mentions_commit(text: str, commit: str) -> bool:
    """True iff ``text`` contains a hex token that is a case-sensitive
    prefix of ``commit``'s full sha (git's own abbreviation convention --
    e.g. the ADR quoting ``2883703a`` for a commit whose full sha starts
    with ``2883703a``). Length-7+ only, matching git's minimum unambiguous
    abbreviation, so a short decimal-looking token can't false-positive."""
    for token in _HEX_TOKEN_RE.findall(text):
        if commit.startswith(token):
            return True
    return False


def _adr_facts(
    symbol_commit_pairs: Iterable[tuple[str, str]],
) -> list[str]:
    """ADR metadata (``adr_status/2``, ``adr_title/2``, ``adr_date/2``,
    ``adr_reconstructed/1``) for every ADR on disk, plus, for every
    (Symbol, Commit) loss event this module already knows about,
    ``discussed_in/4`` when an ADR's text names both -- and
    ``rationale_source(Symbol, Commit, adr, AdrId)`` too, UNLESS that ADR is
    itself flagged reconstructed (module docstring: reconstructed evidence
    documents an audit, not a contemporaneous rationale)."""
    facts: list[str] = []
    adrs: list[tuple[str, str]] = []  # (adr_id, text)
    for path in _adr_files():
        adr_id = _adr_id(path)
        if adr_id is None:
            continue
        text = path.read_text(encoding="utf-8")
        adrs.append((adr_id, text))

        title_match = _ADR_TITLE_RE.search(text)
        if title_match:
            facts.append(
                f"adr_title({_pl_atom(adr_id)}, {_pl_atom(title_match.group(1))})."
            )
        status_match = _ADR_STATUS_RE.search(text)
        if status_match:
            facts.append(
                f"adr_status({_pl_atom(adr_id)}, {_pl_atom(status_match.group(1))})."
            )
        date_match = _ADR_DATE_RE.search(text)
        if date_match:
            facts.append(
                f"adr_date({_pl_atom(adr_id)}, {_pl_atom(date_match.group(1))})."
            )
        if "reconstructed" in text.lower():
            facts.append(f"adr_reconstructed({_pl_atom(adr_id)}).")

    pairs = list(dict.fromkeys(symbol_commit_pairs))  # de-dup, keep order
    for symbol, commit in pairs:
        for adr_id, text in adrs:
            if not _adr_mentions_commit(text, commit):
                continue
            if not _commit_names_symbol(text, symbol):
                continue
            facts.append(
                f"discussed_in({_pl_atom(symbol)}, {_pl_atom(commit)}, adr, "
                f"{_pl_atom(adr_id)})."
            )
            if "reconstructed" not in text.lower():
                facts.append(
                    f"rationale_source({_pl_atom(symbol)}, {_pl_atom(commit)}, "
                    f"adr, {_pl_atom(adr_id)})."
                )
    return facts


# ---------------------------------------------------------------------------
# extract -- orchestration
# ---------------------------------------------------------------------------

_PREDICATE_LIST = (
    "symbol_kind/2, deleted_in/2, deleted_from_file/3, "
    "commit_date/2, commit_message/2, deletion_mechanism/3, renamed_to/3, "
    "moved_to/3, retyped_to/4, citation_mismatch/4, mismatch_evidence/4, "
    "rationale_source/4, discussed_in/4, adr_status/2, adr_title/2, "
    "adr_date/2, adr_reconstructed/1"
)
_DYNAMIC_DECL = f":- dynamic {_PREDICATE_LIST}."
# Facts below are grouped per SOURCE RECORD (all of one symbol's loss-event
# facts together, then the next symbol), not per predicate -- more readable
# for a human scanning one event at a time, at the cost of every predicate
# above being genuinely discontiguous; declared so SWI's consult stays
# warning-free instead of noting each interleaving as if it were an error.
_DISCONTIGUOUS_DECL = f":- discontiguous {_PREDICATE_LIST}."


def extract_facts(
    evidence_json: Path = EVIDENCE_JSON,
    output_path: Path = HISTORY_FACTS,
) -> dict[str, int]:
    """Regenerate ``output_path`` from the T1 evidence JSON, the curated
    worked examples, and the ADR corpus. Returns per-predicate-family
    counts for the caller to report."""
    records = _load_evidence_records(evidence_json)
    curated_facts, curated_commits = _curated_facts()
    curated_keys = frozenset((ev.symbol, ev.commit) for ev in CURATED_LOSS_EVENTS)
    bulk_facts, bulk_commits = _bulk_facts(records, curated_keys)

    all_commits = bulk_commits | curated_commits
    commit_facts = _commit_facts(all_commits)
    commit_messages = {c: _commit_message(c) for c in all_commits}

    pairs = [(rec["deleted_symbol"], rec["deleting_commit"]) for rec in records]
    pairs += [(ev.symbol, ev.commit) for ev in CURATED_LOSS_EVENTS]
    rationale_facts = _rationale_source_facts(pairs, commit_messages)
    adr_facts = _adr_facts(pairs)

    lines = [
        "% history_facts.pl -- GENERATED by scripts/history_intel.py extract.",
        "% Do not hand-edit; regenerate via `python scripts/history_intel.py extract`.",
        "",
        _DYNAMIC_DECL,
        _DISCONTIGUOUS_DECL,
        "",
        *curated_facts,
        "",
        *bulk_facts,
        "",
        *commit_facts,
        "",
        *rationale_facts,
        "",
        *adr_facts,
        "",
    ]
    output_path.parent.mkdir(parents=True, exist_ok=True)
    output_path.write_text("\n".join(lines) + "\n", encoding="utf-8")

    bulk_keys = {(r["deleted_symbol"], r["deleting_commit"]) for r in records}
    return {
        "loss_events": len(bulk_keys | curated_keys),
        "citation_mismatches": len(records),
        "commits": len(all_commits),
        "rationale_from_commit_message": len(rationale_facts),
        "adr_facts": len(adr_facts),
    }


def cmd_extract(_args: argparse.Namespace) -> int:
    counts = extract_facts()
    print(
        "history_intel: wrote "
        f"{counts['loss_events']} loss event(s), "
        f"{counts['citation_mismatches']} citation-mismatch record(s), "
        f"{counts['commits']} commit(s), "
        f"{counts['rationale_from_commit_message']} commit-message rationale "
        f"fact(s), {counts['adr_facts']} ADR-derived fact(s) to {HISTORY_FACTS}",
        file=sys.stderr,
    )
    return 0


def _needs_refresh(artifact: Path, *sources: Path) -> bool:
    if not artifact.exists():
        return True
    stamp = artifact.stat().st_mtime
    return any(s.exists() and stamp < s.stat().st_mtime for s in sources)


def _extraction_sources() -> list[Path]:
    sources = [EVIDENCE_JSON, Path(__file__)]
    sources.extend(_adr_files())
    return sources


def ensure_kb(no_refresh: bool) -> None:
    if not no_refresh and _needs_refresh(HISTORY_FACTS, *_extraction_sources()):
        print(
            f"history_intel: {HISTORY_FACTS} missing/stale -- running extract",
            file=sys.stderr,
        )
        extract_facts()
    if not HISTORY_FACTS.exists():
        raise SystemExit(
            f"history_intel: {HISTORY_FACTS} does not exist -- run `extract` "
            "first, or drop --no-refresh"
        )


# ---------------------------------------------------------------------------
# Prolog session -- one-shot subprocess (facts are small; no warm-worker
# machinery needed at this scale, unlike trace_intel.py's ~13MB structural KB)
# ---------------------------------------------------------------------------


def _kb_files() -> list[Path]:
    files = [EXPLAIN_PL, HISTORY_PL, HISTORY_FACTS]
    if TRACEABILITY_FACTS.exists():
        files.append(TRACEABILITY_FACTS)
    return files


def _run_prolog(goal: str, kb_files: list[Path] | None = None) -> dict[str, Any]:
    """Consult ``kb_files`` (default: the standard KB) then run ``goal`` in
    one ``swipl`` subprocess, returning its ``R`` binding as a Python dict.
    Mirrors ``scripts/trace_intel.py``'s ``_run_prolog_oneshot`` protocol."""
    files = kb_files if kb_files is not None else _kb_files()
    SCRATCH_DIR.mkdir(parents=True, exist_ok=True)
    query_fd, query_path = tempfile.mkstemp(suffix=".pl", dir=str(SCRATCH_DIR))
    os.close(query_fd)
    result_fd, result_path = tempfile.mkstemp(suffix=".json", dir=str(SCRATCH_DIR))
    os.close(result_fd)
    with contextlib.suppress(OSError):
        os.unlink(result_path)

    try:
        with open(query_path, "w", encoding="utf-8") as fh:
            fh.write(":- use_module(library(http/json)).\n")
            for kb_file in files:
                escaped = str(kb_file).replace("'", "''")
                fh.write(f":- consult('{escaped}').\n")
            escaped_result = result_path.replace("'", "''")
            fh.write(f"run :- {goal}, ")
            fh.write("setup_call_cleanup(")
            fh.write(f"open('{escaped_result}', write, Out, [encoding(utf8)]),")
            fh.write("(json_write_dict(Out, R, [width(0)]), nl(Out)),")
            fh.write("close(Out)).\n")

        cmd = ["swipl", "-q", "-g", "run", "-g", "halt", query_path]
        proc = subprocess.run(
            cmd, capture_output=True, text=True, timeout=_PROLOG_TIMEOUT_SECONDS
        )

        if os.path.exists(result_path):
            with open(result_path, encoding="utf-8") as fh:
                content = fh.read().strip()
            if content:
                parsed = json.loads(content)
                if isinstance(parsed, dict):
                    return parsed
                return {"error": "json_not_object", "raw": content[:500]}

        return {
            "error": "prolog_failed",
            "returncode": proc.returncode,
            "stderr": proc.stderr[:1000],
            "stdout": proc.stdout[:1000],
        }
    except subprocess.TimeoutExpired:
        return {"error": "timeout"}
    finally:
        with contextlib.suppress(OSError):
            os.unlink(query_path)
        with contextlib.suppress(OSError):
            os.unlink(result_path)


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


def _commit_filter(args: argparse.Namespace) -> str:
    return (
        f", Commit == {_pl_atom(args.commit)}" if getattr(args, "commit", None) else ""
    )


def _cmd_what(args: argparse.Namespace) -> dict[str, Any]:
    goal = (
        "( \\+ history:loss_known(Symbol) "
        "-> R = json{symbol: Symbol, status: no_loss_on_record} "
        "; findall(json{commit: Commit, kind: Kind, file: File}, "
        "( history:deleted_in(Symbol, Commit)"
        f"{_commit_filter(args)}, "
        "( symbol_kind(Symbol, Kind) -> true ; Kind = unknown ), "
        "( deleted_from_file(Symbol, Commit, File) -> true ; File = unknown ) ), "
        "Events), R = json{symbol: Symbol, status: loss_known, events: Events} )"
    )
    return _run_prolog(f"Symbol = {_pl_atom(args.symbol)}, {goal}")


def _cmd_when(args: argparse.Namespace) -> dict[str, Any]:
    goal = (
        "( \\+ history:loss_known(Symbol) "
        "-> R = json{symbol: Symbol, status: no_loss_on_record} "
        "; findall(json{commit: Commit, date: Date}, "
        "( history:deleted_in(Symbol, Commit)"
        f"{_commit_filter(args)}, "
        "( commit_date(Commit, Date) -> true ; Date = unknown ) ), "
        "Events), R = json{symbol: Symbol, status: loss_known, events: Events} )"
    )
    return _run_prolog(f"Symbol = {_pl_atom(args.symbol)}, {goal}")


def _cmd_how(args: argparse.Namespace) -> dict[str, Any]:
    goal = (
        "( \\+ history:loss_known(Symbol) "
        "-> R = json{symbol: Symbol, status: no_loss_on_record} "
        "; findall(json{commit: Commit, mechanism: Mechanism, "
        "renamed_to: RenamedTo, moved_to: MovedTo}, "
        "( history:deleted_in(Symbol, Commit)"
        f"{_commit_filter(args)}, "
        "( deletion_mechanism(Symbol, Commit, Mechanism) -> true ; Mechanism = unknown ), "
        "( renamed_to(Symbol, Commit, RenamedTo) -> true ; RenamedTo = null ), "
        "( moved_to(Symbol, Commit, MovedTo) -> true ; MovedTo = null ) ), "
        "Events), R = json{symbol: Symbol, status: loss_known, events: Events} )"
    )
    return _run_prolog(f"Symbol = {_pl_atom(args.symbol)}, {goal}")


def _cmd_why(args: argparse.Namespace) -> dict[str, Any]:
    goal = (
        "( \\+ history:loss_known(Symbol) "
        "-> R = json{symbol: Symbol, status: no_loss_on_record} "
        "; findall(json{commit: Commit, rationale: Status}, "
        "( history:deleted_in(Symbol, Commit)"
        f"{_commit_filter(args)}, "
        "history:rationale_status_json(Symbol, Commit, Status) ), "
        "Events), R = json{symbol: Symbol, status: loss_known, events: Events} )"
    )
    return _run_prolog(f"Symbol = {_pl_atom(args.symbol)}, {goal}")


def _cmd_no_recorded_rationale(args: argparse.Namespace) -> dict[str, Any]:
    goal = (
        "( \\+ history:loss_known(Symbol) "
        "-> R = json{symbol: Symbol, status: no_loss_on_record} "
        "; findall(json{commit: Commit, no_recorded_rationale: Bool}, "
        "( history:deleted_in(Symbol, Commit)"
        f"{_commit_filter(args)}, "
        "( history:no_recorded_rationale(Symbol, Commit) -> Bool = true ; Bool = false ) ), "
        "Events), R = json{symbol: Symbol, status: loss_known, events: Events} )"
    )
    return _run_prolog(f"Symbol = {_pl_atom(args.symbol)}, {goal}")


def _cmd_explain(args: argparse.Namespace) -> dict[str, Any]:
    commit = getattr(args, "commit", None)
    if commit:
        goal_atom = f"history:rationale_status({_pl_atom(args.symbol)}, {_pl_atom(commit)}, Status)"
    else:
        goal_atom = f"history:rationale_status({_pl_atom(args.symbol)}, Commit, Status)"
    goal = f"explain:explain_or_whynot_json({goal_atom}, R)"
    return _run_prolog(goal)


def _cmd_report(_args: argparse.Namespace) -> dict[str, Any]:
    goal = (
        "aggregate_all(count, deleted_in(_, _), LossEvents), "
        "aggregate_all(count, "
        "distinct([Symbol, Commit], rationale_source(Symbol, Commit, _, _)), "
        "RationaleRecorded), "
        "aggregate_all(count, "
        "( deleted_in(Symbol, Commit), history:no_recorded_rationale(Symbol, Commit) ), "
        "NoRationale), "
        "aggregate_all(count, citation_mismatch(_, _, _, _), CitationMismatches), "
        "aggregate_all(count, adr_reconstructed(_), ReconstructedAdrs), "
        "R = json{loss_events: LossEvents, rationale_recorded: RationaleRecorded, "
        "no_recorded_rationale: NoRationale, citation_mismatches: CitationMismatches, "
        "reconstructed_adrs: ReconstructedAdrs}"
    )
    return _run_prolog(goal)


_QUERY_DISPATCH = {
    "what": _cmd_what,
    "when": _cmd_when,
    "how": _cmd_how,
    "why": _cmd_why,
    "no-recorded-rationale": _cmd_no_recorded_rationale,
    "explain": _cmd_explain,
    "report": _cmd_report,
}


def _print_result(result: dict[str, Any], json_mode: bool) -> None:
    if json_mode:
        print(json.dumps(result, indent=2, sort_keys=True))
        return
    if "error" in result:
        print(f"error: {result}", file=sys.stderr)
        return
    for key, value in result.items():
        if isinstance(value, list):
            print(f"{key}: ({len(value)})")
            for item in value:
                print(f"  - {item}")
        else:
            print(f"{key}: {value}")


def _add_common_args(parser: argparse.ArgumentParser) -> None:
    parser.add_argument("--json", action="store_true", help="Print raw JSON, not text.")
    parser.add_argument(
        "--no-refresh",
        action="store_true",
        help="Skip the auto-extract freshness check; query the KB as-is.",
    )


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="history_intel",
        description=(
            "The TIME dimension over the Prolog code KB -- WHAT/WHEN/HOW/WHY "
            "a symbol disappeared, or the documented absence of a why."
        ),
    )
    sub = parser.add_subparsers(dest="command", required=True)

    sub.add_parser(
        "extract",
        help="Regenerate history_facts.pl from the T1 evidence JSON + curated "
        "worked examples + docs/adr/*.md.",
    )
    report_p = sub.add_parser("report", help="Corpus-wide counts.")
    _add_common_args(report_p)

    for name, help_text in (
        ("what", "WHAT disappeared -- kind + file, per matching loss event."),
        ("when", "WHEN -- deleting commit + its date."),
        ("how", "HOW -- mechanism (dropped/renamed/moved/retyped) + target."),
        ("why", "WHY -- rationale_status per matching loss event."),
        (
            "no-recorded-rationale",
            "TRUE/FALSE per matching loss event -- a positive answer, not "
            "a failure.",
        ),
        (
            "explain",
            "Proof tree (via explain.pl) for the WHY of one symbol's loss.",
        ),
    ):
        p = sub.add_parser(name, help=help_text)
        p.add_argument("symbol", help="Bare symbol name (not a dotted qualname).")
        p.add_argument(
            "--commit", default=None, help="Disambiguate when multiple commits match."
        )
        _add_common_args(p)

    return parser


def main(argv: list[str] | None = None) -> int:
    parser = _build_parser()
    args = parser.parse_args(argv)

    if args.command == "extract":
        return cmd_extract(args)

    ensure_kb(no_refresh=args.no_refresh)
    result = _QUERY_DISPATCH[args.command](args)
    _print_result(result, json_mode=args.json)
    return 1 if "error" in result else 0


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