sawera/sawera-boards
Sawera Boards - a team task board for the Dodil Boards Challenge. Boards, lists, cards, roles, activity log, MCP actions and AI extras. FastAPI + DataK3 + appid + Models + MCP.
A remix template · 0 remixes · Remix sawera/sawera-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.
Add a use case here, then expose it in doors/web.py and doors/mcp.py (one line each).
"""
import hashlib
import math
import uuid
from datetime import date, timedelta
from typing import Optional
from boards.application.ports import Assistant, BoardRepository
from boards.domain.model import (DEFAULT_LISTS, Board, BoardList, Card, Comment,
Forbidden, Invalid, Person, Role, clean_labels)
class NotFound(Exception):
pass
def _new_id() -> str:
return uuid.uuid4().hex[:12]
SEMANTIC_MIN_SCORE = 0.40 # cosine; below this a card is not a meaning-match
def _card_text(card: Card) -> str:
"""The text a card is searched and embedded by: title, description, labels."""
return " ".join([card.title, card.description, " ".join(card.labels)]).strip()
def _text_hash(text: str) -> str:
return hashlib.sha1(text.encode("utf-8")).hexdigest()[:16]
def _cosine(a, b) -> float:
if not a or not b:
return 0.0
dot = sum(x * y for x, y in zip(a, b))
na = math.sqrt(sum(x * x for x in a))
nb = math.sqrt(sum(y * y for y in b))
return dot / (na * nb) if na and nb else 0.0
class Boards:
def __init__(self, repo: BoardRepository, assistant: Optional[Assistant] = None):
self.repo, self.assistant = repo, assistant
# --- helpers --------------------------------------------------------------
def _board(self, board_id: str, person: Person) -> Board:
board = self.repo.get(board_id)
if board is None:
raise NotFound("no such board")
try:
board.require_view(person)
except Forbidden:
raise NotFound("no such board") # a board you can't see doesn't exist for you
return board
def _find_card(self, card_id: str, person: Person) -> tuple:
card = self.repo.get_card(card_id)
if card is None:
raise NotFound("no such card")
return self._board(card.board_id, person), card
def _cc(self, board_id: str, cache: Optional[dict] = None) -> dict:
"""Comment counts for one board ({card_id: n}), memoized into `cache`
when the caller is building dicts for many boards at once."""
if cache is None:
return self.repo.comment_counts(board_id)
if board_id not in cache:
cache[board_id] = self.repo.comment_counts(board_id)
return cache[board_id]
def _card_dict(self, board: Board, card: Card, today: Optional[date] = None,
cc: Optional[dict] = None) -> dict:
today = today or date.today()
lst = board.list_by_id(card.list_id)
cards = {c.id: c for c in self.repo.cards_on(board.id)}
blockers = [cards[b] for b in card.blocked_by if b in cards]
blocked = any(not board.list_by_id(b.list_id).is_done for b in blockers)
return {"id": card.id, "board_id": card.board_id, "title": card.title,
"description": card.description, "list_id": card.list_id,
"list_name": lst.name, "assignee": card.assignee,
"due_date": card.due_date.isoformat() if card.due_date else None,
"labels": list(card.labels), "position": card.position,
"comments": (cc or {}).get(card.id, 0),
"overdue": card.is_overdue(today, lst.is_done),
"blocked": blocked,
"blocked_by": [{"id": b.id, "title": b.title} for b in blockers]}
def _resolve_list(self, board: Board, list_name: Optional[str],
list_id: Optional[str]) -> BoardList:
if list_id:
return board.list_by_id(list_id)
if list_name:
lst = board.list_by_name(list_name)
if lst is None:
raise Invalid(f"no list called {list_name!r} on this board")
return lst
if not board.lists:
raise Invalid("this board has no lists")
return board.lists[0]
# --- boards & members -------------------------------------------------------
def create_board(self, person: Person, title: str) -> dict:
if not title.strip():
raise Invalid("a board needs a title")
board = Board(id=_new_id(), title=title.strip(), owner_email=person.email)
board.lists = [BoardList(id=_new_id(), board_id=board.id, name=name, position=float(i))
for i, name in enumerate(DEFAULT_LISTS)]
self.repo.save_board(board)
self.repo.save_lists(board)
self.repo.log(board.id, None, person.email, f"created board {board.title!r}")
return {"id": board.id, "title": board.title}
def delete_board(self, person: Person, board_id: str) -> dict:
"""Owner only. Everything on the board goes with it: lists, cards,
comments, memberships and the activity log."""
board = self._board(board_id, person)
board.check_deletable(person)
self.repo.delete_board(board.id)
return {"deleted": board.id}
def _embed_card(self, card: Card) -> None:
"""Store the card's meaning-vector at write time, so a meaning search
only has to embed the question, not every card, and answers in
seconds instead of a minute. Best-effort: a card without a stored
vector is still found by the keyword half of search, and the search
path backfills any vector it finds missing."""
embed = getattr(self.assistant, "embed", None)
if embed is None:
return
text = _card_text(card)
if len(text.split()) < 3:
return
try:
vec = embed([text])[0]
if vec:
self.repo.save_embedding(card.id, _text_hash(text), tuple(vec))
except Exception:
pass # an embedding must never break saving the card itself
def my_boards(self, person: Person) -> list:
return [{"id": b.id, "title": b.title, "role": b.role_of(person).value}
for b in self.repo.boards_for(person.email)]
def show_board(self, person: Person, board_id: str) -> dict:
board = self._board(board_id, person)
today = date.today()
cc = self._cc(board.id)
cards = [self._card_dict(board, c, today, cc) for c in self.repo.cards_on(board.id)]
return {"id": board.id, "title": board.title, "role": board.role_of(person).value,
"owner": board.owner_email,
"members": [{"email": e, "role": r.value}
for e, r in sorted(board.members.items())],
"lists": [{"id": l.id, "name": l.name, "is_done": l.is_done}
for l in board.lists],
"cards": cards}
def add_member(self, person: Person, board_id: str, email: str, role: str) -> dict:
board = self._board(board_id, person)
try:
role_e = Role(role)
except ValueError:
raise Invalid("role must be member or viewer")
board.add_member(person, email, role_e)
self.repo.save_board(board)
self.repo.log(board.id, None, person.email, f"added {email.strip().lower()} as {role_e.value}")
return {"board_id": board.id, "email": email.strip().lower(), "role": role_e.value}
def remove_member(self, person: Person, board_id: str, email: str) -> dict:
board = self._board(board_id, person)
board.remove_member(person, email)
self.repo.save_board(board)
self.repo.log(board.id, None, person.email, f"removed {email.strip().lower()}")
return {"board_id": board.id, "removed": email.strip().lower()}
# --- lists -------------------------------------------------------------------
def add_list(self, person: Person, board_id: str, name: str) -> dict:
board = self._board(board_id, person)
pos = max((l.position for l in board.lists), default=-1.0) + 1.0
lst = board.add_list(person, BoardList(id=_new_id(), board_id=board.id,
name=name, position=pos))
self.repo.save_lists(board)
self.repo.log(board.id, None, person.email, f"added list {lst.name!r}")
return {"id": lst.id, "name": lst.name}
def rename_list(self, person: Person, board_id: str, list_id: str, name: str) -> dict:
board = self._board(board_id, person)
lst = board.rename_list(person, list_id, name)
self.repo.save_lists(board)
self.repo.log(board.id, None, person.email, f"renamed a list to {lst.name!r}")
return {"id": lst.id, "name": lst.name}
def remove_list(self, person: Person, board_id: str, list_id: str) -> dict:
board = self._board(board_id, person)
count = sum(1 for c in self.repo.cards_on(board.id) if c.list_id == list_id)
board.remove_list(person, list_id, count)
self.repo.save_lists(board)
self.repo.log(board.id, None, person.email, "removed a list")
return {"board_id": board.id, "removed_list": list_id}
# --- cards -------------------------------------------------------------------
def add_card(self, person: Person, board_id: str, title: str,
list_name: Optional[str] = None, list_id: Optional[str] = None,
description: str = "", assignee: Optional[str] = None,
due_date: Optional[date] = None, labels=()) -> dict:
board = self._board(board_id, person)
lst = self._resolve_list(board, list_name, list_id)
siblings = [c for c in self.repo.cards_on(board.id) if c.list_id == lst.id]
card = Card(id=_new_id(), board_id=board.id, list_id=lst.id,
title=(title or "").strip(), description=(description or "").strip(),
assignee=assignee.strip().lower() if assignee else None,
due_date=due_date, labels=clean_labels(labels),
position=max((c.position for c in siblings), default=0.0) + 1.0)
board.check_new_card(person, card)
self.repo.save_card(card)
self._embed_card(card)
self.repo.log(board.id, card.id, person.email,
f"added card {card.title!r} to {lst.name}")
return self._card_dict(board, card, cc=self._cc(board.id))
def update_card(self, person: Person, card_id: str, **fields) -> dict:
board, card = self._find_card(card_id, person)
board.require_edit(person)
changed = []
if "title" in fields and fields["title"] is not None:
t = str(fields["title"]).strip()
if not t:
raise Invalid("a card needs a title")
card.title, _ = t, changed.append("title")
if "description" in fields and fields["description"] is not None:
card.description = str(fields["description"]).strip()
changed.append("description")
if "assignee" in fields:
a = fields["assignee"]
card.assignee = a.strip().lower() if a else None
changed.append("assignee")
if "due_date" in fields:
card.due_date = fields["due_date"]
changed.append("due date")
if "labels" in fields and fields["labels"] is not None:
card.labels = clean_labels(fields["labels"])
changed.append("labels")
self.repo.save_card(card)
if "title" in changed or "description" in changed:
self._embed_card(card)
if changed:
self.repo.log(board.id, card.id, person.email,
f"updated {', '.join(changed)} on {card.title!r}")
return self._card_dict(board, card, cc=self._cc(board.id))
def assign_card(self, person: Person, card_id: str, email: Optional[str]) -> dict:
return self.update_card(person, card_id, assignee=email)
def move_card(self, person: Person, card_id: str, list_name: Optional[str] = None,
list_id: Optional[str] = None, index: Optional[int] = None) -> dict:
board, card = self._find_card(card_id, person)
board.require_edit(person)
target = self._resolve_list(board, list_name, list_id)
before = board.list_by_id(card.list_id).name
siblings = [c for c in self.repo.cards_on(board.id)
if c.list_id == target.id and c.id != card.id]
siblings.sort(key=lambda c: c.position)
if index is None or index >= len(siblings):
pos = (siblings[-1].position + 1.0) if siblings else 1.0
elif index <= 0:
pos = siblings[0].position - 1.0
else:
pos = (siblings[index - 1].position + siblings[index].position) / 2.0
card.list_id, card.position = target.id, pos
self.repo.save_card(card)
if target.name != before:
what = f"moved {card.title!r} from {before} to {target.name}"
else:
what = f"reordered {card.title!r} in {target.name}"
self.repo.log(board.id, card.id, person.email, what)
return self._card_dict(board, card, cc=self._cc(board.id))
def delete_card(self, person: Person, card_id: str) -> dict:
board, card = self._find_card(card_id, person)
board.require_edit(person)
for other in self.repo.cards_on(board.id):
if card.id in other.blocked_by:
other.blocked_by = tuple(b for b in other.blocked_by if b != card.id)
self.repo.save_card(other)
self.repo.delete_card(card.id)
self.repo.log(board.id, card.id, person.email, f"deleted card {card.title!r}")
return {"deleted": card.id}
# --- comments ------------------------------------------------------------------
def add_comment(self, person: Person, card_id: str, body: str) -> dict:
"""Editors discuss the work on the card itself. Viewers can read the
discussion but not join it (the same edit rule as every other change)."""
board, card = self._find_card(card_id, person)
board.require_edit(person)
text = (body or "").strip()
if not text:
raise Invalid("a comment needs some text")
if len(text) > 2000:
raise Invalid("a comment can be at most 2000 characters")
comment = Comment(id=_new_id(), card_id=card.id, author=person.email, body=text)
self.repo.save_comment(comment)
self.repo.log(board.id, card.id, person.email, f"commented on {card.title!r}")
saved = self.repo.comments_on(card.id)
mine = [c for c in saved if c.id == comment.id]
return {"id": comment.id, "card_id": card.id, "author": comment.author,
"body": comment.body, "at": mine[0].at if mine else None}
def card_comments(self, person: Person, card_id: str) -> list:
self._find_card(card_id, person) # any board member may read
return [{"id": c.id, "card_id": c.card_id, "author": c.author,
"body": c.body, "at": c.at} for c in self.repo.comments_on(card_id)]
# --- blocked-by ----------------------------------------------------------------
def set_blockers(self, person: Person, card_id: str, blocker_ids: list) -> dict:
board, card = self._find_card(card_id, person)
board.require_edit(person)
all_cards = {c.id: c for c in self.repo.cards_on(board.id)}
proposed = tuple(dict.fromkeys(blocker_ids or ()))
board.check_block_links(card, proposed, all_cards) # validate BEFORE storing
card.blocked_by = proposed
self.repo.save_card(card)
names = [all_cards[b].title for b in card.blocked_by]
self.repo.log(board.id, card.id, person.email,
f"set blockers of {card.title!r} to {names or 'none'}")
return self._card_dict(board, card, cc=self._cc(board.id))
def what_is_blocking(self, person: Person, card_id: str) -> dict:
"""The unfinished cards holding this one up, following links transitively."""
board, card = self._find_card(card_id, person)
all_cards = {c.id: c for c in self.repo.cards_on(board.id)}
cc = self._cc(board.id)
found, seen, queue = [], set(), [(b, 1) for b in card.blocked_by]
while queue:
bid, depth = queue.pop(0)
if bid in seen or bid not in all_cards:
continue
seen.add(bid)
b = all_cards[bid]
if not board.list_by_id(b.list_id).is_done:
found.append({**self._card_dict(board, b, cc=cc), "depth": depth})
queue.extend((nxt, depth + 1) for nxt in b.blocked_by)
return {"card": self._card_dict(board, card, cc=cc), "blocking": found}
# --- cross-board views -----------------------------------------------------------
def _my_cards(self, person: Person) -> list:
out = []
today = date.today()
for board in self.repo.boards_for(person.email):
done_ids = {l.id for l in board.lists if l.is_done}
for c in self.repo.cards_on(board.id):
if c.list_id not in done_ids:
out.append((board, c))
return out
def my_cards(self, person: Person, days: int = 7) -> list:
"""Your plate: cards assigned to you that aren't finished. Overdue first,
then due within `days`, then the rest by due date; undated last."""
today = date.today()
horizon = today + timedelta(days=max(days, 0))
mine = [(b, c) for b, c in self._my_cards(person) if c.assignee == person.email]
def rank(item):
_, c = item
if c.due_date and c.due_date < today:
return (0, c.due_date)
if c.due_date and c.due_date <= horizon:
return (1, c.due_date)
if c.due_date:
return (2, c.due_date)
return (3, today)
cache = {}
return [self._card_dict(b, c, today, self._cc(b.id, cache))
for b, c in sorted(mine, key=rank)][:50]
def overdue_cards(self, person: Person) -> list:
today = date.today()
out, cache = [], {}
for board in self.repo.boards_for(person.email):
done_ids = {l.id for l in board.lists if l.is_done}
for c in self.repo.cards_on(board.id):
if c.is_overdue(today, c.list_id in done_ids):
out.append(self._card_dict(board, c, today, self._cc(board.id, cache)))
return sorted(out, key=lambda d: d["due_date"] or "")
def search_cards(self, person: Person, query: str, semantic: bool = True) -> list:
"""Keyword matches first, then semantic (meaning) matches, best first.
Semantic search embeds the query and each card's text with Dodil Models and
ranks by cosine similarity, so "money owed" finds "Send the invoice". Card
vectors are stored when a card is saved and recomputed only when the text
changed. The semantic half is best-effort: if the AI is off or fails, the
keyword results still come back. Callers that need an instant answer pass
semantic=False for the keyword half alone (the web page asks for the
meaning matches in a second call, so the person sees word hits at once
instead of staring at a spinner while the model thinks)."""
q = (query or "").strip().lower()
if not q:
raise Invalid("a search needs some text")
boards = self.repo.boards_for(person.email)
out, seen, cache = [], set(), {}
for board in boards:
for c in self.repo.cards_on(board.id):
if q in _card_text(c).lower():
out.append(self._card_dict(board, c, cc=self._cc(board.id, cache)))
seen.add(c.id)
if semantic:
try:
out += self._semantic_matches(boards, q, seen, cache)
except Exception:
pass # search must never break because the AI is down
return out[:30]
def _semantic_matches(self, boards: list, q: str, seen: set, cache=None) -> list:
embed = getattr(self.assistant, "embed", None)
if embed is None:
return []
qv = embed([q])[0]
scored = []
for board in boards:
# Cards with only a word or two of text carry no reliable meaning —
# short generic titles ("First task") sit close to every query in
# vector space and would drown the real matches. They are still
# found by the keyword half above; meaning needs a little text.
cards = [c for c in self.repo.cards_on(board.id)
if len(_card_text(c).split()) >= 3]
stored = self.repo.get_embeddings(board.id)
fresh, stale = {}, []
for c in cards:
text = _card_text(c)
got = stored.get(c.id)
if got and got[0] == _text_hash(text) and got[1]:
fresh[c.id] = got[1]
else:
stale.append((c, text))
if stale:
for (c, text), vec in zip(stale, embed([t for _, t in stale])):
if vec:
self.repo.save_embedding(c.id, _text_hash(text), tuple(vec))
fresh[c.id] = tuple(vec)
by_id = {c.id: c for c in cards}
for cid, vec in fresh.items():
if cid in seen:
continue
score = _cosine(qv, vec)
if score >= SEMANTIC_MIN_SCORE:
d = self._card_dict(board, by_id[cid], cc=self._cc(board.id, cache))
d["score"] = round(score, 2)
scored.append((score, d))
scored.sort(key=lambda pair: -pair[0])
return [d for _, d in scored[:10]]
def board_activity(self, person: Person, board_id: str, limit: int = 50) -> list:
self._board(board_id, person)
return self.repo.activity(board_id, max(1, min(limit, 200)))
# --- AI extras ---------------------------------------------------------------------
def suggest_checklist(self, person: Person, card_id: str) -> dict:
"""An AI suggestion only: it changes nothing. A person decides what to keep."""
_, card = self._find_card(card_id, person)
if self.assistant is None:
raise Invalid("no AI assistant configured")
return {"card_id": card.id,
"suggested_checklist": self.assistant.checklist(card.title)}
def daily_summary(self, person: Person, board_id: str) -> dict:
board = self._board(board_id, person)
if self.assistant is None:
raise Invalid("no AI assistant configured")
today = date.today()
dicts = [self._card_dict(board, c, today) for c in self.repo.cards_on(board.id)]
counts = {}
for d in dicts:
counts[d["list_name"]] = counts.get(d["list_name"], 0) + 1
facts = [f"Board: {board.title}",
"Cards per list: " + ", ".join(f"{k}: {v}" for k, v in counts.items()),
"Overdue: " + (", ".join(d["title"] for d in dicts if d["overdue"]) or "none"),
"Blocked: " + (", ".join(d["title"] for d in dicts if d["blocked"]) or "none"),
"Recent activity:"]
facts += [f"- {e['actor']} {e['action']}" for e in self.repo.activity(board.id, 8)]
return {"board_id": board.id,
"summary": self.assistant.summarize("\n".join(facts))}