"""Project domain logic: the objects behind /api/projects, the snapshot the status rollup consumes, the append-only event stream, and the side resolver ("manager" = holds projects.manage on this project; "client" = access without it). Routers stay thin; everything stateful happens here. """ import logging from datetime import datetime, timezone from sqlalchemy import func, select from sqlalchemy.orm import Session, selectinload from app import accounts from app.models import ( Attachment, Comment, CommentRevision, Event, MockupFile, Option, OptionVariant, Pin, ProjectDetails, ProjectVisit, Screen, StepApproval, Thread, TourProgress, WalkthroughStep, ) from app.tour import TOOL_TOUR_VERSION from app.services.rollup import ProjectSnapshot, ScreenSnapshot, rollup log = logging.getLogger(__name__) class DomainError(Exception): def __init__(self, message: str, code: str, status: int = 400): super().__init__(message) self.code = code self.status = status def utcnow() -> datetime: return datetime.now(timezone.utc) # ------------------------------------------------------------------ sides def side_of(username: str, project_id: str) -> str | None: """'manager' | 'client' | None (no access). Capability-based, never a level name: any level carrying projects.manage is the manager side.""" if accounts.is_owner(username): return "manager" if accounts.effective_level(username, project_id) is None: return None if accounts.instance_can(username, project_id, accounts.PERM_PROJECTS_MANAGE): return "manager" return "client" def visible_screens(side: str | None, screens: list[Screen]) -> list[Screen]: """What this side may see. A draft screen is ours while it is being built — the client meets it when we publish it, not when we start it.""" if side == "manager": return list(screens) return [s for s in screens if s.published] def visible_options(side: str | None, options) -> list: """What this side is allowed to see. An option with no mockup is work in progress: the manager needs it (that is where the upload control lives), the client must never meet it — a fileless option renders as a dead frame and invites feedback on nothing. Mirrors rollup's "ready" definition.""" if side == "manager": return list(options) return [o for o in options if o.files] def option_visible_to(db: Session, username: str, option) -> bool: """Guard for the per-option routes, so hiding an option in the project view is not merely cosmetic — the client cannot reach it by id either.""" screen = db.get(Screen, option.screen_id) side = side_of(username, screen.instance_id) if side is None: return False return side == "manager" or bool(option.files) def capabilities(username: str, project_id: str) -> dict: side = side_of(username, project_id) return { "can_view": side is not None, "can_manage": side == "manager", "can_comment": side is not None, "can_select": side is not None, # Approving a walkthrough beat is the client's act; a manager approving # their own presentation would be noise (and rollup ignores it anyway). "can_approve": side == "client", } # ------------------------------------------------------------------ events def record(db: Session, project_id: str, actor: str, kind: str, payload: dict | None = None) -> None: """Append-only. Nothing in the app updates or deletes an Event row.""" db.add(Event(instance_id=project_id, actor=actor, kind=kind, payload=payload)) db.flush() # ------------------------------------------------------------------ details def ensure_details(db: Session, project_id: str) -> ProjectDetails: row = db.get(ProjectDetails, project_id) if row is None: row = ProjectDetails(instance_id=project_id) db.add(row) db.flush() return row def screens_of(db: Session, project_id: str) -> list[Screen]: return list(db.scalars( select(Screen).where(Screen.instance_id == project_id) .order_by(Screen.position, Screen.id) .options(selectinload(Screen.options).selectinload(Option.files), selectinload(Screen.options).selectinload(Option.variants)))) def screen_project(db: Session, screen_id: int) -> tuple[Screen, str]: screen = db.get(Screen, screen_id) if screen is None: raise DomainError("No such screen.", "NO_SUCH_SCREEN", 404) return screen, screen.instance_id def option_project(db: Session, option_id: int) -> tuple[Option, str]: option = db.get(Option, option_id) if option is None: raise DomainError("No such option.", "NO_SUCH_OPTION", 404) screen = db.get(Screen, option.screen_id) return option, screen.instance_id def slugify(name: str) -> str: out = "".join(c if c.isalnum() else "-" for c in name.lower()) out = "-".join(p for p in out.split("-") if p) return out[:60] or "screen" def add_screen(db: Session, project_id: str, title: str) -> Screen: base = slugify(title) existing = {s.slug for s in screens_of(db, project_id)} slug, n = base, 2 while slug in existing: slug, n = f"{base}-{n}", n + 1 position = (db.scalar(select(func.max(Screen.position)).where( Screen.instance_id == project_id)) or 0) + 1 screen = Screen(instance_id=project_id, title=title, slug=slug, position=position) db.add(screen) db.flush() return screen def add_option(db: Session, screen: Screen, title: str, concept_tag: str) -> Option: position = (db.scalar(select(func.max(Option.position)).where( Option.screen_id == screen.id)) or 0) + 1 option = Option(screen_id=screen.id, title=title, concept_tag=concept_tag, position=position) db.add(option) db.flush() return option def axes_of(option: Option) -> dict[str, list[OptionVariant]]: """The option's variant choices, grouped by axis and ordered.""" out: dict[str, list[OptionVariant]] = {} for v in option.variants: out.setdefault(v.axis, []).append(v) return out def default_variants(option: Option) -> dict[str, str]: """What the client sees before touching anything — the first choice on each axis unless one is flagged default.""" picked: dict[str, str] = {} for axis, choices in axes_of(option).items(): chosen = next((c for c in choices if c.is_default), choices[0] if choices else None) if chosen is not None: picked[axis] = chosen.key return picked def variants_complete(option: Option, chosen: dict | None) -> bool: """Every axis answered with a value that actually exists. This is what makes a variant choice part of the locked direction rather than decoration: a screen whose option has axes is not decided until they are all answered, so the rollup cannot report `done` on half a decision. """ axes = axes_of(option) if not axes: return True if not chosen: return False for axis, choices in axes.items(): value = chosen.get(axis) if value is None or not any(c.key == value for c in choices): return False return True def screen_decided(db: Session, screen: Screen) -> bool: if screen.selected_option_id is None: return False option = db.get(Option, screen.selected_option_id) if option is None: return False return variants_complete(option, screen.selected_variants) def last_seen(db: Session, project_id: str, username: str) -> datetime | None: """When this person last looked. None on a first visit — and a first visit is deliberately NOT "everything is new": arriving to find every single thing flagged tells you nothing, so newness only starts counting once there is a previous visit to be new since.""" row = db.scalar(select(ProjectVisit).where( ProjectVisit.instance_id == project_id, ProjectVisit.username == username)) return row.last_seen_at if row else None def mark_seen(db: Session, project_id: str, username: str) -> None: row = db.scalar(select(ProjectVisit).where( ProjectVisit.instance_id == project_id, ProjectVisit.username == username)) if row is None: db.add(ProjectVisit(instance_id=project_id, username=username)) else: row.last_seen_at = utcnow() db.flush() def touch_active(db: Session, project_id: str, username: str) -> None: """The heartbeat clock. Never touches `last_seen_at` — see the model.""" row = db.scalar(select(ProjectVisit).where( ProjectVisit.instance_id == project_id, ProjectVisit.username == username)) if row is None: # A first heartbeat should not backdate newness to "now" and swallow a # first visit's flags, so seed the seen marker far in the past. row = ProjectVisit(instance_id=project_id, username=username, last_seen_at=datetime(1970, 1, 1, tzinfo=timezone.utc)) db.add(row) row.last_active_at = utcnow() db.flush() def client_presence(db: Session, project_id: str) -> list[dict]: """Who on the client side has been here, and when. The single most common anxiety in this workflow is "has he even looked?" — and until now nothing in easel could answer it, though the data was one column away. """ seen = {v.username: v for v in db.scalars(select(ProjectVisit).where( ProjectVisit.instance_id == project_id))} out = [] from app import bw_accounts as bwa for m in bwa.members(): member = m["username"] # Client side = has access to THIS project without projects.manage. if side_of(member, project_id) != "client": continue row = seen.get(member) out.append({"username": member, "last_active_at": row.last_active_at if row else None, "last_seen_at": row.last_seen_at if row else None}) out.sort(key=lambda m: (m["last_active_at"] is None, -(m["last_active_at"].timestamp() if m["last_active_at"] else 0))) return out def newness(db: Session, screens: list[Screen], since: datetime | None) -> tuple[dict[int, bool], dict[int, str]]: """Per-screen and per-option newness against a previous visit. An option is `new` if it did not exist last time, or `changed` if it did but has since gained a mockup file or a walkthrough beat. The distinction earns its keep: "there is another design to look at" and "the one you already read has moved on" are different messages to a returning client. """ screen_new: dict[int, bool] = {} option_state: dict[int, str] = {} if since is None: return screen_new, option_state option_ids = [o.id for s in screens for o in s.options] touched: set[int] = set() if option_ids: for oid in db.scalars(select(MockupFile.option_id).where( MockupFile.option_id.in_(option_ids), MockupFile.created_at > since)): touched.add(oid) for oid in db.scalars(select(WalkthroughStep.option_id).where( WalkthroughStep.option_id.in_(option_ids), WalkthroughStep.created_at > since)): touched.add(oid) for screen in screens: screen_new[screen.id] = screen.created_at > since for option in screen.options: if option.created_at > since: option_state[option.id] = "new" elif option.id in touched: option_state[option.id] = "changed" return screen_new, option_state def thread_is_live(thread: Thread) -> bool: """A thread with nothing left to read is not outstanding feedback. Comments are SOFT-deleted — the standard's rule, and the right one, since a conversation is a record. But a thread whose every comment has been removed still had `resolved == False`, so it went on turning the board and counting as something we owed an answer to, forever. What is open is the reading, not the row. """ return any(c.deleted_at is None for c in thread.comments) def screen_thread(db: Session, screen: Screen) -> Thread | None: """The one thread about this screen. `UNIQUE(subject_type, subject_id)` in the standard means a screen has AT MOST ONE — a running conversation, not a pile of notes. That is the point: pins are the pile.""" return db.scalar(select(Thread).where( Thread.subject_type == "screen", Thread.subject_id == str(screen.id))) def ensure_screen_thread(db: Session, screen: Screen, actor: str) -> Thread: thread = screen_thread(db, screen) if thread is None: thread = Thread(project_id=screen.instance_id, subject_type="screen", subject_id=str(screen.id), title=screen.title, created_by=actor) db.add(thread) db.flush() return thread def step_thread(db: Session, step: WalkthroughStep) -> Thread | None: """The one conversation about a walkthrough beat — the same shape as a screen's: a beat is a subject, and a subject has at most one thread. The question about a point is asked inside the tour, where the point is shown, instead of pausing the tour to drop a pin somewhere near the spot.""" return db.scalar(select(Thread).where( Thread.subject_type == "step", Thread.subject_id == str(step.id))) def ensure_step_thread(db: Session, step: WalkthroughStep, project_id: str, actor: str) -> Thread: thread = step_thread(db, step) if thread is None: thread = Thread(project_id=project_id, subject_type="step", subject_id=str(step.id), title=step.title, created_by=actor) db.add(thread) db.flush() return thread def option_thread(db: Session, option: Option) -> Thread | None: """The one conversation about a concept as a whole — "I like this one but the header feels heavy" — a fourth subject on the same seam as screens and beats.""" return db.scalar(select(Thread).where( Thread.subject_type == "option", Thread.subject_id == str(option.id))) def ensure_option_thread(db: Session, option: Option, project_id: str, actor: str) -> Thread: thread = option_thread(db, option) if thread is None: thread = Thread(project_id=project_id, subject_type="option", subject_id=str(option.id), title=option.title, created_by=actor) db.add(thread) db.flush() return thread def move_thread(db: Session, source: Thread, target: Thread, project_id: str, actor: str) -> Thread: """Re-file a conversation under another subject of the same design — a note on a spot that turns out to be about the whole page, a page remark that belongs to one version. The comments keep their authors and times and join the target's conversation (which the caller ensured exists); the source thread goes, and with it its marker if it was a note on the design. A thread is a record, so nothing is deleted but the empty shell. """ if source.id == target.id: return target fresh = not any(c.deleted_at is None for c in target.comments) for c in list(source.comments): # Through the relationship: comments are delete-orphan children of # their thread, so they must be re-parented, not left behind. c.thread = target if fresh: target.resolved = source.resolved db.flush() pin = db.scalar(select(Pin).where(Pin.thread_id == source.id)) if pin is not None: db.delete(pin) db.flush() db.delete(source) db.flush() record(db, project_id, actor, "thread_moved", { "from": source.id, "to": target.id, "subject_type": target.subject_type, "subject_id": target.subject_id}) return target def comment_count(thread: Thread | None) -> int: """Live comments on a thread — what a count badge should say.""" if thread is None: return 0 return sum(1 for c in thread.comments if c.deleted_at is None) def delete_pin(db: Session, pin: Pin) -> None: """Remove a note and its marker entirely. Deleting a COMMENT is soft — a conversation is a record. Deleting the PIN is a different act: it retracts the note itself, marker and all, and leaving a marker on the design with nothing behind it is worse than either. Own notes are yours to retract; a manager may retract any, which is the same rule that already governs removing a comment. Order matters now that foreign keys are enforced: revisions and attachments hang off comments, comments off the thread, and the pin points at the thread. """ thread = db.get(Thread, pin.thread_id) comments = list(db.scalars(select(Comment).where( Comment.thread_id == pin.thread_id))) if thread else [] if comments: ids = [c.id for c in comments] for revision in db.scalars(select(CommentRevision).where( CommentRevision.comment_id.in_(ids))): db.delete(revision) for attachment in db.scalars(select(Attachment).where( Attachment.comment_id.in_(ids))): db.delete(attachment) db.flush() for comment in comments: db.delete(comment) db.flush() db.delete(pin) db.flush() if thread is not None: db.delete(thread) db.flush() def pin_count(db: Session, option_id: int) -> int: return int(db.scalar(select(func.count()).select_from(Pin) .where(Pin.option_id == option_id)) or 0) def delete_option(db: Session, option: Option) -> None: """Remove an option and everything that is OURS on it. Three things this has to get right, none of which the bare `db.delete()` did: 1. **A conversation is never deleted.** If the option carries pins, the caller refuses (see the router) — the Interaction Standard's rule, and the practical one: a mis-click must not be able to destroy a client's feedback. This function is only reached when there is none. 2. **Walkthrough steps and their approvals are ours**, and they have no cascade declared, so on Postgres the FK would block the delete outright — a 500 for the manager, not a friendly refusal. Deleted explicitly. 3. **A screen may be POINTING at this option** as its locked direction. Left alone, `selected_option_id` survives as a dangling id and the board reports a direction that no longer exists. Cleared first. """ steps = list(db.scalars(select(WalkthroughStep) .where(WalkthroughStep.option_id == option.id))) if steps: step_ids = [st.id for st in steps] for approval in db.scalars(select(StepApproval) .where(StepApproval.step_id.in_(step_ids))): db.delete(approval) for step in steps: db.delete(step) screen = db.get(Screen, option.screen_id) if screen is not None and screen.selected_option_id == option.id: screen.selected_option_id = None screen.selected_variants = None db.flush() db.delete(option) db.flush() def select_option(db: Session, actor: str, screen: Screen, option_id: int | None, variants: dict | None = None) -> None: """The direction-locking act. option_id None clears the selection.""" if option_id is None: screen.selected_option_id = None screen.selected_variants = None else: option = db.get(Option, option_id) if option is None or option.screen_id != screen.id: raise DomainError("That option is not on this screen.", "BAD_OPTION") picked = dict(variants or {}) # Unanswered axes fall back to what the client was looking at, so a # client who never touched a toggle still records a complete choice. for axis, value in default_variants(option).items(): picked.setdefault(axis, value) if not variants_complete(option, picked): raise DomainError( "That is not one of the choices offered for this option.", "BAD_VARIANT") screen.selected_option_id = option_id screen.selected_variants = picked or None db.flush() record(db, screen.instance_id, actor, "selection", {"screen_id": screen.id, "option_id": option_id, "variants": screen.selected_variants}) def send_project(db: Session, actor: str, project_id: str) -> ProjectDetails: details = ensure_details(db, project_id) details.sent_at = utcnow() # A new round: whatever the client said about the last one is answered by # the fact that we are presenting again. details.feedback_completed_at = None details.feedback_completed_by = None db.flush() record(db, project_id, actor, "sent", None) return details # ---------------------------------------------------------------- onboarding def tour_state(db: Session, project_id: str, username: str) -> dict: row = db.scalar(select(TourProgress).where( TourProgress.instance_id == project_id, TourProgress.username == username)) seen = row.seen_version if row else 0 return { "version": TOOL_TOUR_VERSION, "seen_version": seen, "completed": bool(row and row.completed), # The whole "must not be a wall" rule in one boolean: offered once per # person per board, and never again once seen or dismissed, at ANY # version (rian, 2026-09-07: a client who has walked it should not # meet the welcome again; a changed introduction is theirs to replay # from the board). `version`/`seen_version` remain the record. "should_offer": seen == 0, } def mark_tour_seen(db: Session, project_id: str, username: str, completed: bool) -> None: """Finishing and skipping both record the version — a client who said 'not now' has been asked, and asking again next visit is the wall we promised not to build. Re-running it on demand does not un-set this.""" row = db.scalar(select(TourProgress).where( TourProgress.instance_id == project_id, TourProgress.username == username)) if row is None: row = TourProgress(instance_id=project_id, username=username) db.add(row) row.seen_version = max(row.seen_version or 0, TOOL_TOUR_VERSION) row.completed = bool(row.completed or completed) db.flush() def contents(db: Session, project_id: str, username: str) -> dict: """The sidebar table of contents: named, grouped tours plus a progress read. Grouping falls out of the structure we already have — a section per screen, an entry per option, which is what the designer notes assume ("the walkthrough lives once, on the entry option; every other option carries only what changed"). No tours table is invented for it. "Done" is answered from the asker's chair: a client sees what THEY have been through, a manager sees what the CLIENT has been through — which is the question each of them actually has. """ side = side_of(username, project_id) screens = visible_screens(side, screens_of(db, project_id)) if side != "manager": screens = [s for s in screens if visible_options(side, s.options)] state = tour_state(db, project_id, username) sections: list[dict] = [{ "key": "how-to", "kind": "tool", "title": "Tool Introduction", "items": [{ "key": "tool-tour", "title": "A quick introduction to the tool", "href": None, "steps": 0, "required": 0, "approved": 0, "done": state["seen_version"] >= TOOL_TOUR_VERSION, "open_pins": 0, "chosen": False, }], }] option_ids = [o.id for s in screens for o in visible_options(side, s.options)] steps_by_option: dict[int, list[WalkthroughStep]] = {} approvals_by_step: dict[int, list[StepApproval]] = {} threads_by_step: dict[int, Thread] = {} open_by_option: dict[int, int] = {} if option_ids: for step in db.scalars(select(WalkthroughStep).where( WalkthroughStep.option_id.in_(option_ids))): steps_by_option.setdefault(step.option_id, []).append(step) step_ids = [s.id for steps in steps_by_option.values() for s in steps] if step_ids: for ap in db.scalars(select(StepApproval).where( StepApproval.step_id.in_(step_ids), StepApproval.status == "approved")): approvals_by_step.setdefault(ap.step_id, []).append(ap) for thread in db.scalars(select(Thread).where( Thread.subject_type == "step", Thread.subject_id.in_([str(i) for i in step_ids]))): threads_by_step[int(thread.subject_id)] = thread for pin, thread in db.execute( select(Pin, Thread).join(Thread, Pin.thread_id == Thread.id) .where(Pin.option_id.in_(option_ids), Thread.resolved.is_(False))).all(): open_by_option[pin.option_id] = open_by_option.get(pin.option_id, 0) + 1 def _counts_as_done(step: WalkthroughStep) -> bool: approvals = approvals_by_step.get(step.id, []) if side == "manager": return any(side_of(a.username, project_id) != "manager" for a in approvals) return any(a.username == username for a in approvals) def _beat(step: WalkthroughStep) -> dict: # One point of a walkthrough, as the outline shows it: what it is, # whether it has been approved, and whether anyone has said anything. thread = threads_by_step.get(step.id) live = comment_count(thread) return { "id": step.id, "title": step.title, "requires_approval": step.requires_approval, "approved": step.requires_approval and _counts_as_done(step), "comment_count": live, "open": thread is not None and live > 0 and not thread.resolved, } for screen in screens: items = [] for option in visible_options(side, screen.options): steps = steps_by_option.get(option.id, []) required = [s for s in steps if s.requires_approval] approved = [s for s in required if _counts_as_done(s)] items.append({ "key": f"option-{option.id}", "title": option.title, "href": f"/p/{project_id}/o/{option.id}", "steps": len(steps), "beats": [_beat(s) for s in sorted(steps, key=lambda s: (s.step_order, s.id))], "required": len(required), "approved": len(approved), # An entry with nothing to approve cannot be "done" — it is a # place to go, not a beat to complete. Saying otherwise would # make the progress read a lie. "done": bool(required) and len(approved) == len(required), "open_pins": open_by_option.get(option.id, 0), "chosen": screen.selected_option_id == option.id, }) sections.append({ "key": f"screen-{screen.id}", "kind": "screen", "title": screen.title, "items": items, }) for section in sections: completable = [i for i in section["items"] if i["required"] or section["kind"] == "tool"] section["total"] = len(completable) section["done_count"] = sum(1 for i in completable if i["done"]) section["done"] = bool(completable) and section["done_count"] == section["total"] return {"tool_tour": state, "sections": sections} # ------------------------------------------------------------------ snapshot def snapshot(db: Session, project_id: str) -> ProjectSnapshot: # Published only: the rollup answers "where does the client's decision # stand", and a draft is not in front of them to decide on. screens = [s for s in screens_of(db, project_id) if s.published] details = db.get(ProjectDetails, project_id) option_ids = [o.id for s in screens for o in s.options] open_client = open_agency = 0 if option_ids: rows = db.execute( select(Pin, Thread).join(Thread, Pin.thread_id == Thread.id) .where(Pin.option_id.in_(option_ids), Thread.resolved.is_(False)) ).all() for pin, _thread in rows: if side_of(pin.created_by, project_id) == "manager": open_agency += 1 else: # Client-side authors AND authors who since lost access both # count as client feedback — losing access must not silently # resolve their open questions. open_client += 1 # Screen-level and beat-level conversations count the same. Feedback that # does not turn the board is feedback the client left into silence — the # worst outcome there is, and the reason this is in the rollup rather than # beside it. for thread in db.scalars(select(Thread).where( Thread.project_id == project_id, Thread.subject_type.in_(("screen", "step", "option")), Thread.resolved.is_(False))): if not thread_is_live(thread): continue if side_of(thread.created_by, project_id) == "manager": open_agency += 1 else: open_client += 1 required = approved = 0 if option_ids: steps = list(db.scalars(select(WalkthroughStep).where( WalkthroughStep.option_id.in_(option_ids), WalkthroughStep.requires_approval.is_(True)))) required = len(steps) for step in steps: approvals = db.scalars(select(StepApproval).where( StepApproval.step_id == step.id, StepApproval.status == "approved")).all() if any(side_of(a.username, project_id) != "manager" for a in approvals): approved += 1 return ProjectSnapshot( screens=tuple(ScreenSnapshot( option_count=len(s.options), selected=screen_decided(db, s), ready_count=sum(1 for o in s.options if o.files)) for s in screens), sent=bool(details and details.sent_at), open_client_pins=open_client, feedback_complete=(details is not None and details.feedback_completed_at is not None), open_agency_pins=open_agency, required_steps=required, approved_required_steps=approved, ) # Which states are a given side's move. The same project is "needs you" to # different people at different moments, so this is answered per side rather # than baked into the state. _NEEDS_YOU = { "manager": {"waiting_agency", "needs_attention"}, "client": {"waiting_client"}, } def needs_you(state: str, side: str | None) -> bool: return state in _NEEDS_YOU.get(side or "", set()) def activity_index(db: Session, project_ids: list[str]) -> dict[str, datetime]: """Latest activity per project — the newest of its events and its comments. Comments are deliberately included: they are not events, and a project whose only recent news is a reply would otherwise read as idle on the index, which is exactly the project you most want to see. """ out: dict[str, datetime] = {} if not project_ids: return out for pid, when in db.execute( select(Event.instance_id, func.max(Event.occurred_at)) .where(Event.instance_id.in_(project_ids)) .group_by(Event.instance_id)).all(): if when is not None: out[pid] = when for pid, when in db.execute( select(Thread.project_id, func.max(Comment.created_at)) .join(Comment, Comment.thread_id == Thread.id) .where(Thread.project_id.in_(project_ids)) .group_by(Thread.project_id)).all(): if when is not None and (pid not in out or when > out[pid]): out[pid] = when return out def open_feedback_index(db: Session, project_ids: list[str]) -> dict[str, int]: """Open, client-authored threads per project.""" out: dict[str, int] = {} if not project_ids: return out rows = db.execute( select(Thread.project_id, Pin.created_by) .join(Pin, Pin.thread_id == Thread.id) .where(Thread.project_id.in_(project_ids), Thread.resolved.is_(False))).all() for pid, author in rows: if side_of(author, pid) != "manager": out[pid] = out.get(pid, 0) + 1 for thread in db.scalars(select(Thread).where( Thread.project_id.in_(project_ids), Thread.subject_type == "screen", Thread.resolved.is_(False))): pid = thread.project_id or "" if thread_is_live(thread) and side_of(thread.created_by, pid) != "manager": out[pid] = out.get(pid, 0) + 1 return out def waiting_on(db: Session, project_id: str) -> list[str]: """Whose move it is, by name. This fills the tool contract's `actors_waiting`, which has been an empty list since the rebuild. caddie's whole job is telling a client where their project stands, and "waiting" without "on whom" is the half that does not help. """ state = state_of(db, project_id) if state in ("waiting_agency", "needs_attention"): want = "manager" elif state == "waiting_client": want = "client" else: return [] from app import bw_accounts as bwa return sorted(m["username"] for m in bwa.members() if side_of(m["username"], project_id) == want) def manager_side(project_id: str) -> list[str]: """Everyone on the manager side of this project (the owner is a code-level super admin, not a member row, and is not listed).""" from app import bw_accounts as bwa return sorted(m["username"] for m in bwa.members() if side_of(m["username"], project_id) == "manager") def waiting_or_client_side(db: Session, project_id: str) -> list[str]: """Everyone on the client side of this project.""" from app import bw_accounts as bwa return sorted(m["username"] for m in bwa.members() if side_of(m["username"], project_id) == "client") def round_summary(db: Session, project_id: str) -> dict: """What was decided, what is outstanding, and whose move it is. The answer to a round is currently scattered across selections, approvals and two kinds of thread; reading it means opening every option in turn. This gathers it — and it is deliberately shaped like the contract's status body, so the work is not thrown away when caddie's drop-in lands. """ screens = [s for s in screens_of(db, project_id) if s.published] decided: list[dict] = [] undecided: list[dict] = [] for screen in screens: ready = [o for o in screen.options if o.files] if screen.selected_option_id is not None: option = db.get(Option, screen.selected_option_id) picked = screen.selected_variants or {} decided.append({ "screen": screen.title, "option": option.title if option else "—", "variants": [v.label for v in (option.variants if option else []) if picked.get(v.axis) == v.key], }) elif ready: undecided.append({"screen": screen.title, "options": len(ready)}) option_ids = [o.id for s in screens for o in s.options] titles = {o.id: o.title for s in screens for o in s.options} outstanding: list[dict] = [] if option_ids: rows = db.execute( select(Pin, Thread).join(Thread, Pin.thread_id == Thread.id) .where(Pin.option_id.in_(option_ids), Thread.resolved.is_(False)) .order_by(Pin.id)).all() for pin, thread in rows: first = thread.comments[0] if thread.comments else None outstanding.append({ "where": titles.get(pin.option_id, "—"), "author": pin.created_by, "excerpt": (first.body_md[:140] if first else ""), "url": f"/p/{project_id}/o/{pin.option_id}?pin={pin.id}", "kind": "note", }) for screen in screens: thread = screen_thread(db, screen) if thread is not None and not thread.resolved and thread_is_live(thread): first = thread.comments[0] if thread.comments else None outstanding.append({ "where": screen.title, "author": thread.created_by, "excerpt": (first.body_md[:140] if first else ""), "url": f"/p/{project_id}", "kind": "page", }) snap = snapshot(db, project_id) return { "state": rollup(snap), "waiting_on": waiting_on(db, project_id), "decided": decided, "undecided": undecided, "outstanding": outstanding, "approvals_required": snap.required_steps, "approvals_done": snap.approved_required_steps, } def state_of(db: Session, project_id: str) -> str: return rollup(snapshot(db, project_id))