gate.py

gate.py

1. #!/usr/bin/env python3
2. """
3. gate.py: the deterministic path gate in front of the better-memory distiller.
5. A plain script, not a model. Each run wipes STAGING_DIR/staging, walks the vaults
6. with a fail-closed denylist (checked after resolving traversal and symlinks), skips
7. vault scaffolding, hashes each safe note's post-gate content and stages only the
8. notes whose hash differs from the manifest. A manifest of content hashes, rather
9. than a record of seen notes, means an edited note always re-stages.
11. Env: VAULT_ROOTS (colon-separated vault paths), STAGING_DIR (staging root, default
12. ~/cowork-staging), EXTRA_DENY (comma-separated extra denied prefixes, optional).
13. """
15. from __future__ import annotations
17. import os
18. import re
19. import json
20. import shutil
21. import hashlib
22. from pathlib import Path
24. # --- Configuration
26. VAULT_ROOTS = [
27. Path(p).resolve()
28. for p in os.environ.get("VAULT_ROOTS", "").split(":")
29. if p.strip()
30. ]
31. STAGING_ROOT = Path(os.environ.get("STAGING_DIR", str(Path.home() / "cowork-staging"))).resolve()
32. STAGING_DIR = STAGING_ROOT / "staging"
33. MANIFEST_PATH = STAGING_ROOT / ".manifest.json"
35. # Fail-closed denylist. Prefixes are anchored to a vault root; components match a
36. # folder name at any depth. The destination is a private database only the owner
37. # reads, so the list covers genuinely private content; research folders may stage.
38. _BASE_DENY_PREFIXES = [
39. "_private", # the wall
40. "efforts/finance", # financial records
41. "efforts/journal", # the personal journal, whole tree
42. ]
44. DENY_COMPONENTS = [
45. ".obsidian", # plugin configs, which can hold API keys: root and nested vaults
46. ".git", # repo internals, at any depth (nested vaults can be repos)
47. ".trash", # Obsidian deleted-notes, any depth
48. "_excalidraw", # excalidraw scratch dirs, any depth
49. ".basic-memory", # Basic Memory's SQLite index + internal state
50. ".venv", # virtualenv noise
51. ]
53. _EXTRA_DENY = [
54. p.strip().lower().strip("/")
55. for p in os.environ.get("EXTRA_DENY", "").split(",")
56. if p.strip()
57. ]
59. DENY_PREFIXES = _BASE_DENY_PREFIXES + _EXTRA_DENY
61. # Only these are ever staged.
62. STAGE_SUFFIXES = {".md", ".markdown", ".txt"}
64. # --- Scaffolding: vault machinery that is not memory about the owner
65. #
66. # Operational instructions, changelogs, model guides, run logs and templates are
67. # not facts about the owner, and distilling them pollutes recall. They are skipped,
68. # except that each nested vault keeps one explainer (see EXPLAINER_PRIORITY) so it
69. # still contributes a "what is this vault for" note.
71. # Exact filenames (case-insensitive, basename match) that are always scaffolding.
72. SCAFFOLD_FILENAMES = {
73. "heartbeat.md", "setup.md", "model_guide.md", "changelog.md",
74. "skills_reference.md", "subagents.md", "memory.md",
75. "knowledge_index.md", "inventory_index.md", "topics_index.md",
76. "sources_index.md",
77. # Project plumbing. decisions.md is kept for _meta only; see SCAFFOLD_KEEP_PATHS.
78. "decisions.md", "skill.md", "agent.md", "automation.md",
79. "backlog.md", "investigation-plan.md",
80. "test.md", "scratch.md", "url-inventory.md",
81. }
84. # Also scaffolding by pattern: per-folder instruction files (claude.md, claude-*.md),
85. # index files (NN-index.md, *_index.md, *-index.md), done and deprecated notes, probe
86. # files and bare-date daily notes. A vault's chosen explainer is kept regardless.
87. def _is_scaffold_name(basename: str) -> bool:
88. b = basename.lower()
89. if b in SCAFFOLD_FILENAMES:
90. return True
91. if b == "claude.md" or b.startswith("claude-"):
92. return True
93. if re.match(r"^\d{2}-index\.md$", b) or b.endswith("_index.md") or b.endswith("-index.md"):
94. return True
95. if "deprecated" in b:
96. return True
97. if b.endswith("-block-done.md") or b.endswith("-done.md"):
98. return True
99. if re.match(r"^probe\d*\.txt$", b):
100. return True
101. # bare date-named daily scratch at any level (e.g. 2026-06-17.md), but not
102. # dated content notes that have a descriptive suffix (those keep their words)
103. if re.match(r"^\d{4}-\d{2}-\d{2}\.md$", b):
104. return True
105. return False
108. # Directory segments whose notes are templates, run logs, session transcripts or
109. # scratch. A note whose vault-relative path contains one is skipped wholesale.
110. SCAFFOLD_PATH_FRAGMENTS = (
111. "/_templates/",
112. "/08 - templates/",
113. "/02-run-logs/",
114. "/01-sessions/",
115. "/initiation-prompts/",
116. "/_scratch/",
117. "/_scratch-junk/",
118. )
120. # A scaffolding-named file is kept when its lowercased vault-relative path ends with
121. # one of these: decisions.md is plumbing everywhere except the _meta identity layer,
122. # where it records the owner's system decisions.
123. SCAFFOLD_KEEP_PATHS = (
124. "_meta/decisions.md",
125. )
127. # When picking the single explainer note for a nested vault, prefer in this order.
128. EXPLAINER_PRIORITY = ("readme.md", "00-index.md", "claude.md")
130. # A "nested vault" for the one-explainer rule = an immediate child directory of
131. # one of these parents (relative to a vault root). Each such folder keeps exactly
132. # one explainer; the root vault itself also keeps one.
133. EXPLAINER_PARENT_DIRS = ("efforts", "atlas")
135. # --- Section gate: stage only a note's framing when it mixes in sensitive detail
136. #
137. # Off by design: both lists are empty, so every note stages whole. Private material
138. # is already path-denied before it reaches this point, and the patterns only ever
139. # truncated benign notes. Adding a pattern or a link name re-enables it.
142. def _alternation_or_never(patterns: list[str], before: str = "",
143. after: str = "") -> re.Pattern[str]:
144. """Compile `before + "|".join(patterns) + after`, case-insensitive."""
145. # Guard: an empty alternation matches everywhere (for the link pattern, every
146. # "[["), framing everything. When the list is empty, use a pattern that can
147. # never match.
148. if not patterns:
149. return re.compile(r"(?!x)x")
150. return re.compile(before + "|".join(patterns) + after, re.IGNORECASE)
153. SENSITIVE_PATTERNS: list[str] = []
154. _SENSITIVE_RE = _alternation_or_never(SENSITIVE_PATTERNS)
156. # Wikilink targets whose presence marks a note as mixed.
157. EXCLUDED_LINK_NAMES: list[str] = []
158. _EXCLUDED_LINK_RE = _alternation_or_never(
159. [re.escape(n) for n in EXCLUDED_LINK_NAMES], before=r"\[\[\s*(", after=r")"
160. )
162. # First structured heading: framing is everything before it.
163. _FIRST_SECTION_RE = re.compile(r"^##\s+(observations|relations)\b", re.IGNORECASE | re.MULTILINE)
166. # --- The path gate
168. class PathDenied(Exception):
169. pass
172. def _is_denied(rel: str) -> bool:
173. """True if the lowercased vault-relative path is, or sits under, a denied prefix."""
174. return any(rel == deny or rel.startswith(deny + "/") for deny in DENY_PREFIXES)
177. def _resolve_safe(root: Path, rel: str) -> Path:
178. """Resolve a vault-relative path and fail closed on traversal, symlink or deny."""
179. rel_norm = rel.replace("\\", "/").lstrip("/")
180. candidate = (root / rel_norm).resolve()
181. try:
182. candidate.relative_to(root)
183. except ValueError:
184. raise PathDenied(f"escapes root: {rel!r}")
185. rel_from_root = candidate.relative_to(root).as_posix().lower()
186. if _is_denied(rel_from_root):
187. raise PathDenied(f"denied prefix: {rel!r}")
188. if set(rel_from_root.split("/")) & set(DENY_COMPONENTS):
189. raise PathDenied(f"denied component: {rel!r}")
190. return candidate
193. def _iter_safe_files(root: Path):
194. """Walk one vault root, pruning denied dirs, yielding gate-passing notes."""
195. deny_components = set(DENY_COMPONENTS)
196. for dirpath, dirnames, filenames in os.walk(root):
197. here = Path(dirpath)
198. rel_dir = here.relative_to(root).as_posix().lower()
199. rel_dir = "" if rel_dir == "." else rel_dir
200. kept = []
201. for d in dirnames:
202. if d.lower() in deny_components:
203. continue
204. if _is_denied(f"{rel_dir}/{d}".lstrip("/").lower()):
205. continue
206. kept.append(d)
207. dirnames[:] = kept
208. for fn in filenames:
209. fp = here / fn
210. if fp.suffix.lower() not in STAGE_SUFFIXES:
211. continue
212. try:
213. _resolve_safe(root, fp.relative_to(root).as_posix())
214. except (PathDenied, ValueError):
215. continue
216. yield fp
219. def _self_check(root: Path) -> None:
220. """Refuse to run unless every known-private probe is denied."""
221. for probe in ("_private", "_Private", "Efforts/finance",
222. "Efforts/Journal/anything.md", ".obsidian",
223. "Efforts/some-vault/.obsidian/x.md", ".basic-memory"):
224. try:
225. _resolve_safe(root, probe)
226. except PathDenied:
227. pass
228. else:
229. raise SystemExit(f"DENYLIST FAILED to block {probe!r}: refusing to run.")
232. # --- Section gate: whole note or framing only
234. def _framing_only(text: str) -> str:
235. """Return the H1 and framing prose before the first ## Observations/Relations,
236. capped at 1500 characters when there is no such heading, with excluded-note
237. wikilinks redacted."""
238. m = _FIRST_SECTION_RE.search(text)
239. framing = text[: m.start()] if m else text[:1500]
240. framing = _EXCLUDED_LINK_RE.sub("[[redacted", framing)
241. return framing.rstrip() + "\n"
244. def _gate_content(text: str) -> str:
245. """Return the note whole, or its framing only if it matches a sensitive pattern
246. or links to an excluded note."""
247. mixed = bool(_SENSITIVE_RE.search(text)) or bool(_EXCLUDED_LINK_RE.search(text))
248. return _framing_only(text) if mixed else text
251. # --- Scaffolding skip and one explainer per vault
253. def _vault_key(vault_name: str, rel_posix: str) -> str:
254. """The explainer bucket for a note: `<vault>/Efforts/<X>` or `<vault>/Atlas/<X>`
255. for notes inside a nested vault, otherwise the root vault's name."""
256. parts = rel_posix.split("/")
257. low = [p.lower() for p in parts]
258. if len(parts) >= 2 and low[0] in EXPLAINER_PARENT_DIRS:
259. return f"{vault_name}/{parts[0]}/{parts[1]}"
260. return vault_name # root-level vault bucket
263. def _has_path_fragment_skip(rel_posix: str) -> bool:
264. low = "/" + rel_posix.lower() + "/"
265. return any(frag in low for frag in SCAFFOLD_PATH_FRAGMENTS)
268. def _choose_explainers(safe_files):
269. """Given (vault_name, fp, rel_posix) tuples, pick one explainer per vault bucket
270. in EXPLAINER_PRIORITY order. Returns the manifest keys to keep even though
271. they would otherwise be skipped as scaffolding."""
272. by_vault: dict[str, dict[str, str]] = {}
273. for vault_name, fp, rel in safe_files:
274. base = fp.name.lower()
275. if base in EXPLAINER_PRIORITY:
276. vk = _vault_key(vault_name, rel)
277. by_vault.setdefault(vk, {})[base] = f"{vault_name}/{rel}"
278. keep = set()
279. for vk, candidates in by_vault.items():
280. for pref in EXPLAINER_PRIORITY:
281. if pref in candidates:
282. keep.add(candidates[pref])
283. break
284. return keep
287. # --- Manifest
289. def _load_manifest() -> dict:
290. if MANIFEST_PATH.exists():
291. try:
292. return json.loads(MANIFEST_PATH.read_text(encoding="utf-8"))
293. except (OSError, json.JSONDecodeError):
294. return {}
295. return {}
298. def _save_manifest(m: dict) -> None:
299. MANIFEST_PATH.write_text(json.dumps(m, indent=2, sort_keys=True), encoding="utf-8")
302. def _hash(s: str) -> str:
303. return hashlib.sha256(s.encode("utf-8")).hexdigest()
306. # --- Main
308. def main() -> int:
309. if not VAULT_ROOTS:
310. raise SystemExit("Set VAULT_ROOTS (colon-separated absolute vault paths).")
311. for r in VAULT_ROOTS:
312. if not r.is_dir():
313. raise SystemExit(f"Not a directory: {r}")
315. _self_check(VAULT_ROOTS[0])
317. # 1. Wipe staging, so the distiller never reads a previous run's leftovers.
318. if STAGING_DIR.exists():
319. shutil.rmtree(STAGING_DIR)
320. STAGING_DIR.mkdir(parents=True, exist_ok=True)
322. manifest = _load_manifest()
323. new_manifest: dict = {}
324. staged = skipped = framed = scaffold_skipped = 0
326. # Pre-pass: collect all safe files, then resolve the one explainer per vault.
327. safe_files = []
328. for root in VAULT_ROOTS:
329. vault_name = root.name
330. for fp in _iter_safe_files(root):
331. rel = fp.relative_to(root).as_posix()
332. safe_files.append((vault_name, fp, rel))
333. explainer_keep = _choose_explainers(safe_files)
335. # 2-3. Walk, apply scaffolding skip, gate, hash, compare, stage new/changed.
336. for vault_name, fp, rel in safe_files:
337. key = f"{vault_name}/{rel}"
339. # Path-fragment skips (templates, run logs) are absolute: never kept, even
340. # if named like an explainer. Other scaffolding is kept only as the vault's
341. # chosen explainer or a SCAFFOLD_KEEP_PATHS exception.
342. if _has_path_fragment_skip(rel):
343. scaffold_skipped += 1
344. continue
345. if _is_scaffold_name(fp.name) and key not in explainer_keep:
346. rel_low = rel.lower()
347. if not any(rel_low.endswith(p) for p in SCAFFOLD_KEEP_PATHS):
348. scaffold_skipped += 1
349. continue
351. try:
352. raw = fp.read_text(encoding="utf-8", errors="ignore")
353. except OSError:
354. continue
356. content = _gate_content(raw)
357. if not content.strip():
358. continue
359. if content != raw:
360. framed += 1
362. # Hash what is staged, so a change the gate strips never re-stages a note.
363. h = _hash(content)
364. new_manifest[key] = h
366. if manifest.get(key) == h:
367. skipped += 1
368. continue
370. safe_name = key.replace("/", "__")
371. (STAGING_DIR / safe_name).write_text(content, encoding="utf-8")
372. staged += 1
374. # 4. The manifest holds only this run's notes, so deleted notes drop out of it.
375. _save_manifest(new_manifest)
377. print(f"gate: staged={staged} skipped={skipped} framing-only={framed} "
378. f"scaffold-skipped={scaffold_skipped} total-staged={len(new_manifest)} "
379. f"-> {STAGING_DIR}")
380. return 0
383. if __name__ == "__main__":
384. raise SystemExit(main())