damien-kaiser/taskloft
Taskloft — team task boards (owner / member / viewer), card history, my plate, overdue, blockers, AI checklists and summaries, usable from Claude and ChatGPT over MCP.
A remix template · 0 remixes · Remix damien-kaiser/taskloft
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.
References are forgiving on purpose, because people (and their AI) talk in names:
a board is its id or its title, a list is its id or its name, a card is its number (12, "#12")
or its id. Anything the person can't see resolves to "not found", never to "forbidden".
"""
import hashlib
import uuid
from datetime import date, timedelta
from typing import Callable, Optional
from boards.application.ports import Assistant, BoardRepository
from boards.domain.model import (DEFAULT_LISTS, Board, BoardList, Card, Forbidden, Invalid,
Person, Role, check_blocker, clean_labels, clean_title, parse_due)
UNSET = object() # "this field was not sent", distinct from "set it to empty"
DONE_NAMES = ("done", "finished", "shipped", "complete", "completed")
class NotFound(Exception):
pass
def _id() -> str:
return uuid.uuid4().hex[:12]
def _meaning_text(card: Card) -> str:
"""What a card is about, as embedded: its title, its labels and the start of its description."""
return " · ".join(x for x in (card.title, ", ".join(card.labels), card.description[:1000]) if x)
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
# ---------------------------------------------------------------- resolving references
def _board(self, ref: str, person: Person) -> Board:
board = self.repo.get(str(ref))
if board is None: # a title, among the boards this person is on
key = " ".join(str(ref or "").split()).lower()
matches = [b for b in self.repo.boards_for(person.email) if b.title.lower() == key
or b.title.lower() == key.removesuffix(" board")]
if len(matches) > 1:
ids = ", ".join(f"{b.id} ({b.title})" for b in matches)
raise Invalid(f"several of your boards match {ref!r}: {ids}; use the board id")
board = matches[0] if matches else None
if board is None or board.role_of(person) is None:
raise NotFound("no such board") # a board you can't see doesn't exist for you
return board
def _card(self, ref, person: Person) -> tuple:
s = str(ref).strip().lstrip("#")
card = self.repo.card_by_number(int(s)) if s.isdigit() else self.repo.get_card(s)
if card is None:
raise NotFound("no such card")
board = self.repo.get(card.board_id)
if board is None or board.role_of(person) is None:
raise NotFound("no such card")
return board, card
def _waiting_on(self, board: Board, card: Card, index: Optional[dict] = None) -> list:
"""Numbers of the cards this one waits on that are NOT finished yet (a done blocker stops blocking)."""
done, out = board.done_list_ids(), []
for bid in card.blocked_by:
c = index.get(bid) if index is not None else self.repo.get_card(bid)
if c is not None and c.list_id not in done:
out.append(c.number)
return sorted(out)
def _view(self, board: Board, card: Card, with_extras: bool = False, index: Optional[dict] = None) -> dict:
lst = next((lst for lst in board.lists if lst.id == card.list_id), None)
d = {"id": card.id, "number": card.number, "board_id": board.id, "board": board.title,
"title": card.title, "description": card.description, "list": card.list_name,
"list_id": card.list_id, "assignee": card.assignee,
"due_date": card.due_date.isoformat() if card.due_date else None,
"labels": card.labels, "position": card.position,
"done": bool(lst and lst.is_done),
"overdue": card.is_overdue(self.today(), board.done_list_ids()),
"blocked_by": card.blocked_by, "waiting_on": self._waiting_on(board, card, index)}
if with_extras:
d["checklist"] = self.repo.checklist(card.id)
d["history"] = self.repo.history(board.id, card.id)
return d
def _log(self, board: Board, card: Optional[Card], person: Person, action: str,
changes: Optional[dict] = None) -> None:
self.repo.log(board.id, card.id if card else None, person.email, action, changes)
# ---------------------------------------------------------------- boards
def create_board(self, person: Person, title: str, lists: Optional[list] = None) -> dict:
board = Board(id=_id(), title=clean_title(title, "a board"), owner_email=person.email)
names = [n for n in (lists or DEFAULT_LISTS) if str(n).strip()] or list(DEFAULT_LISTS)
for i, name in enumerate(names):
n = board.check_new_list_name(name)
board.lists.append(BoardList(_id(), board.id, n, i, is_done=n.lower() in DONE_NAMES))
self.repo.save_board(board)
self._log(board, None, person, f"created board {board.title!r}")
return {"id": board.id, "title": board.title, "lists": [x.name for x in board.lists]}
def my_boards(self, person: Person) -> list:
return [{"id": b.id, "title": b.title, "role": b.role_of(person).value,
"people": 1 + len(b.members)}
for b in sorted(self.repo.boards_for(person.email), key=lambda b: b.title.lower())]
def show_board(self, person: Person, board_ref: str) -> dict:
board = self._board(board_ref, person)
cards = self.repo.cards_on(board.id)
index = {c.id: c for c in cards}
lists = []
for lst in board.lists:
mine = sorted((c for c in cards if c.list_id == lst.id), key=lambda c: (c.position, c.number))
lists.append({"id": lst.id, "name": lst.name, "done": lst.is_done,
"cards": [self._view(board, c, index=index) for c in mine]})
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": lists}
def rename_board(self, person: Person, board_ref: str, title: str) -> dict:
board = self._board(board_ref, person)
board.require_edit(person)
before, board.title = board.title, clean_title(title, "a board")
self.repo.save_board(board)
self._log(board, None, person, "renamed the board", {"title": [before, board.title]})
return {"id": board.id, "title": board.title}
# ---------------------------------------------------------------- people
def members(self, person: Person, board_ref: str) -> dict:
board = self._board(board_ref, person)
return {"owner": board.owner_email,
"members": [{"email": e, "role": r.value} for e, r in sorted(board.members.items())]}
def invite_member(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'")
before = board.members.get(email.strip().lower())
email = board.invite(person, email, r)
self.repo.save_board(board)
verb = f"changed {email} from {before.value} to {r.value}" if before else f"invited {email} as {r.value}"
self._log(board, None, person, verb)
return {"board_id": board.id, "email": email, "role": r.value}
add_member = invite_member # the starter's name, kept so old clients still work
def remove_member(self, person: Person, board_ref: str, email: str) -> dict:
board = self._board(board_ref, person)
email = board.remove(person, email)
self.repo.remove_member(board.id, email)
for card in self.repo.cards_on(board.id): # their cards go back to nobody
if card.assignee == email:
card.edit(assignee=None)
self.repo.save_card(card)
self._log(board, card, person, "unassigned (left the board)", {"assignee": [email, None]})
self._log(board, None, person, f"removed {email}")
return {"board_id": board.id, "removed": email}
# ---------------------------------------------------------------- lists
def add_list(self, person: Person, board_ref: str, name: str, done: bool = False) -> dict:
board = self._board(board_ref, person)
board.require_edit(person)
lst = BoardList(_id(), board.id, board.check_new_list_name(name), len(board.lists), bool(done))
board.lists.append(lst)
self.repo.save_board(board)
self._log(board, None, person, f"added list {lst.name!r}")
return {"id": lst.id, "name": lst.name, "done": lst.is_done}
def rename_list(self, person: Person, board_ref: str, list_ref: str, name: str) -> dict:
board = self._board(board_ref, person)
board.require_edit(person)
lst = board.find_list(list_ref)
new = board.check_new_list_name(name) if name.strip().lower() != lst.name.lower() else lst.name
before, lst.name = lst.name, new
self.repo.save_board(board)
for card in self.repo.cards_on(board.id):
if card.list_id == lst.id:
card.list_name = new
self.repo.save_card(card)
self._log(board, None, person, f"renamed list {before!r} to {new!r}")
return {"id": lst.id, "name": lst.name}
# ---------------------------------------------------------------- cards
def add_card(self, person: Person, board_ref: str, title: str, list_ref: Optional[str] = None,
description: str = "", assignee: Optional[str] = None, due_date=None,
labels=None) -> dict:
board = self._board(board_ref, person)
board.require_edit(person)
lst = board.find_list(list_ref) if list_ref else board.lists[0]
who = self._assignee(board, assignee)
cid = _id()
siblings = [c.position for c in self.repo.cards_on(board.id) if c.list_id == lst.id]
card = Card(id=cid, board_id=board.id, number=0, title=clean_title(title),
list_id=lst.id, list_name=lst.name, description=(description or "").strip(),
assignee=who, due_date=parse_due(due_date), labels=clean_labels(labels),
position=(max(siblings) + 1.0) if siblings else 1.0)
card.number = self.repo.claim_number(cid)
self.repo.save_card(card)
self._log(board, card, person, f"added card #{card.number} {card.title!r} to {lst.name}",
{k: v for k, v in {"assignee": who, "due_date": card.due_date and card.due_date.isoformat(),
"labels": card.labels or None}.items() if v})
return self._view(board, card)
def card(self, person: Person, card_ref) -> dict:
board, card = self._card(card_ref, person)
return self._view(board, card, with_extras=True)
def update_card(self, person: Person, card_ref, title=UNSET, description=UNSET, due_date=UNSET,
labels=UNSET) -> dict:
board, card = self._card(card_ref, person)
board.require_edit(person)
changes = {}
if title is not UNSET and title is not None:
changes["title"] = clean_title(title)
if description is not UNSET and description is not None:
changes["description"] = str(description).strip()
if due_date is not UNSET:
changes["due_date"] = parse_due(due_date)
if labels is not UNSET and labels is not None:
changes["labels"] = clean_labels(labels)
diff = card.edit(**changes)
if diff:
self.repo.save_card(card)
self._log(board, card, person, "edited " + ", ".join(diff), diff)
return self._view(board, card)
def _assignee(self, board: Board, email: Optional[str]) -> Optional[str]:
if email in (None, "", "nobody", "none"):
return None
e = str(email).strip().lower()
if e in ("me", "myself"):
raise Invalid("say who: use their email")
if not board.can_be_assigned(e):
raise Invalid(f"{e} is not a member of {board.title!r} (viewers can't take cards)")
return e
def assign_card(self, person: Person, card_ref, email: Optional[str]) -> dict:
board, card = self._card(card_ref, person)
board.require_edit(person)
e = person.email if str(email or "").strip().lower() in ("me", "myself") else email
diff = card.edit(assignee=self._assignee(board, e))
if diff:
self.repo.save_card(card)
self._log(board, card, person, f"assigned to {card.assignee}" if card.assignee else "unassigned", diff)
return self._view(board, card)
def move_card(self, person: Person, card_ref, list_ref: str, before_ref=None) -> dict:
"""Move a card to a list; with before_ref, just above that card (drag and drop order)."""
board, card = self._card(card_ref, person)
board.require_edit(person)
lst = board.find_list(list_ref)
others = sorted((c for c in self.repo.cards_on(board.id) if c.list_id == lst.id and c.id != card.id),
key=lambda c: (c.position, c.number))
if before_ref not in (None, ""):
_, anchor = self._card(before_ref, person)
idx = next((i for i, c in enumerate(others) if c.id == anchor.id), None)
if idx is None:
raise Invalid(f"#{anchor.number} is not in {lst.name}")
lo = others[idx - 1].position if idx > 0 else others[idx].position - 1.0
pos = (lo + others[idx].position) / 2
else:
pos = (others[-1].position + 1.0) if others else 1.0
before = card.list_name
diff = card.edit(list_id=lst.id, list_name=lst.name, position=pos)
if diff:
self.repo.save_card(card)
if before != lst.name:
self._log(board, card, person, f"moved from {before} to {lst.name}", {"list": [before, lst.name]})
else:
self._log(board, card, person, f"reordered in {lst.name}")
return self._view(board, card)
# ---------------------------------------------------------------- what needs me
def my_plate(self, person: Person, days: int = 7) -> dict:
"""Cards assigned to me, not finished: overdue first, then due within `days`, then undated."""
today, horizon = self.today(), self.today() + timedelta(days=max(0, int(days)))
overdue, soon, later = [], [], []
for board in self.repo.boards_for(person.email):
done, cards = board.done_list_ids(), self.repo.cards_on(board.id)
index = {c.id: c for c in cards}
for c in cards:
if c.assignee != person.email or c.list_id in done:
continue
v = self._view(board, c, index=index)
if c.due_date and c.due_date < today:
overdue.append(v)
elif c.due_date and c.due_date <= horizon:
soon.append(v)
elif not c.due_date:
later.append(v)
key = lambda v: (v["due_date"] or "9999", v["number"])
return {"today": today.isoformat(), "until": horizon.isoformat(),
"overdue": sorted(overdue, key=key), "due_soon": sorted(soon, key=key),
"no_due_date": sorted(later, key=lambda v: v["number"])}
def overdue(self, person: Person, board_ref: Optional[str] = None) -> list:
boards = [self._board(board_ref, person)] if board_ref else self.repo.boards_for(person.email)
out = []
for board in boards:
done, cards = board.done_list_ids(), self.repo.cards_on(board.id)
index = {c.id: c for c in cards}
out += [self._view(board, c, index=index) for c in cards if c.is_overdue(self.today(), done)]
return sorted(out, key=lambda v: (v["due_date"], v["number"]))
def search_cards(self, person: Person, text: str, board_ref: Optional[str] = None) -> list:
words = [w for w in str(text or "").lower().split() if w]
if not words:
raise Invalid("say what to look for")
boards = [self._board(board_ref, person)] if board_ref else self.repo.boards_for(person.email)
out = []
for board in boards:
cards = self.repo.cards_on(board.id)
index = {c.id: c for c in cards}
for c in cards:
hay = " ".join([c.title, c.description, " ".join(c.labels), c.assignee or ""]).lower()
if all(w in hay for w in words):
out.append(self._view(board, c, index=index))
return sorted(out, key=lambda v: v["number"])
# ---------------------------------------------------------------- history
def card_history(self, person: Person, card_ref) -> list:
board, card = self._card(card_ref, person)
return self.repo.history(board.id, card.id)
def board_activity(self, person: Person, board_ref: str, limit: int = 30) -> list:
board = self._board(board_ref, person)
return self.repo.history(board.id, None, max(1, min(int(limit), 200)))
# ---------------------------------------------------------------- blockers
def set_blocked_by(self, person: Person, card_ref, blocker_ref, blocked: bool = True) -> dict:
board, card = self._card(card_ref, person)
board.require_edit(person)
_, blocker = self._card(blocker_ref, person)
if blocked:
def blocked_by_of(cid):
c = self.repo.get_card(cid)
return c.blocked_by if c else []
check_blocker(card, blocker, blocked_by_of)
if blocker.id not in card.blocked_by:
self.repo.set_blocker(card.id, blocker.id, True)
card.blocked_by.append(blocker.id)
self._log(board, card, person, f"blocked by #{blocker.number}", {"blocked_by": [None, blocker.number]})
elif blocker.id in card.blocked_by:
self.repo.set_blocker(card.id, blocker.id, False)
card.blocked_by.remove(blocker.id)
self._log(board, card, person, f"no longer blocked by #{blocker.number}", {"blocked_by": [blocker.number, None]})
return self._view(board, card)
def what_is_blocking(self, person: Person, card_ref) -> dict:
"""Everything still unfinished that this card waits on, directly or further up the chain."""
board, card = self._card(card_ref, person)
done = board.done_list_ids()
seen, chain, todo = set(), [], [(cid, 1) for cid in card.blocked_by]
while todo:
cid, depth = todo.pop(0)
if cid in seen:
continue
seen.add(cid)
c = self.repo.get_card(cid)
if c is None:
continue
if c.list_id not in done:
chain.append({**self._view(board, c), "depth": depth})
todo += [(x, depth + 1) for x in c.blocked_by]
return {"card": self._view(board, card), "blocking": chain,
"ready": not chain}
# ---------------------------------------------------------------- checklist
def add_checklist_items(self, person: Person, card_ref, items: list) -> dict:
board, card = self._card(card_ref, person)
board.require_edit(person)
existing = self.repo.checklist(card.id)
start = len(existing)
added = []
for i, text in enumerate(t for t in (" ".join(str(x).split()) for x in items) if t):
item = {"id": _id(), "card_id": card.id, "text": text[:200], "done": False, "position": start + i}
self.repo.save_checklist_item(item)
added.append(item["text"])
if added:
self._log(board, card, person, f"added {len(added)} checklist item(s)", {"checklist": [None, added]})
return {"card": card.number, "checklist": self.repo.checklist(card.id)}
def tick_checklist_item(self, person: Person, card_ref, item_ref, done: bool = True) -> dict:
board, card = self._card(card_ref, person)
board.require_edit(person)
items = self.repo.checklist(card.id)
ref = str(item_ref).strip()
item = next((x for x in items if x["id"] == ref), None)
if item is None and ref.isdigit() and 1 <= int(ref) <= len(items):
item = items[int(ref) - 1] # "tick item 2"
if item is None:
raise Invalid("no such checklist item")
if bool(item["done"]) != bool(done):
item["done"] = bool(done)
self.repo.save_checklist_item(item)
self._log(board, card, person, ("ticked " if done else "unticked ") + repr(item["text"]))
return {"card": card.number, "checklist": self.repo.checklist(card.id)}
# ---------------------------------------------------------------- AI (Dodil Models)
def suggest_checklist(self, person: Person, card_ref) -> dict:
"""An AI suggestion only: it changes nothing. A person decides what to keep."""
_, card = self._card(card_ref, person)
if self.assistant is None:
raise Invalid("no AI assistant configured")
return {"card": card.number, "suggested_checklist": self.assistant.checklist(card.title, card.description)}
def make_checklist(self, person: Person, card_ref) -> dict:
"""'Turn this card into a checklist': the AI drafts the steps and they are added to the card."""
board, card = self._card(card_ref, person)
board.require_edit(person)
steps = self.suggest_checklist(person, card_ref)["suggested_checklist"]
return self.add_checklist_items(person, card.id, steps)
def board_summary(self, person: Person, board_ref: str) -> dict:
"""A short daily summary of a board, written by Dodil Models from the board's own facts."""
board = self._board(board_ref, person)
if self.assistant is None:
raise Invalid("no AI assistant configured")
view = self.show_board(person, board.id)
facts = [f"Today is {self.today().isoformat()}."]
for lst in view["lists"]:
facts.append(f"List {lst['name']}{' (finished)' if lst['done'] else ''}: {len(lst['cards'])} card(s).")
for c in lst["cards"]:
bits = [f"#{c['number']} {c['title']}"]
if c["assignee"]:
bits.append(f"assigned to {c['assignee']}")
if c["due_date"]:
bits.append(f"due {c['due_date']}" + (" OVERDUE" if c["overdue"] else ""))
if c["waiting_on"]:
bits.append("waiting on " + ", ".join(f"#{n}" for n in c["waiting_on"]))
facts.append(" - " + ", ".join(bits))
recent = self.repo.history(board.id, None, 10)
facts += [f"Recent: {h['actor']} {h['action']} ({h['at']})" for h in recent]
# One summary per board and per state of the board: the same facts on the same day never wait twice.
key = board.id + ":" + hashlib.sha1("\n".join(facts).encode()).hexdigest()[:20]
cached = self.repo.cached_summary(key)
if cached is not None:
return {"board": board.title, "summary": cached, "cached": True}
text = self.assistant.summary(board.title, "\n".join(facts))
self.repo.save_summary(key, board.id, text)
return {"board": board.title, "summary": text, "cached": False}
# ---------------------------------------------------------------- meaning (Dodil Models embeddings)
def _fresh_vectors(self, boards: list) -> dict:
"""Embed every card whose text changed since it was last embedded (in batches), on these boards only."""
known = self.repo.vector_hashes([b.id for b in boards])
todo = []
for board in boards:
for c in self.repo.cards_on(board.id):
text = _meaning_text(c)
h = hashlib.sha1(text.encode()).hexdigest()[:16]
if known.get(c.id) != h:
todo.append((c, text, h))
for i in range(0, len(todo), 32):
chunk = todo[i:i + 32]
for (c, _, h), vec in zip(chunk, self.assistant.embed([t for _, t, _ in chunk])):
self.repo.save_vector(c.id, c.board_id, h, vec)
return {b.id: b for b in boards}
def _ranked(self, person: Person, boards: dict, hits: list) -> list:
out = []
for cid, dist in hits:
c = self.repo.get_card(cid)
if c is None or c.board_id not in boards: # only what this person can see
continue
out.append({**self._view(boards[c.board_id], c), "similarity": round(1.0 - float(dist), 3)})
return out
def similar_cards(self, person: Person, card_ref, k: int = 5) -> list:
"""'Find cards like this one': nearest cards by meaning, on the boards this person can see."""
if self.assistant is None:
raise Invalid("no AI assistant configured")
_, card = self._card(card_ref, person)
boards = self._fresh_vectors(self.repo.boards_for(person.email))
vec = self.assistant.embed([_meaning_text(card)])[0]
return self._ranked(person, boards, self.repo.nearest(list(boards), vec, max(1, min(int(k), 20)), card.id))
def search_by_meaning(self, person: Person, text: str, board_ref: Optional[str] = None, k: int = 8) -> list:
"""Cards about this, even when they share no word with it ('bill the client' finds 'Send invoice')."""
if self.assistant is None:
raise Invalid("no AI assistant configured")
if not str(text or "").strip():
raise Invalid("say what to look for")
visible = [self._board(board_ref, person)] if board_ref else self.repo.boards_for(person.email)
boards = self._fresh_vectors(visible)
vec = self.assistant.embed([str(text).strip()[:1000]])[0]
return self._ranked(person, boards, self.repo.nearest(list(boards), vec, max(1, min(int(k), 20))))