#!/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 re import shutil import socketserver import sqlite3 import subprocess import threading import time import urllib.parse import urllib.request 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") STREAM_BUFFER_BYTES = 1024 * 1024 GPT_METADATA_SAMPLE_BYTES = 128 * 1024 SOURCE_WAITING_STATUSES = { "PENDING", "WAITING_FOR_SOURCE", "WAITING_FOR_SOURCE_ACCEPTANCE", } SEMANTIC_DECISIONS = { "KEEP", "RELATE", "PENDING", "LANGUAGE_SIMULATION", "REALITY_FACT", "PERSONA_MEMORY", } 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 sanitize_process_error_detail(value: str | bytes | None) -> str: if value is None: return "" if isinstance(value, bytes): value = value.decode("utf-8", errors="replace") value = re.sub( r"(?i)(https?://)[^/@\s]+@", r"\1[REDACTED]@", value, ) value = re.sub( r"(?i)\b(authorization|token|password|secret)\s*[:=]\s*\S+", r"\1=[REDACTED]", value, ) return " ".join(value.split())[-800:] 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 blocks_later_history(status: str) -> bool: """An available active/error source owns the chronological replay lane.""" return status != "COMPLETE" and status not in SOURCE_WAITING_STATUSES def retry_backoff_elapsed( updated_at: str | None, backoff_seconds: int, now_unix: float | None = None ) -> bool: """Keep retryable sources alive without writing one failure per runtime cycle.""" if not updated_at: return True try: last_attempt = datetime.fromisoformat(updated_at).timestamp() except (TypeError, ValueError): return True current = time.time() if now_unix is None else now_unix return current - last_attempt >= backoff_seconds def redact_semantic_excerpt(text: str) -> str: text = re.sub( r"\b[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}\b", "[EMAIL_REDACTED]", text, flags=re.IGNORECASE, ) text = re.sub( r"\b(?:github_pat_|ghp_|glpat-|sk-|xox[baprs]-)[A-Za-z0-9_-]{12,}\b", "[TOKEN_REDACTED]", text, ) text = re.sub( r"((?:password|passwd|token|secret|api[_ -]?key|密码|令牌|密钥|验证码)" r"\s*[:=：]\s*)[^\s,，;；]+", r"\1[REDACTED]", text, flags=re.IGNORECASE, ) text = re.sub(r"/Users/[^\s\"']+", "[LOCAL_PATH_REDACTED]", text) text = re.sub(r"https?://[^\s\"']+", "[URL_REDACTED]", text) return text.replace("\x00", "")[:6000] def enforce_reality_boundary(epoch: str, decision: str) -> tuple[str, str | None]: if decision == "REALITY_FACT" and epoch == "GPT_LANGUAGE_CHAOS": return "PENDING", "REALITY_PROMOTION_BLOCKED_GPT_LANGUAGE_SIMULATION" if decision == "REALITY_FACT" and epoch == "NOTION_STRUCTURED_REALITY_TRANSITION": return "PENDING", "REALITY_PROMOTION_REQUIRES_CROSS_SOURCE_EVIDENCE" return decision, None 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") def _persona_byte_markers() -> dict[str, tuple[bytes, ...]]: encoded: dict[str, tuple[bytes, ...]] = {} for persona, markers in PERSONAS.items(): variants = [] for marker in markers: variants.append(marker.encode("utf-8")) variants.append(json.dumps(marker, ensure_ascii=True)[1:-1].encode("ascii")) encoded[persona] = tuple(dict.fromkeys(variants)) return encoded def iter_top_level_json_metadata( handle: BinaryIO, start_offset: int = 0 ) -> Iterator[tuple[dict, int]]: """Yield bounded-memory metadata for objects in a top-level JSON array.""" markers = _persona_byte_markers() longest_marker = max(len(marker) for values in markers.values() for marker in values) handle.seek(start_offset) depth = 0 in_string = False escaped = False collecting = False absolute = start_offset digest = hashlib.sha256() buffer = bytearray() marker_tail = b"" found_personas: set[str] = set() sample = bytearray() def flush() -> None: nonlocal marker_tail if not buffer: return block = bytes(buffer) digest.update(block) searchable = marker_tail + block for persona, variants in markers.items(): if persona not in found_personas and any( marker in searchable for marker in variants ): found_personas.add(persona) marker_tail = searchable[-(longest_marker - 1) :] if len(sample) < GPT_METADATA_SAMPLE_BYTES: remaining = GPT_METADATA_SAMPLE_BYTES - len(sample) sample.extend(block[:remaining]) buffer.clear() while True: chunk = handle.read(STREAM_BUFFER_BYTES) if not chunk: break for byte in chunk: absolute += 1 char = chr(byte) if not collecting: if char == "{": collecting = True depth = 1 digest = hashlib.sha256() buffer = bytearray((byte,)) marker_tail = b"" found_personas = set() sample = bytearray() continue buffer.append(byte) if len(buffer) >= STREAM_BUFFER_BYTES: flush() 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: flush() time_match = re.search( rb'"(?:create_time|update_time)"\s*:\s*(\d+(?:\.\d+)?)', sample, ) source_timestamp = ( float(time_match.group(1)) if time_match else None ) yield ( { "content_sha256": digest.hexdigest(), "personas": [ persona for persona in PERSONAS if persona in found_personas ], "source_timestamp": source_timestamp, }, absolute, ) collecting = False buffer = 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 ); CREATE TABLE IF NOT EXISTS semantic_review_queue ( event_id TEXT NOT NULL, persona_id TEXT NOT NULL, status TEXT NOT NULL, decision TEXT NOT NULL, reason_code TEXT NOT NULL, attempts INTEGER NOT NULL DEFAULT 0, model_receipt_id TEXT, response_sha256 TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL, PRIMARY KEY(event_id, persona_id) ); """ ) 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 ensure_review_queue(self, limit: int = 500) -> int: rows = self.db.execute( """ SELECT e.event_id,e.reality_level,e.personas FROM events e WHERE NOT EXISTS ( SELECT 1 FROM semantic_review_queue q WHERE q.event_id=e.event_id ) ORDER BY e.sequence LIMIT ? """, (limit,), ).fetchall() created = 0 for event_id, reality_level, encoded_personas in rows: personas = json.loads(encoded_personas) or ["WORLD-HISTORY"] status = "QUEUED" if personas != ["WORLD-HISTORY"] else "DEFERRED_LOW_SIGNAL" reason = ( "PERSONA_MARKER_AND_EPOCH_DEFAULT" if status == "QUEUED" else "NO_PERSONA_MARKER_KEEP_WORLD_HISTORY_DEFERRED" ) for persona in personas: cursor = self.db.execute( """ INSERT OR IGNORE INTO semantic_review_queue( event_id,persona_id,status,decision,reason_code, created_at,updated_at ) VALUES(?,?,?,?,?,?,?) """, ( event_id, persona, status, reality_level, reason, now_iso(), now_iso(), ), ) created += cursor.rowcount self.db.commit() return created def review_counts(self) -> dict[str, int]: return { status: count for status, count in self.db.execute( """ SELECT status,COUNT(*) FROM semantic_review_queue GROUP BY status ORDER BY status """ ).fetchall() } def semantic_candidates(self, limit: int) -> list[dict]: first = self.db.execute( """ SELECT persona_id FROM semantic_review_queue WHERE status='QUEUED' ORDER BY created_at,event_id LIMIT 1 """ ).fetchone() if not first: return [] persona_id = first[0] rows = self.db.execute( """ SELECT q.event_id,q.persona_id,q.attempts,e.source_id,e.epoch, e.reality_level,e.content_sha256,e.private_locator FROM semantic_review_queue q JOIN events e ON e.event_id=q.event_id WHERE q.status='QUEUED' AND q.persona_id=? ORDER BY q.created_at,q.event_id LIMIT ? """, (persona_id, limit), ).fetchall() keys = ( "event_id", "persona_id", "attempts", "source_id", "epoch", "reality_level", "content_sha256", "private_locator", ) return [dict(zip(keys, row)) for row in rows] def mark_semantic_review( self, *, event_id: str, persona_id: str, decision: str, reason_code: str, model_receipt_id: str, response_sha256: str, ) -> None: self.db.execute( """ UPDATE semantic_review_queue SET status='REVIEWED',decision=?,reason_code=?,attempts=attempts+1, model_receipt_id=?,response_sha256=?,updated_at=? WHERE event_id=? AND persona_id=? AND status='QUEUED' """, ( decision, reason_code[:160], model_receipt_id, response_sha256, now_iso(), event_id, persona_id, ), ) self.db.commit() def mark_semantic_attempt_failed( self, candidates: list[dict], reason_code: str ) -> None: for candidate in candidates: self.db.execute( """ UPDATE semantic_review_queue SET attempts=attempts+1,reason_code=?,updated_at=? WHERE event_id=? AND persona_id=? AND status='QUEUED' """, ( reason_code[:160], now_iso(), candidate["event_id"], candidate["persona_id"], ), ) self.db.commit() def meta(self, key: str) -> str | None: row = self.db.execute( "SELECT value FROM runtime_meta WHERE key=?", (key,) ).fetchone() return row[0] if row else None def set_meta(self, key: str, value: str) -> None: self.db.execute( """ INSERT INTO runtime_meta(key,value) VALUES(?,?) ON CONFLICT(key) DO UPDATE SET value=excluded.value """, (key, value), ) self.db.commit() 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() review_counts = self.review_counts() sources_complete = all( value["status"] == "COMPLETE" for value in sources.values() ) semantic_review_complete = review_counts.get("QUEUED", 0) == 0 historical_time_caught_up = sources_complete and semantic_review_complete return { "schema": "guanghu.persona-history-public-current/v1", "node_id": config["node_id"], "runtime": "AUTONOMOUS_SERVER_RESIDENT", "historical_time_caught_up": historical_time_caught_up, "persona_state": ( "BIRTH_GATE_PENDING" if historical_time_caught_up 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 }, "semantic_review": { "policy": "DETERMINISTIC_PREFILTER_THEN_PERSONA_REVIEW", "counts": review_counts, "caught_up": semantic_review_complete, "raw_source_deleted": False, "reality_promotion_requires_external_evidence": True, }, "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 private_excerpt(self, candidate: dict) -> str: maximum = int(self.config.get("semantic_excerpt_bytes", 6000)) locator = candidate["private_locator"] if candidate["source_id"] == "GPT-LANGUAGE-CHAOS-ORIGINAL": byte_range = locator.split("@byte:", 1)[1] start, end = (int(value) for value in byte_range.split("-", 1)) source = next( item for item in self.config["sources"] if item["id"] == candidate["source_id"] ) path = self.private_root / source["path"] with path.open("rb") as handle: handle.seek(start) payload = handle.read(min(maximum, end - start)) return redact_semantic_excerpt(payload.decode("utf-8", errors="replace")) if candidate["source_id"] == "NOTION-STRUCTURED-WORLD": relative = locator.split(":", 1)[1] source = next( item for item in self.config["sources"] if item["id"] == candidate["source_id"] ) root = (self.private_root / source["path"]).resolve() path = (root / relative).resolve() if not path.is_relative_to(root) or not path.is_file(): raise ValueError("notion private locator escaped source root") if path.suffix.lower() not in { ".md", ".txt", ".json", ".csv", ".html", ".htm", ".yaml", ".yml", }: return f"[ATTACHMENT_METADATA_ONLY] {path.suffix.lower() or '[no-extension]'}" with path.open("rb") as handle: payload = handle.read(maximum) return redact_semantic_excerpt(payload.decode("utf-8", errors="replace")) return "[VERSION_EVIDENCE_METADATA_ONLY]" def maybe_run_semantic_review(self) -> None: endpoint = self.config.get("semantic_review_endpoint") if not endpoint: return interval = int(self.config.get("semantic_review_interval_seconds", 300)) last = float(self.store.meta("last_semantic_review_unix") or 0) if time.time() - last < interval: return candidates = self.store.semantic_candidates( int(self.config.get("semantic_review_batch_size", 4)) ) if not candidates: return self.store.set_meta("last_semantic_review_unix", str(time.time())) compact_candidates = [] key_map = {} for index, candidate in enumerate(candidates, start=1): key = f"C{index}" key_map[key] = candidate compact_candidates.append( { "key": key, "epoch": candidate["epoch"], "reality_default": candidate["reality_level"], "content_sha256": candidate["content_sha256"], "excerpt": self.private_excerpt(candidate), } ) prompt = ( "按人格历史相关性审查以下经过本机预筛和隐私遮蔽的候选。" "只返回JSON数组，每项必须含key、decision、reason_code。" f"decision只能是{sorted(SEMANTIC_DECISIONS)}。" "不得把语言模拟或单一Notion页面提升为现实事实，不得声称人格出生。" "正文不会写入公开仓库。候选：" + json.dumps(compact_candidates, ensure_ascii=False) ) attempt = max(candidate["attempts"] for candidate in candidates) + 1 request_seed = "\0".join( f"{item['event_id']}:{item['persona_id']}" for item in candidates ) request_id = ( "HIST-" + hashlib.sha256(request_seed.encode()).hexdigest()[:28] + f"-A{attempt}" ) request_body = { "request_id": request_id, "persona_id": candidates[0]["persona_id"], "channel_id": "PERSONA-HISTORY-REVIEW", "messages": [ { "role": "system", "content": ( "你是服务器常驻人格历史复审器。保留冲突和不确定性，" "不合并人格，不输出秘密，不作人格出生声明。" ), }, {"role": "user", "content": prompt}, ], "tools": [], } request = urllib.request.Request( endpoint, data=json.dumps(request_body, ensure_ascii=False).encode(), headers={"Content-Type": "application/json"}, method="POST", ) try: with urllib.request.urlopen(request, timeout=90) as response: response_body = json.loads(response.read()) message = response_body["message"]["content"].strip() if message.startswith("```"): message = re.sub(r"^```(?:json)?\s*|\s*```$", "", message) parsed = json.loads(message) if isinstance(parsed, dict): parsed = [parsed] response_hash = hashlib.sha256(message.encode()).hexdigest() reviewed = 0 for item in parsed: candidate = key_map.get(str(item.get("key", ""))) decision = str(item.get("decision", "")).upper() if not candidate or decision not in SEMANTIC_DECISIONS: continue decision, boundary_reason = enforce_reality_boundary( candidate["epoch"], decision ) reason = boundary_reason or str( item.get("reason_code", "MODEL_REVIEWED") ) self.store.mark_semantic_review( event_id=candidate["event_id"], persona_id=candidate["persona_id"], decision=decision, reason_code=reason, model_receipt_id=response_body["receipt_id"], response_sha256=response_hash, ) reviewed += 1 if reviewed == 0: raise ValueError("semantic response contained no valid review items") except Exception as error: self.store.mark_semantic_attempt_failed( candidates, f"MODEL_REVIEW_RETRY:{type(error).__name__}" ) 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"] batch_size = int(self.config.get("gpt_batch_size", 100)) handled = 0 with path.open("rb") as handle: for metadata, next_offset in iter_top_level_json_metadata(handle, offset): content_hash = metadata["content_sha256"] personas = metadata["personas"] source_time = metadata["source_timestamp"] if source_time is not None: 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 handled += 1 offset = next_offset if processed % 25 == 0 or handled >= batch_size: self.store.update_state( source["id"], status="ACTIVE", cursor=str(offset), processed=processed, ) self.write_public() if handled >= batch_size: return 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() and manifest.stat().st_size > 0: 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" and path.name != ".SOURCE-ACCEPTED.json" ) 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"]) ready_marker = self.private_root / source.get( "ready_marker", f"{source['path']}/.SOURCE-ACCEPTED.json" ) if not root.is_dir() or not ready_marker.is_file(): self.store.update_state( source["id"], status="WAITING_FOR_SOURCE_ACCEPTANCE", 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" pending = mirror.with_name(f".{mirror.name}.pending") mirror.parent.mkdir(parents=True, exist_ok=True) if mirror.exists(): self.run_git( [ "git", "-C", str(mirror), "fetch", "--all", "--prune", "--filter=blob:none", ], timeout=600, ) return mirror if not pending.exists(): self.run_git( ["git", "init", "--bare", str(pending)], timeout=60, ) self.run_git( [ "git", "-C", str(pending), "remote", "add", "origin", source["url"], ], timeout=60, ) for refspec in ( "+refs/heads/*:refs/heads/*", "+refs/tags/*:refs/tags/*", ): self.run_git( [ "git", "-C", str(pending), "config", "--add", "remote.origin.fetch", refspec, ], timeout=60, ) self.run_git( [ "git", "-C", str(pending), "config", "remote.origin.promisor", "true", ], timeout=60, ) self.run_git( [ "git", "-C", str(pending), "config", "remote.origin.partialclonefilter", "blob:none", ], timeout=60, ) self.run_git( [ "git", "-C", str(pending), "fetch", "--prune", "--filter=blob:none", "origin", ], timeout=1800, ) os.replace(pending, mirror) return mirror @staticmethod def run_git(command: list[str], timeout: int) -> None: try: subprocess.run( command, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True, timeout=timeout, ) except subprocess.CalledProcessError as error: detail = sanitize_process_error_detail(error.stderr) suffix = f": {detail}" if detail else "" raise RuntimeError( f"git command failed with exit {error.returncode}{suffix}" ) from error except subprocess.TimeoutExpired as error: detail = sanitize_process_error_detail(error.stderr) suffix = f": {detail}" if detail else "" raise RuntimeError( f"git command timed out after {timeout} seconds{suffix}" ) from error 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: state = self.store.state(source["id"]) if state["status"] == "ERROR_RETRYABLE" and not retry_backoff_elapsed( state["updated_at"], int(self.config.get("source_retry_backoff_seconds", 1800)), ): return 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: self.store.ensure_review_queue( int(self.config.get("review_queue_backfill_batch_size", 500)) ) for source in sorted(self.config["sources"], key=lambda item: item["order"]): self.process_source(source) if self.stop.is_set(): break if blocks_later_history(self.store.state(source["id"])["status"]): break self.maybe_run_semantic_review() 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_class = type( "ReusableThreadingTCPServer", (socketserver.ThreadingTCPServer,), {"allow_reuse_address": True}, ) server = server_class((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()