theaio/theaio-boards
Laneway, team task boards you can run from Claude or ChatGPT. FastAPI + DataK3 + appid + Models + MCP.
A remix template · 0 remixes · Remix theaio/theaio-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.
Visibility lives in two methods only: _visible() (the boards a person is on) and _board() (one
board, by id or title, that the person is on). Every read and write starts from one of them, so a
board you are not on does not exist for you: NotFound, never a hint that it is there.
Add a use case here, then expose it in doors/web.py and doors/mcp.py (one line each).
"""
import hashlib
import re
import uuid
from datetime import datetime, timedelta, timezone
from typing import Callable, Optional
from boards.application.ports import Assistant, BoardRepository
from boards.domain.model import (DEFAULT_LISTS, MAX_TEXT, Board, Card, Change, HistoryEntry,
Invalid, Person, Role, blocking_path, clean_title, parse_due)
CLAIM_ATTEMPTS = 8 # card numbers: re-read max and retry when someone took ours
EMBED_BATCH = 64
NUMBER = re.compile(r"^(?:card\s*)?(?:no\.?\s*)?#?\s*(\d+)$", re.IGNORECASE)
class NotFound(Exception):
pass
def _new_id() -> str:
return uuid.uuid4().hex[:12]
def _utcnow() -> datetime:
return datetime.now(timezone.utc)
def _role(text: str) -> Role:
try:
return Role((text or "").strip().lower())
except ValueError:
raise Invalid(f"unknown role {text!r}: use 'member' or 'viewer'")
def _meaning(card: Card) -> str:
"""The text a card's embedding is made from."""
parts = [card.title, card.description, " ".join(card.labels)]
return "\n".join(p for p in parts if p)
def _digest(card: Card) -> str:
return hashlib.sha256(_meaning(card).encode()).hexdigest()[:16]
class Boards:
def __init__(self, repo: BoardRepository, assistant: Optional[Assistant] = None,
clock: Callable[[], datetime] = _utcnow):
self.repo, self.assistant, self.clock = repo, assistant, clock
def _today(self):
return self.clock().date()
# --- visibility: the one place ---------------------------------------------------------
def _visible(self, person: Person) -> list:
"""The boards this person is on. Lists, searches and AI features never look further."""
return [b for b in self.repo.boards_for(person.email) if b.role_of(person) is not None]
def _board(self, person: Person, ref: str) -> Board:
"""A board by id or title, among the person's boards only."""
ref = str(ref if ref is not None else "").strip()
if not ref:
raise Invalid("which board? give its title or id (my_boards lists them)")
board = self.repo.get(ref)
if board is not None and board.role_of(person) is not None:
return board
hits = [b for b in self._visible(person) if b.title.lower() == ref.lower()]
if len(hits) == 1:
return hits[0]
if hits:
raise Invalid(f"{len(hits)} of your boards are called {ref!r}; use an id: "
+ ", ".join(b.id for b in hits))
raise NotFound(f"no board {ref!r} among your boards")
def _card(self, person: Person, ref, board_ref: Optional[str] = None) -> tuple:
"""A card by id, number (12, #12, card 12) or title, on a board the person is on.
Without board_ref a number or title is looked up on all their boards; if it matches
cards on several boards, we ask which board instead of guessing."""
ref = str(ref if ref is not None else "").strip()
if not ref:
raise Invalid("which card? give its number (12), its id or its title")
scope = [self._board(person, board_ref)] if board_ref else None
card = self.repo.get_card(ref)
if card is not None:
board = self.repo.get(card.board_id)
if board is not None and board.role_of(person) is not None and \
(scope is None or scope[0].id == board.id):
return board, card
boards = scope if scope is not None else self._visible(person)
m = NUMBER.match(ref)
if m:
n = int(m.group(1))
hits = [(b, c) for b in boards for c in [self.repo.card_by_number(b.id, n)] if c]
else:
key = ref.lower()
every = [(b, c) for b in boards for c in self.repo.cards_on(b.id)]
hits = ([(b, c) for b, c in every if c.title.lower() == key]
or [(b, c) for b, c in every if key in c.title.lower()])
if len(hits) == 1:
return hits[0]
if not hits:
where = f"on {boards[0].title}" if scope else "on your boards"
raise NotFound(f"no card {ref!r} {where}")
listed = "; ".join(f"card {c.number} {c.title!r} on {b.title}" for b, c in hits[:6])
raise Invalid(f"{ref!r} matches several cards ({listed}); say which board, or use the id")
# --- how cards look from the outside ----------------------------------------------------
def _views(self, board: Board) -> list:
"""Every card on the board as the doors show it, with overdue and blocked worked out."""
today, done = self._today(), board.done_list.id
names = {l.id: l.name for l in board.lists}
cards = self.repo.cards_on(board.id)
by_id = {c.id: c for c in cards}
waiting = {c for c, b in self.repo.links_on(board.id)
if b in by_id and by_id[b].list_id != done}
order = {l.id: i for i, l in enumerate(board.ordered())}
cards.sort(key=lambda c: (order.get(c.list_id, 99), c.pos, c.number))
return [{"id": c.id, "number": c.number, "board_id": board.id, "board": board.title,
"list": names.get(c.list_id, ""), "list_id": c.list_id, "title": c.title,
"description": c.description, "assignee": c.assignee,
"due_date": c.due_date.isoformat() if c.due_date else None,
"labels": list(c.labels), "position": c.pos + 1, "done": c.list_id == done,
"overdue": c.is_overdue(today, done), "blocked": c.id in waiting}
for c in cards]
def _view(self, board: Board, card: Card) -> dict:
return next(v for v in self._views(board) if v["id"] == card.id)
def _record(self, board: Board, card: Card, person: Person, changes: list) -> None:
at = self.clock()
self.repo.record([HistoryEntry(uuid.uuid4().hex, board.id, card.id, person.email, at,
ch.field, ch.old, ch.new) for ch in changes])
def _ai(self) -> Assistant:
if self.assistant is None:
raise Invalid("no AI assistant is configured for this app")
return self.assistant
# --- boards and lists -------------------------------------------------------------------
def my_boards(self, person: Person) -> list:
return [{"id": b.id, "title": b.title, "role": b.role_of(person).value}
for b in sorted(self._visible(person), key=lambda b: b.title.lower())]
def create_board(self, person: Person, title: str) -> dict:
board = Board.new(_new_id(), title, person, [_new_id() for _ in DEFAULT_LISTS])
self.repo.save_board(board)
self.repo.log(board.id, None, person.email, f"created board {board.title!r}")
return {"id": board.id, "title": board.title, "role": Role.OWNER.value,
"lists": [l.name for l in board.ordered()]}
def show_board(self, person: Person, board_ref: str) -> dict:
board = self._board(person, board_ref)
views = self._views(board)
return {"id": board.id, "title": board.title, "role": board.role_of(person).value,
"today": self._today().isoformat(),
"members": [{"email": e, "role": r.value} for e, r in board.people()],
"lists": [{"id": l.id, "name": l.name, "position": i + 1,
"cards": [v for v in views if v["list_id"] == l.id]}
for i, l in enumerate(board.ordered())],
"overdue": sum(v["overdue"] for v in views)}
def _lists(self, board: Board) -> dict:
return {"board": board.title,
"lists": [{"id": l.id, "name": l.name, "position": i + 1}
for i, l in enumerate(board.ordered())]}
def add_list(self, person: Person, board_ref: str, name: str) -> dict:
board = self._board(person, board_ref)
lst = board.add_list(person, _new_id(), name)
self.repo.save_board(board)
self.repo.log(board.id, None, person.email, f"added list {lst.name!r}")
return self._lists(board)
def rename_list(self, person: Person, board_ref: str, list_ref: str, name: str) -> dict:
board = self._board(person, board_ref)
lst = board.find_list(list_ref)
old = board.rename_list(person, lst, name)
self.repo.save_board(board)
self.repo.log(board.id, None, person.email, f"renamed list {old!r} to {lst.name!r}")
return self._lists(board)
def move_list(self, person: Person, board_ref: str, list_ref: str, position: int) -> dict:
board = self._board(person, board_ref)
lst = board.find_list(list_ref)
board.move_list(person, lst, int(position) - 1)
self.repo.save_board(board)
self.repo.log(board.id, None, person.email, f"moved list {lst.name!r} to {position}")
return self._lists(board)
# --- people -----------------------------------------------------------------------------
def _members(self, board: Board) -> dict:
return {"board": board.title,
"members": [{"email": e, "role": r.value} for e, r in board.people()]}
def invite_member(self, person: Person, board_ref: str, email: str,
role: str = "member") -> dict:
board = self._board(person, board_ref)
email = board.invite(person, email, _role(role))
self.repo.save_board(board)
self.repo.log(board.id, None, person.email, f"invited {email} as {role}")
return self._members(board)
def set_member_role(self, person: Person, board_ref: str, email: str, role: str) -> dict:
board = self._board(person, board_ref)
email = board.set_role(person, email, _role(role))
self.repo.save_board(board)
if board.members[email] is Role.VIEWER:
self._unassign(person, board, email)
self.repo.log(board.id, None, person.email, f"made {email} a {role}")
return self._members(board)
def remove_member(self, person: Person, board_ref: str, email: str) -> dict:
board = self._board(person, board_ref)
email = board.remove(person, email)
self.repo.remove_member(board.id, email)
self._unassign(person, board, email)
self.repo.log(board.id, None, person.email, f"removed {email}")
return self._members(board)
def _unassign(self, person: Person, board: Board, email: str) -> None:
"""Someone who can no longer edit can't stay assigned: their cards become unassigned."""
for card in self.repo.cards_on(board.id):
if card.assignee == email:
changes = card.update(assignee=None)
self.repo.save_card(card)
self._record(board, card, person, changes)
# --- cards ------------------------------------------------------------------------------
def add_card(self, person: Person, board_ref: str, title: str, list_ref: str = "",
description: str = "") -> dict:
board = self._board(person, board_ref)
board.require_edit(person)
lst = board.find_list(list_ref) if (list_ref or "").strip() else board.ordered()[0]
title = clean_title(title)
pos = 1 + max((c.pos for c in self.repo.cards_on(board.id) if c.list_id == lst.id),
default=-1)
for _ in range(CLAIM_ATTEMPTS):
number = self.repo.max_card_number(board.id) + 1
card = Card(id=f"{board.id}:{number}", board_id=board.id, number=number,
list_id=lst.id, title=title, pos=pos, created_by=person.email)
card.update(description=description)
if self.repo.insert_card(card):
break
else:
raise Invalid("the board is busy right now; try again")
self._record(board, card, person, [Change("created", "", f"{title} (in {lst.name})")])
return self._view(board, card)
def show_card(self, person: Person, card_ref, board_ref: Optional[str] = None) -> dict:
board, card = self._card(person, card_ref, board_ref)
views = {v["id"]: v for v in self._views(board)}
links = self.repo.links_on(board.id)
brief = ("number", "title", "list", "assignee", "due_date", "done")
pick = lambda cid: {k: views[cid][k] for k in brief} # noqa: E731
return {**views[card.id], "role": board.role_of(person).value,
"blocked_by": [pick(b) for c, b in links if c == card.id and b in views],
"blocking": [pick(c) for c, b in links if b == card.id and c in views],
"history": self._history(card)}
def _edit(self, person: Person, board: Board, card: Card, fields: dict) -> dict:
if "due_date" in fields:
fields["due_date"] = parse_due(fields["due_date"], self._today())
if (fields.get("assignee") or "").strip().lower() == "me":
fields["assignee"] = person.email
changes = board.edit_card(person, card, **fields)
if changes:
self.repo.save_card(card)
self._record(board, card, person, changes)
return self._view(board, card)
def update_card(self, person: Person, card_ref, board_ref: Optional[str] = None,
**fields) -> dict:
"""Change any of title, description, assignee, due_date, labels; each is recorded."""
board, card = self._card(person, card_ref, board_ref)
return self._edit(person, board, card, fields)
def assign_card(self, person: Person, card_ref, email: Optional[str],
board_ref: Optional[str] = None) -> dict:
return self.update_card(person, card_ref, board_ref, assignee=email)
def set_due_date(self, person: Person, card_ref, due: Optional[str],
board_ref: Optional[str] = None) -> dict:
return self.update_card(person, card_ref, board_ref, due_date=due)
def describe_card(self, person: Person, card_ref, description: str,
board_ref: Optional[str] = None) -> dict:
return self.update_card(person, card_ref, board_ref, description=description)
def label_card(self, person: Person, card_ref, add=(), remove=(),
board_ref: Optional[str] = None) -> dict:
board, card = self._card(person, card_ref, board_ref)
drop = {str(l).strip().lower() for l in remove or ()}
labels = [l for l in card.labels if l.lower() not in drop] + list(add or ())
return self._edit(person, board, card, {"labels": labels})
def move_card(self, person: Person, card_ref, list_ref: str,
position: Optional[int] = None, board_ref: Optional[str] = None) -> dict:
"""Move a card to a list, at a 1-based position (the end when not given)."""
board, card = self._card(person, card_ref, board_ref)
board.require_edit(person)
to = board.find_list(list_ref)
siblings = sorted((card if c.id == card.id else c
for c in self.repo.cards_on(board.id) if c.list_id == to.id),
key=lambda c: (c.pos, c.number))
index = None if position in (None, "") else int(position) - 1
before = {c.id: c.pos for c in siblings}
changes = board.move_card(person, card, to, siblings, index)
self.repo.save_card(card)
for c in siblings:
if c is not card and before[c.id] != c.pos:
self.repo.save_card(c)
self._record(board, card, person, changes)
return self._view(board, card)
def comment_card(self, person: Person, card_ref, text: str,
board_ref: Optional[str] = None) -> dict:
board, card = self._card(person, card_ref, board_ref)
board.require_edit(person)
text = (text or "").strip()
if not text:
raise Invalid("a comment needs some text")
if len(text) > MAX_TEXT:
raise Invalid(f"keep a comment under {MAX_TEXT} characters")
self._record(board, card, person, [Change("comment", "", text)])
return {"card": card.number, "board": board.title, "comment": text}
def _history(self, card: Card) -> list:
return [{"at": e.at.isoformat(timespec="seconds"), "who": e.actor, "field": e.field,
"old": e.old, "new": e.new}
for e in sorted(self.repo.history(card.id), key=lambda e: e.at)]
def card_history(self, person: Person, card_ref, board_ref: Optional[str] = None) -> dict:
board, card = self._card(person, card_ref, board_ref)
return {"card": card.number, "title": card.title, "board": board.title,
"history": self._history(card)}
# --- across boards: my plate and what's late --------------------------------------------
def my_cards(self, person: Person, within_days: Optional[int] = None) -> dict:
"""Open cards assigned to me. With within_days: only those due by then (overdue too)."""
today = self._today()
if within_days is not None and int(within_days) < 0:
raise Invalid("within_days can't be negative")
horizon = today + timedelta(days=int(within_days)) if within_days is not None else None
cards = [v for b in self._visible(person) for v in self._views(b)
if v["assignee"] == person.email and not v["done"]
and (horizon is None or (v["due_date"] and v["due_date"] <= horizon.isoformat()))]
cards.sort(key=lambda v: (v["due_date"] is None, v["due_date"] or "", v["board"],
v["number"]))
return {"today": today.isoformat(), "within_days": within_days, "cards": cards}
def list_overdue(self, person: Person) -> dict:
cards = [v for b in self._visible(person) for v in self._views(b) if v["overdue"]]
cards.sort(key=lambda v: (v["due_date"], v["board"], v["number"]))
return {"today": self._today().isoformat(), "cards": cards}
# --- dependencies -----------------------------------------------------------------------
def block_card(self, person: Person, card_ref, blocked_by, board_ref=None) -> dict:
"""card can't finish before blocked_by. Both on the same board; loops are refused."""
board, card = self._card(person, card_ref, board_ref)
board.require_edit(person)
_, blocker = self._card(person, blocked_by, board.id)
links = self.repo.links_on(board.id)
if (card.id, blocker.id) not in links:
loop = blocking_path(links, card.id, blocker.id)
if loop is not None:
if len(loop) == 1:
raise Invalid("a card can't block itself")
nums = {c.id: c.number for c in self.repo.cards_on(board.id)}
chain = " ← ".join(f"card {nums.get(i, '?')}" for i in loop)
raise Invalid(f"that would make a loop: {chain}; card {blocker.number} "
f"already waits on card {card.number}")
self.repo.add_link(board.id, card.id, blocker.id)
self._record(board, card, person,
[Change("blocked by", "", f"card {blocker.number}")])
return self.show_card(person, card.id)
def unblock_card(self, person: Person, card_ref, blocked_by, board_ref=None) -> dict:
board, card = self._card(person, card_ref, board_ref)
board.require_edit(person)
_, blocker = self._card(person, blocked_by, board.id)
if (card.id, blocker.id) not in self.repo.links_on(board.id):
raise Invalid(f"card {card.number} is not blocked by card {blocker.number}")
self.repo.remove_link(card.id, blocker.id)
self._record(board, card, person, [Change("blocked by", f"card {blocker.number}", "")])
return self.show_card(person, card.id)
def what_is_blocking(self, person: Person, ref: str) -> dict:
"""ref is a board (title or id): every waiting card and what it waits on. Or a card
(number, id or title): the open cards it waits on, directly and further down."""
board = None
try:
board = self._board(person, ref)
except NotFound:
pass
brief = ("id", "number", "title", "list", "assignee", "due_date", "overdue")
if board is not None:
views = {v["id"]: v for v in self._views(board)}
waits = {}
for c, b in self.repo.links_on(board.id):
if c in views and b in views and not views[b]["done"] and not views[c]["done"]:
waits.setdefault(c, []).append({k: views[b][k] for k in brief})
return {"board": board.title,
"blocked": [{"card": {k: views[c][k] for k in brief}, "waiting_on": w}
for c, w in sorted(waits.items(),
key=lambda i: views[i[0]]["number"])]}
try:
board, card = self._card(person, ref)
except NotFound:
raise NotFound(f"no board or card called {ref!r} among yours")
views = {v["id"]: v for v in self._views(board)}
blocked_by = {}
for c, b in self.repo.links_on(board.id):
blocked_by.setdefault(c, []).append(b)
is_open = lambda cid: cid in views and not views[cid]["done"] # noqa: E731
direct = [b for b in blocked_by.get(card.id, []) if is_open(b)]
chain, todo, seen = [], list(direct), {card.id}
while todo:
nxt = todo.pop(0)
if nxt in seen:
continue
seen.add(nxt)
chain.append(nxt)
todo += [b for b in blocked_by.get(nxt, []) if is_open(b)]
return {"board": board.title, "card": {k: views[card.id][k] for k in brief},
"blocked": bool(direct),
"waiting_on": [{k: views[b][k] for k in brief} for b in direct],
"whole_chain": [{k: views[b][k] for k in brief} for b in chain]}
# --- AI (Dodil Models through the Assistant port) ---------------------------------------
def suggest_checklist(self, person: Person, card_ref, board_ref=None) -> dict:
"""An AI suggestion only: it changes nothing. A person decides what to keep."""
_, card = self._card(person, card_ref, board_ref)
return {"card": card.number, "suggested_checklist": self._ai().checklist(card.title)}
def similar_cards(self, person: Person, card_ref, board_ref=None, limit: int = 5) -> dict:
"""Cards that mean something close to this one, on any board the person is on.
Embeddings are made lazily: only cards whose text changed since last time."""
board, card = self._card(person, card_ref, board_ref)
ai = self._ai()
boards = {b.id: b for b in self._visible(person)}
cards = {c.id: c for b in boards.values() for c in self.repo.cards_on(b.id)}
known = self.repo.embedding_digests(list(boards))
stale = [c for c in cards.values() if known.get(c.id) != _digest(c)]
note = None
try:
for i in range(0, len(stale), EMBED_BATCH):
chunk = stale[i:i + EMBED_BATCH]
for c, vector in zip(chunk, ai.embed([_meaning(c) for c in chunk])):
self.repo.save_embedding(c, _digest(c), vector)
except Exception: # a model outage answers with what is already embedded, never a 500
note = "the AI search is unavailable right now; results cover cards embedded earlier"
vector = self.repo.vector_of(card.id)
hits = [(cid, d) for cid, d in
self.repo.nearest(list(boards), vector, card.id, max(1, min(int(limit), 20)))
if cid in cards and d < 1] if vector else [] # d >= 1: nothing in common
views = {}
for bid in {cards[cid].board_id for cid, _ in hits}:
views.update({v["id"]: v for v in self._views(boards[bid])})
return {"card": card.number, "board": board.title,
"similar": [{**{k: views[cid][k] for k in
("id", "number", "board", "list", "title", "assignee")},
"similarity": round(1 - float(d), 3)} for cid, d in hits],
**({"note": note} if note else {})}
def board_summary(self, person: Person, board_ref: str) -> dict:
"""A short daily summary. The facts are worked out here; the model only words them."""
board = self._board(person, board_ref)
today, views = self._today(), self._views(board)
numbers = {v["id"]: v["number"] for v in views}
soon = (today + timedelta(days=2)).isoformat()
line = lambda v: (f"card {v['number']} {v['title']!r}" # noqa: E731
f"{' due ' + v['due_date'] if v['due_date'] else ''}"
f"{' (' + v['assignee'] + ')' if v['assignee'] else ''}")
facts = [f"Board {board.title!r}, today {today.isoformat()}."]
facts += [f"{l.name}: {sum(v['list_id'] == l.id for v in views)} cards"
for l in board.ordered()]
facts += ["Overdue: " + line(v) for v in views if v["overdue"]]
facts += ["Due soon: " + line(v) for v in views
if not v["done"] and not v["overdue"] and v["due_date"]
and v["due_date"] <= soon]
facts += ["Blocked: " + line(v) for v in views if v["blocked"] and not v["done"]]
recent = sorted(self.repo.board_history(board.id, self.clock() - timedelta(days=1)),
key=lambda e: e.at)
facts += [f"Change: {e.actor} {e.field} on card {numbers.get(e.card_id, '?')}: "
f"{e.old or '-'} -> {e.new or '-'}" for e in recent[-30:]]
summary, note = None, None
if self.assistant is not None:
try:
summary = self.assistant.summarize("\n".join(facts))
except Exception: # a model outage must not hide the facts we already have
note = "the AI summary is unavailable right now; these are the facts"
return {"board": board.title, "today": today.isoformat(),
"summary": summary or "\n".join(facts), "facts": facts,
**({"note": note} if note else {})}