theaio/theaio-boards
Laneway, team task boards you can run from Claude or ChatGPT. FastAPI + DataK3 + appid + Models + MCP.
A remix template · 0 remixes · Remix theaio/theaio-boards
tests/fakes.py
"""In-memory stand-ins for DataK3 and Dodil Models. They store copies, like a database does, so a
use case that forgets to save is caught by the tests instead of hidden by a shared object."""
import hashlib
import math
from copy import deepcopy
from datetime import datetime, timezone
class MemoryBoards:
def __init__(self):
self.boards, self.cards, self.events = {}, {}, []
self.entries, self.links, self.vectors = [], {}, {}
self.stale_max_once = False # simulate a concurrent writer taking our card number
# boards
def get(self, board_id):
return deepcopy(self.boards.get(board_id))
def boards_for(self, email):
return [deepcopy(b) for b in self.boards.values()
if b.owner_email == email or email in b.members]
def save_board(self, board):
self.boards[board.id] = deepcopy(board)
def remove_member(self, board_id, email):
self.boards[board_id].members.pop(email, None)
def log(self, board_id, card_id, actor, action):
self.events.append((board_id, card_id, actor, action))
# cards
def get_card(self, card_id):
return deepcopy(self.cards.get(card_id))
def card_by_number(self, board_id, number):
return next((deepcopy(c) for c in self.cards.values()
if c.board_id == board_id and c.number == number), None)
def cards_on(self, board_id):
return [deepcopy(c) for c in self.cards.values() if c.board_id == board_id]
def max_card_number(self, board_id):
top = max((c.number for c in self.cards.values() if c.board_id == board_id), default=0)
if self.stale_max_once:
self.stale_max_once = False
return top - 1 if top else top
return top
def insert_card(self, card):
if card.id in self.cards: # the primary key refuses it: 23505
return False
self.cards[card.id] = deepcopy(card)
return True
def save_card(self, card):
self.cards[card.id] = deepcopy(card)
# history
def record(self, entries):
self.entries.extend(entries)
def history(self, card_id):
return [e for e in self.entries if e.card_id == card_id]
def board_history(self, board_id, since):
return [e for e in self.entries if e.board_id == board_id and e.at >= since]
# links
def links_on(self, board_id):
return [(c, b) for (c, b), bid in self.links.items() if bid == board_id]
def add_link(self, board_id, card_id, blocked_by):
self.links[(card_id, blocked_by)] = board_id
def remove_link(self, card_id, blocked_by):
self.links.pop((card_id, blocked_by), None)
# embeddings: the SQL version orders by pgvector's <=> (cosine distance); so does this
def embedding_digests(self, board_ids):
return {cid: d for cid, (bid, d, _) in self.vectors.items() if bid in board_ids}
def save_embedding(self, card, digest, vector):
self.vectors[card.id] = (card.board_id, digest, list(vector))
def vector_of(self, card_id):
v = self.vectors.get(card_id)
return v[2] if v else None
def nearest(self, board_ids, vector, exclude_id, limit):
def dist(v):
dot = sum(a * b for a, b in zip(v, vector))
norm = math.sqrt(sum(a * a for a in v)) * math.sqrt(sum(b * b for b in vector))
return 1 - dot / norm if norm else 1.0
hits = [(cid, dist(v)) for cid, (bid, _, v) in self.vectors.items()
if bid in board_ids and cid != exclude_id]
return sorted(hits, key=lambda h: h[1])[:limit]
class FakeAI:
"""Embeddings are word counts hashed into 32 buckets: same words -> close vectors."""
def __init__(self, fail=False):
self.fail, self.embedded = fail, 0
def checklist(self, title):
return ["step one", "step two"]
def embed(self, texts):
self.embedded += len(texts)
out = []
for t in texts:
v = [0.0] * 32
for w in t.lower().split():
v[int(hashlib.md5(w.encode()).hexdigest(), 16) % 32] += 1
out.append(v)
return out
def summarize(self, facts):
if self.fail:
raise RuntimeError("model down")
return "SUMMARY: " + facts.splitlines()[0]
class Clock:
def __init__(self, at=datetime(2026, 11, 2, 9, 0, tzinfo=timezone.utc)):
self.at = at
def __call__(self):
return self.at