tui/cli.py

"""TinyLiquid Analyst - headless terminal client (also scripts/SSH-friendly).

Usage:
  .venv/bin/python tui/cli.py --ckpt ckpt/dpo                     # interactive REPL
  .venv/bin/python tui/cli.py --ckpt ckpt/dpo --once "Verify: ..."
  echo "Verify: ..." | .venv/bin/python tui/cli.py --ckpt ckpt/dpo
"""

import argparse
import sys

from tui.engine import AnalystEngine, parse_cmd, BANNER

HELP = """commands:
  <text>                     chat with the current persona
  /case <sop> <text>         SOP-driven analysis (verdict + confidence + skeptic)
  /skeptic <text>            skeptic attack on a claim
  /search <query>            search the library/corpus index
  /read <key>                read a document from the index
  /web <topic>               search the live web (news/archive/wiki)
  /fetch <url>               pull one URL into the library
  /pull <topic>              search + pull top docs into the library
  /tor on|off|status         route .onion fetches through local Tor SOCKS
  /sop [name]                show one or all procedures
  /persona <analyst|skeptic|none>
  /rag on|off                retrieval-grounded chat (auto search + context)
  /memory on|off             multi-turn conversation memory
  /mem on|off|stats|<claim>  DNA-helix long-term memory (prior-case recall)
  /agent <task>              run the procedural agent loop (search/read/verdict)
  /agents <task>             parallel research swarm: 4 agents, merged report
  /synth [title]             render case ledger into saved research documents
  /chart <t>|<l>:<v,..>|...  series chart + saved artifact (cross-synthesis)
  /opinion <claim>           dual-mind fusion: the model's calibrated opinion
  /research <question>       research-partner loop: pull, verify, opine, reply
  /journal [name]           full journalism suite -> CaseFile notebook
  /gaps                      list the case's open questions (gap ledger)
  /save <name>               save this case
  /load <name>               load a case
  /status                    show state
  /help                      this help
  /quit                      exit"""


def run_once(eng, text):
    if not text.strip():
        return
    if text.startswith("/case"):
        rest = text[5:].strip()
        sop = ""
        if rest and " " in rest[:60]:
            maybe, body = rest.split(maxsplit=1)
            if maybe in eng.__class__.__module__:  # noqa - simple guard
                pass
            sop = maybe
            text = body
        elif rest:
            text = rest
        rep = eng.analyze(text, sop or None)
        print(json_dump(rep))
    elif text.startswith("/"):
        print(handle_cmd(eng, text))
    else:
        print(eng.chat(text))


def json_dump(rep):
    import json
    return json.dumps(rep, indent=2, ensure_ascii=False)


