From 41365287753bef34739b25ae89b8b6799168f65e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=86=B0=E6=9C=94?= <565183519@qq.com> Date: Sat, 1 Aug 2026 18:38:16 +0800 Subject: [PATCH] fix(history): bound ingestion and stage semantic review --- .../config/BS-SH-005.json | 4 +- .../runtime/guanghu_history_runtime.py | 234 ++++++++++++++++-- .../runtime/test_guanghu_history_runtime.py | 99 +++++++- .../scripts/install-bs-sh-005.sh | 1 + 4 files changed, 317 insertions(+), 21 deletions(-) diff --git a/engineering/persona-history-runtime/config/BS-SH-005.json b/engineering/persona-history-runtime/config/BS-SH-005.json index 282ea73..0ce9524 100644 --- a/engineering/persona-history-runtime/config/BS-SH-005.json +++ b/engineering/persona-history-runtime/config/BS-SH-005.json @@ -6,8 +6,10 @@ "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, "sources": [ { "id": "GPT-LANGUAGE-CHAOS-ORIGINAL", @@ -23,6 +25,7 @@ "epoch": "NOTION_STRUCTURED_REALITY_TRANSITION", "reality_default": "MIXED_REQUIRES_EVIDENCE", "path": "notion", + "ready_marker": "notion/.SOURCE-ACCEPTED.json", "order": 20 }, { @@ -68,4 +71,3 @@ } ] } - diff --git a/engineering/persona-history-runtime/runtime/guanghu_history_runtime.py b/engineering/persona-history-runtime/runtime/guanghu_history_runtime.py index 22c399e..513519e 100644 --- a/engineering/persona-history-runtime/runtime/guanghu_history_runtime.py +++ b/engineering/persona-history-runtime/runtime/guanghu_history_runtime.py @@ -8,6 +8,7 @@ import hashlib import json import os import pathlib +import re import shutil import socketserver import sqlite3 @@ -27,6 +28,8 @@ PERSONAS = { "NINGYUAN": ("凝渊",), } PRIVATE_MARKERS = ("email", "token", "password", "secret", "api_key", "private_key") +STREAM_BUFFER_BYTES = 1024 * 1024 +GPT_METADATA_SAMPLE_BYTES = 128 * 1024 def now_iso() -> str: @@ -110,6 +113,115 @@ def iter_top_level_json_objects( 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) @@ -150,6 +262,19 @@ class Store: 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() @@ -233,6 +358,61 @@ class Store: 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 add_error(self, source_id: str, error: Exception) -> None: message = str(error).replace("\n", " ")[:1000] self.db.execute( @@ -280,6 +460,12 @@ class Store: 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(), } @@ -334,19 +520,14 @@ class Runtime: 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 raw, next_offset in iter_top_level_json_objects(handle, offset): - content_hash = hashlib.sha256(raw).hexdigest() - obj = json.loads(raw) - searchable = " ".join( - ( - str(obj.get("title", "")), - json.dumps(obj.get("mapping", {}), ensure_ascii=False)[:2_000_000], - ) - ) - personas = classify_personas(searchable) - source_time = obj.get("create_time") or obj.get("update_time") - if isinstance(source_time, (int, float)): + 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() @@ -362,8 +543,9 @@ class Runtime: private_locator=f"{source['id']}@byte:{offset}-{next_offset}", ) processed += 1 + handled += 1 offset = next_offset - if processed % 25 == 0: + if processed % 25 == 0 or handled >= batch_size: self.store.update_state( source["id"], status="ACTIVE", @@ -371,6 +553,8 @@ class Runtime: processed=processed, ) self.write_public() + if handled >= batch_size: + return if self.stop.is_set(): return self.store.update_state( @@ -379,7 +563,7 @@ class Runtime: def notion_manifest(self, source: dict) -> pathlib.Path: manifest = self.state_root / "private" / f"{source['id']}-files.jsonl" - if manifest.exists(): + if manifest.exists() and manifest.stat().st_size > 0: return manifest root = self.private_root / source["path"] if not root.is_dir(): @@ -389,7 +573,9 @@ class Runtime: paths = sorted( path.relative_to(root).as_posix() for path in root.rglob("*") - if path.is_file() and path.name != ".DS_Store" + 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: @@ -400,10 +586,13 @@ class Runtime: def process_notion(self, source: dict) -> None: root = self.private_root / source["path"] state = self.store.state(source["id"]) - if not root.is_dir(): + 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", + status="WAITING_FOR_SOURCE_ACCEPTANCE", cursor=state["cursor"], processed=state["processed"], ) @@ -557,6 +746,9 @@ class Runtime: 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(): @@ -612,7 +804,12 @@ class ApiHandler(BaseHTTPRequestHandler): def start_api(runtime: Runtime) -> None: host, port = runtime.config["listen"].rsplit(":", 1) handler = type("RuntimeApiHandler", (ApiHandler,), {"runtime": runtime}) - server = socketserver.ThreadingTCPServer((host, int(port)), handler) + 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() @@ -652,4 +849,3 @@ def main() -> None: 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 index 342138a..c0b435f 100644 --- a/engineering/persona-history-runtime/runtime/test_guanghu_history_runtime.py +++ b/engineering/persona-history-runtime/runtime/test_guanghu_history_runtime.py @@ -1,4 +1,5 @@ import io +import hashlib import json import pathlib import tempfile @@ -18,6 +19,24 @@ class RuntimeTests(unittest.TestCase): 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"]) @@ -52,7 +71,85 @@ class RuntimeTests(unittest.TestCase): 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 index 16fc6ae..1a9b558 100644 --- a/engineering/persona-history-runtime/scripts/install-bs-sh-005.sh +++ b/engineering/persona-history-runtime/scripts/install-bs-sh-005.sh @@ -51,6 +51,7 @@ install -m 0640 -o root -g guanghu-history \ "${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}"