diff --git a/engineering/persona-history-runtime/config/BS-SH-005.json b/engineering/persona-history-runtime/config/BS-SH-005.json index 0ce9524..6182721 100644 --- a/engineering/persona-history-runtime/config/BS-SH-005.json +++ b/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", diff --git a/engineering/persona-history-runtime/runtime/guanghu_history_runtime.py b/engineering/persona-history-runtime/runtime/guanghu_history_runtime.py index 9010aa1..22cf6ff 100644 --- a/engineering/persona-history-runtime/runtime/guanghu_history_runtime.py +++ b/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: 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 88707c2..4f6f892 100644 --- a/engineering/persona-history-runtime/runtime/test_guanghu_history_runtime.py +++ b/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)