def handle_cmd(eng, line):
    cmd, arg = parse_cmd(line) or ("chat", line)
    if cmd == "chat":
        return eng.chat(arg)
    if cmd == "case":
        sop = ""
        body = arg
        if " " in body:
            maybe, rest = body.split(maxsplit=1)
            if maybe in {"claim_verification", "cross_source_discrepancy", "pattern_finding",
                         "timeline_reconstruction", "historical_truth", "politics_analysis",
                         "dark_web_research", "terminal_control", "source_triage"}:
                sop, body = maybe, rest
        return json_dump(eng.analyze(body, sop or None))
    if cmd == "skeptic":
        return eng.generate(arg, persona="skeptic")
    if cmd == "search":
        return "\n".join(f"{k} (score {s:.2f})" for k, s in eng.search(arg)) or "(no hits)"
    if cmd == "read":
        return eng.read(arg)
    if cmd == "sop":
        return eng.sop_text(arg) if arg else "\n".join(list_sops_str())
    if cmd == "persona":
        if arg in eng.PERSONA_ID if hasattr(eng, "PERSONA_ID") else arg in ("analyst", "skeptic", "none"):
            eng.persona = arg
            return f"persona -> {arg}"
        return "persona: analyst|skeptic|none"
    if cmd == "save":
        return "saved " + eng.save_case(arg or "default")
    if cmd == "load":
        return eng.load_case(arg or "default")
    if cmd == "rag":
        eng.rag = arg not in ("off", "0", "false")
        return f"retrieval-grounded chat: {'on' if eng.rag else 'off'}"
    if cmd == "memory":
        eng.memory = arg not in ("off", "0", "false")
        return f"multi-turn memory: {'on' if eng.memory else 'off'}"
    if cmd == "mem":
        if arg in ("on", "1", "true", "yes"):
            eng.helix_on = True
            return "long-term helix memory: on"
        if arg in ("off", "0", "false", "no"):
            eng.helix_on = False
            return "long-term helix memory: off"
        if arg == "stats":
            return json_dump(eng.mem_stats())
        return eng.recall(arg)
    if cmd == "agent":
        return json_dump(eng.agent(arg))
    if cmd == "agents":
        return json_dump(eng.agents(arg, n=4))
    if cmd == "synth":
        return eng.synthesize(arg or None)
    if cmd == "chart":
        return eng.chart(arg)
    if cmd == "opinion":
        return json_dump(eng.opinion(arg))
    if cmd == "research":
        return json_dump(eng.research(arg))
    if cmd == "journal":
        return eng.journal(arg or None)
    if cmd == "gaps":
        return "\n".join(f"  {g}" for g in eng.gaps())
    if cmd == "web":
        hits = eng.web_search(arg, n=8)
        if not hits:
            return "(no web hits)"
        return "\n".join("[{0}] {1}\n   {2}\n   {3}".format(
            r["source"], r["title"], r["url"], r["snippet"][:100]) for r in hits)
    if cmd == "fetch":
        return json_dump(eng.web_fetch(arg))
    if cmd == "pull":
        return json_dump(eng.web_pull(arg, n=3))
    if cmd == "tor":
        if arg in ("on", "1", "true", "yes"):
            eng.tor = True
            return "tor: on"
        if arg in ("off", "0", "false", "no"):
            eng.tor = False
            return "tor: off"
        return "tor: {0} | {1}".format(eng.tor, eng.tor_status_msg())
    if cmd == "status":
        return (f"persona={eng.persona} memory={'on' if eng.memory else 'off'} "
                f"helix={'on' if getattr(eng, 'helix_on', False) else 'off'} "
                f"rag={'on' if eng.rag else 'off'} case={eng.case.name} "
                f"msgs={len(eng.case.chat)} notes={len(eng.case.ledger)} ckpt={eng.ckpt}")
    if cmd in ("help", "?"):
        return HELP
    if cmd in ("quit", "exit"):
        raise SystemExit(0)
    return f"unknown command /{cmd}; try /help"


def list_sops_str():
    from research.agent import SOP_ALIASES
    return [f"  {k:28s} {', '.join(SOP_ALIASES[k][:3])}" for k in SOP_ALIASES]


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--ckpt", default="ckpt/v8_lora/best.pt")
    ap.add_argument("--tok", default="data/tokenizer.json")
    ap.add_argument("--once", default=None)
    ap.add_argument("--threads", type=int, default=4)
    args = ap.parse_args()

    eng = AnalystEngine(ckpt=args.ckpt, tok_path=args.tok, threads=args.threads)
    print(BANNER, flush=True)
    print(f"model: {eng.ckpt} | persona: analyst | type /help\n", flush=True)

    if args.once:
        run_once(eng, args.once)
        return
    if not sys.stdin.isatty():
        for line in sys.stdin:
            line = line.strip()
            if not line:
                continue
            if line in ("/quit", "/exit"):
                break
            print(handle_cmd(eng, line), flush=True)
        return
    for line in sys.stdin:
        line = line.strip()
        if not line:
            continue
        try:
            print(handle_cmd(eng, line), flush=True)
        except SystemExit:
            break


if __name__ == "__main__":
    main()