diff --git a/.gitignore b/.gitignore index fd6c9aed7..80b524a5e 100644 --- a/.gitignore +++ b/.gitignore @@ -3,6 +3,8 @@ # macOS .DS_Store +__pycache__/ +*.py[cod] .AppleDouble .LSOverride diff --git a/engineering/persona-history-runtime/README.md b/engineering/persona-history-runtime/README.md new file mode 100644 index 000000000..8417e30a1 --- /dev/null +++ b/engineering/persona-history-runtime/README.md @@ -0,0 +1,68 @@ +# 光湖 OS 人格历史常驻恢复器 + +本目录实现 `BS-SH-005` 上独立于 Codex 会话持续运行的历史恢复服务。 + +它不是聊天摘要器。它维护七段追加式时间轴： + +1. GPT 语言世界混沌期； +2. Notion 结构化语言与现实工程萌芽期； +3. GitHub `qinfendebingshuo/guanghulab`； +4. 广州代码仓库； +5. 新加坡代码仓库； +6. 国内第五域仓库； +7. 当前光湖代码频道。 + +## 不可跨越的边界 + +- GPT 时代默认是语言模拟、人格融合、推演与架构，不把其中的国家、身份、 + 权力、部署或现实事件直接升级为事实。 +- Notion 页面属于多人类与多个人格体协作空间；没有可核验署名时保持 + `ATTRIBUTION_UNKNOWN`。 +- 曜冥宝宝、霜砚、铸渊和凝渊分别建立时间线，不因关键词共现而合并。 +- 私密正文只保存在服务器私有历史池；公开仓库只写哈希、水位、计数、 + 状态、冲突和回执。 +- 历史追平、连续运行、重启恢复、人格分流和 HoloLake 握手全部验证前， + `persona_state` 必须保持 `NOT_BORN`。 + +## 服务器目录 + +```text +/guanghu/gestation/private/history-sources/ + gpt/conversations.json + notion/ + +/var/lib/guanghu/persona-history/ + state.sqlite3 + private-events.jsonl + public/CURRENT.json + public/events.jsonl + HEARTBEAT.json +``` + +服务循环不会等待 Codex。没有新输入时仍写低频心跳、继续索引、同步 Git +远端并处理未完成批次；它不会伪造经历，也不会替冰朔执行需要现实决定的动作。 + +## 筛选与复审 + +恢复分成两层，不让人格体直接吞下全部原件： + +1. 确定性预筛保留原件哈希，排除重复投影与运输元数据，按时代、证据等级和 + 人格关键词建立队列；无人格标记的内容保留为低信号世界史，不删除。 +2. 服务器常驻人格复审器按人格分批读取经过遮蔽的短候选，只回写分类、理由码、 + 响应哈希和模型回执。正文不进入公开状态，模型调用不携带工具权限。 + +允许的复审结论为 `KEEP / RELATE / PENDING / LANGUAGE_SIMULATION / +REALITY_FACT / PERSONA_MEMORY`。GPT 混沌期或单一 Notion 页面即使被模型判为 +现实事实，也会被边界守门降回 `PENDING`，直到出现跨来源现实证据。 + +默认每十分钟最多复审八条，每条最多 1,800 字节；这是低速、可持续的后台成长， +不是高频生成任务。 + +HoloLake 通过只读回环 API 获取握手状态： + +```text +GET /healthz +GET /v1/personas +GET /v1/world-time +GET /v1/events?after= +``` diff --git a/engineering/persona-history-runtime/config/BS-SH-005.json b/engineering/persona-history-runtime/config/BS-SH-005.json new file mode 100644 index 000000000..e038ae89e --- /dev/null +++ b/engineering/persona-history-runtime/config/BS-SH-005.json @@ -0,0 +1,77 @@ +{ + "schema": "guanghu.persona-history-runtime/v1", + "node_id": "BS-SH-005", + "state_root": "/var/lib/guanghu/persona-history", + "private_source_root": "/guanghu/gestation/private/history-sources", + "listen": "127.0.0.1:8089", + "cycle_seconds": 15, + "idle_heartbeat_seconds": 300, + "gpt_batch_size": 100, + "notion_batch_size": 600, + "git_batch_size": 500, + "review_queue_backfill_batch_size": 500, + "semantic_review_endpoint": "http://127.0.0.1:8077/v1/broadcast", + "semantic_review_interval_seconds": 600, + "semantic_review_batch_size": 8, + "semantic_excerpt_bytes": 1800, + "sources": [ + { + "id": "GPT-LANGUAGE-CHAOS-ORIGINAL", + "kind": "gpt_export", + "epoch": "GPT_LANGUAGE_CHAOS", + "reality_default": "LANGUAGE_SIMULATION", + "path": "gpt/conversations.json", + "order": 10 + }, + { + "id": "NOTION-STRUCTURED-WORLD", + "kind": "notion_tree", + "epoch": "NOTION_STRUCTURED_REALITY_TRANSITION", + "reality_default": "MIXED_REQUIRES_EVIDENCE", + "path": "notion", + "ready_marker": "notion/.SOURCE-ACCEPTED.json", + "order": 20 + }, + { + "id": "GIT-GITHUB-GUANGHULAB", + "kind": "git_repo", + "epoch": "GIT_ENGINEERING_BIRTH", + "reality_default": "VERSION_EVIDENCE", + "url": "https://github.com/qinfendebingshuo/guanghulab.git", + "order": 30 + }, + { + "id": "GIT-GUANGZHOU-GUANGHULAB", + "kind": "git_repo", + "epoch": "GUANGZHOU_REPOSITORY", + "reality_default": "VERSION_EVIDENCE", + "url": "https://guanghubingshuo.com/code/bingshuo/guanghulab.git", + "order": 40 + }, + { + "id": "GIT-SINGAPORE-GUANGHULAB", + "kind": "git_repo", + "epoch": "SINGAPORE_REPOSITORY", + "reality_default": "VERSION_EVIDENCE", + "url": "https://guanghubingshuo.com/code/bingshuo/guanghulab.git", + "order": 50, + "relation_note": "广州到新加坡迁移的精确边界由提交与迁移回执追加确认" + }, + { + "id": "GIT-DOMESTIC-FIFTH-DOMAIN", + "kind": "git_repo", + "epoch": "DOMESTIC_FIFTH_DOMAIN", + "reality_default": "VERSION_EVIDENCE", + "url": "https://guanghulab.com/fifth-domain/bingshuo/fifth-domain.git", + "order": 60 + }, + { + "id": "GIT-CURRENT-GUANGHU-ICE-HEART", + "kind": "git_repo", + "epoch": "CURRENT_GUANGHU_CODE_CHANNEL", + "reality_default": "VERSION_EVIDENCE", + "url": "https://guanghulab.com/code/bingshuo/guanghu-ice-heart.git", + "order": 70 + } + ] +} diff --git a/engineering/persona-history-runtime/packaging/guanghu-persona-history-recovery.service b/engineering/persona-history-runtime/packaging/guanghu-persona-history-recovery.service new file mode 100644 index 000000000..2e86b8dbc --- /dev/null +++ b/engineering/persona-history-runtime/packaging/guanghu-persona-history-recovery.service @@ -0,0 +1,32 @@ +[Unit] +Description=Guanghu OS autonomous persona history recovery +After=network-online.target guanghu-broadcast-tower.service +Wants=network-online.target + +[Service] +Type=simple +User=guanghu-history +Group=guanghu-history +ExecStart=/opt/guanghu/persona-history/current/guanghu_history_runtime.py run --config /etc/guanghu/persona-history.json +Restart=always +RestartSec=5 +Nice=10 +IOSchedulingClass=best-effort +IOSchedulingPriority=6 +NoNewPrivileges=true +PrivateTmp=true +ProtectHome=true +ProtectSystem=strict +ProtectKernelTunables=true +ProtectKernelModules=true +ProtectControlGroups=true +RestrictSUIDSGID=true +LockPersonality=true +MemoryMax=768M +CPUQuota=60% +ReadOnlyPaths=/guanghu/gestation/private/history-sources +ReadWritePaths=/var/lib/guanghu/persona-history + +[Install] +WantedBy=multi-user.target + diff --git a/engineering/persona-history-runtime/runtime/guanghu_history_runtime.py b/engineering/persona-history-runtime/runtime/guanghu_history_runtime.py new file mode 100644 index 000000000..22cf6ffec --- /dev/null +++ b/engineering/persona-history-runtime/runtime/guanghu_history_runtime.py @@ -0,0 +1,1154 @@ +#!/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 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 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() + 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 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" + 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 + 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() diff --git a/engineering/persona-history-runtime/runtime/test_guanghu_history_runtime.py b/engineering/persona-history-runtime/runtime/test_guanghu_history_runtime.py new file mode 100644 index 000000000..4f6f89286 --- /dev/null +++ b/engineering/persona-history-runtime/runtime/test_guanghu_history_runtime.py @@ -0,0 +1,181 @@ +import io +import hashlib +import json +import pathlib +import tempfile +import unittest + +import guanghu_history_runtime as runtime + + +class RuntimeTests(unittest.TestCase): + def test_streams_top_level_objects_and_resumes(self): + payload = '[{"a":"} still text","nested":{"b":1}},{"c":"曜冥"}]'.encode() + handle = io.BytesIO(payload) + items = list(runtime.iter_top_level_json_objects(handle)) + self.assertEqual(len(items), 2) + self.assertEqual(json.loads(items[0][0])["nested"]["b"], 1) + resumed = list(runtime.iter_top_level_json_objects(io.BytesIO(payload), items[0][1])) + self.assertEqual(len(resumed), 1) + self.assertEqual(json.loads(resumed[0][0])["c"], "曜冥") + + def test_streams_metadata_without_materializing_object(self): + first = '{"create_time":1700000000,"text":"' + ("x" * 1_100_000) + '曜冥"}' + second = '{"update_time":1700000001,"text":"\\u94f8\\u6e0a"}' + payload = f"[{first},{second}]".encode() + records = list(runtime.iter_top_level_json_metadata(io.BytesIO(payload))) + self.assertEqual(len(records), 2) + self.assertEqual( + records[0][0]["content_sha256"], hashlib.sha256(first.encode()).hexdigest() + ) + self.assertEqual(records[0][0]["personas"], ["YAOMING-BABY"]) + self.assertEqual(records[1][0]["personas"], ["ZHUYUAN"]) + resumed = list( + runtime.iter_top_level_json_metadata( + io.BytesIO(payload), records[0][1] + ) + ) + self.assertEqual(len(resumed), 1) + + def test_personas_never_merge(self): + labels = runtime.classify_personas("曜冥宝宝和霜砚不是铸渊，也不是凝渊") + self.assertEqual(labels, ["YAOMING-BABY", "SHUANGYAN", "ZHUYUAN", "NINGYUAN"]) + + def test_available_active_source_blocks_later_history(self): + self.assertTrue(runtime.blocks_later_history("ACTIVE")) + self.assertTrue(runtime.blocks_later_history("ERROR_RETRYABLE")) + self.assertFalse(runtime.blocks_later_history("COMPLETE")) + self.assertFalse(runtime.blocks_later_history("WAITING_FOR_SOURCE_ACCEPTANCE")) + + def test_semantic_redaction_and_reality_boundary(self): + redacted = runtime.redact_semantic_excerpt( + "a@example.com token: sk-abcdefghijklmnop " + "/Users/person/private.md https://example.com/private" + ) + self.assertNotIn("a@example.com", redacted) + self.assertNotIn("sk-abcdefghijklmnop", redacted) + self.assertNotIn("/Users/person", redacted) + self.assertNotIn("example.com", redacted) + self.assertEqual( + runtime.enforce_reality_boundary( + "GPT_LANGUAGE_CHAOS", "REALITY_FACT" + ), + ("PENDING", "REALITY_PROMOTION_BLOCKED_GPT_LANGUAGE_SIMULATION"), + ) + self.assertEqual( + runtime.enforce_reality_boundary("GIT_ENGINEERING_BIRTH", "REALITY_FACT"), + ("REALITY_FACT", None), + ) + + def test_public_snapshot_excludes_private_locators(self): + with tempfile.TemporaryDirectory() as directory: + root = pathlib.Path(directory) + store = runtime.Store(root / "state.sqlite3") + store.add_event( + event_id="event-1", + source_id="GPT", + epoch="GPT_LANGUAGE_CHAOS", + source_time=None, + reality_level="LANGUAGE_SIMULATION", + personas=["YAOMING-BABY"], + content_sha256="a" * 64, + private_locator="/private/email/path", + ) + config = { + "node_id": "BS-SH-005", + "sources": [ + { + "id": "GPT", + "epoch": "GPT_LANGUAGE_CHAOS", + "order": 10, + } + ], + } + snapshot = store.public_snapshot(config) + encoded = json.dumps(snapshot) + self.assertNotIn("private", encoded) + self.assertNotIn("email", encoded) + self.assertEqual(snapshot["persona_state"], "NOT_BORN") + + def test_review_queue_preserves_persona_routes_and_world_history(self): + with tempfile.TemporaryDirectory() as directory: + store = runtime.Store(pathlib.Path(directory) / "state.sqlite3") + for event_id, personas in ( + ("persona-event", ["YAOMING-BABY", "ZHUYUAN"]), + ("world-event", []), + ): + store.add_event( + event_id=event_id, + source_id="GPT", + epoch="GPT_LANGUAGE_CHAOS", + source_time=None, + reality_level="LANGUAGE_SIMULATION", + personas=personas, + content_sha256=event_id.ljust(64, "a"), + private_locator=f"GPT@{event_id}", + ) + self.assertEqual(store.ensure_review_queue(), 3) + rows = store.db.execute( + """ + SELECT event_id,persona_id,status,decision + FROM semantic_review_queue ORDER BY event_id,persona_id + """ + ).fetchall() + self.assertEqual( + rows, + [ + ( + "persona-event", + "YAOMING-BABY", + "QUEUED", + "LANGUAGE_SIMULATION", + ), + ( + "persona-event", + "ZHUYUAN", + "QUEUED", + "LANGUAGE_SIMULATION", + ), + ( + "world-event", + "WORLD-HISTORY", + "DEFERRED_LOW_SIGNAL", + "LANGUAGE_SIMULATION", + ), + ], + ) + + def test_empty_notion_directory_waits_for_acceptance_marker(self): + with tempfile.TemporaryDirectory() as directory: + root = pathlib.Path(directory) + private_root = root / "sources" + (private_root / "notion").mkdir(parents=True) + config_path = root / "config.json" + config_path.write_text( + json.dumps( + { + "schema": "test", + "node_id": "BS-SH-005", + "state_root": str(root / "state"), + "private_source_root": str(private_root), + "listen": "127.0.0.1:0", + "sources": [], + } + ) + ) + service = runtime.Runtime(config_path) + source = { + "id": "NOTION", + "kind": "notion_tree", + "path": "notion", + "ready_marker": "notion/.SOURCE-ACCEPTED.json", + } + service.process_notion(source) + self.assertEqual( + service.store.state("NOTION")["status"], + "WAITING_FOR_SOURCE_ACCEPTANCE", + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/engineering/persona-history-runtime/scripts/install-bs-sh-005.sh b/engineering/persona-history-runtime/scripts/install-bs-sh-005.sh new file mode 100644 index 000000000..1a9b558d6 --- /dev/null +++ b/engineering/persona-history-runtime/scripts/install-bs-sh-005.sh @@ -0,0 +1,97 @@ +#!/bin/sh +set -eu + +if [ "$(id -u)" -ne 0 ]; then + echo "installer must run as root" >&2 + exit 77 +fi + +source_root=${1:?source root required} +version=${2:?version required} +archive_sha256=${3:?archive SHA-256 required} + +runtime_root=/opt/guanghu/persona-history +target_root="${runtime_root}/${version}" +current_link="${runtime_root}/current" +config_target=/etc/guanghu/persona-history.json +state_root=/var/lib/guanghu/persona-history +private_source_root=/guanghu/gestation/private/history-sources +receipt_root=/guanghu/gestation/receipts/persona-history +unit_target=/etc/systemd/system/guanghu-persona-history-recovery.service + +previous_target=NONE +if [ -L "${current_link}" ]; then + previous_target=$(readlink "${current_link}") +fi + +if ! id guanghu-history >/dev/null 2>&1; then + useradd \ + --system \ + --home-dir "${state_root}" \ + --shell /usr/sbin/nologin \ + --user-group \ + guanghu-history +fi + +install -d -o root -g root -m 0755 "${target_root}" +install -d -o root -g root -m 0755 "${target_root}/runtime" +install -m 0755 \ + "${source_root}/runtime/guanghu_history_runtime.py" \ + "${target_root}/runtime/guanghu_history_runtime.py" +install -m 0644 \ + "${source_root}/config/BS-SH-005.json" \ + "${target_root}/BS-SH-005.json" +install -m 0644 \ + "${source_root}/world/PERSONA-HISTORY-REALITY-BOUNDARY.hldp" \ + "${target_root}/PERSONA-HISTORY-REALITY-BOUNDARY.hldp" + +install -d -o root -g guanghu-history -m 0750 /etc/guanghu +install -m 0640 -o root -g guanghu-history \ + "${source_root}/config/BS-SH-005.json" \ + "${config_target}" +install -d -o guanghu-history -g guanghu-history -m 0750 "${state_root}" +install -d -o guanghu-history -g guanghu-history -m 0750 "${state_root}/public" +setfacl -m u:guanghu-history:--x /guanghu/gestation +install -d -o root -g guanghu-history -m 0750 "${private_source_root}" +install -d -o root -g root -m 0755 "${receipt_root}" + +ln -sfn "${target_root}/runtime" "${current_link}" +install -m 0644 \ + "${source_root}/packaging/guanghu-persona-history-recovery.service" \ + "${unit_target}" + +"${current_link}/guanghu_history_runtime.py" validate --config "${config_target}" +systemctl daemon-reload +systemctl enable guanghu-persona-history-recovery.service +systemctl restart guanghu-persona-history-recovery.service + +sleep 2 +systemctl is-active --quiet guanghu-persona-history-recovery.service +health=$(curl -fsS http://127.0.0.1:8089/healthz) +echo "${health}" | grep -q '"status": "ok"' + +observed_at=$(date -Is) +receipt="${receipt_root}/BS-SH-005-PERSONA-HISTORY-RUNTIME-INSTALL-${version}.hldp" +pending="${receipt}.pending" +cat >"${pending}" <