"""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 and_, false, or_, 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 CommentAck, CommentFlag, DiscussionComment, Notification, Thread, ThreadAsk, ThreadRead 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", "thread_reads", "thread_asks", "comment_acks", "comment_flags", ) #: The subject vocabulary; keys are frozen and never renamed (plan §8). `topic` is a thread #: that belongs to no section (T15): started from the panel with a title, its id the slug. SUBJECT_TYPES = frozenset({"decision", "feature", "page", "structure", "quote", "todo", "item", "topic"}) #: 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", "topic": "/discuss?tab=review", } TOPIC_LABEL = "Topic › " _TOPIC_SLUG_LEN = 60 #: 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"}) #: The one mark a thread carries beside its state (rian's card design, 14 Sep): `later` on an #: open thread (parked: the Later tab), `decision` on a resolved one (a decision was made: the #: Decisions view, the green outline); NULL otherwise. `done` is accepted and stored as NULL. OUTCOMES = frozenset({"done", "decision", "later"}) STAGES = frozenset({"current", "later"}) #: The kinds of ask whose button is the action: the ask completes when the person asked does it. ASK_KINDS = frozenset({"later", "resolve", "reply"}) ASK_DEFAULT_NOTE = {"later": "Confirm: move this to Later", "resolve": "Confirm: resolve this", "reply": "Please reply"} 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 page_of(subject_type: str, subject_id: str) -> str: """The page a subject lives on: PAGE_OF by type, except the page subjects that belong to a page of their own (`settings` and its sections live on /settings).""" if subject_type == "page" and (subject_id == "settings" or subject_id.startswith("settings-")): return "/settings" return PAGE_OF.get(subject_type, "/discuss") 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(subject_type, subject_id)}#{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, acks: list[tuple[int, str]] | None = None, flagged: bool = False) -> dict: """`mine` is the one identity fact the page needs: the UI consumes booleans, never ids. `acks` are the "Got it"s on the comment as (account id, display name), `acked` whether the viewer's is among them; `flagged` is the viewer's own flag, nobody else's.""" acks = acks or [] 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, "moved_from_thread_id": c.moved_from_thread_id, "acks": [name for _, name in acks], "acked": viewer_id is not None and any(i == viewer_id for i, _ in acks), "flagged": flagged, } def flags_of(db: Session, comment_ids: list[int], viewer_id: int | None) -> set[int]: """The comments among `comment_ids` the viewer flagged.""" if not comment_ids or viewer_id is None: return set() return set(db.scalars(select(CommentFlag.comment_id).where( CommentFlag.comment_id.in_(comment_ids), CommentFlag.account_id == viewer_id))) def flag_comment(db: Session, c: DiscussionComment, *, actor: Actor) -> dict: """A personal flag on a comment: one tap sets it, a second takes it back; nobody else sees it and nobody is notified.""" row = db.scalar(select(CommentFlag).where(CommentFlag.comment_id == c.id, CommentFlag.account_id == actor.id)) if row is not None: db.delete(row) db.flush() return {"flagged": False} try: with db.begin_nested(): db.add(CommentFlag(comment_id=c.id, account_id=actor.id)) db.flush() except IntegrityError: pass return {"flagged": True} def set_stage(db: Session, t: Thread, *, actor: Actor, stage: str) -> Thread: """Mark an open thread as Later (parked, the Later tab) or Current (the everyday view). A resolved thread is reopened first, so the mark always sits on an open conversation.""" if stage not in STAGES: raise ValueError("unknown stage") people = people_for(db, t.subject_type) # before the first write, as everywhere if t.resolved: resolve_thread(db, t, actor=actor, resolved=False) t.outcome = "later" if stage == "later" else None db.flush() if stage == "later": complete_asks_by_action(db, t, actor=actor, kind="later", people=people) return t def acks_of(db: Session, comment_ids: list[int]) -> dict[int, list[tuple[int, str]]]: """The "Got it"s on a set of comments, oldest first, with the names joined at read time.""" out: dict[int, list[tuple[int, str]]] = {} if not comment_ids: return out rows = db.execute( select(CommentAck.comment_id, CommentAck.account_id, Account.display_name) .join(Account, Account.id == CommentAck.account_id) .where(CommentAck.comment_id.in_(comment_ids)).order_by(CommentAck.created_at, CommentAck.id) ).all() for cid, aid, name in rows: out.setdefault(cid, []).append((aid, name)) return out def asks_out(db: Session, t: Thread, viewer_id: int | None = None) -> list[dict]: """The asks on a thread, open first then done, oldest first; names joined at read time.""" rows = list(db.scalars(select(ThreadAsk).where(ThreadAsk.thread_id == t.id).order_by(ThreadAsk.done_at.isnot(None), ThreadAsk.id))) if not rows: return [] ids = {r.for_id for r in rows} | {r.by_id for r in rows if r.by_id} | {r.done_by_id for r in rows if r.done_by_id} names = dict(db.execute(select(Account.id, Account.display_name).where(Account.id.in_(ids))).all()) return [{ "id": r.id, "for": names.get(r.for_id, ""), "for_id": r.for_id, "note": r.note, "by": names.get(r.by_id, "") if r.by_id else "", "by_id": r.by_id, "at": _iso(r.created_at), "done_at": _iso(r.done_at), "done_by": names.get(r.done_by_id, "") if r.done_by_id else None, "kind": r.kind, "mine": viewer_id is not None and r.for_id == viewer_id, "by_me": viewer_id is not None and r.by_id == viewer_id, } for r in rows] def _iso(value: datetime | None) -> str | None: return _aware(value).isoformat() if value is not None else None def thread_out(db: Session, t: Thread | None, viewer_id: int | None = None, *, read_at: datetime | None = None, needs_you: bool = False) -> dict | None: """The thread with its live comments, and what it is to the viewer: `unread` is a comment by someone else after `read_at` (the viewer's own words are never unread to them; no read row means never opened), `needs_you` is the caller's finding from the notification rows (`needs_you_threads`), so the bell and the panel agree by construction. `started_at` is the first comment, `last_activity_at` the newest comment or edit: the two sorts.""" if t is None: return None comments = live_comments(db, t.id) started = comments[0].created_at if comments else t.created_at last = max((max(_aware(c.created_at), _aware(c.edited_at)) if c.edited_at else _aware(c.created_at) for c in comments), default=_aware(t.created_at)) unread = viewer_id is not None and any( c.author_id != viewer_id and (read_at is None or _aware(c.created_at) > _aware(read_at)) for c in comments) followup = None if t.followup_note: who = db.get(Account, t.followup_by_id) if t.followup_by_id else None followup = {"note": t.followup_note, "by": who.display_name if who else "", "by_id": t.followup_by_id, "at": _iso(t.followup_at)} resolver = db.get(Account, t.resolved_by_id) if t.resolved and t.resolved_by_id else None closing = next((c for c in comments if c.id == t.closing_comment_id), None) if t.closing_comment_id else None acks = acks_of(db, [c.id for c in comments]) flags = flags_of(db, [c.id for c in comments], viewer_id) 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, "resolved_by": resolver.display_name if resolver else None, "outcome": t.outcome, "later": (not t.resolved) and t.outcome == "later", "decision": t.resolved and t.outcome == "decision", "flagged": bool(flags), "closing_word": closing.body if closing else None, "archived": t.archived_at is not None, "started_at": _iso(started), "last_activity_at": last.isoformat(), "reply_count": max(len(comments) - 1, 0), "read_at": _iso(read_at), "unread": unread, "needs_you": needs_you and not t.resolved, "followup": followup, "asks": asks_out(db, t, viewer_id), "comments": [comment_out(c, viewer_id, acks.get(c.id), c.id in flags) for c in comments], } 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 mark_thread_unread(db: Session, t: Thread, account_id: int) -> dict: """The viewer wants the thread back as unread (the circle beside the link, rian 14 Sep): their read row goes, so any comment by someone else counts as unread again. A thread with nobody else's words in it has nothing to be unread; the row goes all the same.""" row = db.scalar(select(ThreadRead).where(ThreadRead.thread_id == t.id, ThreadRead.account_id == account_id)) if row is not None: db.delete(row) db.flush() return {"thread_id": t.id, "read_at": None} def needs_you_threads(db: Session, viewer_id: int) -> set[int]: """The threads that need the viewer (the workflow plan §3): an unread, unresolved mention of them (the SAME rows the bell's needs-you state comes from, mapped to their threads), or an open ask for them.""" ids = [int(i) for i in db.scalars(select(Notification.source_id).where( Notification.recipient_id == viewer_id, Notification.read.is_(False), Notification.resolved.is_(False), Notification.kind.in_(sorted(NEEDS_YOU)), Notification.source_type == "comment")) if i.isdigit()] out: set[int] = set() if ids: out |= set(db.scalars(select(DiscussionComment.thread_id).where( DiscussionComment.id.in_(ids), DiscussionComment.deleted_at.is_(None), DiscussionComment.thread_id.isnot(None)))) out |= set(db.scalars(select(ThreadAsk.thread_id).where(ThreadAsk.for_id == viewer_id, ThreadAsk.done_at.is_(None)))) return out def threads_for(db: Session, username: str | None, viewer_id: int | None = None, since: datetime | None = None) -> list[dict]: """Every thread the viewer may see, with its live comments and what it is to them; a page joins its own subject list onto this and lists the rest as starters, the panel lists it whole. Running-list threads only for `plan.view`. `since` keeps the threads that moved on or after it (a comment or an edit).""" sees_items = accounts.can(username, accounts.PERM_PLAN_VIEW) reads: dict[int, datetime] = {} needs: set[int] = set() if viewer_id is not None: reads = {r.thread_id: r.read_at for r in db.scalars(select(ThreadRead).where(ThreadRead.account_id == viewer_id))} needs = needs_you_threads(db, viewer_id) 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, read_at=reads.get(t.id), needs_you=t.id in needs) if not th or not th["comments"]: continue if since is not None and datetime.fromisoformat(th["last_activity_at"]) < _aware(since): continue out.append(th) return out def mark_thread_read(db: Session, t: Thread, account_id: int) -> dict: """The viewer opened the thread: their read row moves to now (made if absent, inside a savepoint because two tabs can open the same thread at once), and their notifications the thread raised leave the unread count, so the bell and the panel say the same thing.""" now = utcnow() row = db.scalar(select(ThreadRead).where(ThreadRead.thread_id == t.id, ThreadRead.account_id == account_id)) if row is None: try: with db.begin_nested(): row = ThreadRead(thread_id=t.id, account_id=account_id, read_at=now) db.add(row) db.flush() except IntegrityError: row = db.scalar(select(ThreadRead).where(ThreadRead.thread_id == t.id, ThreadRead.account_id == account_id)) assert row is not None, (t.id, account_id) row.read_at = now else: row.read_at = now ids = [str(i) for i in db.scalars(select(DiscussionComment.id).where(DiscussionComment.thread_id == t.id))] ask_ids = [str(i) for i in db.scalars(select(ThreadAsk.id).where(ThreadAsk.thread_id == t.id))] stale = db.execute(update(Notification).where( Notification.recipient_id == account_id, Notification.read.is_(False), or_(and_(Notification.source_type == "comment", Notification.source_id.in_(ids)) if ids else false(), and_(Notification.source_type == "ask", Notification.source_id.in_(ask_ids)) if ask_ids else false(), and_(Notification.source_type == "thread", Notification.source_id == str(t.id))), ).values(read=True)) db.flush() return {"thread_id": t.id, "read_at": now.isoformat(), "notifications_read": int(stale.rowcount or 0)} # --- 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, created_at: datetime | None = None, resolved: bool = False) -> Notification | None: """One row, or None when it must not exist: the author of the action, or a key already emitted (a retry, a replay, the backfill after a hand emission). `created_at` lets the legacy backfill date a row by the comment it rings for; `resolved` files it behind "Show resolved" from birth (a thread resolved before the row existed).""" assert kind in KINDS, kind if recipient.id == actor_id: return None # 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 None row = 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, resolved=resolved, ) if created_at is not None: row.created_at = created_at db.add(row) return row 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, people: People | 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. A caller that has already resolved the people (a resolve with a closing word) passes them, so the permission lookup stays ahead of the first write.""" if not valid_subject(subject_type, subject_id): raise ValueError("unknown subject") body = (body or "").strip()[:2000] if not body: raise ValueError("empty comment") if people is None: 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) complete_asks_by_action(db, t, actor=actor, kind="reply", people=people) return c def topic_slug(title: str) -> str: """The id a topic gets from its title: lowercase, words joined by hyphens, cut to leave room for a collision suffix. Empty when the title has no letter or digit.""" return _SLUG.sub("-", (title or "").lower()).strip("-")[:_TOPIC_SLUG_LEN].strip("-") def start_topic(db: Session, *, title: str, body: str, actor: Actor) -> Thread: """A thread that belongs to no section: a title and a first comment, from the panel. The slug is the id; a second topic with the same words gets `-2`, `-3`, never the first one's conversation. Written through the one comment path, so mentions and the reply fan-out behave as on any subject.""" title = re.sub(r"\s+", " ", (title or "")).strip()[:120] base = topic_slug(title) if not base: raise ValueError("empty title") slug, n = base, 1 while thread_of(db, "topic", slug) is not None: n += 1 slug = f"{base}-{n}" c = add_comment(db, subject_type="topic", subject_id=slug, actor=actor, body=body, label=TOPIC_LABEL + title) t = db.get(Thread, c.thread_id) assert t is not None return t 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, comment: str | None = None, outcome: str = "done") -> Thread: """Resolve or reopen, with an optional closing (or reopening) word (T17) and an outcome (the workflow plan §2: `done`, concluded; `later`, parked, the thread kept and the discussion free to go on): the words are posted first as the thread's last comment by the actor, through the one comment path, then the state changes, in the same write; the closing comment's id is kept on the thread as the record. Resolving flags every notification the thread raised (the closing word's own reply rows included, so it never rings twice) and tells the people in it once, the `resolved` event carrying the words; reopening un-flags them, clears the outcome, and the reopening word reaches them as an ordinary reply. Recorded against the account (`resolved_by_id`, the comment's `author_id`).""" if outcome not in OUTCOMES: raise ValueError("unknown outcome") words = (comment or "").strip() people = people_for(db, t.subject_type) # before the first write, as everywhere closing = None if words: closing = add_comment(db, subject_type=t.subject_type, subject_id=t.subject_id, actor=actor, body=words, label=t.label, people=people) db.flush() # the sessions never autoflush: the word's own reply rows must exist to be flagged below if t.resolved == resolved: if resolved and closing is not None: t.closing_comment_id = closing.id # a later word on a resolved thread becomes its record db.flush() return t 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 # The mark: `decision` (or the older `later`) on a resolution; a plain `done` is NULL; a # reopened thread carries none. t.outcome = outcome if (resolved and outcome != "done") else None t.closing_comment_id = closing.id if (resolved and closing is not None) else None for n in _thread_notifications(db, t): n.resolved = resolved db.flush() if resolved: complete_asks_by_action(db, t, actor=actor, kind="resolve", people=people) # Microseconds, not seconds: resolve, reopen and resolve again inside one second is one # event each, not one event deduped away. stamp = t.resolved_at.timestamp() fallback = "Parked for later." if outcome == "later" else "Decision made." if outcome == "decision" else "Resolved." 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=words or fallback, 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 archive_thread(db: Session, t: Thread, *, actor: Actor, archived: bool) -> Thread: """Put a resolved conversation away for good, or bring it back (the workflow plan §2, a curator's verb): incorporated into the build, or no longer relevant. Hidden from every list unless asked for, never deleted, its closing word still in the decisions log. An open thread is resolved (Done) first, so nothing archived is ever an open issue.""" if archived: if not t.resolved: resolve_thread(db, t, actor=actor, resolved=True) t.archived_at = utcnow() t.archived_by_id = actor.id else: t.archived_at = None t.archived_by_id = None db.flush() return t def participants_of(db: Session, t: Thread) -> set[int]: """The account ids that wrote a live comment in the thread: who may resolve or reopen it (with the curators), per the workflow plan §7.""" return {c.author_id for c in live_comments(db, t.id) if c.author_id is not None} def set_followup(db: Session, t: Thread, *, actor: Actor, note: str) -> Thread: """A curator marks the thread as needing follow-up (T13): one line saying what, who and when, on the thread itself. Not a task system: one mark per thread, replaced by a new one, cleared with `clear_followup`. Recorded against the account (`followup_by_id`).""" note = re.sub(r"\s+", " ", (note or "")).strip()[:400] if not note: raise ValueError("empty note") t.followup_note = note t.followup_by_id = actor.id t.followup_at = utcnow() db.flush() return t def clear_followup(db: Session, t: Thread, *, actor: Actor, note: str | None = None) -> Thread: """The follow-up is done: the mark goes, and the note, when given, is posted as the curator's comment through the one comment path (the people in the thread hear it as a reply). The permission lookup precedes the write, as everywhere.""" words = (note or "").strip() if words: people = people_for(db, t.subject_type) add_comment(db, subject_type=t.subject_type, subject_id=t.subject_id, actor=actor, body=words, label=t.label, people=people) t.followup_note = None t.followup_by_id = None t.followup_at = None db.flush() return t def move_comment(db: Session, c: DiscussionComment, *, subject_type: str, subject_id: str, actor: Actor, label: str | None = None) -> DiscussionComment: """Re-file a comment under another subject (T18; the standard's §4.1 re-parents through the relationship, never by deleting). The words, the author and the times never change; the comment keeps who moved it and the thread it came from; the legacy key follows so the one-release reads agree; the notifications the comment raised are re-addressed (url, context, category) so the bell's links still land. A comment already on the target is left alone, so a re-run changes nothing. The target thread is found or made (a starter until the comment arrives); the emptied thread stays, a starter again by the reads.""" if not valid_subject(subject_type, subject_id): raise ValueError("unknown subject") target = thread_for(db, subject_type, subject_id, label=(label or "").strip()[:160] or None) if c.thread_id == target.id: return c c.moved_from_thread_id = c.thread_id c.moved_by_id = actor.id c.moved_at = utcnow() c.thread_id = target.id c.item_id, c.feature_key = legacy_of(subject_type, subject_id) db.flush() for n in db.scalars(select(Notification).where(Notification.source_type == "comment", Notification.source_id == str(c.id))): n.url = deep_link(subject_type, subject_id, c.id)[:400] n.context_label = _label(target)[:220] n.category = subject_type db.flush() return c def rethread(db: Session, plan: dict, *, actor: Actor, check: bool = False) -> list[str]: """Apply a re-threading map (T18; `import/rethread-.json`): the topics it names are made with their titles, each move re-files one comment, and the listed comments are soft-deleted. Idempotent: a topic that exists is kept, a comment already on its target or already deleted is reported and left, a comment id the database lacks is reported and skipped. `check` walks the map and writes NOTHING (no savepoint, no flush), rather than writing and rolling back: a check must not depend on transaction semantics. Returns one line per entry, for the operator to read before and after.""" lines: list[str] = [] for topic in plan.get("topics", []): label = TOPIC_LABEL + topic["title"] before = thread_of(db, "topic", topic["id"]) if before is not None: lines.append(f"topic {topic['id']}: exists ({before.label})") elif check: lines.append(f"topic {topic['id']}: would create ({label})") else: t = thread_for(db, "topic", topic["id"], label=label) lines.append(f"topic {topic['id']}: created ({t.label})") for move in plan.get("moves", []): c = db.get(DiscussionComment, int(move["comment"])) head, sep, tail = str(move["to"]).partition(":") if c is None: lines.append(f"move {move['comment']} -> {move['to']}: no such comment, skipped") continue if not sep or not valid_subject(head, tail): lines.append(f"move {move['comment']} -> {move['to']}: not a subject, skipped") continue target = thread_of(db, head, tail) if target is not None and c.thread_id == target.id: lines.append(f"move {move['comment']} -> {move['to']}: already there") continue origin = db.get(Thread, c.thread_id) if c.thread_id else None where = f"{origin.subject_type}:{origin.subject_id}" if origin else "(no thread)" if check: lines.append(f"move {c.id} -> {move['to']}: would move from {where}") continue move_comment(db, c, subject_type=head, subject_id=tail, actor=actor, label=move.get("label")) lines.append(f"move {c.id} -> {move['to']}: moved from {where}") for cid in plan.get("soft_delete", []): c = db.get(DiscussionComment, int(cid)) if c is None: lines.append(f"delete {cid}: no such comment, skipped") elif c.deleted_at is not None: lines.append(f"delete {cid}: already deleted") elif check: lines.append(f"delete {cid}: would soft-delete ({c.body[:40]!r})") else: delete_comment(db, c) lines.append(f"delete {cid}: soft-deleted ({c.body[:40]!r})") return lines def ask(db: Session, t: Thread, *, actor: Actor, for_username: str, note: str, kind: str | None = None) -> ThreadAsk: """The one hand-off (the workflow plan §2): on a thread, for one person, a line saying what. `kind` (rian, 14 Sep) makes the ask an action: `later` (confirm and move to Later), `resolve` (confirm and resolve), `reply` (please reply); the person asked gets the button, and the ask completes when they do it (`complete_asks_by_action`); with a kind the words are optional and default to the kind's line. The person must be someone `@name` could reach on the subject (the same directory the mentions use, so a plan thread can never be handed to a client); asking oneself is allowed (a build note to Rian). Rings the person as a `turn`, needs-you, under the ask's own dedupe key; the asker never rings themselves. The permission lookup precedes the write.""" if kind is not None and kind not in ASK_KINDS: raise ValueError("unknown kind") note = re.sub(r"\s+", " ", (note or "")).strip()[:400] if not note: if kind is None: raise ValueError("empty note") note = ASK_DEFAULT_NOTE[kind] people = people_for(db, t.subject_type) target = people.directory.get((for_username or "").strip().lower()) if target is None: raise LookupError("not someone this conversation can reach") row = ThreadAsk(thread_id=t.id, for_id=target.id, by_id=actor.id, note=note, kind=kind) db.add(row) db.flush() _emit(db, recipient=target, actor_id=actor.id, kind="turn", context_label=_label(t), category=t.subject_type, body=note, url=deep_link(t.subject_type, t.subject_id), source_type="ask", source_id=str(row.id), dedupe=f"ask:{row.id}") db.flush() return row def _finish_ask(db: Session, row: ThreadAsk, t: Thread, actor: Actor, people: People, word: str | None) -> None: row.done_at = utcnow() row.done_by_id = actor.id # The person's turn row leaves the bell's count; the asker hears it is done, as news. for n in db.scalars(select(Notification).where(Notification.source_type == "ask", Notification.source_id == str(row.id))): n.read = True n.resolved = True asker = db.get(Account, row.by_id) if row.by_id else None if asker is not None and asker.username in people.allowed: _emit(db, recipient=asker, actor_id=actor.id, kind="status", context_label=_label(t), category=t.subject_type, body=f"Done: {row.note}" + (f" ({word})" if word else ""), url=deep_link(t.subject_type, t.subject_id), source_type="ask", source_id=str(row.id), dedupe=f"askdone:{row.id}") def finish_ask(db: Session, row: ThreadAsk, *, actor: Actor, word: str | None = None) -> ThreadAsk: """The ask is done, by the person asked or a curator; the word, if any, is their comment on the thread through the one comment path. Doing a done ask again changes nothing.""" t = db.get(Thread, row.thread_id) assert t is not None people = people_for(db, t.subject_type) words = (word or "").strip() if words: add_comment(db, subject_type=t.subject_type, subject_id=t.subject_id, actor=actor, body=words, label=t.label, people=people) if row.done_at is None: _finish_ask(db, row, t, actor, people, words or None) db.flush() return row def edit_ask(db: Session, row: ThreadAsk, *, note: str | None, kind: str | None, keep_kind: bool) -> ThreadAsk: """The asker (or a curator) rewrites an open ask: the words, the kind, or both. The person stays (to ask someone else, delete and ask again). The turn row's preview follows the words.""" if not keep_kind: if kind is not None and kind not in ASK_KINDS: raise ValueError("unknown kind") row.kind = kind if note is not None: words = re.sub(r"\s+", " ", note).strip()[:400] if not words: if row.kind is None: raise ValueError("empty note") words = ASK_DEFAULT_NOTE[row.kind] row.note = words elif row.kind is not None and row.note in ASK_DEFAULT_NOTE.values(): row.note = ASK_DEFAULT_NOTE[row.kind] for n in db.scalars(select(Notification).where(Notification.source_type == "ask", Notification.source_id == str(row.id))): n.body = _preview(row.note) db.flush() return row def delete_ask(db: Session, row: ThreadAsk) -> None: """The asker (or a curator) withdraws an ask: the row goes and its turn row leaves the person's count. An ask is a hand-off, not a conversation, so nothing is kept.""" for n in db.scalars(select(Notification).where(Notification.source_type == "ask", Notification.source_id == str(row.id))): n.read = True n.resolved = True db.delete(row) db.flush() def complete_asks_by_action(db: Session, t: Thread, *, actor: Actor, kind: str, people: People) -> int: """The person asked did the thing (replied, resolved, marked Later): their open asks of that kind on the thread complete, the asker told as news. Nobody else's action completes them.""" n = 0 for a in db.scalars(select(ThreadAsk).where(ThreadAsk.thread_id == t.id, ThreadAsk.for_id == actor.id, ThreadAsk.kind == kind, ThreadAsk.done_at.is_(None))): _finish_ask(db, a, t, actor, people, None) n += 1 if n: db.flush() return n def acknowledge(db: Session, c: DiscussionComment, *, actor: Actor) -> dict: """"Got it" (the workflow plan §2): seen and agreed, one tap, no words; a second tap takes it back. Setting it also completes the person's open asks on that thread (the asker hears it as news), which is how "confirm you have seen this" closes without a 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 # before the first write row = db.scalar(select(CommentAck).where(CommentAck.comment_id == c.id, CommentAck.account_id == actor.id)) if row is not None: db.delete(row) db.flush() return {"acked": False, "asks_done": 0} try: with db.begin_nested(): db.add(CommentAck(comment_id=c.id, account_id=actor.id)) db.flush() except IntegrityError: return {"acked": True, "asks_done": 0} done = 0 if t is not None and people is not None: for a in db.scalars(select(ThreadAsk).where(ThreadAsk.thread_id == t.id, ThreadAsk.for_id == actor.id, ThreadAsk.done_at.is_(None))): _finish_ask(db, a, t, actor, people, None) done += 1 db.flush() return {"acked": True, "asks_done": done} 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}") is not None 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, then the legacy mentions rung (`backfill_legacy_mentions`). 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() rung = backfill_legacy_mentions(db) return f"threads: {threads_made} thread(s) created, {linked} comment(s) linked; {rung}" def backfill_thread_reads_seed(db: Session, args=None) -> str: """The one-time read seed (the workflow plan §3): everything before today is read for everyone, so "unread" starts at zero and means new activity from now on. A read row is inserted, stamped now, for every active account and every thread with a live comment where none exists; a row that exists is left alone, so a second run inserts nothing.""" now = utcnow() people = list(db.scalars(select(Account).where(Account.status == "active", Account.username.isnot(None)))) threads = list(db.scalars(select(Thread.id).where(Thread.id.in_( select(DiscussionComment.thread_id).where(DiscussionComment.deleted_at.is_(None), DiscussionComment.thread_id.isnot(None)))))) have = {(r.thread_id, r.account_id) for r in db.scalars(select(ThreadRead))} inserted = 0 for a in people: for tid in threads: if (tid, a.id) in have: continue db.add(ThreadRead(thread_id=tid, account_id=a.id, read_at=now)) inserted += 1 db.commit() return f"thread_reads_seed: {inserted} row(s) inserted for {len(people)} account(s) over {len(threads)} thread(s)" def backfill_legacy_mentions(db: Session, args=None) -> str: """The mention that never rang (T14): a comment written before the write path emitted notifications, or linked to its thread by the backfill, named `@someone` and nobody heard. For every live comment on a thread, the STORED body's mentions are resolved against the subject's directory (the same rule as the write path: a name that cannot see the subject is dropped, `@claude` rings nobody, never the author) and a `mention` row emitted under the write path's own dedupe key (`mention::`), so a row already emitted by hand or by the write path is left alone and a second run reports zero. The row is dated by the comment, and filed resolved from birth when its thread already is.""" directories: dict[str, dict[str, Account | None]] = {} labels: dict[int, Thread] = {} emitted = comments = 0 for c in db.scalars(select(DiscussionComment).where(DiscussionComment.deleted_at.is_(None), DiscussionComment.thread_id.isnot(None)) .order_by(DiscussionComment.id)): handles = mentions_in(c.body) if not handles: continue t = labels.get(c.thread_id) or db.get(Thread, c.thread_id) if t is None: continue labels[t.id] = t directory = directories.get(t.subject_type) if directory is None: directory = directories[t.subject_type] = mention_directory(db, t.subject_type) url = deep_link(t.subject_type, t.subject_id, c.id) rang = 0 for h in handles: recipient = directory.get(h) if recipient is None: continue row = _emit(db, recipient=recipient, actor_id=c.author_id, kind="mention", context_label=_label(t), category=t.subject_type, body=c.body, url=url, source_type="comment", source_id=str(c.id), dedupe=f"mention:{c.id}", created_at=c.created_at, resolved=t.resolved) rang += row is not None if rang: emitted += rang comments += 1 db.commit() return f"legacy_mentions: {emitted} mention(s) rung for {comments} comment(s)"