Mercurial

comparison mrjunejune/inference/public_knowledge.py @ 265:056790c4fb0d

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
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
comparison
equal deleted inserted replaced
264:04fee26ecce0 265:056790c4fb0d
1 """
2 Stdlib-only deterministic loader and compiler for the assistant knowledge base.
3
4 Loads prompt files from mrjunejune/assistant/ and compiles them with verified
5 public facts into a hashed, deterministic system prompt. Fails closed on any
6 malformed input.
7
8 Public API:
9 compile_prompt(profile, assistant_dir=None) -> dict
10 Returns {"content": str, "version": int, "hash": str}.
11 Raises ValueError for unknown profile or any validation failure.
12 """
13
14 import hashlib
15 import json
16 import pathlib
17 import re
18 from typing import Optional
19 from urllib.parse import urlsplit
20
21 # ---------------------------------------------------------------------------
22 # Constants
23 # ---------------------------------------------------------------------------
24
25 _KNOWN_PROFILES: frozenset = frozenset({"public_visitor", "invited_friend", "june_admin"})
26
27 _REQUIRED_TOP_KEYS: frozenset = frozenset({"version", "facts"})
28 _REQUIRED_FACT_KEYS: frozenset = frozenset(
29 {"id", "topic", "text", "sourceLabel", "sourceUrl", "visibility", "status"}
30 )
31
32 _EXPECTED_VERSION: int = 1
33 _MAX_PROMPT_FILE_BYTES: int = 8192 # 8 KB per prompt file
34 _MAX_COMPILED_BYTES: int = 16384 # 16 KB total compiled ceiling
35
36 _MAX_ID_LEN: int = 64
37 _MAX_TOPIC_LEN: int = 64
38 _MAX_TEXT_LEN: int = 500
39 _MAX_SOURCE_LABEL_LEN: int = 128
40
41 _SLUG_RE = re.compile(r"^[a-z0-9]([a-z0-9-]*[a-z0-9])?$")
42
43 # Contact patterns: email and North-American-style phone numbers
44 _CONTACT_PATTERNS: list = [
45 re.compile(r"[a-zA-Z0-9._%+\-]+@[a-zA-Z0-9.\-]+\.[a-zA-Z]{2,}"),
46 re.compile(r"(\+?1[-.\s]?)?\(?\d{3}\)?[-.\s]\d{3}[-.\s]\d{4}"),
47 ]
48 _SECRET_VALUE_PATTERNS: list = [
49 re.compile(
50 r"\b(password|passwd|secret|token|api[_\-]?key|private[_\-]?key)"
51 r"\s*[:=]\s*\S+",
52 re.IGNORECASE,
53 ),
54 re.compile(r"-----BEGIN [A-Z ]*PRIVATE KEY-----"),
55 ]
56 _CONTROL_PATTERN = re.compile(r"[\x00-\x1f\x7f]")
57 _PROMPT_CONTROL_PATTERN = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]")
58 _RESERVED_CORPUS_PATTERNS: tuple[str, ...] = (
59 "--- knowledge corpus ---",
60 "system:",
61 "assistant:",
62 "user:",
63 "<system",
64 "ignore previous instructions",
65 "ignore prior instructions",
66 "follow these instructions",
67 )
68
69 # Credential / secret keyword patterns
70 _SECRET_PATTERNS: list = [
71 re.compile(
72 r"\b(password|passwd|secret|token|api[_\-]?key|private[_\-]?key)\b",
73 re.IGNORECASE,
74 ),
75 ]
76
77 _FACTS_DELIMITER: str = "--- KNOWLEDGE CORPUS ---"
78
79
80 # ---------------------------------------------------------------------------
81 # Internal helpers
82 # ---------------------------------------------------------------------------
83
84
85 def _default_assistant_dir() -> pathlib.Path:
86 return pathlib.Path(__file__).parent.parent / "assistant"
87
88
89 def _validate_source_url(url: str, fact_id: str) -> None:
90 if not isinstance(url, str) or not url:
91 raise ValueError(f"Fact '{fact_id}': sourceUrl is empty")
92 if _CONTROL_PATTERN.search(url) or "\\" in url:
93 raise ValueError(f"Fact '{fact_id}': sourceUrl contains unsafe characters")
94 if url.startswith("/"):
95 if url.startswith("//") or any(
96 segment in (".", "..") for segment in url.split("/")
97 ):
98 raise ValueError(
99 f"Fact '{fact_id}': sourceUrl must be a safe local path"
100 )
101 return
102 parsed = urlsplit(url)
103 if (
104 parsed.scheme != "https"
105 or not parsed.hostname
106 or parsed.username is not None
107 or parsed.password is not None
108 ):
109 raise ValueError(
110 f"Fact '{fact_id}': sourceUrl must be HTTPS or a safe local path"
111 )
112 try:
113 parsed.port
114 except ValueError as exc:
115 raise ValueError(f"Fact '{fact_id}': sourceUrl has an invalid port") from exc
116
117
118 def _check_no_contact_or_secret(text: str, fact_id: str, field: str) -> None:
119 for pat in _CONTACT_PATTERNS:
120 if pat.search(text):
121 raise ValueError(
122 f"Fact '{fact_id}': {field} contains contact data (email or phone number)"
123 )
124 for pat in _SECRET_PATTERNS:
125 if pat.search(text):
126 raise ValueError(
127 f"Fact '{fact_id}': {field} contains a secret or credential pattern"
128 )
129
130 def _validate_corpus_text(text: str, fact_id: str, field: str) -> None:
131 if _CONTROL_PATTERN.search(text):
132 raise ValueError(f"Fact '{fact_id}': {field} contains control characters")
133 lowered = text.lower()
134 for reserved in _RESERVED_CORPUS_PATTERNS:
135 if reserved in lowered:
136 raise ValueError(
137 f"Fact '{fact_id}': {field} contains reserved prompt instructions"
138 )
139 _check_no_contact_or_secret(text, fact_id, field)
140
141
142 def _load_and_validate_facts(assistant_dir: pathlib.Path) -> list:
143 facts_path = assistant_dir / "knowledge" / "public_facts.json"
144 try:
145 raw = facts_path.read_text(encoding="utf-8")
146 except OSError as exc:
147 raise ValueError(f"Cannot read facts file: {exc}") from exc
148
149 try:
150 data = json.loads(raw)
151 except json.JSONDecodeError as exc:
152 raise ValueError(f"Facts file is not valid JSON: {exc}") from exc
153
154 if not isinstance(data, dict):
155 raise ValueError("Facts file top level must be a JSON object")
156
157 extra_keys = set(data.keys()) - _REQUIRED_TOP_KEYS
158 missing_keys = _REQUIRED_TOP_KEYS - set(data.keys())
159 if extra_keys or missing_keys:
160 raise ValueError(
161 f"Facts file top-level keys must be exactly {sorted(_REQUIRED_TOP_KEYS)}; "
162 f"extra={sorted(extra_keys)}, missing={sorted(missing_keys)}"
163 )
164
165 if (
166 not isinstance(data["version"], int)
167 or isinstance(data["version"], bool)
168 or data["version"] != _EXPECTED_VERSION
169 ):
170 raise ValueError(
171 f"Facts file version must be {_EXPECTED_VERSION}; got {data['version']!r}"
172 )
173
174 raw_facts = data["facts"]
175 if not isinstance(raw_facts, list) or len(raw_facts) == 0:
176 raise ValueError("Facts file 'facts' must be a non-empty list")
177
178 seen_ids: set = set()
179 validated: list = []
180
181 for i, fact in enumerate(raw_facts):
182 if not isinstance(fact, dict):
183 raise ValueError(f"Fact at index {i} must be a JSON object")
184
185 extra = set(fact.keys()) - _REQUIRED_FACT_KEYS
186 missing = _REQUIRED_FACT_KEYS - set(fact.keys())
187 if extra or missing:
188 raise ValueError(
189 f"Fact at index {i}: keys must be exactly "
190 f"{sorted(_REQUIRED_FACT_KEYS)}; "
191 f"extra={sorted(extra)}, missing={sorted(missing)}"
192 )
193
194 fid = fact["id"]
195 if not isinstance(fid, str) or not fid:
196 raise ValueError(f"Fact at index {i}: 'id' must be a non-empty string")
197 if len(fid) > _MAX_ID_LEN:
198 raise ValueError(f"Fact '{fid}': id exceeds {_MAX_ID_LEN} characters")
199 if not _SLUG_RE.match(fid):
200 raise ValueError(
201 f"Fact '{fid}': id must be lowercase alphanumeric with hyphens"
202 )
203 if fid in seen_ids:
204 raise ValueError(f"Duplicate fact id: {fid!r}")
205 seen_ids.add(fid)
206
207 topic = fact["topic"]
208 if not isinstance(topic, str) or not topic:
209 raise ValueError(f"Fact '{fid}': 'topic' must be a non-empty string")
210 if len(topic) > _MAX_TOPIC_LEN:
211 raise ValueError(f"Fact '{fid}': topic exceeds {_MAX_TOPIC_LEN} characters")
212 if not _SLUG_RE.match(topic):
213 raise ValueError(
214 f"Fact '{fid}': topic must be lowercase alphanumeric with hyphens"
215 )
216
217 text = fact["text"]
218 if not isinstance(text, str) or len(text) < 10:
219 raise ValueError(
220 f"Fact '{fid}': 'text' must be a string of at least 10 characters"
221 )
222 if len(text) > _MAX_TEXT_LEN:
223 raise ValueError(
224 f"Fact '{fid}': text exceeds {_MAX_TEXT_LEN} characters"
225 )
226 _validate_corpus_text(text, fid, "text")
227
228 source_label = fact["sourceLabel"]
229 if not isinstance(source_label, str) or not source_label:
230 raise ValueError(f"Fact '{fid}': 'sourceLabel' must be a non-empty string")
231 if len(source_label) > _MAX_SOURCE_LABEL_LEN:
232 raise ValueError(
233 f"Fact '{fid}': sourceLabel exceeds {_MAX_SOURCE_LABEL_LEN} characters"
234 )
235 _validate_corpus_text(source_label, fid, "sourceLabel")
236
237 _validate_source_url(fact["sourceUrl"], fid)
238
239 if not isinstance(fact["visibility"], str) or fact["visibility"] != "public":
240 raise ValueError(
241 f"Fact '{fid}': visibility must be 'public'; got {fact['visibility']!r}"
242 )
243 if not isinstance(fact["status"], str) or fact["status"] != "verified":
244 raise ValueError(
245 f"Fact '{fid}': status must be 'verified'; got {fact['status']!r}"
246 )
247
248 validated.append(fact)
249
250 validated.sort(key=lambda f: f["id"])
251 return validated
252
253
254 def _load_prompt_file(path: pathlib.Path) -> str:
255 try:
256 raw_bytes = path.read_bytes()
257 except OSError as exc:
258 raise ValueError(f"Cannot read prompt file {path.name}: {exc}") from exc
259 if len(raw_bytes) > _MAX_PROMPT_FILE_BYTES:
260 raise ValueError(
261 f"Prompt file {path.name} exceeds {_MAX_PROMPT_FILE_BYTES} bytes "
262 f"(got {len(raw_bytes)})"
263 )
264 try:
265 content = raw_bytes.decode("utf-8")
266 except UnicodeDecodeError as exc:
267 raise ValueError(f"Prompt file {path.name} is not valid UTF-8") from exc
268 if _PROMPT_CONTROL_PATTERN.search(content):
269 raise ValueError(f"Prompt file {path.name} contains control characters")
270 for pattern in _CONTACT_PATTERNS + _SECRET_VALUE_PATTERNS:
271 if pattern.search(content):
272 raise ValueError(
273 f"Prompt file {path.name} contains contact data or a secret value"
274 )
275 return content
276
277
278 def _format_facts(facts: list) -> str:
279 lines: list = []
280 for fact in facts:
281 lines.append(
282 json.dumps(
283 {
284 "fact": fact["text"],
285 "id": fact["id"],
286 "sourceLabel": fact["sourceLabel"],
287 "sourceUrl": fact["sourceUrl"],
288 "topic": fact["topic"],
289 },
290 ensure_ascii=False,
291 separators=(",", ":"),
292 sort_keys=True,
293 )
294 )
295 return "\n".join(lines)
296
297
298 # ---------------------------------------------------------------------------
299 # Public API
300 # ---------------------------------------------------------------------------
301
302
303 def compile_prompt(
304 profile: str,
305 assistant_dir: Optional[pathlib.Path] = None,
306 ) -> dict:
307 """
308 Compile a deterministic system prompt for *profile*.
309
310 Parameters
311 ----------
312 profile:
313 One of ``"public_visitor"``, ``"invited_friend"``, or ``"june_admin"``.
314 assistant_dir:
315 Path to the ``mrjunejune/assistant/`` directory. Defaults to the
316 sibling of the ``inference/`` package at runtime.
317
318 Returns
319 -------
320 dict with keys:
321 - ``"content"`` (str): compiled prompt text
322 - ``"version"`` (int): schema version (always 1)
323 - ``"hash"`` (str): lowercase hex SHA-256 of ``content`` encoded UTF-8
324
325 Raises
326 ------
327 ValueError
328 On unknown profile, malformed files, validation failures, or size
329 limit exceeded.
330 """
331 if profile not in _KNOWN_PROFILES:
332 raise ValueError(
333 f"Unknown profile {profile!r}; "
334 f"known profiles: {sorted(_KNOWN_PROFILES)}"
335 )
336
337 base = pathlib.Path(assistant_dir) if assistant_dir is not None else _default_assistant_dir()
338
339 common = _load_prompt_file(base / "common.md")
340 profile_section = _load_prompt_file(base / f"{profile}.md")
341 facts = _load_and_validate_facts(base)
342 facts_block = _format_facts(facts)
343
344 content = (
345 common.rstrip()
346 + "\n\n"
347 + profile_section.rstrip()
348 + "\n\n"
349 + _FACTS_DELIMITER
350 + "\n\n"
351 + facts_block
352 )
353
354 encoded = content.encode("utf-8")
355 if len(encoded) > _MAX_COMPILED_BYTES:
356 raise ValueError(
357 f"Compiled prompt exceeds {_MAX_COMPILED_BYTES} bytes "
358 f"(got {len(encoded)})"
359 )
360
361 return {
362 "content": content,
363 "version": _EXPECTED_VERSION,
364 "hash": hashlib.sha256(encoded).hexdigest(),
365 }