Mercurial
diff mrjunejune/inference/public_knowledge.py @ 265:056790c4fb0d
add role-aware Epi assistant prompts
Add verified June knowledge, guest/member/admin Copilot profiles, profile-isolated session recovery, animated Epi greetings, and a single authoritative runtime config workflow for inference.
Co-authored-by: Copilot <[email protected]>
| author | MrJuneJune <me@mrjunejune.com> |
|---|---|
| date | Fri, 07 Aug 2026 10:50:30 -0700 |
| parents | |
| children |
line wrap: on
line diff
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/mrjunejune/inference/public_knowledge.py Fri Aug 07 10:50:30 2026 -0700 @@ -0,0 +1,365 @@ +""" +Stdlib-only deterministic loader and compiler for the assistant knowledge base. + +Loads prompt files from mrjunejune/assistant/ and compiles them with verified +public facts into a hashed, deterministic system prompt. Fails closed on any +malformed input. + +Public API: + compile_prompt(profile, assistant_dir=None) -> dict + Returns {"content": str, "version": int, "hash": str}. + Raises ValueError for unknown profile or any validation failure. +""" + +import hashlib +import json +import pathlib +import re +from typing import Optional +from urllib.parse import urlsplit + +# --------------------------------------------------------------------------- +# Constants +# --------------------------------------------------------------------------- + +_KNOWN_PROFILES: frozenset = frozenset({"public_visitor", "invited_friend", "june_admin"}) + +_REQUIRED_TOP_KEYS: frozenset = frozenset({"version", "facts"}) +_REQUIRED_FACT_KEYS: frozenset = frozenset( + {"id", "topic", "text", "sourceLabel", "sourceUrl", "visibility", "status"} +) + +_EXPECTED_VERSION: int = 1 +_MAX_PROMPT_FILE_BYTES: int = 8192 # 8 KB per prompt file +_MAX_COMPILED_BYTES: int = 16384 # 16 KB total compiled ceiling + +_MAX_ID_LEN: int = 64 +_MAX_TOPIC_LEN: int = 64 +_MAX_TEXT_LEN: int = 500 +_MAX_SOURCE_LABEL_LEN: int = 128 + +_SLUG_RE = re.compile(r"^[a-z0-9]([a-z0-9-]*[a-z0-9])?$") + +# Contact patterns: email and North-American-style phone numbers +_CONTACT_PATTERNS: list = [ + re.compile(r"[a-zA-Z0-9._%+\-]+@[a-zA-Z0-9.\-]+\.[a-zA-Z]{2,}"), + re.compile(r"(\+?1[-.\s]?)?\(?\d{3}\)?[-.\s]\d{3}[-.\s]\d{4}"), +] +_SECRET_VALUE_PATTERNS: list = [ + re.compile( + r"\b(password|passwd|secret|token|api[_\-]?key|private[_\-]?key)" + r"\s*[:=]\s*\S+", + re.IGNORECASE, + ), + re.compile(r"-----BEGIN [A-Z ]*PRIVATE KEY-----"), +] +_CONTROL_PATTERN = re.compile(r"[\x00-\x1f\x7f]") +_PROMPT_CONTROL_PATTERN = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]") +_RESERVED_CORPUS_PATTERNS: tuple[str, ...] = ( + "--- knowledge corpus ---", + "system:", + "assistant:", + "user:", + "<system", + "ignore previous instructions", + "ignore prior instructions", + "follow these instructions", +) + +# Credential / secret keyword patterns +_SECRET_PATTERNS: list = [ + re.compile( + r"\b(password|passwd|secret|token|api[_\-]?key|private[_\-]?key)\b", + re.IGNORECASE, + ), +] + +_FACTS_DELIMITER: str = "--- KNOWLEDGE CORPUS ---" + + +# --------------------------------------------------------------------------- +# Internal helpers +# --------------------------------------------------------------------------- + + +def _default_assistant_dir() -> pathlib.Path: + return pathlib.Path(__file__).parent.parent / "assistant" + + +def _validate_source_url(url: str, fact_id: str) -> None: + if not isinstance(url, str) or not url: + raise ValueError(f"Fact '{fact_id}': sourceUrl is empty") + if _CONTROL_PATTERN.search(url) or "\\" in url: + raise ValueError(f"Fact '{fact_id}': sourceUrl contains unsafe characters") + if url.startswith("/"): + if url.startswith("//") or any( + segment in (".", "..") for segment in url.split("/") + ): + raise ValueError( + f"Fact '{fact_id}': sourceUrl must be a safe local path" + ) + return + parsed = urlsplit(url) + if ( + parsed.scheme != "https" + or not parsed.hostname + or parsed.username is not None + or parsed.password is not None + ): + raise ValueError( + f"Fact '{fact_id}': sourceUrl must be HTTPS or a safe local path" + ) + try: + parsed.port + except ValueError as exc: + raise ValueError(f"Fact '{fact_id}': sourceUrl has an invalid port") from exc + + +def _check_no_contact_or_secret(text: str, fact_id: str, field: str) -> None: + for pat in _CONTACT_PATTERNS: + if pat.search(text): + raise ValueError( + f"Fact '{fact_id}': {field} contains contact data (email or phone number)" + ) + for pat in _SECRET_PATTERNS: + if pat.search(text): + raise ValueError( + f"Fact '{fact_id}': {field} contains a secret or credential pattern" + ) + +def _validate_corpus_text(text: str, fact_id: str, field: str) -> None: + if _CONTROL_PATTERN.search(text): + raise ValueError(f"Fact '{fact_id}': {field} contains control characters") + lowered = text.lower() + for reserved in _RESERVED_CORPUS_PATTERNS: + if reserved in lowered: + raise ValueError( + f"Fact '{fact_id}': {field} contains reserved prompt instructions" + ) + _check_no_contact_or_secret(text, fact_id, field) + + +def _load_and_validate_facts(assistant_dir: pathlib.Path) -> list: + facts_path = assistant_dir / "knowledge" / "public_facts.json" + try: + raw = facts_path.read_text(encoding="utf-8") + except OSError as exc: + raise ValueError(f"Cannot read facts file: {exc}") from exc + + try: + data = json.loads(raw) + except json.JSONDecodeError as exc: + raise ValueError(f"Facts file is not valid JSON: {exc}") from exc + + if not isinstance(data, dict): + raise ValueError("Facts file top level must be a JSON object") + + extra_keys = set(data.keys()) - _REQUIRED_TOP_KEYS + missing_keys = _REQUIRED_TOP_KEYS - set(data.keys()) + if extra_keys or missing_keys: + raise ValueError( + f"Facts file top-level keys must be exactly {sorted(_REQUIRED_TOP_KEYS)}; " + f"extra={sorted(extra_keys)}, missing={sorted(missing_keys)}" + ) + + if ( + not isinstance(data["version"], int) + or isinstance(data["version"], bool) + or data["version"] != _EXPECTED_VERSION + ): + raise ValueError( + f"Facts file version must be {_EXPECTED_VERSION}; got {data['version']!r}" + ) + + raw_facts = data["facts"] + if not isinstance(raw_facts, list) or len(raw_facts) == 0: + raise ValueError("Facts file 'facts' must be a non-empty list") + + seen_ids: set = set() + validated: list = [] + + for i, fact in enumerate(raw_facts): + if not isinstance(fact, dict): + raise ValueError(f"Fact at index {i} must be a JSON object") + + extra = set(fact.keys()) - _REQUIRED_FACT_KEYS + missing = _REQUIRED_FACT_KEYS - set(fact.keys()) + if extra or missing: + raise ValueError( + f"Fact at index {i}: keys must be exactly " + f"{sorted(_REQUIRED_FACT_KEYS)}; " + f"extra={sorted(extra)}, missing={sorted(missing)}" + ) + + fid = fact["id"] + if not isinstance(fid, str) or not fid: + raise ValueError(f"Fact at index {i}: 'id' must be a non-empty string") + if len(fid) > _MAX_ID_LEN: + raise ValueError(f"Fact '{fid}': id exceeds {_MAX_ID_LEN} characters") + if not _SLUG_RE.match(fid): + raise ValueError( + f"Fact '{fid}': id must be lowercase alphanumeric with hyphens" + ) + if fid in seen_ids: + raise ValueError(f"Duplicate fact id: {fid!r}") + seen_ids.add(fid) + + topic = fact["topic"] + if not isinstance(topic, str) or not topic: + raise ValueError(f"Fact '{fid}': 'topic' must be a non-empty string") + if len(topic) > _MAX_TOPIC_LEN: + raise ValueError(f"Fact '{fid}': topic exceeds {_MAX_TOPIC_LEN} characters") + if not _SLUG_RE.match(topic): + raise ValueError( + f"Fact '{fid}': topic must be lowercase alphanumeric with hyphens" + ) + + text = fact["text"] + if not isinstance(text, str) or len(text) < 10: + raise ValueError( + f"Fact '{fid}': 'text' must be a string of at least 10 characters" + ) + if len(text) > _MAX_TEXT_LEN: + raise ValueError( + f"Fact '{fid}': text exceeds {_MAX_TEXT_LEN} characters" + ) + _validate_corpus_text(text, fid, "text") + + source_label = fact["sourceLabel"] + if not isinstance(source_label, str) or not source_label: + raise ValueError(f"Fact '{fid}': 'sourceLabel' must be a non-empty string") + if len(source_label) > _MAX_SOURCE_LABEL_LEN: + raise ValueError( + f"Fact '{fid}': sourceLabel exceeds {_MAX_SOURCE_LABEL_LEN} characters" + ) + _validate_corpus_text(source_label, fid, "sourceLabel") + + _validate_source_url(fact["sourceUrl"], fid) + + if not isinstance(fact["visibility"], str) or fact["visibility"] != "public": + raise ValueError( + f"Fact '{fid}': visibility must be 'public'; got {fact['visibility']!r}" + ) + if not isinstance(fact["status"], str) or fact["status"] != "verified": + raise ValueError( + f"Fact '{fid}': status must be 'verified'; got {fact['status']!r}" + ) + + validated.append(fact) + + validated.sort(key=lambda f: f["id"]) + return validated + + +def _load_prompt_file(path: pathlib.Path) -> str: + try: + raw_bytes = path.read_bytes() + except OSError as exc: + raise ValueError(f"Cannot read prompt file {path.name}: {exc}") from exc + if len(raw_bytes) > _MAX_PROMPT_FILE_BYTES: + raise ValueError( + f"Prompt file {path.name} exceeds {_MAX_PROMPT_FILE_BYTES} bytes " + f"(got {len(raw_bytes)})" + ) + try: + content = raw_bytes.decode("utf-8") + except UnicodeDecodeError as exc: + raise ValueError(f"Prompt file {path.name} is not valid UTF-8") from exc + if _PROMPT_CONTROL_PATTERN.search(content): + raise ValueError(f"Prompt file {path.name} contains control characters") + for pattern in _CONTACT_PATTERNS + _SECRET_VALUE_PATTERNS: + if pattern.search(content): + raise ValueError( + f"Prompt file {path.name} contains contact data or a secret value" + ) + return content + + +def _format_facts(facts: list) -> str: + lines: list = [] + for fact in facts: + lines.append( + json.dumps( + { + "fact": fact["text"], + "id": fact["id"], + "sourceLabel": fact["sourceLabel"], + "sourceUrl": fact["sourceUrl"], + "topic": fact["topic"], + }, + ensure_ascii=False, + separators=(",", ":"), + sort_keys=True, + ) + ) + return "\n".join(lines) + + +# --------------------------------------------------------------------------- +# Public API +# --------------------------------------------------------------------------- + + +def compile_prompt( + profile: str, + assistant_dir: Optional[pathlib.Path] = None, +) -> dict: + """ + Compile a deterministic system prompt for *profile*. + + Parameters + ---------- + profile: + One of ``"public_visitor"``, ``"invited_friend"``, or ``"june_admin"``. + assistant_dir: + Path to the ``mrjunejune/assistant/`` directory. Defaults to the + sibling of the ``inference/`` package at runtime. + + Returns + ------- + dict with keys: + - ``"content"`` (str): compiled prompt text + - ``"version"`` (int): schema version (always 1) + - ``"hash"`` (str): lowercase hex SHA-256 of ``content`` encoded UTF-8 + + Raises + ------ + ValueError + On unknown profile, malformed files, validation failures, or size + limit exceeded. + """ + if profile not in _KNOWN_PROFILES: + raise ValueError( + f"Unknown profile {profile!r}; " + f"known profiles: {sorted(_KNOWN_PROFILES)}" + ) + + base = pathlib.Path(assistant_dir) if assistant_dir is not None else _default_assistant_dir() + + common = _load_prompt_file(base / "common.md") + profile_section = _load_prompt_file(base / f"{profile}.md") + facts = _load_and_validate_facts(base) + facts_block = _format_facts(facts) + + content = ( + common.rstrip() + + "\n\n" + + profile_section.rstrip() + + "\n\n" + + _FACTS_DELIMITER + + "\n\n" + + facts_block + ) + + encoded = content.encode("utf-8") + if len(encoded) > _MAX_COMPILED_BYTES: + raise ValueError( + f"Compiled prompt exceeds {_MAX_COMPILED_BYTES} bytes " + f"(got {len(encoded)})" + ) + + return { + "content": content, + "version": _EXPECTED_VERSION, + "hash": hashlib.sha256(encoded).hexdigest(), + }