rgboards/rg-boards
Lanes, team task boards you run from the web or from Claude/ChatGPT (MCP). FastAPI + DataK3 + appid + Dodil Models.
A remix template · 0 remixes · Remix rgboards/rg-boards
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.
Reads are scoped by the caller inside the use case: a board you are not on does not exist for
you (404), whichever door you came through, so Claude sees exactly what its person sees.
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 concurrent.futures import ThreadPoolExecutor, TimeoutError as Late
from datetime import date, datetime, timedelta, timezone
from typing import Callable, Optional
from boards.application.ports import Assistant, BoardRepository, DuplicateKey
from boards.domain.model import (DEFAULT_LISTS, Board, Card, Conflict, Invalid,
Person, Role, check_no_cycle, clean_description, clean_labels,
clean_title, parse_due)
class NotFound(Exception):
"""Doesn't exist, or exists on a board you are not on: the two look the same on purpose."""
class Unavailable(Exception):
"""Dodil Models didn't answer. 503 / tool error; nothing was written."""
def _utcnow() -> datetime:
return datetime.now(timezone.utc)
def _iso(d: Optional[date]) -> Optional[str]:
return d.isoformat() if d else None
class Boards:
def __init__(self, repo: BoardRepository, assistant: Optional[Assistant] = None,
today: Callable[[], date] = date.today, ai_wait: float = 40.0,
now: Callable[[], datetime] = _utcnow):
self.repo, self.assistant, self.today, self.now = repo, assistant, today, now
self.ai_wait = ai_wait # stay under the platform's request timeout
self._ai_pool = ThreadPoolExecutor(max_workers=4)
self._writing: dict = {} # board_id:day -> Future (summary in flight)
# ------------------------------------------------------------------ lookups (scoped)
def _board(self, board_ref: str, person: Person) -> Board:
"""A board by id, or by title among the boards this person is on."""
ref = (board_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
matches = [b for b in self.repo.boards_for(person.email)
if b.title.lower() == ref.lower()]
if not matches: # also accept "the Ops board" -> "Ops"
short = ref.lower().removeprefix("the ").removesuffix(" board").strip()
matches = [b for b in self.repo.boards_for(person.email)
if b.title.lower() == short]
if len(matches) == 1:
return matches[0]
if len(matches) > 1:
raise Invalid(f"you are on {len(matches)} boards called {ref!r}; use the board id")
raise NotFound("no such board") # a board you can't see doesn't exist for you
def _card(self, card_ref, person: Person, board_ref: Optional[str] = None) -> tuple:
"""A card by id ('ab12cd34ef56-7'), or by its number ('7', '#7') on a board you can see."""
ref = str(card_ref or "").strip().lstrip("#")
if not ref:
raise Invalid("which card? give its number or id")
if not ref.isdigit():
card = self.repo.get_card(ref)
if card is not None and not card.archived:
board = self.repo.get(card.board_id)
if board is not None and board.role_of(person) is not None:
return board, card
raise NotFound("no such card") # same answer as a card that doesn't exist
boards = ([self._board(board_ref, person)] if board_ref
else self.repo.boards_for(person.email))
by_id = {b.id: b for b in boards}
if not ref.isdigit():
return self._card_by_title(ref, by_id)
number = int(ref)
hits = [c for c in self.repo.cards_on(list(by_id)) if c.number == number]
if not hits:
raise NotFound(f"no card #{number}" + (" on that board" if board_ref else ""))
if len(hits) > 1:
names = ", ".join(sorted(by_id[c.board_id].title for c in hits))
raise Invalid(f"card #{number} exists on several boards ({names}); say which board")
return by_id[hits[0].board_id], hits[0]
def _card_by_title(self, text: str, by_id: dict) -> tuple:
"""'the launch card' -> the one card whose title matches: exact title first, then a
title that starts with the words, then one that contains them all. More than one
match is refused with the candidates, so the caller (often an AI) can pick."""
want = text.lower().removeprefix("the ").removesuffix(" card").strip()
words = want.split()
cards = self.repo.cards_on(list(by_id)) if by_id else []
for test in (lambda t: t == want, lambda t: t.startswith(want),
lambda t: all(w in t for w in words)):
hits = [c for c in cards if test(c.title.lower())]
if len(hits) == 1:
return by_id[hits[0].board_id], hits[0]
if len(hits) > 1:
names = "; ".join(f"#{c.number} {c.title} ({by_id[c.board_id].title})"
for c in sorted(hits, key=lambda c: c.number)[:8])
raise Invalid(f"{len(hits)} cards match {text!r}: {names}. Use the number")
raise NotFound("no such card")
def _retrying(self, write: Callable[[], dict]) -> dict:
"""Moves, assignments and archiving re-read the card and try again when someone else
changed it at the same moment (the compare-and-set lost), instead of failing the drag."""
for _ in range(3):
try:
return write()
except Conflict:
continue
return write()
def _log(self, board_id: str, card_id: Optional[str], actor: str, action: str,
field: Optional[str] = None, old: Optional[str] = None,
new: Optional[str] = None) -> None:
self.repo.log(board_id, card_id, actor, action, field, old, new, at=self.now())
def _view(self, board: Board, card: Card, blockers: Optional[dict] = None,
cards_by_id: Optional[dict] = None) -> dict:
today = self.today()
open_blockers = []
for bid in (blockers or {}).get(card.id, []):
b = (cards_by_id or {}).get(bid)
if b and b.list_name != board.done_list and not b.archived:
open_blockers.append({"id": b.id, "number": b.number, "title": b.title})
return {"id": card.id, "number": card.number, "board_id": card.board_id,
"board": board.title, "title": card.title, "description": card.description,
"list_name": card.list_name, "assignee": card.assignee,
"due_date": _iso(card.due_date), "labels": list(card.labels),
"position": card.position, "version": card.version,
"overdue": card.is_overdue(today, board.done_list),
"blocked_by": open_blockers}
def _blockers(self, board_id: str) -> dict:
out: dict = {}
for card_id, blocked_by in self.repo.links_on(board_id):
out.setdefault(card_id, []).append(blocked_by)
return out
def _save(self, card: Card, expected_version: Optional[int]) -> None:
version = card.version if expected_version is None else int(expected_version)
if expected_version is not None and int(expected_version) != card.version:
raise Conflict("this card changed since you opened it; reload and try again")
card.version = version + 1
if not self.repo.update_card(card, version):
card.version = version
raise Conflict("this card changed since you opened it; reload and try again")
def _log_changes(self, board: Board, card: Card, actor: str, before: dict) -> list:
changes = []
for name, old in before.items():
new = getattr(card, name)
if name == "due_date":
old, new = _iso(old), _iso(new)
if name == "labels":
old, new = ", ".join(old), ", ".join(new)
if old != new:
self._log(board.id, card.id, actor, f"changed {name.replace('_', ' ')}",
field=name, old=None if old is None else str(old),
new=None if new is None else str(new))
changes.append(name)
return changes
# ------------------------------------------------------------------ boards
def create_board(self, person: Person, title: str) -> dict:
board = Board(id=uuid.uuid4().hex[:12], title=clean_title(title, "board"),
owner_email=person.email, lists=list(DEFAULT_LISTS))
self.repo.save_board(board)
self._log(board.id, None, person.email, f"created board {board.title!r}")
return {"id": board.id, "title": board.title, "lists": board.lists, "role": "owner"}
def my_boards(self, person: Person) -> list:
boards = self.repo.boards_for(person.email)
cards = self.repo.cards_on([b.id for b in boards]) if boards else []
today, out = self.today(), []
for b in sorted(boards, key=lambda b: b.title.lower()):
mine = [c for c in cards if c.board_id == b.id]
out.append({"id": b.id, "title": b.title, "role": b.role_of(person).value,
"cards": len(mine),
"overdue": sum(c.is_overdue(today, b.done_list) for c in mine)})
return out
def show_board(self, person: Person, board_ref: str) -> dict:
board = self._board(board_ref, person)
cards = self.repo.cards_on([board.id])
by_id = {c.id: c for c in cards}
blockers = self._blockers(board.id)
views = [self._view(board, c, blockers, by_id) for c in cards]
lists = []
for name in board.lists:
in_list = sorted((v for v in views if v["list_name"] == name),
key=lambda v: (v["position"], v["number"]))
lists.append({"name": name, "cards": in_list})
return {"id": board.id, "title": board.title, "role": board.role_of(person).value,
"owner": board.owner_email, "done_list": board.done_list, "lists": lists,
"members": self._members(board)}
def _members(self, board: Board) -> list:
return ([{"email": board.owner_email, "role": "owner"}] +
[{"email": e, "role": r.value} for e, r in sorted(board.members.items())])
def rename_board(self, person: Person, board_ref: str, title: str) -> dict:
board = self._board(board_ref, person)
board.require_owner(person, "renames the board")
old, board.title = board.title, clean_title(title, "board")
self.repo.save_board(board)
self._log(board.id, None, person.email, "renamed board", "title", old, board.title)
return {"id": board.id, "title": board.title}
def add_list(self, person: Person, board_ref: str, name: str) -> dict:
board = self._board(board_ref, person)
name = board.add_list(person, name)
self.repo.save_board(board)
self._log(board.id, None, person.email, f"added list {name!r}")
return {"board_id": board.id, "lists": board.lists}
def remove_list(self, person: Person, board_ref: str, name: str) -> dict:
board = self._board(board_ref, person)
name = board.list_named(name)
count = sum(1 for c in self.repo.cards_on([board.id]) if c.list_name == name)
board.remove_list(person, name, count)
self.repo.save_board(board)
self._log(board.id, None, person.email, f"removed list {name!r}")
return {"board_id": board.id, "lists": board.lists}
# ------------------------------------------------------------------ people
def members(self, person: Person, board_ref: str) -> list:
return self._members(self._board(board_ref, person))
def add_member(self, person: Person, board_ref: str, email: str,
role: str = "member") -> dict:
board = self._board(board_ref, person)
try:
r = Role((role or "member").strip().lower())
except ValueError:
raise Invalid("role is 'member' or 'viewer'")
before = board.members.get((email or "").strip().lower())
email = board.add_member(person, email, r)
if before is r:
return {"board_id": board.id, "email": email, "role": r.value, "changed": False}
self.repo.save_board(board)
self._log(board.id, None, person.email,
f"added {email} as {r.value}" if before is None
else f"changed {email} from {before.value} to {r.value}",
"member", before.value if before else None, r.value)
return {"board_id": board.id, "email": email, "role": r.value, "changed": True}
def remove_member(self, person: Person, board_ref: str, email: str) -> dict:
board = self._board(board_ref, person)
email = board.remove_member(person, email)
self.repo.save_board(board)
for c in self.repo.cards_on([board.id]): # nobody stays assigned to a board they left
if c.assignee == email:
before = {"assignee": c.assignee}
c.assignee = None
self._save(c, None)
self._log_changes(board, c, person.email, before)
self._log(board.id, None, person.email, f"removed {email}")
return {"board_id": board.id, "removed": email}
# ------------------------------------------------------------------ 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=None) -> dict:
board = self._board(board_ref, person)
board.require_edit(person)
lst = board.list_named(list_name) if list_name else board.lists[0]
who = self._assignee(board, person, assignee)
existing = [c for c in self.repo.cards_on([board.id]) if c.list_name == lst]
top = max((c.position for c in existing), default=0.0) + 1.0
for _ in range(5): # the card number is claimed by a plain INSERT; retry on a race
n = self.repo.next_number(board.id)
card = Card(id=f"{board.id}-{n}", board_id=board.id, number=n,
title=clean_title(title), list_name=lst,
description=clean_description(description), assignee=who,
due_date=parse_due(due_date, self.today()), labels=clean_labels(labels),
position=top, created_by=person.email)
try:
self.repo.insert_card(card)
break
except DuplicateKey:
continue
else:
raise Conflict("too many cards were added at once; try again")
self._log(board.id, card.id, person.email, f"added card #{n} {card.title!r} to {lst}")
return self._view(board, card)
def _assignee(self, board: Board, person: Person, assignee) -> Optional[str]:
if assignee in (None, "", "nobody", "none"):
return None
email = person.email if str(assignee).strip().lower() == "me" else \
str(assignee).strip().lower()
if not board.can_be_assigned(email):
raise Invalid(f"{email} can't be assigned here: assign an owner or member of "
f"the board")
return email
def card_details(self, person: Person, card_ref, board_ref: Optional[str] = None) -> dict:
board, card = self._card(card_ref, person, board_ref)
cards = {c.id: c for c in self.repo.cards_on([board.id])}
blockers = self._blockers(board.id)
view = self._view(board, card, blockers, cards)
view["blocks"] = [{"id": c, "number": cards[c].number, "title": cards[c].title}
for c, bs in blockers.items() if card.id in bs and c in cards]
view["all_blockers"] = [{"id": b, "number": cards[b].number, "title": cards[b].title,
"list_name": cards[b].list_name}
for b in blockers.get(card.id, []) if b in cards]
view["checklist"] = self.repo.checklist(card.id)
view["history"] = self.repo.history(card.id)
view["can_edit"] = board.role_of(person) is not Role.VIEWER
return view
def update_card(self, person: Person, card_ref, board_ref: Optional[str] = None,
title: Optional[str] = None, description: Optional[str] = None,
due_date: Optional[str] = None, labels=None,
expected_version: Optional[int] = None) -> dict:
"""Change any of title / description / due date ('' clears it) / labels."""
board, card = self._card(card_ref, person, board_ref)
board.require_edit(person)
before = {"title": card.title, "description": card.description,
"due_date": card.due_date, "labels": list(card.labels)}
if title is not None:
card.title = clean_title(title)
if description is not None:
card.description = clean_description(description)
if due_date is not None:
card.due_date = parse_due(due_date, self.today())
if labels is not None:
card.labels = clean_labels(labels)
if any(getattr(card, k) != v for k, v in before.items()):
self._save(card, expected_version)
self._log_changes(board, card, person.email, before)
return self._view(board, card)
def assign_card(self, person: Person, card_ref, assignee: Optional[str],
board_ref: Optional[str] = None) -> dict:
"""Assign to an email, 'me', or nobody (empty)."""
def write() -> dict:
board, card = self._card(card_ref, person, board_ref)
board.require_edit(person)
before = {"assignee": card.assignee}
card.assignee = self._assignee(board, person, assignee)
if card.assignee != before["assignee"]:
self._save(card, None)
self._log_changes(board, card, person.email, before)
return self._view(board, card)
return self._retrying(write)
def move_card(self, person: Person, card_ref, list_name: str,
board_ref: Optional[str] = None, before_card: Optional[str] = None) -> dict:
"""Move to a list; with before_card, place it just above that card (drag and drop)."""
return self._retrying(lambda: self._move(person, card_ref, list_name, board_ref,
before_card))
def _move(self, person: Person, card_ref, list_name: str, board_ref: Optional[str],
before_card: Optional[str]) -> dict:
board, card = self._card(card_ref, person, 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, c.number))
if before_card:
idx = next((i for i, c in enumerate(others) if c.id == before_card), None)
if idx is None:
position = (others[-1].position + 1.0) if others else 1.0
else:
prev = others[idx - 1].position if idx > 0 else others[idx].position - 2.0
position = (prev + others[idx].position) / 2.0
else:
position = (others[-1].position + 1.0) if others else 1.0
old_list = card.list_name
card.list_name, card.position = target, position
self._save(card, None)
if old_list != target:
self._log(board.id, card.id, person.email, f"moved from {old_list} to {target}",
field="list", old=old_list, new=target)
return self._view(board, card)
def archive_card(self, person: Person, card_ref, board_ref: Optional[str] = None) -> dict:
def write() -> dict:
board, card = self._card(card_ref, person, board_ref)
board.require_edit(person)
card.archived = True
self._save(card, None)
self._log(board.id, card.id, person.email, f"archived card #{card.number}",
field="archived", old="false", new="true")
return {"id": card.id, "number": card.number, "title": card.title,
"archived": True}
return self._retrying(write)
def card_history(self, person: Person, card_ref, board_ref: Optional[str] = None) -> list:
_, card = self._card(card_ref, person, board_ref)
return self.repo.history(card.id)
def board_activity(self, person: Person, board_ref: str, limit: int = 50) -> list:
board = self._board(board_ref, person)
return self.repo.activity(board.id, max(1, min(int(limit), 200)))
# ------------------------------------------------------------------ "what's on my plate"
def _visible_cards(self, person: Person) -> list:
boards = {b.id: b for b in self.repo.boards_for(person.email)}
if not boards:
return []
cards = self.repo.cards_on(list(boards))
out = []
for bid, board in boards.items():
mine = [c for c in cards if c.board_id == bid]
by_id, blockers = {c.id: c for c in mine}, self._blockers(bid)
out += [(board, c, self._view(board, c, blockers, by_id)) for c in mine]
return out
def my_cards(self, person: Person, days: Optional[int] = None) -> list:
"""Open cards assigned to you; with days, only those due within that many days
(overdue ones included). 'This week' = 7."""
today = self.today()
out = []
for board, card, view in self._visible_cards(person):
if card.assignee != person.email or card.list_name == board.done_list:
continue
if days is not None and not (card.due_date and
card.due_date <= today + timedelta(days=int(days))):
continue
out.append(view)
return sorted(out, key=lambda v: (v["due_date"] or "9999", v["board"], v["number"]))
def overdue(self, person: Person, board_ref: Optional[str] = None) -> list:
"""Cards past their due date and not done, on every board you can see (or one)."""
board_id = self._board(board_ref, person).id if board_ref else None
out = [v for b, _, v in self._visible_cards(person)
if v["overdue"] and (board_id is None or b.id == board_id)]
return sorted(out, key=lambda v: (v["due_date"], v["board"], v["number"]))
def search_cards(self, person: Person, text: str, board_ref: Optional[str] = None) -> list:
"""Plain text search over titles, descriptions and labels."""
words = [w for w in (text or "").lower().split() if w]
if not words:
raise Invalid("search for some words")
board_id = self._board(board_ref, person).id if board_ref else None
out = []
for b, c, v in self._visible_cards(person):
hay = f"{c.title} {c.description} {' '.join(c.labels)}".lower()
if (board_id is None or b.id == board_id) and all(w in hay for w in words):
out.append(v)
return out[:50]
# ------------------------------------------------------------------ dependencies
def add_blocker(self, person: Person, card_ref, blocked_by_ref,
board_ref: Optional[str] = None) -> dict:
"""Record that a card is blocked by another card on the same board."""
board, card = self._card(card_ref, person, board_ref)
board.require_edit(person)
_, blocker = self._card(blocked_by_ref, person, board.id)
if blocker.board_id != board.id:
raise Invalid("a card can only be blocked by a card on the same board")
links = self._blockers(board.id)
if blocker.id in links.get(card.id, []):
return self.card_details(person, card.id)
check_no_cycle(card.id, blocker.id, lambda cid: links.get(cid, []))
self.repo.add_link(board.id, card.id, blocker.id, person.email)
self._log(board.id, card.id, person.email, f"blocked by #{blocker.number}",
field="blocked_by", old=None, new=f"#{blocker.number}")
return self.card_details(person, card.id)
def remove_blocker(self, person: Person, card_ref, blocked_by_ref,
board_ref: Optional[str] = None) -> dict:
board, card = self._card(card_ref, person, board_ref)
board.require_edit(person)
_, blocker = self._card(blocked_by_ref, person, board.id)
if blocker.id not in self._blockers(board.id).get(card.id, []):
raise Invalid(f"#{card.number} is not blocked by #{blocker.number}")
self.repo.remove_link(card.id, blocker.id)
self._log(board.id, card.id, person.email, f"no longer blocked by #{blocker.number}",
field="blocked_by", old=f"#{blocker.number}", new=None)
return self.card_details(person, card.id)
def what_blocks(self, person: Person, card_ref, board_ref: Optional[str] = None) -> dict:
"""Everything still open that stands between this card and done, following the
'blocked by' links all the way down, plus the cards nothing blocks (where to start)."""
board, card = self._card(card_ref, person, board_ref)
cards = {c.id: c for c in self.repo.cards_on([board.id])}
links = self._blockers(board.id)
chain, seen, stack = [], set(), list(links.get(card.id, []))
while stack:
cid = stack.pop()
if cid in seen or cid not in cards:
continue
seen.add(cid)
c = cards[cid]
if c.list_name != board.done_list:
chain.append(c)
stack.extend(links.get(cid, []))
def open_blockers(c):
return [b for b in links.get(c.id, [])
if b in cards and cards[b].list_name != board.done_list]
view = lambda c: {"number": c.number, "id": c.id, "title": c.title,
"list_name": c.list_name, "assignee": c.assignee,
"due_date": _iso(c.due_date),
"overdue": c.is_overdue(self.today(), board.done_list)}
return {"card": view(card), "blocked": bool(chain),
"open_blockers": [view(c) for c in sorted(chain, key=lambda c: c.number)],
"start_here": [view(c) for c in chain if not open_blockers(c)]}
def blocked_cards(self, person: Person, board_ref: Optional[str] = None) -> list:
"""Open cards that are waiting on another open card, with what they wait on: the
quick answer to "what is blocked?" on every board you can see (or one)."""
board_id = self._board(board_ref, person).id if board_ref else None
out = [v for b, c, v in self._visible_cards(person)
if v["blocked_by"] and c.list_name != b.done_list
and (board_id is None or b.id == board_id)]
return sorted(out, key=lambda v: (v["board"], v["number"]))
# ------------------------------------------------------------------ AI (Dodil Models)
def _ai(self) -> Assistant:
if self.assistant is None:
raise Unavailable("no AI assistant configured")
return self.assistant
def _ask(self, method: str, *args):
"""One call to Dodil Models; a timeout or an HTTP error becomes Unavailable."""
try:
return getattr(self._ai(), method)(*args)
except Unavailable:
raise
except Exception as e: # httpx errors, a malformed body, a model that said nothing
raise Unavailable(f"Dodil Models didn't answer ({type(e).__name__}); "
f"try again in a moment")
def suggest_checklist(self, person: Person, card_ref,
board_ref: Optional[str] = None) -> dict:
"""An AI suggestion only: it changes nothing. A person decides what to keep."""
_, card = self._card(card_ref, person, board_ref)
steps = self._ask("checklist", card.title, card.description)
return {"card_id": card.id, "number": card.number, "suggested_checklist": steps}
def make_checklist(self, person: Person, card_ref, items=None,
board_ref: Optional[str] = None) -> dict:
"""Turn a card into a checklist: the items you give, or the AI's suggestion."""
board, card = self._card(card_ref, person, board_ref)
board.require_edit(person)
if isinstance(items, str): # "a\nb" or "a; b" from an assistant
items = [x for x in items.replace(";", "\n").splitlines()]
steps = [str(s).strip()[:300] for s in (items or []) if str(s).strip()] or \
self._ask("checklist", card.title, card.description)
if not steps:
raise Unavailable("Dodil Models suggested no steps; try again or give the items")
existing = self.repo.checklist(card.id)
start = max((i["position"] for i in existing), default=0)
for i, text in enumerate(steps[:20], start=1):
self.repo.save_check_item(board.id, {"id": uuid.uuid4().hex[:12], "card_id": card.id,
"text": text[:300], "done": False,
"position": start + i})
self._log(board.id, card.id, person.email, f"added a checklist of {len(steps[:20])}",
field="checklist", old=None, new="; ".join(steps[:20])[:500])
return {"card_id": card.id, "number": card.number,
"checklist": self.repo.checklist(card.id)}
def tick_item(self, person: Person, card_ref, item_id: str, done: bool = True,
board_ref: Optional[str] = None) -> dict:
board, card = self._card(card_ref, person, board_ref)
board.require_edit(person)
item = next((i for i in self.repo.checklist(card.id) if i["id"] == item_id), None)
if item is None:
raise NotFound("no such checklist item")
if item["done"] != bool(done):
item["done"] = bool(done)
self.repo.save_check_item(board.id, item)
self._log(board.id, card.id, person.email,
f"{'ticked' if done else 'unticked'} {item['text']!r}",
field="checklist", old=None, new=item["text"])
return {"card_id": card.id, "checklist": self.repo.checklist(card.id)}
def find_similar(self, person: Person, card_ref=None, text: Optional[str] = None,
board_ref: Optional[str] = None, limit: int = 5) -> list:
"""Search by meaning: cards like a given card, or like a sentence. Only boards you
can see are searched. Embeddings come from Dodil Models and are cached per card."""
source = None
if card_ref not in (None, ""):
_, source = self._card(card_ref, person, board_ref)
text = f"{source.title}\n{source.description}"
if not (text or "").strip():
raise Invalid("give a card or some text to compare with")
rows = [(b, c, v) for b, c, v in self._visible_cards(person)
if not source or c.id != source.id]
if not rows:
return []
board_ids = sorted({b.id for b, _, _ in rows})
cache = self.repo.vectors_on(board_ids)
texts = {c.id: f"{c.title}\n{c.description}" for _, c, _ in rows}
hashes = {cid: hashlib.sha1(t.encode()).hexdigest()[:16] for cid, t in texts.items()}
missing = [cid for cid in texts if cache.get(cid, (None,))[0] != hashes[cid]]
if missing:
vectors = self._ask("embed", [texts[cid] for cid in missing])
for cid, vec in zip(missing, vectors):
cache[cid] = (hashes[cid], vec)
self.repo.save_vector(cid.rsplit("-", 1)[0], cid, hashes[cid], vec)
query = self._ask("embed", [text])[0]
scored = []
for _, c, v in rows:
vec = cache.get(c.id, (None, None))[1]
if vec:
scored.append((_cosine(query, vec), v))
scored.sort(key=lambda s: -s[0])
return [{**v, "similarity": round(s, 3)} for s, v in scored[:max(1, min(limit, 20))]]
def daily_summary(self, person: Person, board_ref: str, refresh: bool = False) -> dict:
"""A short AI summary of a board: what moved in the last day, what's overdue, what's
blocked, who has what. Written once a day and kept (refresh=True rewrites it). A
reasoning model can take a while: after ai_wait seconds this answers 'writing' and the
summary is saved when it lands, so asking again a minute later returns it."""
board = self._board(board_ref, person)
day = self.today().isoformat()
key = f"{board.id}:{day}"
cached = self.repo.get_summary(board.id, day)
if cached and not refresh:
return {"board": board.title, "day": day, "status": "ready", **cached}
fut = self._writing.get(key)
if fut is None or fut.done():
facts = self._summary_facts(person, board)
self._ai()
def write() -> str:
text = self._ask("summary", board.title, facts)
self.repo.save_summary(board.id, day, text)
return text
fut = self._ai_pool.submit(write)
self._writing[key] = fut
try:
text = fut.result(timeout=self.ai_wait)
except Late:
return {"board": board.title, "day": day, "status": "writing",
"summary": (cached or {}).get("summary"),
"hint": "Dodil Models is still writing it; ask again in a minute"}
finally:
if fut.done():
self._writing.pop(key, None)
return {"board": board.title, "day": day, "status": "ready", "summary": text}
def _summary_facts(self, person: Person, board: Board) -> str:
data = self.show_board(person, board.id)
lines = []
for lst in data["lists"]:
lines.append(f"List {lst['name']}: " + ("; ".join(
f"#{c['number']} {c['title']} (assignee {c['assignee'] or 'none'}, due "
f"{c['due_date'] or 'none'}{', OVERDUE' if c['overdue'] else ''}"
f"{', blocked' if c['blocked_by'] else ''})" for c in lst["cards"]) or "empty"))
recent = self.repo.activity(board.id, 25)
lines.append("Recent changes: " + "; ".join(
f"{a['at'][:16]} {a['actor']} {a['action']}" for a in recent))
lines.append(f"Today is {self.today().isoformat()}.")
return "\n".join(lines)
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