eduard/cobra-boards
Cobra Boards, team task boards you can run from Claude or ChatGPT. FastAPI + DataK3 + appid + Dodil Models + MCP.
A remix template · 1 remixes · Remix eduard/cobra-boards
app/boards/application/use_cases.py
"""Every action a person can take. The web routes and the MCP tools both call these and nothing
else, so a rule cannot be skipped by using the other door.
People (and Claude) name things the way people do: a board by its title, a card by its number
("card 12"), a list by its name. The resolvers below turn those into records, and never guess:
two boards called "Ops" is a question back to the caller, not a coin flip.
"""
import hashlib
import re
import uuid
from datetime import date, timedelta
from typing import Optional
from boards.application.ports import Assistant, AssistantUnavailable, BoardRepository, ObjectStore
from boards.domain.model import (
MAX_ATTACHMENT,
MAX_ATTACHMENTS,
MAX_COMMENT,
MAX_TEXT,
MAX_TITLE,
SAFE_INLINE,
TEMPLATES,
Board,
BoardList,
Card,
Conflict,
Event,
Forbidden,
Invalid,
Person,
Role,
changes_between,
clean_email,
clean_filename,
clean_labels,
clean_text,
describe,
mentions_in,
now,
would_cycle,
)
class NotFound(Exception):
pass
def new_id() -> str:
return uuid.uuid4().hex[:12]
def parse_due(value, today: Optional[date] = None) -> Optional[date]:
"""ISO dates, plus the words people actually type: today, tomorrow, +3d, next week."""
if value in (None, ""):
return None
if isinstance(value, date):
return value
v = str(value).strip().lower()
today = today or date.today()
if v in ("none", "no date", "clear", "remove"):
return None
if v == "today":
return today
if v == "tomorrow":
return today + timedelta(days=1)
if v in ("next week", "in a week"):
return today + timedelta(days=7)
m = re.fullmatch(r"\+?(\d{1,3})\s*d(ays?)?", v)
if m:
return today + timedelta(days=int(m.group(1)))
try:
return date.fromisoformat(v[:10])
except ValueError:
raise Invalid(f"{value!r} is not a date; use YYYY-MM-DD, 'today', 'tomorrow' or '+3d'") from None
class Boards:
def __init__(self, repo: BoardRepository, assistant: Optional[Assistant] = None,
today=date.today, objects: Optional[ObjectStore] = None):
self.repo, self.assistant, self.today, self.objects = repo, assistant, today, objects
# ------------------------------------------------------------------ finding things
def _board(self, ref: str, person: Person) -> Board:
"""A board by id, or by title among the boards this person is on."""
ref = (ref or "").strip()
if not ref:
raise Invalid("which board? give its name or id")
board = self.repo.get(ref)
if board is not None:
try:
board.require_view(person)
except Forbidden:
raise NotFound("no such board") from None # a board you can't see doesn't exist for you
return board
mine = self.repo.boards_for(person.email)
key = ref.lower()
exact = [b for b in mine if b.title.lower() == key]
hits = exact or [b for b in mine if key in b.title.lower()]
if len(hits) == 1:
return hits[0]
if not hits:
raise NotFound(f"you have no board called {ref!r}")
names = ", ".join(f"{b.title} ({b.id})" for b in hits)
raise Invalid(f"{len(hits)} boards match {ref!r}: {names}. Say which one.")
def _card(self, ref, person: Person, board_ref: Optional[str] = None) -> tuple:
"""A card by id ('ab12cd-7'), or by number ('7', '#7') on a named board."""
raw = str(ref).strip().lstrip("#")
card = self.repo.get_card(raw) if not raw.isdigit() else None
if card is None and raw.isdigit():
number = int(raw)
if board_ref:
board = self._board(board_ref, person)
card = self.repo.card_by_number(board.id, number)
else:
found = [c for b in self.repo.boards_for(person.email)
if (c := self.repo.card_by_number(b.id, number))]
if len(found) > 1:
titles = ", ".join(f"{self.repo.get(c.board_id).title} #{c.number} "
f"'{c.title}'" for c in found)
raise Invalid(f"card {number} exists on several of your boards: {titles}. "
"Say which board.")
card = found[0] if found else None
if card is None:
raise NotFound(f"no card {ref}")
board = self.repo.get(card.board_id)
if board is None or board.role_of(person) is None:
raise NotFound(f"no card {ref}")
return board, card
def _list(self, board: Board, name_or_id: Optional[str]) -> BoardList:
if not name_or_id:
return board.lists[0]
for lst in board.lists:
if lst.id == name_or_id:
return lst
return board.list_named(name_or_id)
# ------------------------------------------------------------------ views
def _card_view(self, board: Board, c: Card, blockers: dict, open_ids: set) -> dict:
lst = next((l for l in board.lists if l.id == c.list_id), None)
blocked_by = sorted(b for b in blockers.get(c.id, ()) if b in open_ids)
return {"id": c.id, "number": c.number, "board_id": board.id, "board": board.title,
"title": c.title, "description": c.description, "list_id": c.list_id,
"list": lst.name if lst else None, "assignee": c.assignee,
"due_date": c.due_date.isoformat() if c.due_date else None,
"labels": list(c.labels), "position": c.position, "version": c.version,
"overdue": c.is_overdue(board, self.today()), "done": c.is_done(board),
"blocked_by": [int(b.rsplit("-", 1)[1]) for b in blocked_by],
"updated_at": c.updated_at.isoformat() if c.updated_at else None}
def _board_cards(self, board: Board) -> tuple:
"""The board's live cards (archived ones left), its blocker edges, and the open ids."""
gone = self.repo.archived_on(board.id)
cards = [c for c in self.repo.cards_on(board.id) if c.id not in gone]
done = board.done_list_ids()
open_ids = {c.id for c in cards if c.list_id not in done}
return cards, self.repo.blockers_on(board.id), open_ids
def my_boards(self, person: Person) -> list:
out = []
for b in self.repo.boards_for(person.email):
cards, _, open_ids = self._board_cards(b)
today = self.today()
out.append({"id": b.id, "title": b.title, "role": b.role_of(person).value,
"open_cards": len(open_ids),
"overdue": sum(1 for c in cards if c.is_overdue(b, today))})
return sorted(out, key=lambda x: x["title"].lower())
def show_board(self, person: Person, board_ref: str) -> dict:
board = self._board(board_ref, person)
cards, blockers, open_ids = self._board_cards(board)
views = [self._card_view(board, c, blockers, open_ids) for c in cards]
counts = self.repo.card_counts(board.id)
for v in views:
n = counts.get(v["id"], {})
v["comments"], v["files"] = n.get("comments", 0), n.get("files", 0)
lists = []
for lst in board.lists:
lists.append({"id": lst.id, "name": lst.name, "is_done": lst.is_done,
"cards": sorted((v for v in views if v["list_id"] == lst.id),
key=lambda v: v["position"])})
role = board.role_of(person)
return {"id": board.id, "title": board.title, "role": role.value,
"can_edit": role is not Role.VIEWER, "is_owner": role is Role.OWNER,
"owner": board.owner_email,
"members": [{"email": board.owner_email, "role": "owner"}] +
[{"email": e, "role": r.value} for e, r in sorted(board.members.items())],
"lists": lists}
# ------------------------------------------------------------------ boards & members
def first_visit(self, person: Person) -> list:
"""My boards; on a person's very first visit, first give them a private sample board.
Its id comes from their email, so two tabs opening at once still make only one."""
mine = self.my_boards(person)
if mine:
return mine
from boards.application.sample import build_welcome_board
welcome_id = "w" + hashlib.sha1(person.email.encode()).hexdigest()[:11]
if self.repo.get(welcome_id) is None:
try:
build_welcome_board(self, person, self.today(), board_id=welcome_id)
except Exception:
if self.repo.get(welcome_id) is None: # not the twin-tab race: a real failure
raise
return self.my_boards(person)
@staticmethod
def templates() -> list:
"""The board templates: the lists a new board can start with."""
return [{"name": k, "lists": [n for n, _ in v]} for k, v in TEMPLATES.items()]
def create_board(self, person: Person, title: str, board_id: Optional[str] = None,
template: Optional[str] = None) -> dict:
title = clean_text(title, "a board title", 80, required=True)
key = (template or "basic").strip().lower()
if key not in TEMPLATES:
raise Invalid(f"no template called {template!r}; there are " + ", ".join(TEMPLATES))
board = Board(id=board_id or new_id(), title=title, owner_email=person.email,
created_at=now())
board.lists = [BoardList(id=new_id(), board_id=board.id, name=n, position=float(i + 1),
is_done=d) for i, (n, d) in enumerate(TEMPLATES[key])]
self.repo.create_board(board)
self.repo.log(self._event(board.id, None, person, "board_created",
f"created the board {title!r}"))
return {"id": board.id, "title": board.title, "lists": [l.name for l in board.lists]}
def rename_board(self, person: Person, board_ref: str, title: str) -> dict:
board = self._board(board_ref, person)
board.require_owner(person)
title = clean_text(title, "a board title", 80, required=True)
old = board.title
self.repo.rename_board(board.id, title)
self.repo.log(self._event(board.id, None, person, "board_renamed",
f"renamed the board from {old!r} to {title!r}"))
return {"id": board.id, "title": title}
def invite(self, person: Person, board_ref: str, email: str, role: str = "member") -> dict:
board = self._board(board_ref, person)
try:
r = Role(str(role).strip().lower())
except ValueError:
raise Invalid("role must be 'member' or 'viewer'") from None
existing = board.members.get(clean_email(email))
email = board.set_member(person, email, r)
self.repo.save_member(board.id, email, r.value)
what = (f"changed {email} from {existing.value} to {r.value}" if existing
else f"added {email} as {r.value}")
self.repo.log(self._event(board.id, None, person, "member_set", what))
return {"board": board.title, "email": email, "role": r.value}
def remove_member(self, person: Person, board_ref: str, email: str) -> dict:
board = self._board(board_ref, person)
email = board.remove_member(person, email)
self.repo.delete_member(board.id, email)
unassigned = self.repo.unassign_all(
board.id, email, lambda card: self._event(
board.id, card.id, person, "updated", f"unassigned {email} (removed from board)",
{"assignee": [email, None]}))
self.repo.log(self._event(board.id, None, person, "member_removed",
f"removed {email} from the board"))
return {"board": board.title, "removed": email, "cards_unassigned": unassigned}
# ------------------------------------------------------------------ lists
def add_list(self, person: Person, board_ref: str, name: str, is_done: bool = False) -> dict:
board = self._board(board_ref, person)
board.require_edit(person)
name = board.check_new_list_name(name)
pos = max((l.position for l in board.lists), default=0.0) + 1
lst = BoardList(id=new_id(), board_id=board.id, name=name, position=pos,
is_done=bool(is_done))
self.repo.save_list(lst)
self.repo.log(self._event(board.id, None, person, "list_added", f"added the list {name!r}"))
return {"id": lst.id, "name": lst.name, "is_done": lst.is_done}
def update_list(self, person: Person, board_ref: str, list_ref: str,
name: Optional[str] = None, is_done: Optional[bool] = None,
position: Optional[float] = None) -> dict:
board = self._board(board_ref, person)
board.require_edit(person)
lst = self._list(board, list_ref)
parts = []
if name is not None and name.strip() != lst.name:
new = board.check_new_list_name(name, except_id=lst.id)
parts.append(f"renamed the list {lst.name!r} to {new!r}")
lst.name = new
if is_done is not None and bool(is_done) != lst.is_done:
lst.is_done = bool(is_done)
parts.append(f"marked {lst.name!r} as " + ("a done list" if lst.is_done
else "an open list"))
if position is not None:
lst.position = float(position)
parts.append(f"moved the list {lst.name!r}")
if parts:
self.repo.save_list(lst)
self.repo.log(self._event(board.id, None, person, "list_updated", "; ".join(parts)))
return {"id": lst.id, "name": lst.name, "is_done": lst.is_done, "position": lst.position}
# ------------------------------------------------------------------ cards
def _event(self, board_id, card_id, person, action, summary, changes=None) -> Event:
return Event(id=new_id() + new_id(), board_id=board_id, card_id=card_id,
actor=person.email, action=action, summary=summary,
changes=changes or {}, at=now())
def _check_assignee(self, board: Board, assignee: Optional[str],
person: Optional[Person] = None) -> Optional[str]:
if assignee in (None, "", "nobody", "none"):
return None
if person and str(assignee).strip().lower() in ("me", "myself", "i", "self"):
assignee = person.email # what people (and their agents) actually say
email = clean_email(assignee)
if email not in board.people():
raise Invalid(f"{email} is not on this board; the owner can invite them first")
return email
def add_card(self, person: Person, board_ref: str, title: str, list_name: str = "",
description: str = "", assignee: Optional[str] = None, due_date=None,
labels=None) -> dict:
board = self._board(board_ref, person)
board.require_edit(person)
title = clean_text(title, "a card title", MAX_TITLE, required=True)
lst = self._list(board, list_name)
cards = self.repo.cards_on(board.id)
pos = max((c.position for c in cards if c.list_id == lst.id), default=0.0) + 1
card = Card(id="", board_id=board.id, number=0, list_id=lst.id, title=title,
description=clean_text(description, "a description", MAX_TEXT),
assignee=self._check_assignee(board, assignee, person),
due_date=parse_due(due_date, self.today()),
labels=clean_labels(labels), position=pos, created_by=person.email,
created_at=now(), updated_at=now())
for _ in range(8): # the number is claimed by a plain INSERT: a clash just tries the next
card.number = self.repo.next_number(board.id)
card.id = f"{board.id}-{card.number}"
event = self._event(board.id, card.id, person, "created",
f"added #{card.number} {title!r} to {lst.name}",
changes_between({}, card.snapshot(board)))
if self.repo.insert_card(card, event):
break
else:
raise Conflict("too many cards were added at once; try again")
if card.assignee and card.assignee != person.email:
self._notify(board, card, person, "assigned", [card.assignee],
f"assigned you #{card.number} {card.title!r}")
cards, blockers, open_ids = self._board_cards(board)
return self._card_view(board, card, blockers, open_ids | {card.id})
def update_card(self, person: Person, card_ref, board_ref: Optional[str] = None,
expected_version: Optional[int] = None, title: Optional[str] = None,
description: Optional[str] = None, assignee=..., due_date=...,
labels=None, list_name: Optional[str] = None,
position: Optional[float] = None) -> dict:
"""Change any fields of a card. Pass expected_version (from when you read it) and a card
someone else changed in between is refused instead of silently overwritten."""
board, card = self._card(card_ref, person, board_ref)
board.require_edit(person)
if card.id in self.repo.archived_on(board.id):
raise Invalid(f"card #{card.number} is archived; restore it first")
if expected_version is not None and int(expected_version) != card.version:
raise Conflict(f"card #{card.number} changed since you opened it "
f"(version {card.version}, you had {expected_version}); reload it")
before = card.snapshot(board)
if title is not None:
card.title = clean_text(title, "a card title", MAX_TITLE, required=True)
if description is not None:
card.description = clean_text(description, "a description", MAX_TEXT)
if assignee is not ...:
card.assignee = self._check_assignee(board, assignee, person)
if due_date is not ...:
card.due_date = parse_due(due_date, self.today())
if labels is not None:
card.labels = clean_labels(labels)
if list_name:
target = self._list(board, list_name)
if target.id != card.list_id and position is None:
cards = self.repo.cards_on(board.id)
position = max((c.position for c in cards if c.list_id == target.id),
default=0.0) + 1
card.list_id = target.id
if position is not None:
card.position = float(position)
changes = changes_between(before, card.snapshot(board))
moved_only = position is not None and not changes
if not changes and not moved_only:
cards, blockers, open_ids = self._board_cards(board)
return self._card_view(board, card, blockers, open_ids)
action = "moved" if set(changes) == {"list"} or moved_only else "updated"
summary = describe(changes) if changes else "reordered the card"
old_version = card.version
card.version += 1
card.updated_at = now()
event = self._event(board.id, card.id, person, action, f"#{card.number}: {summary}",
changes)
if not self.repo.update_card(card, old_version, event):
raise Conflict(f"card #{card.number} was changed by someone else just now; reload it")
new_owner = changes.get("assignee", [None, None])[1]
if new_owner and new_owner != person.email:
self._notify(board, card, person, "assigned", [new_owner],
f"assigned you #{card.number} {card.title!r}")
cards, blockers, open_ids = self._board_cards(board)
return self._card_view(board, card, blockers, open_ids)
def move_card(self, person: Person, card_ref, list_name: str, board_ref: Optional[str] = None,
before_card=None, expected_version: Optional[int] = None) -> dict:
"""Move a card to a list; optionally right above another card (for drag and drop)."""
board, card = self._card(card_ref, person, board_ref)
target = self._list(board, list_name)
position = None
if before_card not in (None, ""):
_, anchor = self._card(before_card, person, board.id)
siblings = sorted((c for c in self.repo.cards_on(board.id)
if c.list_id == target.id and c.id != card.id),
key=lambda c: c.position)
idx = next((i for i, c in enumerate(siblings) if c.id == anchor.id), None)
if idx is None:
raise Invalid(f"card #{anchor.number} is not in {target.name}")
prev = siblings[idx - 1].position if idx > 0 else anchor.position - 1
position = (prev + anchor.position) / 2
return self.update_card(person, card.id, expected_version=expected_version,
list_name=target.id, position=position)
def card_detail(self, person: Person, card_ref, board_ref: Optional[str] = None) -> dict:
board, card = self._card(card_ref, person, board_ref)
cards, blockers, open_ids = self._board_cards(board)
view = self._card_view(board, card, blockers, open_ids)
by_id = {c.id: c for c in cards}
view["blocks"] = sorted(by_id[cid].number for cid, bs in blockers.items()
if card.id in bs and cid in by_id)
view["blocked_by_all"] = [{"number": by_id[b].number, "title": by_id[b].title,
"done": b not in open_ids}
for b in sorted(blockers.get(card.id, ())) if b in by_id]
view["checklist"] = self.repo.checklist(card.id)
view["comments"] = [self._comment_view(x) for x in self.repo.comments(card.id)]
view["attachments"] = [self._att_view(a) for a in self.repo.attachments(card.id)]
arch = self.repo.archived_on(board.id).get(card.id)
view["archived"] = bool(arch)
view["archived_by"] = arch["by"] if arch else None
view["history"] = [self._event_view(e) for e in self.repo.history(card.id)]
view["people"] = board.people()
view["can_edit"] = board.role_of(person) is not Role.VIEWER
return view
@staticmethod
def _event_view(e: Event) -> dict:
return {"actor": e.actor, "action": e.action, "summary": e.summary,
"changes": e.changes, "at": e.at.isoformat() if e.at else None}
def card_history(self, person: Person, card_ref, board_ref: Optional[str] = None) -> list:
_, card = self._card(card_ref, person, board_ref)
return [self._event_view(e) for e in self.repo.history(card.id)]
def board_activity(self, person: Person, board_ref: str, limit: int = 30) -> list:
board = self._board(board_ref, person)
return [self._event_view(e) for e in self.repo.board_activity(board.id, limit)]
# ------------------------------------------------------------------ my work
def _visible(self, person: Person) -> list:
return self.repo.boards_for(person.email)
def my_cards(self, person: Person, days: Optional[int] = 7) -> dict:
"""What's on my plate: my open cards that are overdue or due within `days`, then the
rest of my open cards without a date. Across every board I'm on."""
today = self.today()
horizon = today + timedelta(days=int(days or 7))
boards = {b.id: b for b in self._visible(person)}
gone = {}
overdue, this_week, later, undated = [], [], [], []
for c in self.repo.cards_assigned_to(person.email):
b = boards.get(c.board_id)
if b is None or c.is_done(b):
continue
if b.id not in gone:
gone[b.id] = self.repo.archived_on(b.id)
if c.id in gone[b.id]:
continue
_, blockers, open_ids = self._board_cards(b)
v = self._card_view(b, c, blockers, open_ids)
if c.due_date is None:
undated.append(v)
elif c.due_date < today:
overdue.append(v)
elif c.due_date <= horizon:
this_week.append(v)
else:
later.append(v)
key = lambda v: (v["due_date"] or "9999", v["board"], v["number"])
return {"today": today.isoformat(), "overdue": sorted(overdue, key=key),
"due_soon": sorted(this_week, key=key), "no_due_date": sorted(undated, key=key),
"later": sorted(later, key=key)}
def overdue(self, person: Person, board_ref: Optional[str] = None) -> list:
boards = [self._board(board_ref, person)] if board_ref else self._visible(person)
today = self.today()
out = []
for b in boards:
cards, blockers, open_ids = self._board_cards(b)
out += [self._card_view(b, c, blockers, open_ids) for c in cards
if c.is_overdue(b, today)]
return sorted(out, key=lambda v: (v["due_date"], v["board"], v["number"]))
def search_cards(self, person: Person, text: str, board_ref: Optional[str] = None) -> list:
"""Plain text search over titles, descriptions and labels."""
key = (text or "").strip().lower()
if not key:
raise Invalid("search for what?")
boards = [self._board(board_ref, person)] if board_ref else self._visible(person)
out = []
for b in boards:
cards, blockers, open_ids = self._board_cards(b)
for c in cards:
hay = " ".join([c.title, c.description, " ".join(c.labels)]).lower()
if key in hay or key.lstrip("#") == str(c.number):
out.append(self._card_view(b, c, blockers, open_ids))
return out[:50]
# ------------------------------------------------------------------ dependencies
def add_blocker(self, person: Person, card_ref, blocker_ref,
board_ref: Optional[str] = None) -> dict:
"""card_ref is blocked by blocker_ref (both on the same board)."""
board, card = self._card(card_ref, person, board_ref)
board.require_edit(person)
_, blocker = self._card(blocker_ref, person, board.id)
if blocker.board_id != board.id:
raise Invalid("a card can only be blocked by a card on the same board")
edges = self.repo.blockers_on(board.id)
if blocker.id in edges.get(card.id, set()):
return {"card": card.number, "blocked_by": blocker.number, "already": True}
if would_cycle(card.id, blocker.id, edges):
raise Invalid(f"#{blocker.number} already waits on #{card.number}; "
"that would make them wait for each other forever")
self.repo.add_dependency(board.id, card.id, blocker.id)
self.repo.log(self._event(board.id, card.id, person, "blocked",
f"#{card.number} is now blocked by #{blocker.number}",
{"blocked_by": [None, blocker.number]}))
return {"card": card.number, "blocked_by": blocker.number}
def remove_blocker(self, person: Person, card_ref, blocker_ref,
board_ref: Optional[str] = None) -> dict:
board, card = self._card(card_ref, person, board_ref)
board.require_edit(person)
_, blocker = self._card(blocker_ref, person, board.id)
self.repo.remove_dependency(card.id, blocker.id)
self.repo.log(self._event(board.id, card.id, person, "unblocked",
f"#{card.number} no longer waits on #{blocker.number}",
{"blocked_by": [blocker.number, None]}))
return {"card": card.number, "unblocked_from": blocker.number}
def what_is_blocking(self, person: Person, card_ref, board_ref: Optional[str] = None) -> dict:
"""Every unfinished card standing in the way, followed down the chain, and where to
start: the blockers that are not waiting on anything themselves."""
board, card = self._card(card_ref, person, board_ref)
cards, edges, open_ids = self._board_cards(board)
by_id = {c.id: c for c in cards}
chain, seen = [], set()
def walk(cid, depth):
for b in sorted(edges.get(cid, ())):
if b in seen or b not in open_ids or b not in by_id:
continue
seen.add(b)
c = by_id[b]
chain.append({"number": c.number, "title": c.title, "depth": depth,
"assignee": c.assignee,
"list": board.list_by_id(c.list_id).name,
"waits_on": sorted(by_id[x].number for x in edges.get(b, ())
if x in open_ids and x in by_id)})
walk(b, depth + 1)
walk(card.id, 1)
start = [c for c in chain if not c["waits_on"]]
return {"card": card.number, "title": card.title, "blocked": bool(chain),
"chain": chain, "start_with": start}
# ------------------------------------------------------------------ checklists
def suggest_checklist(self, person: Person, card_ref, board_ref: Optional[str] = None) -> dict:
"""An AI suggestion only: it changes nothing. A person decides what to keep."""
_, card = self._card(card_ref, person, board_ref)
if self.assistant is None:
raise Invalid("no AI assistant configured")
try:
raw = self.assistant.checklist(card.title, card.description)
except AssistantUnavailable:
raise Invalid("Dodil Models didn't answer just now; "
"try the suggestion again in a minute") from None
steps = [clean_text(s, "a step", 200) for s in raw]
return {"card": card.number, "title": card.title,
"suggested_steps": [s for s in steps if s][:8],
"note": "Nothing was saved. Add the steps you want with add_checklist_items."}
def add_checklist_items(self, person: Person, card_ref, items: list,
board_ref: Optional[str] = None) -> dict:
board, card = self._card(card_ref, person, board_ref)
board.require_edit(person)
texts = [clean_text(t, "a step", 200, required=True) for t in (items or [])]
if not texts:
raise Invalid("add which steps?")
existing = self.repo.checklist(card.id)
if len(existing) + len(texts) > 40:
raise Invalid("a checklist takes at most 40 steps")
pos = max((i["position"] for i in existing), default=0.0)
for t in texts:
pos += 1
self.repo.save_checklist_item(card.id, new_id(), t, False, pos)
self.repo.log(self._event(board.id, card.id, person, "checklist",
f"#{card.number}: added {len(texts)} checklist step"
+ ("s" if len(texts) > 1 else "")))
return {"card": card.number, "checklist": self.repo.checklist(card.id)}
def set_checklist_item(self, person: Person, card_ref, item_id: str,
done: Optional[bool] = None, text: Optional[str] = None,
remove: bool = False, board_ref: Optional[str] = None) -> dict:
board, card = self._card(card_ref, person, board_ref)
board.require_edit(person)
items = {i["id"]: i for i in self.repo.checklist(card.id)}
item = items.get(item_id)
if item is None:
raise NotFound("no such checklist step")
if remove:
self.repo.delete_checklist_item(card.id, item_id)
summary = f"removed the step {item['text']!r}"
else:
new_text = clean_text(text, "a step", 200, required=True) if text else item["text"]
new_done = item["done"] if done is None else bool(done)
self.repo.save_checklist_item(card.id, item_id, new_text, new_done, item["position"])
summary = (f"checked {new_text!r}" if new_done and not item["done"] else
f"unchecked {new_text!r}" if item["done"] and not new_done else
f"edited the step {new_text!r}")
self.repo.log(self._event(board.id, card.id, person, "checklist",
f"#{card.number}: {summary}"))
return {"card": card.number, "checklist": self.repo.checklist(card.id)}
# ------------------------------------------------------------------ AI over the board
@staticmethod
def _text_of(c: Card) -> str:
return f"{c.title}\n{c.description}\nlabels: {', '.join(c.labels)}".strip()
def _ensure_embeddings(self, boards: list) -> None:
"""Embeds cards whose text changed since last time (capped, so a big board stays fast)."""
done = 0
for b in boards:
for c in self.repo.cards_on(b.id):
text = self._text_of(c)
h = hashlib.sha1(text.encode()).hexdigest()
if self.repo.embedding_hash(c.id) == h:
continue
self.repo.save_embedding(c.id, b.id, self.assistant.embed(text), h)
done += 1
if done >= 100:
return
def similar_cards(self, person: Person, card_ref=None, text: Optional[str] = None,
board_ref: Optional[str] = None, limit: int = 5) -> dict:
"""Cards that mean the same thing, even with different words. Only boards you're on."""
if self.assistant is None:
raise Invalid("no AI assistant configured")
boards = [self._board(board_ref, person)] if board_ref else self._visible(person)
source = None
if card_ref not in (None, ""):
b, source = self._card(card_ref, person, board_ref)
if all(x.id != b.id for x in boards):
boards.append(b)
text = self._text_of(source)
if not (text or "").strip():
raise Invalid("similar to which card, or to what text?")
try:
self._ensure_embeddings(boards)
query = self.assistant.embed(text)
except AssistantUnavailable:
raise Invalid("meaning search is unavailable for a moment (Dodil Models didn't "
"answer); plain search still works") from None
hits = self.repo.similar_cards([b.id for b in boards], query, int(limit) + 1)
by_board = {b.id: b for b in boards}
gone = set().union(*(self.repo.archived_on(b.id) for b in boards)) if boards else set()
out = []
for card_id, score in hits:
if (source is not None and card_id == source.id) or card_id in gone:
continue
c = self.repo.get_card(card_id)
b = by_board.get(c.board_id) if c else None
if c and b and b.role_of(person):
out.append({"number": c.number, "board": b.title, "title": c.title,
"list": b.list_by_id(c.list_id).name,
"similarity": round(1 - float(score), 3)})
return {"similar_to": source.title if source else text, "cards": out[:int(limit)]}
def daily_summary(self, person: Person, board_ref: str) -> dict:
"""A short morning briefing. The numbers are counted here; the AI only puts them into
words, so it can't invent a card that isn't on the board."""
board = self._board(board_ref, person)
cards, edges, open_ids = self._board_cards(board)
today = self.today()
since = today - timedelta(days=1)
recent = [e for e in self.repo.board_activity(board.id, 200)
if e.at and e.at.date() >= since]
overdue = [c for c in cards if c.is_overdue(board, today)]
due_today = [c for c in cards if c.due_date == today and c.id in open_ids]
blocked = [c for c in cards if c.id in open_ids
and any(b in open_ids for b in edges.get(c.id, ()))]
per_list = {l.name: sum(1 for c in cards if c.list_id == l.id) for l in board.lists}
finished = [e for e in recent if e.action == "moved" and
any(e.changes.get("list", [None, None])[1] == l.name
for l in board.lists if l.is_done)]
fact = lambda c: f"#{c.number} {c.title}" + (f" ({c.assignee})" if c.assignee else "")
facts = "\n".join([
f"Date: {today.isoformat()}",
"Cards per list: " + ", ".join(f"{k} {v}" for k, v in per_list.items()),
f"Finished in the last day: {len(finished)}",
"Overdue: " + ("; ".join(fact(c) for c in overdue) or "none"),
"Due today: " + ("; ".join(fact(c) for c in due_today) or "none"),
"Blocked: " + ("; ".join(fact(c) for c in blocked) or "none"),
"Changes in the last day: " + ("; ".join(f"{e.actor}: {e.summary}"
for e in recent[:15]) or "none")])
text = None
if self.assistant is not None:
try:
text = self.assistant.summary(board.title, facts)
except Exception: # the briefing still works without the AI
text = None
return {"board": board.title, "date": today.isoformat(), "summary": text or facts,
"facts": {"per_list": per_list, "overdue": [c.number for c in overdue],
"due_today": [c.number for c in due_today],
"blocked": [c.number for c in blocked],
"finished_last_day": len(finished)}}
# ------------------------------------------------------------------ live
def board_stamp(self, person: Person, board_ref: str) -> dict:
"""What the open page polls: the newest history line on the board. When it changes,
someone changed the board, and the page reloads it (and says who did what)."""
board = self._board(board_ref, person)
e = self.repo.latest_event(board.id)
if e is None:
return {"board": board.id, "stamp": "", "by": None, "summary": None, "at": None}
return {"board": board.id, "stamp": e.id, "by": e.actor, "summary": e.summary,
"card_id": e.card_id, "at": e.at.isoformat() if e.at else None}
# ------------------------------------------------------------------ comments & notifications
@staticmethod
def _comment_view(x: dict) -> dict:
at = x.get("created_at")
return {"id": x["id"], "author": x["author"], "body": x["body"],
"mentions": list(x.get("mentions") or []),
"at": at.isoformat() if hasattr(at, "isoformat") else at}
def _notify(self, board: Board, card: Optional[Card], actor: Person, kind: str,
emails: list, summary: str) -> int:
"""One notification per person, never to the one who did it, only to people on the board."""
people = set(board.people())
items = [{"id": new_id() + new_id(), "email": e, "board_id": board.id,
"card_id": card.id if card else None, "kind": kind, "actor": actor.email,
"summary": summary, "created_at": now()}
for e in dict.fromkeys(emails) if e != actor.email and e in people]
self.repo.add_notifications(items)
return len(items)
def add_comment(self, person: Person, card_ref, body: str,
board_ref: Optional[str] = None) -> dict:
"""A comment on a card. @name mentions a teammate and notifies them; the card's assignee
hears about it too. Viewers read comments but can't write them."""
board, card = self._card(card_ref, person, board_ref)
board.require_edit(person)
body = clean_text(body, "a comment", MAX_COMMENT, required=True)
mentioned = mentions_in(body, board.people())
comment = {"id": new_id() + new_id(), "board_id": board.id, "card_id": card.id,
"author": person.email, "body": body, "mentions": mentioned,
"created_at": now()}
short = body if len(body) <= 80 else body[:77] + "…"
event = self._event(board.id, card.id, person, "commented",
f"#{card.number}: commented {short!r}")
self.repo.add_comment(comment, event)
told = self._notify(board, card, person, "mention", mentioned,
f"mentioned you on #{card.number} {card.title!r}: {short}")
if card.assignee and card.assignee not in mentioned:
told += self._notify(board, card, person, "comment", [card.assignee],
f"commented on your card #{card.number} {card.title!r}: {short}")
return {"card": card.number, "comment": self._comment_view(comment),
"mentioned": mentioned, "notified": told}
def comments(self, person: Person, card_ref, board_ref: Optional[str] = None) -> list:
_, card = self._card(card_ref, person, board_ref)
return [self._comment_view(x) for x in self.repo.comments(card.id)]
def my_notifications(self, person: Person, unread_only: bool = False) -> dict:
"""What changed for me: mentions, cards given to me, comments on my cards. Only from
boards I'm still on."""
mine = {b.id: b for b in self._visible(person)}
items = []
for n in self.repo.notifications_for(person.email, 100):
b = mine.get(n["board_id"])
if b is None or (unread_only and n.get("read_at")):
continue
num = int(n["card_id"].rsplit("-", 1)[1]) if n.get("card_id") else None
at = n.get("created_at")
items.append({"id": n["id"], "kind": n["kind"], "actor": n["actor"],
"summary": n["summary"], "board": b.title, "board_id": b.id,
"card_id": n.get("card_id"), "card": num,
"read": bool(n.get("read_at")),
"at": at.isoformat() if hasattr(at, "isoformat") else at})
return {"unread": sum(1 for i in items if not i["read"]), "items": items[:50]}
def mark_notifications_read(self, person: Person, ids: Optional[list] = None) -> dict:
return {"marked": self.repo.mark_read(person.email, ids or None, now())}
# ------------------------------------------------------------------ attachments
@staticmethod
def _att_view(a: dict) -> dict:
at = a.get("created_at")
return {"id": a["id"], "name": a["name"], "size": int(a.get("size") or 0),
"content_type": a.get("content_type") or "application/octet-stream",
"uploaded_by": a.get("uploaded_by"), "url": f"/api/attachments/{a['id']}",
"at": at.isoformat() if hasattr(at, "isoformat") else at}
def add_attachment(self, person: Person, card_ref, name: str, data: bytes,
content_type: str = "", board_ref: Optional[str] = None) -> dict:
"""A file on a card. The bytes go to the bucket's object storage; the card keeps a record
and a history line."""
if self.objects is None:
raise Invalid("attachments are not configured")
board, card = self._card(card_ref, person, board_ref)
board.require_edit(person)
name = clean_filename(name)
if not data:
raise Invalid("the file is empty")
if len(data) > MAX_ATTACHMENT:
raise Invalid(f"a file can be at most {MAX_ATTACHMENT // (1024 * 1024)} MB")
if len(self.repo.attachments(card.id)) >= MAX_ATTACHMENTS:
raise Invalid(f"a card takes at most {MAX_ATTACHMENTS} files")
att_id = new_id() + new_id()
att = {"id": att_id, "board_id": board.id, "card_id": card.id, "name": name,
"content_type": (content_type or "application/octet-stream")[:100],
"size": len(data), "object_key": f"attachments/{board.id}/{card.id}/{att_id}",
"uploaded_by": person.email, "created_at": now()}
self.objects.put(att["object_key"], data, att["content_type"])
self.repo.add_attachment(att, self._event(board.id, card.id, person, "attached",
f"#{card.number}: attached {name!r}"))
return {"card": card.number, "attachment": self._att_view(att)}
def _attachment(self, person: Person, att_id: str) -> tuple:
att = self.repo.get_attachment(att_id)
if att is None:
raise NotFound("no such file")
board = self.repo.get(att["board_id"])
if board is None or board.role_of(person) is None:
raise NotFound("no such file") # a file on a board you can't see doesn't exist
return board, att
def download_attachment(self, person: Person, att_id: str) -> dict:
board, att = self._attachment(person, att_id)
ctype = att.get("content_type") or "application/octet-stream"
return {"name": att["name"], "data": self.objects.get(att["object_key"]),
"content_type": ctype if ctype in SAFE_INLINE else "application/octet-stream",
"inline": ctype in SAFE_INLINE}
def delete_attachment(self, person: Person, att_id: str) -> dict:
board, att = self._attachment(person, att_id)
board.require_edit(person)
card = self.repo.get_card(att["card_id"])
number = card.number if card else "?"
self.repo.delete_attachment(att_id, self._event(
board.id, att["card_id"], person, "detached",
f"#{number}: removed the file {att['name']!r}"))
self.objects.delete(att["object_key"])
return {"removed": att["name"]}
# ------------------------------------------------------------------ archive
def archive_card(self, person: Person, card_ref, board_ref: Optional[str] = None) -> dict:
"""Take a card off the board without losing it: its number, history and comments stay,
and it can be restored. There is no delete, by design."""
board, card = self._card(card_ref, person, board_ref)
board.require_edit(person)
if card.id in self.repo.archived_on(board.id):
return {"card": card.number, "title": card.title, "archived": True, "already": True}
self.repo.archive_card(card.id, board.id, person.email, now(), self._event(
board.id, card.id, person, "archived", f"#{card.number}: archived the card",
{"archived": [False, True]}))
return {"card": card.number, "title": card.title, "archived": True}
def restore_card(self, person: Person, card_ref, board_ref: Optional[str] = None) -> dict:
board, card = self._card(card_ref, person, board_ref)
board.require_edit(person)
if card.id not in self.repo.archived_on(board.id):
raise Invalid(f"card #{card.number} is not archived")
self.repo.restore_card(card.id, self._event(
board.id, card.id, person, "restored", f"#{card.number}: restored the card",
{"archived": [True, False]}))
return {"card": card.number, "title": card.title, "archived": False,
"list": board.list_by_id(card.list_id).name}
def archived_cards(self, person: Person, board_ref: str) -> list:
board = self._board(board_ref, person)
out = []
for card_id, info in self.repo.archived_on(board.id).items():
c = self.repo.get_card(card_id)
if c:
at = info.get("at")
out.append({"id": c.id, "number": c.number, "title": c.title,
"archived_by": info.get("by"),
"at": at.isoformat() if hasattr(at, "isoformat") else at})
return sorted(out, key=lambda x: x["number"])