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(),
+    }