new_brain_server.py

"""New Brain — cartridge #3 as a local MCP server (FastMCP, stdio).

The MEMORY organ of the coprocessor family: a fact store with a metabolism (verdict, provenance,
lifecycle) — distinct from Open Brain, which stores what was SAID. This process is the ONLY thing
that touches the `new_brain` database (spec §1 seam convention).

HARD RULE (MCP stdio): nothing prints to stdout — stdout is the JSONRPC wire. All diagnostics in
the project code already route to stderr; keep it that way.

Needs the DB password in the env (NB_DB_PASSWORD or DB_PASSWORD) — supplied at registration, the
same way Open Brain receives it.

Run standalone (stdio):  DB_PASSWORD=... .venv/Scripts/python.exe new_brain_server.py
"""

from __future__ import annotations

from mcp.server.fastmcp import FastMCP

from new_brain_read import recall as _recall
from new_brain_store import NewBrainStore
from new_brain_write import remember as _remember

mcp = FastMCP("new-brain")
STORE = NewBrainStore()          # raises at startup if the DB password is absent — fail loud, early


@mcp.tool()
def describe() -> dict:
    """What the New Brain is and what it currently holds. The MEMORY organ: every fact carries a
    verdict, provenance, and lifecycle — it stores what is KNOWN, not merely what was said."""
    return {
        "id": "new-brain",
        "role": "fact store with a metabolism (verdict + provenance + lifecycle)",
        "kinds": ["DECISION", "RULE", "STATE", "PREFERENCE", "REFERENCE", "RESIDUAL", "EVENT"],
        "write": "remember(text) — extract typed facts, tag, conflict-check, write with evidence",
        "read": "recall(query) — trinary retrieval: each fact labeled verdict+age+contested",
        "contents": STORE.stats(),
    }


@mcp.tool()
def remember(text: str, source_ref: str | None = None) -> dict:
    """Run a capture through the write path: extract typed facts (seven-kind personal schema),
    translate topics to registry tags, run the SYNCHRONOUS conflict-density check, and write each
    fact with its verdict, verbatim-quote evidence, and temporal frame. Untypeable input is stored
    as a raw document, never forced into a fact.

    Returns what was written, what auto-resolved (explicit-supersession / ground-truth lanes), and
    any CONFLICT INTERRUPTS — contested head-to-heads to resolve now, while context is warm, via
    resolve_conflict(). RULE-kind conflicts ALWAYS interrupt (never auto-supersede)."""
    return _remember(STORE, text, source_type="utterance", source_ref=source_ref,
                     captured_by="opus-4.8")


@mcp.tool()
def recall(query: str, k: int = 8) -> dict:
    """Retrieve facts relevant to a query, EACH honestly labeled: verdict, age, verified-since, and
    contested partner if any. An expired STATE serves as "last known, unverified since <date>"; an
    EVENT never serves as a present-tense claim; contested facts serve both sides. When the fact
    layer is thin, falls through to the documents layer (labeled unverified) and says which layer
    answered."""
    return _recall(STORE, query, k=k)


@mcp.tool()
def resolve_conflict(conflict_id: int, winner_id: int, reasoning: str) -> dict:
    """Close a contested head-to-head: the winner stays active, the loser is superseded WITH the
    resolution reasoning attached (supersession is structural, not prose). winner_id must be one of
    the two facts in the conflict."""
    return STORE.resolve_conflict(conflict_id, winner_id, reasoning)


@mcp.tool()
def resolve_both_stand(conflict_id: int, reasoning: str) -> dict:
    """Close a contested head-to-head as a NON-conflict: the density check false-fired on
    COMPLEMENTARY facts that are both true and do not actually contradict. BOTH facts return to
    active and NEITHER is superseded. Use this — not resolve_conflict — when the interrupt's two
    facts can both stand (e.g. fragments of one inventory sentence the extractor shredded). For a
    genuine either/or head-to-head, use resolve_conflict."""
    return STORE.resolve_both_stand(conflict_id, reasoning)


@mcp.tool()
def retire_fact(fact_id: int, superseded_by_id: int, reasoning: str) -> dict:
    """Manually retire a known-false fact the conflict-density check MISSED, superseding it with
    the fact that corrects it. Same structural supersession as resolve_conflict (loser marked
    superseded_by the winner, reasoning attached, NO deletion) — for the case where no conflict
    ever opened, so no resolution tool could reach it. reasoning is mandatory; an already-superseded
    fact cannot be retired again; a fact cannot supersede itself."""
    return STORE.retire_fact(fact_id, superseded_by_id, reasoning)


@mcp.tool()
def open_conflicts() -> dict:
    """List the contested pairs still awaiting resolution."""
    return {"open": STORE.open_conflicts()}


@mcp.tool()
def registry_query(prefix: str = "") -> dict:
    """List registry tags, optionally under a dotted prefix (e.g. 'Project.RiftSuite'). Tags are
    machine-owned; Aaron speaks naturally and the write path translates."""
    return {"tags": STORE.registry_query(prefix)}


@mcp.tool()
def registry_add_leaf(tag: str, description: str = "") -> dict:
    """Mint a LEAF tag under an EXISTING parent, e.g. 'Project.NewBrain.Phase2' under
    'Project.NewBrain'. Leaves extend an existing branch and are NOT Aaron-gated (unlike roots via
    registry_propose_root). Refuses a bare root or a tag whose dotted parent doesn't exist yet."""
    return STORE.add_leaf(tag, description or None)


@mcp.tool()
def registry_propose_root(root: str, description: str = "") -> dict:
    """Add a new ROOT tag to the registry. Per governance, new roots are Aaron's call (leaves
    auto-mint under existing parents; roots do not) — this tool records his approval."""
    STORE.add_root(root, description or None)
    return {"added_root": root, "note": "new root registered (Aaron-gated action)"}


if __name__ == "__main__":
    mcp.run()