new_brain_write.py

"""Write path — the learning loop generalized (new-brain-mvp-spec.md §2).

input -> extract (personal schema) -> tag -> SYNCHRONOUS conflict check (pgvector kNN density)
-> triage -> write with verdict + evidence + temporal frame. Untypeable input -> documents.

Triage lanes on a near-hit with a differing value:
  1. explicit supersession stated  -> auto-resolve, supersede, link
       ** EXCEPTION (Aaron 2026-07-03): RULE-kind NEVER takes this lane — a rule replacing a
          rule ALWAYS runs the resolution round, explicit or not. **
  2. mechanically verifiable (path/URL ground truth; carts for code-STATE) -> resolve by truth
  3. else -> write BOTH as contested + INTERRUPT with the head-to-head

STDOUT LAW: stderr only.
"""

from __future__ import annotations

import os
import re
import sys

from new_brain_extract import extract_facts
from new_brain_store import NewBrainStore, embed_one

# A near-hit of the SAME kind with cosine sim >= NEAR_HIT_SIM is a conflict candidate — UNLESS
# the statements are an exact (normalized) restatement, which is a pure re-capture, not a
# conflict. Note the deliberate tradeoff: a near-paraphrase of the SAME fact (same value, reworded)
# can trip a false interrupt. That is the SAFE failure mode — it asks Aaron, it never silently
# merges — and matches the spec's interrupt-on-detect/recoverability stance. The precision of this
# boundary (paraphrase-vs-changed-value) is UNMEASURED: logged as a null, tuning is future work.
NEAR_HIT_SIM = 0.72

# only these per-kind field keys are load-bearing columns; anything else the model emits is dropped
KIND_FIELD_COLS = {"scope", "status_detail", "trigger_predicate", "risk_weight",
                   "earned_by", "subject", "expected_lifetime"}

# scribe/model names are PROVENANCE (captured_by), never aboutness — a topic matching one is
# dropped before tag resolution (the provenance-is-not-aboutness law, enforced in code)
_MODEL_NAMES = {"opus", "fable", "sonnet", "haiku", "claude", "qwen", "gpt", "gemini",
                "deepseek", "llama", "mistral", "ollama"}


def _gate_topics(topics: list[str]) -> tuple[list[str], list[str]]:
    """Deterministic topic hygiene before tag resolution: drop scribe/model names, dedup
    case-insensitively. Returns (kept, dropped) — dropped is logged to meta, never lost."""
    kept, dropped, seen = [], [], set()
    for t in topics:
        low = t.strip().lower()
        if not low or low in seen:
            continue
        seen.add(low)
        if low in _MODEL_NAMES:
            dropped.append(t)
            continue
        kept.append(t.strip())
    return kept, dropped


def _entity_shaped(topic: str) -> bool:
    """Only entity-shaped unresolved topics surface to Aaron as candidate roots: something with
    an uppercase letter, a dot path, or a digit ('Open Brain', 'Domain.UE') — never lowercase
    common words ('count', 'parity'), which are extraction noise, kept in meta only."""
    return any(c.isupper() or c.isdigit() or c == "." for c in topic)

_PATH_RE = re.compile(r"[A-Za-z]:\\[^\s\"']+|/(?:[\w.-]+/)+[\w.-]+")
_URL_RE = re.compile(r"https?://[^\s\"')]+")


def _paths(text: str) -> list[str]:
    return _PATH_RE.findall(text) + _URL_RE.findall(text)


def _mechanical_resolution(new_stmt: str, old_stmt: str) -> str | None:
    """Ground-truth resolver for STATE/REFERENCE facts that name a filesystem path: the fact
    whose path actually EXISTS wins. Returns 'new' | 'old' | None (can't mechanically decide).
    URLs are treated as un-checkable here (no network in the write path) -> None."""
    new_p = [p for p in _paths(new_stmt) if not p.startswith("http")]
    old_p = [p for p in _paths(old_stmt) if not p.startswith("http")]
    if not new_p and not old_p:
        return None
    new_true = any(os.path.exists(p) for p in new_p) if new_p else None
    old_true = any(os.path.exists(p) for p in old_p) if old_p else None
    if new_true and old_true is False:
        return "new"
    if old_true and new_true is False:
        return "old"
    return None


