dachdigitalwerksorg/kanvaflow
KanvaFlow — Remixable team task board with RBAC, drag-and-drop Kanban, dependencies, Dodil Models AI, and MCP tools.
A remix template · 0 remixes · Remix dachdigitalwerksorg/kanvaflow
app/boards/application/use_cases.py
"""KanvaFlow Application Layer: every action a user or AI agent can perform.
Both `doors/web.py` (FastAPI) and `doors/mcp.py` (Claude / ChatGPT MCP tools) delegate
exclusively to `Boards` methods here, guaranteeing a single source of truth for RBAC,
visibility, activity logging, dependencies, overdue tracking, and Dodil Models AI helpers.
"""
import uuid
from datetime import date, timedelta
from typing import Any, Optional
from boards.application.ports import Assistant, BoardRepository
from boards.domain.model import (
Board,
Card,
Forbidden,
Invalid,
Person,
Role,
cosine_similarity,
lexical_similarity,
normalize_checklist,
normalize_labels,
unresolved_blockers,
validate_no_cycle,
)
_UNSET = object()
class NotFound(Exception):
"""Target resource does not exist or caller is not permitted to see it (HTTP 404)."""
def _parse_due_date(val: Any) -> Optional[date]:
if val is None or val == "":
return None
if isinstance(val, date):
return val
try:
return date.fromisoformat(str(val).strip()[:10])
except ValueError as exc:
raise Invalid(f"invalid due_date {val!r}; expected YYYY-MM-DD") from exc
def _serialize_card(
c: Card,
today: date,
done_lists: set[str],
blockers_map: Optional[dict[str, list[str]]] = None,
cards_by_id: Optional[dict[str, Card]] = None,
) -> dict:
b_ids = list((blockers_map or {}).get(c.id, []))
unresolved_ids = []
if cards_by_id:
unresolved_ids = [
bid for bid in b_ids if bid in cards_by_id and not cards_by_id[bid].is_done(done_lists)
]
return {
"id": c.id,
"board_id": c.board_id,
"title": c.title,
"description": c.description,
"list_name": c.list_name,
"assignee": c.assignee,
"due_date": c.due_date.isoformat() if c.due_date else None,
"labels": list(c.labels),
"checklist": list(c.checklist),
"position": c.position,
"overdue": c.is_overdue(today, done_lists),
"blocked_by": b_ids,
"unresolved_blockers": unresolved_ids,
"is_blocked": len(unresolved_ids) > 0,
}
class Boards:
def __init__(
self,
repo: BoardRepository,
assistant: Optional[Assistant] = None,
clock=date.today,
):
self.repo = repo
self.assistant = assistant
self._today = clock
# ------------------------------------------------------------------
# Internal lookup & authorization helpers
# ------------------------------------------------------------------
def _board(self, board_id_or_title: str, person: Person) -> Board:
key = (board_id_or_title or "").strip()
if not key:
raise NotFound("no such board")
board = self.repo.get(key)
if board is not None:
try:
board.require_view(person)
return board
except Forbidden as exc:
raise NotFound("no such board") from exc
# Allow natural-language lookup by exact board title among visible boards (for MCP ergonomics)
visible = self.repo.boards_for(person.email.strip().lower())
matches = [b for b in visible if b.title.lower() == key.lower()]
if len(matches) == 1:
return matches[0]
raise NotFound("no such board")
def _find_card(self, card_id: str, person: Person) -> tuple[Board, Card]:
key = (card_id or "").strip()
if not key:
raise NotFound("no such card")
card = self.repo.get_card(key)
if card is not None:
board = self._board(card.board_id, person)
return board, card
# Support short unambiguous ID prefix across visible boards (e.g. "move card 9a3f to Done")
if len(key) >= 4:
matches: list[tuple[Board, Card]] = []
for b in self.repo.boards_for(person.email.strip().lower()):
for c in self.repo.cards_on(b.id):
if c.id.startswith(key.lower()):
matches.append((b, c))
if len(matches) == 1:
return matches[0]
raise NotFound("no such card")
def _maybe_embed(self, card: Card) -> None:
if self.assistant is None:
return
try:
vec = self.assistant.embed(card.searchable_text())
if vec:
card.embedding = vec
except Exception:
# Never fail core task operations if the optional embedding endpoint is temporarily unreachable
pass
# ------------------------------------------------------------------
# Boards, Memberships & Custom Lists
# ------------------------------------------------------------------
def create_board(self, person: Person, title: str, description: str = "") -> dict:
board = Board(
id=uuid.uuid4().hex[:12],
title=title,
description=description,
owner_email=person.email,
)
self.repo.save_board(board)
self.repo.log(board.id, None, person.email, f"created board {board.title!r}")
return {
"id": board.id,
"title": board.title,
"description": board.description,
"role": Role.OWNER.value,
"lists": [{"name": l.name, "position": l.position, "is_done": l.is_done} for l in board.lists],
}
def my_boards(self, person: Person) -> list[dict]:
today = self._today()
email = person.email.strip().lower()
out = []
for b in self.repo.boards_for(email):
role = b.role_of(person)
if role is None:
continue
cards = self.repo.cards_on(b.id)
done_set = b.done_list_names()
overdue_count = sum(1 for c in cards if c.is_overdue(today, done_set))
out.append({
"id": b.id,
"title": b.title,
"description": b.description,
"owner_email": b.owner_email,
"role": role.value,
"member_count": 1 + len(b.members),
"card_count": len(cards),
"overdue_count": overdue_count,
})
return out
def show_board(self, person: Person, board_id: str) -> dict:
today = self._today()
board = self._board(board_id, person)
cards = sorted(self.repo.cards_on(board.id), key=lambda c: (c.position, c.id))
cards_by_id = {c.id: c for c in cards}
deps = self.repo.dependencies_on(board.id)
done_set = board.done_list_names()
members_list = [{"email": board.owner_email, "role": Role.OWNER.value}] + [
{"email": em, "role": rl.value} for em, rl in sorted(board.members.items())
]
return {
"id": board.id,
"title": board.title,
"description": board.description,
"owner_email": board.owner_email,
"viewer_email": person.email.strip().lower(),
"role": board.role_of(person).value,
"lists": [
{"name": l.name, "position": l.position, "is_done": l.is_done}
for l in sorted(board.lists, key=lambda x: x.position)
],
"members": members_list,
"cards": [_serialize_card(c, today, done_set, deps, cards_by_id) for c in cards],
}
def add_member(self, person: Person, board_id: str, email: str, role: str = "member") -> dict:
board = self._board(board_id, person)
try:
parsed_role = Role(role.strip().lower())
except ValueError as exc:
raise Invalid(f"unknown role {role!r}; choose 'member' or 'viewer'") from exc
board.add_member(person, email, parsed_role)
clean_email = email.strip().lower()
self.repo.save_board(board)
self.repo.log(board.id, None, person.email, f"added {clean_email} as {parsed_role.value}")
return {"board_id": board.id, "email": clean_email, "role": parsed_role.value}
def remove_member(self, person: Person, board_id: str, email: str) -> dict:
board = self._board(board_id, person)
clean_email = (email or "").strip().lower()
board.remove_member(person, clean_email)
self.repo.remove_member_row(board.id, clean_email)
self.repo.save_board(board)
self.repo.log(board.id, None, person.email, f"removed {clean_email} from board")
return {"board_id": board.id, "removed": clean_email}
def add_list(self, person: Person, board_id: str, name: str, is_done: bool = False) -> dict:
board = self._board(board_id, person)
new_lst = board.add_list(person, name, is_done=is_done)
self.repo.save_board(board)
self.repo.log(board.id, None, person.email, f"added list {new_lst.name!r}")
return {"board_id": board.id, "name": new_lst.name, "position": new_lst.position, "is_done": new_lst.is_done}
def remove_list(self, person: Person, board_id: str, name: str) -> dict:
board = self._board(board_id, person)
canonical = board.canonical_list(name)
count = sum(1 for c in self.repo.cards_on(board.id) if c.list_name == canonical)
board.remove_list(person, canonical, count)
self.repo.save_board(board)
self.repo.log(board.id, None, person.email, f"removed list {canonical!r}")
return {"board_id": board.id, "removed": canonical}
# ------------------------------------------------------------------
# Cards & Drag-and-Drop Lifecycle
# ------------------------------------------------------------------
def add_card(
self,
person: Person,
board_id: str,
title: str,
list_name: str = "To do",
description: str = "",
assignee: Optional[str] = None,
due_date: Any = None,
labels: Optional[list[str]] = None,
) -> dict:
board = self._board(board_id, person)
board.require_edit(person)
canonical_list = board.canonical_list(list_name)
existing = [c for c in self.repo.cards_on(board.id) if c.list_name == canonical_list]
next_pos = max((c.position for c in existing), default=-1) + 1
card = Card(
id=uuid.uuid4().hex[:12],
board_id=board.id,
title=title,
list_name=canonical_list,
description=description,
assignee=assignee,
due_date=_parse_due_date(due_date),
labels=normalize_labels(labels),
position=next_pos,
)
self._maybe_embed(card)
self.repo.save_card(card)
self.repo.log(board.id, card.id, person.email, f"added card {card.title!r} to {canonical_list}")
return _serialize_card(card, self._today(), board.done_list_names())
def update_card(
self,
person: Person,
card_id: str,
title: Any = _UNSET,
description: Any = _UNSET,
assignee: Any = _UNSET,
due_date: Any = _UNSET,
labels: Any = _UNSET,
checklist: Any = _UNSET,
) -> dict:
board, card = self._find_card(card_id, person)
board.require_edit(person)
changes = []
text_changed = False
if title is not _UNSET and title is not None:
new_title = str(title).strip()
if not new_title:
raise Invalid("a card needs a title")
if new_title != card.title:
changes.append(f"title {card.title!r} -> {new_title!r}")
card.title = new_title[:240]
text_changed = True
if description is not _UNSET and description is not None:
new_desc = str(description).strip()
if new_desc != card.description:
changes.append("updated description")
card.description = new_desc
text_changed = True
if assignee is not _UNSET:
new_assignee = str(assignee).strip().lower() if assignee else None
if new_assignee != card.assignee:
changes.append(f"assignee -> {new_assignee or 'unassigned'}")
card.assignee = new_assignee
if due_date is not _UNSET:
new_due = _parse_due_date(due_date)
if new_due != card.due_date:
changes.append(f"due_date -> {new_due.isoformat() if new_due else 'none'}")
card.due_date = new_due
if labels is not _UNSET and labels is not None:
new_labels = normalize_labels(labels)
if new_labels != card.labels:
changes.append(f"labels -> [{', '.join(new_labels)}]")
card.labels = new_labels
text_changed = True
if checklist is not _UNSET and checklist is not None:
new_cl = normalize_checklist(checklist)
if new_cl != card.checklist:
done_n = sum(1 for i in new_cl if i["done"])
changes.append(f"checklist ({done_n}/{len(new_cl)} done)")
card.checklist = new_cl
if text_changed:
self._maybe_embed(card)
self.repo.save_card(card)
if changes:
self.repo.log(board.id, card.id, person.email, f"updated {card.title!r}: " + "; ".join(changes))
cards_by_id = {c.id: c for c in self.repo.cards_on(board.id)}
deps = self.repo.dependencies_on(board.id)
return _serialize_card(card, self._today(), board.done_list_names(), deps, cards_by_id)
def assign_card(self, person: Person, card_id: str, assignee: Optional[str]) -> dict:
return self.update_card(person, card_id, assignee=assignee)
def move_card(
self,
person: Person,
card_id: str,
list_name: str,
position: Optional[int] = None,
) -> dict:
board, card = self._find_card(card_id, person)
board.require_edit(person)
before = card.list_name
canonical = board.canonical_list(list_name)
card.move_to(canonical, position=position)
self.repo.save_card(card)
if before != canonical:
self.repo.log(board.id, card.id, person.email, f"moved {card.title!r} from {before} to {canonical}")
elif position is not None:
self.repo.log(board.id, card.id, person.email, f"reordered {card.title!r} in {canonical}")
cards_by_id = {c.id: c for c in self.repo.cards_on(board.id)}
deps = self.repo.dependencies_on(board.id)
return _serialize_card(card, self._today(), board.done_list_names(), deps, cards_by_id)
def delete_card(self, person: Person, card_id: str) -> dict:
board, card = self._find_card(card_id, person)
board.require_edit(person)
self.repo.delete_card(card.id)
self.repo.log(board.id, card.id, person.email, f"deleted card {card.title!r}")
return {"deleted": card.id, "board_id": board.id, "title": card.title}
# ------------------------------------------------------------------
# Activity History, Personal Agenda & Overdue Detection
# ------------------------------------------------------------------
def board_activity(self, person: Person, board_id: str, limit: int = 40) -> list[dict]:
board = self._board(board_id, person)
return self.repo.activity_for_board(board.id, limit=max(1, min(int(limit), 200)))
def card_history(self, person: Person, card_id: str) -> list[dict]:
_, card = self._find_card(card_id, person)
return self.repo.activity_for_card(card.id, limit=50)
def list_overdue(self, person: Person, board_id: Optional[str] = None) -> list[dict]:
today = self._today()
boards_to_scan = (
[self._board(board_id, person)]
if board_id
else self.repo.boards_for(person.email.strip().lower())
)
overdue_items = []
for b in boards_to_scan:
if b.role_of(person) is None:
continue
done_set = b.done_list_names()
cards = self.repo.cards_on(b.id)
cards_by_id = {c.id: c for c in cards}
deps = self.repo.dependencies_on(b.id)
for c in cards:
if c.is_overdue(today, done_set):
item = _serialize_card(c, today, done_set, deps, cards_by_id)
item["board_title"] = b.title
item["days_overdue"] = (today - c.due_date).days if c.due_date else 0
overdue_items.append(item)
overdue_items.sort(key=lambda x: x["due_date"] or "")
return overdue_items
def my_agenda(self, person: Person, days: int = 7) -> dict:
"""Answer 'what's on my plate this week?' across every board the signed-in person can see."""
today = self._today()
horizon = today + timedelta(days=max(1, int(days)))
me = person.email.strip().lower()
overdue, due_soon, in_progress, other_assigned = [], [], [], []
for b in self.repo.boards_for(me):
if b.role_of(person) is None:
continue
done_set = b.done_list_names()
cards = self.repo.cards_on(b.id)
cards_by_id = {c.id: c for c in cards}
deps = self.repo.dependencies_on(b.id)
for c in cards:
if c.is_done(done_set):
continue
# Include cards explicitly assigned to me, or unassigned cards on boards I own solo
is_mine = (c.assignee == me) or (c.assignee is None and b.owner_email == me and len(b.members) == 0)
if not is_mine:
continue
entry = _serialize_card(c, today, done_set, deps, cards_by_id)
entry["board_title"] = b.title
if c.is_overdue(today, done_set):
overdue.append(entry)
elif c.due_date and today <= c.due_date <= horizon:
due_soon.append(entry)
elif c.list_name.lower() == "doing":
in_progress.append(entry)
else:
other_assigned.append(entry)
return {
"person": me,
"as_of": today.isoformat(),
"horizon_days": days,
"overdue": overdue,
"due_this_week": due_soon,
"doing": in_progress,
"backlog_assigned": other_assigned,
"total_open": len(overdue) + len(due_soon) + len(in_progress) + len(other_assigned),
}
# ------------------------------------------------------------------
# Card Dependencies ("blocked by" & "what is blocking the launch?")
# ------------------------------------------------------------------
def add_blocker(self, person: Person, card_id: str, blocked_by_id: str) -> dict:
board, card = self._find_card(card_id, person)
board.require_edit(person)
b2, blocker = self._find_card(blocked_by_id, person)
if board.id != b2.id:
raise Invalid("dependencies must link cards on the same board")
deps = self.repo.dependencies_on(board.id)
validate_no_cycle(card.id, blocker.id, deps)
if blocker.id not in deps.get(card.id, []):
self.repo.add_dependency(board.id, card.id, blocker.id)
self.repo.log(
board.id,
card.id,
person.email,
f"marked {card.title!r} as blocked by {blocker.title!r}",
)
deps = self.repo.dependencies_on(board.id)
cards_by_id = {c.id: c for c in self.repo.cards_on(board.id)}
return _serialize_card(card, self._today(), board.done_list_names(), deps, cards_by_id)
def remove_blocker(self, person: Person, card_id: str, blocked_by_id: str) -> dict:
board, card = self._find_card(card_id, person)
board.require_edit(person)
_, blocker = self._find_card(blocked_by_id, person)
self.repo.remove_dependency(card.id, blocker.id)
self.repo.log(
board.id,
card.id,
person.email,
f"removed blocker {blocker.title!r} from {card.title!r}",
)
deps = self.repo.dependencies_on(board.id)
cards_by_id = {c.id: c for c in self.repo.cards_on(board.id)}
return _serialize_card(card, self._today(), board.done_list_names(), deps, cards_by_id)
def what_is_blocking(self, person: Person, target: str, board_id: Optional[str] = None) -> dict:
"""Answer 'what is blocking the launch?' by card ID or title keyword across visible boards."""
today = self._today()
needle = (target or "").strip()
if not needle:
raise Invalid("provide a card id or title keyword to inspect blockers")
boards_to_scan = (
[self._board(board_id, person)]
if board_id
else self.repo.boards_for(person.email.strip().lower())
)
reports = []
for b in boards_to_scan:
if b.role_of(person) is None:
continue
cards = self.repo.cards_on(b.id)
cards_by_id = {c.id: c for c in cards}
deps = self.repo.dependencies_on(b.id)
done_set = b.done_list_names()
matched = [
c
for c in cards
if c.id == needle or c.id.startswith(needle.lower()) or needle.lower() in c.title.lower()
]
for c in matched:
direct_open, transitive_open = unresolved_blockers(c.id, cards_by_id, deps, done_set)
reports.append({
"board_id": b.id,
"board_title": b.title,
"card": _serialize_card(c, today, done_set, deps, cards_by_id),
"is_blocked": bool(direct_open or transitive_open),
"direct_blockers": [_serialize_card(x, today, done_set, deps, cards_by_id) for x in direct_open],
"upstream_blockers": [_serialize_card(x, today, done_set, deps, cards_by_id) for x in transitive_open],
})
if not reports:
raise NotFound(f"no visible card matching {needle!r}")
return {"query": needle, "matches": reports}
# ------------------------------------------------------------------
# Dodil Models AI Features (Checklist, Semantic Search, Daily Summary)
# ------------------------------------------------------------------
def suggest_checklist(self, person: Person, card_id: str, apply: bool = False) -> dict:
board, card = self._find_card(card_id, person)
if self.assistant is None:
raise Invalid("no AI assistant configured")
if apply:
board.require_edit(person)
steps = self.assistant.checklist(card.title, card.description)
if apply and steps:
existing_texts = {item["text"].lower() for item in card.checklist}
merged = list(card.checklist) + [
{"text": s[:240], "done": False} for s in steps if s.lower() not in existing_texts
]
card.checklist = merged[:30]
self.repo.save_card(card)
self.repo.log(board.id, card.id, person.email, f"applied AI checklist ({len(steps)} steps) to {card.title!r}")
return {
"card_id": card.id,
"title": card.title,
"applied": bool(apply),
"suggested_checklist": steps,
"checklist": card.checklist,
}
def search_similar_cards(
self,
person: Person,
query_or_card_id: str,
board_id: Optional[str] = None,
limit: int = 5,
) -> list[dict]:
"""Semantic search ('find cards like this one') using Dodil Models embeddings + lexical hybrid."""
today = self._today()
q = (query_or_card_id or "").strip()
if not q:
raise Invalid("search query cannot be empty")
exclude_id: Optional[str] = None
query_text = q
query_vec: Optional[list[float]] = None
# Check if `q` refers to an existing card ID
existing_card = self.repo.get_card(q)
if existing_card is not None:
try:
self._board(existing_card.board_id, person)
exclude_id = existing_card.id
query_text = existing_card.searchable_text()
query_vec = existing_card.embedding
except NotFound:
existing_card = None
if query_vec is None and self.assistant is not None:
try:
query_vec = self.assistant.embed(query_text)
except Exception:
query_vec = None
boards_to_scan = (
[self._board(board_id, person)]
if board_id
else self.repo.boards_for(person.email.strip().lower())
)
scored = []
for b in boards_to_scan:
if b.role_of(person) is None:
continue
done_set = b.done_list_names()
cards = self.repo.cards_on(b.id)
cards_by_id = {c.id: c for c in cards}
deps = self.repo.dependencies_on(b.id)
for c in cards:
if c.id == exclude_id:
continue
if c.embedding is None and query_vec is not None and self.assistant is not None:
self._maybe_embed(c)
if c.embedding is not None:
self.repo.save_card(c)
sem = cosine_similarity(query_vec, c.embedding) if (query_vec and c.embedding) else 0.0
lex = lexical_similarity(query_text, c.searchable_text())
score = (0.75 * sem + 0.25 * lex) if (query_vec and c.embedding) else lex
if score > 0.02:
item = _serialize_card(c, today, done_set, deps, cards_by_id)
item["board_title"] = b.title
item["similarity"] = round(score, 4)
scored.append(item)
scored.sort(key=lambda x: x["similarity"], reverse=True)
return scored[: max(1, min(int(limit), 20))]
def summarize_board(self, person: Person, board_id: str) -> dict:
today = self._today()
board = self._board(board_id, person)
if self.assistant is None:
raise Invalid("no AI assistant configured")
detail = self.show_board(person, board.id)
recent = self.repo.activity_for_board(board.id, limit=12)
by_list = {l["name"]: 0 for l in detail["lists"]}
overdue_titles, blocked_titles = [], []
for c in detail["cards"]:
by_list[c["list_name"]] = by_list.get(c["list_name"], 0) + 1
if c["overdue"]:
overdue_titles.append(f"{c['title']} (due {c['due_date']}, assignee: {c['assignee'] or 'unassigned'})")
if c["is_blocked"]:
blocked_titles.append(f"{c['title']} (blocked by {len(c['unresolved_blockers'])} card(s))")
snapshot_lines = [
f"Board: {board.title} (as of {today.isoformat()})",
"Counts by list: " + ", ".join(f"{k}: {v}" for k, v in by_list.items()),
"Overdue cards: " + ("; ".join(overdue_titles) if overdue_titles else "none"),
"Blocked cards: " + ("; ".join(blocked_titles) if blocked_titles else "none"),
"Recent activity: " + ("; ".join(f"{e['actor']} {e['action']}" for e in recent[:8]) if recent else "none"),
]
summary_text = self.assistant.summarize_board(board.title, "\n".join(snapshot_lines))
return {
"board_id": board.id,
"board_title": board.title,
"counts_by_list": by_list,
"overdue_count": len(overdue_titles),
"blocked_count": len(blocked_titles),
"summary": summary_text,
}
def self_check(self) -> dict:
return {"ok": True, "app": "kanvaflow", "tables": self.repo.health_counts()}