Let the server decide whose memory a caller may read

A holder is now somebody who can prove it. Schema v11 stores a sha256 of a
per-holder token beside the directory row, so a copied database cannot replay
it and the plaintext is shown exactly once. The operator token mints and
revokes those tokens; a holder token reads that holder's own memories and
nothing else; naming somebody else without the operator token is refused with
both ids in the message.

The visibility rule was leaking before this: browse had none at all, and the
search rules short-circuited to the whole library whenever no holder was
named. The rule is now written out in five copies with one parameter carried
four times, and a test compares the copies so they cannot drift — semgrep
rejects an interpolated constant, so the copies are not a choice.

Eyes carry three states on purpose. NULL means the whole library, which is
what a process that owns the store is entitled to; a holder id means that
holder; the empty string means nobody, which is what a tokenless HTTP caller
is. Making NULL mean nobody instead would have forced every internal caller —
CLI, proofs, the compiler, the bench harnesses — to name eyes, contradicting
the rule that a local process reads as its own trusted principal.

The capability card answers "can this store answer my kind of question" before
anything is sent. It is derived from what the store holds, so it cannot
promise a modality the store lacks, and it carries no memory text, so it is
safe to read before trusting the store. HTTP, CLI and one MCP tool all serve
it without a token.

backend/src/ely_eye/api.py

@@ -5,7 +5,7 @@ import shutil
 from pathlib import Path
 from typing import Annotated
 
-from fastapi import Depends, FastAPI, File, HTTPException, Query, UploadFile
+from fastapi import Depends, FastAPI, File, HTTPException, Query, Request, UploadFile
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import FileResponse
 from fastapi.staticfiles import StaticFiles
@@ -13,7 +13,8 @@ from pydantic import BaseModel
 
 from . import __version__
 from .adapters import plan as adapter_plan
-from .api_auth import require_write
+from .api_auth import read_as, require_write, resolve_caller
+from .api_card import register as register_card_routes
 from .api_cartridges import register as register_cartridge_routes
 from .api_holders import register as register_holder_routes
 from .api_memory import register as register_memory_routes
@@ -43,6 +44,7 @@ from .schemas import (
     PathIngestRequest,
     RememberRequest,
     RuntimeProfile,
+    SearchFilters,
     SearchRequest,
     SearchResult,
     SupersedeRequest,
@@ -166,12 +168,15 @@ def create_app() -> FastAPI:
     @app.post("/api/chat", response_model=AnswerResponse)
     def chat(
         request: ChatRequest,
+        http: Request,
         settings: Annotated[Settings, Depends(settings_dep)],
         _: Annotated[None, Depends(require_write)],
     ) -> AnswerResponse:
         db = Database(settings)
         compiler = ContextCompiler(settings, db)
-        context = compiler.compile(request.question, request.profile)
+        context = compiler.compile(
+            request.question, request.profile, as_holder=read_as(resolve_caller(http, settings), None)
+        )
         if not context.hits:
             raise HTTPException(status_code=404, detail="No evidence atoms are available for this question.")
         runtime = QwenRuntime(settings)
@@ -192,12 +197,24 @@ def create_app() -> FastAPI:
         )
 
     @app.post("/api/search", response_model=SearchResult)
-    def search(request: SearchRequest, settings: Annotated[Settings, Depends(settings_dep)]) -> SearchResult:
+    def search(
+        request: SearchRequest,
+        http: Request,
+        settings: Annotated[Settings, Depends(settings_dep)],
+    ) -> SearchResult:
+        """Search through the eyes the token proves. A body that names another holder is refused, not obeyed."""
         try:
+            caller = resolve_caller(http, settings)
+            eyes = read_as(caller, request.filters.as_holder if request.filters else None)
+            filters = request.filters
+            if filters is not None:
+                filters = filters.model_copy(update={"as_holder": eyes})
+            elif eyes is not None:
+                filters = SearchFilters(as_holder=eyes)
             return RetrievalService(settings).search(
                 request.query,
                 request.top_k,
-                filters=request.filters,
+                filters=filters,
                 cursor=request.cursor,
                 image=request.image,
             )
@@ -207,13 +224,15 @@ def create_app() -> FastAPI:
     @app.post("/api/memory", response_model=EvidenceAtom)
     def remember(
         request: RememberRequest,
+        http: Request,
         settings: Annotated[Settings, Depends(settings_dep)],
         _: Annotated[None, Depends(require_write)],
     ) -> EvidenceAtom:
