1174 lines
42 KiB
Python
1174 lines
42 KiB
Python
#!/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 retry_backoff_elapsed(
|
||
updated_at: str | None, backoff_seconds: int, now_unix: float | None = None
|
||
) -> bool:
|
||
"""Keep retryable sources alive without writing one failure per runtime cycle."""
|
||
if not updated_at:
|
||
return True
|
||
try:
|
||
last_attempt = datetime.fromisoformat(updated_at).timestamp()
|
||
except (TypeError, ValueError):
|
||
return True
|
||
current = time.time() if now_unix is None else now_unix
|
||
return current - last_attempt >= backoff_seconds
|
||
|
||
|
||
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:
|
||
state = self.store.state(source["id"])
|
||
if state["status"] == "ERROR_RETRYABLE" and not retry_backoff_elapsed(
|
||
state["updated_at"],
|
||
int(self.config.get("source_retry_backoff_seconds", 1800)),
|
||
):
|
||
return
|
||
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()
|