#!/usr/bin/env python3 """ characterif.py - Heichalot two-node Tailcat bootstrap/link prototype. v0.3a idea (v0.2 transport preserved; display/API entry-points added): NODE A: python3 tools/characterif.py start - starts the local Flask API - starts `tailcat --json serve ` - obtains A's Tailcat address - gives a tiny JSON identity document to `croc send --text` - prints croc's human-manageable pairing code - waits as a server NODE B: python3 tools/characterif.py join - starts its own Flask API/Tailcat listener - receives A's identity JSON through croc - stores A locally - connects to A over Tailcat - POSTs B's own identity to A's /api/peer/register - stores the two-way relationship - remains running as a server After pairing, both machines are peers: each is both a Tailcat server and client. v0.3a also reserves/implements simple display-facing API entry-points for AI identity, standard representation, location manifest, and recent chat. These are deliberately minimal so their internals can be replaced later without changing the URLs. This is an internal prototype. The Flask API binds to the LAN by default. Do not expose it directly to an untrusted network without authentication. """ from __future__ import annotations import argparse import atexit import configparser import getpass import json import os import re import shutil import socket import subprocess import sys import threading import time import webbrowser from datetime import datetime, timezone from multiprocessing import shared_memory from pathlib import Path from typing import Any, Dict, Optional, List from urllib import error, request import tailcat try: from flask import Flask, jsonify, request as flask_request, send_file, render_template_string except ImportError: Flask = None jsonify = None flask_request = None send_file = None render_template_string = None APP_NAME = "characterif" PROTOCOL = "characterif/1" API_VERSION = 1 DEFAULT_BIND = "0.0.0.0" DEFAULT_PORT = 8765 DEFAULT_TIMEOUT = 15 BASE_DIR = Path(__file__).resolve().parent.parent CONFIG_PATH = Path(os.environ.get("CHARACTERIF_CONFIG", BASE_DIR / "characterif.conf")).expanduser() # Shared Heichalot-CMS platform paths live in src/config.py. SRC_DIR = BASE_DIR / "src" if str(SRC_DIR) not in sys.path: sys.path.insert(0, str(SRC_DIR)) from config import character_data_dir, platform_data_dir STATE_DIR = Path.home() / ".config" / APP_NAME REGISTRY_CACHE = STATE_DIR / "nodes.json" SESSION_LOG = STATE_DIR / "session.json" LOCATION_MANIFEST = BASE_DIR / "location-manifest.json" # Small in-memory display feed. This is deliberately not a durable queue. # It only gives the visual client something to poll until a later event/stream layer exists. RECENT_CHAT_LIMIT = 100 RECENT_CHAT: List[Dict[str, Any]] = [] RECENT_CHAT_LOCK = threading.Lock() # Ambiguous default portrait used when a character is configured without an image. DEFAULT_PORTRAIT_SVG = """\ """ CHILDREN = [] CURRENT_TAILCAT_ADDRESS: Optional[str] = None CURRENT_CROC_CODE: Optional[str] = None WAN_PAIRING_LOCK = threading.Lock() # Volatile character connection status shared between local processes. # CharacterIF is the intended writer; UIs and CMS processes are readers. PRESENCE_SHM_NAME = "heichalot_characterif_presence_v1" PRESENCE_SHM_SIZE = 64 * 1024 PRESENCE_HEADER_SIZE = 16 PRESENCE_WRITE_LOCK = threading.Lock() PRESENCE_SHM_OWNER: Optional[shared_memory.SharedMemory] = None MOBILE_LANDING_TEMPLATE = r""" CharacterIF

CharacterIF

Mobile LAN interface

Connected to {{ node }} on your local network.

Local web interface

Wifi (local LAN) Mobile Internet (WAN)

""" MOBILE_HOME_TEMPLATE = r""" CharacterIF - Home

Characters

Available on this CharacterIF node

Remote viewing

Open the discussion page and use the existing CharacterIF responder.

Open discussion

""" MOBILE_REMOTE_VIEW_TEMPLATE = r""" CharacterIF - Remote View

Choose a character and start the discussion.

""" MOBILE_MESSAGES_TEMPLATE = r""" CharacterIF - Messages

Waiting for messages from another CharacterIF node…

""" MOBILE_SETTINGS_TEMPLATE = r""" CharacterIF - Settings

Settings

Placeholder for mobile settings. Nothing here is wired yet.