def _norm(s: str) -> str:
    return re.sub(r"\s+", " ", s).strip().lower().rstrip(".")


def _candidate(fact: dict, near: list[dict]) -> dict | None:
    """Top near-hit that is a genuine conflict candidate: same kind, sim >= NEAR_HIT_SIM, and not
    an exact restatement of this fact."""
    for n in near:
        if n["kind"] != fact["kind"] or n["sim"] < NEAR_HIT_SIM:
            continue
        if _norm(fact["statement"]) == _norm(n["statement"]):
            continue                    # pure re-capture of the identical fact — not a conflict
        return n
    return None


def _fact_row(fact: dict, tags: list[str], emb, source_type: str,
              source_ref: str | None, captured_by: str, unresolved: list[str],
              dropped_models: list[str] | None = None, extra_meta: dict | None = None) -> dict:
    fields = {k: v for k, v in (fact.get("fields") or {}).items()
              if k in KIND_FIELD_COLS and v not in (None, "")}
    meta = {"topics_raw": fact.get("topics", [])}
    if unresolved:
        meta["unresolved_topics"] = unresolved       # nothing silently lost
    if dropped_models:
        meta["dropped_model_topics"] = dropped_models  # scribe-as-topic, gated but recorded
    if fact.get("absorbed"):
        meta["absorbed_outcomes"] = fact["absorbed"]   # folded outcome clauses, verbatim quotes kept
    if fact.get("replaces"):
        meta["replaces_text"] = fact["replaces"]
    if extra_meta:
        meta.update(extra_meta)                        # caller-supplied provenance (e.g. trust:external)
    row = {
        "statement": fact["statement"],
        "kind": fact["kind"],
        "tags": tags,
        "verdict": "active",
        "evidence_quote": fact["quote"],             # verbatim, guaranteed by the extractor
        "evidence_source_type": source_type,
        "evidence_source_ref": source_ref,
        "evidence_captured_by": captured_by,
        "embedding": emb,
        "meta": meta,
        **fields,
    }
    return row


def remember(store: NewBrainStore, text: str, *, source_type: str = "utterance",
             source_ref: str | None = None, captured_by: str = "opus-4.8",
             extra_meta: dict | None = None) -> dict:
    """Run one capture through the write path. Returns a structured result: facts written,
    auto-resolutions (lanes 1/2), conflict interrupts (lane 3), documents, unresolved tags.

    extra_meta (optional): provenance merged into EVERY fact this capture mints — used by the
    acquisition ladder to stamp trust:external + tier + re-verify cadence on L2 write-backs."""
    result = {"written": [], "auto_resolved": [], "conflicts": [],
              "documents": [], "unresolved_tags": []}

    facts = extract_facts(text)
    if not facts:
        # untypeable or index-gated capture: its ONLY home is the documents layer (stored once here)
        doc_id = store.insert_document(text, source_ref, embed_one(text),
                                       {"reason": "no typed facts extracted"})
        result["documents"].append(doc_id)
        return result

    # LOSSLESS FLOOR (Aaron 2026-07-03): a capture that yields facts ALSO lands whole in the
    # documents layer. Anchor hygiene can drop a true-but-unquotable detail from the facts (the
    # field-case's "576 == 576" clause), so keeping the raw capture makes recall failures
    # RE-MINABLE for live captures, not just migrated ones. Stored exactly ONCE per capture: the
    # not-facts branch above already documents untypeable/index-gated input and returns, so this
    # line only runs when it hasn't — never double-storing.
    doc_id = store.insert_document(text, source_ref, embed_one(text),
                                   {"reason": "raw-capture floor; capture also yielded typed facts"})
    result["documents"].append(doc_id)

    for fact in facts:
        write_fact(store, fact, result, source_type=source_type,
                   source_ref=source_ref, captured_by=captured_by, extra_meta=extra_meta)
    return result


