From 1d12fe96d2de17a94892b90b9145de6db76f7bcf Mon Sep 17 00:00:00 2001 From: Rohit C Prasad Date: Sun, 16 Aug 2026 06:44:20 -0700 Subject: [PATCH] Journal splits from the board: case-keyed store, grants ride assignment Cases outlive boards/teams, so entries+per-case chains move to journal.db with a grant table (creator-on-attach, assignment-fed, explicit shares). Adds the raw capture kind: excerpt inline under a body cap, full payload as a sha256-referenced artifact; reads skip raw unless asked. --- coworker/server/manager.py | 12 +- coworker/teams/__init__.py | 7 +- coworker/teams/journal.py | 412 +++++++++++++++++++++++++++++++++++++ coworker/teams/model.py | 9 +- coworker/teams/store.py | 156 +++----------- coworker/teams/tools.py | 25 ++- tests/test_team_journal.py | 229 ++++++++++++++------- 7 files changed, 623 insertions(+), 227 deletions(-) create mode 100644 coworker/teams/journal.py diff --git a/coworker/server/manager.py b/coworker/server/manager.py index 2a5693be..3e923eea 100644 --- a/coworker/server/manager.py +++ b/coworker/server/manager.py @@ -82,7 +82,7 @@ from ..providers import ( ) from ..secrets import SecretStore, state_dir from ..sessions import SessionRecord -from ..teams import TeamStore +from ..teams import JournalStore, TeamStore from ..skills import ( SessionSkillStore, SkillLoader, @@ -197,10 +197,12 @@ 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") + # Agent teams: two append-only stores, one record discipline. The journal is + # case-keyed (knowledge outlives boards/teams); the board log is space-scoped, + # and assignment feeds journal-case grants. Verbs register per-session behind + # the persona's `team:` trait (wake plumbing lands separately). + self.journal_store = JournalStore(base / "journal.db") + self.team_store = TeamStore(base / "teams.db", journal=self.journal_store) # 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 index 7d4ad2aa..c2a2a9b4 100644 --- a/coworker/teams/__init__.py +++ b/coworker/teams/__init__.py @@ -1,6 +1,8 @@ -"""Agent teams substrate: one append-only event store; the board, journal, and -per-agent deliveries are projections of it.""" +"""Agent teams substrate. Two append-only stores, one record discipline: +the board log (space-scoped — a board lives and dies with its team) and the +journal store (case-keyed — knowledge that outlives boards and teams).""" +from .journal import JournalStore from .model import ( Actor, AuthorityError, @@ -18,6 +20,7 @@ __all__ = [ "BoardError", "ChainError", "ItemState", + "JournalStore", "Role", "TeamStore", "board_tools", diff --git a/coworker/teams/journal.py b/coworker/teams/journal.py new file mode 100644 index 00000000..c20260f6 --- /dev/null +++ b/coworker/teams/journal.py @@ -0,0 +1,412 @@ +"""The journal store — case-keyed knowledge that outlives boards and teams. + +Split from the board log on purpose (decided 2026-08-16): a board is a team-scoped +artifact and can be archived with its team, but a journal case follows the +INVESTIGATION — it may span two boards, survive a team, or belong to an Ops case no +board ever references. So cases live in their own store, hash-chained per case, with +their own grant table. What stays unified with the board is the record shape and the +discipline: attributed, timestamped, append-only, taint-flagged — the policy/audit +choke point is the API layer, not table co-location. + +Access model: the user is never gated. Everyone else needs a grant on the case: +- creating a case (first append) grants its creator; +- assignment feeds grants automatically (assign an item carrying a case → the + assignee gains it; reassignment moves it) — "sharing rides assignment"; +- explicit grants cover cross-team sharing. + +Backing is SQLite for now (same as everything else in the state dir); the store is +deliberately small enough to swap the backing later without touching the verb +surface. Retrieval order stays: filters (here) → entity index → vectors as a +derived index. +""" + +from __future__ import annotations + +import json +import sqlite3 +import threading +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Optional + +from .model import ( + JOURNAL_BODY_LIMIT, + JOURNAL_KINDS, + Actor, + AuthorityError, + BoardError, + ChainError, + Role, +) +from .store import GENESIS, _canonical, _hash + +_HASHED_FIELDS = ( + "ts", + "case_id", + "kind", + "actor", + "actor_role", + "space", + "item_id", + "payload", + "taint", + "prev_hash", +) + + +class JournalStore: + 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 journal_entries ( + seq INTEGER PRIMARY KEY AUTOINCREMENT, + ts TEXT NOT NULL, + case_id 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 '', + space TEXT, + item_id INTEGER, + 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_journal_case + ON journal_entries (case_id, seq); + CREATE INDEX IF NOT EXISTS idx_journal_item + ON journal_entries (case_id, space, item_id, seq); + CREATE TABLE IF NOT EXISTS journal_grants ( + case_id TEXT NOT NULL, + principal TEXT NOT NULL, + source TEXT NOT NULL, + space TEXT DEFAULT '', + item_id INTEGER, + UNIQUE (case_id, principal, source, space, item_id) + ); + CREATE TABLE IF NOT EXISTS journal_meta ( + case_id TEXT PRIMARY KEY, + head_hash TEXT NOT NULL, + created_ts TEXT NOT NULL + ); + """) + self._conn.commit() + + # ---------------------------------------------------------------------- verbs + + def append( + self, + actor: Actor, + case: str, + body: str, + *, + kind: str = "note", + space: Optional[str] = None, + item: Optional[int] = None, + entities: Optional[list[str]] = None, + refs: Optional[list[str]] = None, + taint: bool = False, + ) -> dict[str, Any]: + 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})") + if len(body) > JOURNAL_BODY_LIMIT: + raise BoardError( + f"entry body over {JOURNAL_BODY_LIMIT} chars — store the full" + " payload as an artifact and journal an excerpt with a" + " sha256-qualified ref" + ) + with self._lock: + exists = self._case_exists(case) + if exists: + self._check_access(actor, case) + ts = datetime.now(timezone.utc).isoformat() + prev = self._head_hash(case) + record = { + "ts": ts, + "case_id": case, + "kind": kind, + "actor": actor.id, + "actor_role": actor.role.value, + "space": space, + "item_id": item, + "payload": _canonical( + { + "body": body, + "entities": sorted(set(entities or [])), + "refs": [str(ref) for ref in refs or []], + } + ), + "taint": 1 if taint else 0, + "prev_hash": prev, + } + record["hash"] = _hash(record, fields=_HASHED_FIELDS) + try: + cursor = self._conn.execute( + """ + INSERT INTO journal_entries + (ts, case_id, kind, actor, actor_role, persona, model, + session_id, space, item_id, payload, taint, prev_hash, hash) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + ts, + case, + kind, + actor.id, + actor.role.value, + actor.persona, + actor.model, + actor.session_id, + space, + item, + record["payload"], + record["taint"], + prev, + record["hash"], + ), + ) + if not exists: + self._conn.execute( + "INSERT INTO journal_meta (case_id, head_hash, created_ts)" + " VALUES (?, ?, ?)", + (case, record["hash"], ts), + ) + # A new case belongs to whoever opened it. + self._grant_locked(case, actor.id, source="creator") + else: + self._conn.execute( + "UPDATE journal_meta SET head_hash = ? WHERE case_id = ?", + (record["hash"], case), + ) + self._conn.commit() + except Exception: + self._conn.rollback() + raise + return {**record, "seq": cursor.lastrowid} + + def read( + self, + actor: Actor, + case: str, + *, + item: Optional[int] = None, + author: Optional[str] = None, + kind: Optional[str] = None, + entity: Optional[str] = None, + since_seq: int = 0, + include_raw: bool = False, + limit: int = 100, + ) -> list[dict[str, Any]]: + """Filtered read. `raw` captures are skipped unless asked for (by + `kind="raw"` or `include_raw`) so dumps never bury the signal entries.""" + with self._lock: + self._check_access(actor, case) + where = ["case_id = ?", "seq > ?"] + params: list[Any] = [case, since_seq] + if item is not None: + where.append("item_id = ?") + params.append(item) + if author: + where.append("actor = ?") + params.append(author) + if kind: + if kind not in JOURNAL_KINDS: + raise BoardError(f"unknown entry kind: {kind}") + where.append("kind = ?") + params.append(kind) + elif not include_raw: + where.append("kind != 'raw'") + rows = self._conn.execute( + "SELECT * FROM journal_entries WHERE " + + " AND ".join(where) + + " ORDER BY seq", + params, + ).fetchall() + out = [] + cap = max(1, min(int(limit or 100), 1000)) + for row in rows: + entry = _row_to_entry(row) + if entity and entity not in entry["entities"]: + continue + out.append(entry) + if len(out) >= cap: + break + return out + + def cases(self, actor: Actor) -> list[str]: + """Cases visible to this actor (all of them for the user).""" + with self._lock: + if actor.role == Role.USER: + rows = self._conn.execute( + "SELECT case_id FROM journal_meta ORDER BY case_id" + ).fetchall() + else: + rows = self._conn.execute( + "SELECT DISTINCT case_id FROM journal_grants WHERE principal = ?" + " ORDER BY case_id", + (actor.id,), + ).fetchall() + return [row["case_id"] for row in rows] + + # ---------------------------------------------------------------------- grants + + def grant(self, actor: Actor, case: str, principal: str) -> None: + """Explicit cross-team sharing. The user may grant any case; a lead may + grant cases it holds. Workers never grant — evidence flows up, access + flows down.""" + if actor.role == Role.WORKER or actor.role == Role.SYSTEM: + raise AuthorityError("only the user or a lead may grant a case") + with self._lock: + if not self._case_exists(case): + raise BoardError(f"no case '{case}'") + if actor.role == Role.LEAD: + self._check_access(actor, case) + self._grant_locked(case, principal, source="grant") + self._conn.commit() + + def revoke(self, actor: Actor, case: str, principal: str) -> None: + if actor.role == Role.WORKER or actor.role == Role.SYSTEM: + raise AuthorityError("only the user or a lead may revoke a case grant") + with self._lock: + if actor.role == Role.LEAD: + self._check_access(actor, case) + self._conn.execute( + "DELETE FROM journal_grants WHERE case_id = ? AND principal = ?" + " AND source = 'grant'", + (case, principal), + ) + self._conn.commit() + + def ensure_case(self, case: str, creator: str) -> None: + """Create a case (empty, chain at genesis) if it doesn't exist, granting + its creator. Called by the board when an item attaches a case ref — so + the case belongs to whoever attached it, not to whichever assignee + happens to journal first. Standalone cases (no board) are still created + by their first append.""" + if not (case or "").strip(): + return + with self._lock: + if not self._case_exists(case): + self._conn.execute( + "INSERT INTO journal_meta (case_id, head_hash, created_ts)" + " VALUES (?, ?, ?)", + (case, GENESIS, datetime.now(timezone.utc).isoformat()), + ) + self._grant_locked(case, creator, source="creator") + self._conn.commit() + + def sync_assignment( + self, + case: str, + *, + space: str, + item_id: int, + assignee: str, + previous: str = "", + ) -> None: + """Called by the board on assign: access rides assignment. The previous + assignee loses the grant THIS item carried (grants from its other items + or explicit shares survive).""" + if not case: + return + with self._lock: + if previous: + self._conn.execute( + "DELETE FROM journal_grants WHERE case_id = ? AND principal = ?" + " AND source = 'assignment' AND space = ? AND item_id = ?", + (case, previous, space, item_id), + ) + self._grant_locked( + case, assignee, source="assignment", space=space, item_id=item_id + ) + self._conn.commit() + + # ----------------------------------------------------------------- integrity + + def verify_chain(self, case: str) -> int: + with self._lock: + rows = self._conn.execute( + "SELECT * FROM journal_entries WHERE case_id = ? ORDER BY seq", + (case,), + ).fetchall() + prev = GENESIS + for row in rows: + record = {key: row[key] for key in _HASHED_FIELDS} + if row["prev_hash"] != prev: + raise ChainError(f"entry {row['seq']}: chain linkage broken") + if _hash(record, fields=_HASHED_FIELDS) != row["hash"]: + raise ChainError(f"entry {row['seq']}: content does not match hash") + prev = row["hash"] + return len(rows) + + def close(self) -> None: + self._conn.close() + + # ------------------------------------------------------------------ internals + + def _check_access(self, actor: Actor, case: str) -> None: + if actor.role == Role.USER: + return + row = self._conn.execute( + "SELECT 1 FROM journal_grants WHERE case_id = ? AND principal = ?" + " LIMIT 1", + (case, actor.id), + ).fetchone() + if row is None: + raise AuthorityError(f"{actor.id} has no grant on case '{case}'") + + def _grant_locked( + self, + case: str, + principal: str, + *, + source: str, + space: str = "", + item_id: Optional[int] = None, + ) -> None: + self._conn.execute( + "INSERT OR IGNORE INTO journal_grants" + " (case_id, principal, source, space, item_id) VALUES (?, ?, ?, ?, ?)", + (case, principal, source, space, item_id), + ) + + def _case_exists(self, case: str) -> bool: + return ( + self._conn.execute( + "SELECT 1 FROM journal_meta WHERE case_id = ?", (case,) + ).fetchone() + is not None + ) + + def _head_hash(self, case: str) -> str: + row = self._conn.execute( + "SELECT head_hash FROM journal_meta WHERE case_id = ?", (case,) + ).fetchone() + return row["head_hash"] if row else GENESIS + + +def _row_to_entry(row: sqlite3.Row) -> dict[str, Any]: + entry = dict(row) + try: + payload = json.loads(entry.pop("payload") or "{}") + except json.JSONDecodeError: + payload = {} + entry["body"] = payload.get("body") + entry["entities"] = payload.get("entities") or [] + entry["refs"] = payload.get("refs") or [] + entry["author"] = entry.pop("actor") + entry["role"] = entry.pop("actor_role") + entry["item"] = entry.pop("item_id") + return entry diff --git a/coworker/teams/model.py b/coworker/teams/model.py index aa09983a..37c057c4 100644 --- a/coworker/teams/model.py +++ b/coworker/teams/model.py @@ -62,7 +62,14 @@ class Actor: 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") +# `note` is any observation — the journal is not only for investigations. `raw` is +# a capture (log excerpt, command output); reads skip raw unless asked, and large +# payloads belong in the artifact store with a sha256-qualified ref on the entry. +JOURNAL_KINDS = ("finding", "evidence", "decision", "note", "raw") + +# An entry body is an excerpt/summary, never a blob: oversized payloads make every +# chain-verify and unfiltered read drag. Full captures go to the artifact store. +JOURNAL_BODY_LIMIT = 16_000 class BoardError(Exception): diff --git a/coworker/teams/store.py b/coworker/teams/store.py index 1678aecc..e5180316 100644 --- a/coworker/teams/store.py +++ b/coworker/teams/store.py @@ -1,10 +1,12 @@ -"""The team event store — ONE append-only log; board, journal, and deliveries are -projections of it. +"""The board event store — an append-only log per space; the board and per-agent +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. +Doctrine (agent-teams design): board events and chat messages are one attributed, +timestamped, immutable record shape in one space-scoped 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. Journal entries share the shape +and discipline but live in their own case-keyed store (teams.journal): cases outlive +boards and teams, so their lifecycle can't be chained to a board's. Mechanics, kept boring: - Append and projection-fold happen in the same transaction via the same `_apply` @@ -29,7 +31,6 @@ from typing import Any, Optional from .model import ( EDGES, - JOURNAL_KINDS, LINK_KINDS, WORKER_TARGETS, Actor, @@ -43,12 +44,13 @@ from .model import ( GENESIS = "genesis" # Event kinds. Chat lands later with the chat surface; the record shape already fits. +# Journal entries live in their own case-keyed store (teams.journal) — cases outlive +# boards, so they don't belong in a board's space-scoped log. 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", @@ -66,7 +68,11 @@ _HASHED_FIELDS = ( class TeamStore: - def __init__(self, db_path: str | Path) -> None: + def __init__(self, db_path: str | Path, *, journal: Any = None) -> None: + # `journal` is a teams.journal.JournalStore when wired: assignment feeds + # case grants ("sharing rides assignment"). Optional so the board works + # standalone (tests, boards with no journal). + self.journal = journal self.db_path = str(db_path) if self.db_path != ":memory:": Path(self.db_path).expanduser().parent.mkdir(parents=True, exist_ok=True) @@ -369,6 +375,8 @@ class TeamStore: "case": case, }, ) + if self.journal is not None and case: + self.journal.ensure_case(case, actor.id) return self.get_item(space, item_id, seq=event["seq"]) def list_items( @@ -507,6 +515,14 @@ class TeamStore: recipient=assignee, payload={"assignee": assignee, "previous": item["assignee"] or ""}, ) + if self.journal is not None and item["case_id"]: + self.journal.sync_assignment( + item["case_id"], + space=space, + item_id=item_id, + assignee=assignee, + previous=item["assignee"] or "", + ) return self.get_item(space, item_id, seq=event["seq"]) def link( @@ -555,112 +571,6 @@ class TeamStore: ) 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() @@ -777,18 +687,6 @@ class TeamStore: 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]: # Assigned items, items the worker filed itself, and items directly # linked to either — its slice of the board, nothing more. @@ -876,8 +774,8 @@ 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}) +def _hash(record: dict[str, Any], *, fields: tuple[str, ...] = _HASHED_FIELDS) -> str: + material = _canonical({key: record[key] for key in fields}) return hashlib.sha256(material.encode("utf-8")).hexdigest() diff --git a/coworker/teams/tools.py b/coworker/teams/tools.py index 2a9476f5..60fcccae 100644 --- a/coworker/teams/tools.py +++ b/coworker/teams/tools.py @@ -15,6 +15,7 @@ from typing import Callable, Optional import aisuite as ai +from .journal import JournalStore from .model import Actor, BoardError, Role from .store import TeamStore @@ -150,10 +151,10 @@ def board_tools( def journal_tools( - store: TeamStore, + journal: "JournalStore", *, - space: str, actor: Actor, + space: str = "", taint: Callable[[], bool] = lambda: False, ) -> list: def journal_append( @@ -165,16 +166,18 @@ def journal_tools( 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).""" + evidence, decision, note (any observation), or raw (a capture — put an + excerpt here and store the full payload as an artifact, referenced with + a sha256-qualified ref). `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, artifact).""" return _call( - store.journal_append, - space, + journal.append, actor, case, body, kind=kind, + space=space or None, item=item, entities=[str(entity) for entity in entities or []], refs=[str(ref) for ref in refs or []], @@ -187,20 +190,22 @@ def journal_tools( author: str = "", kind: str = "", entity: str = "", + include_raw: bool = False, 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.""" + Prefer narrow filtered reads over pulling the whole case. Raw captures + are skipped unless you pass include_raw or kind="raw".""" try: return { - "entries": store.journal_read( - space, + "entries": journal.read( actor, case, item=item, author=author or None, kind=kind or None, entity=entity or None, + include_raw=include_raw, limit=limit, ) } diff --git a/tests/test_team_journal.py b/tests/test_team_journal.py index 1f24b842..e9924902 100644 --- a/tests/test_team_journal.py +++ b/tests/test_team_journal.py @@ -1,110 +1,179 @@ -"""Journal cases: filtered reads, access-rides-assignment, lazy case lifecycle.""" +"""Journal store: case-keyed and board-independent, grants ride assignment, +filtered reads, raw-capture discipline, per-case hash chains.""" import pytest -from coworker.teams import Actor, AuthorityError, BoardError, Role, TeamStore +from coworker.teams import ( + Actor, + AuthorityError, + BoardError, + ChainError, + JournalStore, + Role, + TeamStore, +) +from coworker.teams.model import JOURNAL_BODY_LIMIT USER = Actor(id="user", role=Role.USER) LEAD = Actor(id="lead-1", role=Role.LEAD) +LEAD2 = Actor(id="lead-2", 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 journal(tmp_path): + journal = JournalStore(tmp_path / "journal.db") + yield journal + journal.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) +@pytest.fixture +def board(tmp_path, journal): + board = TeamStore(tmp_path / "teams.db", journal=journal) + yield board + board.close() + + +def case_item(board, case="findings", assignee="worker-1", space=SPACE): + item = board.create_item(space, LEAD, title="Task", criteria="c", case=case) + board.transition(space, USER, item["id"], "approved") + board.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_case_creation_grants_the_creator(journal): + journal.append(LEAD, "findings", "case opened") + assert journal.cases(LEAD) == ["findings"] + assert journal.cases(OTHER) == [] + with pytest.raises(AuthorityError, match="no grant"): + journal.read(OTHER, "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") +def test_assignment_feeds_grants_and_reassignment_moves_them(board, journal): + item_id = case_item(board, assignee="worker-1") + journal.append(WORKER, "findings", "assignee writes", item=item_id, space=SPACE) + with pytest.raises(AuthorityError): + journal.read(OTHER, "findings") + board.assign(SPACE, LEAD, item_id, "worker-2") + journal.append(OTHER, "findings", "successor picks up the case") + with pytest.raises(AuthorityError, match="no grant"): + journal.append(WORKER, "findings", "predecessor lost access") - 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") + +def test_grants_from_other_items_survive_reassignment(board, journal): + first = case_item(board, assignee="worker-1") + second = case_item(board, assignee="worker-1") + board.assign(SPACE, LEAD, first, "worker-2") + # worker-1 still holds the case through its second item + journal.append(WORKER, "findings", "still on the case", item=second, space=SPACE) + + +def test_cases_span_boards_and_teams(board, journal): + case_item(board, case="ops-incident", space="alpha") + case_item(board, case="ops-incident", space="beta", assignee="worker-1") + journal.append(WORKER, "ops-incident", "one case, two boards") + entries = journal.read(USER, "ops-incident") + assert len(entries) == 1 + # and a case with NO board at all is fine — an Ops scratch investigation + journal.append(USER, "loose-threads", "observation with no board") + assert "loose-threads" in journal.cases(USER) + + +def test_explicit_grant_shares_across_teams(journal): + journal.append(LEAD, "findings", "opened by lead-1") + with pytest.raises(AuthorityError): + journal.read(LEAD2, "findings") + journal.grant(LEAD, "findings", "lead-2") + assert journal.read(LEAD2, "findings")[0]["body"] == "opened by lead-1" + journal.revoke(LEAD, "findings", "lead-2") + with pytest.raises(AuthorityError): + journal.read(LEAD2, "findings") + + +def test_workers_never_grant(journal, board): + case_item(board) + with pytest.raises(AuthorityError, match="lead"): + journal.grant(WORKER, "findings", "worker-2") + + +def test_a_lead_cannot_grant_a_case_it_does_not_hold(journal): + journal.append(LEAD, "findings", "lead-1's case") + with pytest.raises(AuthorityError, match="no grant"): + journal.grant(LEAD2, "findings", "worker-2") + + +def test_filtered_reads(journal): + journal.append( + LEAD, "findings", "logos bucket is world-readable", kind="finding", + entities=["aws_s3_bucket.assets", "uploads.ts"], refs=["services/uploads.ts:41"], + ) + journal.append( + LEAD, "findings", "invoice PDFs stream from the API", kind="evidence", + entities=["uploads.ts"], + ) + journal.grant(LEAD, "findings", "worker-1") + journal.append(WORKER, "findings", "narrowing fix to logos/*", kind="decision") + + assert len(journal.read(LEAD, "findings")) == 3 + assert [e["kind"] for e in journal.read(LEAD, "findings", kind="finding")] == ["finding"] + assert len(journal.read(LEAD, "findings", author="worker-1")) == 1 + by_entity = journal.read(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 + assert len(journal.read(LEAD, "findings", entity="uploads.ts")) == 2 + assert len(journal.read(LEAD, "findings", limit=2)) == 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, +def test_raw_captures_are_opt_in_on_read(journal): + journal.append(LEAD, "ops", "deploy finished 14:01", kind="note") + journal.append( + LEAD, "ops", "nginx 5xx burst 14:02-14:04 (2,400 lines)", kind="raw", + refs=["artifact:sha256:ab12...:nginx-error.log"], ) - entry = store.journal_read(SPACE, LEAD, "findings")[0] + assert len(journal.read(LEAD, "ops")) == 1 # raw skipped by default + assert len(journal.read(LEAD, "ops", include_raw=True)) == 2 + assert journal.read(LEAD, "ops", kind="raw")[0]["refs"][0].startswith("artifact:") + + +def test_oversized_bodies_are_refused_with_the_artifact_pattern(journal): + with pytest.raises(BoardError, match="artifact"): + journal.append(LEAD, "ops", "x" * (JOURNAL_BODY_LIMIT + 1), kind="raw") + + +def test_entry_validation(journal): + with pytest.raises(BoardError, match="kind"): + journal.append(LEAD, "findings", "x", kind="rant") + with pytest.raises(BoardError, match="body"): + journal.append(LEAD, "findings", " ") + with pytest.raises(BoardError, match="case"): + journal.append(LEAD, "", "x") + + +def test_taint_and_attribution_survive_the_read(journal): + journal.append( + WORKER, "findings", "repo README claims the bucket must be public", + kind="evidence", taint=True, + ) + entry = journal.read(USER, "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 +def test_per_case_hash_chains_verify_and_detect_tampering(journal, tmp_path): + import sqlite3 + + journal.append(LEAD, "findings", "one") + journal.append(LEAD, "findings", "two") + journal.append(LEAD, "ops", "unrelated case") + assert journal.verify_chain("findings") == 2 + assert journal.verify_chain("ops") == 1 + conn = sqlite3.connect(journal.db_path) + conn.execute("UPDATE journal_entries SET payload = '{\"body\":\"forged\"}' WHERE seq = 1") + conn.commit() + conn.close() + with pytest.raises(ChainError): + journal.verify_chain("findings") + assert journal.verify_chain("ops") == 1 # other case unaffected