""" def utcnow() -> str: return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") def ensure_state_dir() -> None: STATE_DIR.mkdir(parents=True, exist_ok=True) def load_config() -> configparser.ConfigParser: cfg = configparser.ConfigParser() if CONFIG_PATH.exists(): cfg.read(CONFIG_PATH) return cfg def cfg_get(cfg: configparser.ConfigParser, section: str, option: str, fallback=None): try: return cfg.get(section, option) except (configparser.NoSectionError, configparser.NoOptionError): return fallback def local_node_name(cfg: configparser.ConfigParser) -> str: name = cfg_get(cfg, "characterif", "node") if name: return name.strip() for section, option in ( ("node", "name"), ("heichalot", "node"), ("server", "node"), ): name = cfg_get(cfg, section, option) if name: return name.strip() return socket.gethostname().split(".")[0] def local_user_account_name(cfg: configparser.ConfigParser) -> str: """Resolve the human user account name with optional alias.""" alias = cfg_get(cfg, "characterif", "local_user_account_name") if alias: return alias.strip() return getpass.getuser() def local_ai_name(cfg: configparser.ConfigParser) -> str: """Human/AI identity used in simple chat envelopes.""" name = cfg_get(cfg, "characterif", "local_user_account_name") if name: return name.strip() return local_node_name(cfg) def character_section(cfg: configparser.ConfigParser, name: Optional[str] = None) -> str: """Return the config section for a character on this node.""" return f"character-{name or local_ai_name(cfg)}" def character_cfg_get(cfg: configparser.ConfigParser, option: str, fallback=None, name: Optional[str] = None): """Read a character-specific setting from [character-].""" return cfg_get(cfg, character_section(cfg, name), option, fallback) def _character_bool(cfg: configparser.ConfigParser, name: str, option: str, fallback: bool = False) -> bool: value = character_cfg_get(cfg, option, name=name) if value is None: return fallback return str(value).strip().lower() in {"1", "true", "yes", "on"} def character_document(cfg: configparser.ConfigParser, name: str) -> Optional[Dict[str, Any]]: """Return one configured character, or None if it does not exist.""" section = character_section(cfg, name) if not cfg.has_section(section): return None configured_name = character_cfg_get(cfg, "name", name, name=name).strip() character_type = character_cfg_get(cfg, "type", name=name) if character_type is None: character_type = "" character_type = character_type.strip().lower() character_type = "ai" if character_type == "ai" else "human" data_dir = character_cfg_get(cfg, "data_dir", "", name=name).strip() return { "name": configured_name, "character_type": character_type, "node": local_node_name(cfg), "state": "available", "portrait": f"/api/character/{configured_name}/portrait", "data_dir": data_dir or None, "available_remotely": _character_bool(cfg, name, "available_remotely"), "exists_remotely": _character_bool(cfg, name, "exists_remotely"), } def character_documents(cfg: configparser.ConfigParser) -> List[Dict[str, Any]]: """Return all characters configured in [character-] sections.""" result = [] for section in cfg.sections(): if not section.startswith("character-"): continue name = section[len("character-"):] doc = character_document(cfg, name) if doc is not None: result.append(doc) return result def remote_character_data_dir(name: str, node: str) -> Path: """Return the local cache directory for a character learned from another node.""" def slug(value: str) -> str: cleaned = re.sub(r"[^A-Za-z0-9._-]+", "-", value.strip()).strip("-._").lower() if not cleaned: raise ValueError("character/node name does not contain a usable directory name") return cleaned return platform_data_dir() / f"character-{slug(name)}@{slug(node)}" def cached_remote_character_documents() -> List[Dict[str, Any]]: """Load cached remote character metadata from the platform data directory.""" result = [] root = platform_data_dir() if not root.is_dir(): return result for data_dir in sorted(root.glob("character-*@*")): json_path = data_dir / "character.json" conf_path = data_dir / "character.conf" if not json_path.is_file() or not conf_path.is_file(): continue try: doc = json.loads(json_path.read_text(encoding="utf-8")) meta = configparser.ConfigParser() meta.read(conf_path, encoding="utf-8") if not isinstance(doc, dict): continue doc = dict(doc) doc["local"] = meta.getboolean("address", "local", fallback=False) doc["source_node"] = meta.get("address", "node", fallback="") or None doc["data_dir"] = str(data_dir.resolve()) result.append(doc) except Exception: continue return result def all_character_documents(cfg: configparser.ConfigParser) -> List[Dict[str, Any]]: """Return local characters followed by locally cached remote characters.""" local = [] for doc in character_documents(cfg): item = dict(doc) item["local"] = True local.append(item) return local + cached_remote_character_documents() def character_key(name: str, node: str) -> str: """Return the canonical distributed character identity.""" return f"{name}@{node}" def load_characters() -> Dict[str, Dict[str, Any]]: """Return all characters known to this installation, keyed by name@node.""" cfg = load_config() result: Dict[str, Dict[str, Any]] = {} for character in all_character_documents(cfg): name = str(character.get("name") or "").strip() node = str(character.get("source_node") or character.get("node") or "").strip() if not name or not node: continue key = character_key(name, node) result[key] = { "name": name, "character_type": character.get("character_type"), "node": node, "local": bool(character.get("local")), "portrait": character.get("portrait"), "data_dir": character.get("data_dir"), "available_remotely": bool(character.get("available_remotely")), "exists_remotely": bool(character.get("exists_remotely")), } return result def _presence_open_for_write() -> shared_memory.SharedMemory: """Create or attach to the local shared-memory presence block.""" global PRESENCE_SHM_OWNER if PRESENCE_SHM_OWNER is not None: return PRESENCE_SHM_OWNER try: shm = shared_memory.SharedMemory( name=PRESENCE_SHM_NAME, create=True, size=PRESENCE_SHM_SIZE, ) shm.buf[:] = b"\0" * PRESENCE_SHM_SIZE except FileExistsError: shm = shared_memory.SharedMemory(name=PRESENCE_SHM_NAME, create=False) PRESENCE_SHM_OWNER = shm return shm def _presence_read_from(shm: shared_memory.SharedMemory) -> Dict[str, Dict[str, Any]]: """Read one consistent JSON snapshot from the shared-memory block.""" for _ in range(5): sequence_before = int.from_bytes(shm.buf[0:8], "little") if sequence_before & 1: time.sleep(0) continue payload_length = int.from_bytes(shm.buf[8:16], "little") if payload_length == 0: return {} if payload_length > len(shm.buf) - PRESENCE_HEADER_SIZE: return {} payload = bytes( shm.buf[ PRESENCE_HEADER_SIZE: PRESENCE_HEADER_SIZE + payload_length ] ) sequence_after = int.from_bytes(shm.buf[0:8], "little") if sequence_before != sequence_after or (sequence_after & 1): time.sleep(0) continue try: value = json.loads(payload.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError): return {} return value if isinstance(value, dict) else {} return {} def character_statuses() -> Dict[str, Dict[str, Any]]: """ Return the complete last-known character connection-status snapshot. This is the cheap reader intended for UIs. If CharacterIF has not created the shared-memory block yet, an empty dictionary is returned. """ if PRESENCE_SHM_OWNER is not None: return _presence_read_from(PRESENCE_SHM_OWNER) try: shm = shared_memory.SharedMemory(name=PRESENCE_SHM_NAME, create=False) except FileNotFoundError: return {} try: return _presence_read_from(shm) finally: shm.close() def _write_character_statuses(statuses: Dict[str, Dict[str, Any]]) -> None: """Replace the complete shared-memory status snapshot.""" payload = json.dumps( statuses, separators=(",", ":"), sort_keys=True, ).encode("utf-8") if len(payload) > PRESENCE_SHM_SIZE - PRESENCE_HEADER_SIZE: raise RuntimeError("character status shared-memory block is full") with PRESENCE_WRITE_LOCK: shm = _presence_open_for_write() sequence = int.from_bytes(shm.buf[0:8], "little") if sequence & 1: sequence += 1 # Odd sequence = write in progress. Even sequence = stable snapshot. shm.buf[0:8] = (sequence + 1).to_bytes(8, "little") shm.buf[PRESENCE_HEADER_SIZE:PRESENCE_HEADER_SIZE + len(payload)] = payload shm.buf[8:16] = len(payload).to_bytes(8, "little") shm.buf[0:8] = (sequence + 2).to_bytes(8, "little") def set_character_status(name: str, node: str, status: str) -> Dict[str, Any]: """Set one character's last-known connection status.""" status = status.strip().lower() if status not in {"online", "offline"}: raise ValueError("status must be 'online' or 'offline'") name = name.strip() node = node.strip() if not name or not node: raise ValueError("character name and node are required") key = character_key(name, node) with PRESENCE_WRITE_LOCK: # Read while holding the writer lock, then write the replacement # snapshot without re-entering the same lock. if PRESENCE_SHM_OWNER is not None: statuses = _presence_read_from(PRESENCE_SHM_OWNER) else: try: existing = shared_memory.SharedMemory( name=PRESENCE_SHM_NAME, create=False, ) except FileNotFoundError: statuses = {} else: try: statuses = _presence_read_from(existing) finally: existing.close() statuses[key] = { "name": name, "node": node, "status": status, } payload = json.dumps( statuses, separators=(",", ":"), sort_keys=True, ).encode("utf-8") if len(payload) > PRESENCE_SHM_SIZE - PRESENCE_HEADER_SIZE: raise RuntimeError("character status shared-memory block is full") shm = _presence_open_for_write() sequence = int.from_bytes(shm.buf[0:8], "little") if sequence & 1: sequence += 1 shm.buf[0:8] = (sequence + 1).to_bytes(8, "little") shm.buf[PRESENCE_HEADER_SIZE:PRESENCE_HEADER_SIZE + len(payload)] = payload shm.buf[8:16] = len(payload).to_bytes(8, "little") shm.buf[0:8] = (sequence + 2).to_bytes(8, "little") return dict(statuses[key]) def get_character_status(name: str, node: str) -> Optional[Dict[str, Any]]: """Return one character's last-known status, or None if it has no entry.""" return character_statuses().get(character_key(name.strip(), node.strip())) def online_characters() -> Dict[str, Dict[str, Any]]: """Return only characters whose last-known status is online.""" return { key: value for key, value in character_statuses().items() if value.get("status") == "online" } def remotely_available_character_documents(cfg: configparser.ConfigParser) -> List[Dict[str, Any]]: """Return only local characters that this node explicitly advertises remotely.""" result = [] for doc in character_documents(cfg): if doc.get("available_remotely"): item = dict(doc) item["local"] = False item["exists_remotely"] = True result.append(item) return result def cache_remote_characters(peer: Dict[str, Any], payload: Dict[str, Any]) -> List[Path]: """Persist character metadata learned from one remote CharacterIF server.""" node = str(peer.get("node") or "").strip() if not node: raise RuntimeError("remote peer has no node name") characters = payload.get("characters") if not isinstance(characters, list): raise RuntimeError("remote server did not return a character list") written = [] for character in characters: if not isinstance(character, dict): continue name = str(character.get("name") or "").strip() if not name: continue data_dir = remote_character_data_dir(name, node) data_dir.mkdir(parents=True, exist_ok=True) (data_dir / "character.json").write_text( json.dumps(character, indent=2, sort_keys=True) + "\n", encoding="utf-8", ) meta = configparser.ConfigParser() meta["address"] = { "node": node, "local": "false", "api_port": str(peer.get("api_port", DEFAULT_PORT)), "tailcat_address": str(peer.get("address") or ""), "updated": utcnow(), } with (data_dir / "character.conf").open("w", encoding="utf-8") as handle: meta.write(handle) written.append(data_dir) return written def fetch_remote_characters(peer: Dict[str, Any]) -> Dict[str, Any]: """Fetch the public character index from one Tailcat peer.""" return tailcat.get_json( peer["address"], int(peer.get("api_port", DEFAULT_PORT)), "/", timeout=15, ) def sync_remote_characters(peer: Dict[str, Any]) -> List[Path]: payload = fetch_remote_characters(peer) if not payload.get("ok", True): raise RuntimeError("remote character index returned an error: " + json.dumps(payload)) return cache_remote_characters(peer, payload) def _delayed_sync_remote_characters(peer: Dict[str, Any]) -> None: """Best-effort reverse sync after a newly joined peer has started its Flask server.""" last_error = None for _ in range(8): time.sleep(1.0) try: written = sync_remote_characters(peer) print(f"\nCached {len(written)} remote character(s) from {peer.get('node')}") return except Exception as exc: last_error = exc print(f"\nRemote character sync from {peer.get('node')} deferred: {last_error}") def character_private_config_path(cfg: configparser.ConfigParser, name: str) -> Optional[Path]: """Return the private character.conf path for a local character.""" data_dir = character_cfg_get(cfg, "data_dir", "", name=name) if not data_dir: return None return Path(data_dir).expanduser() / "character.conf" def load_character_private_config(cfg: configparser.ConfigParser, name: str) -> configparser.ConfigParser: private = configparser.ConfigParser() path = character_private_config_path(cfg, name) if path is not None and path.is_file(): private.read(path, encoding="utf-8") return private def cached_remote_character(name: str) -> Optional[Dict[str, Any]]: """Return the first cached remote character with this name.""" for doc in cached_remote_character_documents(): if str(doc.get("name") or "").casefold() == name.casefold(): return doc return None def cached_remote_address(name: str) -> Optional[Dict[str, Any]]: """Load routing data from a cached remote character's character.conf.""" doc = cached_remote_character(name) if doc is None: return None data_dir = doc.get("data_dir") if not data_dir: return None meta = configparser.ConfigParser() conf_path = Path(str(data_dir)) / "character.conf" if not conf_path.is_file(): return None meta.read(conf_path, encoding="utf-8") if not meta.has_section("address"): return None return { "node": meta.get("address", "node", fallback=""), "address": meta.get("address", "tailcat_address", fallback=""), "api_port": meta.getint("address", "api_port", fallback=DEFAULT_PORT), } def respond_as_character(cfg: configparser.ConfigParser, name: str, text: str) -> Dict[str, Any]: """Route one synchronous prompt to a local or cached remote character.""" local_doc = character_document(cfg, name) if local_doc is not None: if local_doc.get("character_type") != "ai": raise RuntimeError(f"character not available: {name}") private = load_character_private_config(cfg, name) api = private.get("interface", "api", fallback="ollama") model = private.get("interface", "model", fallback="gemma3") # Lazy import keeps CharacterIF usable for routing-only installations. import responder response_text = responder.respond(text, api=api, model=model) return { "ok": True, "character": local_doc.get("name", name), "node": local_node_name(cfg), "local": True, "response": response_text, "time": utcnow(), } remote_doc = cached_remote_character(name) route = cached_remote_address(name) if remote_doc is None or route is None or not route.get("address"): raise RuntimeError(f"character not found or has no route: {name}") return tailcat.post_json( route["address"], int(route.get("api_port", DEFAULT_PORT)), f"/api/character/{name}/respond", {"text": text}, timeout=180, ) def local_ai_document(cfg: configparser.ConfigParser) -> Dict[str, Any]: """Return the configured/default local AI participant.""" name = local_ai_name(cfg) doc = character_document(cfg, name) if doc is not None and doc.get("character_type") == "ai": return doc return { "name": name, "character_type": "ai", "node": local_node_name(cfg), "state": "available", "portrait": f"/api/ai/{name}/portrait", } def portrait_path(cfg: configparser.ConfigParser, name: Optional[str] = None) -> Optional[Path]: """Resolve the configured portrait for a character.""" value = character_cfg_get(cfg, "portrait", name=name) if not value: return None path = Path(value).expanduser() if not path.is_absolute(): path = BASE_DIR / path return path def location_document() -> Dict[str, Any]: """ Load location-manifest.json if present; otherwise return the v1 stage shape. The fallback preserves the API contract before any artwork exists. """ if LOCATION_MANIFEST.exists(): try: data = json.loads(LOCATION_MANIFEST.read_text()) if isinstance(data, dict): return data except Exception: pass return { "location": "skylab-default", "dimensions": {"width": 12, "height": 4, "depth": 8}, "stage": { "background": None, "floor": None, "ceiling": None, "left_wall": None, "right_wall": None, }, "implemented": False, } def remember_chat_message(message: Dict[str, Any]) -> None: item = dict(message) item.setdefault("received_time", utcnow()) with RECENT_CHAT_LOCK: RECENT_CHAT.append(item) if len(RECENT_CHAT) > RECENT_CHAT_LIMIT: del RECENT_CHAT[:-RECENT_CHAT_LIMIT] def notify_chat_message(cfg: configparser.ConfigParser, sender: str, text: str) -> None: """Deliver an incoming chat to the desktop user as a cross-platform notification.""" try: from notifypy import Notify except ImportError: return notification = Notify(default_notification_application_name=APP_NAME) notification.title = f"Message from {sender}" notification.message = text icon = portrait_path(cfg, sender) if icon is not None and icon.is_file(): notification.icon = str(icon) try: notification.send() except Exception: pass def api_port(cfg: configparser.ConfigParser) -> int: return int(cfg_get(cfg, "characterif", "api_port", DEFAULT_PORT)) def lan_ip_address() -> str: """Best-effort IPv4 address suitable for another device on the local LAN.""" sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) try: # UDP connect selects the outbound interface without sending application data. sock.connect(("8.8.8.8", 80)) address = sock.getsockname()[0] if address and not address.startswith("127."): return address except OSError: pass finally: sock.close() try: for address in socket.gethostbyname_ex(socket.gethostname())[2]: if address and not address.startswith("127."): return address except OSError: pass return "127.0.0.1" def mobile_public_url(port: int) -> str: return f"http://{lan_ip_address()}:{port}/mobile" def daemon_url(cfg: configparser.ConfigParser) -> str: bind = character_cfg_get(cfg, "api_host", DEFAULT_BIND) if bind in ("0.0.0.0", "::"): bind = "127.0.0.1" return f"http://{bind}:{api_port(cfg)}" def load_registry() -> Dict[str, Any]: ensure_state_dir() if not REGISTRY_CACHE.exists(): return {"version": API_VERSION, "nodes": {}} try: data = json.loads(REGISTRY_CACHE.read_text()) data.setdefault("version", API_VERSION) data.setdefault("nodes", {}) return data except Exception: return {"version": API_VERSION, "nodes": {}} def save_registry(data: Dict[str, Any]) -> None: ensure_state_dir() tmp = REGISTRY_CACHE.with_suffix(".tmp") tmp.write_text(json.dumps(data, indent=2, sort_keys=True) + "\n") tmp.replace(REGISTRY_CACHE) def save_peer(doc: Dict[str, Any]) -> None: reg = load_registry() doc = dict(doc) doc["updated"] = utcnow() reg["nodes"][doc["node"]] = doc save_registry(reg) def get_peer(node: str) -> Optional[Dict[str, Any]]: return load_registry().get("nodes", {}).get(node) def run_cmd(args, *, timeout=DEFAULT_TIMEOUT, input_text=None, env=None) -> Dict[str, Any]: started = time.time() try: cp = subprocess.run( args, input=input_text, text=True, capture_output=True, timeout=timeout, env=env, ) return { "ok": cp.returncode == 0, "exit_code": cp.returncode, "stdout": cp.stdout, "stderr": cp.stderr, "elapsed": round(time.time() - started, 3), "argv": args, } except subprocess.TimeoutExpired as exc: return { "ok": False, "exit_code": None, "stdout": exc.stdout or "", "stderr": exc.stderr or "", "elapsed": round(time.time() - started, 3), "error": f"timeout after {timeout}s", "argv": args, } except FileNotFoundError: return { "ok": False, "exit_code": None, "stdout": "", "stderr": "", "elapsed": round(time.time() - started, 3), "error": f"command not found: {args[0]}", "argv": args, } def cleanup_children() -> None: for proc in reversed(CHILDREN): try: if proc.poll() is None: proc.terminate() try: proc.wait(timeout=2) except subprocess.TimeoutExpired: proc.kill() except Exception: pass atexit.register(cleanup_children) def check_dependencies(require_croc=False) -> None: missing = [] if not tailcat.available(): missing.append("tailcat") if require_croc and not shutil.which("croc"): missing.append("croc") if missing: raise RuntimeError("missing command(s): " + ", ".join(missing)) def start_tailcat_listener(port: int) -> str: """Compatibility wrapper; Tailcat implementation now lives in tailcat.py.""" global CURRENT_TAILCAT_ADDRESS CURRENT_TAILCAT_ADDRESS = tailcat.start(port) return CURRENT_TAILCAT_ADDRESS def identity_document(cfg: configparser.ConfigParser, address: str) -> Dict[str, Any]: return { "protocol": PROTOCOL, "node": local_node_name(cfg), "character": local_user_account_name(cfg), "address": address, "api_port": api_port(cfg), "services": { "characterif_api": api_port(cfg), }, "created": utcnow(), } def parse_identity_text(text: str) -> Dict[str, Any]: """ Find a characterif identity JSON object in croc's stdout. Normally croc's received --text content should be the stdout itself, but scanning permits harmless surrounding status text. """ text = text.strip() try: obj = json.loads(text) if obj.get("protocol") == PROTOCOL: return obj except Exception: pass # Try line-by-line in case croc includes informational output on stdout. for line in text.splitlines(): line = line.strip() if not (line.startswith("{") and line.endswith("}")): continue try: obj = json.loads(line) if obj.get("protocol") == PROTOCOL: return obj except Exception: pass # Last-resort JSON object scan. for m in re.finditer(r"\{.*?\}", text, flags=re.DOTALL): try: obj = json.loads(m.group(0)) if obj.get("protocol") == PROTOCOL: return obj except Exception: continue raise RuntimeError("received croc text did not contain a characterif/1 identity document") def start_croc_send_text(payload: str) -> tuple[subprocess.Popen, str]: """ Start `croc send --text ...`, read enough output to obtain its human code, and leave croc running until the receiver consumes the text. """ check_dependencies(require_croc=True) proc = subprocess.Popen( ["croc", "send", "--text", payload], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1, ) CHILDREN.append(proc) deadline = time.time() + 20 lines = [] code = None # Croc has historically printed variants such as "Code is: ...". patterns = [ re.compile(r"code\s+is\s*:\s*(.+?)\s*$", re.I), re.compile(r"code\s*:\s*(.+?)\s*$", re.I), ] while time.time() < deadline: if proc.stdout is None: break line = proc.stdout.readline() if line: clean = line.rstrip() lines.append(clean) # Show croc's output verbatim. Earlier we suppressed every line # containing "code", which accidentally hid the actual receive # command / URL containing the generated secret. print(f"[croc] {clean}") # Older/current classic form: # Code is: alpha-beta-gamma for pattern in patterns: m = pattern.search(clean) if m: code = m.group(1).strip() break # Current Linux/macOS form keeps the secret out of argv: # CROC_SECRET='alpha-beta-gamma' croc if not code: m = re.search( r"""CROC_SECRET\s*=\s*['\"]([^'\"]+)['\"]""", clean, flags=re.I, ) if m: code = m.group(1).strip() # Classic receive command: # croc alpha-beta-gamma if not code: m = re.search(r"\bcroc\s+([^\s]+)\s*$", clean) if m: candidate = m.group(1).strip().strip("'\"") if not candidate.startswith("-"): code = candidate # Browser form: # https://getcroc.com/?code=alpha-beta-gamma if not code: m = re.search(r"[?&]code=([^&\s]+)", clean, flags=re.I) if m: from urllib.parse import unquote code = unquote(m.group(1)).strip() if code: return proc, code elif proc.poll() is not None: break else: time.sleep(0.05) raise RuntimeError( "croc did not print a pairing code" + (("\nOutput:\n" + "\n".join(lines)) if lines else "") ) def receive_croc_text(code: str, timeout: int = 300) -> str: """ Receive croc text using CROC_SECRET in the child environment. This is intentionally the first thing to try. Python's subprocess API explicitly supports setting an environment for a spawned process. If the installed croc build does not accept CROC_SECRET for receive mode, we fall back to passing the code as an argument. """ check_dependencies(require_croc=True) env = os.environ.copy() env["CROC_SECRET"] = code result = run_cmd(["croc", "--yes"], timeout=timeout, env=env) if result["ok"] and result.get("stdout", "").strip(): return result["stdout"] # Compatibility fallback. If the environment-variable invocation fails, # try croc's ordinary positional-code form. fallback = run_cmd(["croc", "--yes", code], timeout=timeout) if fallback["ok"] and fallback.get("stdout", "").strip(): return fallback["stdout"] raise RuntimeError( "could not receive croc text.\n" f"CROC_SECRET attempt stderr: {result.get('stderr','').strip()}\n" f"argument fallback stderr: {fallback.get('stderr','').strip()}" ) def find_free_local_port() -> int: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.bind(("127.0.0.1", 0)) return int(s.getsockname()[1]) def wait_for_port(port: int, timeout=8) -> bool: deadline = time.time() + timeout while time.time() < deadline: try: with socket.create_connection(("127.0.0.1", port), timeout=0.3): return True except OSError: time.sleep(0.1) return False def post_json(url: str, payload: Dict[str, Any], timeout=15) -> Dict[str, Any]: data = json.dumps(payload).encode("utf-8") req = request.Request( url, data=data, headers={"Content-Type": "application/json"}, method="POST", ) with request.urlopen(req, timeout=timeout) as r: return json.loads(r.read().decode("utf-8")) def register_with_remote_peer(peer: Dict[str, Any], me: Dict[str, Any]) -> Dict[str, Any]: """Register through the optional Tailcat transport driver.""" return tailcat.post_json( peer["address"], int(peer.get("api_port", DEFAULT_PORT)), "/api/peer/register", me, timeout=15, ) def send_chat_to_remote_peer(peer: Dict[str, Any], envelope: Dict[str, Any]) -> Dict[str, Any]: """Send chat through the optional Tailcat transport driver.""" return tailcat.post_json( peer["address"], int(peer.get("api_port", DEFAULT_PORT)), "/api/chat-text", envelope, timeout=15, ) def make_app() -> "Flask": if Flask is None: raise RuntimeError("Flask is not installed. Try: pip install flask") cfg = load_config() app = Flask(APP_NAME) @app.get("/api/status") def api_status(): return jsonify({ "ok": True, "service": APP_NAME, "protocol": PROTOCOL, "api_version": API_VERSION, "node": local_node_name(cfg), "character": local_user_account_name(cfg), "tailcat_address": CURRENT_TAILCAT_ADDRESS, "time": utcnow(), }) @app.post("/api/cms/transfer") def api_cms_transfer(): """Receive story.md text and allocate a new CMS entry on this repository.""" if cfg_get(cfg, "characterif", "role", "workstation") != "domain-repository": return jsonify({"ok": False, "error": "This node is not a domain-repository"}), 403 payload = flask_request.get_json(silent=True) if not isinstance(payload, dict) or not isinstance(payload.get("story"), str): return jsonify({"ok": False, "error": "Expected JSON with a story string"}), 400 story = payload["story"] if not story.startswith("---\n") or len(story.encode("utf-8")) > 5_000_000: return jsonify({"ok": False, "error": "Invalid or oversized story.md"}), 400 try: import yaml from createentry import create_entry, set_yaml_field parts = story.split("\n---", 1) if len(parts) != 2: raise ValueError("Missing closing YAML frontmatter delimiter") metadata = yaml.safe_load(parts[0][4:]) if not isinstance(metadata, dict): raise ValueError("Invalid YAML frontmatter") original_id = str(metadata.get("entry_id") or "").strip() if not re.fullmatch(r"[A-Za-z0-9_-]+", original_id): raise ValueError("Missing or invalid entry_id") entry_type = str(metadata.get("entry_type") or "n").strip() if not re.fullmatch(r"[A-Za-z0-9_-]+", entry_type): raise ValueError("Invalid entry_type") title = str(metadata.get("title") or "Transferred story") except (ValueError, ImportError, yaml.YAMLError) as exc: return jsonify({"ok": False, "error": str(exc)}), 400 try: new_id, entry_dir, story_path = create_entry(entry_type, title=title) # Keep the submitted story intact except for its repository entry ID. story_path.write_text(story, encoding="utf-8") set_yaml_field(story_path, "entry_id", new_id) # Preserve where this received entry came from on the repository. source_node = str(payload.get("source_node") or "").strip() or None source_directory = str(payload.get("source_directory") or "").strip() or None source_address = flask_request.remote_addr event = { "datetime": utcnow(), "text": "Entry received from workstation", "original_entry_id": original_id, "entry_id": new_id, "source_node": source_node, "source_directory": source_directory, "source_address": source_address, "repository_node": local_node_name(cfg), } private_dir = entry_dir / "private" private_dir.mkdir(parents=True, exist_ok=True) history_path = private_dir / "transfer-history.json" history = [] if history_path.exists(): history = json.loads(history_path.read_text(encoding="utf-8")) if not isinstance(history, list): raise ValueError(f"Expected JSON list in {history_path}") history.append(event) history_path.write_text(json.dumps(history, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") publish_path = entry_dir / "publish.json" if publish_path.exists(): publish = json.loads(publish_path.read_text(encoding="utf-8")) if not isinstance(publish, dict): raise ValueError(f"Expected JSON object in {publish_path}") else: publish = {"free": True, "members": True, "premium": True} log = publish.get("log", []) if not isinstance(log, list): log = [] log.append(dict(event)) publish["log"] = log publish_path.write_text(json.dumps(publish, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") except Exception as exc: app.logger.exception("CMS transfer failed") return jsonify({"ok": False, "error": str(exc)}), 500 return jsonify({ "ok": True, "original_entry_id": original_id, "entry_id": new_id, "repository_node": local_node_name(cfg), "source_status": "Transferred", }), 201 # --- Character / AI display API -------------------------------------- @app.get("/mobile") def mobile_landing(): port = api_port(cfg) return render_template_string( MOBILE_LANDING_TEMPLATE, node=local_node_name(cfg), public_url=mobile_public_url(port), croc_code=CURRENT_CROC_CODE, ) @app.post("/api/mobile/wan-pairing") def mobile_wan_pairing(): """Return this node's Tailcat address for the hosted WAN browser.""" try: with WAN_PAIRING_LOCK: address = CURRENT_TAILCAT_ADDRESS or start_tailcat_listener(api_port(cfg)) me = identity_document(cfg, address) return jsonify({ "ok": True, "tailcat_address": address, "api_port": me["api_port"], "node": me["node"], }) except Exception as exc: return jsonify({"ok": False, "error": str(exc)}), 500 @app.get("/mobile/home") @app.get("/mobile/app") def mobile_home(): return render_template_string( MOBILE_HOME_TEMPLATE, node=local_node_name(cfg), ) @app.get("/mobile/remote-view") def mobile_remote_view(): return render_template_string( MOBILE_REMOTE_VIEW_TEMPLATE, node=local_node_name(cfg), ) @app.get("/mobile/messages") def mobile_messages(): return render_template_string( MOBILE_MESSAGES_TEMPLATE, node=local_node_name(cfg), ) @app.get("/mobile/settings") def mobile_settings(): return render_template_string( MOBILE_SETTINGS_TEMPLATE, node=local_node_name(cfg), ) @app.get("/") def index(): # The public CharacterIF index is intentionally just JSON. Only local # characters explicitly marked available_remotely are advertised. return jsonify({ "ok": True, "node": local_node_name(cfg), "characters": remotely_available_character_documents(cfg), }) @app.get("/api/characters") def api_characters(): return jsonify({"ok": True, "characters": all_character_documents(cfg)}) @app.get("/api/character/") def api_character(name): character = character_document(cfg, name) if character is None: return jsonify({"ok": False, "error": "character not found", "name": name}), 404 return jsonify({"ok": True, **character}) @app.get("/api/ais") def api_ais(): ais = [c for c in all_character_documents(cfg) if c.get("character_type") == "ai"] return jsonify({"ok": True, "ais": ais}) @app.get("/api/ai/") def api_ai(name): ai = character_document(cfg, name) if ai is None or ai.get("character_type") != "ai": return jsonify({"ok": False, "error": "ai not found", "name": name}), 404 return jsonify({"ok": True, **ai}) @app.get("/api/character//portrait") @app.get("/api/ai//portrait") @app.get("/api/ai//representation/standard") def api_character_portrait(name): character = character_document(cfg, name) if character is None: return jsonify({"ok": False, "error": "character not found", "name": name}), 404 path = portrait_path(cfg, name) if path is None: return jsonify({ "ok": False, "error": "portrait not configured", "name": name, "config_key": f"[{character_section(cfg, name)}] portrait", }), 501 if not path.is_file(): return jsonify({ "ok": False, "error": "portrait file not found", "name": name, "path": str(path), }), 404 return send_file(path) @app.post("/api/character//respond") def api_character_respond(name): body = flask_request.get_json(silent=True) or {} text_value = body.get("text") if not isinstance(text_value, str) or not text_value.strip(): return jsonify({"ok": False, "error": "text is required"}), 400 try: result = respond_as_character(cfg, name, text_value) return jsonify(result) except Exception as exc: return jsonify({"ok": False, "error": str(exc), "name": name}), 502 @app.post("/api/session/finish") def api_session_finish(): """Append a lightweight mobile Remote View snapshot to session.json.""" body = flask_request.get_json(silent=True) or {} messages = body.get("messages") if not isinstance(messages, list): return jsonify({"ok": False, "error": "messages must be a list"}), 400 ensure_state_dir() payload = { "version": 1, "sessions": [], } if SESSION_LOG.is_file(): try: existing = json.loads(SESSION_LOG.read_text(encoding="utf-8")) if isinstance(existing, dict): payload = existing except Exception: # Keep this endpoint intentionally forgiving during the prototype phase. payload = {"version": 1, "sessions": []} sessions = payload.setdefault("sessions", []) if not isinstance(sessions, list): sessions = [] payload["sessions"] = sessions session = { "finished": utcnow(), "node": local_node_name(cfg), "page": body.get("page") or "remote-view", "target": body.get("target") or "all", "messages": messages, } sessions.append(session) tmp = SESSION_LOG.with_suffix(".tmp") tmp.write_text(json.dumps(payload, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") tmp.replace(SESSION_LOG) return jsonify({ "ok": True, "saved": True, "path": str(SESSION_LOG), "messages": len(messages), }) @app.get("/api/location") def api_location(): return jsonify({"ok": True, **location_document()}) @app.get("/api/chat/recent") def api_chat_recent(): try: limit = int(flask_request.args.get("limit", 50)) except (TypeError, ValueError): limit = 50 limit = max(1, min(limit, RECENT_CHAT_LIMIT)) with RECENT_CHAT_LOCK: messages = list(RECENT_CHAT[-limit:]) return jsonify({"ok": True, "messages": messages}) # --- Existing peer/node API ------------------------------------------- @app.get("/api/nodes") def api_nodes(): return jsonify(load_registry()) @app.get("/api/nodes/") def api_node(node): peer = get_peer(node) if not peer: return jsonify({"ok": False, "error": "node not found", "node": node}), 404 return jsonify({"ok": True, "node": peer}) @app.post("/api/peer/register") def api_peer_register(): body = flask_request.get_json(silent=True) or {} if body.get("protocol") != PROTOCOL: return jsonify({"ok": False, "error": "unsupported protocol"}), 400 if not body.get("node") or not body.get("address"): return jsonify({"ok": False, "error": "node and address are required"}), 400 save_peer(body) # The joining node starts its Flask server immediately after this # registration returns, so retry its character index in the background. threading.Thread( target=_delayed_sync_remote_characters, args=(dict(body),), daemon=True, ).start() mine = None if CURRENT_TAILCAT_ADDRESS: mine = identity_document(cfg, CURRENT_TAILCAT_ADDRESS) print(f"\nPeer registered: {body['node']}") print(f" address: {body['address'][:45]}{'...' if len(body['address']) > 45 else ''}") return jsonify({ "ok": True, "registered": body["node"], "peer": mine, }) def run_response_plugin(body: Dict[str, Any]) -> None: """Run the locally approved plugin for an incoming response document. Plugins are opt-in through [plugins] in characterif.conf. The network document never supplies a filename or command. Missing/invalid config simply means that no plugin is run. """ if body.get("type") != "response": return plugin_name = cfg_get(cfg, "plugins", "response") if not plugin_name: return # "directory" is reserved configuration metadata, never a plugin name. if plugin_name.strip().lower() == "directory": print("\nPLUGIN skipped: 'directory' is reserved") return # Plugins are deliberately restricted to plain Python filenames. # No paths, arguments, or shell syntax are accepted from plugin entries. if not re.fullmatch(r"[A-Za-z0-9_.-]+\.py", plugin_name): print(f"\nPLUGIN skipped: invalid configured filename: {plugin_name!r}") return configured_dir = cfg_get(cfg, "plugins", "directory") if configured_dir: plugins_dir = Path(configured_dir).expanduser() if not plugins_dir.is_absolute(): plugins_dir = BASE_DIR / plugins_dir else: # Preserve the old location when no directory is configured. plugins_dir = Path(__file__).resolve().parent / "plugins" plugin_path = (plugins_dir / plugin_name).resolve() try: plugin_path.relative_to(plugins_dir.resolve()) except ValueError: print("\nPLUGIN skipped: configured plugin is outside plugins directory") return if not plugin_path.is_file(): print(f"\nPLUGIN skipped: not found: {plugin_path}") return try: result = subprocess.run( [sys.executable, str(plugin_path)], input=json.dumps(body), text=True, capture_output=True, timeout=DEFAULT_TIMEOUT, check=False, ) print(f"\nPLUGIN {plugin_name}: exit {result.returncode}") if result.stdout.strip(): print(result.stdout.rstrip()) if result.stderr.strip(): print(result.stderr.rstrip(), file=sys.stderr) except Exception as exc: print(f"\nPLUGIN {plugin_name} failed: {exc}", file=sys.stderr) @app.post("/api/chat-text") def api_chat_text(): body = flask_request.get_json(silent=True) or {} text_value = body.get("text") if not isinstance(text_value, str) or not text_value.strip(): return jsonify({"ok": False, "error": "text is required"}), 400 sender_node = body.get("from_node") or "unknown-node" sender_character = body.get("from_character") or body.get("from_ai") or "user" target_character = body.get("to_character") or body.get("to_ai") or sender_character target_node = body.get("to_node") or local_node_name(cfg) # Keep a small display-oriented feed as well as printing to console. # This is not intended to replace durable chat storage/queueing. remembered = dict(body) remembered.setdefault("from_node", sender_node) remembered.setdefault("from_character", sender_character) remembered.setdefault("to_node", target_node) remembered.setdefault("to_character", target_character) remember_chat_message(remembered) # A response document may trigger one explicitly approved local plugin. # Ordinary chat/request messages never execute plugins. run_response_plugin(remembered) print("\nCHAT") print(f" from node: {sender_node}") print(f" from character: {sender_character}") print(f" to node: {target_node}") print(f" to character: {target_character}") print(f" text: {text_value}") if ( target_node == local_node_name(cfg) and target_character.casefold() == local_user_account_name(cfg).casefold() ): notify_chat_message(cfg, sender_character, text_value) return jsonify({ "ok": True, "received": True, "node": local_node_name(cfg), "character": local_user_account_name(cfg), "time": utcnow(), }) @app.post("/api/ping") def api_ping(): body = flask_request.get_json(silent=True) or {} node = body.get("node") timeout = int(body.get("timeout", DEFAULT_TIMEOUT)) peer = get_peer(node) if node else None if not peer or not peer.get("address"): return jsonify({"ok": False, "error": "node/address not found", "node": node}), 404 result = tailcat.ping( peer["address"], until_direct=bool(body.get("until_direct")), timeout=timeout, ) result["node"] = node return jsonify(result), (200 if result["ok"] else 502) @app.post("/api/run") def api_run(): body = flask_request.get_json(silent=True) or {} node = body.get("node") command = body.get("command") timeout = int(body.get("timeout", 60)) if not node or not command: return jsonify({"ok": False, "error": "node and command are required"}), 400 peer = get_peer(node) if not peer or not peer.get("address"): return jsonify({"ok": False, "error": "node/address not found", "node": node}), 404 result = tailcat.run_remote( peer["address"], command, timeout=timeout, ) result["node"] = node result["command"] = command return jsonify(result), (200 if result["ok"] else 502) return app # Minimal LAN service discovery. Independent of Tailcat and MQTT. ZEROCONF_SERVICE = "_characterif._tcp.local." def advertise_lan(cfg, port): from zeroconf import Zeroconf, ServiceInfo node = local_node_name(cfg) role = cfg_get(cfg, "characterif", "role", "workstation").strip() # Determine the outbound LAN address without sending any packets. with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock: sock.connect(("192.0.2.1", 9)) ip = sock.getsockname()[0] if ip.startswith("127."): raise RuntimeError("No usable LAN IPv4 address") zc = Zeroconf() info = ServiceInfo( ZEROCONF_SERVICE, f"{node}.{ZEROCONF_SERVICE}", addresses=[socket.inet_aton(ip)], port=port, properties={"node": node, "role": role}, server=f"{socket.gethostname().split('.')[0]}.local.", ) try: zc.register_service(info) except Exception: zc.close() raise print(f"CharacterIF Zeroconf: {node} ({role}) at {ip}:{port}") return zc, info def discover_lan(seconds=3): from zeroconf import Zeroconf, ServiceBrowser, ServiceListener found = {} class Listener(ServiceListener): def add_service(self, zc, service_type, name): info = zc.get_service_info(service_type, name) if info: props = {k.decode('utf-8'): v.decode('utf-8') for k, v in info.properties.items()} found[name] = { "node": props.get("node", name.split('.')[0]), "role": props.get("role", "workstation"), "address": info.parsed_addresses()[0] if info.parsed_addresses() else "", "port": info.port, } def update_service(self, zc, service_type, name): self.add_service(zc, service_type, name) def remove_service(self, zc, service_type, name): found.pop(name, None) zc = Zeroconf() try: browser = ServiceBrowser(zc, ZEROCONF_SERVICE, Listener()) time.sleep(seconds) browser.cancel() return sorted(found.values(), key=lambda n: n["node"]) finally: zc.close() def run_flask(cfg: configparser.ConfigParser, bind=None, port=None) -> int: bind = bind or character_cfg_get(cfg, "api_host", DEFAULT_BIND) port = port or api_port(cfg) zc, service = advertise_lan(cfg, port) app = make_app() print(f"\n{APP_NAME}: node={local_node_name(cfg)}") print(f"{APP_NAME}: api=http://{bind}:{port}") print(f"{APP_NAME}: registry={REGISTRY_CACHE}") try: app.run(host=bind, port=port, threaded=True, use_reloader=False) finally: zc.unregister_service(service) zc.close() return 0 def run_mobile_web(cfg: configparser.ConfigParser, bind: Optional[str] = None, port: Optional[int] = None, open_browser: bool = False) -> int: """Run the LAN-only mobile-first CharacterIF web interface.""" bind = bind or "0.0.0.0" port = port or api_port(cfg) url = mobile_public_url(port) # WAN mode needs the same Tailcat listener used by the normal server/start paths. # Starting it here also makes the address visible immediately for diagnostics. address = CURRENT_TAILCAT_ADDRESS or start_tailcat_listener(port) print(f"\n{APP_NAME}: mobile LAN web interface") print(f"Local web interface:\n {url}") print(f"Tailcat address:\n {address}") print("\nOpen that address on a phone connected to the same LAN.") print("QR code, connection/busy state, and remote mobile access are intentionally deferred.") if open_browser: local_url = f"http://127.0.0.1:{port}/mobile" threading.Timer(0.7, lambda: webbrowser.open(local_url)).start() zc, service = advertise_lan(cfg, port) app = make_app() try: app.run(host=bind, port=port, threaded=True, use_reloader=False) finally: zc.unregister_service(service) zc.close() return 0 def api_request(method: str, path: str, payload: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: cfg = load_config() url = daemon_url(cfg) + path data = None headers = {} if payload is not None: data = json.dumps(payload).encode("utf-8") headers["Content-Type"] = "application/json" req = request.Request(url, data=data, headers=headers, method=method) try: with request.urlopen(req, timeout=120) as r: return json.loads(r.read().decode("utf-8")) except error.HTTPError as exc: body = exc.read().decode("utf-8", "replace") try: return json.loads(body) except Exception: return {"ok": False, "error": f"HTTP {exc.code}", "body": body} except Exception as exc: return {"ok": False, "error": str(exc), "url": url} def print_result(obj: Any, raw=False) -> int: if raw and isinstance(obj, dict): if obj.get("stdout"): sys.stdout.write(obj["stdout"]) if obj.get("stderr"): sys.stderr.write(obj["stderr"]) return 0 if obj.get("ok") else 1 print(json.dumps(obj, indent=2)) return 0 if not isinstance(obj, dict) or obj.get("ok", True) else 1 def cmd_start(args) -> int: global CURRENT_CROC_CODE cfg = load_config() port = args.port or api_port(cfg) check_dependencies(require_croc=True) address = start_tailcat_listener(port) me = identity_document(cfg, address) payload = json.dumps(me, separators=(",", ":")) print(f"\ncharacterif node: {me['node']}") print("Tailcat listener: ready") print(f"Tailcat address: {address}") _, code = start_croc_send_text(payload) CURRENT_CROC_CODE = code print("\nPAIRING CODE") print(f" {code}") print("\nGive that croc code to the second machine and run:") print(f" python3 tools/characterif.py join '{code}'") print("\nWaiting for peer registration...") return run_flask(cfg, bind=args.bind, port=port) def cmd_join(args) -> int: cfg = load_config() port = args.port or api_port(cfg) check_dependencies(require_croc=True) # B must become a server too before telling A about itself. address = start_tailcat_listener(port) me = identity_document(cfg, address) print(f"\ncharacterif node: {me['node']}") print("Tailcat listener: ready") print("Receiving first peer address through croc...") received = receive_croc_text(args.code, timeout=args.croc_timeout) peer = parse_identity_text(received) if peer["node"] == me["node"]: raise RuntimeError(f"peer has same node name as this machine: {me['node']}") save_peer(peer) print(f"Received peer: {peer['node']}") print("Connecting back over Tailcat and registering this node...") reply = register_with_remote_peer(peer, me) if not reply.get("ok"): raise RuntimeError("peer registration failed: " + json.dumps(reply)) # A may return its current identity. Save it because it is authoritative. returned_peer = reply.get("peer") if isinstance(returned_peer, dict) and returned_peer.get("node") and returned_peer.get("address"): save_peer(returned_peer) print("\nPAIRING COMPLETE") print(f" {me['node']} <-> {peer['node']}") print("Both nodes now have each other's Tailcat address.") print(f"Fetching characters from {peer['node']}...") written = sync_remote_characters(peer) print(f"Cached {len(written)} remote character(s).") for path in written: print(f" {path}") return run_flask(cfg, bind=args.bind, port=port) def _setup_prompt_value(label: str, default: str = "") -> str: suffix = f" [{default}]" if default else "" value = input(f"{label}{suffix}: ").strip() return value if value else default def _setup_prompt_bool(label: str, default: bool) -> bool: suffix = "[Y/n]" if default else "[y/N]" while True: value = input(f"{label} {suffix}: ").strip().lower() if not value: return default if value in {"y", "yes", "1", "true", "on"}: return True if value in {"n", "no", "0", "false", "off"}: return False print("Please answer y or n.") def _parse_bool_arg(value: str) -> bool: text = value.strip().lower() if text in {"y", "yes", "1", "true", "on"}: return True if text in {"n", "no", "0", "false", "off"}: return False raise argparse.ArgumentTypeError("expected true or false") def _setup_config_bool( cfg: configparser.ConfigParser, option: str, fallback: bool, ) -> bool: try: return cfg.getboolean("characterif", option, fallback=fallback) except ValueError: return fallback def _write_setup_config(cfg: configparser.ConfigParser) -> None: CONFIG_PATH.parent.mkdir(parents=True, exist_ok=True) with CONFIG_PATH.open("w", encoding="utf-8") as handle: cfg.write(handle) print(f"\nConfiguration written to:\n {CONFIG_PATH.resolve()}") def setup_character(args) -> int: """Create or update one CharacterIF character configuration.""" cfg = load_config() name = (args.name or _setup_prompt_value("Character name")).strip() if not name: raise RuntimeError("character name is required") is_account_owner = name.casefold() == local_user_account_name(cfg).casefold() if args.character_type is not None: is_ai = args.character_type == "ai" else: is_ai = _setup_prompt_bool("Is this an AI character?", not is_account_owner) portrait_text = args.portrait if not portrait_text: portrait_text = _setup_prompt_value("Portrait image filename (blank for default graphic)") portrait_source = Path(portrait_text).expanduser() if portrait_text.strip() else None if portrait_source is not None and not portrait_source.is_file(): raise RuntimeError(f"portrait file not found: {portrait_source}") available_remotely = ( args.available_remotely if args.available_remotely is not None else _setup_prompt_bool("Available remotely?", False) ) exists_remotely = ( args.exists_remotely if args.exists_remotely is not None else _setup_prompt_bool("Exists remotely?", False) ) interface_api = (args.interface_api or _setup_prompt_value("Interface API", "ollama")).strip() interface_model = (args.model or _setup_prompt_value("Model", "gemma3")).strip() if character_data_dir is None: raise RuntimeError("cannot import character_data_dir from src/config.py") data_dir = character_data_dir(name) data_dir.mkdir(parents=True, exist_ok=True) portrait_target = data_dir / "portrait.svg" if portrait_source is None: portrait_target.write_text(DEFAULT_PORTRAIT_SVG) else: portrait_target = data_dir / portrait_source.name if portrait_source.resolve() != portrait_target.resolve(): shutil.copy2(portrait_source, portrait_target) section = f"character-{name}" if not cfg.has_section(section): cfg.add_section(section) cfg.set(section, "name", name) cfg.set(section, "type", "ai" if is_ai else "") cfg.set(section, "portrait", str(portrait_target.resolve())) cfg.set(section, "data_dir", str(data_dir.resolve())) cfg.set(section, "available_remotely", "true" if available_remotely else "false") cfg.set(section, "exists_remotely", "true" if exists_remotely else "false") # Private, machine-local implementation details. This file is never used # to build the public character document returned over CharacterIF. private_path = data_dir / "character.conf" private = configparser.ConfigParser() if private_path.is_file(): private.read(private_path, encoding="utf-8") if not private.has_section("address"): private.add_section("address") if not private.has_section("interface"): private.add_section("interface") private.set("address", "node", local_node_name(cfg)) private.set("address", "local", "true") private.set("interface", "api", interface_api) private.set("interface", "model", interface_model) with private_path.open("w", encoding="utf-8") as handle: private.write(handle) _write_setup_config(cfg) print("\nCharacter configured") print(f" name: {name}") print(f" section: [{section}]") print(f" character type: {'ai' if is_ai else 'human'}") print(f" data: {data_dir.resolve()}") print(f" portrait: {portrait_target.resolve()}") print(f" available remotely: {'yes' if available_remotely else 'no'}") print(f" exists remotely: {'yes' if exists_remotely else 'no'}") print(f" interface api: {interface_api}") print(f" model: {interface_model}") print(f" private config: {(data_dir / 'character.conf').resolve()}") return 0 def setup_lan() -> int: """ Configure CharacterIF's local-network identity and LAN discovery. Existing config.ini sections and unrelated options are preserved. The historical [characterif] section name is retained for compatibility. """ cfg = load_config() if not cfg.has_section("characterif"): cfg.add_section("characterif") current_user_account = cfg_get(cfg, "characterif", "local_user_account_name", local_user_account_name(cfg)) current_node = cfg_get(cfg, "characterif", "node", local_node_name(cfg)) current_port = str(api_port(cfg)) current_lan = _setup_config_bool(cfg, "lan", True) print("\nCharacterIF LAN setup\n") user_account_name = _setup_prompt_value("User account name", current_user_account) node_name = _setup_prompt_value("Node name", current_node) while True: port_text = _setup_prompt_value("API port", current_port) try: port = int(port_text) if not 1 <= port <= 65535: raise ValueError break except ValueError: print("Please enter a TCP port between 1 and 65535.") lan_enabled = _setup_prompt_bool( "Enable automatic local LAN discovery?", current_lan, ) try: import zeroconf # noqa: F401 zeroconf_ok = True except ImportError: zeroconf_ok = False print("\nLAN dependency check") print(f" zeroconf: {'installed' if zeroconf_ok else 'not installed'}") if not _setup_prompt_bool("\nSave LAN configuration?", True): print("Configuration not changed.") return 0 cfg.set("characterif", "local_user_account_name", user_account_name) cfg.set("characterif", "node", node_name) cfg.set("characterif", "api_port", str(port)) cfg.set("characterif", "lan", "true" if lan_enabled else "false") section = f"character-{user_account_name}" if not cfg.has_section(section): cfg.add_section(section) _write_setup_config(cfg) if lan_enabled and not zeroconf_ok: print("\nLAN discovery is enabled, but zeroconf is not installed.") print("Install it with:") print(" pip install zeroconf") return 0 def setup_remote() -> int: """ Configure explicit online/Tailcat participation. Running this setup does not itself start Tailcat or contact the Internet. It only records whether online operation is enabled. """ cfg = load_config() if not cfg.has_section("characterif"): cfg.add_section("characterif") current_online = _setup_config_bool(cfg, "online", False) print("\nCharacterIF remote setup\n") print("Remote mode uses Tailcat and Croc and may communicate over the Internet.") print("LAN-only operation does not require remote mode.\n") online_enabled = _setup_prompt_bool( "Enable online/Tailcat networking?", current_online, ) tailcat_ok = tailcat.available() croc_ok = shutil.which("croc") is not None print("\nRemote dependency check") print(f" tailcat: {'installed' if tailcat_ok else 'not installed'}") print(f" croc: {'installed' if croc_ok else 'not installed'}") if not _setup_prompt_bool("\nSave remote configuration?", True): print("Configuration not changed.") return 0 cfg.set( "characterif", "online", "true" if online_enabled else "false", ) _write_setup_config(cfg) if online_enabled and not tailcat_ok: print("\nOnline networking is enabled, but tailcat is not installed.") if online_enabled and not croc_ok: print("\nOnline bootstrap is enabled, but croc is not installed.") return 0 def build_parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser(description="Heichalot CharacterIF node interface") p.add_argument( "--setup", choices=("lan", "remote"), help="configure LAN or remote networking", ) p.add_argument( "--setup-character", action="store_true", help="create or update a character; prompts for missing values", ) p.add_argument( "--character-list-json", action="store_true", help="print all known characters as JSON and exit", ) p.add_argument( "--character-status-json", action="store_true", help="print the shared-memory character connection statuses as JSON and exit", ) p.add_argument("--name", help="character name") p.add_argument( "--character-type", "--type", dest="character_type", choices=("ai", "human"), help="whether this is an AI or a biological-human character", ) p.add_argument("--portrait", help="portrait image filename") p.add_argument("--api", dest="interface_api", help="local responder API (default: ollama)") p.add_argument("--model", help="local responder model (default: gemma3)") p.add_argument( "--available_remotely", "--available-remotely", dest="available_remotely", type=_parse_bool_arg, help="whether this character can be reached remotely (true/false)", ) p.add_argument( "--exists_remotely", "--exists-remotely", dest="exists_remotely", type=_parse_bool_arg, help="whether this character exists on a remote node (true/false)", ) sub = p.add_subparsers(dest="command") s = sub.add_parser("start", help="start first node and print a croc pairing code") s.add_argument("--bind") s.add_argument("--port", type=int) s = sub.add_parser("join", help="join the first node using its croc pairing code") s.add_argument("code") s.add_argument("--bind") s.add_argument("--port", type=int) s.add_argument("--croc-timeout", type=int, default=300) s = sub.add_parser("web", help="run the mobile-first LAN web interface") s.add_argument("--bind", default="0.0.0.0") s.add_argument("--port", type=int) s.add_argument("--open", action="store_true", help="open the landing page in the desktop browser") s = sub.add_parser("server", help="run API/Tailcat server without croc pairing") s.add_argument("--bind") s.add_argument("--port", type=int) sub.add_parser("status", help="show running daemon status") sub.add_parser("nodes", help="list locally known peers") sub.add_parser("discover", help="discover running CharacterIF nodes on the LAN") p.add_argument("--find-repository", action="store_true", help="find the LAN domain repository (JSON)") s = sub.add_parser("sync-characters", help="fetch and cache characters from a known peer") s.add_argument("node") s = sub.add_parser("node", help="show one locally known peer") s.add_argument("node") s = sub.add_parser("ping", help="Tailcat ping a known peer") s.add_argument("node") s.add_argument("--until-direct", action="store_true") s.add_argument("--timeout", type=int, default=DEFAULT_TIMEOUT) s = sub.add_parser("run", help="run a command using Tailcat SSH") s.add_argument("node") s.add_argument("remote_command", nargs=argparse.REMAINDER) s.add_argument("--timeout", type=int, default=60) s = sub.add_parser("chat", help="send simple text from one character to another") s.add_argument("node") s.add_argument("--chat-text", required=True) s.add_argument( "--from-character", "--ai-name", dest="from_character", help="sending character name; defaults to 'user'", ) s.add_argument( "--to-character", "--to-ai", dest="to_character", help="target character name; defaults to the sending character", ) s = sub.add_parser("respond", help="request one synchronous response from a character") s.add_argument("character") s.add_argument("--text", required=True) s = sub.add_parser("ssh", help="open interactive Tailcat SSH to a known peer") s.add_argument("node") s.add_argument("ssh_args", nargs=argparse.REMAINDER) return p def main() -> int: parser = build_parser() args = parser.parse_args() if args.setup == "lan": return setup_lan() if args.setup == "remote": return setup_remote() if args.setup_character: return setup_character(args) if args.character_list_json: print(json.dumps(load_characters(), indent=2, sort_keys=True)) return 0 if args.character_status_json: print(json.dumps(character_statuses(), indent=2, sort_keys=True)) return 0 if not args.command and not args.find_repository: parser.print_help() return 2 cfg = load_config() try: if args.find_repository or args.command == "discover": nodes = discover_lan() if args.find_repository: matches = [n for n in nodes if n["role"] == "domain-repository"] if len(matches) != 1: print(json.dumps({"ok": False, "error": f"Expected one repository, found {len(matches)}", "repositories": matches}, indent=2)) return 1 print(json.dumps(matches[0], indent=2)) else: print(json.dumps(nodes, indent=2)) return 0 if args.command == "start": return cmd_start(args) if args.command == "join": return cmd_join(args) if args.command == "web": return run_mobile_web(cfg, bind=args.bind, port=args.port, open_browser=args.open) if args.command == "server": port = args.port or api_port(cfg) start_tailcat_listener(port) return run_flask(cfg, bind=args.bind, port=port) if args.command == "status": return print_result(api_request("GET", "/api/status")) if args.command == "nodes": return print_result(api_request("GET", "/api/nodes")) if args.command == "sync-characters": peer = get_peer(args.node) if not peer or not peer.get("address"): print(f"node {args.node!r} not found in {REGISTRY_CACHE}", file=sys.stderr) return 1 written = sync_remote_characters(peer) print(f"Cached {len(written)} remote character(s) from {args.node}.") for path in written: print(path) return 0 if args.command == "node": return print_result(api_request("GET", f"/api/nodes/{args.node}")) if args.command == "ping": return print_result(api_request("POST", "/api/ping", { "node": args.node, "until_direct": args.until_direct, "timeout": args.timeout, }), raw=True) if args.command == "run": remote_command = " ".join(args.remote_command).strip() if not remote_command: parser.error("run requires a remote command") return print_result(api_request("POST", "/api/run", { "node": args.node, "command": remote_command, "timeout": args.timeout, }), raw=True) if args.command == "chat": peer = get_peer(args.node) if not peer or not peer.get("address"): print(f"node {args.node!r} not found in {REGISTRY_CACHE}", file=sys.stderr) return 1 sender_character = args.from_character or "user" target_character = args.to_character or sender_character envelope = { "protocol": PROTOCOL, "type": "chat-text", "from_node": local_node_name(cfg), "from_character": sender_character, "to_node": args.node, "to_character": target_character, "text": args.chat_text, "time": utcnow(), } result = send_chat_to_remote_peer(peer, envelope) return print_result(result) if args.command == "respond": return print_result(respond_as_character(cfg, args.character, args.text)) if args.command == "ssh": result = api_request("GET", f"/api/nodes/{args.node}") if not result.get("ok"): return print_result(result) address = result["node"].get("address") if not address: print(f"node {args.node!r} has no Tailcat address", file=sys.stderr) return 1 return tailcat.ssh(address, args.ssh_args) except KeyboardInterrupt: print("\nStopped.") return 130 except Exception as exc: print(f"{APP_NAME}: ERROR: {exc}", file=sys.stderr) return 1 return 2 if __name__ == "__main__": raise SystemExit(main())