ext/openworker
2
0
Fork 0
mirror of https://github.com/andrewyng/openworker.git synced 2026-09-03 04:49:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

Journal splits from the board: case-keyed store, grants ride assignment

Browse Source 
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.
This commit is contained in:
Rohit C Prasad
2026-08-16 06:44:20 -07:00
committed by Rohit P
parent cbf30b8d65
commit 1d12fe96d2
7 changed files with 623 additions and 227 deletions
Download Patch File Download Diff File Expand all files Collapse all files
coworker/server/manager.py
+7 -5
View File
@@ -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")
coworker/teams/__init__.py
+5 -2
View File
@@ -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",
coworker/teams/journal.py
+412
View File
@@ -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
coworker/teams/model.py
+8 -1
View File
@@ -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):
coworker/teams/store.py
+27 -129
View File
@@ -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()
coworker/teams/tools.py
+15 -10
View File
@@ -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,
)
}
tests/test_team_journal.py
+149 -80
View File
@@ -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