hololake-system-architecture/engineering/persona-history-runtime/runtime/guanghu_history_runtime.py

#!/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
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


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")


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 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
            },
            "semantic_review": {
                "policy": "DETERMINISTIC_PREFILTER_THEN_PERSONA_REVIEW",
                "counts": self.review_counts(),
                "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 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"
        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:
        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
        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()