def write_fact(store: NewBrainStore, fact: dict, result: dict, *,
               source_type: str = "utterance", source_ref: str | None = None,
               captured_by: str = "opus-4.8", extra_meta: dict | None = None) -> None:
    """Tag -> density conflict check -> triage -> write, for ONE extracted fact. Split out from
    remember() so the conflict lanes can be exercised deterministically (no LLM in the loop)."""
    topics, dropped_models = _gate_topics(fact.get("topics", []))
    tags, unresolved = store.resolve_tags(topics)
    # surface only entity-shaped unresolved topics as candidate roots, deduped across the
    # capture; lowercase content-word noise stays in meta.topics_raw, never reaches Aaron
    seen = {u.lower() for u in result["unresolved_tags"]}
    for u in unresolved:
        if _entity_shaped(u) and u.lower() not in seen:
            result["unresolved_tags"].append(u)
            seen.add(u.lower())
    emb = embed_one(fact["statement"])
    cand = _candidate(fact, store.knn_facts(emb, k=8, tags=tags or None))
    row = _fact_row(fact, tags, emb, source_type, source_ref, captured_by, unresolved,
                    dropped_models, extra_meta)

    # no near-hit -> straight write
    if cand is None:
        new_id = store.insert_fact(row)
        result["written"].append({"id": new_id, "kind": fact["kind"],
                                  "statement": fact["statement"]})
        return

    # LANE ROUTING on a conflict candidate
    # RULE exception FIRST: RULE never auto-resolves, explicit replacement or not (Aaron override)
    if fact["kind"] == "RULE":
        new_id = store.insert_fact(row)
        cid = store.open_conflict(cand["id"], new_id)
        result["conflicts"].append(_interrupt(store, cid, cand["id"], new_id,
                                              "RULE-kind: resolution round mandatory (never auto-supersedes)"))
        return

    # lane 1: explicit supersession stated
    if fact.get("replaces"):
        new_id = store.insert_fact(row)
        store.supersede(cand["id"], new_id, "explicit supersession stated in capture")
        result["auto_resolved"].append({"lane": "explicit-supersession", "new_id": new_id,
                                        "superseded": cand["id"], "kind": fact["kind"],
                                        "statement": fact["statement"]})
        return

    # lane 2: mechanically verifiable (path ground truth)
    gt = _mechanical_resolution(fact["statement"], cand["statement"]) \
        if fact["kind"] in ("STATE", "REFERENCE") else None
    if gt is not None:
        new_id = store.insert_fact(row)
        if gt == "new":
            store.supersede(cand["id"], new_id, "ground truth: new fact's path resolves, old's does not")
            winner, loser = new_id, cand["id"]
        else:
            store.supersede(new_id, cand["id"], "ground truth: existing fact's path resolves, new's does not")
            winner, loser = cand["id"], new_id
        result["auto_resolved"].append({"lane": "ground-truth", "winner": winner,
                                        "superseded": loser, "kind": fact["kind"],
                                        "statement": fact["statement"]})
        return

    # lane 3: genuinely contested -> both facts frozen contested, INTERRUPT
    new_id = store.insert_fact(row)
    cid = store.open_conflict(cand["id"], new_id)
    result["conflicts"].append(_interrupt(store, cid, cand["id"], new_id,
                                          "contested: no explicit supersession, not mechanically resolvable"))


def _interrupt(store: NewBrainStore, conflict_id: int, existing_id: int, incoming_id: int,
               reason: str) -> dict:
    """The head-to-head shape the session surfaces so Aaron can resolve it while context is warm."""
    ex, inc = store.get_fact(existing_id), store.get_fact(incoming_id)
    def _side(f):
        return {"id": f["id"], "statement": f["statement"], "kind": f["kind"],
                "evidence_quote": f["evidence_quote"], "t_captured": str(f["t_captured"])}
    return {"type": "conflict_interrupt", "conflict_id": conflict_id, "reason": reason,
            "head_to_head": {"existing": _side(ex), "incoming": _side(inc)},
            "resolve_with": f"resolve_conflict({conflict_id}, <winner_id>, '<reasoning>')"}


if __name__ == "__main__":
    import json
    store = NewBrainStore()
    r = remember(store, "Aaron's favorite planet is Naboo, and he dislikes Tatooine.",
                 source_type="utterance", source_ref="smoke-test")
    print(json.dumps(r, indent=2), file=sys.stderr)
    store.close()