"""Discussions on the Interaction Standard: threads, comments, mentions, the inbox. Sources of truth: this module, `app/models/discussion.py`, `app/routers/discussion.py`, `app/routers/notifications.py`, `app/cli_refresh.py` (reads `CLIENT_WRITTEN_TABLES`), `docs/CLIENT-SURFACES.md`, `docs/ACCOUNTS.md`, `tests/test_discussion_*.py`. Design: the accounts plan §8; the extraction source is the server's Interaction Standard reference implementation, ported verbatim where it is standard. What is DFP's own is five hooks: the deep link (`deep_link`), the subject resolver (`subject_of_legacy`, `legacy_of`, `permission_of`), the recipient policy (`_recipients`), the mention directory (`mention_directory`) and the base URL (`settings.public_base_url`). Rules carried over, each earned elsewhere: * Mentions are parsed SERVER-side from the stored body; the client payload never decides who is notified. An `@` of someone who cannot see the subject is dropped silently, so a plan thread never notifies a client. * Never notify the author of their own action; dedupe every emit, so a retry or a double-click cannot produce two bells. * Resolved propagates: resolving a thread flags the notifications it raised (they leave the unread count); reopening un-flags them. * Never hard-delete a conversation: a comment is soft-deleted and drops out of reads. * A decision is an EVENT kind (something was decided; it closes a loop and reads as news), never a thread state: state stays on `owner_item_states` and `discussion_items`. * Every permission lookup happens BEFORE a write path's first write (`people_for`): the kit answers `can()` from a session of its own, and the test suite's single shared connection reads that session's close as a rollback of the request's pending rows. The order costs nothing in production and keeps the suite honest. `CLIENT_WRITTEN_TABLES` is the one list of tables whose rows are written by people using the site; the staging refresh reads it, never a copy, so a table added here is preserved from the day it exists. Order matters: a table is listed after every table it references. """ from __future__ import annotations import re from dataclasses import dataclass from datetime import UTC, datetime from sqlalchemy import select, update from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session from app.config import settings from app.models import Account from app.models.discussion import DiscussionComment, Notification, Thread from app.services import accounts from app.services.identity import Actor #: Written by people on the site, in dependency order; the staging refresh preserves these. CLIENT_WRITTEN_TABLES: tuple[str, ...] = ( "discussion_items", "threads", "discussion_comments", "feature_priorities", "quote_selections", "quote_requests", "client_todos", "client_uploads", "owner_item_states", "notifications", "account_preferences", ) #: The subject vocabulary; keys are frozen and never renamed (plan §8). SUBJECT_TYPES = frozenset({"decision", "feature", "page", "structure", "quote", "todo", "item"}) #: The page each subject lives on, for the deep link: /discuss's tabs (T8), the running list. PAGE_OF = { "decision": "/discuss", "feature": "/discuss", "page": "/discuss", "structure": "/discuss?tab=structure", "quote": "/discuss?tab=quote", "todo": "/discuss?tab=todo", "item": "/issues", } #: The standard's vocabulary. `mention` and `turn` are addressed to you; the rest is news. KINDS = frozenset({"mention", "reply", "turn", "status", "resolved", "decision"}) NEEDS_YOU = frozenset({"mention", "turn"}) APP_NAME = "dfp" MENTION = re.compile(r"(? datetime: return datetime.now(UTC) # --- the subject resolver -------------------------------------------------------------------- def subject_of_legacy(item_id: int | None, feature_key: str | None) -> tuple[str, str]: """The pre-thread key a comment was written under, as a subject. A decision card's comments hung off `item_id`; a bare board key is a feature; the literal `general` is the discuss page itself; everything else already carries its type before the colon.""" if item_id is not None: return "decision", str(item_id) key = (feature_key or "").strip() if key == "general": return "page", "discuss" head, sep, tail = key.partition(":") if sep and head in SUBJECT_TYPES and tail: return head, tail return "feature", key def legacy_of(subject_type: str, subject_id: str) -> tuple[int | None, str | None]: """The legacy columns for a subject, kept for one release so the old reads still work.""" if subject_type == "decision": return int(subject_id), None if subject_type == "feature": return None, subject_id if subject_type == "page" and subject_id == "discuss": return None, "general" return None, f"{subject_type}:{subject_id}" def permission_of(subject_type: str) -> str: """What a person must hold to see a subject: the running list's items are the owner's page (`plan.view`); every other subject is a client surface (`client.view`).""" return accounts.PERM_PLAN_VIEW if subject_type == "item" else accounts.PERM_CLIENT_VIEW def anchor(subject_type: str, subject_id: str) -> str: """The DOM id of a subject's card: `t--`, so every page anchors the same way.""" return f"t-{subject_type}-{_SLUG.sub('-', subject_id.lower()).strip('-')}" def deep_link(subject_type: str, subject_id: str, comment_id: int | None = None) -> str: """Where a subject lives in the site, absolute when PUBLIC_BASE_URL is set (the inbox is cross-app by design). `#c-` lands on the comment; the client resolves it by id.""" path = f"{PAGE_OF.get(subject_type, '/discuss')}#{anchor(subject_type, subject_id)}" if comment_id is not None: path += f"#c-{comment_id}" base = (settings.public_base_url or "").rstrip("/") return f"{base}{path}" def valid_subject(subject_type: str, subject_id: str) -> bool: return subject_type in SUBJECT_TYPES and 0 < len(subject_id) <= 80 and ":" not in subject_id # --- threads --------------------------------------------------------------------------------- def thread_of(db: Session, subject_type: str, subject_id: str) -> Thread | None: return db.scalar(select(Thread).where(Thread.subject_type == subject_type, Thread.subject_id == subject_id)) def thread_for(db: Session, subject_type: str, subject_id: str, label: str | None = None) -> Thread: """Get or create, safe under two concurrent first comments: the insert runs in a savepoint and an IntegrityError re-reads the row the other writer made (agents.md: every shared row must tolerate a race; a plain select-then-insert lost whole runs).""" t = thread_of(db, subject_type, subject_id) if t is None: try: with db.begin_nested(): t = Thread(subject_type=subject_type, subject_id=subject_id, label=label) db.add(t) db.flush() except IntegrityError: t = thread_of(db, subject_type, subject_id) assert t is not None, (subject_type, subject_id) if label and not t.label: t.label = label return t def live_comments(db: Session, thread_id: int) -> list[DiscussionComment]: """The conversation as a reader sees it: soft-deleted rows excluded.""" return list(db.scalars( select(DiscussionComment) .where(DiscussionComment.thread_id == thread_id, DiscussionComment.deleted_at.is_(None)) .order_by(DiscussionComment.created_at, DiscussionComment.id) )) def comment_out(c: DiscussionComment, viewer_id: int | None = None) -> dict: """`mine` is the one identity fact the page needs: the UI consumes booleans, never ids.""" return { "id": c.id, "author": c.author, "author_id": c.author_id, "body": c.body, "at": c.created_at.isoformat() if c.created_at else None, "edited_at": c.edited_at.isoformat() if c.edited_at else None, "mine": viewer_id is not None and c.author_id == viewer_id, } def thread_out(db: Session, t: Thread | None, viewer_id: int | None = None) -> dict | None: if t is None: return None return { "id": t.id, "subject_type": t.subject_type, "subject_id": t.subject_id, "label": t.label, "anchor": anchor(t.subject_type, t.subject_id), "resolved": t.resolved, "resolved_by_id": t.resolved_by_id, "resolved_at": t.resolved_at.isoformat() if t.resolved_at else None, "comments": [comment_out(c, viewer_id) for c in live_comments(db, t.id)], } def discussion_of(db: Session, subjects: list[dict], viewer_id: int | None = None) -> dict: """One read per work surface: every subject the surface has, in the order given, each with its conversation or `None` (a starter). The caller decides which subjects the reader may see (`permission_of`).""" out, open_n, total = [], 0, 0 for s in subjects: th = thread_out(db, thread_of(db, s["subject_type"], s["subject_id"]), viewer_id) if th and not th["comments"]: th = None # every comment deleted: back to a starter if th: total += 1 open_n += 0 if th["resolved"] else 1 out.append({**s, "thread": th}) return {"subjects": out, "open": open_n, "total": total} def threads_for(db: Session, username: str | None, viewer_id: int | None = None) -> list[dict]: """Every thread the viewer may see, with its live comments; a page joins its own subject list onto this and lists the rest as starters. Running-list threads only for `plan.view`.""" sees_items = accounts.can(username, accounts.PERM_PLAN_VIEW) rows = db.scalars(select(Thread).order_by(Thread.id)) out = [] for t in rows: if t.subject_type == "item" and not sees_items: continue th = thread_out(db, t, viewer_id) if th and th["comments"]: out.append(th) return out # --- people ---------------------------------------------------------------------------------- def _holders(db: Session, permission: str) -> dict[str, Account]: """Active accounts holding a permission (the owner included), keyed by username.""" people = list(db.scalars(select(Account).where(Account.status == "active", Account.username.isnot(None)))) return {a.username: a for a in people if accounts.can(a.username, permission)} @dataclass(frozen=True) class People: """Who a subject can reach, resolved once per write, before the write.""" allowed: dict[str, Account] # holders of the subject's permission curators: frozenset[str] # of those, the `discussion.curate` holders directory: dict[str, Account | None] # what `@name` may reach here def people_for(db: Session, subject_type: str) -> People: allowed = _holders(db, permission_of(subject_type)) curators = frozenset(u for u in _holders(db, accounts.PERM_DISCUSSION_CURATE) if u in allowed) return People(allowed=allowed, curators=curators, directory=mention_directory(db, subject_type, allowed)) def mention_directory(db: Session, subject_type: str, allowed: dict[str, Account] | None = None) -> dict[str, Account | None]: """Who `@name` may reach on a subject: the holders of the subject's permission, and the `claude` handle (no account) where the subject is the owner's: a comment naming it there is a subject a session reads on its next run, and never a notification to anyone.""" out: dict[str, Account | None] = dict(allowed if allowed is not None else _holders(db, permission_of(subject_type))) if permission_of(subject_type) == accounts.PERM_PLAN_VIEW: out.setdefault(CLAUDE_HANDLE, None) return out def mentions_in(body: str) -> list[str]: seen: list[str] = [] for handle in MENTION.findall(body or ""): h = handle.lower().rstrip("._-") # "@mark." in prose names mark if len(h) >= 2 and h not in seen: seen.append(h) return seen def _preview(text: str) -> str: return re.sub(r"\s+", " ", (text or "")).strip()[:_PREVIEW_LEN] def _label(t: Thread) -> str: return t.label or f"{t.subject_type} › {t.subject_id}" # --- emit ------------------------------------------------------------------------------------ def _emit(db: Session, *, recipient: Account, actor_id: int | None, kind: str, context_label: str, category: str, body: str, url: str, source_type: str, source_id: str, dedupe: str) -> bool: assert kind in KINDS, kind if recipient.id == actor_id: return False # never notify the author of their own action key = f"{dedupe}:{recipient.username}" if db.scalar(select(Notification.id).where(Notification.dedupe_key == key)) is not None: return False db.add(Notification( recipient_id=recipient.id, actor_id=actor_id, app=APP_NAME, kind=kind, category=category, context_label=context_label[:220], body=_preview(body), url=url[:400], source_type=source_type, source_id=str(source_id), dedupe_key=key, )) return True def _participants(db: Session, t: Thread, people: People) -> set[str]: by_id = {a.id: u for u, a in people.allowed.items()} return {by_id[c.author_id] for c in live_comments(db, t.id) if c.author_id in by_id} def _notify_comment(db: Session, c: DiscussionComment, t: Thread, people: People, *, only_new_mentions: bool = False) -> None: """Mentions from the STORED body, then the reply fan-out: everyone who took part in the thread plus the curators, each only if they hold the subject's permission. A `@claude` on the owner's subjects notifies nobody (a session reads it); anywhere else it is dropped.""" mentioned = {h: people.directory[h] for h in mentions_in(c.body) if people.directory.get(h) is not None} url = deep_link(t.subject_type, t.subject_id, c.id) common = dict(actor_id=c.author_id, context_label=_label(t), category=t.subject_type, body=c.body, url=url, source_type="comment", source_id=str(c.id)) for u in sorted(mentioned): _emit(db, recipient=mentioned[u], kind="mention", dedupe=f"mention:{c.id}", **common) if only_new_mentions: return # an edit never re-rings the reply fan-out for u in sorted((_participants(db, t, people) | people.curators) - set(mentioned)): _emit(db, recipient=people.allowed[u], kind="reply", dedupe=f"reply:{c.id}", **common) # --- the write paths ------------------------------------------------------------------------- def add_comment(db: Session, *, subject_type: str, subject_id: str, actor: Actor, body: str, label: str | None = None) -> DiscussionComment: """The one write path for conversation. The thread is found or made, the legacy key is set beside it for one release, and mentions are parsed here, from the stored body.""" if not valid_subject(subject_type, subject_id): raise ValueError("unknown subject") body = (body or "").strip()[:2000] if not body: raise ValueError("empty comment") people = people_for(db, subject_type) t = thread_for(db, subject_type, subject_id, label=(label or "").strip()[:160] or None) item_id, feature_key = legacy_of(subject_type, subject_id) c = DiscussionComment(thread_id=t.id, item_id=item_id, feature_key=feature_key, author=actor.display_name, author_id=actor.id, body=body) db.add(c) db.flush() _notify_comment(db, c, t, people) return c def edit_comment(db: Session, c: DiscussionComment, *, body: str) -> DiscussionComment: """The author (or a curator) rewrites the words; `edited_at` says so. A newly named person is mentioned; nobody else hears an edit.""" body = (body or "").strip()[:2000] if not body: raise ValueError("empty comment") t = db.get(Thread, c.thread_id) if c.thread_id is not None else None people = people_for(db, t.subject_type) if t is not None else None c.body = body c.edited_at = utcnow() db.flush() if t is not None and people is not None: _notify_comment(db, c, t, people, only_new_mentions=True) return c def delete_comment(db: Session, c: DiscussionComment) -> None: """Soft: the row stays, the reader stops seeing it.""" c.deleted_at = utcnow() db.flush() def _thread_notifications(db: Session, t: Thread) -> list[Notification]: ids = [str(i) for i in db.scalars(select(DiscussionComment.id).where(DiscussionComment.thread_id == t.id))] if not ids: return [] return list(db.scalars(select(Notification).where( Notification.source_type == "comment", Notification.source_id.in_(ids)))) def resolve_thread(db: Session, t: Thread, *, actor: Actor, resolved: bool) -> Thread: """Resolve or reopen. Resolving flags every notification the thread raised (they leave the unread count) and tells the people in it, once.""" if t.resolved == resolved: return t people = people_for(db, t.subject_type) recipients = sorted((_participants(db, t, people) | people.curators)) if resolved else [] t.resolved = resolved t.resolved_by_id = actor.id if resolved else None t.resolved_at = utcnow() if resolved else None for n in _thread_notifications(db, t): n.resolved = resolved db.flush() if resolved: stamp = int(t.resolved_at.timestamp()) for u in recipients: _emit(db, recipient=people.allowed[u], actor_id=actor.id, kind="resolved", context_label=_label(t), category=t.subject_type, body="Resolved.", url=deep_link(t.subject_type, t.subject_id), source_type="thread", source_id=str(t.id), dedupe=f"resolved:{t.id}:{stamp}") return t def emit_decision(db: Session, *, subject_type: str, subject_id: str, actor: Actor, body: str, label: str | None = None) -> int: """Something was decided on a subject (a running-list decision recorded, a decision card resolved): the people in its thread and the curators hear it as news, never as "needs you". Returns how many rows were written; a subject nobody has spoken about tells nobody but the curators.""" people = people_for(db, subject_type) t = thread_of(db, subject_type, subject_id) context = (label or "").strip()[:160] or (_label(t) if t else f"{subject_type} › {subject_id}") recipients = set(people.curators) | (_participants(db, t, people) if t is not None else set()) stamp = int(utcnow().timestamp()) n = 0 for u in sorted(recipients): n += _emit(db, recipient=people.allowed[u], actor_id=actor.id, kind="decision", context_label=context, category=subject_type, body=body, url=deep_link(subject_type, subject_id), source_type="subject", source_id=f"{subject_type}:{subject_id}", dedupe=f"decision:{subject_type}:{subject_id}:{stamp}") db.flush() return n # --- the inbox ------------------------------------------------------------------------------- def inbox(db: Session, account_id: int, limit: int = 40) -> dict: rows = list(db.scalars( select(Notification).where(Notification.recipient_id == account_id) .order_by(Notification.read, Notification.created_at.desc(), Notification.id.desc()).limit(limit) )) actor_ids = {r.actor_id for r in rows if r.actor_id is not None} names = {} if actor_ids: names = dict(db.execute(select(Account.id, Account.display_name).where(Account.id.in_(actor_ids))).all()) # A resolved row is out of the count and the default list, and stays reachable behind the # bell's "Show resolved". unread = [r for r in rows if not r.read and not r.resolved] return { "unread": len(unread), # "needs you" is stronger than "unread": a mention is addressed to you personally. "needs_you": any(r.kind in NEEDS_YOU for r in unread), "notifications": [{ "id": r.id, "actor": names.get(r.actor_id, "") if r.actor_id else "", "app": r.app, "kind": r.kind, "category": r.category, "context_label": r.context_label, "body": r.body, "url": r.url, "read": r.read, "resolved": r.resolved, "at": r.created_at.isoformat(), } for r in rows], } def mark_read(db: Session, account_id: int, ids: list[int] | None = None) -> int: q = select(Notification).where(Notification.recipient_id == account_id, Notification.read.is_(False)) if ids: q = q.where(Notification.id.in_(ids)) n = 0 for r in db.scalars(q): r.read = True n += 1 db.flush() return n # --- what rian said to Claude ---------------------------------------------------------------- def claude_inbox(db: Session, since: datetime | None = None) -> list[dict]: """The comments that name `@claude` on the owner's subjects (the running list), oldest first: a place for rian to talk to a session, which reads it on its next run (`app.cli discussion inbox --for claude`). A client-visible subject never appears here, because the handle is not in its directory and such a mention is dropped.""" out = [] for c in db.scalars(select(DiscussionComment).where(DiscussionComment.deleted_at.is_(None)) .order_by(DiscussionComment.created_at, DiscussionComment.id)): if CLAUDE_HANDLE not in mentions_in(c.body): continue t = db.get(Thread, c.thread_id) if c.thread_id is not None else None subject = (t.subject_type, t.subject_id) if t else subject_of_legacy(c.item_id, c.feature_key) if permission_of(subject[0]) != accounts.PERM_PLAN_VIEW: continue if since is not None and c.created_at is not None and _aware(c.created_at) < since: continue out.append({ "id": c.id, "at": c.created_at.isoformat() if c.created_at else None, "author": c.author, "subject_type": subject[0], "subject_id": subject[1], "label": t.label if t else None, "resolved": bool(t and t.resolved), "body": c.body, "url": deep_link(subject[0], subject[1], c.id), }) return out def _aware(value: datetime) -> datetime: return value if value.tzinfo is not None else value.replace(tzinfo=UTC) # --- the backfill ---------------------------------------------------------------------------- def backfill_threads(db: Session, args=None) -> str: """Migration #6's data move: a thread per legacy key and `thread_id` set on every comment that lacks one. Idempotent: a second run reports zero. It marks NOTHING as edited: the plan proposed reading an edit from `updated_at` moving past `created_at`, but the who-column backfill of 11 Sep moved `updated_at` on every row it relinked, so the signal cannot tell a relink from a rewrite (empty beats guessed); `edited_at` is stamped by edits from now on. The update below names `updated_at` so the column's onupdate does not move it again.""" threads_made = linked = 0 for c in db.scalars(select(DiscussionComment).where(DiscussionComment.thread_id.is_(None)) .order_by(DiscussionComment.id)): subject_type, subject_id = subject_of_legacy(c.item_id, c.feature_key) before = thread_of(db, subject_type, subject_id) t = thread_for(db, subject_type, subject_id) threads_made += before is None db.execute(update(DiscussionComment).where(DiscussionComment.id == c.id) .values(thread_id=t.id, updated_at=DiscussionComment.updated_at)) linked += 1 db.commit() return f"threads: {threads_made} thread(s) created, {linked} comment(s) linked"