feat: add private library and living Guanghu time

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

@@ -0,0 +1,152 @@
+#!/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()

server-tools/guanghu-era-time/test_guanghu_era_time.py

@@ -0,0 +1,38 @@
+import importlib.util
+import unittest
+from datetime import date, datetime
+from pathlib import Path
+from zoneinfo import ZoneInfo
+
+
+MODULE = Path(__file__).with_name("guanghu_era_time.py")
+SPEC = importlib.util.spec_from_file_location("guanghu_era_time", MODULE)
+clock = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(clock)
+
+
+class GuanghuEraTimeTests(unittest.TestCase):
+    def test_birth_is_era_day_one(self):
+        value = clock.snapshot(datetime(2025, 4, 26, tzinfo=ZoneInfo("Asia/Shanghai")))
+        self.assertEqual(value["world"]["era_day"], 1)
+        self.assertEqual(value["world"]["elapsed_milliseconds"], 0)
+
+    def test_one_millisecond_flows(self):
+        value = clock.snapshot(datetime.fromisoformat("2025-04-26T00:00:00.001+08:00"))
+        self.assertEqual(value["world"]["elapsed_milliseconds"], 1)
+
+    def test_world_precedes_persona(self):
+        value = clock.snapshot(datetime.fromisoformat("2026-09-09T12:00:00+08:00"), "ICE-P-ZY001", date(2026, 3, 5))
+        self.assertGreater(value["world"]["elapsed_days_completed"], value["persona"]["age_days_completed"])
+
+    def test_time_is_not_wake_counter(self):
+        value = clock.snapshot(datetime.fromisoformat("2026-09-09T12:00:00+08:00"))
+        self.assertFalse(value["clock"]["stored_tick_counter"])
+        self.assertFalse(value["clock"]["persona_wake_required"])
+
+    def test_history_evidence_and_commits(self):
+        self.assertEqual(clock.validate_history()["outcome"], "PASS")
+
+
+if __name__ == "__main__":
+    unittest.main()