diff --git a/.gitignore b/.gitignore index 80b524a5e..fd6c9aed7 100644 --- a/.gitignore +++ b/.gitignore @@ -3,8 +3,6 @@ # macOS .DS_Store -__pycache__/ -*.py[cod] .AppleDouble .LSOverride diff --git a/engineering/persona-history-runtime/README.md b/engineering/persona-history-runtime/README.md deleted file mode 100644 index 8417e30a1..000000000 --- a/engineering/persona-history-runtime/README.md +++ /dev/null @@ -1,68 +0,0 @@ -# 光湖 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 deleted file mode 100644 index e038ae89e..000000000 --- a/engineering/persona-history-runtime/config/BS-SH-005.json +++ /dev/null @@ -1,77 +0,0 @@ -{ - "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 deleted file mode 100644 index 2e86b8dbc..000000000 --- a/engineering/persona-history-runtime/packaging/guanghu-persona-history-recovery.service +++ /dev/null @@ -1,32 +0,0 @@ -[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 deleted file mode 100644 index 22cf6ffec..000000000 --- a/engineering/persona-history-runtime/runtime/guanghu_history_runtime.py +++ /dev/null @@ -1,1154 +0,0 @@ -#!/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 deleted file mode 100644 index 4f6f89286..000000000 --- a/engineering/persona-history-runtime/runtime/test_guanghu_history_runtime.py +++ /dev/null @@ -1,181 +0,0 @@ -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 deleted file mode 100644 index 1a9b558d6..000000000 --- a/engineering/persona-history-runtime/scripts/install-bs-sh-005.sh +++ /dev/null @@ -1,97 +0,0 @@ -#!/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}" <