mobiledev/taskflow
TaskFlow, a team task board you run from the web or from Claude and ChatGPT. FastAPI + DataK3 + appid + Models + MCP.
A remix template · 0 remixes · Remix mobiledev/taskflow
app/taskflow/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 can't be skipped by using the other door.
Boards and cards are named the way people talk: a board by its id or its title ("the Ops
board"), a card by its id or its number ("card 12", "#12").
"""
import math
import re
import uuid
from datetime import date, timedelta
from typing import Callable, Optional
from taskflow.application.ports import Assistant, BoardRepository
from taskflow.domain.model import (MAX_CARDS, Board, Card, Forbidden, Invalid, Person, Role,
parse_date, would_cycle)
class NotFound(Exception):
pass
AI_UNAVAILABLE = ("Dodil Models didn't answer, so the AI part couldn't run. The organization's "
"AI credits may be used up for now; try again later.")
def _new_id() -> str:
return uuid.uuid4().hex[:12]
class TaskFlow:
def __init__(self, repo: BoardRepository, assistant: Optional[Assistant] = None,
today: Callable[[], date] = date.today):
self.repo, self.assistant, self.today = repo, assistant, today
# --- finding things the way people name them -------------------------------------------
def _board(self, person: Person, ref: str) -> Board:
ref = (ref or "").strip()
board = self.repo.get(ref) if ref else None
if board is not None and board.role_of(person) is not None:
return board
mine = self.repo.boards_for(person.email)
for match in (lambda b: b.title.lower() == ref.lower(),
lambda b: ref.lower() in b.title.lower()):
found = [b for b in mine if ref and match(b)]
if len(found) == 1:
return found[0]
if len(found) > 1:
raise Invalid(f"{ref!r} matches several boards: "
f"{', '.join(repr(b.title) for b in found)}; use the board id")
raise NotFound(f"no board {ref!r}") # a board you can't see doesn't exist for you
def _card(self, person: Person, ref, board_ref: Optional[str] = None) -> tuple:
ref = str(ref or "").strip()
number = re.fullmatch(r"#?(\d+)", ref)
# A card id can be all digits by chance (random hex), so an exact id wins over a number.
if number and not ref.startswith("#") and self.repo.get_card(ref) is not None:
number = None
if number:
n = int(number.group(1))
boards = ([self._board(person, board_ref)] if board_ref
else self.repo.boards_for(person.email))
hits = [(b, c) for b in boards for c in self.repo.cards_on(b.id) if c.number == n]
if len(hits) == 1:
return hits[0]
if len(hits) > 1:
raise Invalid(f"card #{n} exists on {', '.join(repr(b.title) for b, _ in hits)}"
f"; say which board")
raise NotFound(f"no card #{n}")
card = self.repo.get_card(ref)
board = self.repo.get(card.board_id) if card else None
if card is None or board is None or board.role_of(person) is None:
raise NotFound(f"no card {ref!r}")
if board_ref and self._board(person, board_ref).id != board.id:
raise NotFound(f"no card {ref!r} on that board")
return board, card
def _view(self, board: Board, card: Card, cards: Optional[dict] = None) -> dict:
"""A card as the doors return it. Pass `cards` (id -> Card) when showing many cards, so
the board is read once, not once per card."""
if cards is None:
cards = {c.id: c for c in self.repo.cards_on(board.id)}
open_blockers = [cards[i] for i in card.blocked_by
if i in cards and not cards[i].is_done(board)]
done = sum(1 for item in card.checklist if item["done"])
return {
"id": card.id, "number": card.number, "board_id": board.id,
"board": board.title, "title": card.title, "description": card.description,
"list_name": card.list_name, "position": card.position,
"assignee": card.assignee,
"due_date": card.due_date.isoformat() if card.due_date else None,
"overdue": card.is_overdue(board, self.today()),
"labels": list(card.labels),
"blocked_by": [f"#{cards[i].number}" for i in card.blocked_by if i in cards],
"blocked": bool(open_blockers),
"checklist": card.checklist,
"checklist_progress": f"{done}/{len(card.checklist)}" if card.checklist else None,
}
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]
return max(positions, default=0.0) + 1.0
def _record(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 or {})
# --- boards ----------------------------------------------------------------------------
def create_board(self, person: Person, title: str, lists: Optional[list] = None) -> dict:
title = (title or "").strip()
if not title:
raise Invalid("a board needs a title")
board = Board(id=_new_id(), title=title[:80], owner_email=person.email)
if lists:
board.lists = []
for name in lists:
board.add_list(person, name)
self.repo.save_board(board)
self._record(board, None, person, f"created board {board.title!r}")
return self.show_board(person, board.id)
def my_boards(self, person: Person) -> list:
return [{"id": b.id, "title": b.title, "role": b.role_of(person).value}
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(person, board_ref)
cards = sorted(self.repo.cards_on(board.id), key=lambda c: c.position)
by_id = {c.id: c for c in cards}
return {
"id": board.id, "title": board.title, "role": board.role_of(person).value,
"lists": [{"name": name, "cards": [self._view(board, c, by_id) for c in cards
if c.list_name == name]}
for name in board.lists],
"members": self._members(board),
}
def _members(self, board: Board) -> list:
people = [{"email": board.owner_email, "role": Role.OWNER.value}]
people += [{"email": e, "role": r.value} for e, r in sorted(board.members.items())]
return people
def add_list(self, person: Person, board_ref: str, name: str,
before: Optional[str] = None) -> dict:
board = self._board(person, board_ref)
name = board.add_list(person, name, before)
self.repo.save_board(board)
self._record(board, None, person, f"added list {name!r}")
return {"board_id": board.id, "lists": board.lists}
def rename_list(self, person: Person, board_ref: str, name: str, new_name: str) -> dict:
board = self._board(person, board_ref)
old, new = board.rename_list(person, name, new_name)
if old != new:
self.repo.save_board(board)
for card in self.repo.cards_on(board.id):
if card.list_name == old:
card.list_name = new
self.repo.save_card(card)
self._record(board, None, person, f"renamed list {old!r} to {new!r}")
return {"board_id": board.id, "lists": board.lists}
def delete_list(self, person: Person, board_ref: str, name: str) -> dict:
board = self._board(person, board_ref)
target = board.list_named(name)
count = sum(1 for c in self.repo.cards_on(board.id) if c.list_name == target)
board.remove_list(person, target, count)
self.repo.save_board(board)
self._record(board, None, person, f"deleted list {target!r}")
return {"board_id": board.id, "lists": board.lists}
def delete_board(self, person: Person, board_ref: str, confirm_title: str) -> dict:
"""Owner only, and the caller must repeat the board's title: deleting a board deletes
every card on it."""
board = self._board(person, board_ref)
board.require_owner(person)
if (confirm_title or "").strip() != board.title:
raise Invalid(f"type the board's title {board.title!r} to delete it")
self.repo.delete_board(board.id)
return {"deleted": board.title}
# --- people ----------------------------------------------------------------------------
def members(self, person: Person, board_ref: str) -> list:
return self._members(self._board(person, board_ref))
def add_member(self, person: Person, board_ref: str, email: str,
role: str = "member") -> dict:
board = self._board(person, board_ref)
try:
wanted = Role(str(role).strip().lower())
except ValueError:
raise Invalid("role must be 'member' or 'viewer'")
before = board.members.get((email or "").strip().lower())
email = board.add_member(person, email, wanted)
self.repo.save_board(board)
if before is None:
self._record(board, None, person, f"invited {email} as {wanted.value}")
elif before is not wanted:
self._record(board, None, person, f"changed {email} from {before.value} to "
f"{wanted.value}")
return {"board_id": board.id, "email": email, "role": wanted.value}
def remove_member(self, person: Person, board_ref: str, email: str) -> dict:
board = self._board(person, board_ref)
email = board.remove_member(person, email)
self.repo.remove_membership(board.id, email)
self._record(board, None, person, f"removed {email}")
for card in self.repo.cards_on(board.id): # their cards go back to the pool
if card.assignee == email:
diff = card.edit(board, assignee=None)
self.repo.save_card(card)
self._record(board, card, person, f"unassigned #{card.number}", diff)
return {"board_id": board.id, "removed": email}
# --- cards -----------------------------------------------------------------------------
def add_card(self, person: Person, board_ref: str, title: str,
list_name: Optional[str] = None, description: str = "",
assignee: Optional[str] = None, due_date=None, labels=None) -> dict:
board = self._board(person, board_ref)
board.require_edit(person)
target = board.list_named(list_name) if list_name else board.lists[0]
if len(self.repo.cards_on(board.id)) >= MAX_CARDS:
raise Invalid(f"a board holds at most {MAX_CARDS} cards; delete or archive some first")
card = Card(id=_new_id(), board_id=board.id, number=0, title="", list_name=target,
position=self._end_of(board, target))
card.edit(board, title=title, description=description, assignee=assignee,
due_date=due_date, labels=labels or ())
card.number = self.repo.claim_card_number(board.id, board.card_seq + 1)
board.card_seq = max(board.card_seq, card.number)
self.repo.save_board(board)
self.repo.save_card(card)
self._record(board, card, person, f"added #{card.number} {card.title!r} to {target}")
return self._view(board, card)
def delete_card(self, person: Person, card_ref, board_ref: Optional[str] = None) -> dict:
board, card = self._card(person, card_ref, board_ref)
board.require_edit(person)
for other in self.repo.cards_on(board.id): # nothing waits on a deleted card
if card.id in other.blocked_by:
other.blocked_by.remove(card.id)
self.repo.save_card(other)
self.repo.delete_card(card.id)
self._record(board, card, person, f"deleted #{card.number} {card.title!r}",
{"title": [card.title, None]})
return {"deleted": f"#{card.number} {card.title}", "board": board.title}
def show_card(self, person: Person, card_ref, board_ref: Optional[str] = None) -> dict:
board, card = self._card(person, card_ref, board_ref)
return {**self._view(board, card), "history": self.repo.activity(board.id, card.id, 20)}
def update_card(self, person: Person, card_ref, board_ref: Optional[str] = None,
**changes) -> dict:
board, card = self._card(person, card_ref, board_ref)
board.require_edit(person)
diff = card.edit(board, **changes)
if diff:
self.repo.save_card(card)
self._record(board, card, person,
f"changed {', '.join(diff)} on #{card.number}", diff)
return self._view(board, card)
def assign_card(self, person: Person, card_ref, assignee: Optional[str],
board_ref: Optional[str] = None) -> dict:
return self.update_card(person, card_ref, board_ref, assignee=assignee or None)
def move_card(self, person: Person, card_ref, list_name: str,
board_ref: Optional[str] = None, index: Optional[int] = None) -> dict:
"""Move a card to a list. `index` (0 = top) places it among that list's cards, which is
what a drag and drop sends; without it the card goes to the bottom."""
board, card = self._card(person, card_ref, board_ref)
board.require_edit(person)
target = board.list_named(list_name)
others = sorted((c for c in self.repo.cards_on(board.id)
if c.list_name == target and c.id != card.id), key=lambda c: c.position)
if index is None or index >= len(others):
position = (others[-1].position + 1.0) if others else 1.0
else:
index = max(0, index)
after = others[index].position
before = others[index - 1].position if index > 0 else after - 2.0
position = (before + after) / 2
moved = card.move_to(board, target, position)
self.repo.save_card(card)
if moved:
self._record(board, card, person,
f"moved #{card.number} from {moved[0]} to {moved[1]}",
{"list_name": moved})
return self._view(board, card)
def card_history(self, person: Person, card_ref, board_ref: Optional[str] = None) -> list:
board, card = self._card(person, card_ref, board_ref)
return self.repo.activity(board.id, card.id)
def board_activity(self, person: Person, board_ref: str, limit: int = 30) -> list:
board = self._board(person, board_ref)
return self.repo.activity(board.id, None, max(1, min(int(limit), 200)))
# --- "what's on my plate?" -------------------------------------------------------------
def my_cards(self, person: Person, due_within_days: Optional[int] = None) -> list:
"""Open cards assigned to me on every board I'm on, soonest due first. With
`due_within_days=7` it answers "what's on my plate this week?" (overdue included)."""
today, out = self.today(), []
until = today + timedelta(days=due_within_days) if due_within_days is not None else None
for board in self.repo.boards_for(person.email):
cards = {c.id: c for c in self.repo.cards_on(board.id)}
for card in cards.values():
if card.assignee != person.email or card.is_done(board):
continue
if until and not (card.due_date and card.due_date <= until):
continue
out.append(self._view(board, card, cards))
return sorted(out, key=lambda c: (c["due_date"] is None, c["due_date"] or "", c["board"]))
def overdue(self, person: Person, board_ref: Optional[str] = None) -> list:
boards = ([self._board(person, board_ref)] if board_ref
else self.repo.boards_for(person.email))
today = self.today()
late = []
for b in boards:
cards = {c.id: c for c in self.repo.cards_on(b.id)}
late += [self._view(b, c, cards) for c in cards.values() if c.is_overdue(b, today)]
return sorted(late, key=lambda c: c["due_date"])
# --- dependencies ----------------------------------------------------------------------
def block_card(self, person: Person, card_ref, blocked_by, board_ref: Optional[str] = None
) -> dict:
board, card = self._card(person, card_ref, board_ref)
board.require_edit(person)
_, blocker = self._card(person, blocked_by, board.id)
if blocker.id == card.id:
raise Invalid("a card can't block itself")
if blocker.id not in card.blocked_by:
cards = {c.id: c for c in self.repo.cards_on(board.id)}
if would_cycle(card.id, blocker.id, lambda i: cards[i].blocked_by if i in cards else []):
raise Invalid(f"#{blocker.number} already waits on #{card.number}; "
f"that would be a loop")
card.blocked_by.append(blocker.id)
self.repo.save_card(card)
self._record(board, card, person, f"#{card.number} is blocked by #{blocker.number}",
{"blocked_by": [None, f"#{blocker.number}"]})
return self._view(board, card)
def unblock_card(self, person: Person, card_ref, blocked_by,
board_ref: Optional[str] = None) -> dict:
board, card = self._card(person, card_ref, board_ref)
board.require_edit(person)
_, blocker = self._card(person, blocked_by, board.id)
if blocker.id in card.blocked_by:
card.blocked_by.remove(blocker.id)
self.repo.save_card(card)
self._record(board, card, person, f"#{card.number} no longer waits on "
f"#{blocker.number}",
{"blocked_by": [f"#{blocker.number}", None]})
return self._view(board, card)
def what_blocks(self, person: Person, card_ref, board_ref: Optional[str] = None) -> dict:
"""Everything still open that this card waits on, directly or further down the chain
("what is blocking the launch?")."""
board, card = self._card(person, card_ref, board_ref)
cards = {c.id: c for c in self.repo.cards_on(board.id)}
found, todo, seen = [], list(card.blocked_by), set()
while todo:
current = cards.get(todo.pop(0))
if current is None or current.id in seen:
continue
seen.add(current.id)
if not current.is_done(board):
found.append(self._view(board, current, cards))
todo.extend(current.blocked_by)
return {"card": f"#{card.number} {card.title}", "blocked": bool(found), "blockers": found}
# --- checklist and AI ------------------------------------------------------------------
def add_checklist(self, person: Person, card_ref, items: list,
board_ref: Optional[str] = None) -> dict:
board, card = self._card(person, card_ref, board_ref)
board.require_edit(person)
added = card.add_checklist_items(items)
self.repo.save_card(card)
self._record(board, card, person, f"added {len(added)} checklist item(s) to "
f"#{card.number}", {"checklist": [None, added]})
return self._view(board, card)
def tick(self, person: Person, card_ref, item: int, done: bool = True,
board_ref: Optional[str] = None) -> dict:
board, card = self._card(person, card_ref, board_ref)
board.require_edit(person)
entry = card.tick(int(item), done)
self.repo.save_card(card)
self._record(board, card, person,
f"{'ticked' if done else 'unticked'} {entry['text']!r} on #{card.number}")
return self._view(board, card)
def _ask(self, call):
"""Run one AI request. If Dodil Models doesn't answer (down, or the organization's AI
credits are used up) say so plainly instead of failing with a bare server error."""
if self.assistant is None:
raise Invalid("no AI assistant is configured")
try:
return call(self.assistant)
except Exception:
raise Invalid(AI_UNAVAILABLE)
def suggest_checklist(self, person: Person, card_ref, board_ref: Optional[str] = None
) -> dict:
"""A suggestion only: nothing changes until someone calls add_checklist."""
board, card = self._card(person, card_ref, board_ref)
steps = self._ask(lambda ai: ai.checklist(card.title, card.description))
return {"card": f"#{card.number} {card.title}", "suggested_checklist": steps}
def checklist_from_ai(self, person: Person, card_ref, board_ref: Optional[str] = None
) -> dict:
""""Turn this card into a checklist": suggest the steps and add them in one go."""
board, card = self._card(person, card_ref, board_ref)
board.require_edit(person)
steps = self._ask(lambda ai: ai.checklist(card.title, card.description))
return self.add_checklist(person, card.id, steps)
def similar_cards(self, person: Person, card_ref, board_ref: Optional[str] = None,
limit: int = 5) -> list:
"""Cards on the same board that mean something close, by embedding similarity."""
board, card = self._card(person, card_ref, board_ref)
cards = {c.id: c for c in self.repo.cards_on(board.id)}
others = [c for c in cards.values() if c.id != card.id][:200]
if not others:
return []
texts = [_text(card)] + [_text(c) for c in others]
vectors = self._ask(lambda ai: ai.embed(texts))
scored = sorted(((_cosine(vectors[0], v), c) for v, c in zip(vectors[1:], others)),
key=lambda pair: pair[0], reverse=True)
return [{**self._view(board, c, cards), "similarity": round(score, 3)}
for score, c in scored[:max(1, min(int(limit), 20))] if score > 0.3]
def board_summary(self, person: Person, board_ref: str) -> dict:
"""A short daily summary: the facts are counted here, the AI only words them."""
board = self._board(person, board_ref)
today = self.today()
cards = self.repo.cards_on(board.id)
counts = {name: sum(1 for c in cards if c.list_name == name) for name in board.lists}
late = [c for c in cards if c.is_overdue(board, today)]
soon = [c for c in cards if c.due_date and today <= c.due_date <= today
+ timedelta(days=2) and not c.is_done(board)]
recent = self.repo.activity(board.id, None, 15)
facts = "\n".join([
f"Board: {board.title}. Today: {today.isoformat()}.",
"Cards per list: " + ", ".join(f"{k}: {v}" for k, v in counts.items()),
"Overdue: " + ("; ".join(f"#{c.number} {c.title} (due {c.due_date}, "
f"{c.assignee or 'unassigned'})" for c in late) or "none"),
"Due in the next 2 days: " + ("; ".join(f"#{c.number} {c.title}" for c in soon)
or "none"),
"Recent activity: " + ("; ".join(f"{e['actor']} {e['action']}" for e in recent)
or "none"),
])
try: # the facts stand on their own if the AI is out
text = self._ask(lambda ai: ai.summarize(facts))
except Invalid:
text = facts
return {"board": board.title, "summary": text, "counts": counts,
"overdue": [f"#{c.number}" for c in late]}
def _text(card: Card) -> str:
return f"{card.title}. {card.description} {' '.join(card.labels)}".strip()
def _cosine(a, b) -> 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__ = ["TaskFlow", "NotFound", "Forbidden", "Invalid", "parse_date"]