"""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: chat with the current persona /case SOP-driven analysis (verdict + confidence + skeptic) /skeptic skeptic attack on a claim /search search the library/corpus index /read read a document from the index /web search the live web (news/archive/wiki) /fetch pull one URL into the library /pull 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 /rag on|off retrieval-grounded chat (auto search + context) /memory on|off multi-turn conversation memory /mem on|off|stats| DNA-helix long-term memory (prior-case recall) /agent run the procedural agent loop (search/read/verdict) /agents parallel research swarm: 4 agents, merged report /synth [title] render case ledger into saved research documents /chart |:|... series chart + saved artifact (cross-synthesis) /opinion dual-mind fusion: the model's calibrated opinion /research research-partner loop: pull, verify, opine, reply /journal [name] full journalism suite -> CaseFile notebook /gaps list the case's open questions (gap ledger) /save save this case /load 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()