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