sirenamc/stackline
Stackline, a team task board you run from the web or from Claude/ChatGPT. FastAPI + DataK3 + appid + Models + MCP.
A remix template · 0 remixes · Remix sirenamc/stackline
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 Callable, Optional
from boards.application.ports import Assistant, BoardRepository, Event
from boards.domain.model import (DEFAULT_LISTS, Board, Card, Forbidden, Invalid, Person, Role,
check_blocker, clean_labels, clean_text, edit_card,
normalize_email)
class NotFound(Exception):
pass
def _new_id() -> str:
return uuid.uuid4().hex[:12]
def parse_number(ref) -> int:
"""'12', '#12' and 12 all mean card 12."""
try:
return int(str(ref).strip().lstrip("#"))
except ValueError:
raise Invalid(f"{ref!r} is not a card number; cards are numbered like #12")
def parse_date(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"{value!r} is not a date; use YYYY-MM-DD")
def _text(card: Card) -> str:
return f"{card.title}\n{card.description}".strip()
def _hash(text: str) -> str:
return hashlib.sha1(text.encode()).hexdigest()[:16]
def _cosine(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
# The example board a new team starts from: what launching a small website really involves.
# (title, list, days from today for the due date, numbers of earlier steps it waits on)
EXAMPLE_STEPS = (
("Agree the pages and who writes each", "Done", -9, ()),
("Collect logos and photos", "Done", -6, ()),
("Write the home page copy", "Doing", -2, (1,)),
("Design the home page", "Doing", 2, (2, 3)),
("Set up the domain and email", "To do", -1, ()),
("Build the pages", "To do", 6, (4,)),
("Proofread every page", "To do", 8, (3, 6)),
("Test on phones", "To do", 9, (6,)),
("Launch the site", "To do", 11, (5, 7, 8)),
)
MAX_PLAN = 15
def _whole(value, i: int, field: str, limit: int) -> int:
"""A whole number from a plan step, or a clear refusal (never a crash on "soon" or 1e9)."""
try:
n = int(str(value).strip())
except ValueError:
raise Invalid(f"step {i + 1}: {field} must be a whole number")
if abs(n) > limit:
raise Invalid(f"step {i + 1}: {field} is out of range")
return n
class Boards:
def __init__(self, repo: BoardRepository, assistant: Optional[Assistant] = None,
today: Callable[[], date] = date.today, demo_accounts: tuple = ()):
self.repo, self.assistant, self.today = repo, assistant, today
# (owner, member, viewer) emails of the public demo, if this copy runs one
self.demo = tuple(normalize_email(e) for e in demo_accounts) if demo_accounts else ()
# ---- finding things the person is allowed to see -------------------------------------
def _board(self, ref: str, person: Person) -> Board:
"""A board by id or by title. A board you can't see doesn't exist for you."""
board = self.repo.get(ref) if ref else None
if board is None and ref:
key = ref.strip().lower()
matches = [b for b in self.repo.boards_for(person.email) if b.title.lower() == key]
if len(matches) > 1:
raise Invalid(f"more than one board is called {ref!r}; use its id: "
+ ", ".join(b.id for b in matches))
board = matches[0] if matches else None
if board is None or board.role_of(person) is None:
raise NotFound(f"no board {ref!r} that you can see")
return board
def _card(self, ref, person: Person) -> tuple:
card = self.repo.get_card(parse_number(ref))
if card is None:
raise NotFound(f"no card {ref}")
try:
board = self._board(card.board_id, person)
except NotFound:
raise NotFound(f"no card {ref}")
return board, card
def _visible_boards(self, person: Person, ref: Optional[str] = None) -> list:
return [self._board(ref, person)] if ref else self.repo.boards_for(person.email)
def _done(self, board: Board, card: Card) -> bool:
return any(lst.id == card.list_id and lst.is_done for lst in board.lists)
def _view(self, board: Board, card: Card, open_blockers: Optional[list] = None) -> dict:
lst = next((x for x in board.lists if x.id == card.list_id), None)
done = bool(lst and lst.is_done)
view = {"id": card.id, "number": card.number, "board_id": board.id,
"board": board.title, "list": lst.name if lst else None, "list_id": card.list_id,
"title": card.title, "description": card.description,
"assignee": card.assignee,
"due_date": card.due_date.isoformat() if card.due_date else None,
"labels": list(card.labels), "blocked_by": list(card.blocked_by),
"checklist": [{"text": t, "done": d} for t, d in card.checklist],
"done": done, "overdue": card.is_overdue(self.today(), done)}
if open_blockers is not None:
view["open_blockers"] = open_blockers
return view
def _open_blockers(self, board: Board, card: Card, cards_by_number: dict) -> list:
return [n for n in card.blocked_by
if n in cards_by_number and not self._done(board, cards_by_number[n])]
def _log(self, board: Board, card: Optional[Card], person: Person, action: str,
field: Optional[str] = None, before=None, after=None) -> None:
def s(v):
if v is None or v == ():
return None
return ", ".join(v) if isinstance(v, tuple) else str(v)
self.repo.log(Event(board.id, card.number if card else None, person.email, action,
field, s(before), s(after)))
# ---- boards, lists and people ---------------------------------------------------------
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 create_board(self, person: Person, title: str) -> dict:
title = clean_text(title, "a board needs a title", 80)
board = Board(id=_new_id(), title=title, owner_email=person.email)
for name in DEFAULT_LISTS:
board.add_list(_new_id(), name)
board.lists[-1].is_done = True
self.repo.save_board(board)
for lst in board.lists:
self.repo.save_list(board.id, lst)
self._log(board, None, person, "created board", after=title)
return {"id": board.id, "title": board.title, "lists": [x.name for x in board.lists]}
def show_board(self, person: Person, board_ref: str) -> dict:
board = self._board(board_ref, person)
cards = self.repo.cards_on(board.id)
by_number = {c.number: c for c in cards}
lists = []
for lst in board.lists:
here = 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, "is_done": lst.is_done,
"cards": [self._view(board, c, self._open_blockers(board, c, by_number))
for c in here]})
return {"id": board.id, "title": board.title, "role": board.role_of(person).value,
"lists": lists, "members": self._members(board)}
def _members(self, board: Board) -> list:
people = [{"email": board.owner_email, "role": "owner"}]
people += [{"email": e, "role": r.value} for e, r in sorted(board.members.items())]
return people
def members(self, person: Person, board_ref: str) -> list:
return self._members(self._board(board_ref, person))
def invite_member(self, person: Person, board_ref: str, email: str,
role: str = "member") -> dict:
"""Add someone to a board, or change their role. Owner only."""
board = self._board(board_ref, person)
try:
new_role = Role(role)
except ValueError:
raise Invalid("role must be 'member' or 'viewer'")
email = normalize_email(email)
before = board.members.get(email)
board.set_member(person, email, new_role)
self.repo.save_membership(board.id, email, new_role)
if before is None:
self._log(board, None, person, "member added", "member", None, f"{email} ({role})")
elif before is not new_role:
self._log(board, None, person, "role changed", email, before.value, role)
return {"board_id": board.id, "email": email, "role": role}
def remove_member(self, person: Person, board_ref: str, email: str) -> dict:
board = self._board(board_ref, person)
board.remove_member(person, email)
email = normalize_email(email)
self.repo.remove_membership(board.id, email)
self._log(board, None, person, "member removed", "member", email, None)
for card in self.repo.cards_on(board.id):
if card.assignee == email: # their cards go back to nobody, and the log says so
card.assignee = None
self.repo.save_card(card)
self._log(board, card, person, "unassigned", "assignee", email, None)
return {"board_id": board.id, "removed": email}
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)
lst = board.add_list(_new_id(), name)
lst.is_done = bool(is_done)
self.repo.save_list(board.id, lst)
self._log(board, None, person, "list added", "list", None, lst.name)
return {"id": lst.id, "name": lst.name, "is_done": lst.is_done}
def rename_list(self, person: Person, board_ref: str, list_ref: str, new_name: str) -> dict:
board = self._board(board_ref, person)
board.require_edit(person)
before = board.find_list(list_ref).name
lst = board.rename_list(list_ref, new_name)
self.repo.save_list(board.id, lst)
self._log(board, None, person, "list renamed", "list", before, lst.name)
return {"id": lst.id, "name": lst.name}
# ---- cards ----------------------------------------------------------------------------
def add_card(self, person: Person, board_ref: str, title: str, list_name: str = "",
description: str = "", assignee: Optional[str] = None,
due_date: Optional[str] = None, labels=()) -> dict:
board = self._board(board_ref, person)
board.require_edit(person)
lst = board.find_list(list_name) if list_name else board.first_list()
card = Card(number=0, board_id=board.id, list_id=lst.id,
title=clean_text(title, "a card needs a title", 200),
description=(description or "").strip()[:4000],
assignee=self._assignee(board, person, assignee),
due_date=parse_date(due_date), labels=clean_labels(labels),
position=self._end_of(board.id, lst.id))
card = self.repo.insert_card(card)
self._log(board, card, person, "created", "list", None, lst.name)
return self._view(board, card)
def _assignee(self, board: Board, person: Person, who: Optional[str]) -> Optional[str]:
if who in (None, ""):
return None
email = person.email if who.strip().lower() == "me" else normalize_email(who)
if not board.is_on_board(email):
raise Invalid(f"{email} is not on this board; the owner can invite them first")
return email
def _end_of(self, board_id: str, list_id: str) -> float:
here = [c.position for c in self.repo.cards_on(board_id) if c.list_id == list_id]
return (max(here) + 1.0) if here else 1.0
def card_details(self, person: Person, card_ref) -> dict:
board, card = self._card(card_ref, person)
by_number = {c.number: c for c in self.repo.cards_on(board.id)}
view = self._view(board, card, self._open_blockers(board, card, by_number))
view["history"] = [self._event(e) for e in self.repo.history(board.id, card.number)]
return view
def update_card(self, person: Person, card_ref, **changes) -> dict:
"""Change title, description, due date or labels. Only real changes are logged."""
board, card = self._card(card_ref, person)
board.require_edit(person)
changes = {k: v for k, v in changes.items() if v is not None}
if "due_date" in changes:
changes["due_date"] = parse_date(changes["due_date"])
new, diff = edit_card(card, changes)
if diff:
self.repo.save_card(new)
for field, before, after in diff:
self._log(board, new, person, "edited", field, before, after)
return self._view(board, new)
def assign_card(self, person: Person, card_ref, assignee: Optional[str]) -> dict:
"""Assign to a board member ('me' works), or pass nothing to unassign."""
board, card = self._card(card_ref, person)
board.require_edit(person)
email = self._assignee(board, person, assignee)
if email != card.assignee:
before, card.assignee = card.assignee, email
self.repo.save_card(card)
self._log(board, card, person, "assigned" if email else "unassigned",
"assignee", before, email)
return self._view(board, card)
def move_card(self, person: Person, card_ref, list_name: str,
position: Optional[int] = None) -> dict:
"""Move a card to a list (by name or id). position is its place in that list, from 0;
leave it out to put the card at the bottom."""
board, card = self._card(card_ref, person)
board.require_edit(person)
target = board.find_list(list_name)
before = next((x.name for x in board.lists if x.id == card.list_id), None)
others = sorted((c for c in self.repo.cards_on(board.id)
if c.list_id == target.id and c.number != card.number),
key=lambda c: (c.position, c.number))
if position is None or position >= len(others):
card.position = (others[-1].position + 1.0) if others else 1.0
else:
i = max(0, int(position))
after = others[i].position
prev = others[i - 1].position if i > 0 else after - 2.0
card.position = (prev + after) / 2
card.list_id = target.id
self.repo.save_card(card)
if before != target.name:
self._log(board, card, person, "moved", "list", before, target.name)
return self._view(board, card)
def delete_card(self, person: Person, card_ref) -> dict:
board, card = self._card(card_ref, person)
board.require_edit(person)
self.repo.delete_card(card.number)
self._log(board, card, person, "deleted", "title", card.title, None)
for other in self.repo.cards_on(board.id):
if card.number in other.blocked_by:
other.blocked_by = tuple(n for n in other.blocked_by if n != card.number)
self.repo.save_card(other)
return {"deleted": card.number}
# ---- a whole board from one sentence ----------------------------------------------------
def set_up_board(self, person: Person, title: str, goal: str = "",
deadline: Optional[str] = None, people=(), steps=None) -> dict:
"""A ready board from a goal and a date: lists, cards with due dates counted back from
the deadline, owners and "blocked by" links, all in the log.
steps: the plan, if Claude (or a person) already wrote one, as a list of
{"title", "list"?, "due_date"? or "days_before"?, "assignee"?, "after"?: [step numbers]}
where "after" names earlier steps (1 = the first) this one waits on.
Leave steps out and Dodil Models drafts the plan from the goal.
Everything is checked before anything is written, so a bad plan changes nothing."""
title = clean_text(title, "a board needs a title", 80)
end = parse_date(deadline)
invited = [normalize_email(e) for e in (people or ()) if str(e).strip()]
if len(invited) > 20:
raise Invalid("invite at most 20 people when setting up a board")
if steps is None:
if not (goal or "").strip():
raise Invalid("say what the board is for, or give the steps")
steps = self._model_call(self._ai().plan, goal.strip(), (end or self.today()).isoformat(),
self.today().isoformat())
plan = self._check_plan(person, steps, end, set(invited))
created = self.create_board(person, title)
board = self._board(created["id"], person)
for email in invited:
if email != person.email:
board.set_member(person, email, Role.MEMBER)
self.repo.save_membership(board.id, email, Role.MEMBER)
self._log(board, None, person, "member added", "member", None,
f"{email} (member)")
# Written straight through the repository, one card at a time in plan order, so a
# step's blockers already have their numbers: a plan of 12 cards stays well inside
# Ignite's 60 s even from far away. The same rules as add_card and block_card apply:
# _check_plan has already run them.
numbers, ends = [], {}
for step in plan:
lst = board.find_list(step["list"])
ends[lst.id] = ends.get(lst.id, 0.0) + 1.0
card = self.repo.insert_card(Card(
number=0, board_id=board.id, list_id=lst.id, title=step["title"],
description=step["description"],
assignee=self._assignee(board, person, step["assignee"]),
due_date=step["due_date"], position=ends[lst.id],
blocked_by=tuple(numbers[i] for i in step["after"])))
numbers.append(card.number)
self._log(board, card, person, "created", "list", None, lst.name)
for n in card.blocked_by:
self._log(board, card, person, "blocked", "blocked_by", None, f"#{n}")
out = self.show_board(person, board.id)
out["created_cards"] = numbers
return out
def _check_plan(self, person: Person, steps, end: Optional[date], invited: set) -> list:
if not isinstance(steps, (list, tuple)) or not steps:
raise Invalid("the plan has no steps")
if len(steps) > MAX_PLAN:
raise Invalid(f"keep a plan to {MAX_PLAN} steps; split bigger work into boards")
lists = {name.lower(): name for name in DEFAULT_LISTS}
plan = []
for i, raw in enumerate(steps):
if isinstance(raw, str):
raw = {"title": raw}
if not isinstance(raw, dict):
raise Invalid(f"step {i + 1} should be an object with a title")
list_name = lists.get(str(raw.get("list") or "To do").strip().lower())
if list_name is None:
raise Invalid(f"step {i + 1}: list must be one of {', '.join(DEFAULT_LISTS)}")
due = parse_date(raw.get("due_date"))
if due is None and raw.get("days_before") not in (None, "") and end:
due = end - timedelta(days=_whole(raw["days_before"], i, "days_before", 3650))
who = raw.get("assignee") or None
if who and who.strip().lower() != "me":
who = normalize_email(who)
if who not in invited and who != person.email:
raise Invalid(f"step {i + 1}: {who} isn't in people; add them there first")
after = []
after_refs = raw.get("after") or ()
if not isinstance(after_refs, (list, tuple)):
after_refs = [after_refs]
for ref in after_refs:
j = _whole(ref, i, "after", MAX_PLAN) - 1
if not 0 <= j < i:
raise Invalid(f"step {i + 1} can only wait on earlier steps (1 to {i})")
after.append(j)
plan.append({"title": clean_text(str(raw.get("title") or ""),
f"step {i + 1} needs a title", 200),
"list": list_name, "due_date": due, "assignee": who,
"description": str(raw.get("description") or "")[:4000],
"after": after})
return plan
def start_example(self, person: Person) -> dict:
"""The example board for someone who has just signed in to an empty copy: nine cards of
a website launch, two of them late, with real blockers, so every feature has something
to show. Yours to change or delete."""
today = self.today()
steps = [{"title": t, "list": lst, "due_date": (today + timedelta(days=d)).isoformat(),
"assignee": "me", "after": list(after)}
for t, lst, d, after in EXAMPLE_STEPS]
return self.set_up_board(person, "Example: website launch", steps=steps)
def reset_demo(self, person: Person) -> dict:
"""The public demo's three accounts can put its boards back as they were, so whatever
the last visitor did, the next one starts from the same story."""
if not self.demo or person.email not in self.demo:
raise Forbidden("only the demo accounts can reset the demo")
owner_email, member_email, viewer_email = self.demo
owner, member = Person(owner_email), Person(member_email)
for b in self.repo.boards_for(owner_email):
if b.owner_email == owner_email:
self.repo.delete_board(b.id)
today = self.today()
launch = self.set_up_board(owner, "Launch", people=[member_email], steps=[
{"title": t, "list": lst, "due_date": (today + timedelta(days=d)).isoformat(),
"assignee": member_email if i % 3 == 1 else "me", "after": list(after)}
for i, (t, lst, d, after) in enumerate(EXAMPLE_STEPS)])
self.invite_member(owner, launch["id"], viewer_email, "viewer")
n = launch["created_cards"]
self.move_card(member, n[4], "Doing")
self.add_checklist(member, n[2], ["Headline", "Three reasons to call us", "Contact block"])
self.check_item(member, n[2], 0)
ops = self.set_up_board(owner, "Ops", people=[member_email], steps=[
{"title": "Send invoice to Acme", "list": "Doing", "assignee": member_email,
"due_date": (today - timedelta(days=3)).isoformat()},
{"title": "Renew the office lease", "assignee": "me",
"due_date": (today + timedelta(days=5)).isoformat()},
{"title": "Pay the hosting bill", "assignee": member_email,
"due_date": (today + timedelta(days=2)).isoformat()}])
return {"boards": [launch["title"], ops["title"]],
"accounts": {"owner": owner_email, "member": member_email,
"viewer": viewer_email}}
# ---- what needs attention -------------------------------------------------------------
def my_plate(self, person: Person, days: int = 7) -> dict:
"""Open cards assigned to you: overdue, due in the next `days` days, and undated."""
today = self.today()
horizon = today + timedelta(days=max(0, int(days)))
out = {"today": today.isoformat(), "overdue": [], "due_soon": [], "no_due_date": []}
for board in self.repo.boards_for(person.email):
for card in self.repo.cards_on(board.id):
if card.assignee != person.email or self._done(board, card):
continue
view = self._view(board, card)
if card.due_date is None:
out["no_due_date"].append(view)
elif card.due_date < today:
out["overdue"].append(view)
elif card.due_date <= horizon:
out["due_soon"].append(view)
for key in ("overdue", "due_soon"):
out[key].sort(key=lambda v: v["due_date"])
return out
def overdue(self, person: Person, board_ref: Optional[str] = None) -> list:
today, found = self.today(), []
for board in self._visible_boards(person, board_ref):
for card in self.repo.cards_on(board.id):
if card.is_overdue(today, self._done(board, card)):
view = self._view(board, card)
view["days_late"] = (today - card.due_date).days
found.append(view)
return sorted(found, key=lambda v: -v["days_late"])
def search_cards(self, person: Person, text: str, board_ref: Optional[str] = None) -> list:
"""Plain words, matched in titles, descriptions and labels."""
words = [w for w in (text or "").lower().split() if w]
if not words:
raise Invalid("say what to look for")
found = []
for board in self._visible_boards(person, board_ref):
for card in self.repo.cards_on(board.id):
hay = f"{card.title} {card.description} {' '.join(card.labels)}".lower()
if all(w in hay for w in words):
found.append(self._view(board, card))
return found
# ---- dependencies ---------------------------------------------------------------------
def block_card(self, person: Person, card_ref, blocked_by_ref) -> dict:
"""Record that a card can't finish before another one does."""
board, card = self._card(card_ref, person)
board.require_edit(person)
_, blocker = self._card(blocked_by_ref, person)
cards = {c.number: c for c in self.repo.cards_on(board.id)}
check_blocker(card, blocker, lambda n: cards[n].blocked_by if n in cards else ())
if blocker.number not in card.blocked_by:
card.blocked_by = card.blocked_by + (blocker.number,)
self.repo.save_card(card)
self._log(board, card, person, "blocked", "blocked_by", None, f"#{blocker.number}")
return self._view(board, card, self._open_blockers(board, card, cards))
def unblock_card(self, person: Person, card_ref, blocked_by_ref) -> dict:
board, card = self._card(card_ref, person)
board.require_edit(person)
n = parse_number(blocked_by_ref)
if n in card.blocked_by:
card.blocked_by = tuple(x for x in card.blocked_by if x != n)
self.repo.save_card(card)
self._log(board, card, person, "unblocked", "blocked_by", f"#{n}", None)
return self._view(board, card)
def what_is_blocking(self, person: Person, card_ref) -> dict:
"""Every unfinished card standing between this card and done, following the chain."""
board, card = self._card(card_ref, person)
cards = {c.number: c for c in self.repo.cards_on(board.id)}
chain, seen, todo = [], set(), list(card.blocked_by)
while todo:
n = todo.pop(0)
if n in seen or n not in cards:
continue
seen.add(n)
other = cards[n]
if not self._done(board, other):
chain.append(self._view(board, other))
todo.extend(other.blocked_by)
return {"card": self._view(board, card), "blocking": chain,
"ready": not chain}
# ---- checklist and AI -----------------------------------------------------------------
def _ai(self) -> Assistant:
if self.assistant is None:
raise Invalid("no AI assistant configured")
return self.assistant
@staticmethod
def _model_call(fn, *args):
"""The model is a remote service: a slow or failed answer becomes a clear message for
the person (or Claude) instead of a 500, and nothing has been changed."""
try:
return fn(*args)
except Exception:
raise Invalid("the AI model is busy or slow right now; nothing was changed, "
"try again in a minute")
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)
steps = self._model_call(self._ai().checklist, card.title, card.description)
return {"card": card.number, "suggested_checklist": steps}
def add_checklist(self, person: Person, card_ref, items: list) -> dict:
board, card = self._card(card_ref, person)
board.require_edit(person)
new = [clean_text(str(i), "an empty checklist item", 200) for i in items or []]
if not new:
raise Invalid("give at least one checklist item")
if len(card.checklist) + len(new) > 30:
raise Invalid("at most 30 checklist items on a card")
card.checklist = card.checklist + tuple((t, False) for t in new)
self.repo.save_card(card)
self._log(board, card, person, "checklist added", "checklist", None, "; ".join(new))
return self._view(board, card)
def check_item(self, person: Person, card_ref, index: int, done: bool = True) -> dict:
board, card = self._card(card_ref, person)
board.require_edit(person)
items = list(card.checklist)
if not 0 <= index < len(items):
raise Invalid(f"card #{card.number} has {len(items)} checklist items (from 0)")
text, was = items[index]
if was != done:
items[index] = (text, bool(done))
card.checklist = tuple(items)
self.repo.save_card(card)
self._log(board, card, person, "checked" if done else "unchecked", "checklist",
None, text)
return self._view(board, card)
def similar_cards(self, person: Person, card_ref, limit: int = 5) -> list:
"""Cards whose meaning is close to this one, on the boards you can see."""
_, card = self._card(card_ref, person)
boards = {b.id: b for b in self.repo.boards_for(person.email)}
pool = [c for b in boards.values() for c in self.repo.cards_on(b.id)]
vectors = self._vectors(pool)
mine = vectors.get(card.number)
if mine is None:
return []
ranked = sorted(((round(_cosine(mine, vectors[c.number]), 3), c) for c in pool
if c.number != card.number and c.number in vectors),
key=lambda x: -x[0])
return [{**self._view(boards[c.board_id], c), "similarity": s}
for s, c in ranked[:max(1, int(limit))]]
def _vectors(self, cards: list) -> dict:
"""Embeddings are cached per card and recomputed only when its text changed."""
stored = self.repo.embeddings(list({c.board_id for c in cards}))
out, missing = {}, []
for c in cards:
h, vec = stored.get(c.number, (None, None))
if h == _hash(_text(c)):
out[c.number] = vec
else:
missing.append(c)
for i in range(0, len(missing), 32):
batch = missing[i:i + 32]
vectors = self._model_call(self._ai().embed, [_text(c) for c in batch])
for c, vec in zip(batch, vectors):
self.repo.save_embedding(c.number, _hash(_text(c)), vec)
out[c.number] = vec
return out
def board_summary(self, person: Person, board_ref: str) -> dict:
"""A short daily stand-up for a board: what moved in the last day, what's late, what's
blocked. The facts are counted here; the AI only writes them up."""
board = self._board(board_ref, person)
cards = {c.number: c for c in self.repo.cards_on(board.id)}
today = self.today()
overdue = [c for c in cards.values() if c.is_overdue(today, self._done(board, c))]
blocked = [c for c in cards.values() if self._open_blockers(board, c, cards)
and not self._done(board, c)]
recent = [e for e in self.repo.activity(board.id, 200)
if e.at is None or e.at.date() >= today - timedelta(days=1)]
per_list = {lst.name: sum(1 for c in cards.values() if c.list_id == lst.id)
for lst in board.lists}
def section(title: str, lines: list) -> list:
return [title, *(lines or ["- none"])]
facts = "\n".join([
f"Board: {board.title}. Today: {today.isoformat()}.",
"Cards per list: " + ", ".join(f"{k} {v}" for k, v in per_list.items()) + ".",
*section("Changes in the last day:", [f"- {self._sentence(e)}" for e in recent[:40]]),
*section("Overdue:", [f"- #{c.number} {c.title} (due {c.due_date}, "
f"{c.assignee or 'nobody'})" for c in overdue]),
*section("Blocked:", [f"- #{c.number} {c.title} waits on "
+ ", ".join(f"#{n}" for n in self._open_blockers(board, c, cards))
for c in blocked])])
try:
text, by_ai = self._ai().summary(facts), True
except Exception: # the summary must not depend on the model being up
text, by_ai = facts, False
return {"board": board.title, "summary": text, "written_by_ai": by_ai,
"overdue": len(overdue),
"blocked": len(blocked), "changes_last_day": len(recent)}
# ---- history --------------------------------------------------------------------------
def board_activity(self, person: Person, board_ref: str, limit: int = 50) -> list:
board = self._board(board_ref, person)
return [self._event(e) for e in self.repo.activity(board.id, max(1, min(int(limit), 200)))]
def _event(self, e: Event) -> dict:
return {"at": e.at.isoformat(timespec="seconds") if e.at else None, "who": e.actor,
"card": e.card_number, "action": e.action, "field": e.field,
"before": e.before, "after": e.after, "text": self._sentence(e)}
@staticmethod
def _sentence(e: Event) -> str:
if e.action == "created" and e.card_number and e.after:
return f"{e.actor} created #{e.card_number} in {e.after}"
if e.action == "moved" and e.card_number:
return f"{e.actor} moved #{e.card_number} from {e.before} to {e.after}"
what = f"#{e.card_number} " if e.card_number else ""
change = ""
if e.field and (e.before or e.after):
change = f" {e.field}: {e.before or '(none)'} -> {e.after or '(none)'}"
return f"{e.actor} {e.action} {what}{change}".replace(" ", " ").strip()