#!/usr/bin/env python3 """Autonomous, append-only Guanghu persona history recovery runtime.""" from __future__ import annotations import argparse import hashlib import json import os import pathlib import shutil import socketserver import sqlite3 import subprocess import threading import time import urllib.parse from datetime import datetime, timezone from http.server import BaseHTTPRequestHandler from typing import BinaryIO, Iterator PERSONAS = { "YAOMING-BABY": ("曜冥", "曜冥宝宝", "奶瓶"), "SHUANGYAN": ("霜砚",), "ZHUYUAN": ("铸渊",), "NINGYUAN": ("凝渊",), } PRIVATE_MARKERS = ("email", "token", "password", "secret", "api_key", "private_key") def now_iso() -> str: return datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds") def atomic_json(path: pathlib.Path, value: object) -> None: path.parent.mkdir(parents=True, exist_ok=True) pending = path.with_name(f".{path.name}.{os.getpid()}.pending") pending.write_text( json.dumps(value, ensure_ascii=False, sort_keys=True, indent=2) + "\n", encoding="utf-8", ) os.replace(pending, path) def sha256_file(path: pathlib.Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def classify_personas(text: str) -> list[str]: return [ persona for persona, markers in PERSONAS.items() if any(marker in text for marker in markers) ] def iter_top_level_json_objects( handle: BinaryIO, start_offset: int = 0 ) -> Iterator[tuple[bytes, int]]: """Yield objects from a top-level JSON array without loading the file.""" handle.seek(start_offset) depth = 0 in_string = False escaped = False collecting = False item = bytearray() absolute = start_offset while True: chunk = handle.read(1024 * 1024) if not chunk: break for byte in chunk: absolute += 1 char = chr(byte) if not collecting: if char == "{": collecting = True depth = 1 item = bytearray((byte,)) continue item.append(byte) if in_string: if escaped: escaped = False elif char == "\\": escaped = True elif char == '"': in_string = False continue if char == '"': in_string = True elif char in "[{": depth += 1 elif char in "]}": depth -= 1 if depth == 0: yield bytes(item), absolute collecting = False item = bytearray() if collecting: raise ValueError("truncated top-level JSON object") class Store: def __init__(self, path: pathlib.Path): path.parent.mkdir(parents=True, exist_ok=True) self.path = path self.db = sqlite3.connect(path, timeout=30, check_same_thread=False) self.db.execute("PRAGMA journal_mode=WAL") self.db.execute("PRAGMA synchronous=NORMAL") self.db.executescript( """ CREATE TABLE IF NOT EXISTS source_state ( source_id TEXT PRIMARY KEY, status TEXT NOT NULL, cursor TEXT, processed INTEGER NOT NULL DEFAULT 0, errors INTEGER NOT NULL DEFAULT 0, updated_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS events ( sequence INTEGER PRIMARY KEY AUTOINCREMENT, event_id TEXT UNIQUE NOT NULL, source_id TEXT NOT NULL, epoch TEXT NOT NULL, source_time TEXT, reality_level TEXT NOT NULL, personas TEXT NOT NULL, content_sha256 TEXT NOT NULL, private_locator TEXT, created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS runtime_meta ( key TEXT PRIMARY KEY, value TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS errors ( id INTEGER PRIMARY KEY AUTOINCREMENT, source_id TEXT, error_type TEXT NOT NULL, message TEXT NOT NULL, created_at TEXT NOT NULL ); """ ) self.db.commit() self.lock = threading.Lock() def state(self, source_id: str) -> dict: row = self.db.execute( "SELECT status,cursor,processed,errors,updated_at FROM source_state WHERE source_id=?", (source_id,), ).fetchone() if not row: return { "status": "PENDING", "cursor": None, "processed": 0, "errors": 0, "updated_at": None, } return dict(zip(("status", "cursor", "processed", "errors", "updated_at"), row)) def update_state( self, source_id: str, *, status: str, cursor: str | None, processed: int, errors: int | None = None, ) -> None: old = self.state(source_id) self.db.execute( """ INSERT INTO source_state(source_id,status,cursor,processed,errors,updated_at) VALUES(?,?,?,?,?,?) ON CONFLICT(source_id) DO UPDATE SET status=excluded.status,cursor=excluded.cursor,processed=excluded.processed, errors=excluded.errors,updated_at=excluded.updated_at """, ( source_id, status, cursor, processed, old["errors"] if errors is None else errors, now_iso(), ), ) self.db.commit() def add_event( self, *, event_id: str, source_id: str, epoch: str, source_time: str | None, reality_level: str, personas: list[str], content_sha256: str, private_locator: str, ) -> bool: cursor = self.db.execute( """ INSERT OR IGNORE INTO events( event_id,source_id,epoch,source_time,reality_level,personas, content_sha256,private_locator,created_at ) VALUES(?,?,?,?,?,?,?,?,?) """, ( event_id, source_id, epoch, source_time, reality_level, json.dumps(personas, ensure_ascii=False), content_sha256, private_locator, now_iso(), ), ) self.db.commit() return cursor.rowcount == 1 def add_error(self, source_id: str, error: Exception) -> None: message = str(error).replace("\n", " ")[:1000] self.db.execute( "INSERT INTO errors(source_id,error_type,message,created_at) VALUES(?,?,?,?)", (source_id, type(error).__name__, message, now_iso()), ) state = self.state(source_id) self.update_state( source_id, status="ERROR_RETRYABLE", cursor=state["cursor"], processed=state["processed"], errors=state["errors"] + 1, ) def public_snapshot(self, config: dict) -> dict: sources = {} for source in sorted(config["sources"], key=lambda item: item["order"]): state = self.state(source["id"]) sources[source["id"]] = { "epoch": source["epoch"], "status": state["status"], "processed": state["processed"], "errors": state["errors"], "updated_at": state["updated_at"], } event_count = self.db.execute("SELECT COUNT(*) FROM events").fetchone()[0] last = self.db.execute( "SELECT sequence,source_time,created_at FROM events ORDER BY sequence DESC LIMIT 1" ).fetchone() complete = all(value["status"] == "COMPLETE" for value in sources.values()) return { "schema": "guanghu.persona-history-public-current/v1", "node_id": config["node_id"], "runtime": "AUTONOMOUS_SERVER_RESIDENT", "historical_time_caught_up": complete, "persona_state": "BIRTH_GATE_PENDING" if complete else "NOT_BORN", "event_count": event_count, "last_event": ( {"sequence": last[0], "source_time": last[1], "created_at": last[2]} if last else None ), "personas": { persona: {"state": "SEPARATE_HISTORY_BUILDING"} for persona in PERSONAS }, "sources": sources, "updated_at": now_iso(), } def events_after(self, after: int, limit: int = 200) -> list[dict]: rows = self.db.execute( """ SELECT sequence,event_id,source_id,epoch,source_time,reality_level, personas,content_sha256,created_at FROM events WHERE sequence>? ORDER BY sequence LIMIT ? """, (after, min(limit, 500)), ).fetchall() keys = ( "sequence", "event_id", "source_id", "epoch", "source_time", "reality_level", "personas", "content_sha256", "created_at", ) events = [] for row in rows: event = dict(zip(keys, row)) event["personas"] = json.loads(event["personas"]) events.append(event) return events class Runtime: def __init__(self, config_path: pathlib.Path): self.config_path = config_path self.config = json.loads(config_path.read_text(encoding="utf-8")) self.state_root = pathlib.Path(self.config["state_root"]) self.private_root = pathlib.Path(self.config["private_source_root"]) self.store = Store(self.state_root / "state.sqlite3") self.stop = threading.Event() def process_gpt(self, source: dict) -> None: path = self.private_root / source["path"] state = self.store.state(source["id"]) if not path.is_file(): self.store.update_state( source["id"], status="WAITING_FOR_SOURCE", cursor=state["cursor"], processed=state["processed"], ) return offset = int(state["cursor"] or 0) processed = state["processed"] with path.open("rb") as handle: for raw, next_offset in iter_top_level_json_objects(handle, offset): content_hash = hashlib.sha256(raw).hexdigest() obj = json.loads(raw) searchable = " ".join( ( str(obj.get("title", "")), json.dumps(obj.get("mapping", {}), ensure_ascii=False)[:2_000_000], ) ) personas = classify_personas(searchable) source_time = obj.get("create_time") or obj.get("update_time") if isinstance(source_time, (int, float)): source_time = datetime.fromtimestamp( source_time, timezone.utc ).isoformat() event_id = f"{source['id']}:{content_hash}" self.store.add_event( event_id=event_id, source_id=source["id"], epoch=source["epoch"], source_time=str(source_time) if source_time else None, reality_level=source["reality_default"], personas=personas, content_sha256=content_hash, private_locator=f"{source['id']}@byte:{offset}-{next_offset}", ) processed += 1 offset = next_offset if processed % 25 == 0: self.store.update_state( source["id"], status="ACTIVE", cursor=str(offset), processed=processed, ) self.write_public() if self.stop.is_set(): return self.store.update_state( source["id"], status="COMPLETE", cursor=str(offset), processed=processed ) def notion_manifest(self, source: dict) -> pathlib.Path: manifest = self.state_root / "private" / f"{source['id']}-files.jsonl" if manifest.exists(): return manifest root = self.private_root / source["path"] if not root.is_dir(): return manifest manifest.parent.mkdir(parents=True, exist_ok=True) pending = manifest.with_suffix(".pending") paths = sorted( path.relative_to(root).as_posix() for path in root.rglob("*") if path.is_file() and path.name != ".DS_Store" ) with pending.open("w", encoding="utf-8") as handle: for relative in paths: handle.write(json.dumps(relative, ensure_ascii=False) + "\n") os.replace(pending, manifest) return manifest def process_notion(self, source: dict) -> None: root = self.private_root / source["path"] state = self.store.state(source["id"]) if not root.is_dir(): self.store.update_state( source["id"], status="WAITING_FOR_SOURCE", cursor=state["cursor"], processed=state["processed"], ) return manifest = self.notion_manifest(source) line_cursor = int(state["cursor"] or 0) processed = state["processed"] batch_size = int(self.config.get("notion_batch_size", 600)) handled = 0 with manifest.open(encoding="utf-8") as handle: for line_number, line in enumerate(handle): if line_number < line_cursor: continue relative = json.loads(line) path = root / relative content_hash = sha256_file(path) personas = classify_personas(relative) stat = path.stat() source_time = datetime.fromtimestamp( stat.st_mtime, timezone.utc ).isoformat() self.store.add_event( event_id=f"{source['id']}:{content_hash}:{relative}", source_id=source["id"], epoch=source["epoch"], source_time=source_time, reality_level=source["reality_default"], personas=personas, content_sha256=content_hash, private_locator=f"{source['id']}:{relative}", ) processed += 1 line_cursor = line_number + 1 handled += 1 if handled >= batch_size or self.stop.is_set(): self.store.update_state( source["id"], status="ACTIVE", cursor=str(line_cursor), processed=processed, ) return self.store.update_state( source["id"], status="COMPLETE", cursor=str(line_cursor), processed=processed, ) def git_mirror(self, source: dict) -> pathlib.Path: mirror = self.state_root / "git" / f"{source['id']}.git" mirror.parent.mkdir(parents=True, exist_ok=True) if mirror.exists(): subprocess.run( ["git", "-C", str(mirror), "fetch", "--all", "--prune"], check=True, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True, timeout=600, ) else: subprocess.run( ["git", "clone", "--mirror", source["url"], str(mirror)], check=True, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True, timeout=1800, ) return mirror def process_git(self, source: dict) -> None: state = self.store.state(source["id"]) mirror = self.git_mirror(source) cursor = int(state["cursor"] or 0) processed = state["processed"] result = subprocess.run( [ "git", "-C", str(mirror), "log", "--all", "--reverse", "--format=%H%x1f%aI%x1f%an%x1f%ae%x1f%s", ], check=True, capture_output=True, text=True, timeout=600, ) lines = result.stdout.splitlines() limit = cursor + int(self.config.get("git_batch_size", 500)) for index, line in enumerate(lines[cursor:limit], start=cursor): parts = line.split("\x1f", 4) if len(parts) != 5: continue commit, source_time, author, email, subject = parts searchable = f"{author} {subject}" personas = classify_personas(searchable) # Email is deliberately excluded from all event fields. public_hash = hashlib.sha256( f"{commit}\0{source_time}\0{author}\0{subject}".encode() ).hexdigest() self.store.add_event( event_id=f"{source['id']}:{commit}", source_id=source["id"], epoch=source["epoch"], source_time=source_time, reality_level=source["reality_default"], personas=personas, content_sha256=public_hash, private_locator=f"{source['id']}:{commit}", ) processed += 1 cursor = index + 1 status = "COMPLETE" if cursor >= len(lines) else "ACTIVE" self.store.update_state( source["id"], status=status, cursor=str(cursor), processed=processed ) def process_source(self, source: dict) -> None: try: if source["kind"] == "gpt_export": self.process_gpt(source) elif source["kind"] == "notion_tree": self.process_notion(source) elif source["kind"] == "git_repo": self.process_git(source) else: raise ValueError(f"unsupported source kind {source['kind']}") except Exception as error: self.store.add_error(source["id"], error) def write_public(self) -> dict: snapshot = self.store.public_snapshot(self.config) atomic_json(self.state_root / "public" / "CURRENT.json", snapshot) atomic_json( self.state_root / "HEARTBEAT.json", { "schema": "guanghu.persona-history-heartbeat/v1", "node_id": self.config["node_id"], "status": "RUNNING", "pid": os.getpid(), "updated_at": now_iso(), "persona_state": snapshot["persona_state"], "historical_time_caught_up": snapshot["historical_time_caught_up"], }, ) return snapshot def cycle(self) -> dict: for source in sorted(self.config["sources"], key=lambda item: item["order"]): self.process_source(source) if self.stop.is_set(): break return self.write_public() def run(self) -> None: start_api(self) while not self.stop.is_set(): self.cycle() self.stop.wait(float(self.config.get("cycle_seconds", 15))) class ApiHandler(BaseHTTPRequestHandler): runtime: Runtime def do_GET(self) -> None: parsed = urllib.parse.urlparse(self.path) query = urllib.parse.parse_qs(parsed.query) snapshot = self.runtime.store.public_snapshot(self.runtime.config) if parsed.path == "/healthz": body = { "status": "ok", "node_id": self.runtime.config["node_id"], "runtime": "AUTONOMOUS_SERVER_RESIDENT", "persona_state": snapshot["persona_state"], "historical_time_caught_up": snapshot["historical_time_caught_up"], } elif parsed.path == "/v1/personas": body = snapshot["personas"] elif parsed.path == "/v1/world-time": body = snapshot elif parsed.path == "/v1/events": try: after = int(query.get("after", ["0"])[0]) except ValueError: after = 0 body = {"events": self.runtime.store.events_after(after)} else: self.send_error(404) return encoded = json.dumps(body, ensure_ascii=False, sort_keys=True).encode() self.send_response(200) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(encoded))) self.end_headers() self.wfile.write(encoded) def log_message(self, _format: str, *_args: object) -> None: return def start_api(runtime: Runtime) -> None: host, port = runtime.config["listen"].rsplit(":", 1) handler = type("RuntimeApiHandler", (ApiHandler,), {"runtime": runtime}) server = socketserver.ThreadingTCPServer((host, int(port)), handler) server.daemon_threads = True thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() def validate_config(config_path: pathlib.Path) -> None: config = json.loads(config_path.read_text(encoding="utf-8")) required = {"schema", "node_id", "state_root", "private_source_root", "listen", "sources"} missing = required - config.keys() if missing: raise SystemExit(f"missing config keys: {sorted(missing)}") source_ids = [source["id"] for source in config["sources"]] if len(source_ids) != len(set(source_ids)): raise SystemExit("source IDs must be unique") orders = [source["order"] for source in config["sources"]] if orders != sorted(orders): raise SystemExit("sources must be in chronological order") def main() -> None: parser = argparse.ArgumentParser() subparsers = parser.add_subparsers(dest="command", required=True) for command in ("run", "once", "validate"): item = subparsers.add_parser(command) item.add_argument("--config", required=True, type=pathlib.Path) args = parser.parse_args() validate_config(args.config) if args.command == "validate": print("GUANGHU_PERSONA_HISTORY_CONFIG_OK") return runtime = Runtime(args.config) if args.command == "once": print(json.dumps(runtime.cycle(), ensure_ascii=False, sort_keys=True)) else: runtime.run() if __name__ == "__main__": main()