guanghu-ice-heart/server-tools/guanghu-era-time/guanghu_era_time.py

#!/usr/bin/env python3
"""Deterministic Guanghu Era living clock and verifiable history reader."""

from __future__ import annotations

import argparse
import hashlib
import json
import subprocess
import time
from datetime import date, datetime, timezone
from pathlib import Path
from zoneinfo import ZoneInfo


BEIJING = ZoneInfo("Asia/Shanghai")
EPOCH = datetime(2025, 4, 26, 0, 0, 0, tzinfo=BEIJING)
EPOCH_DATE = EPOCH.date()
DAY_MS = 86_400_000
REPO = Path(__file__).resolve().parents[2]
MAP_PATH = REPO / "routing/guanghu-era-living-time-map.json"
HISTORY_PATH = REPO / "time-control/history-index.json"


def stable(value: object) -> bytes:
    return (json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + "\n").encode()


def digest(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()


def parse_instant(value: str | None) -> datetime:
    if value is None:
        return datetime.now(timezone.utc).astimezone(BEIJING)
    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    if parsed.tzinfo is None:
        raise ValueError("INSTANT_TIMEZONE_REQUIRED")
    return parsed.astimezone(BEIJING)


def snapshot(now: datetime, persona_id: str | None = None, birth: date | None = None) -> dict:
    elapsed_ms = int((now.astimezone(timezone.utc) - EPOCH.astimezone(timezone.utc)).total_seconds() * 1000)
    if elapsed_ms < 0:
        raise ValueError("CURRENT_TIME_PRECEDES_GUANGHU_EPOCH")
    persona = None
    if persona_id:
        if birth is None:
            raise ValueError("PERSONA_BIRTH_REQUIRED")
        age_days = (now.date() - birth).days
        if age_days < 0:
            raise ValueError("CURRENT_TIME_PRECEDES_PERSONA_BIRTH")
        persona = {
            "persona_id": persona_id,
            "birth_date_beijing": birth.isoformat(),
            "age_days_completed": age_days,
            "age_day_number": age_days + 1,
        }
    value = {
        "schema": "guanghu.era-living-time-snapshot/v1",
        "state": "LIVE_REALITY_TIME_DERIVED",
        "observed_at": now.isoformat(timespec="milliseconds"),
        "observed_at_unix_ms": int(now.timestamp() * 1000),
        "world": {
            "id": "SYS-GLW-0001",
            "era_id": "GUANGHU-ERA-0001",
            "visible_birth_date_beijing": EPOCH_DATE.isoformat(),
            "birth_fact_precision": "DATE",
            "exact_birth_instant": None,
            "calendar_epoch_boundary": EPOCH.isoformat(),
            "calendar_epoch_boundary_kind": "DETERMINISTIC_COMPUTATION_CONVENTION_NOT_CLAIMED_BIRTH_INSTANT",
            "era_day": elapsed_ms // DAY_MS + 1,
            "elapsed_days_completed": elapsed_ms // DAY_MS,
            "elapsed_milliseconds": elapsed_ms,
        },
        "persona": persona,
        "time_control_channel": {
            "id": "CH-GLW-TIME-0001",
            "path": "glw://time/control",
            "history_index": str(HISTORY_PATH),
        },
        "clock": {
            "source": "REALITY_SYSTEM_CLOCK_UTC",
            "stored_tick_counter": False,
            "persona_wake_required": False,
            "scheduler_required": False,
        },
    }
    value["snapshot_sha256"] = hashlib.sha256(stable(value)).hexdigest()
    return value


def validate_history() -> dict:
    history = json.loads(HISTORY_PATH.read_text())
    if history.get("schema") != "guanghu.time-control-history-index/v1":
        raise ValueError("HISTORY_SCHEMA_INVALID")
    checked = []
    for entry in history.get("entries", []):
        paths = []
        for relative in entry.get("evidence_paths", []):
            target = (REPO / relative).resolve()
            if REPO.resolve() not in target.parents or not target.is_file() or target.is_symlink():
                raise ValueError(f"HISTORY_EVIDENCE_INVALID:{relative}")
            paths.append({"path": relative, "sha256": digest(target)})
        commit = entry.get("commit_sha")
        if commit:
            subprocess.run(["git", "-C", str(REPO), "cat-file", "-e", f"{commit}^{{commit}}"], check=True, capture_output=True)
        checked.append({"event_id": entry["event_id"], "evidence": paths, "commit_sha": commit})
    return {"outcome": "PASS", "channel_id": history["channel_id"], "entries": checked, "history_sha256": digest(HISTORY_PATH)}


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("command", choices=["now", "stream", "history", "verify"])
    parser.add_argument("--at")
    parser.add_argument("--persona-id")
    parser.add_argument("--birth-date")
    parser.add_argument("--count", type=int, default=5)
    parser.add_argument("--interval-ms", type=int, default=100)
    args = parser.parse_args()
    birth = date.fromisoformat(args.birth_date) if args.birth_date else None
    if args.command == "now":
        print(json.dumps(snapshot(parse_instant(args.at), args.persona_id, birth), ensure_ascii=False, indent=2))
    elif args.command == "stream":
        if args.at:
            raise ValueError("STREAM_AT_OVERRIDE_FORBIDDEN")
        if not 1 <= args.count <= 100 or not 10 <= args.interval_ms <= 60_000:
            raise ValueError("STREAM_BOUNDS_INVALID")
        for _ in range(args.count):
            print(json.dumps(snapshot(parse_instant(None), args.persona_id, birth), ensure_ascii=False, separators=(",", ":")), flush=True)
            time.sleep(args.interval_ms / 1000)
    elif args.command == "history":
        print(json.dumps(validate_history(), ensure_ascii=False, indent=2))
    else:
        first = snapshot(parse_instant(None), args.persona_id, birth)
        time.sleep(0.02)
        second = snapshot(parse_instant(None), args.persona_id, birth)
        if second["world"]["elapsed_milliseconds"] <= first["world"]["elapsed_milliseconds"]:
            raise ValueError("LIVING_TIME_NOT_ADVANCING")
        result = validate_history()
        result.update({
            "clock": "PASS_MONOTONIC_LIVE_MILLISECONDS",
            "delta_ms": second["world"]["elapsed_milliseconds"] - first["world"]["elapsed_milliseconds"],
            "map_sha256": digest(MAP_PATH),
            "first": first,
            "second": second,
        })
        print(json.dumps(result, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()