north-scape/relayboard
Relayboard: accessible team boards with audited actions, dependency maps, and shared web/MCP use cases.
A remix template · 0 remixes · Remix north-scape/relayboard
app/boards/application/use_cases.py
"""Web and MCP are two doors onto these identity-scoped use cases.
Every mutation + audit event is committed together. The repository serializes board writes.
"""
import math
import uuid
from dataclasses import asdict
from datetime import date, datetime, timedelta, timezone
from functools import wraps
from boards.domain.model import Board, Card, Conflict, Forbidden, Invalid, LISTS, Person, Role, email_address, text
class NotFound(Exception): pass
def mutation(fn):
@wraps(fn)
def wrapped(self, *args, **kwargs):
with self.repo.atomic():
return fn(self, *args, **kwargs)
return wrapped
class Boards:
def __init__(self, repo, assistant=None, today=None):
self.repo, self.assistant = repo, assistant
self.today = today or (lambda: datetime.now(timezone.utc).date())
def _card(self, c):
d = asdict(c)
d['reference'] = c.reference
d['due_date'] = c.due_date.isoformat() if c.due_date else None
d['is_overdue'] = c.overdue(self.today())
return d
def _board(self, board_id, person):
board = self.repo.get(board_id)
if board is None or board.role_of(person) is None:
raise NotFound('no such board')
return board
def _find_card(self, card_id, person):
card = self.repo.get_card(card_id)
if card is None:
raise NotFound('no such card')
try:
board = self._board(card.board_id, person)
except NotFound:
raise NotFound('no such card') from None
# Re-read after the aggregate lock: another writer may have changed the card
# between the initial lookup and acquiring the board lock.
card = self.repo.get_card(card_id)
if card is None or card.board_id != board.id: raise NotFound('no such card')
return board, card
def _fields(self, board, changes, card=None):
allowed = {'title', 'description', 'assignee', 'due_date', 'labels', 'checklist'}
if not isinstance(changes, dict) or set(changes) - allowed:
raise Invalid('unknown card field')
out = {}
for key, value in changes.items():
if key == 'title': out[key] = text(value, 'title', 180, True)
elif key == 'description': out[key] = text(value, 'description', 6000)
elif key == 'assignee':
value = email_address(value) if value else None
if value and board.role_of(Person(value)) not in (Role.OWNER, Role.MEMBER):
raise Invalid('assignee must be an owner or member of this board')
out[key] = value
elif key == 'due_date':
try: out[key] = date.fromisoformat(value) if value else None
except (TypeError, ValueError): raise Invalid('due date must be YYYY-MM-DD') from None
elif key == 'checklist': out[key] = self._checklist(value, card)
elif key == 'labels':
if not isinstance(value, list) or len(value) > 8:
raise Invalid('use at most 8 labels')
out[key] = list(dict.fromkeys(text(v, 'label', 30, True) for v in value))
return out
def _checklist(self, items, card):
if not isinstance(items, list) or len(items) > 50:
raise Invalid('checklist must contain at most 50 items')
existing = {item['id'] for item in card.checklist} if card else set()
seen, result = set(), []
for item in items:
if not isinstance(item, dict) or set(item) - {'id', 'text', 'done'}:
raise Invalid('checklist items accept only id, text and done')
label = text(item.get('text'), 'checklist step', 300, True)
done = item.get('done', False)
if type(done) is not bool: raise Invalid('checklist done must be true or false')
if 'id' in item:
item_id = item['id']
if not isinstance(item_id, str) or item_id not in existing or item_id in seen:
raise Invalid('checklist ids must be unique existing items on this card')
else:
item_id = uuid.uuid4().hex
seen.add(item_id)
result.append({'id': item_id, 'text': label, 'done': done})
return result
def _record(self, person, card, before, action):
after = self._card(card)
changes = {k: {'before': before.get(k), 'after': v} for k, v in after.items()
if before.get(k) != v and k not in ('revision','is_overdue','reference')}
self.repo.save_card(card)
self.repo.log(card.board_id, card.id, person.email, action, changes)
return after
def _revision(self, card, expected_revision):
# Door validation is not the trust boundary: every application caller must
# supply the revision it read. Check after visibility/edit permission and
# the locked re-read, but before validation, no-op handling or any write.
if type(expected_revision) is not int:
raise Invalid('expected_revision must be an integer; read the card first')
if expected_revision != card.revision:
raise Conflict('card changed since you opened it; reload and try again')
@mutation
def create_board(self, person, title):
board = Board(uuid.uuid4().hex, text(title, 'board title', 100, True), person.email)
self.repo.save_board(board)
self.repo.log(board.id, None, person.email, 'created board')
return {'id': board.id, 'title': board.title}
def my_boards(self, person):
return [{'id': b.id, 'title': b.title, 'role': b.role_of(person).value}
for b in self.repo.boards_for(person.email) if b.role_of(person)]
def _resolution(self, matches):
"""Never pick an arbitrary candidate, including when the list is truncated."""
count = len(matches)
return {'status': 'resolved' if count == 1 else 'ambiguous' if count else 'not_found',
'matches': matches[:20], 'total_matches': count, 'truncated': count > 20}
def resolve_board(self, person, reference):
reference = text(reference, 'board reference', 100, True)
visible = self.my_boards(person)
exact = [b for b in visible if b['id'] == reference]
matches = exact or [b for b in visible if b['title'].casefold() == reference.casefold()]
return self._resolution(sorted(matches, key=lambda b: (b['title'].casefold(), b['id'])))
def resolve_card(self, person, reference, board_id=None):
"""Read-only discovery. Writes continue to require a canonical card id."""
reference = text(reference, 'card reference', 180, True)
boards = [self._board(board_id, person)] if board_id is not None else [
b for b in self.repo.boards_for(person.email) if b.role_of(person)]
visible = [(b, c) for b in boards for c in self.repo.cards_on(b.id)]
# Canonical ids take priority; shortened aliases and titles never silently
# disambiguate one another. A title equal to an alias is still ambiguous.
exact = [(b, c) for b, c in visible if c.id == reference]
matches = exact or [(b, c) for b, c in visible
if c.reference.casefold() == reference.casefold()
or c.title.casefold() == reference.casefold()]
matches.sort(key=lambda bc: (bc[0].title.casefold(), bc[0].id, bc[1].title.casefold(), bc[1].id))
return self._resolution([{'id': c.id, 'reference': c.reference, 'title': c.title,
'board_id': b.id, 'board_title': b.title,
'list_name': c.list_name, 'revision': c.revision}
for b, c in matches])
def show_board(self, person, board_id):
board = self._board(board_id, person)
return {'id': board.id, 'title': board.title, 'role': board.role_of(person).value,
'lists': list(LISTS), 'members': [{'email': board.owner_email, 'role': 'owner'}] +
[{'email': e, 'role': r.value} for e, r in sorted(board.members.items())],
'cards': [self._card(c) for c in self.repo.cards_on(board_id)]}
@mutation
def add_member(self, person, board_id, email, role='member'):
board = self._board(board_id, person)
board.add_member(person, email, role)
email = email_address(email)
# An owner changing a member to viewer also removes active assignments.
if role == 'viewer':
for card in self.repo.cards_on(board_id):
if card.assignee == email:
before = self._card(card)
card.assignee, card.revision = None, card.revision + 1
self._record(person, card, before, 'unassigned after role changed to viewer')
self.repo.save_board(board)
self.repo.log(board.id, None, person.email, f'set {email} role to {role}')
return {'board_id': board_id, 'email': email, 'role': role}
@mutation
def remove_member(self, person, board_id, email):
board = self._board(board_id, person)
previous_role = board.remove_member(person, email)
email = email_address(email)
result = {'board_id': board_id, 'email': email, 'removed': previous_role is not None,
'unassigned_cards': 0}
if previous_role is None:
return result
self.repo.delete_membership(board.id, email)
for card in self.repo.cards_on(board.id):
if card.assignee == email:
before = self._card(card)
card.assignee, card.revision = None, card.revision + 1
self._record(person, card, before, 'unassigned after member removal')
result['unassigned_cards'] += 1
self.repo.log(board.id, None, person.email, f'removed {email} from board',
{'membership': {'before': {'email': email, 'role': previous_role.value}, 'after': None}})
return result
@mutation
def add_card(self, person, board_id, title, list_name='To do', description='', assignee=None, due_date=None, labels=None):
board = self._board(board_id, person)
board.require_edit(person)
fields = self._fields(board, dict(title=title,description=description,assignee=assignee,due_date=due_date,labels=labels or []))
card = Card(uuid.uuid4().hex, board_id, **fields)
card.move_to(list_name)
return self._record(person, card, {}, 'added card')
@mutation
def update_card(self, person, card_id, changes, expected_revision=None):
board, card = self._find_card(card_id, person)
board.require_edit(person)
self._revision(card, expected_revision)
fields = self._fields(board, changes, card)
before = self._card(card)
if all(getattr(card,k) == v for k,v in fields.items()): return before
for k,v in fields.items(): setattr(card,k,v)
card.revision += 1
return self._record(person, card, before, 'updated card')
def update_checklist(self, person, card_id, items, expected_revision):
return self.update_card(person, card_id, {'checklist': items}, expected_revision)
def assign_card(self, person, card_id, email, expected_revision=None):
return self.update_card(person, card_id, {'assignee':email}, expected_revision)
@mutation
def move_card(self, person, card_id, list_name, expected_revision=None):
board, card = self._find_card(card_id, person)
board.require_edit(person)
self._revision(card, expected_revision)
before = self._card(card)
card.move_to(list_name)
if before['list_name'] == list_name: return before
card.revision += 1
return self._record(person, card, before, f"moved from {before['list_name']} to {list_name}")
def activity(self, person, board_id):
self._board(board_id, person)
return self.repo.activity(board_id)
def _visible_cards(self, person):
return [c for b in self.repo.boards_for(person.email) if b.role_of(person)
for c in self.repo.cards_on(b.id)]
def my_cards(self, person, days=7):
if type(days) is not int or not 1 <= days <= 90: raise Invalid('days must be 1 to 90')
end = self.today() + timedelta(days=days)
cards = [c for c in self._visible_cards(person) if c.assignee == person.email
and c.list_name != 'Done' and (c.due_date is None or c.due_date < end)]
return [self._card(c) for c in sorted(cards,key=lambda c:(c.due_date or date.max,c.id))]
def overdue(self, person):
return [self._card(c) for c in sorted(self._visible_cards(person),key=lambda c:(c.due_date or date.max,c.id))
if c.overdue(self.today())]
@mutation
def set_blockers(self, person, card_id, blocked_by, expected_revision=None):
board, card = self._find_card(card_id, person)
board.require_edit(person)
self._revision(card, expected_revision)
if not isinstance(blocked_by,list) or len(blocked_by)>50 or any(not isinstance(x,str) for x in blocked_by):
raise Invalid('blocked_by must contain at most 50 card ids')
proposed = list(dict.fromkeys(blocked_by))
cards = {c.id:c for c in self.repo.cards_on(board.id)}
if any(i not in cards for i in proposed): raise Invalid('blockers must belong to the same board')
def reaches(start, target):
todo, visited = [start], set()
while todo:
current=todo.pop()
if current == target: return True
if current in visited: continue
visited.add(current)
todo.extend(cards[current].blocked_by if current in cards else [])
return False
if any(reaches(i,card_id) for i in proposed): raise Invalid('dependency would create a cycle')
before=self._card(card)
if proposed == card.blocked_by: return before
card.blocked_by, card.revision = proposed, card.revision+1
return self._record(person,card,before,'updated blockers')
def blockers(self, person, card_id):
board,card=self._find_card(card_id,person)
cards={c.id:c for c in self.repo.cards_on(board.id)}
seen,todo,out=set(),list(card.blocked_by),[]
while todo:
i=todo.pop()
if i in seen or i not in cards: continue
seen.add(i)
blocker=cards[i]
if blocker.list_name == 'Done': continue
out.append(self._card(blocker));todo.extend(blocker.blocked_by)
return {'card_id':card_id,'blockers':out,'is_blocked':bool(out)}
def _ai(self):
if self.assistant is None: raise Invalid('Dodil Models is not configured')
return self.assistant
def suggest_checklist(self, person, card_id):
_,card=self._find_card(card_id,person)
steps = self._ai().checklist(card.title)
if not isinstance(steps, list) or not 1 <= len(steps) <= 8:
raise Invalid('model returned an invalid checklist; try again')
steps = [text(step, 'suggested checklist step', 300, True) for step in steps]
return {'card_id':card.id,'suggested_checklist':steps,'source_revision':card.revision,'suggestion_only':True}
def board_summary(self, person, board_id):
board=self.show_board(person,board_id)
if len(board['cards'])>100: raise Invalid('summary supports at most 100 cards per board')
# Include identifiers so the model can resolve dependency links within
# this already-authorized board, rather than guessing what UUIDs mean.
payload = {'title': board['title'], 'cards': [
{**{k: c[k] for k in ('id', 'reference', 'title', 'list_name', 'assignee',
'due_date', 'is_overdue', 'blocked_by')},
'checklist_progress': {'done': sum(item['done'] for item in c['checklist']),
'total': len(c['checklist'])}}
for c in board['cards']]}
return {'board_id':board_id,'summary':self._ai().summary(payload),'suggestion_only':True}
def similar_cards(self, person, card_id, limit=5):
board,card=self._find_card(card_id,person)
if type(limit) is not int or not 1<=limit<=10: raise Invalid('limit must be 1 to 10')
candidates=[c for c in self.repo.cards_on(board.id) if c.id!=card.id]
if len(candidates)>99: raise Invalid('semantic search supports at most 100 cards per board')
if not candidates: return {'card_id':card_id,'matches':[]}
vectors=self._ai().embeddings([c.title+'\n'+c.description for c in [card]+candidates])
if len(vectors)!=len(candidates)+1: raise Invalid('model returned an incomplete embedding batch')
def cosine(v):
a=vectors[0]
if len(a)!=len(v) or not a or any(not math.isfinite(x) for x in a+v): raise Invalid('invalid embedding')
norm=math.sqrt(sum(x*x for x in a)*sum(x*x for x in v))
return sum(x*y for x,y in zip(a,v))/norm if norm else 0
scores=sorted(zip(candidates,vectors[1:]),key=lambda cv:cosine(cv[1]),reverse=True)[:limit]
return {'card_id':card_id,'matches':[{'card':self._card(c),'similarity':round(cosine(v),4)} for c,v in scores]}