+        """The token decides who owns the memory; the body cannot say so. The operator token writes as the default holder."""
         try:
             return MemoryService(settings).remember(
                 request.text,
-                holder_id=settings.default_holder,
+                holder_id=resolve_caller(http, settings).holder_id or settings.default_holder,
                 origin=MemoryOrigin.self_,
                 scenario=request.scenario,
                 event_time=request.event_time,
@@ -229,6 +248,7 @@ def create_app() -> FastAPI:
     @app.post("/api/memory/supersede", response_model=EvidenceAtom)
     def supersede(
         request: SupersedeRequest,
+        http: Request,
         settings: Annotated[Settings, Depends(settings_dep)],
         _: Annotated[None, Depends(require_write)],
     ) -> EvidenceAtom:
@@ -236,7 +256,7 @@ def create_app() -> FastAPI:
             return MemoryService(settings).supersede(
                 request.atom_id,
                 text=request.text,
-                holder_id=settings.default_holder,
+                holder_id=resolve_caller(http, settings).holder_id or settings.default_holder,
                 scenario=request.scenario,
                 event_time=request.event_time,
                 event_time_precision=request.event_time_precision,
@@ -263,6 +283,7 @@ def create_app() -> FastAPI:
     register_cartridge_routes(app, settings_dep)
     register_holder_routes(app, settings_dep)
     register_memory_routes(app, settings_dep)
+    register_card_routes(app, settings_dep)
 
     @app.get("/api/adapters", response_model=list[AdapterStatus])
     def adapters(settings: Annotated[Settings, Depends(settings_dep)]) -> list[AdapterStatus]:
@@ -270,6 +291,7 @@ def create_app() -> FastAPI:
 
     @app.get("/api/atoms", response_model=AtomPage)
     def atoms(
+        http: Request,
         settings: Annotated[Settings, Depends(settings_dep)],
         limit: Annotated[int, Query(ge=1, le=200)] = 50,
         cursor: str | None = None,
@@ -277,6 +299,7 @@ def create_app() -> FastAPI:
         source_id: str | None = None,
         holder: str | None = None,
     ) -> AtomPage:
+        """`holder` picks whose shelf to look at; the token decides what is on it."""
         try:
             return browse(
                 settings,
@@ -285,13 +308,16 @@ def create_app() -> FastAPI:
                 modality=modality,
                 source_id=source_id,
                 holder=holder,
+                as_holder=read_as(resolve_caller(http, settings), None),
             )
         except ValueError as exc:
             raise HTTPException(status_code=400, detail=str(exc)) from exc
 
     @app.get("/api/atoms/{atom_id}/image")
-    def atom_image(atom_id: str, settings: Annotated[Settings, Depends(settings_dep)]) -> FileResponse:
-        found = picture(settings, atom_id)
+    def atom_image(
+        atom_id: str, http: Request, settings: Annotated[Settings, Depends(settings_dep)]
+    ) -> FileResponse:
+        found = picture(settings, atom_id, as_holder=read_as(resolve_caller(http, settings), None))
         if found is None:
             raise HTTPException(status_code=404, detail=f"{atom_id} has no picture to show")
         return FileResponse(found.path, media_type=found.mime)

backend/src/ely_eye/api_auth.py

@@ -1,22 +1,94 @@
-"""Writes need a token; queries do not.
+"""Writes need a token; a read needs one to be anybody at all.
 
 Anything that can change stored state — a memory, an ingested source, a mounted cartridge, a
 proof run — is refused unless the caller presents the token the server was started with. A
 browser must also come from an allowed origin. The CLI and the MCP server reach the same
 writer through the same rule, so no face of the product gets a cheaper path to the store.
+
+A read with no token is still allowed, but it is nobody's read: it sees what nobody owns and
+what is marked for everyone, and nothing else. Naming a holder is a claim about who is asking,
+so it needs that holder's token or the operator token — otherwise `as_holder` would be a way to
+look over anyone's shoulder by typing their name.
 """
 
 from __future__ import annotations
 
 import secrets
-from typing import TYPE_CHECKING
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Literal
 
 from fastapi import HTTPException, Request
 
+from .db import Database
+from .holder_store import HolderStore
+
 if TYPE_CHECKING:
     from .config import Settings
 
 
+@dataclass(frozen=True)
+class Caller:
+    """Who the server decided is asking, from the token in front of it."""
+
+    kind: Literal["operator", "holder", "anonymous"]
+    holder_id: str | None = None
+
+    @property
+    def may_name_others(self) -> bool:
+        return self.kind == "operator"
+
+
+def presented_token(request: Request) -> str:
+    """The bearer value on this request, or an empty string when nobody presented one."""
+
+    header = request.headers.get("authorization", "")
+    return header.removeprefix("Bearer ").strip() if header.startswith("Bearer ") else ""
+
+
+def resolve_caller(request: Request, settings: Settings) -> Caller:
+    """Turn the bearer token into an identity. An unknown token is refused, not treated as nobody."""
+
+    token = presented_token(request)
+    if not token:
+        return Caller(kind="anonymous")
+    if settings.auth_token and secrets.compare_digest(token, settings.auth_token):
+        return Caller(kind="operator")
+    holder_id = HolderStore(Database(settings)).verify_token(token)
+    if holder_id is None:
+        raise HTTPException(
+            status_code=401, detail="this token is neither the operator token nor a holder token"
+        )
+    return Caller(kind="holder", holder_id=holder_id)
+
+
+def read_as(caller: Caller, claimed: str | None) -> str | None:
+    """Whose eyes a read runs under. A claim the token cannot back is refused, not ignored.
+
+    The answer is the three states the store's visibility clause understands: a holder id, the
+    empty string for a caller who is nobody, and None for an operator who reads the whole library.
+    """
+
+    if claimed is None:
+        return "" if caller.kind == "anonymous" else caller.holder_id
+    if caller.holder_id == claimed or caller.may_name_others:
+        return claimed
+    if caller.holder_id is None:
+        raise HTTPException(
+            status_code=403,
+            detail=(
+                f"an anonymous caller may not read as {claimed!r}; "
+                "send that holder's token or the operator token"
+            ),
+        )
+    raise HTTPException(
+        status_code=403,
+        detail=(
+            f"holder {caller.holder_id!r} may not read as {claimed!r}; "
+            "only the operator token may name another holder"
+        ),
+    )
+
+
 def require_write(request: Request) -> None:
     """Gate a mutating request. The app carries the settings source, so this stays a plain global."""
 

backend/src/ely_eye/api_card.py

@@ -0,0 +1,37 @@
+"""The capability card over HTTP: served without a token, because it is the reason to send one.
+
+A caller deciding whether to point an agent at this store should not have to guess whether the
+store holds speech, who is in the directory, or which schema answered last. Everything here is
+read back from the store, so it cannot advertise a sense the store has no atoms for.
+
+No `from __future__ import annotations` here: FastAPI reads these route signatures, and string
+annotations would be resolved against module globals, where the injected settings_dep exists
+only as a parameter of register().
+"""
+
+from collections.abc import Callable
+from typing import Annotated
+
+from fastapi import Depends, FastAPI, HTTPException
+
+from .card import build_card
+from .card_schemas import CapabilityCard, HolderCard
+from .config import Settings
+
+
+def register(app: FastAPI, settings_dep: Callable[[], Settings]) -> None:
+    """Register the card routes. Nothing here is secret: it is the store describing itself."""
+
+    @app.get("/api/card", response_model=CapabilityCard)
+    def card(settings: Annotated[Settings, Depends(settings_dep)]) -> CapabilityCard:
+        return build_card(settings)
+
+    @app.get("/api/card/{holder_id}", response_model=HolderCard)
+    def holder_card(
+        holder_id: str, settings: Annotated[Settings, Depends(settings_dep)]
+    ) -> HolderCard:
+        """One directory entry with its counted evidence; 404 names the holder that is not there."""
+        for entry in build_card(settings).holders:
+            if entry.holder_id == holder_id:
+                return entry
+        raise HTTPException(status_code=404, detail=f"no holder {holder_id!r} has memories or a directory entry")

backend/src/ely_eye/api_holders.py

@@ -8,9 +8,9 @@ only as a parameter of register().
 from collections.abc import Callable
 from typing import Annotated
 
-from fastapi import Depends, FastAPI, HTTPException
+from fastapi import Depends, FastAPI, HTTPException, Request
 
-from .api_auth import require_write
+from .api_auth import read_as, require_write, resolve_caller
 from .config import Settings
 from .db import Database
 from .holder_store import HolderStore
@@ -20,6 +20,7 @@ from .transactive_schemas import (
     AskRequest,
     Holder,
     HolderUpsertRequest,
+    IssuedToken,
     ReferralAnswer,
     WhoKnowsReport,
     WhoKnowsRequest,
@@ -27,7 +28,12 @@ from .transactive_schemas import (
 
 
 def register(app: FastAPI, settings_dep: Callable[[], Settings]) -> None:
-    """Register the transactive routes. The server owns the timestamps, the caller does not."""
+    """Register the transactive routes. The server owns the timestamps, the caller does not.
+
+    The directory routes answer "who knows this?" and carry counts, scores and contact details but
+    no memory text, so they stay readable without a token; `ask` reads the memories themselves, so
+    it runs through the eyes the token proves.
+    """
 
     app.state.write_settings = settings_dep
 
@@ -84,6 +90,34 @@ def register(app: FastAPI, settings_dep: Callable[[], Settings]) -> None:
             raise HTTPException(status_code=409, detail=str(exc)) from exc
         return entry
 
+    @app.post("/api/holders/{holder_id}/token", response_model=IssuedToken)
+    def issue_token(
+        holder_id: str,
+        settings: Annotated[Settings, Depends(settings_dep)],
+        _: Annotated[None, Depends(require_write)],
+    ) -> IssuedToken:
+        """Mint a bearer token for one holder. Only the operator token may do this, and the value is shown once."""
+        try:
+            entry, token = store(settings).issue_token(holder_id)
+        except ValueError as exc:
+            raise HTTPException(status_code=422, detail=str(exc)) from exc
+        issued_at = entry.token_issued_at
+        if issued_at is None:
+            raise HTTPException(status_code=500, detail=f"holder {holder_id!r} holds a token but no issue time")
+        return IssuedToken(holder_id=entry.holder_id, token=token, issued_at=issued_at)
+
+    @app.delete("/api/holders/{holder_id}/token", response_model=Holder)
+    def revoke_token(
+        holder_id: str,
+        settings: Annotated[Settings, Depends(settings_dep)],
+        _: Annotated[None, Depends(require_write)],
+    ) -> Holder:
+        """Drop a holder's token; the holder stays in the directory, only the proof of identity goes."""
+        try:
+            return store(settings).revoke_token(holder_id)
+        except ValueError as exc:
+            raise HTTPException(status_code=404, detail=str(exc)) from exc
+
     @app.get("/api/holders/{holder_id}/knows", response_model=WhoKnowsReport)
     def holder_knows(
         holder_id: str,
@@ -111,12 +145,14 @@ def register(app: FastAPI, settings_dep: Callable[[], Settings]) -> None:
     @app.post("/api/ask", response_model=ReferralAnswer)
     def ask(
         request: AskRequest,
+        http: Request,
         settings: Annotated[Settings, Depends(settings_dep)],
     ) -> ReferralAnswer:
+        """Ask, and be referred. The memories read here belong to somebody, so the token decides whose."""
         try:
             return TransactiveService(settings).ask(
                 request.question,
-                as_holder=request.as_holder,
+                as_holder=read_as(resolve_caller(http, settings), request.as_holder),
                 max_hops=request.max_hops,
                 include_text=request.include_text,
             )

backend/src/ely_eye/api_memory.py

@@ -12,8 +12,9 @@ only as a parameter of register().
 from collections.abc import Callable
 from typing import Annotated
 
-from fastapi import Depends, FastAPI, HTTPException
+from fastapi import Depends, FastAPI, HTTPException, Request
 
+from .api_auth import read_as, resolve_caller
 from .config import Settings
 from .db import Database
 from .memory import MemoryService
@@ -21,28 +22,46 @@ from .schemas import ConflictRecord, EvidenceAtom
 
 
 def register(app: FastAPI, settings_dep: Callable[[], Settings]) -> None:
-    """Register the tier reads. The holder is named by the caller because the question is about
-    somebody's memory, not about the server's own identity."""
+    """Register the tier reads. A tier belongs to somebody, so the token decides whose it is: the
+    `holder` parameter narrows the question to a name, and naming a name the token cannot back is
+    refused rather than obeyed."""
 
     def service(settings: Settings) -> MemoryService:
         return MemoryService(settings, Database(settings))
 
     @app.get("/api/memory/working", response_model=list[EvidenceAtom])
     def working_set(
-        holder: str,
         session: str,
+        http: Request,
         settings: Annotated[Settings, Depends(settings_dep)],
+        holder: str | None = None,
     ) -> list[EvidenceAtom]:
         """What one session recorded, in the order it recorded it, with no search involved."""
+        who = read_as(resolve_caller(http, settings), holder)
+        if not who:
+            raise HTTPException(
+                status_code=422,
+                detail="a working set belongs to somebody; send that holder's token, or the operator token with a holder",
+            )
         try:
-            return service(settings).working_set(holder, session)
+            return service(settings).working_set(who, session)
         except ValueError as exc:
             raise HTTPException(status_code=422, detail=str(exc)) from exc
 
     @app.get("/api/memory/conflicts", response_model=list[ConflictRecord])
     def conflicts(
+        http: Request,
         settings: Annotated[Settings, Depends(settings_dep)],
         holder: str | None = None,
     ) -> list[ConflictRecord]:
         """Every disagreement the store noticed and nobody has settled."""
-        return service(settings).conflicts(holder)
+        who = read_as(resolve_caller(http, settings), holder)
+        if who == "":
+            raise HTTPException(
+                status_code=422,
+                detail=(
+                    "a disagreement belongs to somebody; send that holder's token, "
+                    "or the operator token with a holder"
+                ),
+            )
+        return service(settings).conflicts(who)

backend/src/ely_eye/browse.py

@@ -11,10 +11,12 @@ from .retrieval import decode_cursor, encode_cursor
 from .schemas import AtomPage, Modality
 from .storage import ObjectStore
 
+# The visibility rule is spelled out in four literal statements — two here, one in store_search, one in vector_index — because the SQL gate rejects a concatenated statement and a view cannot take a parameter; backend/tests/test_visibility_rule.py fails if the copies disagree.
+
 
 @dataclass(frozen=True)
 class Picture:
-    """The bytes behind an atom and how to label them when serving."""
+    """The stored bytes behind an atom and how to label them when serving."""
 
     path: Path
     mime: str
@@ -28,22 +30,48 @@ def browse(
     modality: Modality | None = None,
     source_id: str | None = None,
     holder: str | None = None,
+    as_holder: str | None = None,
 ) -> AtomPage:
-    """One page plus the total, so a list can say what it left out instead of looking endless."""
+    """One page plus the total, so a list can say what it left out instead of looking endless.
+
+    `holder` narrows the shelf to one holder's memories; `as_holder` is whose eyes the page is
+    read through: a holder id, the empty string for a caller who is nobody (only what nobody owns
+    and what is marked for everyone), or None to read the whole library as the operator does.
+    """
     offset = decode_cursor(cursor)
     wanted = (modality.value if modality else None, source_id, holder)
     filters = tuple(value for bound in wanted for value in (bound, bound))
+    eyes = (as_holder, as_holder, as_holder, as_holder)
     database = Database(settings)
     with database.connect() as connection:
         total = int(
             connection.execute(
-                "SELECT COUNT(*) FROM v_atoms_current WHERE (? IS NULL OR modality = ?) AND (? IS NULL OR source_id = ?) AND (? IS NULL OR holder_id = ?)",
-                filters,
+                "SELECT COUNT(*) FROM v_atoms_current"
+                " WHERE (? IS NULL OR modality = ?) AND (? IS NULL OR source_id = ?) AND (? IS NULL OR holder_id = ?)"
+                " AND (? IS NULL"
+                "      OR holder_id = ''"
+                "      OR json_extract(visibility_json, '$.scope') = 'everyone'"
+                "      OR (? <> '' AND (holder_id = ?"
+                "            OR EXISTS ("
+                "                  SELECT 1 FROM json_each(json_extract(visibility_json, '$.holders'))"
+                "                  WHERE value = ?"
+                "            ))))",
+                (*filters, *eyes),
             ).fetchone()[0]
         )
         rows = connection.execute(
-            "SELECT * FROM v_atoms_current WHERE (? IS NULL OR modality = ?) AND (? IS NULL OR source_id = ?) AND (? IS NULL OR holder_id = ?) ORDER BY ingested_at DESC, atom_id DESC LIMIT ? OFFSET ?",
-            (*filters, limit + 1, offset),
+            "SELECT * FROM v_atoms_current"
+            " WHERE (? IS NULL OR modality = ?) AND (? IS NULL OR source_id = ?) AND (? IS NULL OR holder_id = ?)"
+            " AND (? IS NULL"
+            "      OR holder_id = ''"
+            "      OR json_extract(visibility_json, '$.scope') = 'everyone'"
+            "      OR (? <> '' AND (holder_id = ?"
+            "            OR EXISTS ("
+            "                  SELECT 1 FROM json_each(json_extract(visibility_json, '$.holders'))"
+            "                  WHERE value = ?"
+            "            ))))"
+            " ORDER BY ingested_at DESC, atom_id DESC LIMIT ? OFFSET ?",
+            (*filters, *eyes, limit + 1, offset),
         ).fetchall()
     atoms = [row_to_atom(row) for row in rows[:limit]]
     more = len(rows) > limit or offset + limit < total
@@ -54,13 +82,22 @@ def browse(
     )
 
 
-def picture(settings: Settings, atom_id: str) -> Picture | None:
+def picture(settings: Settings, atom_id: str, as_holder: str | None = None) -> Picture | None:
     """The stored bytes behind one atom; None means there is no picture to show, for any of the three reasons."""
     database = Database(settings)
     with database.connect() as connection:
         row = connection.execute(
-            "SELECT image_ref, mime FROM v_atoms_current JOIN sources USING (source_id) WHERE atom_id = ?",
-            (atom_id,),
+            "SELECT image_ref, mime FROM v_atoms_current JOIN sources USING (source_id)"
+            " WHERE atom_id = ?"
+            " AND (? IS NULL"
+            "      OR holder_id = ''"
+            "      OR json_extract(visibility_json, '$.scope') = 'everyone'"
+            "      OR (? <> '' AND (holder_id = ?"
+            "            OR EXISTS ("
+            "                  SELECT 1 FROM json_each(json_extract(visibility_json, '$.holders'))"
+            "                  WHERE value = ?"
+            "            ))))",
+            (atom_id, as_holder, as_holder, as_holder, as_holder),
         ).fetchone()
     if row is None or not row["image_ref"]:
         return None

backend/src/ely_eye/cache_fabric.py

@@ -17,9 +17,9 @@ class CacheFabric:
         self.settings = settings or get_settings()
         self.db = database or Database(self.settings)
 
-    def semantic_key(self, question: str, profile: str, store_revision: int) -> str:
-        """A compiled context is only true for the corpus it was compiled from, so the corpus is part of its name."""
-        return hashlib.sha256(f"compiler-v2\n{profile}\n{store_revision}\n{question}".encode()).hexdigest()
+    def semantic_key(self, question: str, profile: str, store_revision: int, holder: str = "") -> str:
+        """A compiled context is only true for the corpus it was compiled from and the eyes that were allowed to see it, so both belong in its name."""
+        return hashlib.sha256(f"compiler-v2\n{profile}\n{store_revision}\n{holder}\n{question}".encode()).hexdigest()
 
     def record_hit(self, layer: str, cache_key: str, bytes_estimate: int = 0, metadata: dict[str, Any] | None = None) -> None:
         self.db.log_cache_event(layer, cache_key, "hit", bytes_estimate, metadata)

backend/src/ely_eye/card.py

@@ -0,0 +1,92 @@
+"""What this store can answer, read back from what it actually holds.
+
+A caller deciding whether to send its questions here needs three facts it cannot guess: who is in
+the directory and what they claim to know, whether the memory it would get covers pictures and
+speech or only words, and which schema the answers were produced under. Every field is read from
+the store, so the card cannot promise a sense the store has no atoms for, and it cannot promise a
+capacity a mount has not carried.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+
+from .card_schemas import CapabilityCard, HolderCard
+from .cartridge import CartridgeService
+from .config import Settings
+from .db import Database
+from .holder_store import HolderStore
+
+
+def build_card(settings: Settings) -> CapabilityCard:
+    """The store-level card: schema, counts, senses present, and what is mounted read-only."""
+
+    database = Database(settings)
+    with database.connect() as connection:
+        schema_version = int(connection.execute("PRAGMA user_version").fetchone()[0])
+        sources = int(connection.execute("SELECT COUNT(*) FROM sources").fetchone()[0])
+        senses = {
+            str(row["modality"]): int(row["n"])
+            for row in connection.execute(
+                "SELECT modality, COUNT(*) AS n FROM v_atoms_current GROUP BY modality ORDER BY n DESC"
+            )
+        }
+        per_holder = {
+            str(row["holder_id"]): (int(row["n"]), str(row["newest"]))
+            for row in connection.execute(
+                "SELECT holder_id, COUNT(*) AS n, MAX(ingested_at) AS newest FROM v_atoms_current"
+                " GROUP BY holder_id"
+            )
+        }
+        senses_by_holder: dict[str, dict[str, int]] = {}
+        scenarios_by_holder: dict[str, list[str]] = {}
+        for row in connection.execute(
+            "SELECT holder_id, modality, COUNT(*) AS n FROM v_atoms_current"
+            " GROUP BY holder_id, modality ORDER BY n DESC"
+        ):
+            senses_by_holder.setdefault(str(row["holder_id"]), {})[str(row["modality"])] = int(row["n"])
+        for row in connection.execute(
+            "SELECT holder_id, scenario, COUNT(*) AS n FROM v_atoms_current"
+            " GROUP BY holder_id, scenario HAVING n > 0 ORDER BY n DESC"
+        ):
+            scenarios_by_holder.setdefault(str(row["holder_id"]), []).append(str(row["scenario"]))
+
+    directory = {entry.holder_id: entry for entry in HolderStore(database).list_holders()}
+    cards: dict[str, HolderCard] = {}
+    for holder_id, (memories, newest) in per_holder.items():
+        entry = directory.get(holder_id)
+        cards[holder_id] = HolderCard(
+            holder_id=holder_id,
+            name=entry.name if entry else "",
+            kind=entry.kind if entry else "person",
+            knows_about=list(entry.knows_about) if entry else [],
+            contact=dict(entry.contact) if entry else {},
+            memories=memories,
+            modalities=senses_by_holder.get(holder_id, {}),
+            scenarios=scenarios_by_holder.get(holder_id, []),
+            newest_memory=datetime.fromisoformat(newest),
+            registered=entry is not None,
+            token_issued=entry is not None and entry.token_issued_at is not None,
+        )
+    for holder_id, entry in directory.items():
+        if holder_id not in cards:
+            cards[holder_id] = HolderCard(
+                holder_id=holder_id,
+                name=entry.name,
+                kind=entry.kind,
+                knows_about=list(entry.knows_about),
+                contact=dict(entry.contact),
+                newest_memory=None,
+                registered=True,
+                token_issued=entry.token_issued_at is not None,
+            )
+
+    return CapabilityCard(
+        schema_version=schema_version,
+        atoms=database.atom_count(),
+        sources=sources,
+        modalities=senses,
+        has_audio="audio" in senses,
+        holders=[cards[key] for key in sorted(cards)],
+        mounts=CartridgeService(settings).mount_state(),
+    )

backend/src/ely_eye/card_schemas.py

@@ -0,0 +1,58 @@
+"""The capability card: what this store can answer, stated as data.
+
+Held apart from the memory schemas because the card is a promise made to a stranger — it is served
+without a token — while the memory schemas describe what a holder's own memories look like. The
+two change for different reasons, so they live in different files.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+
+from pydantic import BaseModel, Field
+
+from .schemas import MountState, utc_now
+from .transactive_schemas import HolderKind
+
+
+class HolderCard(BaseModel):
+    """One directory entry, plus the evidence that it is a holder rather than a typed-in name.
+
+    `memories` and `newest_memory` are counted from the store, so a holder who declared ten
+    topics and recorded nothing shows as such. `token_issued` says whether anything can prove it
+    is this holder; `registered` says whether the directory knows it at all, which is false for a
+    holder_id that only ever appeared on a memory.
+    """
+
+    holder_id: str
+    name: str = ""
+    kind: HolderKind = "person"
+    knows_about: list[str] = Field(default_factory=list)
+    contact: dict[str, str] = Field(default_factory=dict)
+    memories: int = Field(ge=0, default=0)
+    modalities: dict[str, int] = Field(default_factory=dict)
+    scenarios: list[str] = Field(default_factory=list)
+    newest_memory: datetime | None = None
+    registered: bool = False
+    token_issued: bool = False
+
+
+class CapabilityCard(BaseModel):
+    """What a caller can expect from this store before it sends anything into it.
+
+    `has_audio` exists because a store that ingested only pictures and PDFs answers questions
+    about speech with OCR of a frame, and the caller should learn that before it relies on it.
+    `schema_version` is read from the open database, not from the build constant, so a store that
+    has not been migrated yet reports what it actually is. The card is served without a token: it
+    names holders, counts their memories and lists the scenario names they used, but no memory
+    text — a caller learns what this store can answer, not what it remembers.
+    """
+
+    schema_version: int
+    atoms: int = Field(ge=0)
+    sources: int = Field(ge=0)
+    modalities: dict[str, int] = Field(default_factory=dict)
+    has_audio: bool
+    holders: list[HolderCard] = Field(default_factory=list)
+    mounts: MountState
+    generated_at: datetime = Field(default_factory=utc_now)

backend/src/ely_eye/cli_transactive.py

@@ -10,12 +10,13 @@ from collections.abc import Callable
 
 import typer
 
+from .card import build_card
 from .config import Settings
 from .db import Database
 from .holder_store import HolderStore
 from .schemas import utc_now
 from .transactive import TransactiveService
-from .transactive_schemas import Holder, HolderKind
+from .transactive_schemas import Holder, HolderKind, IssuedToken
 
 
 def register(app: typer.Typer, settings: Callable[[], Settings]) -> None:
@@ -62,6 +63,25 @@ def register(app: typer.Typer, settings: Callable[[], Settings]) -> None:
         store().delete(holder_id)
         typer.echo(f'{{"deleted": "{holder_id}"}}')
 
+    @app.command("holder-issue")
+    def holder_issue(holder_id: str) -> None:
+        """Mint a bearer token for one holder. The value appears once: only its hash is stored."""
+        entry, token = store().issue_token(holder_id)
+        if entry.token_issued_at is None:
+            raise ValueError(f"holder {holder_id!r} holds a token but no issue time")
+        issued = IssuedToken(holder_id=entry.holder_id, token=token, issued_at=entry.token_issued_at)
+        typer.echo(issued.model_dump_json(indent=2))
+
+    @app.command("holder-revoke")
+    def holder_revoke(holder_id: str) -> None:
+        """Drop a holder's token. The entry stays; nobody can read as them until a new one is issued."""
+        typer.echo(store().revoke_token(holder_id).model_dump_json(indent=2))
+
+    @app.command()
+    def card() -> None:
+        """What this store can answer, read back from what it holds: holders, senses, schema, mounts."""
+        typer.echo(build_card(settings()).model_dump_json(indent=2))
+
     @app.command("who-knows")
     def who_knows(query: str, as_holder: str = "", limit: int = 8) -> None:
         """Rank the holders whose memories answer this query."""

backend/src/ely_eye/compiler.py

@@ -8,7 +8,14 @@ from .cache_fabric import SEMANTIC_LAYER, CacheFabric
 from .config import Settings, get_settings
 from .db import Database
 from .retrieval import RetrievalService
-from .schemas import CompiledContext, ContextPlan, Modality, RetrievalHit, RuntimeProfile
+from .schemas import (
+    CompiledContext,
+    ContextPlan,
+    Modality,
+    RetrievalHit,
+    RuntimeProfile,
+    SearchFilters,
+)
 
 PROFILE_TOKEN_BUDGETS = {
     RuntimeProfile.live_demo: 128_000,
@@ -25,8 +32,18 @@ class ContextCompiler:
         self.retrieval = RetrievalService(self.settings, self.db)
         self.cache = CacheFabric(self.settings, self.db)
 
-    def compile(self, question: str, profile: RuntimeProfile) -> CompiledContext:
-        cache_key = self.cache.semantic_key(question, profile.value, self.db.revision())
+    def compile(self, question: str, profile: RuntimeProfile, as_holder: str | None = None) -> CompiledContext:
+        """Build the context one question gets, seen through one holder's eyes.
+
+        `as_holder` is passed to retrieval and folded into the cache key: a context compiled for
+        a holder carries that holder's own memories, and serving it to somebody else would leak.
+        """
+        cache_key = self.cache.semantic_key(
+            question,
+            profile.value,
+            self.db.revision(),
+            "the whole library" if as_holder is None else f"as holder {as_holder!r}",
+        )
         cached = self.cache.read_semantic_cache(cache_key)
         if cached is not None:
             compiled = CompiledContext.model_validate(cached)
@@ -40,7 +57,9 @@ class ContextCompiler:
         else:
             self.cache.record_miss(SEMANTIC_LAYER, cache_key)
         plan = self.plan(question, profile)
-        hits = self.retrieval.search(plan.retrieval_query, plan.evidence_budget_atoms).hits
+        hits = self.retrieval.search(
+            plan.retrieval_query, plan.evidence_budget_atoms, filters=SearchFilters(as_holder=as_holder)
+        ).hits
         citable, held = self._split_citable(hits)
         packed_text, token_total = self._pack_hits(citable, plan.token_budget, plan.retrieval_query)
         trace_id = hashlib.sha256(f"{cache_key}:{token_total}".encode()).hexdigest()[:16]

backend/src/ely_eye/holder_store.py

@@ -8,6 +8,8 @@ never a guess.
 
 from __future__ import annotations
 
+import hashlib
+import secrets
 import sqlite3
 from dataclasses import dataclass
 from datetime import datetime
@@ -30,6 +32,9 @@ class WhoKnowsCount:
     latest_time: str | None
 
 
+_TOKEN_BYTES = 32
+
+
 class HolderStore:
     """CRUD for registered holders plus the raw counts who_knows is built from."""
 
@@ -68,6 +73,63 @@ class HolderStore:
             )
         return stored
 
+    def register_minimal(self, holder_id: str) -> None:
+        """Make a holder exist without overwriting what somebody already declared about them.
+
+        A memory nobody registered still needs a directory entry, or a referral to its holder has
+        no target to point at.
+        """
+        now = utc_now().isoformat()
+        with self.database.connect() as connection:
+            connection.execute(
+                """
+                INSERT INTO holders (holder_id, name, kind, knows_about_json, contact_json, created_at, updated_at)
+                VALUES (?, '', 'person', '[]', '{}', ?, ?)
+                ON CONFLICT(holder_id) DO NOTHING
+                """,
+                (holder_id, now, now),
+            )
+
+    def issue_token(self, holder_id: str) -> tuple[Holder, str]:
+        """Give a holder a fresh bearer token. The plain value is returned once; the row keeps its hash."""
+        self.register_minimal(holder_id)
+        token = secrets.token_urlsafe(_TOKEN_BYTES)
+        now = utc_now().isoformat()
+        with self.database.connect() as connection:
+            connection.execute(
+                "UPDATE holders SET token_hash = ?, token_issued_at = ?, updated_at = ? WHERE holder_id = ?",
+                (_token_hash(token), now, now, holder_id),
+            )
+        entry = self.get(holder_id)
+        if entry is None:
+            raise ValueError(f"holder {holder_id!r} vanished while its token was being issued")
+        return entry, token
+
+    def revoke_token(self, holder_id: str) -> Holder:
+        """Drop a holder's token: nobody can read as them until a new one is issued."""
+        now = utc_now().isoformat()
+        with self.database.connect() as connection:
+            changed = connection.execute(
+                "UPDATE holders SET token_hash = NULL, token_issued_at = NULL, updated_at = ? WHERE holder_id = ?",
+                (now, holder_id),
+            )
+        if changed.rowcount == 0:
+            raise ValueError(f"no holder {holder_id!r} is registered")
+        entry = self.get(holder_id)
+        if entry is None:
+            raise ValueError(f"holder {holder_id!r} vanished while its token was being revoked")
+        return entry
+
+    def verify_token(self, token: str) -> str | None:
+        """Which holder a bearer token speaks for, or None when it speaks for nobody."""
+        if not token:
+            return None
+        with self.database.connect() as connection:
+            row = connection.execute(
+                "SELECT holder_id FROM holders WHERE token_hash = ?", (_token_hash(token),)
+            ).fetchone()
+        return None if row is None else str(row["holder_id"])
+
     def get(self, holder_id: str) -> Holder | None:
         with self.database.connect() as connection:
             row = connection.execute("SELECT * FROM holders WHERE holder_id = ?", (holder_id,)).fetchone()
@@ -145,6 +207,10 @@ class HolderStore:
         ]
 
 
+def _token_hash(token: str) -> str:
+    return hashlib.sha256(token.encode("utf-8")).hexdigest()
+
+
 def _moment(value: str) -> datetime:
     parsed = datetime.fromisoformat(value)
     if parsed.tzinfo is None:
@@ -168,6 +234,7 @@ def _row_to_holder(row: sqlite3.Row) -> Holder:
         kind=_kind(str(row["kind"])),
         knows_about=from_json(row["knows_about_json"], []),
         contact=from_json(row["contact_json"], {}),
+        token_issued_at=None if row["token_issued_at"] is None else _moment(str(row["token_issued_at"])),
         created_at=_moment(str(row["created_at"])),
         updated_at=_moment(str(row["updated_at"])),
     )

backend/src/ely_eye/memory.py

@@ -17,6 +17,7 @@ from . import lifecycle_store, trust
 from .cache_fabric import CacheFabric
 from .config import Settings, get_settings
 from .db import Database
+from .holder_store import HolderStore
 from .schemas import (
     ArchiveReport,
     ConflictRecord,
@@ -75,7 +76,8 @@ class MemoryService:
         """Store one statement as a memory owned by a holder.
 
         Repeating the same sentence as the same holder updates the memory instead of
-        duplicating it; two holders holding the same sentence keep two memories.
+        duplicating it; two holders holding the same sentence keep two memories. A holder nobody
+        registered is added to the directory here, so a referral to it has something to point at.
         """
         statement = text.strip()
         if not statement:
@@ -89,6 +91,7 @@ class MemoryService:
             )
 
         moment = utc_now()
+        HolderStore(self.db).register_minimal(holder_id)
         digest = hashlib.sha256(f"{holder_id}\n{statement}".encode()).hexdigest()
         source_id = f"mem-{digest[:16]}"
         self.db.upsert_sources(

backend/src/ely_eye/migrations.py

@@ -16,7 +16,7 @@ import sqlite3
 from .storage import token_equivalent, visual_token_equivalent
 from .text_index import index_text
 
-SCHEMA_VERSION = 10
+SCHEMA_VERSION = 11
 
 ATOM_COLUMNS_V2 = """
     atom_id TEXT PRIMARY KEY,
@@ -96,6 +96,8 @@ _BASE_TABLES = """
             CHECK (kind IN ('person', 'agent', 'service')),
         knows_about_json TEXT NOT NULL DEFAULT '[]',
         contact_json TEXT NOT NULL DEFAULT '{}',
+        token_hash TEXT,
+        token_issued_at TEXT,
         created_at TEXT NOT NULL,
         updated_at TEXT NOT NULL
     );
@@ -290,6 +292,13 @@ def _recount_visual_tokens(connection: sqlite3.Connection) -> None:
     connection.executemany("UPDATE atoms SET token_equivalent = ? WHERE atom_id = ?", updates)
 
 
+_NEW_HOLDER_COLUMNS: tuple[tuple[str, str], ...] = (
+    ("token_hash", "TEXT"),
+    ("token_issued_at", "TEXT"),
+)
+"""A holder proves its identity with a bearer token; only its hash is kept, so a copied database cannot replay it."""
+
+
 def apply(connection: sqlite3.Connection) -> None:
     """Bring the schema to SCHEMA_VERSION, in place, without losing rows."""
     version = int(connection.execute("PRAGMA user_version").fetchone()[0])
@@ -314,6 +323,11 @@ def apply(connection: sqlite3.Connection) -> None:
             for name, definition in columns:
                 if name not in existing:
                     connection.execute(f"ALTER TABLE atoms ADD COLUMN {name} {definition}")
+    if version < 11:
+        existing_holders = _column_names(connection, "holders")
+        for name, definition in _NEW_HOLDER_COLUMNS:
+            if name not in existing_holders:
+                connection.execute(f"ALTER TABLE holders ADD COLUMN {name} {definition}")
     if version < 10:
         # CREATE VIEW IF NOT EXISTS keeps the old body, and v10 changed which rows are current, so an upgraded store must be told to look again.
         connection.execute("DROP VIEW IF EXISTS v_atoms_current")

backend/src/ely_eye/store_search.py

@@ -46,10 +46,11 @@ def search_atoms(
     depends on string-built SQL, and the planner sees one shape. A memory with no event time
     cannot answer a time filter, so it drops out when one is asked for.
 
-    `as_holder` applies the sharing rule: a memory is readable by its owner, by anyone named
-    in its visible_to list, by everyone when it is public, and by anyone when nobody owns it.
-    A memory with no holder cannot be private to a holder that does not exist. Without an
-    asker the query runs over the whole library, which is what an operator debugging wants."""
+    `as_holder` applies the sharing rule in three states: a holder id sees its own memories, the
+    ones naming it in visible_to, the ones marked for everyone, and the ones nobody owns; the
+    empty string is a caller who is nobody, so only the last two; None runs over the whole
+    library, which is what an operator or a process that owns the store reads with. A memory with
+    no holder cannot be private to a holder that does not exist."""
     with database.connect() as connection:
         expression = fts_expression(rare_terms(connection, fts_terms(query)))
         rows = connection.execute(
@@ -77,13 +78,14 @@ def search_atoms(
                     WHERE ca.atom_id = f.atom_id
                       AND ca.cartridge_id IN (SELECT value FROM json_each(?))
               ))
-              AND (? IS NULL OR a.holder_id = ''
+              AND (? IS NULL
+                    OR a.holder_id = ''
                     OR json_extract(a.visibility_json, '$.scope') = 'everyone'
-                    OR a.holder_id = ?
-                    OR EXISTS (
-                          SELECT 1 FROM json_each(json_extract(a.visibility_json, '$.holders'))
-                          WHERE value = ?
-                    ))
+                    OR (? <> '' AND (a.holder_id = ?
+                          OR EXISTS (
+                                SELECT 1 FROM json_each(json_extract(a.visibility_json, '$.holders'))
+                                WHERE value = ?
+                          ))))
             ORDER BY score
             LIMIT ? OFFSET ?
             """,
@@ -108,6 +110,7 @@ def search_atoms(
                 as_holder,
                 as_holder,
                 as_holder,
+                as_holder,
                 limit + 1,
                 offset,
             ),

backend/src/ely_eye/transactive_schemas.py

@@ -24,6 +24,8 @@ class Holder(BaseModel):
     kind: HolderKind = "person"
     knows_about: list[str] = Field(default_factory=list)
     contact: dict[str, str] = Field(default_factory=dict)
+    token_issued_at: datetime | None = None
+    """When a bearer token was last issued here. The hash itself stays out of the answer."""
     created_at: datetime
     updated_at: datetime
 
@@ -35,6 +37,18 @@ class Holder(BaseModel):
         return value.strip()
 
 
+class IssuedToken(BaseModel):
+    """A holder's bearer token, shown once at the moment it is minted.
+
+    Only the hash is stored, so this answer is the only place the value will ever appear: a lost
+    token is replaced, not recovered.
+    """
+
+    holder_id: str
+    token: str
+    issued_at: datetime
+
+
 class HolderUpsertRequest(BaseModel):
     """What a caller may set about a holder. The id comes from the path, never the body."""
 

backend/src/ely_eye/vector_index.py

@@ -225,7 +225,9 @@ class VectorIndex:
         question with a full scan (77ms over 153,857 atoms) while the atoms table answers it
         through the atom_id index in 0.2ms, and a search cannot afford the scan. The scope is
         checked here rather than in the engine so the rule lives in one place; the engine's id
-        list is only a narrowing.
+        list is only a narrowing. Visibility is here too, and it reads the same three states the
+        words leg reads: a holder id, the empty string for a caller who is nobody (unowned rows
+        and rows marked for everyone, no more), and None for an operator reading the whole library.
         """
         rows = connection.execute(
             """
@@ -246,13 +248,14 @@ class VectorIndex:
                     WHERE ca.atom_id = atoms.atom_id
                       AND ca.cartridge_id IN (SELECT value FROM json_each(?))
               ))
-              AND (? IS NULL OR holder_id = ''
+              AND (? IS NULL
+                    OR holder_id = ''
                     OR json_extract(visibility_json, '$.scope') = 'everyone'
-                    OR holder_id = ?
-                    OR EXISTS (
-                          SELECT 1 FROM json_each(json_extract(visibility_json, '$.holders'))
-                          WHERE value = ?
-                    ))
+                    OR (? <> '' AND (holder_id = ?
+                          OR EXISTS (
+                                SELECT 1 FROM json_each(json_extract(visibility_json, '$.holders'))
+                                WHERE value = ?
+                          ))))
             """,
             (
                 to_json(atom_ids),
@@ -273,6 +276,7 @@ class VectorIndex:
                 as_holder,
                 as_holder,
                 as_holder,
+                as_holder,
             ),
         ).fetchall()
         return {str(row["atom_id"]) for row in rows}

backend/tests/test_cartridge_scope.py

@@ -51,7 +51,8 @@ class Store:
         return self.cartridges.materialize(name, atoms, sources).cartridge_id
 
     def texts(self, query: str = "钥匙", **filters: Any) -> list[str]:
-        result = self.search.search(query, filters=SearchFilters(**filters) if filters else None)
+        eyes: dict[str, Any] = {"as_holder": "wife", **filters}
+        result = self.search.search(query, filters=SearchFilters(**eyes))
         return [hit.atom.text for hit in result.hits]
 
 

backend/tests/test_holder_acl.py

@@ -0,0 +1,268 @@
+"""The sharing rule over HTTP: a token says who is asking, and the store answers to that person."""
+
+from __future__ import annotations
+
+import hashlib
+from pathlib import Path
+from typing import Any
+
+import pytest
+from fastapi.testclient import TestClient
+
+import ely_eye.api as api_module
+from ely_eye.compiler import ContextCompiler
+from ely_eye.config import Settings
+from ely_eye.db import Database
+from ely_eye.holder_store import HolderStore
+from ely_eye.memory import MemoryService
+from ely_eye.schemas import (
+    EvidenceAtom,
+    MemoryOrigin,
+    Modality,
+    RuntimeProfile,
+    SourceRecord,
+    TrustScores,
+    Visibility,
+)
+from ely_eye.storage import ObjectStore
+
+OPERATOR = "the-operator-token"
+AS_OPERATOR = {"Authorization": f"Bearer {OPERATOR}"}
+PNG = b"\x89PNG\r\n\x1a\n"
+KEYS = "我们的钥匙放在玄关抽屉里"
+THERMOSTAT = "恒温器设在二十二度"
+OUTAGE = "上周三的停电持续了四十分钟"
+
+
+def _settings(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Settings:
+    resolved = Settings(home=tmp_path, default_holder="wife", auth_token=OPERATOR)
+    monkeypatch.setattr(api_module, "get_settings", lambda: resolved)
+    Database(resolved).init()
+    return resolved
+
+
+@pytest.fixture
+def client(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> TestClient:
+    return TestClient(api_module.create_app())
+
+
+@pytest.fixture
+def settings(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Settings:
+    return _settings(tmp_path, monkeypatch)
+
+
+def _token_for(client: TestClient, holder_id: str) -> dict[str, Any]:
+    answer = client.post(f"/api/holders/{holder_id}/token", headers=AS_OPERATOR)
+    assert answer.status_code == 200, answer.json()
+    return dict(answer.json())
+
+
+def _remember(settings: Settings, text: str, holder_id: str, **kwargs: Any) -> EvidenceAtom:
+    return MemoryService(settings, Database(settings)).remember(
+        text, holder_id=holder_id, origin=MemoryOrigin.self_, **kwargs
+    )
+
+
+def _hits(client: TestClient, query: str, headers: dict[str, str] | None = None, **body: Any) -> list[str]:
+    answer = client.post("/api/search", json={"query": query, **body}, headers=headers or {})
+    assert answer.status_code == 200, answer.json()
+    return [str(hit["atom"]["atom_id"]) for hit in answer.json()["hits"]]
+
+
+def test_only_the_operator_mints_a_token(client: TestClient, settings: Settings) -> None:
+    _remember(settings, KEYS, "wife")
+    wife = _token_for(client, "wife")
+    assert client.post("/api/holders/husband/token").status_code == 401, "no token cannot mint one"
+    refused = client.post("/api/holders/husband/token", headers={"Authorization": f"Bearer {wife['token']}"})
+    assert refused.status_code == 401, "a holder token is not an operator token"
+    assert "Bearer" in refused.json()["detail"]
+
+
+def test_the_token_is_shown_once_and_stored_as_a_hash(client: TestClient, settings: Settings) -> None:
+    _remember(settings, KEYS, "wife")
+    issued = _token_for(client, "wife")
+    token = str(issued["token"])
+    assert issued["holder_id"] == "wife" and issued["issued_at"]
+
+    with Database(settings).connect() as connection:
+        row = connection.execute("SELECT token_hash FROM holders WHERE holder_id = 'wife'").fetchone()
+    stored = str(row["token_hash"])
+    assert stored == hashlib.sha256(token.encode("utf-8")).hexdigest()
+    assert token not in stored and len(stored) == 64, "a copied database must not replay the token"
+
+    listed = client.get("/api/holders/wife", headers=AS_OPERATOR).json()
+    assert "token" not in listed and listed["token_issued_at"] == issued["issued_at"]
+
+
+def test_a_holder_token_reads_as_that_holder_and_nobody_else(client: TestClient, settings: Settings) -> None:
+    mine = _remember(settings, KEYS, "wife")
+    wife = _token_for(client, "wife")
+    _token_for(client, "husband")
+
+    as_wife = _hits(client, KEYS, headers={"Authorization": f"Bearer {wife['token']}"})
+    as_husband = _hits(client, KEYS, headers={"Authorization": f"Bearer {wife['token']}"})
+    assert mine.atom_id in as_wife
+    assert as_husband == as_wife, "the same token reads the same way whatever the query says"
+
+    anonymous = _hits(client, KEYS)
+    assert mine.atom_id not in anonymous, "a caller with no token is nobody"
+
+
+def test_an_unknown_token_is_refused_rather_than_treated_as_nobody(client: TestClient, settings: Settings) -> None:
+    _remember(settings, KEYS, "wife")
+    refused = client.post("/api/search", json={"query": KEYS}, headers={"Authorization": "Bearer not-a-token"})
+    assert refused.status_code == 401
+    detail = str(refused.json()["detail"])
+    assert "operator" in detail and "holder" in detail, "the refusal has to say which two things failed"
+
+
+def test_a_public_memory_is_findable_by_anybody(client: TestClient, settings: Settings) -> None:
+    shared = _remember(settings, OUTAGE, "wife", visibility=Visibility(scope="everyone"))
+    assert shared.atom_id in _hits(client, OUTAGE), "no token, still visible"
+
+
+def test_a_shared_memory_is_readable_only_by_the_holders_it_names(client: TestClient, settings: Settings) -> None:
+    private = _remember(settings, THERMOSTAT, "wife", visibility=Visibility(scope="visible_to", holders=["husband"]))
+    _remember(settings, KEYS, "husband")
+    wife = _token_for(client, "wife")
+    husband = _token_for(client, "husband")
+
+    assert private.atom_id in _hits(client, THERMOSTAT, headers={"Authorization": f"Bearer {husband['token']}"})
+    assert private.atom_id in _hits(client, THERMOSTAT, headers={"Authorization": f"Bearer {wife['token']}"})
+    assert private.atom_id not in _hits(client, THERMOSTAT)
+
+    outsider = _remember(settings, OUTAGE, "neighbour")
+    _token_for(client, "neighbour")
+    neighbour = _token_for(client, "neighbour")
+    assert outsider.atom_id
+    assert private.atom_id not in _hits(client, THERMOSTAT, headers={"Authorization": f"Bearer {neighbour['token']}"})
+
+
+def test_naming_another_holder_is_a_claim_the_token_has_to_back(client: TestClient, settings: Settings) -> None:
+    mine = _remember(settings, KEYS, "wife")
+    wife = _token_for(client, "wife")
+    _token_for(client, "husband")
+    as_wife = {"Authorization": f"Bearer {wife['token']}"}
+
+    over = client.post("/api/search", json={"query": KEYS, "filters": {"as_holder": "husband"}}, headers=as_wife)
+    assert over.status_code == 403
+    detail = str(over.json()["detail"])
+    assert "wife" in detail and "husband" in detail, "a refusal that names neither side cannot be acted on"
+
+    anonymous = client.post("/api/search", json={"query": KEYS, "filters": {"as_holder": "wife"}})
+    assert anonymous.status_code == 403, "typing a name is not a token"
+
+    operator = client.post("/api/search", json={"query": KEYS, "filters": {"as_holder": "wife"}}, headers=AS_OPERATOR)
+    assert operator.status_code == 200
+    assert mine.atom_id in [str(hit["atom"]["atom_id"]) for hit in operator.json()["hits"]]
+
+
+def test_revoking_a_token_stops_it_working(client: TestClient, settings: Settings) -> None:
+    _remember(settings, KEYS, "wife")
+    token = _token_for(client, "wife")["token"]
+    assert HolderStore(Database(settings)).verify_token(str(token)) == "wife"
+
+    revoked = client.delete("/api/holders/wife/token", headers=AS_OPERATOR)
+    assert revoked.status_code == 200 and revoked.json()["holder_id"] == "wife"
+    assert client.post("/api/search", json={"query": KEYS}, headers={"Authorization": f"Bearer {token}"}).status_code == 401
+    assert client.get("/api/holders/wife", headers=AS_OPERATOR).json()["token_issued_at"] is None
+
+
+def test_a_working_set_refuses_to_belong_to_nobody(client: TestClient, settings: Settings) -> None:
+    _remember(settings, KEYS, "wife", session_id="kitchen-2026")
+    _remember(settings, THERMOSTAT, "husband", session_id="kitchen-2026")
+
+    anonymous = client.get("/api/memory/working", params={"session": "kitchen-2026"})
+    assert anonymous.status_code == 422
+    assert "somebody" in str(anonymous.json()["detail"])
+
+    wife = _token_for(client, "wife")
+    hers = client.get("/api/memory/working", params={"session": "kitchen-2026"}, headers={"Authorization": f"Bearer {wife['token']}"})
+    assert [atom["text"] for atom in hers.json()] == [KEYS], "a session sees its own lane, not the room's"
+
+    both = client.get("/api/memory/working", params={"session": "kitchen-2026", "holder": "husband"}, headers=AS_OPERATOR)
+    assert [atom["text"] for atom in both.json()] == [THERMOSTAT]
+
+
+def test_a_conflict_list_refuses_to_belong_to_nobody(client: TestClient, settings: Settings) -> None:
+    _remember(settings, "恒温器放在书房，设定 22 度", "wife")
+    _remember(settings, "恒温器调到 24 度", "wife")
+    assert client.get("/api/memory/conflicts").status_code == 422
+    wife = _token_for(client, "wife")
+    hers = client.get("/api/memory/conflicts", headers={"Authorization": f"Bearer {wife['token']}"})
+    assert hers.status_code == 200 and len(hers.json()) == 1
+
+
+def test_the_card_is_public_names_holders_and_carries_no_memory_text(client: TestClient, settings: Settings) -> None:
+    _remember(settings, KEYS, "wife", scenario="home")
+    _remember(settings, THERMOSTAT, "husband")
+    _token_for(client, "wife")
+
+    answer = client.get("/api/card")
+    assert answer.status_code == 200, "a capability card exists to be read by a stranger"
+    card = answer.json()
+    assert card["schema_version"] >= 11 and card["has_audio"] is False
+    named = {holder["holder_id"]: holder for holder in card["holders"]}
+    assert set(named) == {"wife", "husband"}
+    assert named["wife"]["memories"] == 1 and named["wife"]["token_issued"] is True
+    assert named["husband"]["token_issued"] is False
+    assert named["wife"]["scenarios"] == ["home"] and named["wife"]["modalities"] == {"text": 1}
+    assert named["wife"]["newest_memory"], "a stranger needs to know how stale the memory is"
+    assert KEYS not in answer.text and THERMOSTAT not in answer.text
+
+    one = client.get("/api/card/wife")
+    assert one.status_code == 200 and one.json()["holder_id"] == "wife"
+    assert client.get("/api/card/nobody").status_code == 404
+
+
+def test_a_picture_is_served_only_to_eyes_that_may_see_it(client: TestClient, settings: Settings) -> None:
+    database = Database(settings)
+    stored = ObjectStore(settings).put_bytes(PNG, ".png", "image")
+    database.upsert_sources(
+        [
+            SourceRecord(
+                source_id=source,
+                path=f"file:///{source}.png",
+                kind="image",
+                mime="image/png",
+                sha256=source * 8,
+                size_bytes=len(PNG),
+            )
+            for source in ("hers", "ours")
+        ]
+    )
+    database.upsert_atoms(
+        [
+            EvidenceAtom(
+                atom_id=f"{source}:frame",
+                modality=Modality.image,
+                source_id=source,
+                source=f"file:///{source}.png",
+                image_ref=stored,
+                trust=TrustScores(parser=1.0, ocr=0.0, model=0.0),
+                holder_id="wife" if source == "hers" else "",
+            )
+            for source in ("hers", "ours")
+        ]
+    )
+
+    assert client.get("/api/atoms/ours:frame/image").status_code == 200, "nobody owns it, so anybody may look"
+    assert client.get("/api/atoms/hers:frame/image").status_code == 404, "the bytes are behind the same rule as the words"
+    assert client.get("/api/atoms", params={"modality": "image"}).json()["total"] == 1
+    wife = _token_for(client, "wife")
+    page = client.get("/api/atoms", params={"modality": "image"}, headers={"Authorization": f"Bearer {wife['token']}"})
+    assert page.json()["total"] == 2
+
+
+def test_an_operator_and_a_stranger_do_not_share_one_compiled_context(settings: Settings) -> None:
+    _remember(settings, KEYS, "wife")
+    compiler = ContextCompiler(settings, Database(settings))
+    # The stranger compiles first: if the cache key ignored whose eyes were used, the operator would be served this.
+    stranger = compiler.compile(KEYS, RuntimeProfile.live_demo, as_holder="")
+    assert stranger.hits == [], "a memory nobody owns and nobody marked public is invisible to a stranger"
+
+    owner = compiler.compile(KEYS, RuntimeProfile.live_demo, as_holder="wife")
+    assert KEYS in owner.packed_text and owner.cache_trace_id != stranger.cache_trace_id
+
+    whole = compiler.compile(KEYS, RuntimeProfile.live_demo)
+    assert KEYS in whole.packed_text, "a process that owns the store still reads the whole library"

backend/tests/test_product_api.py

@@ -18,6 +18,8 @@ from ely_eye.storage import ObjectStore
 
 PNG = b"\x89PNG\r\n\x1a\n"
 BASE = datetime(2026, 5, 1, 12, 0, tzinfo=UTC)
+OPERATOR = "the-shelf-owner"
+AS_OWNER = {"Authorization": f"Bearer {OPERATOR}"}
 
 HTTP_SURFACE = {
     "/api/adapters",
@@ -32,11 +34,14 @@ HTTP_SURFACE = {
     "/api/cartridges/{cartridge_id}/refresh",
     "/api/cartridges/{cartridge_id}/switch",
     "/api/cartridges/{cartridge_id}/unmount",
+    "/api/card",
+    "/api/card/{holder_id}",
     "/api/chat",
     "/api/hashhop",
     "/api/holders",
     "/api/holders/{holder_id}",
     "/api/holders/{holder_id}/knows",
+    "/api/holders/{holder_id}/token",
     "/api/ingest/path",
     "/api/ingest/upload",
     "/api/memory",
@@ -57,7 +62,7 @@ HTTP_SURFACE = {
 
 @pytest.fixture
 def settings(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Settings:
-    resolved = Settings(home=tmp_path, default_holder="wife")
+    resolved = Settings(home=tmp_path, default_holder="wife", auth_token=OPERATOR)
     monkeypatch.setattr(api_module, "get_settings", lambda: resolved)
     Database(resolved).init()
     return resolved
@@ -106,14 +111,15 @@ def _seed(settings: Settings, pictures: int = 3) -> None:
 
 def test_the_store_pages_newest_first_and_says_how_much_is_left(client: TestClient, settings: Settings) -> None:
     _seed(settings)
-    first = client.get("/api/atoms", params={"limit": 2})
+    # A tokenless caller is nobody and would see none of these; this test reads the shelf as its owner.
+    first = client.get("/api/atoms", params={"limit": 2}, headers=AS_OWNER)
     assert first.status_code == 200
     page = first.json()
     assert page["total"] == 3
     assert [atom["atom_id"] for atom in page["atoms"]] == ["shot2:frame", "shot1:frame"]
     assert page["next_cursor"]
 
-    second = client.get("/api/atoms", params={"limit": 2, "cursor": page["next_cursor"]})
+    second = client.get("/api/atoms", params={"limit": 2, "cursor": page["next_cursor"]}, headers=AS_OWNER)
     rest = second.json()
     assert [atom["atom_id"] for atom in rest["atoms"]] == ["shot0:frame"]
     assert rest["next_cursor"] is None, "the last page must not promise another one"
@@ -125,11 +131,11 @@ def test_browse_narrows_by_modality_and_by_holder(client: TestClient, settings:
         "the keys are on the shelf", holder_id="wife", origin=MemoryOrigin.self_
     )
 
-    images = client.get("/api/atoms", params={"modality": "image"}).json()
+    images = client.get("/api/atoms", params={"modality": "image"}, headers=AS_OWNER).json()
     assert images["total"] == 2
     assert {atom["modality"] for atom in images["atoms"]} == {"image"}
 
-    his = client.get("/api/atoms", params={"holder": "husband"}).json()
+    his = client.get("/api/atoms", params={"holder": "husband"}, headers=AS_OWNER).json()
     assert [atom["atom_id"] for atom in his["atoms"]] == ["shot1:frame"]
 
 
@@ -142,7 +148,7 @@ def test_an_unreadable_cursor_is_refused_with_400(client: TestClient, settings:
 
 def test_a_picture_atom_serves_the_bytes_behind_it(client: TestClient, settings: Settings) -> None:
     _seed(settings, pictures=1)
-    response = client.get("/api/atoms/shot0:frame/image")
+    response = client.get("/api/atoms/shot0:frame/image", headers=AS_OWNER)
     assert response.status_code == 200
     assert response.content == PNG
     assert response.headers["content-type"] == "image/png"
@@ -152,15 +158,15 @@ def test_a_memory_with_no_picture_says_so(client: TestClient, settings: Settings
     atom = MemoryService(settings, Database(settings)).remember(
         "the keys are on the shelf", holder_id="wife", origin=MemoryOrigin.self_
     )
-    assert client.get(f"/api/atoms/{atom.atom_id}/image").status_code == 404
-    assert client.get("/api/atoms/nope:image").status_code == 404
+    assert client.get(f"/api/atoms/{atom.atom_id}/image", headers=AS_OWNER).status_code == 404
+    assert client.get("/api/atoms/nope:image", headers=AS_OWNER).status_code == 404
 
 
 def test_a_forgotten_picture_stops_being_served(client: TestClient, settings: Settings) -> None:
     _seed(settings, pictures=1)
     MemoryService(settings, Database(settings)).forget(atom_id="shot0:frame")
-    assert client.get("/api/atoms/shot0:frame/image").status_code == 404
-    assert client.get("/api/atoms").json()["total"] == 0
+    assert client.get("/api/atoms/shot0:frame/image", headers=AS_OWNER).status_code == 404
+    assert client.get("/api/atoms", headers=AS_OWNER).json()["total"] == 0
 
 
 def test_status_says_what_the_machine_can_see_and_what_it_cannot(client: TestClient, settings: Settings) -> None:

backend/tests/test_transactive.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import json
 from pathlib import Path
 from typing import Any
 
@@ -15,6 +16,7 @@ import ely_eye.cli as cli_module
 from ely_eye.cli import app
 from ely_eye.config import Settings
 from ely_eye.db import Database
+from ely_eye.holder_store import HolderStore
 from ely_eye.ingestion import IngestionService
 from ely_eye.memory import MemoryService
 from ely_eye.migrations import SCHEMA_VERSION
@@ -249,3 +251,45 @@ def test_the_cli_asks_and_reports_the_referral(tmp_path: Path, monkeypatch: pyte
     assert asked.exit_code == 0 and '"status": "referred"' in asked.stdout
     assert runner.invoke(app, ["holders"]).stdout.count("wife") >= 1
     assert runner.invoke(app, ["holder-set", "wife", "--contact", "not-a-pair"]).exit_code != 0
+
+
+def test_the_cli_mints_a_token_that_proves_who_is_asking(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    settings = Settings(home=tmp_path / "home", default_holder="husband")
+    monkeypatch.setattr(cli_module, "get_settings", lambda: settings)
+    runner = CliRunner()
+    Store(settings).wife_remembers(HOME_KEYS)
+
+    issued = json.loads(runner.invoke(app, ["holder-issue", "wife"]).stdout)
+    assert issued["holder_id"] == "wife" and issued["token"] and issued["issued_at"]
+    assert HolderStore(Database(settings)).verify_token(issued["token"]) == "wife"
+    assert issued["token"] not in runner.invoke(app, ["card"]).stdout, "the card is public and the token is not"
+
+    revoked = json.loads(runner.invoke(app, ["holder-revoke", "wife"]).stdout)
+    assert revoked["holder_id"] == "wife" and revoked["token_issued_at"] is None
+    assert HolderStore(Database(settings)).verify_token(issued["token"]) is None
+
+
+def test_the_card_is_read_back_from_the_store_rather_than_written_by_hand(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    settings = Settings(home=tmp_path / "home", default_holder="husband")
+    monkeypatch.setattr(cli_module, "get_settings", lambda: settings)
+    runner = CliRunner()
+    store = Store(settings)
+    store.wife_remembers(HOME_KEYS)
+    store.husband_remembers(WORK_KEYS)
+    store.register("neighbour", name="邻居", knows_about=["weather"])
+
+    answer = runner.invoke(app, ["card"])
+    assert answer.exit_code == 0, answer.stdout
+    card = json.loads(answer.stdout)
+    assert card["schema_version"] == SCHEMA_VERSION
+    assert card["atoms"] == 2 and card["sources"] == 2
+    assert card["modalities"] == {"text": 2} and card["has_audio"] is False
+    assert card["mounts"] == {"mounted": [], "scoped": False, "atoms_in_scope": 0}
+
+    named = {entry["holder_id"]: entry for entry in card["holders"]}
+    assert named["wife"]["memories"] == 1 and named["wife"]["scenarios"] == ["home"]
+    assert named["wife"]["newest_memory"], "a caller needs to know how stale the answer would be"
+    assert named["neighbour"]["memories"] == 0 and named["neighbour"]["knows_about"] == ["weather"]
+    assert HOME_KEYS not in answer.stdout, "the card says what a holder holds, never what they hold it to be"

backend/tests/test_visibility_rule.py

@@ -0,0 +1,46 @@
+"""The visibility rule is written out in five literal statements; this test keeps them one rule.
+
+The SQL gate refuses a concatenated statement and a view cannot take a parameter, so the clause
+is copied rather than composed. A copy that drifts is invisible to every other test — the search
+leg, the browse page and the picture route would each keep passing while disagreeing about who
+may see a memory.
+"""
+
+from __future__ import annotations
+
+import re
+from pathlib import Path
+
+SOURCE_DIR = Path(__file__).resolve().parents[1] / "src" / "ely_eye"
+FILES = ("browse.py", "store_search.py", "vector_index.py")
+CLAUSE = re.compile(r"\(\? IS NULL[^()]*?holder_id = ''.*?WHERE value = \?[^)]*\)\)\)\)", re.DOTALL)
+EXPECTED_COPIES = 5
+
+
+def _normalize(clause: str) -> str:
+    """Strip what the copies are allowed to differ in: whitespace, string-literal quotes, table aliases."""
+
+    return re.sub(r"\s+", "", clause).replace('"', "").replace("a.", "")
+
+
+def test_every_copy_of_the_visibility_rule_agrees() -> None:
+    found: list[tuple[str, str]] = [
+        (name, _normalize(match.group(0)))
+        for name in FILES
+        for match in CLAUSE.finditer((SOURCE_DIR / name).read_text(encoding="utf-8"))
+    ]
+
+    assert len(found) == EXPECTED_COPIES, [name for name, _ in found]
+    distinct = {clause for _, clause in found}
+    assert len(distinct) == 1, [name for name, clause in found if clause != next(iter(distinct))]
+
+
+def test_the_rule_still_carries_all_three_states() -> None:
+    match = CLAUSE.search((SOURCE_DIR / "store_search.py").read_text(encoding="utf-8"))
+    assert match is not None, "the words leg no longer carries a visibility clause"
+    clause = _normalize(match.group(0))
+
+    assert "?ISNULL" in clause, "None still reads the whole library"
+    assert "holder_id=''" in clause and "$.scope')='everyone'" in clause, "unowned and public rows stay open to a stranger"
+    assert "?<>''" in clause, "the empty string must stay nobody rather than matching every holder"
+    assert "holder_id=?" in clause and "value=?" in clause, "the owner and the visible_to list are still checked"