"""Section-anchored comments, stored as sibling `.comments.md` files. Atlas is a pure renderer that owns nothing (evolution plan section 3.4): every view reads source-of-truth files, and every write lands in a canonical location. Comments follow the same rule -- one markdown file per subject, parsed on every read, never an event-maintained store. The format is deliberately hand-editable so a person (or a Claude session) can reply by editing the file; the app re-reads it on the next request. File layout (one file per subject, e.g. `comments/guide/01-lab.comments.md`): # Comments -- guide/01-lab ## [t-k3j9x2m1] section=substrate status=open - **adi** @ 2026-09-05T11:40:00-07:00 > Why can't I run docker myself? > (a second line of the same comment) - **rian** @ 2026-09-05T12:00:00-07:00 > Because the gateway is the one audited door. Every thread is a `## [id] section= status=` heading; every entry is a `- **author** @ ` line followed by ` > ` body lines. Anything that does not match those shapes is ignored on read and dropped on the next write, so a stray edit degrades to "that line vanished", never to a crash. Writes take a per-file lock and replace the file atomically. Files are created group-writable (0664, directories 2775) so the owner can edit them from the shell. """ from __future__ import annotations import fcntl import os import re import secrets import tempfile from dataclasses import asdict, dataclass, field from datetime import datetime, timezone from pathlib import Path SLUG_RE = re.compile(r"^[a-z0-9][a-z0-9-]{0,63}$") AUTHOR_RE = re.compile(r"^[a-z0-9][a-z0-9._-]{0,63}$") THREAD_ID_RE = re.compile(r"^t-[a-z0-9]{6,16}$") _THREAD_LINE = re.compile( r"^## \[(t-[a-z0-9]{6,16})\] section=([a-z0-9][a-z0-9-]{0,63}) status=(open|resolved)\s*$") _ENTRY_LINE = re.compile(r"^- \*\*([a-z0-9][a-z0-9._-]{0,63})\*\* @ (\S+)\s*$") _BODY_LINE = re.compile(r"^ >(?: (.*))?$") MAX_BODY_CHARS = 4000 MAX_THREADS_PER_SUBJECT = 500 MAX_ENTRIES_PER_THREAD = 200 # The section slug that means "the subject as a whole" (a chapter-level comment). WHOLE_SUBJECT = "chapter" class CommentsError(Exception): def __init__(self, code: str, summary: str, details: str = "") -> None: super().__init__(summary) self.code = code self.summary = summary self.details = details @dataclass class Entry: author: str at: str body: str @dataclass class Thread: id: str section: str status: str entries: list[Entry] = field(default_factory=list) def as_dict(self) -> dict: d = asdict(self) for i, e in enumerate(d["entries"]): e["index"] = i return d # ------------------------------------------------------------------ parsing def parse(text: str) -> list[Thread]: """Parse the markdown into threads. Tolerant: unknown lines are skipped.""" threads: list[Thread] = [] thread: Thread | None = None entry: Entry | None = None body_lines: list[str] = [] def close_entry() -> None: nonlocal entry, body_lines if entry is not None and thread is not None: entry.body = "\n".join(body_lines).rstrip() if entry.body: thread.entries.append(entry) entry = None body_lines = [] for raw in text.splitlines(): m = _THREAD_LINE.match(raw) if m: close_entry() thread = Thread(id=m.group(1), section=m.group(2), status=m.group(3)) threads.append(thread) continue if raw.startswith("## "): # A heading that is not a well-formed thread ends the current thread: # entries under it are orphans, not part of the previous thread. close_entry() thread = None continue if thread is None: continue m = _ENTRY_LINE.match(raw) if m: close_entry() entry = Entry(author=m.group(1), at=m.group(2), body="") continue m = _BODY_LINE.match(raw) if m and entry is not None: body_lines.append(m.group(1) or "") continue # Any other line ends the current entry (blank lines between entries are # fine; prose outside the shape is ignored). close_entry() close_entry() return [t for t in threads if t.entries] def render(subject: str, threads: list[Thread]) -> str: out = [f"# Comments -- {subject}", "", "", ""] for t in threads: out.append(f"## [{t.id}] section={t.section} status={t.status}") for e in t.entries: out.append(f"- **{e.author}** @ {e.at}") for line in (e.body or "").split("\n"): out.append(f" > {line}" if line else " >") out.append("") return "\n".join(out).rstrip("\n") + "\n" # ------------------------------------------------------------------ the store def _now() -> str: return datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds") def _new_id(existing: set[str]) -> str: while True: cand = "t-" + secrets.token_hex(4) if cand not in existing: return cand def _clean_body(body: str) -> str: body = (body or "").replace("\r\n", "\n").replace("\r", "\n").strip() if not body: raise CommentsError("EMPTY_BODY", "Write something first.", "A comment needs some text.") if len(body) > MAX_BODY_CHARS: raise CommentsError("BODY_TOO_LONG", "That comment is too long.", f"Keep it under {MAX_BODY_CHARS} characters.") return body def check_slug(value: str, what: str) -> str: if not SLUG_RE.match(value or ""): raise CommentsError("BAD_INPUT", f"That {what} name is not valid.", "Lowercase letters, digits and dashes only.") return value class CommentsStore: """All comment files for one root, e.g. `/comments`. Subjects are `/` (both slug-shaped), mapped to `//.comments.md`.""" def __init__(self, root: Path) -> None: self.root = Path(root) # -- paths def _path(self, area: str, slug: str) -> Path: check_slug(area, "area") check_slug(slug, "subject") return self.root / area / f"{slug}.comments.md" def _lock_path(self, area: str, slug: str) -> Path: return self.root / area / f".{slug}.lock" # -- reads def threads(self, area: str, slug: str) -> list[Thread]: path = self._path(area, slug) if not path.is_file(): return [] return parse(path.read_text(encoding="utf-8")) def counts(self, area: str) -> dict[str, dict[str, int]]: """{slug: {"open": n, "total": n}} for every subject in the area.""" check_slug(area, "area") folder = self.root / area out: dict[str, dict[str, int]] = {} if not folder.is_dir(): return out for path in sorted(folder.glob("*.comments.md")): slug = path.name[: -len(".comments.md")] if not SLUG_RE.match(slug): continue threads = parse(path.read_text(encoding="utf-8")) out[slug] = { "open": sum(1 for t in threads if t.status == "open"), "total": len(threads), } return out # -- writes (each one: lock, re-read, mutate, atomic replace) def _mutate(self, area: str, slug: str, fn) -> list[Thread]: path = self._path(area, slug) folder = path.parent folder.mkdir(parents=True, exist_ok=True) try: folder.chmod(0o2775) except OSError: pass lock_path = self._lock_path(area, slug) with open(lock_path, "a+") as lock: fcntl.flock(lock.fileno(), fcntl.LOCK_EX) try: threads = parse(path.read_text(encoding="utf-8")) if path.is_file() else [] threads = fn(threads) text = render(f"{area}/{slug}", threads) fd, tmp = tempfile.mkstemp(prefix=f".{slug}.", suffix=".tmp", dir=folder) try: with os.fdopen(fd, "w", encoding="utf-8") as fh: fh.write(text) os.chmod(tmp, 0o664) os.replace(tmp, path) except BaseException: try: os.unlink(tmp) except OSError: pass raise finally: fcntl.flock(lock.fileno(), fcntl.LOCK_UN) try: lock_path.chmod(0o664) except OSError: pass return threads def add_thread(self, area: str, slug: str, section: str, author: str, body: str) -> Thread: check_slug(section, "section") if not AUTHOR_RE.match(author or ""): raise CommentsError("BAD_INPUT", "Unknown author.") body = _clean_body(body) created: dict[str, Thread] = {} def fn(threads: list[Thread]) -> list[Thread]: if len(threads) >= MAX_THREADS_PER_SUBJECT: raise CommentsError("TOO_MANY_THREADS", "This page has too many threads already.", "Resolve or delete some before starting more.") t = Thread(id=_new_id({x.id for x in threads}), section=section, status="open", entries=[Entry(author, _now(), body)]) threads.append(t) created["t"] = t return threads self._mutate(area, slug, fn) return created["t"] def reply(self, area: str, slug: str, thread_id: str, author: str, body: str) -> Thread: if not THREAD_ID_RE.match(thread_id or ""): raise CommentsError("NO_SUCH_THREAD", "That thread does not exist.") if not AUTHOR_RE.match(author or ""): raise CommentsError("BAD_INPUT", "Unknown author.") body = _clean_body(body) found: dict[str, Thread] = {} def fn(threads: list[Thread]) -> list[Thread]: t = _find(threads, thread_id) if len(t.entries) >= MAX_ENTRIES_PER_THREAD: raise CommentsError("THREAD_FULL", "This thread is full.", "Start a new thread instead.") t.entries.append(Entry(author, _now(), body)) found["t"] = t return threads self._mutate(area, slug, fn) return found["t"] def set_status(self, area: str, slug: str, thread_id: str, status: str) -> Thread: if status not in ("open", "resolved"): raise CommentsError("BAD_INPUT", "Status must be open or resolved.") found: dict[str, Thread] = {} def fn(threads: list[Thread]) -> list[Thread]: t = _find(threads, thread_id) t.status = status found["t"] = t return threads self._mutate(area, slug, fn) return found["t"] def delete_entry(self, area: str, slug: str, thread_id: str, index: int) -> Thread | None: """Remove one entry. Removing the last entry removes the thread (returns None).""" found: dict[str, Thread | None] = {} def fn(threads: list[Thread]) -> list[Thread]: t = _find(threads, thread_id) if index < 0 or index >= len(t.entries): raise CommentsError("NO_SUCH_ENTRY", "That comment does not exist.") del t.entries[index] if not t.entries: threads.remove(t) found["t"] = None else: found["t"] = t return threads self._mutate(area, slug, fn) return found["t"] def _find(threads: list[Thread], thread_id: str) -> Thread: for t in threads: if t.id == thread_id: return t raise CommentsError("NO_SUCH_THREAD", "That thread does not exist.", "It may have been deleted. Reload the page.")