dmslanes/lanes
Lanes - team task boards with roles, history and an MCP endpoint for Claude and ChatGPT
A remix template · 0 remixes · Remix dmslanes/lanes
app/application/use_cases.py
"""Use cases. The web routes and the MCP tools call these and nothing else.
Every method takes the signed-in person (`actor`, an email) first and scopes
every read by it: a board you are not on is "not found", for people and for
their AI assistant alike.
"""
from __future__ import annotations
import math
import uuid
from dataclasses import replace
from datetime import date, datetime, timedelta, timezone
from typing import Callable
from app.application.ports import Assistant, Repository
from app.domain.model import (
DONE_LIST_NAMES,
Activity,
Board,
BoardList,
Card,
DomainError,
Forbidden,
Membership,
NotFound,
Role,
Workspace,
clean_labels,
clean_title,
diff_card,
normalize_email,
parse_checklist,
would_cycle,
)
DEFAULT_LISTS = ("To do", "Doing", "Done")
_UNSET = object()
def _utcnow() -> datetime:
return datetime.now(timezone.utc)
class BoardService:
def __init__(self, repo: Repository, clock: Callable[[], datetime] = _utcnow,
new_id: Callable[[], str] = lambda: uuid.uuid4().hex, ai: Assistant | None = None):
self.repo = repo
self.ai = ai
self._vectors: dict[tuple[str, int], list[float]] = {} # (card id, version) -> embedding
self.clock = clock
self.new_id = new_id
# ---------- access helpers ----------
def _membership(self, actor: str, workspace_id: str) -> Membership:
m = self.repo.get_membership(workspace_id, normalize_email(actor))
if m is None:
raise NotFound("workspace not found")
return m
def _board(self, actor: str, board_id: str) -> tuple[Board, Membership]:
board = self.repo.get_board(board_id)
if board is None:
raise NotFound("board not found")
m = self.repo.get_membership(board.workspace_id, normalize_email(actor))
if m is None:
raise NotFound("board not found") # never confirm a board exists to an outsider
return board, m
def _card(self, actor: str, card_ref: str | int) -> tuple[Card, Membership]:
card = None
if isinstance(card_ref, int) or str(card_ref).lstrip("#").isdigit():
card = self.repo.get_card_by_number(int(str(card_ref).lstrip("#")))
if card is None:
card = self.repo.get_card(str(card_ref))
if card is None:
raise NotFound(f"card {card_ref} not found")
m = self.repo.get_membership(card.workspace_id, normalize_email(actor))
if m is None:
raise NotFound(f"card {card_ref} not found")
return card, m
def _visible_workspace_ids(self, actor: str) -> list[str]:
return [m.workspace_id for m in self.repo.memberships_of_person(normalize_email(actor))]
def _resolve_list(self, board_id: str, list_ref: str) -> BoardList:
lists = self.repo.lists_of_board(board_id)
for lst in lists:
if lst.id == list_ref:
return lst
wanted = (list_ref or "").strip().lower()
for lst in lists:
if lst.name.lower() == wanted:
return lst
names = ", ".join(lst.name for lst in lists)
raise NotFound(f"no list {list_ref!r} on this board (lists: {names})")
def _done_list_ids(self, board_ids: set[str]) -> set[str]:
out = set()
for bid in board_ids:
out |= {l.id for l in self.repo.lists_of_board(bid) if l.name.lower() in DONE_LIST_NAMES}
return out
def _record(self, actor: str, card: Card, action: str, changes: dict) -> None:
self.repo.add_activity(Activity(
id=self.new_id(), card_id=card.id, board_id=card.board_id,
actor=normalize_email(actor), at=self.clock(), action=action, changes=changes))
# ---------- workspaces and people ----------
def create_workspace(self, actor: str, name: str) -> Workspace:
actor = normalize_email(actor)
ws = Workspace(id=self.new_id(), name=clean_title(name, "workspace name"),
created_by=actor, created_at=self.clock())
self.repo.add_workspace(ws)
self.repo.save_membership(Membership(ws.id, actor, Role.OWNER))
return ws
def list_workspaces(self, actor: str) -> list[tuple[Workspace, Role]]:
out = []
for m in self.repo.memberships_of_person(normalize_email(actor)):
ws = self.repo.get_workspace(m.workspace_id)
if ws:
out.append((ws, m.role))
return sorted(out, key=lambda p: p[0].name.lower())
def list_members(self, actor: str, workspace_id: str) -> list[Membership]:
self._membership(actor, workspace_id)
return sorted(self.repo.memberships_of_workspace(workspace_id), key=lambda m: (m.role != Role.OWNER, m.email))
def invite_member(self, actor: str, workspace_id: str, email: str, role: str = "member") -> Membership:
self._membership(actor, workspace_id).require_owner()
role_ = Role(role)
if role_ is Role.OWNER:
raise DomainError("a workspace has one owner; invite as member or viewer")
email = normalize_email(email)
existing = self.repo.get_membership(workspace_id, email)
if existing and existing.role is Role.OWNER:
raise DomainError("the owner's role can't be changed")
m = Membership(workspace_id, email, role_)
self.repo.save_membership(m)
return m
def remove_member(self, actor: str, workspace_id: str, email: str) -> None:
self._membership(actor, workspace_id).require_owner()
email = normalize_email(email)
target = self.repo.get_membership(workspace_id, email)
if target is None:
raise NotFound(f"{email} is not in this workspace")
if target.role is Role.OWNER:
raise DomainError("the owner can't be removed")
self.repo.delete_membership(workspace_id, email)
# ---------- boards and lists ----------
def create_board(self, actor: str, workspace_id: str, name: str,
lists: tuple[str, ...] | list[str] = DEFAULT_LISTS) -> Board:
self._membership(actor, workspace_id).require_edit()
board = Board(id=self.new_id(), workspace_id=workspace_id,
name=clean_title(name, "board name"), created_at=self.clock())
self.repo.add_board(board)
for i, list_name in enumerate(lists):
self.repo.add_list(BoardList(self.new_id(), board.id, clean_title(list_name, "list name"), i))
return board
def list_boards(self, actor: str) -> list[Board]:
return sorted(self.repo.boards_of_workspaces(self._visible_workspace_ids(actor)),
key=lambda b: b.name.lower())
def find_board(self, actor: str, ref: str) -> Board:
"""A board by id or by name ("the Ops board"), among the boards the actor can see."""
boards = self.list_boards(actor)
for b in boards:
if b.id == ref:
return b
wanted = (ref or "").strip().lower().removesuffix(" board").strip()
matches = [b for b in boards if b.name.lower() == wanted] or \
[b for b in boards if wanted and wanted in b.name.lower()]
if len(matches) == 1:
return matches[0]
if not matches:
raise NotFound(f"no board called {ref!r} that you can see")
raise DomainError(f"{ref!r} matches several boards: {', '.join(b.name for b in matches)}")
def get_board(self, actor: str, board_id: str) -> dict:
board, m = self._board(actor, board_id)
lists = sorted(self.repo.lists_of_board(board.id), key=lambda l: l.position)
cards = self.repo.cards_of_board(board.id)
return {
"board": board, "role": m.role, "lists": lists,
"cards": {l.id: sorted([c for c in cards if c.list_id == l.id], key=lambda c: c.position)
for l in lists},
}
def add_list(self, actor: str, board_id: str, name: str) -> BoardList:
board, m = self._board(actor, board_id)
m.require_edit()
position = len(self.repo.lists_of_board(board.id))
lst = BoardList(self.new_id(), board.id, clean_title(name, "list name"), position)
self.repo.add_list(lst)
return lst
# ---------- cards ----------
def add_card(self, actor: str, board_id: str, list_ref: str, title: str, description: str = "",
assignee: str | None = None, due_date: date | None = None,
labels: list[str] | None = None) -> Card:
board, m = self._board(actor, board_id)
m.require_edit()
lst = self._resolve_list(board.id, list_ref)
if assignee:
assignee = self._check_assignee(board.workspace_id, assignee)
now = self.clock()
siblings = [c for c in self.repo.cards_of_board(board.id) if c.list_id == lst.id]
card = Card(id=self.new_id(), workspace_id=board.workspace_id, board_id=board.id, list_id=lst.id,
number=self.repo.claim_card_number(), title=clean_title(title),
description=(description or "").strip()[:5000], assignee=assignee or None,
due_date=due_date, labels=clean_labels(labels),
position=max([c.position for c in siblings], default=-1) + 1,
created_at=now, updated_at=now)
self.repo.add_card(card)
self._record(actor, card, "created", {"title": [None, card.title], "list": [None, lst.name]})
return card
def _check_assignee(self, workspace_id: str, email: str) -> str:
email = normalize_email(email)
if self.repo.get_membership(workspace_id, email) is None:
raise DomainError(f"{email} is not a member of this workspace")
return email
def _save(self, actor: str, before: Card, after: Card, action: str,
expected_version: int | None = None) -> Card:
changes = diff_card(before, after)
if not changes:
return before
if "list_id" in changes: # history speaks list names, not ids
old, new = self.repo.get_list(before.list_id), self.repo.get_list(after.list_id)
changes["list"] = [old.name if old else None, new.name if new else None]
del changes["list_id"]
after = replace(after, updated_at=self.clock(), version=before.version + 1)
self.repo.save_card(after, expected_version if expected_version is not None else before.version)
self._record(actor, after, action, changes)
return after
def update_card(self, actor: str, card_ref: str | int, *, title=_UNSET, description=_UNSET,
assignee=_UNSET, due_date=_UNSET, labels=_UNSET,
expected_version: int | None = None) -> Card:
card, m = self._card(actor, card_ref)
m.require_edit()
after = replace(card)
if title is not _UNSET:
after.title = clean_title(title)
if description is not _UNSET:
after.description = (description or "").strip()[:5000]
if assignee is not _UNSET:
after.assignee = self._check_assignee(card.workspace_id, assignee) if assignee else None
if due_date is not _UNSET:
after.due_date = due_date
if labels is not _UNSET:
after.labels = clean_labels(labels)
return self._save(actor, card, after, "updated", expected_version)
# ---------- dependencies ----------
def block_card(self, actor: str, card_ref: str | int, blocker_ref: str | int) -> Card:
"""`card` can't finish before `blocker` does ("card 4 is blocked by card 2")."""
card, m = self._card(actor, card_ref)
m.require_edit()
blocker, _ = self._card(actor, blocker_ref)
if blocker.workspace_id != card.workspace_id:
raise DomainError("a card can only wait on a card of the same workspace")
if blocker.id == card.id:
raise DomainError("a card can't block itself")
if blocker.number in card.blocked_by:
return card
def blocked_by_of(n: int) -> list[int]:
c = self.repo.get_card_by_number(n)
return c.blocked_by if c else []
if would_cycle(card.number, blocker.number, blocked_by_of):
raise DomainError(f"card {blocker.number} already waits on card {card.number}; that would be a loop")
return self._save(actor, card, replace(card, blocked_by=card.blocked_by + [blocker.number]), "blocked")
def unblock_card(self, actor: str, card_ref: str | int, blocker_ref: str | int) -> Card:
card, m = self._card(actor, card_ref)
m.require_edit()
number = int(str(blocker_ref).lstrip("#")) if str(blocker_ref).lstrip("#").isdigit() \
else self._card(actor, blocker_ref)[0].number
return self._save(actor, card, replace(card, blocked_by=[n for n in card.blocked_by if n != number]),
"unblocked")
def find_card(self, actor: str, ref: str | int) -> Card:
"""A card by number, id, or a title words ("the launch") when exactly one visible card matches."""
try:
return self._card(actor, ref)[0]
except NotFound:
matches = self.search_cards(actor, str(ref).lower().removeprefix("the ").removesuffix(" card"))
if len(matches) == 1:
return matches[0]
if matches:
raise DomainError(f"{ref!r} matches several cards: "
+ ", ".join(f"#{c.number} {c.title}" for c in matches[:8])) from None
raise
def blockers(self, actor: str, card_ref: str | int) -> list[Card]:
"""Open cards that stand between `card` and done, at any depth ("what is blocking the launch?")."""
card = self.find_card(actor, card_ref)
done = self._done_list_ids({card.board_id})
out, seen, todo = [], {card.number}, list(card.blocked_by)
while todo:
n = todo.pop(0)
if n in seen:
continue
seen.add(n)
c = self.repo.get_card_by_number(n)
if c is None or c.workspace_id != card.workspace_id:
continue # deleted blocker
done |= self._done_list_ids({c.board_id})
if c.list_id not in done:
out.append(c)
todo.extend(c.blocked_by)
return out
# ---------- AI (Dodil Models) ----------
def _require_ai(self) -> Assistant:
if self.ai is None:
raise DomainError("AI is not configured for this deployment")
return self.ai
def card_to_checklist(self, actor: str, card_ref: str | int) -> Card:
"""Ask the model for the steps of card N and append them to its description as a checklist."""
card, m = self._card(actor, card_ref)
m.require_edit()
reply = self._require_ai().complete(
"You split a task into the concrete steps a small team would tick off. Answer with 3 to 8 short "
"steps, one per line, imperative mood, no numbering, no commentary.",
f"Task: {card.title}\nDetails: {card.description or '(none)'}")
steps = parse_checklist(reply)
if not steps:
raise DomainError("the AI returned no steps; try again")
body = (card.description.rstrip() + "\n\n" if card.description.strip() else "") + \
"Checklist:\n" + "\n".join(f"- [ ] {s}" for s in steps)
return self._save(actor, card, replace(card, description=body[:5000]), "updated")
def similar_cards(self, actor: str, card_ref: str | int, limit: int = 5) -> list[tuple[Card, float]]:
"""Cards whose meaning is closest to card N (embeddings), among the boards the actor can see."""
card = self.find_card(actor, card_ref)
others = [c for c in self.repo.cards_of_workspaces(self._visible_workspace_ids(actor)) if c.id != card.id]
if not others:
return []
text = lambda c: f"{c.title}. {c.description} {' '.join(c.labels)}".strip()[:2000]
# shortcut: vectors live in process memory keyed by card version; move to a DataK3 vector column past ~5k cards
missing = [c for c in [card, *others] if (c.id, c.version) not in self._vectors]
if missing:
for c, v in zip(missing, self._require_ai().embed([text(c) for c in missing])):
self._vectors[(c.id, c.version)] = v
target = self._vectors[(card.id, card.version)]
scored = [(c, _cosine(target, self._vectors[(c.id, c.version)])) for c in others]
return sorted(scored, key=lambda p: -p[1])[:limit]
def board_summary(self, actor: str, board_id: str) -> str:
"""A short daily stand-up of a board: what moved in the last day, what is late, what is stuck."""
data = self.get_board(actor, board_id)
board, lists = data["board"], data["lists"]
today = self.clock().date()
done = {l.id for l in lists if l.name.lower() in DONE_LIST_NAMES}
lines = []
for l in lists:
for c in data["cards"][l.id]:
flags = [f"due {c.due_date}" + (" LATE" if c.is_overdue(today, done) else "") if c.due_date else "",
f"assigned {c.assignee}" if c.assignee else "unassigned",
f"blocked by {', '.join('#%d' % n for n in c.blocked_by)}" if c.blocked_by else ""]
lines.append(f"[{l.name}] #{c.number} {c.title} ({'; '.join(f for f in flags if f)})")
since = self.clock() - timedelta(days=1)
recent = [a for a in self.repo.activity_of_board(board.id, 200) if a.at >= since]
numbers = {c.id: c.number for cards in data["cards"].values() for c in cards}
changes = [f"{a.actor} {a.action} " + (f"#{numbers[a.card_id]}" if a.card_id in numbers else "a deleted card")
+ f": {', '.join(a.changes)}" for a in recent[:60]]
if not lines:
return f"{board.name} has no cards yet."
return self._require_ai().complete(
"You write the daily summary of a team task board in plain English: at most 6 short sentences, "
"starting with what moved, then what is late or blocked and who owns it. Cite cards as #N. "
"Use only the facts given.",
f"Board: {board.name}\nToday: {today}\nCards:\n" + "\n".join(lines)
+ "\nChanges in the last 24 hours:\n" + ("\n".join(changes) or "(none)"))
def move_card(self, actor: str, card_ref: str | int, to_list: str, position: int | None = None) -> Card:
card, m = self._card(actor, card_ref)
m.require_edit()
lst = self._resolve_list(card.board_id, to_list)
siblings = sorted([c for c in self.repo.cards_of_board(card.board_id)
if c.list_id == lst.id and c.id != card.id], key=lambda c: c.position)
if position is None or position >= len(siblings):
new_pos = (siblings[-1].position + 1) if siblings else 0
else:
position = max(position, 0)
# renumber the target list so positions stay dense and ordered
for i, sib in enumerate(siblings):
want = i if i < position else i + 1
if sib.position != want:
self.repo.save_card(replace(sib, position=want, version=sib.version + 1), sib.version)
new_pos = position
return self._save(actor, card, replace(card, list_id=lst.id, position=new_pos), "moved")
def assign_card(self, actor: str, card_ref: str | int, assignee: str | None) -> Card:
card, m = self._card(actor, card_ref)
m.require_edit()
new = self._check_assignee(card.workspace_id, assignee) if assignee else None
return self._save(actor, card, replace(card, assignee=new), "assigned")
def delete_card(self, actor: str, card_ref: str | int) -> None:
card, m = self._card(actor, card_ref)
m.require_edit()
self._record(actor, card, "deleted", {"title": [card.title, None]})
self.repo.delete_card(card.id)
def get_card(self, actor: str, card_ref: str | int) -> Card:
return self._card(actor, card_ref)[0]
def card_history(self, actor: str, card_ref: str | int) -> list[Activity]:
card, _ = self._card(actor, card_ref)
return sorted(self.repo.activity_of_card(card.id), key=lambda a: a.at)
def board_activity(self, actor: str, board_id: str, limit: int = 50) -> list[Activity]:
board, _ = self._board(actor, board_id)
return self.repo.activity_of_board(board.id, limit)
# ---------- questions people ask ----------
def my_cards(self, actor: str, due_within_days: int | None = None) -> list[Card]:
"""Open cards assigned to me; with `due_within_days`, only those due by then (overdue included)."""
actor = normalize_email(actor)
cards = [c for c in self.repo.cards_of_workspaces(self._visible_workspace_ids(actor)) if c.assignee == actor]
done = self._done_list_ids({c.board_id for c in cards})
cards = [c for c in cards if c.list_id not in done]
if due_within_days is not None:
limit = self.clock().date() + timedelta(days=due_within_days)
cards = [c for c in cards if c.due_date is not None and c.due_date <= limit]
return sorted(cards, key=lambda c: (c.due_date or date.max, c.number))
def overdue(self, actor: str, board_id: str | None = None) -> list[Card]:
if board_id:
self._board(actor, board_id)
cards = self.repo.cards_of_board(board_id)
else:
cards = self.repo.cards_of_workspaces(self._visible_workspace_ids(actor))
done = self._done_list_ids({c.board_id for c in cards})
today = self.clock().date()
return sorted([c for c in cards if c.is_overdue(today, done)], key=lambda c: (c.due_date, c.number))
def search_cards(self, actor: str, text: str) -> list[Card]:
words = [w for w in (text or "").lower().split() if w]
if not words:
return []
cards = self.repo.cards_of_workspaces(self._visible_workspace_ids(actor))
hay = lambda c: f"{c.title} {c.description} {' '.join(c.labels)}".lower()
return [c for c in cards if all(w in hay(c) for w in words)][:50]
def _cosine(a: list[float], b: list[float]) -> float:
dot = sum(x * y for x, y in zip(a, b))
norm = math.sqrt(sum(x * x for x in a)) * math.sqrt(sum(y * y for y in b))
return dot / norm if norm else 0.0
__all__ = ["BoardService", "DEFAULT_LISTS", "Forbidden", "NotFound", "DomainError"]