feat(history): add bounded persona semantic review

engineering/persona-history-runtime/config/BS-SH-005.json

@@ -10,6 +10,10 @@
   "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": 300,
+  "semantic_review_batch_size": 4,
+  "semantic_excerpt_bytes": 6000,
   "sources": [
     {
       "id": "GPT-LANGUAGE-CHAOS-ORIGINAL",

engineering/persona-history-runtime/runtime/guanghu_history_runtime.py

@@ -16,6 +16,7 @@ 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
@@ -35,6 +36,14 @@ SOURCE_WAITING_STATUSES = {
     "WAITING_FOR_SOURCE",
     "WAITING_FOR_SOURCE_ACCEPTANCE",
 }
+SEMANTIC_DECISIONS = {
+    "KEEP",
+    "RELATE",
+    "PENDING",
+    "LANGUAGE_SIMULATION",
+    "REALITY_FACT",
+    "PERSONA_MEMORY",
+}
 
 
 def now_iso() -> str:
@@ -72,6 +81,38 @@ def blocks_later_history(status: str) -> bool:
     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]]:
@@ -423,6 +464,105 @@ class Store:
             ).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(
@@ -517,6 +657,156 @@ class Runtime:
         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"])
@@ -765,6 +1055,7 @@ class Runtime:
                 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:

engineering/persona-history-runtime/runtime/test_guanghu_history_runtime.py

@@ -47,6 +47,26 @@ class RuntimeTests(unittest.TestCase):
         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)