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/domain/model.py
"""KanvaFlow Domain Layer: pure Python entities, value objects, and business invariants.
No database, no HTTP, no MCP imports. Every rule written here holds identically for
both the web UI and Claude / ChatGPT over MCP.
"""
import math
import re
from dataclasses import dataclass, field
from datetime import date
from enum import Enum
from typing import Optional
DEFAULT_LISTS = ("To do", "Doing", "Done")
class Role(str, Enum):
OWNER = "owner"
MEMBER = "member"
VIEWER = "viewer"
class Forbidden(Exception):
"""The signed-in person is not allowed to perform this action (HTTP 403 / MCP tool error)."""
class Invalid(Exception):
"""The action violates a domain invariant (HTTP 422 / MCP tool error)."""
@dataclass(frozen=True)
class Person:
"""Authenticated caller identity injected by the Dodil gateway, never from request payloads."""
email: str
@dataclass
class BoardList:
name: str
position: int
is_done: bool = False
def default_board_lists() -> list[BoardList]:
return [
BoardList(name="To do", position=0, is_done=False),
BoardList(name="Doing", position=1, is_done=False),
BoardList(name="Done", position=2, is_done=True),
]
def normalize_labels(labels: Optional[list[str]]) -> list[str]:
if not labels:
return []
seen = set()
cleaned = []
for raw in labels:
tag = str(raw).strip()
if tag and tag.lower() not in seen:
seen.add(tag.lower())
cleaned.append(tag[:40])
return cleaned[:12]
def normalize_checklist(items: Optional[list]) -> list[dict]:
if not items:
return []
out = []
for item in items:
if isinstance(item, str):
txt = item.strip()
if txt:
out.append({"text": txt[:240], "done": False})
elif isinstance(item, dict):
txt = str(item.get("text", "")).strip()
if txt:
out.append({"text": txt[:240], "done": bool(item.get("done", False))})
return out[:30]
@dataclass
class Card:
id: str
board_id: str
title: str
list_name: str = DEFAULT_LISTS[0]
description: str = ""
assignee: Optional[str] = None
due_date: Optional[date] = None
labels: list[str] = field(default_factory=list)
checklist: list[dict] = field(default_factory=list)
position: int = 0
embedding: Optional[list[float]] = None
def __post_init__(self) -> None:
self.title = (self.title or "").strip()
if not self.title:
raise Invalid("a card needs a title")
if len(self.title) > 240:
raise Invalid("card title cannot exceed 240 characters")
self.description = (self.description or "").strip()
if self.assignee is not None:
cleaned_email = self.assignee.strip().lower()
self.assignee = cleaned_email if cleaned_email else None
self.labels = normalize_labels(self.labels)
self.checklist = normalize_checklist(self.checklist)
def move_to(self, canonical_list_name: str, position: Optional[int] = None) -> None:
self.list_name = canonical_list_name
if position is not None:
self.position = max(0, int(position))
def is_done(self, done_list_names: set[str]) -> bool:
return self.list_name.lower() in {n.lower() for n in done_list_names}
def is_overdue(self, today: date, done_list_names: set[str]) -> bool:
return bool(self.due_date and self.due_date < today and not self.is_done(done_list_names))
def searchable_text(self) -> str:
parts = [self.title, self.description, " ".join(self.labels)]
return " ".join(p for p in parts if p).strip()
@dataclass
class Board:
id: str
title: str
owner_email: str
description: str = ""
members: dict[str, Role] = field(default_factory=dict) # email -> Role (excluding owner)
lists: list[BoardList] = field(default_factory=default_board_lists)
def __post_init__(self) -> None:
self.title = (self.title or "").strip()
if not self.title:
raise Invalid("a board needs a title")
self.owner_email = self.owner_email.strip().lower()
self.description = (self.description or "").strip()
if not self.lists:
self.lists = default_board_lists()
else:
self.lists = sorted(self.lists, key=lambda x: x.position)
def done_list_names(self) -> set[str]:
done = {lst.name for lst in self.lists if lst.is_done}
if not done:
done = {lst.name for lst in self.lists if lst.name.lower() == "done"}
return done
def canonical_list(self, requested_name: str) -> str:
needle = (requested_name or "").strip().lower()
aliases = {"todo": "to do", "in progress": "doing", "completed": "done"}
normalized = aliases.get(needle, needle)
for lst in self.lists:
if lst.name.lower() == needle or lst.name.lower() == normalized:
return lst.name
valid = ", ".join(lst.name for lst in self.lists)
raise Invalid(f"unknown list {requested_name!r}; lists are {valid}")
def role_of(self, person: Person) -> Optional[Role]:
email = person.email.strip().lower()
if email == self.owner_email:
return Role.OWNER
return self.members.get(email)
def require_view(self, person: Person) -> None:
if self.role_of(person) is None:
raise Forbidden("you are not on this board")
def require_edit(self, person: Person) -> None:
role = self.role_of(person)
if role is None:
raise Forbidden("you are not on this board")
if role is Role.VIEWER:
raise Forbidden("viewers can't change the board")
def require_owner(self, person: Person) -> None:
if self.role_of(person) is not Role.OWNER:
raise Forbidden("only the owner manages members")
def add_member(self, by: Person, email: str, role: Role) -> None:
self.require_owner(by)
if role is Role.OWNER:
raise Invalid("a board has one owner")
clean_email = (email or "").strip().lower()
if not clean_email or "@" not in clean_email:
raise Invalid("valid member email required")
if clean_email == self.owner_email:
raise Invalid("owner is already on the board")
self.members[clean_email] = role
def remove_member(self, by: Person, email: str) -> None:
self.require_owner(by)
clean_email = (email or "").strip().lower()
if clean_email == self.owner_email:
raise Invalid("cannot remove the board owner")
if clean_email not in self.members:
raise Invalid(f"{clean_email} is not a member of this board")
del self.members[clean_email]
def add_list(self, by: Person, name: str, is_done: bool = False) -> BoardList:
self.require_edit(by)
clean = (name or "").strip()
if not clean:
raise Invalid("list name cannot be empty")
if any(lst.name.lower() == clean.lower() for lst in self.lists):
raise Invalid(f"list {clean!r} already exists")
next_pos = max((lst.position for lst in self.lists), default=-1) + 1
new_lst = BoardList(name=clean[:60], position=next_pos, is_done=bool(is_done))
self.lists.append(new_lst)
return new_lst
def remove_list(self, by: Person, name: str, card_count_in_list: int) -> str:
self.require_edit(by)
canonical = self.canonical_list(name)
if len(self.lists) <= 1:
raise Invalid("a board must keep at least one list")
if card_count_in_list > 0:
raise Invalid(f"move or delete the {card_count_in_list} card(s) in {canonical!r} first")
self.lists = [lst for lst in self.lists if lst.name != canonical]
for idx, lst in enumerate(self.lists):
lst.position = idx
return canonical
def validate_no_cycle(card_id: str, blocked_by_id: str, blockers_by_card: dict[str, list[str]]) -> None:
"""Ensure adding `card_id is blocked by blocked_by_id` does not create a self-loop or cycle."""
if card_id == blocked_by_id:
raise Invalid("a card cannot block itself")
# If `card_id` is already reachable upstream from `blocked_by_id`, adding `card_id -> blocked_by_id` cycles.
stack = [blocked_by_id]
visited = set()
while stack:
curr = stack.pop()
if curr == card_id:
raise Invalid("dependency cycle detected")
if curr not in visited:
visited.add(curr)
stack.extend(blockers_by_card.get(curr, []))
def unresolved_blockers(
target_id: str,
cards_by_id: dict[str, Card],
blockers_by_card: dict[str, list[str]],
done_lists: set[str],
) -> tuple[list[Card], list[Card]]:
"""Return (direct_unfinished_blockers, transitive_unfinished_blockers) for `target_id`."""
direct_ids = blockers_by_card.get(target_id, [])
direct_open = [
cards_by_id[bid]
for bid in direct_ids
if bid in cards_by_id and not cards_by_id[bid].is_done(done_lists)
]
all_open: list[Card] = []
seen = set()
queue = list(direct_ids)
while queue:
cid = queue.pop(0)
if cid in seen or cid not in cards_by_id:
continue
seen.add(cid)
c = cards_by_id[cid]
if not c.is_done(done_lists):
all_open.append(c)
queue.extend(blockers_by_card.get(cid, []))
direct_set = {c.id for c in direct_open}
transitive_only = [c for c in all_open if c.id not in direct_set]
return direct_open, transitive_only
def cosine_similarity(a: Optional[list[float]], b: Optional[list[float]]) -> float:
if not a or not b or len(a) != len(b):
return 0.0
dot = sum(x * y for x, y in zip(a, b))
na = math.sqrt(sum(x * x for x in a))
nb = math.sqrt(sum(y * y for y in b))
if na == 0.0 or nb == 0.0:
return 0.0
return dot / (na * nb)
_TOKEN_RE = re.compile(r"[a-z0-9]{2,}")
def lexical_similarity(query: str, text: str) -> float:
q_tokens = set(_TOKEN_RE.findall((query or "").lower()))
t_tokens = set(_TOKEN_RE.findall((text or "").lower()))
if not q_tokens or not t_tokens:
return 0.0
overlap = len(q_tokens & t_tokens)
return overlap / math.sqrt(len(q_tokens) * len(t_tokens))