""" chat_repl.py — interactive terminal chat for testing the pipeline locally. Usage: python chat_repl.py # defaults to Socrates python chat_repl.py --character diogenes # switch character python chat_repl.py --user # use a real user_id from Supabase Type /reset to clear the short-term history for this session Type /switch to swap between socrates and diogenes Type /quit to exit """ import argparse import sys from pathlib import Path # Load .env if present (for local testing outside Docker) try: from dotenv import load_dotenv load_dotenv(Path(__file__).parent / ".env") except ImportError: pass # dotenv not installed — rely on env vars already being set from app_nn import run_chat_app from db_5_process_session import _save_history, _load_history from pipeline_proactive import run_proactive_pipeline # ── defaults ──────────────────────────────────────────────────────────────── DEFAULT_USER_ID = "18b4ef7c-8f18-4fcf-98f2-eaa6d5485a54" DEFAULT_USERNAME = "lollo632" # email prefix — must match the FAISS path (user_lollo632/db5) DEFAULT_PROFILE = {"language": "en", "cross_character_opinions": True} parser = argparse.ArgumentParser() parser.add_argument("--character", default="socrates") parser.add_argument("--user", default=DEFAULT_USER_ID) parser.add_argument("--username", default=DEFAULT_USERNAME) args = parser.parse_args() character_id = args.character user_id = args.user DEFAULT_USERNAME = args.username def clear_history(): empty = {"sessions": []} _save_history("chat_history_total", empty, user_id) _save_history("chat_history_short", empty, user_id) print(f"[history cleared for {user_id}]\n") print(f"\n=== Socrates Chat REPL ===") print(f"character : {character_id}") print(f"user_id : {user_id}") print(f"commands : /reset /switch /proactive /quit\n") while True: try: user_input = input("You: ").strip() except (EOFError, KeyboardInterrupt): print("\nBye.") sys.exit(0) if not user_input: continue if user_input == "/quit": sys.exit(0) if user_input == "/reset": clear_history() continue if user_input == "/switch": character_id = "diogenes" if character_id == "socrates" else "socrates" print(f"[switched to {character_id}]\n") continue # @mention switch: "@diogenes ..." or "@socrates ..." switches character mid-conversation character_explicit = False if user_input.startswith("@"): parts = user_input.split(None, 1) mentioned = parts[0][1:].lower() # strip the @ from characters import CHARACTERS if mentioned in CHARACTERS: if mentioned != character_id: character_id = mentioned print(f"[switched to {character_id}]\n") character_explicit = True user_input = parts[1] if len(parts) > 1 else "" if not user_input: continue # if mention not recognised, fall through and send as-is if user_input == "/proactive": print("[proactive] Firing pipeline (force=True, bypasses activity gate)...") result = run_proactive_pipeline(user_id, DEFAULT_USERNAME, force=True) if result: print(f"\n{character_id.capitalize()} (proactive): {result['message']}\n") _img = result.get("image_url") _exp = result.get("image_expires_at", "")[:10] if result.get("image_expires_at") else "-" print(f"[type={result['type']} tone={result['tone']} image={'yes (expires '+_exp+')' if _img else 'none'}]\n") if _img: print(f"[image_url] {_img}\n") # Now let you reply to it so the full chat pipeline is tested too try: followup = input("Your reply (or press Enter to skip): ").strip() except (EOFError, KeyboardInterrupt): print("\nBye.") sys.exit(0) if followup and not followup.startswith("/"): reply = run_chat_app( user_id=user_id, username=DEFAULT_USERNAME, profile=DEFAULT_PROFILE, ui_lang="en", user_msg=followup, character_id=character_id, ) if isinstance(reply, dict): text = reply.get("reply") or str(reply) # Sync REPL character to whoever actually responded character_id = reply.get("character_id_used") or character_id else: text = str(reply) print(f"\n{character_id.capitalize()}: {text}\n") else: print("[proactive] Pipeline returned nothing — check logs above.\n") continue reply = run_chat_app( user_id=user_id, username=DEFAULT_USERNAME, profile=DEFAULT_PROFILE, ui_lang="en", user_msg=user_input, character_id=character_id, character_explicit=character_explicit, ) # reply can be a plain string or a dict with a "reply" key if isinstance(reply, dict): text = reply.get("reply") or str(reply) character_id = reply.get("character_id_used") or character_id else: text = str(reply) print(f"\n{character_id.capitalize()}: {text}\n") # ── Coordinator recommendation: auto-switch character ───────────────── if isinstance(reply, dict): rec = reply.get("recommended_character") if rec: print(f" [debug] recommended_character={rec!r}") if rec and rec != character_id: from characters import CHARACTERS rec_name = CHARACTERS.get(rec, {}).get("display_name", rec.capitalize()) character_id = rec character_explicit = True print(f" [switching to {rec_name} — next message goes to them]\n") # ── Cross-character knock ────────────────────────────────────────────── knock = reply.get("cross_character_knock") if isinstance(reply, dict) else None if knock and knock.get("character_id"): guest_name = knock.get("character_display_name", knock["character_id"].capitalize()) print(f" ✋ {guest_name} wants to add something.") if knock.get("bridge"): print(f" \"{knock['bridge'][:120]}\"") print(f" [1] Hear it [2] Not now [3] Never again on this topic") try: choice_raw = input(" Your choice (1/2/3 or Enter to skip): ").strip() except (EOFError, KeyboardInterrupt): choice_raw = "" if choice_raw == "1": from util_cross_character import generate_guest_opinion guest_reply = generate_guest_opinion( guest_character_id=knock["character_id"], relation=knock.get("relation") or "", bridge=knock.get("bridge") or "", host_reply=text, user_msg=user_input, reply_language="en", ) print(f"\n{guest_name}: {guest_reply}\n") elif choice_raw == "3": from util_cross_character import add_session_exclusion add_session_exclusion(user_id, knock["character_id"], knock.get("topic_key", user_input[:120])) print(f" [exclusion added: {guest_name} on this topic]\n") else: print() # decline — silent