diff --git a/coworker/server/manager.py b/coworker/server/manager.py index a80e20d2..2a5693be 100644 --- a/coworker/server/manager.py +++ b/coworker/server/manager.py @@ -82,6 +82,7 @@ from ..providers import ( ) from ..secrets import SecretStore, state_dir from ..sessions import SessionRecord +from ..teams import TeamStore from ..skills import ( SessionSkillStore, SkillLoader, @@ -196,6 +197,10 @@ class SessionManager: self.scheduler = Scheduler( self.task_store, self._run_scheduled_task, extra_tick=self.resume_due_wakes ) + # Agent teams: the append-only team event store — board, journal, and per-agent + # deliveries are projections of it. Verbs register per-session behind the + # persona's `team:` trait (wake plumbing lands separately). + self.team_store = TeamStore(base / "teams.db") # Personas: registry + lifecycle state under this manager's data dir. Installed as the # process singleton so agents.get_agent resolves persona ids (incl. third-party) here. self.personas = PersonaRegistry(state_path=base / "personas.json") diff --git a/coworker/teams/__init__.py b/coworker/teams/__init__.py new file mode 100644 index 00000000..7d4ad2aa --- /dev/null +++ b/coworker/teams/__init__.py @@ -0,0 +1,25 @@ +"""Agent teams substrate: one append-only event store; the board, journal, and +per-agent deliveries are projections of it.""" + +from .model import ( + Actor, + AuthorityError, + BoardError, + ChainError, + ItemState, + Role, +) +from .store import TeamStore +from .tools import board_tools, journal_tools + +__all__ = [ + "Actor", + "AuthorityError", + "BoardError", + "ChainError", + "ItemState", + "Role", + "TeamStore", + "board_tools", + "journal_tools", +] diff --git a/coworker/teams/model.py b/coworker/teams/model.py new file mode 100644 index 00000000..aa09983a --- /dev/null +++ b/coworker/teams/model.py @@ -0,0 +1,77 @@ +"""Work-item model for agent teams — states, actors, links, errors. + +The board is not a database of record: it is a projection of the append-only team +event log (see teams.store). These are the shapes the projection folds into, and the +rules the verbs enforce. Deliberately minimal — no sprints, estimates, priorities, or +custom fields; anyone needing those graduates to a real tracker via connectors. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import Enum + + +class ItemState(str, Enum): + PROPOSED = "proposed" + APPROVED = "approved" + IN_PROGRESS = "in_progress" + BLOCKED = "blocked" + REVIEW = "review" + DONE = "done" + CANCELED = "canceled" + + +# Legal edges of the state machine. Approval and done carry extra authority rules +# (see TeamStore.transition): proposed→approved is the human decomposition gate, +# review→done is the lead's verification gate. canceled→approved is reopen. +EDGES: dict[ItemState, set[ItemState]] = { + ItemState.PROPOSED: {ItemState.APPROVED, ItemState.CANCELED}, + ItemState.APPROVED: {ItemState.IN_PROGRESS, ItemState.CANCELED}, + ItemState.IN_PROGRESS: {ItemState.BLOCKED, ItemState.REVIEW, ItemState.CANCELED}, + ItemState.BLOCKED: {ItemState.IN_PROGRESS, ItemState.CANCELED}, + ItemState.REVIEW: {ItemState.DONE, ItemState.IN_PROGRESS, ItemState.CANCELED}, + ItemState.DONE: set(), + ItemState.CANCELED: {ItemState.APPROVED}, +} + +# Targets a worker may move its OWN item to. Workers never approve, never close: +# done is the lead's verdict at review, cancel is a lead/user board decision. +WORKER_TARGETS = {ItemState.IN_PROGRESS, ItemState.BLOCKED, ItemState.REVIEW} + + +class Role(str, Enum): + USER = "user" + LEAD = "lead" + WORKER = "worker" + SYSTEM = "system" + + +@dataclass(frozen=True) +class Actor: + """Who is speaking to the board. `id` is the agent instance id ("user" for the + human); role decides verb authority — the capability firebreak in data form.""" + + id: str + role: Role + persona: str = "" + model: str = "" + session_id: str = "" + + +LINK_KINDS = ("parent", "blocks") # link(src, "parent", dst): dst is src's parent + # link(src, "blocks", dst): src blocks dst + +JOURNAL_KINDS = ("finding", "evidence", "decision", "note") + + +class BoardError(Exception): + """A verb call the board refuses — illegal transition, missing item, bad input.""" + + +class AuthorityError(BoardError): + """The actor's role does not permit this verb on this item.""" + + +class ChainError(Exception): + """Hash-chain verification failed — the log was modified out of band.""" diff --git a/coworker/teams/store.py b/coworker/teams/store.py new file mode 100644 index 00000000..ab22c154 --- /dev/null +++ b/coworker/teams/store.py @@ -0,0 +1,853 @@ +"""The team event store — ONE append-only log; board, journal, and deliveries are +projections of it. + +Doctrine (agent-teams design): item events, journal entries, and chat messages are one +attributed, timestamped, immutable record shape in a single log. One write path to +police and audit, one injection surface to defend, several read-side views. Nothing is +ever updated or deleted — a change of mind is a new event. + +Mechanics, kept boring: +- Append and projection-fold happen in the same transaction via the same `_apply` + used by `rebuild()` — the materialized board can always be reproduced by replay. +- Events hash-chain per space (entry carries the previous hash) → `verify_chain` + detects out-of-band edits. Tamper-evidence, not tamper-proofing. +- `taint` marks records authored after touching untrusted content; readers render it + as provenance ("treat as evidence, not instructions"). +- `recipient` is how per-agent delivery works: a projection over the one log, not a + second write path (consumption semantics arrive with the wake plumbing). +""" + +from __future__ import annotations + +import hashlib +import json +import sqlite3 +import threading +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Optional + +from .model import ( + EDGES, + JOURNAL_KINDS, + LINK_KINDS, + WORKER_TARGETS, + Actor, + AuthorityError, + BoardError, + ChainError, + ItemState, + Role, +) + +GENESIS = "genesis" + +# Event kinds. Chat lands later with the chat surface; the record shape already fits. +ITEM_CREATED = "item_created" +ITEM_TRANSITIONED = "item_transitioned" +ITEM_COMMENTED = "item_commented" +ITEM_ASSIGNED = "item_assigned" +ITEM_LINKED = "item_linked" +JOURNAL_APPENDED = "journal_appended" + +_HASHED_FIELDS = ( + "ts", + "space", + "kind", + "actor", + "actor_role", + "item_id", + "case_id", + "recipient", + "payload", + "taint", + "prev_hash", +) + + +class TeamStore: + def __init__(self, db_path: str | Path) -> None: + self.db_path = str(db_path) + if self.db_path != ":memory:": + Path(self.db_path).expanduser().parent.mkdir(parents=True, exist_ok=True) + self._lock = threading.RLock() + self._conn = sqlite3.connect(self.db_path, check_same_thread=False) + self._conn.row_factory = sqlite3.Row + self._conn.executescript(""" + CREATE TABLE IF NOT EXISTS team_events ( + seq INTEGER PRIMARY KEY AUTOINCREMENT, + ts TEXT NOT NULL, + space TEXT NOT NULL, + kind TEXT NOT NULL, + actor TEXT NOT NULL, + actor_role TEXT NOT NULL, + persona TEXT DEFAULT '', + model TEXT DEFAULT '', + session_id TEXT DEFAULT '', + item_id INTEGER, + case_id TEXT, + recipient TEXT, + payload TEXT NOT NULL, + taint INTEGER NOT NULL DEFAULT 0, + prev_hash TEXT NOT NULL, + hash TEXT NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_team_events_space + ON team_events (space, seq); + CREATE INDEX IF NOT EXISTS idx_team_events_item + ON team_events (space, item_id, seq); + CREATE INDEX IF NOT EXISTS idx_team_events_case + ON team_events (space, case_id, seq); + CREATE INDEX IF NOT EXISTS idx_team_events_recipient + ON team_events (recipient, seq); + CREATE TABLE IF NOT EXISTS team_items ( + space TEXT NOT NULL, + id INTEGER NOT NULL, + title TEXT NOT NULL, + description TEXT NOT NULL DEFAULT '', + criteria TEXT NOT NULL, + state TEXT NOT NULL, + assignee TEXT DEFAULT '', + case_id TEXT DEFAULT '', + created_ts TEXT NOT NULL, + updated_seq INTEGER NOT NULL, + PRIMARY KEY (space, id) + ); + CREATE TABLE IF NOT EXISTS team_links ( + space TEXT NOT NULL, + src INTEGER NOT NULL, + kind TEXT NOT NULL, + dst INTEGER NOT NULL, + UNIQUE (space, src, kind, dst) + ); + CREATE TABLE IF NOT EXISTS team_meta ( + space TEXT PRIMARY KEY, + head_hash TEXT NOT NULL, + watermark INTEGER NOT NULL + ); + """) + self._conn.commit() + + # ------------------------------------------------------------------ events core + + def append_event( + self, + space: str, + kind: str, + actor: Actor, + *, + item_id: Optional[int] = None, + case_id: Optional[str] = None, + recipient: Optional[str] = None, + payload: Optional[dict[str, Any]] = None, + taint: bool = False, + ) -> dict[str, Any]: + """Append one record and fold it into the projections, atomically.""" + if not space: + raise BoardError("space is required") + with self._lock: + try: + return self._append_locked( + space, + kind, + actor, + item_id=item_id, + case_id=case_id, + recipient=recipient, + payload=payload or {}, + taint=taint, + ) + except Exception: + self._conn.rollback() + raise + + def _append_locked( + self, + space: str, + kind: str, + actor: Actor, + *, + item_id: Optional[int], + case_id: Optional[str], + recipient: Optional[str], + payload: dict[str, Any], + taint: bool, + ) -> dict[str, Any]: + prev = self._head_hash(space) + record = { + "ts": datetime.now(timezone.utc).isoformat(), + "space": space, + "kind": kind, + "actor": actor.id, + "actor_role": actor.role.value, + "item_id": item_id, + "case_id": case_id, + "recipient": recipient, + "payload": _canonical(payload), + "taint": 1 if taint else 0, + "prev_hash": prev, + } + record["hash"] = _hash(record) + cursor = self._conn.execute( + """ + INSERT INTO team_events + (ts, space, kind, actor, actor_role, persona, model, session_id, + item_id, case_id, recipient, payload, taint, prev_hash, hash) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + record["ts"], + space, + kind, + actor.id, + actor.role.value, + actor.persona, + actor.model, + actor.session_id, + item_id, + case_id, + recipient, + record["payload"], + record["taint"], + prev, + record["hash"], + ), + ) + seq = cursor.lastrowid + self._apply(space, seq, record["ts"], kind, item_id, payload) + self._conn.execute( + """ + INSERT INTO team_meta (space, head_hash, watermark) VALUES (?, ?, ?) + ON CONFLICT(space) DO UPDATE SET head_hash = ?, watermark = ? + """, + (space, record["hash"], seq, record["hash"], seq), + ) + self._conn.commit() + return {**record, "seq": seq, "payload": payload} + + def events( + self, + space: str, + *, + kinds: Optional[list[str]] = None, + item_id: Optional[int] = None, + case_id: Optional[str] = None, + since_seq: int = 0, + limit: int = 500, + ) -> list[dict[str, Any]]: + where = ["space = ?", "seq > ?"] + params: list[Any] = [space, since_seq] + if kinds: + where.append(f"kind IN ({','.join('?' * len(kinds))})") + params.extend(kinds) + if item_id is not None: + where.append("item_id = ?") + params.append(item_id) + if case_id is not None: + where.append("case_id = ?") + params.append(case_id) + sql = ( + "SELECT * FROM team_events WHERE " + + " AND ".join(where) + + " ORDER BY seq LIMIT ?" + ) + params.append(max(1, min(int(limit or 500), 2000))) + with self._lock: + rows = self._conn.execute(sql, params).fetchall() + return [_row_to_event(row) for row in rows] + + def for_recipient( + self, recipient: str, *, since_seq: int = 0, limit: int = 200 + ) -> list[dict[str, Any]]: + """Everything addressed to one agent, in order — the delivery projection.""" + with self._lock: + rows = self._conn.execute( + "SELECT * FROM team_events WHERE recipient = ? AND seq > ?" + " ORDER BY seq LIMIT ?", + (recipient, since_seq, max(1, min(int(limit or 200), 2000))), + ).fetchall() + return [_row_to_event(row) for row in rows] + + def spaces(self) -> list[str]: + with self._lock: + rows = self._conn.execute( + "SELECT space FROM team_meta ORDER BY space" + ).fetchall() + return [row["space"] for row in rows] + + def verify_chain(self, space: str) -> int: + """Recompute the chain; return the number of verified events. + + Raises ChainError at the first record whose hash or linkage does not match — + the log was edited out of band. + """ + with self._lock: + rows = self._conn.execute( + "SELECT * FROM team_events WHERE space = ? ORDER BY seq", (space,) + ).fetchall() + prev = GENESIS + for row in rows: + record = {key: row[key] for key in _HASHED_FIELDS} + if row["prev_hash"] != prev: + raise ChainError(f"event {row['seq']}: chain linkage broken") + if _hash(record) != row["hash"]: + raise ChainError(f"event {row['seq']}: content does not match hash") + prev = row["hash"] + return len(rows) + + def rebuild(self, space: str) -> None: + """Drop the space's projections and replay its log through `_apply`. + + The recovery path (projection bug fix, cache corruption) — never the hot + path; live appends fold incrementally in `append_event`. + """ + with self._lock: + self._conn.execute("DELETE FROM team_items WHERE space = ?", (space,)) + self._conn.execute("DELETE FROM team_links WHERE space = ?", (space,)) + rows = self._conn.execute( + "SELECT seq, ts, kind, item_id, payload FROM team_events" + " WHERE space = ? ORDER BY seq", + (space,), + ).fetchall() + for row in rows: + self._apply( + space, + row["seq"], + row["ts"], + row["kind"], + row["item_id"], + json.loads(row["payload"]), + ) + self._conn.commit() + + # ------------------------------------------------------------------ board verbs + + def create_item( + self, + space: str, + actor: Actor, + *, + title: str, + criteria: str, + description: str = "", + parent: Optional[int] = None, + case: Optional[str] = None, + ) -> dict[str, Any]: + """New item in `proposed`. Acceptance criteria are load-bearing — required.""" + self._require(actor, {Role.USER, Role.LEAD}, "create_item") + if not (title or "").strip(): + raise BoardError("title is required") + if not (criteria or "").strip(): + raise BoardError( + "acceptance criteria are required — they are what gets verified at" + " review" + ) + with self._lock: + if parent is not None: + parent_item = self._item(space, parent) + if case is None: + case = parent_item["case_id"] or None + item_id = self._next_item_id(space) + event = self.append_event( + space, + ITEM_CREATED, + actor, + item_id=item_id, + case_id=case, + payload={ + "title": title.strip(), + "description": description, + "criteria": criteria.strip(), + "parent": parent, + "case": case, + }, + ) + return self.get_item(space, item_id, seq=event["seq"]) + + def list_items( + self, + space: str, + actor: Actor, + *, + state: Optional[str] = None, + assignee: Optional[str] = None, + ) -> list[dict[str, Any]]: + """Items in a space. Workers see only their slice: assigned items plus items + directly linked to those.""" + where = ["space = ?"] + params: list[Any] = [space] + if state: + where.append("state = ?") + params.append(ItemState(state).value) + if assignee: + where.append("assignee = ?") + params.append(assignee) + with self._lock: + rows = self._conn.execute( + "SELECT * FROM team_items WHERE " + + " AND ".join(where) + + " ORDER BY id", + params, + ).fetchall() + items = [dict(row) for row in rows] + if actor.role == Role.WORKER: + visible = self._worker_slice(space, actor.id) + items = [item for item in items if item["id"] in visible] + for item in items: + item["links"] = self._links_of(space, item["id"]) + return items + + def get_item( + self, space: str, item_id: int, *, seq: Optional[int] = None + ) -> dict[str, Any]: + with self._lock: + item = self._item(space, item_id) + item["links"] = self._links_of(space, item_id) + item["comments"] = self.comments(space, item_id) + if seq is not None: + item["seq"] = seq + return item + + def transition( + self, + space: str, + actor: Actor, + item_id: int, + to: str, + *, + comment: str = "", + taint: bool = False, + ) -> dict[str, Any]: + target = ItemState(to) + with self._lock: + item = self._item(space, item_id) + current = ItemState(item["state"]) + if target not in EDGES[current]: + raise BoardError( + f"illegal transition {current.value} → {target.value}" + ) + self._check_transition_authority(actor, item, current, target) + event = self.append_event( + space, + ITEM_TRANSITIONED, + actor, + item_id=item_id, + case_id=item["case_id"] or None, + payload={ + "from": current.value, + "to": target.value, + "comment": comment, + }, + taint=taint, + ) + return self.get_item(space, item_id, seq=event["seq"]) + + def comment( + self, + space: str, + actor: Actor, + item_id: int, + body: str, + *, + taint: bool = False, + ) -> dict[str, Any]: + if not (body or "").strip(): + raise BoardError("comment body is required") + with self._lock: + item = self._item(space, item_id) + if actor.role == Role.WORKER and item_id not in self._worker_slice( + space, actor.id + ): + raise AuthorityError( + f"worker {actor.id} may only comment on its assigned items" + " and items linked to them" + ) + return self.append_event( + space, + ITEM_COMMENTED, + actor, + item_id=item_id, + case_id=item["case_id"] or None, + payload={"body": body}, + taint=taint, + ) + + def assign( + self, space: str, actor: Actor, item_id: int, assignee: str + ) -> dict[str, Any]: + """Set the assignee. Not a message: the event addresses the assignee + (`recipient`), and the worker's prompt derives from the item itself.""" + self._require(actor, {Role.USER, Role.LEAD}, "assign") + if not (assignee or "").strip(): + raise BoardError("assignee is required") + with self._lock: + item = self._item(space, item_id) + state = ItemState(item["state"]) + if state in (ItemState.PROPOSED, ItemState.DONE, ItemState.CANCELED): + raise BoardError( + f"cannot assign an item in state {state.value} — items are" + " assigned after approval" + ) + event = self.append_event( + space, + ITEM_ASSIGNED, + actor, + item_id=item_id, + case_id=item["case_id"] or None, + recipient=assignee, + payload={"assignee": assignee, "previous": item["assignee"] or ""}, + ) + return self.get_item(space, item_id, seq=event["seq"]) + + def link( + self, space: str, actor: Actor, src: int, kind: str, dst: int + ) -> dict[str, Any]: + self._require(actor, {Role.USER, Role.LEAD}, "link") + if kind not in LINK_KINDS: + raise BoardError(f"unknown link kind: {kind} (use one of {LINK_KINDS})") + if src == dst: + raise BoardError("an item cannot link to itself") + with self._lock: + self._item(space, src) + self._item(space, dst) + if kind == "parent" and self._would_cycle(space, src, dst): + raise BoardError("parent link would create a cycle") + return self.append_event( + space, + ITEM_LINKED, + actor, + item_id=src, + payload={"src": src, "kind": kind, "dst": dst}, + ) + + def comments(self, space: str, item_id: int) -> list[dict[str, Any]]: + """Attributed comments on an item — standalone comments plus the notes + carried on transitions (a `blocked` explanation lives with its event).""" + out = [] + for event in self.events( + space, kinds=[ITEM_COMMENTED, ITEM_TRANSITIONED], item_id=item_id + ): + body = ( + event["payload"].get("body") + if event["kind"] == ITEM_COMMENTED + else event["payload"].get("comment") + ) + if body: + out.append( + { + "seq": event["seq"], + "ts": event["ts"], + "author": event["actor"], + "role": event["actor_role"], + "body": body, + "taint": event["taint"], + } + ) + return out + + # ---------------------------------------------------------------------- journal + + def journal_append( + self, + space: str, + actor: Actor, + case: str, + body: str, + *, + kind: str = "note", + item: Optional[int] = None, + entities: Optional[list[str]] = None, + refs: Optional[list[str]] = None, + taint: bool = False, + ) -> dict[str, Any]: + """Append one journal entry to a case. Cases are created lazily on first + append; sharing rides assignment — a worker may only touch cases referenced + by its assigned items.""" + if not (case or "").strip(): + raise BoardError("case is required") + if not (body or "").strip(): + raise BoardError("entry body is required") + if kind not in JOURNAL_KINDS: + raise BoardError(f"unknown entry kind: {kind} (use one of {JOURNAL_KINDS})") + with self._lock: + self._check_case_access(space, actor, case) + if item is not None: + self._item(space, item) + return self.append_event( + space, + JOURNAL_APPENDED, + actor, + item_id=item, + case_id=case, + payload={ + "kind": kind, + "body": body, + "entities": sorted(set(entities or [])), + "refs": list(refs or []), + }, + taint=taint, + ) + + def journal_read( + self, + space: str, + actor: Actor, + case: str, + *, + item: Optional[int] = None, + author: Optional[str] = None, + kind: Optional[str] = None, + entity: Optional[str] = None, + since_seq: int = 0, + limit: int = 100, + ) -> list[dict[str, Any]]: + """Filtered read — the thing that makes shared journals viable. The entity + filter scans extracted entities for now; a dedicated index (then vectors) + drops in behind this same signature.""" + with self._lock: + self._check_case_access(space, actor, case) + events = self.events( + space, + kinds=[JOURNAL_APPENDED], + case_id=case, + item_id=item, + since_seq=since_seq, + limit=2000, + ) + out = [] + for event in events: + payload = event["payload"] + if author and event["actor"] != author: + continue + if kind and payload.get("kind") != kind: + continue + if entity and entity not in (payload.get("entities") or []): + continue + out.append( + { + "seq": event["seq"], + "ts": event["ts"], + "author": event["actor"], + "role": event["actor_role"], + "item": event["item_id"], + "kind": payload.get("kind"), + "body": payload.get("body"), + "entities": payload.get("entities") or [], + "refs": payload.get("refs") or [], + "taint": event["taint"], + } + ) + if len(out) >= max(1, min(int(limit or 100), 1000)): + break + return out + + def cases(self, space: str) -> list[str]: + with self._lock: + rows = self._conn.execute( + "SELECT DISTINCT case_id FROM team_events" + " WHERE space = ? AND kind = ? AND case_id IS NOT NULL" + " ORDER BY case_id", + (space, JOURNAL_APPENDED), + ).fetchall() + return [row["case_id"] for row in rows] + + def close(self) -> None: + self._conn.close() + + # ------------------------------------------------------------------- internals + + def _apply( + self, + space: str, + seq: int, + ts: str, + kind: str, + item_id: Optional[int], + payload: dict[str, Any], + ) -> None: + """Fold one event into the projections. The ONLY writer of team_items and + team_links — shared by live appends and rebuild(), so replay always + reproduces the materialized state.""" + if kind == ITEM_CREATED: + self._conn.execute( + """ + INSERT INTO team_items + (space, id, title, description, criteria, state, assignee, + case_id, created_ts, updated_seq) + VALUES (?, ?, ?, ?, ?, ?, '', ?, ?, ?) + """, + ( + space, + item_id, + payload.get("title") or "", + payload.get("description") or "", + payload.get("criteria") or "", + ItemState.PROPOSED.value, + payload.get("case") or "", + ts, + seq, + ), + ) + if payload.get("parent") is not None: + self._conn.execute( + "INSERT OR IGNORE INTO team_links (space, src, kind, dst)" + " VALUES (?, ?, 'parent', ?)", + (space, item_id, payload["parent"]), + ) + elif kind == ITEM_TRANSITIONED: + self._conn.execute( + "UPDATE team_items SET state = ?, updated_seq = ?" + " WHERE space = ? AND id = ?", + (payload.get("to"), seq, space, item_id), + ) + elif kind == ITEM_ASSIGNED: + self._conn.execute( + "UPDATE team_items SET assignee = ?, updated_seq = ?" + " WHERE space = ? AND id = ?", + (payload.get("assignee") or "", seq, space, item_id), + ) + elif kind == ITEM_LINKED: + self._conn.execute( + "INSERT OR IGNORE INTO team_links (space, src, kind, dst)" + " VALUES (?, ?, ?, ?)", + (space, payload.get("src"), payload.get("kind"), payload.get("dst")), + ) + # Comments and journal entries have no materialized state: their + # projections read straight off the (indexed) log. + + def _check_transition_authority( + self, actor: Actor, item: dict[str, Any], current: ItemState, target: ItemState + ) -> None: + if actor.role == Role.SYSTEM: + raise AuthorityError("system events cannot transition items") + if current == ItemState.PROPOSED and target == ItemState.APPROVED: + if actor.role != Role.USER: + raise AuthorityError( + "only the user approves proposed items — that is the" + " decomposition gate" + ) + return + if target == ItemState.DONE and actor.role == Role.WORKER: + raise AuthorityError( + "workers finish by moving to review — done is the verdict after" + " verification" + ) + if actor.role == Role.WORKER: + if item["assignee"] != actor.id: + raise AuthorityError( + f"worker {actor.id} is not assigned item #{item['id']}" + ) + if target not in WORKER_TARGETS: + raise AuthorityError( + f"workers may move their item to" + f" {sorted(state.value for state in WORKER_TARGETS)} only" + ) + + def _check_case_access(self, space: str, actor: Actor, case: str) -> None: + if actor.role != Role.WORKER: + return + rows = self._conn.execute( + "SELECT DISTINCT case_id FROM team_items WHERE space = ? AND assignee = ?", + (space, actor.id), + ).fetchall() + if case not in {row["case_id"] for row in rows if row["case_id"]}: + raise AuthorityError( + f"worker {actor.id} has no assigned item on case '{case}'" + ) + + def _worker_slice(self, space: str, worker_id: str) -> set[int]: + rows = self._conn.execute( + "SELECT id FROM team_items WHERE space = ? AND assignee = ?", + (space, worker_id), + ).fetchall() + mine = {row["id"] for row in rows} + if not mine: + return set() + linked = self._conn.execute( + "SELECT src, dst FROM team_links WHERE space = ?", (space,) + ).fetchall() + out = set(mine) + for row in linked: + if row["src"] in mine: + out.add(row["dst"]) + if row["dst"] in mine: + out.add(row["src"]) + return out + + def _links_of(self, space: str, item_id: int) -> list[dict[str, Any]]: + rows = self._conn.execute( + "SELECT src, kind, dst FROM team_links WHERE space = ?" + " AND (src = ? OR dst = ?)", + (space, item_id, item_id), + ).fetchall() + out = [] + for row in rows: + if row["src"] == item_id: + out.append({"kind": row["kind"], "item": row["dst"]}) + else: + inverse = "child" if row["kind"] == "parent" else "blocked_by" + out.append({"kind": inverse, "item": row["src"]}) + return out + + def _would_cycle(self, space: str, src: int, dst: int) -> bool: + # Walking up from dst: if we reach src, making dst the parent of src closes + # a loop. + current, hops = dst, 0 + while hops < 1000: + row = self._conn.execute( + "SELECT dst FROM team_links WHERE space = ? AND src = ?" + " AND kind = 'parent'", + (space, current), + ).fetchone() + if row is None: + return False + if row["dst"] == src: + return True + current, hops = row["dst"], hops + 1 + return True + + def _item(self, space: str, item_id: int) -> dict[str, Any]: + row = self._conn.execute( + "SELECT * FROM team_items WHERE space = ? AND id = ?", (space, item_id) + ).fetchone() + if row is None: + raise BoardError(f"no item #{item_id} in space '{space}'") + return dict(row) + + def _next_item_id(self, space: str) -> int: + row = self._conn.execute( + "SELECT MAX(id) AS top FROM team_items WHERE space = ?", (space,) + ).fetchone() + return int(row["top"] or 0) + 1 + + def _head_hash(self, space: str) -> str: + row = self._conn.execute( + "SELECT head_hash FROM team_meta WHERE space = ?", (space,) + ).fetchone() + return row["head_hash"] if row else GENESIS + + def _require(self, actor: Actor, roles: set[Role], verb: str) -> None: + if actor.role not in roles: + raise AuthorityError( + f"{verb} requires one of" + f" {sorted(role.value for role in roles)} (actor {actor.id} is" + f" {actor.role.value})" + ) + + +def _canonical(payload: dict[str, Any]) -> str: + return json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str) + + +def _hash(record: dict[str, Any]) -> str: + material = _canonical({key: record[key] for key in _HASHED_FIELDS}) + return hashlib.sha256(material.encode("utf-8")).hexdigest() + + +def _row_to_event(row: sqlite3.Row) -> dict[str, Any]: + event = dict(row) + try: + event["payload"] = json.loads(event.get("payload") or "{}") + except json.JSONDecodeError: + event["payload"] = {} + return event diff --git a/coworker/teams/tools.py b/coworker/teams/tools.py new file mode 100644 index 00000000..21e75727 --- /dev/null +++ b/coworker/teams/tools.py @@ -0,0 +1,210 @@ +"""Board and journal verbs as agent tools. + +The verbs are generic on purpose (the connector-dialect play): the local TeamStore is +the default backing, and a Jira/Linear-backed dialect can implement the same tool +surface later. Registration is gated by the persona's `team:` trait — a lead gets the +full set, a worker gets the worker set, solo personas get none of this. + +The engine decides `taint` (whether this agent touched untrusted content this +session) and passes it at construction — the model never self-reports provenance. +""" + +from __future__ import annotations + +from typing import Callable, Optional + +import aisuite as ai + +from .model import Actor, BoardError, Role +from .store import TeamStore + +LEAD_VERBS = ("create_item", "list_items", "transition", "comment", "assign", "link") +WORKER_VERBS = ("list_items", "transition", "comment") +JOURNAL_VERBS = ("journal_append", "journal_read") + +# Explicit schema: the auto-generator's normalizer strips every `title` key to drop +# pydantic metadata, which also deletes a PARAMETER named `title` from properties. +# Registered via `__coworker_schema__` (same escape hatch as todo_write). +_CREATE_ITEM_SCHEMA = { + "type": "function", + "function": { + "name": "create_item", + "description": ( + "Create a work item in the proposed state. `criteria` is the acceptance" + " criteria — what gets verified before the item can be done; required." + " `parent` links it under another item; `case` names its journal case" + " (children inherit the parent's case by default)." + ), + "parameters": { + "type": "object", + "properties": { + "title": {"type": "string"}, + "criteria": {"type": "string"}, + "description": {"type": "string"}, + "parent": {"type": "integer"}, + "case": {"type": "string"}, + }, + "required": ["title", "criteria"], + }, + }, +} + + +def board_tools( + store: TeamStore, + *, + space: str, + actor: Actor, + taint: Callable[[], bool] = lambda: False, +) -> list: + """The board verbs for one agent, pre-bound to its space and identity. + + Authority is enforced twice on purpose: the returned set is role-filtered + (a worker never even sees `assign`), and the store re-checks every call — + the tool layer is convenience, the store is the gate. + """ + + def create_item( + title: str, + criteria: str, + description: str = "", + parent: Optional[int] = None, + case: str = "", + ) -> dict: + """Create a work item in the proposed state. `criteria` is the acceptance + criteria — what gets verified before the item can be done; required. + `parent` links it under another item; `case` names its journal case + (children inherit the parent's case by default).""" + return _call( + store.create_item, + space, + actor, + title=title, + criteria=criteria, + description=description, + parent=parent, + case=case or None, + ) + + def list_items(state: str = "", assignee: str = "") -> dict: + """List work items on the board, optionally filtered by state + (proposed/approved/in_progress/blocked/review/done/canceled) or assignee.""" + try: + return {"items": store.list_items(space, actor, state=state or None, assignee=assignee or None)} + except (BoardError, ValueError) as error: + return {"error": str(error)} + + def transition(item: int, to: str, comment: str = "") -> dict: + """Move a work item to a new state. Workers move their own item to + in_progress, blocked, or review (attach the blocker or a hand-off summary + as `comment`); done requires review verification first.""" + return _call( + store.transition, space, actor, item, to, comment=comment, taint=taint() + ) + + def comment(item: int, body: str) -> dict: + """Add a comment to a work item. Comments are durable and attributed — + answers that matter belong here, not in chat.""" + return _call(store.comment, space, actor, item, body, taint=taint()) + + def assign(item: int, assignee: str) -> dict: + """Assign a work item to a worker coworker. The item itself becomes the + worker's assignment — write the description and criteria accordingly.""" + return _call(store.assign, space, actor, item, assignee) + + def link(src: int, kind: str, dst: int) -> dict: + """Link two work items: kind `parent` (dst becomes src's parent) or + `blocks` (src blocks dst).""" + return _call(store.link, space, actor, src, kind, dst) + + verbs = LEAD_VERBS if actor.role in (Role.USER, Role.LEAD) else WORKER_VERBS + local = locals() + out = [] + for name in verbs: + wrapped = _wrap(local[name]) + if name == "create_item": + wrapped.__coworker_schema__ = _CREATE_ITEM_SCHEMA + out.append(wrapped) + return out + + +def journal_tools( + store: TeamStore, + *, + space: str, + actor: Actor, + taint: Callable[[], bool] = lambda: False, +) -> list: + def journal_append( + case: str, + body: str, + kind: str = "note", + item: Optional[int] = None, + entities: Optional[list] = None, + refs: Optional[list] = None, + ) -> dict: + """Append an entry to a journal case as you work: kind is finding, + evidence, decision, or note. `entities` are the concrete things it is + about (file paths, resource names, CVE ids) — they power later recall; + `refs` are pointers (file:line, commit, url).""" + return _call( + store.journal_append, + space, + actor, + case, + body, + kind=kind, + item=item, + entities=[str(entity) for entity in entities or []], + refs=[str(ref) for ref in refs or []], + taint=taint(), + ) + + def journal_read( + case: str, + item: Optional[int] = None, + author: str = "", + kind: str = "", + entity: str = "", + limit: int = 50, + ) -> dict: + """Read a journal case, filtered: by item, author, entry kind, or entity. + Prefer narrow filtered reads over pulling the whole case.""" + try: + return { + "entries": store.journal_read( + space, + actor, + case, + item=item, + author=author or None, + kind=kind or None, + entity=entity or None, + limit=limit, + ) + } + except (BoardError, ValueError) as error: + return {"error": str(error)} + + local = locals() + return [_wrap(local[name]) for name in JOURNAL_VERBS] + + +def _call(func, *args, **kwargs) -> dict: + try: + result = func(*args, **kwargs) + return result if isinstance(result, dict) else {"ok": True} + except (BoardError, ValueError) as error: + return {"error": str(error)} + + +def _wrap(func): + risk = "medium" if func.__name__ == "assign" else "low" + return ai.tool( + func, + metadata=ai.ToolMetadata( + category="team", + risk_level=risk, + capabilities=["team"], + ), + ) diff --git a/tests/test_team_board.py b/tests/test_team_board.py new file mode 100644 index 00000000..02ce3e71 --- /dev/null +++ b/tests/test_team_board.py @@ -0,0 +1,204 @@ +"""Board verbs: state-machine legality and role authority (the capability firebreak).""" + +import pytest + +from coworker.teams import Actor, AuthorityError, BoardError, Role, TeamStore +from coworker.teams.tools import board_tools + +USER = Actor(id="user", role=Role.USER) +LEAD = Actor(id="lead-1", role=Role.LEAD) +WORKER = Actor(id="worker-1", role=Role.WORKER) +OTHER = Actor(id="worker-2", role=Role.WORKER) +SPACE = "proj" + + +@pytest.fixture +def store(tmp_path): + store = TeamStore(tmp_path / "teams.db") + yield store + store.close() + + +def assigned_item(store, assignee="worker-1"): + item = store.create_item(SPACE, LEAD, title="Task", criteria="tests pass") + store.transition(SPACE, USER, item["id"], "approved") + store.assign(SPACE, LEAD, item["id"], assignee) + return item["id"] + + +# ------------------------------------------------------------------ create_item + +def test_acceptance_criteria_are_required(store): + with pytest.raises(BoardError, match="criteria"): + store.create_item(SPACE, LEAD, title="Vague hope", criteria=" ") + + +def test_workers_cannot_create_items(store): + with pytest.raises(AuthorityError): + store.create_item(SPACE, WORKER, title="Scope creep", criteria="c") + + +def test_child_inherits_parent_case(store): + parent = store.create_item( + SPACE, LEAD, title="Root", criteria="c", case="findings" + ) + child = store.create_item( + SPACE, LEAD, title="Child", criteria="c", parent=parent["id"] + ) + assert child["case_id"] == "findings" + assert {"kind": "parent", "item": parent["id"]} in child["links"] + + +# ------------------------------------------------------------------- transitions + +def test_full_happy_path(store): + item_id = assigned_item(store) + store.transition(SPACE, WORKER, item_id, "in_progress") + store.transition(SPACE, WORKER, item_id, "review", comment="branch green") + done = store.transition(SPACE, LEAD, item_id, "done") + assert done["state"] == "done" + + +def test_illegal_edges_rejected(store): + item = store.create_item(SPACE, LEAD, title="T", criteria="c") + with pytest.raises(BoardError, match="illegal transition"): + store.transition(SPACE, USER, item["id"], "done") + with pytest.raises(BoardError, match="illegal transition"): + store.transition(SPACE, USER, item["id"], "in_progress") + + +def test_only_the_user_approves(store): + item = store.create_item(SPACE, LEAD, title="T", criteria="c") + with pytest.raises(AuthorityError, match="decomposition gate"): + store.transition(SPACE, LEAD, item["id"], "approved") + approved = store.transition(SPACE, USER, item["id"], "approved") + assert approved["state"] == "approved" + + +def test_workers_never_mark_done(store): + item_id = assigned_item(store) + store.transition(SPACE, WORKER, item_id, "in_progress") + store.transition(SPACE, WORKER, item_id, "review") + with pytest.raises(AuthorityError, match="review"): + store.transition(SPACE, WORKER, item_id, "done") + + +def test_worker_cannot_touch_someone_elses_item(store): + item_id = assigned_item(store, assignee="worker-1") + with pytest.raises(AuthorityError, match="not assigned"): + store.transition(SPACE, OTHER, item_id, "in_progress") + + +def test_worker_cannot_cancel(store): + item_id = assigned_item(store) + with pytest.raises(AuthorityError): + store.transition(SPACE, WORKER, item_id, "canceled") + + +def test_cancel_is_a_board_verb_and_reopen_works(store): + item_id = assigned_item(store) + store.transition(SPACE, LEAD, item_id, "canceled") + reopened = store.transition(SPACE, LEAD, item_id, "approved") + assert reopened["state"] == "approved" + assert reopened["assignee"] == "worker-1" # reassignable — assignment survives + + +def test_done_is_terminal(store): + item_id = assigned_item(store) + store.transition(SPACE, WORKER, item_id, "in_progress") + store.transition(SPACE, WORKER, item_id, "review") + store.transition(SPACE, LEAD, item_id, "done") + with pytest.raises(BoardError, match="illegal transition"): + store.transition(SPACE, USER, item_id, "in_progress") + + +def test_rework_loop(store): + item_id = assigned_item(store) + store.transition(SPACE, WORKER, item_id, "in_progress") + store.transition(SPACE, WORKER, item_id, "review") + store.transition(SPACE, LEAD, item_id, "in_progress", comment="criteria 2 unmet") + item = store.get_item(SPACE, item_id) + assert item["state"] == "in_progress" + assert item["comments"][-1]["body"] == "criteria 2 unmet" + + +# ---------------------------------------------------------------- assign / link + +def test_cannot_assign_before_approval(store): + item = store.create_item(SPACE, LEAD, title="T", criteria="c") + with pytest.raises(BoardError, match="after approval"): + store.assign(SPACE, LEAD, item["id"], "worker-1") + + +def test_workers_cannot_assign_or_link(store): + item_id = assigned_item(store) + with pytest.raises(AuthorityError): + store.assign(SPACE, WORKER, item_id, "worker-2") + other = store.create_item(SPACE, LEAD, title="B", criteria="c") + with pytest.raises(AuthorityError): + store.link(SPACE, WORKER, item_id, "blocks", other["id"]) + + +def test_parent_cycles_rejected(store): + a = store.create_item(SPACE, LEAD, title="A", criteria="c") + b = store.create_item(SPACE, LEAD, title="B", criteria="c", parent=a["id"]) + with pytest.raises(BoardError, match="cycle"): + store.link(SPACE, LEAD, a["id"], "parent", b["id"]) + + +def test_blocked_by_shows_on_the_other_side(store): + a = store.create_item(SPACE, LEAD, title="A", criteria="c") + b = store.create_item(SPACE, LEAD, title="B", criteria="c") + store.link(SPACE, LEAD, a["id"], "blocks", b["id"]) + assert {"kind": "blocked_by", "item": a["id"]} in store.get_item(SPACE, b["id"])[ + "links" + ] + + +# ------------------------------------------------------------- worker visibility + +def test_worker_sees_only_its_slice(store): + mine = assigned_item(store, assignee="worker-1") + theirs = assigned_item(store, assignee="worker-2") + linked = store.create_item(SPACE, LEAD, title="Dep", criteria="c") + store.link(SPACE, LEAD, linked["id"], "blocks", mine) + visible = {item["id"] for item in store.list_items(SPACE, WORKER)} + assert visible == {mine, linked["id"]} + assert theirs not in visible + # the user sees everything + assert len(store.list_items(SPACE, USER)) == 3 + + +def test_worker_comments_only_on_its_slice(store): + theirs = assigned_item(store, assignee="worker-2") + with pytest.raises(AuthorityError, match="assigned"): + store.comment(SPACE, WORKER, theirs, "drive-by") + + +# ------------------------------------------------------------------- tool layer + +def test_tool_sets_are_role_filtered(store): + lead_names = {tool.__name__ for tool in board_tools(store, space=SPACE, actor=LEAD)} + worker_names = { + tool.__name__ for tool in board_tools(store, space=SPACE, actor=WORKER) + } + assert lead_names == {"create_item", "list_items", "transition", "comment", "assign", "link"} + assert worker_names == {"list_items", "transition", "comment"} + + +def test_tools_return_errors_instead_of_raising(store): + tools = {t.__name__: t for t in board_tools(store, space=SPACE, actor=WORKER)} + result = tools["transition"](item=99, to="in_progress") + assert "error" in result + + +def test_create_item_tool_schema_keeps_the_title_parameter(store): + from coworker.tools.registry import ToolRegistry + + registry = ToolRegistry() + registry.register_all(board_tools(store, space=SPACE, actor=LEAD)) + schema = registry.get("create_item").schema + properties = schema["function"]["parameters"]["properties"] + # The auto-generator strips dict keys named "title" (pydantic metadata cleanup), + # which would delete this parameter; the explicit schema must keep it. + assert "title" in properties diff --git a/tests/test_team_journal.py b/tests/test_team_journal.py new file mode 100644 index 00000000..1f24b842 --- /dev/null +++ b/tests/test_team_journal.py @@ -0,0 +1,110 @@ +"""Journal cases: filtered reads, access-rides-assignment, lazy case lifecycle.""" + +import pytest + +from coworker.teams import Actor, AuthorityError, BoardError, Role, TeamStore + +USER = Actor(id="user", role=Role.USER) +LEAD = Actor(id="lead-1", role=Role.LEAD) +WORKER = Actor(id="worker-1", role=Role.WORKER) +OTHER = Actor(id="worker-2", role=Role.WORKER) +SPACE = "proj" + + +@pytest.fixture +def store(tmp_path): + store = TeamStore(tmp_path / "teams.db") + yield store + store.close() + + +def case_item(store, case="findings", assignee="worker-1"): + item = store.create_item(SPACE, LEAD, title="Task", criteria="c", case=case) + store.transition(SPACE, USER, item["id"], "approved") + store.assign(SPACE, LEAD, item["id"], assignee) + return item["id"] + + +def test_cases_are_lazy_and_listed(store): + assert store.cases(SPACE) == [] + item_id = case_item(store) + store.journal_append( + SPACE, WORKER, "findings", "public ACL on uploads", kind="finding", item=item_id + ) + assert store.cases(SPACE) == ["findings"] + + +def test_filtered_reads(store): + item_id = case_item(store) + store.journal_append( + SPACE, + WORKER, + "findings", + "logos bucket is world-readable", + kind="finding", + item=item_id, + entities=["aws_s3_bucket.assets", "uploads.ts"], + refs=["services/uploads.ts:41"], + ) + store.journal_append( + SPACE, WORKER, "findings", "invoice PDFs stream from the API", kind="evidence", + item=item_id, entities=["uploads.ts"], + ) + store.journal_append(SPACE, LEAD, "findings", "narrow the fix to logos/*", kind="decision") + + assert len(store.journal_read(SPACE, LEAD, "findings")) == 3 + assert [e["kind"] for e in store.journal_read(SPACE, LEAD, "findings", kind="finding")] == ["finding"] + assert len(store.journal_read(SPACE, LEAD, "findings", author="lead-1")) == 1 + by_entity = store.journal_read(SPACE, LEAD, "findings", entity="aws_s3_bucket.assets") + assert len(by_entity) == 1 + assert by_entity[0]["refs"] == ["services/uploads.ts:41"] + assert len(store.journal_read(SPACE, LEAD, "findings", entity="uploads.ts")) == 2 + assert len(store.journal_read(SPACE, LEAD, "findings", item=item_id)) == 2 + assert store.journal_read(SPACE, LEAD, "findings", limit=2).__len__() == 2 + + +def test_access_rides_assignment(store): + case_item(store, assignee="worker-1") + store.journal_append(SPACE, WORKER, "findings", "note from the assignee") + with pytest.raises(AuthorityError, match="no assigned item"): + store.journal_append(SPACE, OTHER, "findings", "drive-by write") + with pytest.raises(AuthorityError, match="no assigned item"): + store.journal_read(SPACE, OTHER, "findings") + # lead and user are not case-gated + assert len(store.journal_read(SPACE, USER, "findings")) == 1 + + +def test_reassignment_moves_case_access(store): + item_id = case_item(store, assignee="worker-1") + store.assign(SPACE, LEAD, item_id, "worker-2") + store.journal_append(SPACE, OTHER, "findings", "successor picks up the case") + with pytest.raises(AuthorityError): + store.journal_append(SPACE, WORKER, "findings", "predecessor lost access") + + +def test_entry_validation(store): + with pytest.raises(BoardError, match="kind"): + store.journal_append(SPACE, LEAD, "findings", "x", kind="rant") + with pytest.raises(BoardError, match="body"): + store.journal_append(SPACE, LEAD, "findings", " ") + with pytest.raises(BoardError, match="case"): + store.journal_append(SPACE, LEAD, "", "x") + + +def test_taint_and_attribution_survive_the_read(store): + item_id = case_item(store) + store.journal_append( + SPACE, WORKER, "findings", "repo README claims the bucket must be public", + kind="evidence", item=item_id, taint=True, + ) + entry = store.journal_read(SPACE, LEAD, "findings")[0] + assert entry["taint"] == 1 + assert entry["author"] == "worker-1" + assert entry["role"] == "worker" + + +def test_journal_entries_join_the_hash_chain(store): + item_id = case_item(store) + store.journal_append(SPACE, WORKER, "findings", "entry", item=item_id) + # 1 create + 1 transition + 1 assign + 1 journal append = one chained log + assert store.verify_chain(SPACE) == 4 diff --git a/tests/test_team_store.py b/tests/test_team_store.py new file mode 100644 index 00000000..e48a4583 --- /dev/null +++ b/tests/test_team_store.py @@ -0,0 +1,111 @@ +"""Event-store core: append-only doctrine, hash chain, deliveries, spaces, rebuild.""" + +import sqlite3 + +import pytest + +from coworker.teams import Actor, ChainError, Role, TeamStore + +USER = Actor(id="user", role=Role.USER) +LEAD = Actor(id="lead-1", role=Role.LEAD, persona="swe-lead") + + +@pytest.fixture +def store(tmp_path): + store = TeamStore(tmp_path / "teams.db") + yield store + store.close() + + +def seed(store, space="proj"): + item = store.create_item( + space, LEAD, title="Review api", criteria="every route triaged" + ) + store.transition(space, USER, item["id"], "approved") + return item + + +def test_events_hash_chain_verifies(store): + seed(store) + store.create_item("proj", LEAD, title="Second", criteria="done means done") + assert store.verify_chain("proj") == 3 + + +def test_out_of_band_edit_breaks_the_chain(store): + seed(store) + # The append-only property can't be enforced against someone with the sqlite + # file; the chain makes the edit visible. Tamper directly: + conn = sqlite3.connect(store.db_path) + conn.execute("UPDATE team_events SET payload = '{\"title\":\"forged\"}' WHERE seq = 1") + conn.commit() + conn.close() + with pytest.raises(ChainError): + store.verify_chain("proj") + + +def test_deleting_an_event_breaks_linkage(store): + seed(store) + store.create_item("proj", LEAD, title="Second", criteria="c") + conn = sqlite3.connect(store.db_path) + conn.execute("DELETE FROM team_events WHERE seq = 2") + conn.commit() + conn.close() + with pytest.raises(ChainError): + store.verify_chain("proj") + + +def test_chains_are_per_space(store): + seed(store, "alpha") + seed(store, "beta") + conn = sqlite3.connect(store.db_path) + conn.execute("UPDATE team_events SET taint = 1 WHERE space = 'beta'") + conn.commit() + conn.close() + assert store.verify_chain("alpha") == 2 # untouched space still verifies + with pytest.raises(ChainError): + store.verify_chain("beta") + + +def test_spaces_created_lazily_and_isolated(store): + assert store.spaces() == [] + seed(store, "alpha") + seed(store, "beta") + assert store.spaces() == ["alpha", "beta"] + assert [i["id"] for i in store.list_items("alpha", USER)] == [1] # ids per space + assert [i["id"] for i in store.list_items("beta", USER)] == [1] + + +def test_assignment_lands_in_the_assignees_deliveries(store): + item = seed(store) + store.assign("proj", LEAD, item["id"], "worker-1") + deliveries = store.for_recipient("worker-1") + assert len(deliveries) == 1 + assert deliveries[0]["kind"] == "item_assigned" + assert deliveries[0]["payload"]["assignee"] == "worker-1" + assert store.for_recipient("worker-2") == [] + + +def test_rebuild_reproduces_the_projection(store): + item = seed(store) + worker = Actor(id="worker-1", role=Role.WORKER) + store.assign("proj", LEAD, item["id"], "worker-1") + store.transition("proj", worker, item["id"], "in_progress") + store.comment("proj", worker, item["id"], "halfway", taint=True) + before = store.list_items("proj", USER) + store.rebuild("proj") + after = store.list_items("proj", USER) + assert after == before + assert after[0]["state"] == "in_progress" + assert after[0]["assignee"] == "worker-1" + # created_ts comes from the event, so replay is deterministic + assert store.get_item("proj", item["id"])["created_ts"] == item["created_ts"] + + +def test_taint_travels_with_the_record(store): + item = seed(store) + store.assign("proj", LEAD, item["id"], "worker-1") + worker = Actor(id="worker-1", role=Role.WORKER) + store.comment("proj", worker, item["id"], "repo says the bucket is public", taint=True) + comments = store.comments("proj", item["id"]) + assert comments[-1]["taint"] == 1 + assert comments[-1]["role"] == "worker"