Board as an open surface: BoardDialect seam, join tokens, /v1/board API

Local (direct stores) and remote (one wire protocol) dialects; trackers become mirrors later, never dialects.
Tokens bind actor+role server-side (sha256-stored); board routes carry their own auth, external writes kick the wake tick.

coworker/server/app.py

@@ -162,6 +162,8 @@ from ..inbox import VIS_INBOX, VIS_INLINE, args_preview
 from ..permissions import Mode
 from ..providers import AssistantTurn
 from .. import toolchain
+from ..teams.model import AuthorityError as TeamsAuthorityError
+from ..teams.model import BoardError as TeamsBoardError
 from .manager import SessionManager
 
 
@@ -216,6 +218,10 @@ def create_app(manager: SessionManager) -> FastAPI:
             not api_token
             or request.method == "OPTIONS"
             or request.url.path in tokenless_paths
+            # `/v1/board` carries its own, stronger auth: per-actor board tokens
+            # (identity + access), designed to be handed to external harnesses and
+            # other machines — which can never hold the machine-local sidecar token.
+            or request.url.path.startswith("/v1/board/")
             or _request_authenticated(request)
         ):
             return await call_next(request)
@@ -762,6 +768,242 @@ def create_app(manager: SessionManager) -> FastAPI:
     def teams_journal() -> dict[str, Any]:
         return {"cases": manager.journal_overview()}
 
