flc/crewboard
Crewboard - team task board (boards, lists, cards, roles, history, drag and drop, MCP, AI extras). FastAPI + DataK3 + appid + Models.
A remix template · 1 remixes · Remix flc/crewboard
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 re
from datetime import date, timedelta
from typing import Callable, Optional
from boards.application.ports import Assistant, BoardRepository, Taken
from boards.domain.model import (Board, Card, Forbidden, Invalid, Conflict, Person, Role,
board_key, check_dependency, clean_labels, open_blockers)
class NotFound(Exception):
pass
CARD_REF = re.compile(r"^\s*([A-Za-z0-9]+)-(\d+)\s*$")
NUMBER_REF = re.compile(r"^\s*(?:card\s*)?#?(\d+)\s*$", re.I)
def parse_due(value) -> Optional[date]:
if value in (None, ""):
return None
if isinstance(value, date):
return value
try:
return date.fromisoformat(str(value).strip()[:10])
except ValueError:
raise Invalid(f"due date {value!r} is not YYYY-MM-DD")
class Boards:
def __init__(self, repo: BoardRepository, assistant: Optional[Assistant] = None,
today: Callable[[], date] = date.today):
self.repo, self.assistant, self.today = repo, assistant, today
self._vectors = {} # (card id, text hash) -> embedding; cheap cache, rebuilt on restart
# --- reading, always scoped to the caller -------------------------------------------
def _board(self, board_id: str, person: Person) -> Board:
board = self.repo.get((board_id or "").strip().upper())
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 _visible(self, person: Person) -> list:
return self.repo.boards_for(person.email)
def _resolve(self, person: Person, ref: str) -> tuple:
"""'OPS-12', '12', '#12' or 'card 12' -> (board, card), among boards the person can see."""
ref = str(ref)
m = CARD_REF.match(ref)
if m:
card = self.repo.get_card(f"{m.group(1).upper()}-{int(m.group(2))}")
if card is None:
raise NotFound(f"no card {ref.strip()}")
return self._board(card.board_id, person), card
m = NUMBER_REF.match(ref)
if not m:
raise Invalid(f"{ref!r} is not a card reference; use the card id like OPS-12")
n = int(m.group(1))
hits = []
for board in self._visible(person):
card = self.repo.get_card(f"{board.id}-{n}")
if card is not None:
hits.append((board, card))
if not hits:
raise NotFound(f"no card {n} on your boards")
if len(hits) > 1:
raise Invalid(f"card {n} exists on several boards: "
f"{', '.join(c.id for _, c in hits)}; say which one")
return hits[0]
def _card_out(self, c: Card, board: Board, by_id: Optional[dict] = None) -> dict:
today = self.today()
blockers = open_blockers(c, by_id or {}, board.done_list) if by_id is not None else None
out = {"id": c.id, "board_id": c.board_id, "number": c.number, "title": c.title,
"description": c.description, "list_name": c.list_name, "position": c.position,
"assignee": c.assignee, "labels": c.labels, "blocked_by": c.blocked_by,
"due_date": c.due_date.isoformat() if c.due_date else None,
"overdue": c.is_overdue(today, board.done_list), "archived": c.archived,
"version": c.version}
if blockers is not None:
out["blocked"] = bool(blockers)
return out
def my_boards(self, person: Person) -> list:
return [{"id": b.id, "title": b.title, "role": b.role_of(person).value, "lists": b.lists}
for b in self._visible(person)]
def show_board(self, person: Person, board_id: str, archived: bool = False) -> dict:
board = self._board(board_id, person)
cards = self.repo.cards_on(board.id)
by_id = {c.id: c for c in cards}
shown = sorted((c for c in cards if c.archived == archived),
key=lambda c: (board.lists.index(c.list_name)
if c.list_name in board.lists else 99, c.position))
return {"id": board.id, "title": board.title, "role": board.role_of(person).value,
"lists": board.lists, "done_list": board.done_list, "people": board.people(),
"cards": [self._card_out(c, board, by_id) for c in shown]}
def show_card(self, person: Person, card_ref: str) -> dict:
board, card = self._resolve(person, card_ref)
by_id = {c.id: c for c in self.repo.cards_on(board.id)}
out = self._card_out(card, board, by_id)
out["waiting_for"] = [self._brief(c) for c in open_blockers(card, by_id, board.done_list)]
out["history"] = self.repo.activity(board.id, card.id, limit=20)
return out
@staticmethod
def _brief(c: Card) -> dict:
return {"id": c.id, "title": c.title, "list_name": c.list_name, "assignee": c.assignee,
"due_date": c.due_date.isoformat() if c.due_date else None}
def my_cards(self, person: Person, days: int = 7) -> dict:
"""'What's on my plate this week?': my unfinished cards, most urgent first."""
today, horizon = self.today(), self.today() + timedelta(days=max(0, int(days)))
plate = {"overdue": [], "due_soon": [], "later": [], "no_due_date": []}
for board in self._visible(person):
for c in self.repo.cards_on(board.id):
if c.assignee != person.email or c.archived or c.list_name == board.done_list:
continue
item = {**self._brief(c), "board": board.title}
if c.due_date is None:
plate["no_due_date"].append(item)
elif c.due_date < today:
plate["overdue"].append(item)
elif c.due_date <= horizon:
plate["due_soon"].append(item)
else:
plate["later"].append(item)
for items in plate.values():
items.sort(key=lambda i: (i["due_date"] or "9999", i["id"]))
plate["window"] = f"{today.isoformat()} to {horizon.isoformat()}"
return plate
def overdue(self, person: Person, board_id: Optional[str] = None) -> list:
"""Unfinished cards past their due date, on one board or every board you can see."""
boards = [self._board(board_id, person)] if board_id else self._visible(person)
today, out = self.today(), []
for board in boards:
for c in self.repo.cards_on(board.id):
if c.is_overdue(today, board.done_list):
out.append({**self._brief(c), "board": board.title,
"days_late": (today - c.due_date).days})
return sorted(out, key=lambda i: -i["days_late"])
def find_cards(self, person: Person, text: str, board_id: Optional[str] = None) -> list:
"""Plain word search over titles, descriptions and labels."""
words = [w for w in re.findall(r"\w+", (text or "").lower()) if len(w) > 1]
if not words:
raise Invalid("say what to look for")
boards = [self._board(board_id, person)] if board_id else self._visible(person)
out = []
for board in boards:
for c in self.repo.cards_on(board.id):
hay = " ".join([c.title, c.description, " ".join(c.labels)]).lower()
if all(w in hay for w in words):
out.append({**self._brief(c), "board": board.title})
return out[:50]
def card_history(self, person: Person, card_ref: str) -> list:
board, card = self._resolve(person, card_ref)
return self.repo.activity(board.id, card.id, limit=100)
def board_activity(self, person: Person, board_id: str, limit: int = 50) -> list:
board = self._board(board_id, person)
return self.repo.activity(board.id, None, limit=max(1, min(int(limit), 200)))
def what_blocks(self, person: Person, card_ref: str) -> dict:
"""'What is blocking the launch?': every unfinished card it waits for, through chains."""
board, card = self._resolve(person, card_ref)
by_id = {c.id: c for c in self.repo.cards_on(board.id)}
chain = open_blockers(card, by_id, board.done_list)
return {"card": self._brief(card), "blocked": bool(chain),
"waiting_for": [self._brief(c) for c in chain]}
# --- boards, lists and people (owner) -----------------------------------------------
def create_board(self, person: Person, title: str) -> dict:
title = (title or "").strip()[:80]
if not title:
raise Invalid("a board needs a title")
key = board_key(title)
for attempt in range(1, 50): # the key is claimed by a plain INSERT: one writer wins
board = Board(id=key if attempt == 1 else f"{key}{attempt}", title=title,
owner_email=person.email)
try:
self.repo.insert_board(board)
break
except Taken:
continue
else:
raise Invalid("could not find a free board key; try another title")
self.repo.log(board.id, None, person.email, f"created board {board.title!r}")
return {"id": board.id, "title": board.title, "lists": board.lists}
def rename_board(self, person: Person, board_id: str, title: str) -> dict:
board = self._board(board_id, person)
board.require_owner(person)
title = (title or "").strip()[:80]
if not title:
raise Invalid("a board needs a title")
before, board.title = board.title, title
self.repo.save_board(board)
self.repo.log(board.id, None, person.email, "renamed the board",
{"title": [before, title]})
return {"id": board.id, "title": board.title}
def add_member(self, person: Person, board_id: str, email: str, role: str = "member") -> dict:
board = self._board(board_id, person)
try:
role_ = Role((role or "member").strip().lower())
except ValueError:
raise Invalid("role is 'member' or 'viewer'")
before = board.members.get((email or "").strip().lower())
board.add_member(person, email, role_)
email = email.strip().lower()
self.repo.save_members(board)
what = f"changed {email} to {role_.value}" if before else f"added {email} as {role_.value}"
self.repo.log(board.id, None, person.email, what,
{"role": [before.value if before else None, role_.value]})
return {"board_id": board.id, "email": email, "role": role_.value}
def remove_member(self, person: Person, board_id: str, email: str) -> dict:
board = self._board(board_id, person)
board.remove_member(person, email)
email = email.strip().lower()
self.repo.save_members(board, removed=(email,))
for c in self.repo.cards_on(board.id): # their cards go back to the pool, on the record
if c.assignee == email:
self._write(person, board, c, c.version, "unassigned (left the board)",
assignee=None)
self.repo.log(board.id, None, person.email, f"removed {email}")
return {"board_id": board.id, "removed": email}
def people(self, person: Person, board_id: str) -> list:
return self._board(board_id, person).people()
def add_list(self, person: Person, board_id: str, name: str,
before: Optional[str] = None) -> dict:
board = self._board(board_id, person)
board.add_list(person, name, before)
self.repo.save_board(board)
self.repo.log(board.id, None, person.email, f"added list {name.strip()!r}")
return {"board_id": board.id, "lists": board.lists}
def rename_list(self, person: Person, board_id: str, old: str, new: str) -> dict:
board = self._board(board_id, person)
old, new = board.rename_list(person, old, new)
self.repo.save_board(board)
for c in self.repo.cards_on(board.id):
if c.list_name == old:
v = c.version
c.list_name, c.version = new, v + 1
self.repo.update_card(c, v)
self.repo.log(board.id, None, person.email, f"renamed list {old!r} to {new!r}")
return {"board_id": board.id, "lists": board.lists}
def remove_list(self, person: Person, board_id: str, name: str) -> dict:
board = self._board(board_id, person)
target = board.list_named(name)
count = sum(1 for c in self.repo.cards_on(board.id)
if c.list_name == target and not c.archived)
removed = board.remove_list(person, target, count)
self.repo.save_board(board)
self.repo.log(board.id, None, person.email, f"removed list {removed!r}")
return {"board_id": board.id, "lists": board.lists}
# --- cards (owner and members) ------------------------------------------------------
def _write(self, person: Person, board: Board, card: Card, expected: int, action: str,
**changes) -> dict:
"""Apply changes, bump the version, write only if nobody else wrote first, then log."""
diff = {}
list_change = changes.pop("list_name", None)
position = changes.pop("position", None)
blocked_by = changes.pop("blocked_by", None)
archived = changes.pop("archived", None)
diff.update(card.edit(**changes))
if list_change is not None and list_change != card.list_name:
diff["list"] = [card.list_name, list_change]
card.list_name = list_change
if position is not None:
card.position = position
if blocked_by is not None and blocked_by != card.blocked_by:
diff["blocked_by"] = [card.blocked_by, blocked_by]
card.blocked_by = blocked_by
if archived is not None and archived != card.archived:
diff["archived"] = [card.archived, archived]
card.archived = archived
if not diff and position is None:
return self._card_out(card, board)
card.version = expected + 1
if not self.repo.update_card(card, expected):
raise Conflict(f"{card.id} was changed by someone else since you opened it; "
"reload and try again")
if diff:
self.repo.log(board.id, card.id, person.email, action, diff)
return self._card_out(card, board)
def _editable(self, person: Person, card_ref: str, version: Optional[int]) -> tuple:
board, card = self._resolve(person, card_ref)
board.require_edit(person)
if version is not None and int(version) != card.version:
raise Conflict(f"{card.id} was changed since you opened it (you had version "
f"{version}, it is now {card.version}); reload and try again")
return board, card
def _check_assignee(self, board: Board, email: Optional[str]) -> Optional[str]:
if email in (None, ""):
return None
email = email.strip().lower()
if email in ("me",):
raise Invalid("say the person's email")
if not board.can_work(email):
raise Invalid(f"{email} is not an owner or member of this board "
"(viewers can't be given cards)")
return email
def add_card(self, person: Person, board_id: str, title: str, list_name: Optional[str] = None,
description: str = "", assignee: Optional[str] = None, due_date=None,
labels: Optional[list] = None) -> dict:
board = self._board(board_id, person)
board.require_edit(person)
target = board.list_named(list_name) if list_name else board.lists[0]
if assignee == "me":
assignee = person.email
card = Card(id="", board_id=board.id, number=0, title="", list_name=target)
card.edit(title=title, description=description,
assignee=self._check_assignee(board, assignee), due_date=parse_due(due_date),
labels=clean_labels(labels))
card.position = self._end_of(board, target)
number = self.repo.max_number(board.id) + 1
for _ in range(50): # card numbers are claimed by a plain INSERT: one writer wins
card.number, card.id = number, f"{board.id}-{number}"
try:
self.repo.insert_card(card)
break
except Taken:
number += 1
else:
raise Conflict("the board is very busy; try again")
self.repo.log(board.id, card.id, person.email, f"added card {card.title!r} to {target}")
return self._card_out(card, board)
def _end_of(self, board: Board, list_name: str) -> float:
positions = [c.position for c in self.repo.cards_on(board.id)
if c.list_name == list_name and not c.archived]
return (max(positions) + 1.0) if positions else 1.0
def update_card(self, person: Person, card_ref: str, version: Optional[int] = None,
**fields) -> dict:
"""Change any of title, description, assignee, due_date, labels."""
board, card = self._editable(person, card_ref, version)
if "assignee" in fields:
a = fields["assignee"]
fields["assignee"] = self._check_assignee(board, person.email if a == "me" else a)
if "due_date" in fields:
fields["due_date"] = parse_due(fields["due_date"])
return self._write(person, board, card, card.version, "edited the card", **fields)
def assign_card(self, person: Person, card_ref: str, email: Optional[str]) -> dict:
board, card = self._editable(person, card_ref, None)
who = self._check_assignee(board, person.email if email == "me" else email)
return self._write(person, board, card, card.version,
f"assigned to {who}" if who else "unassigned", assignee=who)
def move_card(self, person: Person, card_ref: str, list_name: str,
before: Optional[str] = None, version: Optional[int] = None) -> dict:
"""Move a card to a list; with `before`, place it just above that card (drag and drop)."""
board, card = self._editable(person, card_ref, version)
target = board.list_named(list_name)
others = sorted((c for c in self.repo.cards_on(board.id)
if c.list_name == target and not c.archived and c.id != card.id),
key=lambda c: c.position)
if before:
ids = [c.id for c in others]
ref = before.strip().upper()
if ref not in ids:
raise Invalid(f"{before} is not in {target}")
i = ids.index(ref)
above = others[i - 1].position if i > 0 else others[i].position - 2.0
position = (above + others[i].position) / 2.0
else:
position = (others[-1].position + 1.0) if others else 1.0
before_list = card.list_name
action = (f"moved from {before_list} to {target}" if before_list != target
else f"reordered in {target}")
return self._write(person, board, card, card.version, action, list_name=target,
position=position)
def archive_card(self, person: Person, card_ref: str, archived: bool = True) -> dict:
board, card = self._editable(person, card_ref, None)
return self._write(person, board, card, card.version,
"archived the card" if archived else "restored the card",
archived=archived)
def block_card(self, person: Person, card_ref: str, blocked_by: str) -> dict:
"""Record that card_ref can't finish before blocked_by does."""
board, card = self._editable(person, card_ref, None)
_, blocker = self._resolve(person, blocked_by)
by_id = {c.id: c for c in self.repo.cards_on(board.id)}
check_dependency(card, blocker, by_id)
if blocker.id in card.blocked_by:
return self._card_out(card, board, by_id)
return self._write(person, board, card, card.version, f"blocked by {blocker.id}",
blocked_by=card.blocked_by + [blocker.id])
def unblock_card(self, person: Person, card_ref: str, blocked_by: str) -> dict:
board, card = self._editable(person, card_ref, None)
ref = blocked_by.strip().upper()
if ref not in card.blocked_by:
raise Invalid(f"{card.id} is not blocked by {blocked_by}")
return self._write(person, board, card, card.version, f"no longer blocked by {ref}",
blocked_by=[x for x in card.blocked_by if x != ref])
# --- AI (Dodil Models): suggestions only, they never change the board ----------------
def _ai(self) -> Assistant:
if self.assistant is None:
raise Invalid("no AI assistant configured")
return self.assistant
def suggest_checklist(self, person: Person, card_ref: str) -> dict:
"""An AI suggestion only: it changes nothing. A person decides what to keep."""
_, card = self._resolve(person, card_ref)
items = self._ai().checklist(card.title, card.description)
if not items:
raise Invalid("the AI gave no suggestion this time; try again")
return {"card_id": card.id, "suggested_checklist": items}
def apply_checklist(self, person: Person, card_ref: str, items: list) -> dict:
"""Write the checklist a person chose into the card's description."""
board, card = self._editable(person, card_ref, None)
items = [str(i).strip() for i in items or [] if str(i).strip()][:20]
if not items:
raise Invalid("no checklist items")
text = (card.description + "\n\n" if card.description else "") + \
"\n".join(f"- [ ] {i}" for i in items)
return self._write(person, board, card, card.version, "added a checklist",
description=text)
def similar_cards(self, person: Person, card_ref: str, limit: int = 5) -> dict:
"""Cards on the same board that mean the same thing (embeddings, not keywords)."""
board, card = self._resolve(person, card_ref)
others = [c for c in self.repo.cards_on(board.id) if c.id != card.id and not c.archived]
if not others:
return {"card": self._brief(card), "similar": []}
vecs = self._embed([card] + others)
q = vecs[0]
scored = sorted(((self._cos(q, v), c) for v, c in zip(vecs[1:], others)),
key=lambda t: -t[0])
return {"card": self._brief(card),
"similar": [{**self._brief(c), "score": round(s, 3)}
for s, c in scored[:max(1, min(int(limit), 20))]]}
def _embed(self, cards: list) -> list:
keys, missing = [], []
for c in cards:
text = f"{c.title}\n{c.description}\n{' '.join(c.labels)}".strip()
k = (c.id, hashlib.sha1(text.encode()).hexdigest())
keys.append(k)
if k not in self._vectors:
missing.append((k, text))
if missing:
for (k, _), v in zip(missing, self._ai().embed([t for _, t in missing])):
self._vectors[k] = v
return [self._vectors[k] for k in keys]
@staticmethod
def _cos(a: list, b: list) -> float:
dot = sum(x * y for x, y in zip(a, b))
na, nb = math.sqrt(sum(x * x for x in a)), math.sqrt(sum(y * y for y in b))
return dot / (na * nb) if na and nb else 0.0
def board_summary(self, person: Person, board_id: str) -> dict:
"""A short daily summary: what moved in the last day, what is late, what is stuck."""
board = self._board(board_id, person)
cards = [c for c in self.repo.cards_on(board.id) if not c.archived]
by_id = {c.id: c for c in cards}
today = self.today()
since = (today - timedelta(days=1)).isoformat()
events = [e for e in self.repo.activity(board.id, None, limit=200)
if (e.get("at") or "") >= since]
late = [c for c in cards if c.is_overdue(today, board.done_list)]
stuck = [c for c in cards if c.list_name != board.done_list
and open_blockers(c, by_id, board.done_list)]
counts = {name: sum(1 for c in cards if c.list_name == name) for name in board.lists}
facts = "\n".join([
f"Board: {board.title}. Today: {today.isoformat()}.",
"Cards per list: " + ", ".join(f"{k} {v}" for k, v in counts.items()),
"Changes in the last 24 hours:",
*[f"- {e['actor']} {e['action']} ({e.get('card_id') or 'board'})" for e in events[:40]],
"Overdue:", *[f"- {c.id} {c.title} (due {c.due_date}, {c.assignee or 'nobody'})"
for c in late],
"Blocked:", *[f"- {c.id} {c.title}" for c in stuck]])
return {"board_id": board.id, "date": today.isoformat(), "counts": counts,
"overdue": [c.id for c in late], "blocked": [c.id for c in stuck],
"changes_last_24h": len(events), "summary": self._ai().summary(facts)}