research/user_journal.py

"""Per-user journalist's notebook (deterministic suit memory).

The tiny model cannot rederive "who the user is and what they care about" from
weights. So the suit keeps a small, auditable USER JOURNAL: focus topics, active
investigation threads, remembered corrections, and a tone preset. The journal is
injected as a short context prefix on each analysis turn, and is updated with a
hashed-simple key + rationalized text / questions. This is memory in the suit,
not in the brain; when the user returns days later, the model "still knows them"
like a good journalist knows their subject.

Usage:
    from research.user_journal import UserJournal
    j = UserJournal()            # loads ./data/user_journal.json (creates default)
    ctx = j.context()            # compact prompt-prefix string
    j.note_thread(text)          # remember this turn as an active thread
    j.note_fact(fact)            # pin a fact/correction the user cares about
"""
import json
import time
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]

# Editorial tone presets -> the single line we hand the model.
TONES = {
    "spock": "Tone: strictly logical, evidence-first, concise; state what is "
             "unsupported instead of guessing.",  # default
    "journalist": "Tone: probe the question; separate asserted fact from "
                  "speculation; ask what source the user already trusts.",
    "coach": "Tone: explain briefly and support the user's own reasoning, "
             "correcting only where evidence demands.",
    "concise": "Tone: compact and direct, no filler.",
}
DEFAULT_TONE = "spock"
MAX_THREADS = 12
MAX_FACTS = 12


_default = {
    "handle": "guest",
    "tone": DEFAULT_TONE,
    "focus": [],
    "threads": [],
    "facts": [],
    "notes": "",
    "corrections": [],
    "last_seen": "",
}


class UserJournal:
    def __init__(self, path=None):
        self.path = Path(path) if path else ROOT / "data" / "user_journal.json"
        self.data = dict(_default)
        if self.path.exists():
            try:
                import json
                self.data.update(json.loads(self.path.read_text()))
            except Exception:
                pass

    def save(self):
        import json, time
        self.data["last_seen"] = time.strftime("%Y-%m-%d %H:%M")
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self.path.write_text(json.dumps(self.data, indent=2, ensure_ascii=False))

    # ---- reads ----
    def context(self):
        d = self.data
        lines = ["\nJOURNAL (about the user, journal's private notes):"]
        lines.append("handle: " + str(d.get("handle", "guest")))
        if d.get("tone"):
            lines.append(TONES.get(d["tone"], TONES[DEFAULT_TONE]))
        if d.get("threads"):
            lines.append("active threads: " + "; ".join(t if isinstance(t, str) else t.get("title","") for t in d["threads"][-MAX_THREADS:]))
        if d.get("facts"):
            lines.append("notes: " + " | ".join(str(f)[:140] for f in d["facts"][-MAX_FACTS:]))
        if d.get("corrections"):
            lines.append("remembered corrections: " + " | ".join(c[:140] for c in d["corrections"][-6:]))
        lines.append("These are private notes. Use them to be relevant to THIS user, "
                     "but do not state them back verbatim.\n")
        return "\n".join(lines)

    # ---- writes (suit heuristics) ----
    def set_handle(self, name):
        self.data["handle"] = (name or "guest").strip()

    def set_tone(self, preset):
        if preset in TONES:
            self.data["tone"] = preset

    def note_thread(self, text):
        title = " ".join((text or "").split()[:12])
        if not title:
            return
        threads = [t for t in self.data.setdefault("threads", []) if not (isinstance(t,str) and t==title)]
        threads.append(title)
        self.data["threads"] = threads[-MAX_THREADS:]

    def note_fact(self, fact):
        fact = (fact or "").strip()
        if not fact:
            return
        self.data.setdefault("facts", []).append(fact)
        self.data["facts"] = self.data["facts"][-MAX_FACTS:]

    def note_focus(self, terms):
        for t in (terms or []):
            t = str(t).strip()
            if t and t not in self.data.setdefault("focus", []):
                self.data["focus"].append(t)
        self.data["focus"] = self.data["focus"][-16:]

    def remember_correction(self, text):
        # A safety-corpus: if the user explicitly corrects us, keep it short.
        low = text.lower()
        if any(k in low for k in ("you're wrong", "that's wrong", "no, ", "correction", "actually ")):
            self.data.setdefault("corrections", []).append(text[:160])

    def snapshot(self):
        return dict(self.data)


if __name__ == "__main__":
    import sys
    j = UserJournal()
    print(j.context())
    print("---")
    j.note_thread("wanted to verify the 2022 repaint permit narrative")
    j.note_fact("user cares about timeline provenance across agencies")
    print("after write snapshot keys:", sorted(j.snapshot().keys()))