+    # ---- The open board surface (OPE-100): token-authenticated `/v1/board` API.
+    # Identity is the TOKEN (actor+role bound at mint, resolved per request, never
+    # client-asserted); authority is the STORE — the same double gate in-app agents
+    # get. This is the one wire protocol every external front door rides:
+    # RemoteDialect (the `ocw` CLI, the team-board MCP server, headless instances)
+    # today, a hosted board service later. Tokens are required even on loopback —
+    # they carry identity, not just access.
+
+    def _board_actor(request: Request):
+        auth = request.headers.get("authorization", "")
+        token = auth[7:] if auth.lower().startswith("bearer ") else ""
+        return manager.board_tokens.resolve(token)
+
+    def _board(request: Request, handler):
+        actor = _board_actor(request)
+        if actor is None:
+            return JSONResponse(
+                {"error": "board token required (Authorization: Bearer …) — mint"
+                          " one with `ocw board token` on the serving machine"},
+                status_code=401,
+            )
+        try:
+            return handler(actor)
+        except TeamsAuthorityError as error:
+            return JSONResponse({"error": str(error)}, status_code=403)
+        except (TeamsBoardError, ValueError) as error:
+            return JSONResponse({"error": str(error)}, status_code=400)
+
+    @app.get("/v1/board/whoami")
+    def board_whoami(request: Request):
+        return _board(
+            request, lambda actor: {"actor": actor.id, "role": actor.role.value}
+        )
+
+    @app.get("/v1/board/spaces")
+    def board_spaces(request: Request):
+        return _board(request, lambda actor: {"spaces": manager.team_store.spaces()})
+
+    @app.get("/v1/board/items")
+    def board_list_items(
+        request: Request, space: str, state: str = "", assignee: str = ""
+    ):
+        return _board(
+            request,
+            lambda actor: {
+                "items": manager.team_store.list_items(
+                    space, actor, state=state or None, assignee=assignee or None
+                )
+            },
+        )
+
+    @app.get("/v1/board/item")
+    def board_get_item(request: Request, space: str, id: int):
+        return _board(
+            request, lambda actor: manager.team_store.get_item(space, int(id))
+        )
+
+    @app.post("/v1/board/items")
+    def board_create_item(request: Request, body: dict):
+        body = body or {}
+
+        def run(actor):
+            item = manager.team_store.create_item(
+                str(body.get("space", "")),
+                actor,
+                title=str(body.get("title", "")),
+                criteria=str(body.get("criteria", "")),
+                description=str(body.get("description", "")),
+                parent=(
+                    int(body["parent"]) if body.get("parent") is not None else None
+                ),
+                case=str(body.get("case") or "") or None,
+            )
+            manager.kick_team_tick()  # a new filing is lead-subscription news
+            return item
+
+        return _board(request, run)
+
+    @app.post("/v1/board/items/transition")
+    def board_transition_item(request: Request, body: dict):
+        body = body or {}
+
+        def run(actor):
+            item = manager.team_store.transition(
+                str(body.get("space", "")),
+                actor,
+                int(body.get("id", 0)),
+                str(body.get("to", "")),
+                comment=str(body.get("comment", "")),
+                refs=[str(ref) for ref in body.get("refs") or []],
+            )
+            manager.kick_team_tick()  # review/blocked should reach the lead now
+            return item
+
+        return _board(request, run)
+
+    @app.post("/v1/board/items/comment")
+    def board_comment_item(request: Request, body: dict):
+        body = body or {}
+        return _board(
+            request,
+            lambda actor: manager.team_store.comment(
+                str(body.get("space", "")),
+                actor,
+                int(body.get("id", 0)),
+                str(body.get("body", "")),
+                refs=[str(ref) for ref in body.get("refs") or []],
+            ),
+        )
+
+    @app.post("/v1/board/items/assign")
+    def board_assign_item(request: Request, body: dict):
+        body = body or {}
+
+        def run(actor):
+            item = manager.team_store.assign(
+                str(body.get("space", "")),
+                actor,
+                int(body.get("id", 0)),
+                str(body.get("assignee", "")),
+            )
+            manager.kick_team_tick()  # the assignee's queue has news
+            return item
+
+        return _board(request, run)
+
+    @app.post("/v1/board/items/claim")
+    def board_claim_item(request: Request, body: dict):
+        body = body or {}
+
+        def run(actor):
+            item = manager.team_store.claim(
+                str(body.get("space", "")), actor, int(body.get("id", 0))
+            )
+            manager.kick_team_tick()  # claims land in the lead's feed
+            return item
+
+        return _board(request, run)
+
+    @app.post("/v1/board/link")
+    def board_link_items(request: Request, body: dict):
+        body = body or {}
+        return _board(
+            request,
+            lambda actor: manager.team_store.link(
+                str(body.get("space", "")),
+                actor,
+                int(body.get("src", 0)),
+                str(body.get("kind", "")),
+                int(body.get("dst", 0)),
+            ),
+        )
+
+    @app.get("/v1/board/policy")
+    def board_get_policy(request: Request, space: str):
+        return _board(request, lambda actor: manager.team_store.policy(space))
+
+    @app.post("/v1/board/policy")
+    def board_set_policy(request: Request, body: dict):
+        body = body or {}
+        return _board(
+            request,
+            lambda actor: manager.team_store.set_policy(
+                str(body.get("space", "")), actor, claims=str(body.get("claims", ""))
+            ),
+        )
+
+    @app.get("/v1/board/pending")
+    def board_pending(request: Request, limit: int = 200):
+        return _board(
+            request,
+            lambda actor: {
+                "events": manager.team_store.pending_for(actor.id, limit=int(limit))
+            },
+        )
+
+    @app.post("/v1/board/consume")
+    def board_consume(request: Request, body: dict):
+        body = body or {}
+
+        def run(actor):
+            manager.team_store.consume(actor.id, int(body.get("upto_seq", 0)))
+            return {"ok": True}
+
+        return _board(request, run)
+
+    @app.get("/v1/board/journal/cases")
+    def board_journal_cases(request: Request):
+        return _board(
+            request, lambda actor: {"cases": manager.journal_store.overview(actor)}
+        )
+
+    @app.get("/v1/board/journal")
+    def board_journal_read(
+        request: Request,
+        case: str,
+        item: Optional[int] = None,
+        author: str = "",
+        kind: str = "",
+        entity: str = "",
+        include_raw: str = "",
+        limit: int = 100,
+    ):
+        return _board(
+            request,
+            lambda actor: {
+                "entries": manager.journal_store.read(
+                    actor,
+                    case,
+                    item=item,
+                    author=author or None,
+                    kind=kind or None,
+                    entity=entity or None,
+                    include_raw=bool(include_raw),
+                    limit=int(limit),
+                )
+            },
+        )
+
+    @app.post("/v1/board/journal")
+    def board_journal_append(request: Request, body: dict):
+        body = body or {}
+        return _board(
+            request,
+            lambda actor: manager.journal_store.append(
+                actor,
+                str(body.get("case", "")),
+                str(body.get("body", "")),
+                kind=str(body.get("kind") or "note"),
+                space=str(body.get("space") or "") or None,
+                item=int(body["item"]) if body.get("item") is not None else None,
+                entities=[str(e) for e in body.get("entities") or []],
+                refs=[str(ref) for ref in body.get("refs") or []],
+            ),
+        )
+
     @app.get("/v1/memory")
     def memory() -> dict[str, Any]:
         return {"memory": manager.list_memory()}

coworker/teams/__init__.py

@@ -26,3 +26,7 @@ __all__ = [
     "board_tools",
     "journal_tools",
 ]
+
+# BoardDialect / LocalDialect / RemoteDialect live in .dialect, BoardTokens in
+# .tokens — imported directly by their consumers (CLI, MCP server, `/v1/board`)
+# to keep this package root light for the common in-app path.

coworker/teams/dialect.py

@@ -0,0 +1,474 @@
+"""The BoardDialect seam — where "a board" stops meaning "our SQLite file".
+
+A dialect is where the board of record LIVES, seen from a client's chair:
+- LocalDialect: this machine's TeamStore/JournalStore, direct SQLite. For the
+  standalone/headless case where the caller is the only writer.
+- RemoteDialect: one wire protocol (the `/v1/board` HTTP API) to a board served
+  elsewhere — the running OpenWorker sidecar on this machine, a teammate's machine,
+  or a hosted board service later. Identity rides the token; the server binds it to
+  an actor+role and the store enforces authority, so a remote client is safe by
+  construction.
+
+External trackers (Jira/Linear) are deliberately NOT dialects: making a pre-LLM
+tracker the board of record means contorting our state machine and delivery cursors
+onto its API. They join as MIRRORS instead — one more subscriber with a cursor over
+the append-only event log, replaying events outward (decided 2026-08-16). The board
+stays the abstraction and the source of truth.
+
+Every front door — the `team-board` MCP server, the `ocw` CLI, remote OpenWorker
+instances — bottoms out in this one verb surface. Dialect instances are
+identity-bound: one actor per instance, matching the one-identity-per-process shape
+of an external harness.
+
+Cross-process write safety: the store's hash-chain append is read-head-then-write
+under an in-process lock, so two processes must never write one SQLite file
+directly. Rule: when a server is up, clients go remote; LocalDialect is for the
+headless case where this process is the only writer.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Optional, Protocol
+
+from .journal import JournalStore
+from .model import Actor, BoardError, Role
+from .store import TeamStore
+
+
+class BoardDialect(Protocol):
+    """The verb surface a board client sees, identity already bound."""
+
+    def whoami(self) -> dict[str, Any]: ...
+    def spaces(self) -> list[str]: ...
+    def list_items(
+        self,
+        space: str,
+        *,
+        state: Optional[str] = None,
+        assignee: Optional[str] = None,
+    ) -> list[dict[str, Any]]: ...
+    def get_item(self, space: str, item_id: int) -> dict[str, Any]: ...
+    def create_item(
+        self,
+        space: str,
+        *,
+        title: str,
+        criteria: str,
+        description: str = "",
+        parent: Optional[int] = None,
+        case: Optional[str] = None,
+    ) -> dict[str, Any]: ...
+    def transition(
+        self,
+        space: str,
+        item_id: int,
+        to: str,
+        *,
+        comment: str = "",
+        refs: Optional[list[str]] = None,
+    ) -> dict[str, Any]: ...
+    def comment(
+        self,
+        space: str,
+        item_id: int,
+        body: str,
+        *,
+        refs: Optional[list[str]] = None,
+    ) -> dict[str, Any]: ...
+    def assign(self, space: str, item_id: int, assignee: str) -> dict[str, Any]: ...
+    def claim(self, space: str, item_id: int) -> dict[str, Any]: ...
+    def link(self, space: str, src: int, kind: str, dst: int) -> dict[str, Any]: ...
+    def policy(self, space: str) -> dict[str, Any]: ...
+    def set_policy(self, space: str, *, claims: str) -> dict[str, Any]: ...
+    def pending(self, *, limit: int = 200) -> list[dict[str, Any]]: ...
+    def consume(self, upto_seq: int) -> None: ...
+    def journal_append(
+        self,
+        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,
+    ) -> dict[str, Any]: ...
+    def journal_read(
+        self,
+        case: str,
+        *,
+        item: Optional[int] = None,
+        author: Optional[str] = None,
+        kind: Optional[str] = None,
+        entity: Optional[str] = None,
+        include_raw: bool = False,
+        limit: int = 100,
+    ) -> list[dict[str, Any]]: ...
+    def journal_overview(self) -> list[dict[str, Any]]: ...
+
+
+class LocalDialect:
+    """Direct store access, one bound identity. The headless/standalone backing."""
+
+    def __init__(
+        self, store: TeamStore, journal: Optional[JournalStore], actor: Actor
+    ) -> None:
+        self.store = store
+        self.journal = journal
+        self.actor = actor
+
+    def whoami(self) -> dict[str, Any]:
+        return {"actor": self.actor.id, "role": self.actor.role.value}
+
+    def spaces(self) -> list[str]:
+        return self.store.spaces()
+
+    def list_items(
+        self,
+        space: str,
+        *,
+        state: Optional[str] = None,
+        assignee: Optional[str] = None,
+    ) -> list[dict[str, Any]]:
+        return self.store.list_items(space, self.actor, state=state, assignee=assignee)
+
+    def get_item(self, space: str, item_id: int) -> dict[str, Any]:
+        return self.store.get_item(space, item_id)
+
+    def create_item(
+        self,
+        space: str,
+        *,
+        title: str,
+        criteria: str,
+        description: str = "",
+        parent: Optional[int] = None,
+        case: Optional[str] = None,
+    ) -> dict[str, Any]:
+        return self.store.create_item(
+            space,
+            self.actor,
+            title=title,
+            criteria=criteria,
+            description=description,
+            parent=parent,
+            case=case,
+        )
+
+    def transition(
+        self,
+        space: str,
+        item_id: int,
+        to: str,
+        *,
+        comment: str = "",
+        refs: Optional[list[str]] = None,
+    ) -> dict[str, Any]:
+        return self.store.transition(
+            space, self.actor, item_id, to, comment=comment, refs=refs
+        )
+
+    def comment(
+        self,
+        space: str,
+        item_id: int,
+        body: str,
+        *,
+        refs: Optional[list[str]] = None,
+    ) -> dict[str, Any]:
+        return self.store.comment(space, self.actor, item_id, body, refs=refs)
+
+    def assign(self, space: str, item_id: int, assignee: str) -> dict[str, Any]:
+        return self.store.assign(space, self.actor, item_id, assignee)
+
+    def claim(self, space: str, item_id: int) -> dict[str, Any]:
+        return self.store.claim(space, self.actor, item_id)
+
+    def link(self, space: str, src: int, kind: str, dst: int) -> dict[str, Any]:
+        return self.store.link(space, self.actor, src, kind, dst)
+
+    def policy(self, space: str) -> dict[str, Any]:
+        return self.store.policy(space)
+
+    def set_policy(self, space: str, *, claims: str) -> dict[str, Any]:
+        return self.store.set_policy(space, self.actor, claims=claims)
+
+    def pending(self, *, limit: int = 200) -> list[dict[str, Any]]:
+        return self.store.pending_for(self.actor.id, limit=limit)
+
+    def consume(self, upto_seq: int) -> None:
+        self.store.consume(self.actor.id, int(upto_seq))
+
+    def journal_append(
+        self,
+        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,
+    ) -> dict[str, Any]:
+        self._need_journal()
+        return self.journal.append(
+            self.actor,
+            case,
+            body,
+            kind=kind,
+            space=space,
+            item=item,
+            entities=entities,
+            refs=refs,
+        )
+
+    def journal_read(
+        self,
+        case: str,
+        *,
+        item: Optional[int] = None,
+        author: Optional[str] = None,
+        kind: Optional[str] = None,
+        entity: Optional[str] = None,
+        include_raw: bool = False,
+        limit: int = 100,
+    ) -> list[dict[str, Any]]:
+        self._need_journal()
+        return self.journal.read(
+            self.actor,
+            case,
+            item=item,
+            author=author,
+            kind=kind,
+            entity=entity,
+            include_raw=include_raw,
+            limit=limit,
+        )
+
+    def journal_overview(self) -> list[dict[str, Any]]:
+        self._need_journal()
+        return self.journal.overview(self.actor)
+
+    def _need_journal(self) -> None:
+        if self.journal is None:
+            raise BoardError("no journal store is attached to this board")
+
+
+class RemoteDialect:
+    """The `/v1/board` HTTP client. `base_url` is an OpenWorker sidecar or a hosted
+    board service; the Bearer token carries identity — the server resolves it to an
+    actor+role, so this client never states who it is, it proves it."""
+
+    def __init__(
+        self, base_url: str, token: str, *, client: Any = None, timeout: float = 30.0
+    ) -> None:
+        import httpx
+
+        self.base_url = base_url.rstrip("/")
+        self._client = client or httpx.Client(
+            base_url=self.base_url,
+            timeout=timeout,
+        )
+        self._client.headers["Authorization"] = f"Bearer {token}"
+
+    # -- plumbing --------------------------------------------------------------
+
+    def _get(self, path: str, params: Optional[dict] = None) -> Any:
+        response = self._client.get(
+            path, params={k: v for k, v in (params or {}).items() if v is not None}
+        )
+        return self._unwrap(response)
+
+    def _post(self, path: str, body: dict) -> Any:
+        response = self._client.post(
+            path, json={k: v for k, v in body.items() if v is not None}
+        )
+        return self._unwrap(response)
+
+    @staticmethod
+    def _unwrap(response: Any) -> Any:
+        if response.status_code == 401:
+            raise BoardError("board token was not accepted (401) — mint one with"
+                             " `ocw board token` on the serving machine")
+        try:
+            data = response.json()
+        except ValueError:
+            data = {}
+        if response.status_code >= 400:
+            raise BoardError(
+                str(data.get("error") or data.get("detail") or response.text)
+            )
+        return data
+
+    # -- verbs -----------------------------------------------------------------
+
+    def whoami(self) -> dict[str, Any]:
+        return self._get("/v1/board/whoami")
+
+    def spaces(self) -> list[str]:
+        return self._get("/v1/board/spaces")["spaces"]
+
+    def list_items(
+        self,
+        space: str,
+        *,
+        state: Optional[str] = None,
+        assignee: Optional[str] = None,
+    ) -> list[dict[str, Any]]:
+        return self._get(
+            "/v1/board/items",
+            {"space": space, "state": state, "assignee": assignee},
+        )["items"]
+
+    def get_item(self, space: str, item_id: int) -> dict[str, Any]:
+        return self._get("/v1/board/item", {"space": space, "id": item_id})
+
+    def create_item(
+        self,
+        space: str,
+        *,
+        title: str,
+        criteria: str,
+        description: str = "",
+        parent: Optional[int] = None,
+        case: Optional[str] = None,
+    ) -> dict[str, Any]:
+        return self._post(
+            "/v1/board/items",
+            {
+                "space": space,
+                "title": title,
+                "criteria": criteria,
+                "description": description,
+                "parent": parent,
+                "case": case,
+            },
+        )
+
+    def transition(
+        self,
+        space: str,
+        item_id: int,
+        to: str,
+        *,
+        comment: str = "",
+        refs: Optional[list[str]] = None,
+    ) -> dict[str, Any]:
+        return self._post(
+            "/v1/board/items/transition",
+            {
+                "space": space,
+                "id": item_id,
+                "to": to,
+                "comment": comment,
+                "refs": refs or [],
+            },
+        )
+
+    def comment(
+        self,
+        space: str,
+        item_id: int,
+        body: str,
+        *,
+        refs: Optional[list[str]] = None,
+    ) -> dict[str, Any]:
+        return self._post(
+            "/v1/board/items/comment",
+            {"space": space, "id": item_id, "body": body, "refs": refs or []},
+        )
+
+    def assign(self, space: str, item_id: int, assignee: str) -> dict[str, Any]:
+        return self._post(
+            "/v1/board/items/assign",
+            {"space": space, "id": item_id, "assignee": assignee},
+        )
+
+    def claim(self, space: str, item_id: int) -> dict[str, Any]:
+        return self._post("/v1/board/items/claim", {"space": space, "id": item_id})
+
+    def link(self, space: str, src: int, kind: str, dst: int) -> dict[str, Any]:
+        return self._post(
+            "/v1/board/link", {"space": space, "src": src, "kind": kind, "dst": dst}
+        )
+
+    def policy(self, space: str) -> dict[str, Any]:
+        return self._get("/v1/board/policy", {"space": space})
+
+    def set_policy(self, space: str, *, claims: str) -> dict[str, Any]:
+        return self._post("/v1/board/policy", {"space": space, "claims": claims})
+
+    def pending(self, *, limit: int = 200) -> list[dict[str, Any]]:
+        return self._get("/v1/board/pending", {"limit": limit})["events"]
+
+    def consume(self, upto_seq: int) -> None:
+        self._post("/v1/board/consume", {"upto_seq": int(upto_seq)})
+
+    def journal_append(
+        self,
+        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,
+    ) -> dict[str, Any]:
+        return self._post(
+            "/v1/board/journal",
+            {
+                "case": case,
+                "body": body,
+                "kind": kind,
+                "space": space,
+                "item": item,
+                "entities": entities or [],
+                "refs": refs or [],
+            },
+        )
+
+    def journal_read(
+        self,
+        case: str,
+        *,
+        item: Optional[int] = None,
+        author: Optional[str] = None,
+        kind: Optional[str] = None,
+        entity: Optional[str] = None,
+        include_raw: bool = False,
+        limit: int = 100,
+    ) -> list[dict[str, Any]]:
+        return self._get(
+            "/v1/board/journal",
+            {
+                "case": case,
+                "item": item,
+                "author": author,
+                "kind": kind,
+                "entity": entity,
+                "include_raw": "1" if include_raw else None,
+                "limit": limit,
+            },
+        )["entries"]
+
+    def journal_overview(self) -> list[dict[str, Any]]:
+        return self._get("/v1/board/journal/cases")["cases"]
+
+    def close(self) -> None:
+        self._client.close()
+
+
+def local_dialect(
+    db_dir, *, actor: str = "user", role: str = "user"
+) -> LocalDialect:
+    """Open the state dir's stores directly as one bound identity — the headless
+    backing for the CLI and MCP server when no OpenWorker server is running."""
+    from pathlib import Path
+
+    base = Path(db_dir).expanduser()
+    journal = JournalStore(base / "journal.db")
+    store = TeamStore(base / "teams.db", journal=journal)
+    return LocalDialect(
+        store, journal, Actor(id=actor, role=Role(role))
+    )

coworker/teams/tokens.py

@@ -0,0 +1,97 @@
+"""Board join tokens — identity for external board clients.
+
+A token binds an ACTOR and a ROLE server-side: an external harness (another agent
+CLI, a headless OpenWorker, the `ocw` CLI from a second machine) presents the token
+and the server resolves who it is — the client never states its own identity, and a
+worker token cannot claim to be the lead. Authority then falls to the store, same
+as for in-app agents: the token is identity, the store is the gate.
+
+Storage is hash-only (sha256): the plaintext is shown once at mint and never
+persisted, so the registry file leaking doesn't leak the credentials. Revocation is
+per-token, keyed by the display prefix.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import secrets
+import threading
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Optional
+
+from .model import Actor, Role
+
+_TOKEN_PREFIX = "owb_"  # OpenWorker board — greppable in configs, meaningless to guess
+
+
+class BoardTokens:
+    def __init__(self, path: str | Path) -> None:
+        self.path = Path(path).expanduser()
+        self._lock = threading.Lock()
+
+    def mint(self, actor: str, role: str = "worker", *, label: str = "") -> str:
+        """Create a token for one actor identity; returns the plaintext ONCE."""
+        actor = (actor or "").strip()
+        if not actor:
+            raise ValueError("actor is required")
+        Role(role)  # validate early — a bad role should fail at mint, not at use
+        token = _TOKEN_PREFIX + secrets.token_urlsafe(32)
+        with self._lock:
+            entries = self._load()
+            entries[_digest(token)] = {
+                "actor": actor,
+                "role": role,
+                "label": label,
+                "prefix": token[:12],
+                "created_ts": datetime.now(timezone.utc).isoformat(),
+            }
+            self._save(entries)
+        return token
+
+    def resolve(self, token: str) -> Optional[Actor]:
+        if not token:
+            return None
+        with self._lock:
+            entry = self._load().get(_digest(token))
+        if entry is None:
+            return None
+        return Actor(id=entry["actor"], role=Role(entry["role"]))
+
+    def entries(self) -> list[dict[str, Any]]:
+        with self._lock:
+            return sorted(self._load().values(), key=lambda e: e["created_ts"])
+
+    def revoke(self, prefix: str) -> int:
+        """Revoke every token whose display prefix matches; returns the count."""
+        prefix = (prefix or "").strip()
+        if not prefix:
+            return 0
+        with self._lock:
+            entries = self._load()
+            keep = {
+                key: entry
+                for key, entry in entries.items()
+                if not entry["prefix"].startswith(prefix)
+            }
+            removed = len(entries) - len(keep)
+            if removed:
+                self._save(keep)
+        return removed
+
+    def _load(self) -> dict[str, dict[str, Any]]:
+        try:
+            return json.loads(self.path.read_text())
+        except (OSError, ValueError):
+            return {}
+
+    def _save(self, entries: dict[str, dict[str, Any]]) -> None:
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        tmp = self.path.with_suffix(".tmp")
+        tmp.write_text(json.dumps(entries, indent=2))
+        tmp.replace(self.path)
+
+
+def _digest(token: str) -> str:
+    return hashlib.sha256(token.encode("utf-8")).hexdigest()