feat(book): tenant boundary and per-tenant numbering on the Book (T-7)

005 has never run in production (still at 004), so it is amended in place
rather than patched later by a 007: book_records gains owner_tenant (the
tatlock tenant, Caller.tenant from T-4) and number, a short outward id
allocated per tenant via a new book_tenant_counters row locked FOR UPDATE
in the same transaction as the insert — INSERT ... ON CONFLICT DO NOTHING
bootstraps the counter row so a tenant's very first allocation doesn't
race its own creation. Unique on (owner_tenant, number); composite index
on (owner_tenant, kind, status). 006 applies on top unchanged.

Every Book route now requires a tenant (BookService._require_tenant,
mirroring the Scribe's own pattern) and filters on it: a record in
another tenant answers exactly like one that does not exist, in both
directions. link and add's parent convenience both route through
_get_or_404, so a link can never reach another tenant — which is also
what keeps show's links/backlinks free of their own tenant filter, since
every link touching a tenant's record is created tenant-safe already.
Responses include number; list accepts number= to find one.

Verified: core_api_test downgraded to 004 (with the original migration,
to match its already-applied pre-amendment schema), amended, upgraded
back to head — clean round trip. 508 tests pass (489 baseline + 19 new:
3 migration-shape, 16 tenant/numbering). Mutation checks confirmed and
reverted: dropping list's tenant filter, allocating numbers from a global
counter, and dropping link's tenant check on its target each fail the
test built to catch it. Raw REST against a dev server on :8777 against
core_api_test confirms the same behaviour over the wire: two tenants'
first records both land on number 1, and a cross-tenant read answers 404
not_found.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

CHANGELOG.md

@@ -29,7 +29,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   `consumer_invalid`/`tenant_invalid`. Every other kind ignores both headers
   outright - `tatlock` and `scheduler` carry no tenant, and a `user` caller's
   tenant comes only from their own JWT's `tatlock_user` claim, never from a
-  header. Book routes do not filter by tenant yet (core-api T-2).
+  header.
 - **Caller auth** (`src/domains/auth/callers.py`) - the first thing in this repo
   that reads a service key. `require_caller()` resolves a static office/tatlock/
   scheduler key (constant-time compare, checked before any JWT parsing) or falls
@@ -40,7 +40,24 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   `find`, `change`, `close`, `link`/`unlink`, `tag`/`untag`, `history`, all under
   `/book/*`, behind `require_caller("office", "user")`. Migration 005 creates the
   whole schema: `book_records`, `book_tags`, `book_record_links`,
-  `book_attachments`, `book_events` (append-only, DB trigger-enforced).
+  `book_attachments`, `book_events` (append-only, DB trigger-enforced),
+  `book_tenant_counters`.
+- **The Book is behind the tenant boundary** (core-api T-7 / workspace D-73) -
+  every route now requires a tenant on the caller (400 `tenant_required` for
+  a `user`-kind caller with no `tatlock_user` claim; an `office` caller
+  already has one from every call per T-4) and filters on it: a record in
+  another tenant answers exactly like one that does not exist (404
+  `not_found`), in both directions, and `link`/`add`'s `parent` convenience
+  refuse a target in another tenant the same way. `add` sets `owner_tenant`
+  from the caller and allocates `number` - a short outward id numbered per
+  tenant (each tenant's Book starts at 1, unique on `(owner_tenant,
+  number)`, never reused), allocated via a per-tenant counter row
+  (`book_tenant_counters`) locked `FOR UPDATE` in the same transaction as
+  the insert. Responses include `number`; `list` accepts `number=` to find
+  one. Migration 005 (unreleased - production was still at 004) is amended
+  in place to add `book_records.owner_tenant`/`number` and
+  `book_tenant_counters`, rather than patching them in with a later
+  migration.
 
 ### Changed
 

alembic/env.py

@@ -20,7 +20,7 @@ from src.shared.database import Base
 from src.domains.auth.models import User, Role, UserRole, UserPreferences, ApiKey  # noqa: F401
 from src.domains.dashboard.models import QuickLink, DashboardWidget  # noqa: F401
 from src.domains.book.models import (  # noqa: F401
-    BookRecord, BookTag, BookRecordLink, BookAttachment, BookEvent,
+    BookRecord, BookTag, BookRecordLink, BookAttachment, BookEvent, BookTenantCounter,
 )
 from src.domains.scribe.models import (  # noqa: F401
     ScribeAddress, ScribeMessage, ScribeMailboxEntry,

alembic/versions/20261019_1400_005_create_book_tables.py

@@ -1,18 +1,22 @@
-"""Create Book tables (T-1, D-68/D-69/D-70)
+"""Create Book tables (T-1, D-68/D-69/D-70; tenant/number amended by T-7)
 
 Revision ID: 005
 Revises: 004
 Create Date: 2026-10-19
 
 The whole Book schema lands in this one migration — book_records,
-book_tags, book_record_links, book_attachments, book_events — so later
+book_tags, book_record_links, book_attachments, book_events,
-rounds (append, remind, reschedule, pause/resume, redact, ...) need no
+book_tenant_counters — so later rounds (append, remind, reschedule,
-migration of their own for these tables.
+pause/resume, redact, ...) need no migration of their own for these tables.
 
 book_events is append-only: a trigger raises on UPDATE and DELETE, for
 every role including the owning application. `redact` (D-70 rule 4) is the
 one sanctioned exception and is a later round; this migration does not
 soften the trigger.
+
+core-api T-7: amended before 005's first release (production was still at
+004) to add book_records.owner_tenant/number and book_tenant_counters,
+rather than patching them in with a later migration — see T-7's dossier.
 """
 from typing import Sequence, Union
 
@@ -34,6 +38,8 @@ def upgrade() -> None:
     op.create_table(
         "book_records",
         sa.Column("id", UUID(as_uuid=True), nullable=False),
+        sa.Column("owner_tenant", sa.String(length=63), nullable=False),
+        sa.Column("number", sa.Integer(), nullable=False),
         sa.Column("kind", sa.String(length=20), nullable=False),
         sa.Column("title", sa.String(length=500), nullable=True),
         sa.Column("body", sa.Text(), nullable=False, server_default=""),
@@ -65,10 +71,16 @@ def upgrade() -> None:
         sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
         sa.PrimaryKeyConstraint("id"),
         sa.UniqueConstraint("source", "external_id", name="uq_book_records_source_external_id"),
+        sa.UniqueConstraint("owner_tenant", "number", name="uq_book_records_tenant_number"),
     )
     op.create_index("ix_book_records_kind", "book_records", ["kind"])
     op.create_index("ix_book_records_status", "book_records", ["status"])
     op.create_index("ix_book_records_due_at", "book_records", ["due_at"])
+    op.create_index(
+        "ix_book_records_tenant_kind_status",
+        "book_records",
+        ["owner_tenant", "kind", "status"],
+    )
 
     # Generated column — 'simple' because the household writes mixed Dutch
     # and English, and a language-specific config would stem one of the two
@@ -85,6 +97,17 @@ def upgrade() -> None:
         "ix_book_records_search", "book_records", ["search"], postgresql_using="gin"
     )
 
+    # One row per tenant: the next `number` to allocate on book_records.
+    # Locked FOR UPDATE in the same transaction as the record insert it
+    # numbers (BookService._allocate_number) — see that method's docstring
+    # for why this shape is safe under concurrency.
+    op.create_table(
+        "book_tenant_counters",
+        sa.Column("owner_tenant", sa.String(length=63), nullable=False),
+        sa.Column("next_number", sa.Integer(), nullable=False, server_default="1"),
+        sa.PrimaryKeyConstraint("owner_tenant"),
+    )
+
     op.create_table(
         "book_tags",
         sa.Column("record_id", UUID(as_uuid=True), sa.ForeignKey("book_records.id", ondelete="CASCADE"), nullable=False),
@@ -165,7 +188,10 @@ def downgrade() -> None:
     op.drop_index("ix_book_tags_tag", table_name="book_tags")
     op.drop_table("book_tags")
 
+    op.drop_table("book_tenant_counters")
+
     op.drop_index("ix_book_records_search", table_name="book_records")
+    op.drop_index("ix_book_records_tenant_kind_status", table_name="book_records")
     op.drop_index("ix_book_records_due_at", table_name="book_records")
     op.drop_index("ix_book_records_status", table_name="book_records")
     op.drop_index("ix_book_records_kind", table_name="book_records")

src/domains/book/controller.py

@@ -3,7 +3,10 @@ Book Controller
 
 The core Book verbs (D-70): add, show, list, find, change, close,
 link/unlink, tag/untag, history. Every route requires a caller (office or
-user); writes additionally require an identity (T-1).
+user); writes additionally require an identity (T-1). Every route also
+requires a tenant on the caller — enforced in the service layer
+(`BookService._require_tenant`) rather than here, so reads and writes share
+one check (core-api T-7).
 """
 import uuid
 from typing import Optional
@@ -105,14 +108,19 @@ class BookController(BaseController):
             tag: Optional[str] = Query(None),
             scope: Optional[str] = Query(None),
             parent: Optional[uuid.UUID] = Query(None),
+            number: Optional[int] = Query(None, description="find the one record with this outward id"),
             limit: int = Query(50, ge=1, le=200),
             offset: int = Query(0, ge=0),
             caller: Caller = Depends(require_book_reader),
             session: AsyncSession = Depends(get_async_session),
         ):
-            results, total = await service.list_records(
+            try:
-                session, view=view, tag=tag, scope=scope, parent=parent, limit=limit, offset=offset
+                results, total = await service.list_records(
-            )
+                    session, caller, view=view, tag=tag, scope=scope, parent=parent,
+                    number=number, limit=limit, offset=offset,
+                )
+            except BookError as error:
+                raise HTTPException(status_code=error.status_code, detail={"code": error.code}) from error
             return BookRecordListResponse(
                 records=[_record_response(record, tags) for record, tags in results],
                 total=total,
@@ -136,7 +144,10 @@ class BookController(BaseController):
             caller: Caller = Depends(require_book_reader),
             session: AsyncSession = Depends(get_async_session),
         ):
-            results, total = await service.find_records(session, q=q, limit=limit, offset=offset)
+            try:
+                results, total = await service.find_records(session, caller, q=q, limit=limit, offset=offset)
+            except BookError as error:
+                raise HTTPException(status_code=error.status_code, detail={"code": error.code}) from error
             return BookRecordListResponse(
                 records=[_record_response(record, tags) for record, tags in results],
                 total=total,
@@ -159,7 +170,7 @@ class BookController(BaseController):
             session: AsyncSession = Depends(get_async_session),
         ):
             try:
-                record, tags, links, backlinks = await service.show_record(session, record_id)
+                record, tags, links, backlinks = await service.show_record(session, record_id, caller)
             except BookError as error:
                 raise HTTPException(status_code=error.status_code, detail={"code": error.code}) from error
             return _detail_response(record, tags, links, backlinks)
@@ -301,7 +312,7 @@ class BookController(BaseController):
             session: AsyncSession = Depends(get_async_session),
         ):
             try:
-                events = await service.get_history(session, record_id)
+                events = await service.get_history(session, record_id, caller)
             except BookError as error:
                 raise HTTPException(status_code=error.status_code, detail={"code": error.code}) from error
             return BookHistoryResponse(

src/domains/book/models.py

@@ -18,6 +18,7 @@ from sqlalchemy import (
     Computed,
     DateTime,
     ForeignKey,
+    Index,
     Integer,
     String,
     Text,
@@ -62,6 +63,8 @@ class BookRecord(Base):
     __tablename__ = "book_records"
     __table_args__ = (
         UniqueConstraint("source", "external_id", name="uq_book_records_source_external_id"),
+        UniqueConstraint("owner_tenant", "number", name="uq_book_records_tenant_number"),
+        Index("ix_book_records_tenant_kind_status", "owner_tenant", "kind", "status"),
     )
 
     id: Mapped[uuid.UUID] = mapped_column(
@@ -69,6 +72,16 @@ class BookRecord(Base):
         primary_key=True,
         default=uuid.uuid4,
     )
+    # The tatlock tenant the record belongs to (Caller.tenant, T-4/D-73).
+    # Every query in the service layer filters on it — there is no
+    # database-level RLS here, so that filter is load-bearing in Python,
+    # following src/domains/scribe/models.py's same note.
+    owner_tenant: Mapped[str] = mapped_column(String(63), nullable=False)
+    # A short outward id, numbered per tenant (never globally) so a tenant
+    # can never infer another's activity from its numbers. Allocated by
+    # BookService._allocate_number, which locks a per-tenant counter row
+    # FOR UPDATE in the same transaction as this row's insert.
+    number: Mapped[int] = mapped_column(Integer, nullable=False)
     kind: Mapped[str] = mapped_column(String(20), nullable=False, index=True)
     title: Mapped[Optional[str]] = mapped_column(String(500), nullable=True)
     body: Mapped[str] = mapped_column(Text, nullable=False, default="")
@@ -246,3 +259,26 @@ class BookEvent(Base):
 
     def __repr__(self) -> str:
         return f"<BookEvent {self.action} on {self.record_id} by {self.identity}>"
+
+
+class BookTenantCounter(Base):
+    """
+    One row per tenant: the next `number` to allocate on `book_records`.
+
+    `BookService._allocate_number` upserts this row (`INSERT ... ON
+    CONFLICT DO NOTHING`, so a tenant's first `add` never races its own
+    counter row into existence) and then locks it `FOR UPDATE` in the same
+    transaction as the record insert it numbers — a second concurrent `add`
+    in the same tenant blocks on that lock until the first commits, so two
+    concurrent creates never read the same `next_number`. A different
+    tenant's counter row is untouched, so tenants never contend with each
+    other.
+    """
+
+    __tablename__ = "book_tenant_counters"
+
+    owner_tenant: Mapped[str] = mapped_column(String(63), primary_key=True)
+    next_number: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
+
+    def __repr__(self) -> str:
+        return f"<BookTenantCounter {self.owner_tenant} next={self.next_number}>"

src/domains/book/schemas.py

@@ -126,9 +126,13 @@ class BookLinkResponse(BaseSchema):
 
 
 class BookRecordResponse(BaseSchema):
-    """Every write returns the stored record (D-70 rule 7)."""
+    """Every write returns the stored record (D-70 rule 7). `number` is the
+    outward id, numbered per tenant (core-api T-7); `owner_tenant` is not
+    echoed back — it is implicit in the caller, following
+    src/domains/scribe/schemas.py's AddressResponse."""
 
     id: uuid.UUID
+    number: int
     kind: str
     title: Optional[str]
     body: str

src/domains/book/service.py

@@ -6,6 +6,16 @@ Rejected options (D-68, D-70). Every mutating method stamps a `book_events`
 row via `_write_event`; the table itself is append-only (migration 005's
 trigger), so this module never updates or deletes an event.
 
+Every method starts by requiring a tenant on the caller (`_require_tenant`,
+same pattern as `src/domains/scribe/service.py`) and ends by filtering its
+query on it: **a record in another tenant answers exactly like one that
+does not exist** (core-api T-7 / workspace D-73), so no method here ever
+confirms an id across that boundary. `link` and `add`'s `parent` convenience
+both route through `_get_or_404`, so neither can create a link into another
+tenant — which is also what keeps `show`'s links/backlinks free of a
+separate tenant filter: every link touching a tenant's record is created
+tenant-safe in the first place.
+
 Methods take `session: AsyncSession` per call rather than storing it on
 `self`, following `src/domains/dashboard/service.py`.
 """
@@ -14,6 +24,7 @@ from typing import List, Optional, Tuple
 from zoneinfo import ZoneInfo
 
 from sqlalchemy import delete, func, select
+from sqlalchemy.dialects.postgresql import insert as pg_insert
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from src.domains.auth.callers import Caller
@@ -24,6 +35,7 @@ from src.domains.book.models import (
     BookRecord,
     BookRecordLink,
     BookTag,
+    BookTenantCounter,
 )
 from src.domains.book.schemas import BookCloseRequest, BookRecordCreate, BookRecordUpdate
 from src.shared.config import get_settings
@@ -48,12 +60,61 @@ class BookService:
     # Internal helpers
     # =========================================================================
 
-    async def _get_or_404(self, session: AsyncSession, record_id) -> BookRecord:
+    def _require_tenant(self, caller: Caller) -> str:
-        record = await session.get(BookRecord, record_id)
+        """Every Book operation needs a tenant; a caller without one gets a
+        clear error rather than a query that silently matches nothing. An
+        Office caller always has one (callers.py enforces it on every call);
+        a user-kind caller without a `tatlock_user` claim does not."""
+        if not caller.tenant:
+            raise BookError("tenant_required", 400)
+        return caller.tenant
+
+    async def _get_or_404(self, session: AsyncSession, record_id, tenant: str) -> BookRecord:
+        """A record outside `tenant` answers exactly like one that does not
+        exist — `session.get()` alone can't express that filter, since the
+        primary key is `id` only, so this is a `select` rather than a get."""
+        stmt = select(BookRecord).where(BookRecord.id == record_id, BookRecord.owner_tenant == tenant)
+        record = (await session.execute(stmt)).scalar_one_or_none()
         if record is None:
             raise BookError("not_found", 404)
         return record
 
+    async def _allocate_number(self, session: AsyncSession, tenant: str) -> int:
+        """
+        Allocate the next outward `number` for `tenant`, safe under
+        concurrency.
+
+        `INSERT ... ON CONFLICT DO NOTHING` guarantees the tenant's counter
+        row exists without racing its own creation (two simultaneous first
+        `add`s in a brand-new tenant can't both try to INSERT the same row
+        and fail), then `SELECT ... FOR UPDATE` locks that row for the rest
+        of this transaction — a second concurrent `add` in the same tenant
+        blocks on the lock until this one commits or rolls back, so two
+        concurrent creates in one tenant can never read the same
+        `next_number`. A different tenant's row is untouched, so tenants
+        never contend with each other. This is the dossier's suggested
+        shape (a per-tenant counter row taken FOR UPDATE in the same
+        transaction as the insert); the ON CONFLICT DO NOTHING bootstrap is
+        the one addition, needed because the plain shape alone still races
+        on a tenant's very first allocation.
+        """
+        upsert = (
+            pg_insert(BookTenantCounter)
+            .values(owner_tenant=tenant, next_number=1)
+            .on_conflict_do_nothing(index_elements=["owner_tenant"])
+        )
+        await session.execute(upsert)
+
+        locked = (
+            select(BookTenantCounter)
+            .where(BookTenantCounter.owner_tenant == tenant)
+            .with_for_update()
+        )
+        counter = (await session.execute(locked)).scalar_one()
+        number = counter.next_number
+        counter.next_number = number + 1
+        return number
+
     async def _tags_for(self, session: AsyncSession, record_id) -> List[str]:
         stmt = select(BookTag.tag).where(BookTag.record_id == record_id).order_by(BookTag.tag)
         result = await session.execute(stmt)
@@ -97,10 +158,15 @@ class BookService:
         data: BookRecordCreate,
         caller: Caller,
     ) -> Tuple[BookRecord, List[str]]:
-        """`add` — text-only input lands an unfiled Note (D-72)."""
+        """`add` — text-only input lands an unfiled Note (D-72). Sets
+        `owner_tenant` from the caller and allocates `number` (core-api T-7)."""
+        tenant = self._require_tenant(caller)
         status = self._validate_status(data.kind, data.status)
+        number = await self._allocate_number(session, tenant)
 
         record = BookRecord(
+            owner_tenant=tenant,
+            number=number,
             kind=data.kind,
             title=data.title,
             body=data.body,
@@ -132,6 +198,11 @@ class BookService:
             session.add(BookTag(record_id=record.id, tag=tag))
 
         if data.parent is not None:
+            # The parent is a link target (`relation="parent"`), so it goes
+            # through the same tenant-scoped lookup `link` uses — a parent
+            # in another tenant (or no such record at all) answers 404
+            # not_found, the same as an explicit `link` call would.
+            await self._get_or_404(session, data.parent, tenant)
             session.add(BookRecordLink(
                 from_id=record.id,
                 relation="parent",
@@ -143,16 +214,18 @@ class BookService:
         await session.commit()
         await session.refresh(record)
 
-        logger.info(f"Book record added: kind={data.kind} id={record.id} by={caller.identity}")
+        logger.info(f"Book record added: kind={data.kind} id={record.id} number={number} tenant={tenant} by={caller.identity}")
         return record, tag_names
 
     async def show_record(
         self,
         session: AsyncSession,
         record_id,
+        caller: Caller,
     ) -> Tuple[BookRecord, List[str], List[BookRecordLink], List[BookRecordLink]]:
         """`show` — the record, its tags, outgoing links and backlinks."""
-        record = await self._get_or_404(session, record_id)
+        tenant = self._require_tenant(caller)
+        record = await self._get_or_404(session, record_id, tenant)
         tags = await self._tags_for(session, record_id)
 
         links_stmt = select(BookRecordLink).where(BookRecordLink.from_id == record_id)
@@ -166,21 +239,26 @@ class BookService:
     async def list_records(
         self,
         session: AsyncSession,
+        caller: Caller,
         view: Optional[str] = None,
         tag: Optional[str] = None,
         scope: Optional[str] = None,
         parent: Optional[object] = None,
+        number: Optional[int] = None,
         limit: int = 50,
         offset: int = 0,
     ) -> Tuple[List[Tuple[BookRecord, List[str]]], int]:
         """
-        `list` — the nine views of D-70. Status-named views (`inbox`,
+        `list` — the nine views of D-70, scoped to the caller's tenant.
-        `waiting`, `someday`, `done`) filter on the matching task status and
+        Status-named views (`inbox`, `waiting`, `someday`, `done`) filter on
-        simply return nothing for kinds that never carry it. `today` and
+        the matching task status and simply return nothing for kinds that
-        `upcoming` bucket on `due_at` in `book_timezone`; richer due/overdue
+        never carry it. `today` and `upcoming` bucket on `due_at` in
-        handling is `due`/`next`, both next round.
+        `book_timezone`; richer due/overdue handling is `due`/`next`, both
+        next round. `number=` finds the one record with that outward id in
+        the caller's tenant (core-api T-7).
         """
-        conditions = []
+        tenant = self._require_tenant(caller)
+        conditions = [BookRecord.owner_tenant == tenant]
 
         if view in ("inbox", "waiting", "someday", "done"):
             conditions.append(BookRecord.status == view)
@@ -220,6 +298,9 @@ class BookService:
             tag_subquery = select(BookTag.record_id).where(BookTag.tag == tag)
             conditions.append(BookRecord.id.in_(tag_subquery))
 
+        if number is not None:
+            conditions.append(BookRecord.number == number)
+
         base_query = select(BookRecord).where(*conditions)
 
         count_query = select(func.count()).select_from(base_query.subquery())
@@ -243,13 +324,18 @@ class BookService:
     async def find_records(
         self,
         session: AsyncSession,
+        caller: Caller,
         q: str,
         limit: int = 50,
         offset: int = 0,
     ) -> Tuple[List[Tuple[BookRecord, List[str]]], int]:
-        """`find` — full-text search over title and body."""
+        """`find` — full-text search over title and body, scoped to the
+        caller's tenant."""
+        tenant = self._require_tenant(caller)
         tsquery = func.websearch_to_tsquery("simple", q)
-        base_query = select(BookRecord).where(BookRecord.search.op("@@")(tsquery))
+        base_query = select(BookRecord).where(
+            BookRecord.owner_tenant == tenant, BookRecord.search.op("@@")(tsquery)
+        )
 
         count_query = select(func.count()).select_from(base_query.subquery())
         total = (await session.execute(count_query)).scalar() or 0
@@ -276,7 +362,8 @@ class BookService:
         caller: Caller,
     ) -> Tuple[BookRecord, List[str]]:
         """`change` — a partial update; only fields set on the request move."""
-        record = await self._get_or_404(session, record_id)
+        tenant = self._require_tenant(caller)
+        record = await self._get_or_404(session, record_id, tenant)
 
         update_data = data.model_dump(exclude_unset=True)
         diff: dict = {}
@@ -321,8 +408,12 @@ class BookService:
         `close` — a Task is done or dropped; a Question is resolved into a
         Decision (`resolves` link, status `settled`); a Decision is
         superseded by a newer one (`supersedes` link, status `superseded`).
+        `resolved_by`/`superseded_by` are link targets, so each is looked up
+        in the caller's tenant the same as `link` — a target in another
+        tenant answers 404 not_found.
         """
-        record = await self._get_or_404(session, record_id)
+        tenant = self._require_tenant(caller)
+        record = await self._get_or_404(session, record_id, tenant)
         diff: dict = {}
 
         if record.kind == "task":
@@ -334,7 +425,7 @@ class BookService:
         elif record.kind == "question":
             if data.resolved_by is None:
                 raise BookError("invalid_status", 400)
-            decision = await self._get_or_404(session, data.resolved_by)
+            decision = await self._get_or_404(session, data.resolved_by, tenant)
             if decision.kind != "decision":
                 raise BookError("invalid_status", 400)
             session.add(BookRecordLink(
@@ -349,7 +440,7 @@ class BookService:
         elif record.kind == "decision":
             if data.superseded_by is None:
                 raise BookError("invalid_status", 400)
-            new_decision = await self._get_or_404(session, data.superseded_by)
+            new_decision = await self._get_or_404(session, data.superseded_by, tenant)
             if new_decision.kind != "decision":
                 raise BookError("invalid_status", 400)
             session.add(BookRecordLink(
@@ -386,8 +477,11 @@ class BookService:
         relation: str,
         caller: Caller,
     ) -> Tuple[BookRecord, List[str]]:
-        record = await self._get_or_404(session, record_id)
+        """`link` refuses a target in another tenant the same way a missing
-        await self._get_or_404(session, to_id)  # the target must exist
+        one is refused — both are 404 not_found (core-api T-7)."""
+        tenant = self._require_tenant(caller)
+        record = await self._get_or_404(session, record_id, tenant)
+        await self._get_or_404(session, to_id, tenant)  # the target must exist in the same tenant
 
         session.add(BookRecordLink(
             from_id=record_id,
@@ -412,7 +506,8 @@ class BookService:
         relation: str,
         caller: Caller,
     ) -> Tuple[BookRecord, List[str]]:
-        record = await self._get_or_404(session, record_id)
+        tenant = self._require_tenant(caller)
+        record = await self._get_or_404(session, record_id, tenant)
 
         stmt = delete(BookRecordLink).where(
             BookRecordLink.from_id == record_id,
@@ -443,7 +538,8 @@ class BookService:
         tag: str,
         caller: Caller,
     ) -> Tuple[BookRecord, List[str]]:
-        record = await self._get_or_404(session, record_id)
+        tenant = self._require_tenant(caller)
+        record = await self._get_or_404(session, record_id, tenant)
 
         existing = await session.get(BookTag, (record_id, tag))
         if existing is None:
@@ -461,7 +557,8 @@ class BookService:
         tag: str,
         caller: Caller,
     ) -> Tuple[BookRecord, List[str]]:
-        record = await self._get_or_404(session, record_id)
+        tenant = self._require_tenant(caller)
+        record = await self._get_or_404(session, record_id, tenant)
 
         stmt = delete(BookTag).where(BookTag.record_id == record_id, BookTag.tag == tag)
         result = await session.execute(stmt)
@@ -476,8 +573,11 @@ class BookService:
     # history
     # =========================================================================
 
-    async def get_history(self, session: AsyncSession, record_id, limit: int = 100) -> List[BookEvent]:
+    async def get_history(
-        await self._get_or_404(session, record_id)
+        self, session: AsyncSession, record_id, caller: Caller, limit: int = 100
+    ) -> List[BookEvent]:
+        tenant = self._require_tenant(caller)
+        await self._get_or_404(session, record_id, tenant)
 
         stmt = (
             select(BookEvent)

tests/conftest.py

@@ -114,7 +114,10 @@ async def clean_book_tables(book_session):
     from sqlalchemy import text
 
     await book_session.execute(
-        text("TRUNCATE book_events, book_attachments, book_record_links, book_tags, book_records")
+        text(
+            "TRUNCATE book_events, book_attachments, book_record_links, book_tags, "
+            "book_records, book_tenant_counters"
+        )
     )
     await book_session.commit()
     yield book_session

tests/test_book.py

@@ -1,12 +1,14 @@
 """
-Tests for the Book domain (T-1 / D-68, D-70) against `core_api_test`.
+Tests for the Book domain (T-1 / D-68, D-70, T-7) against `core_api_test`.
 
 `get_async_session` is overridden to the test database for every test here
 — the Book routes list `caller` before `session` in their dependencies, but
 overriding is what actually guarantees the suite never opens a connection
 to production regardless of FastAPI's dependency resolution order.
 """
+import asyncio
 import uuid
+from unittest.mock import patch
 
 import pytest
 from fastapi.testclient import TestClient
@@ -18,6 +20,8 @@ from src.shared.database import get_async_session
 
 settings = get_settings()
 
+OTHER_TENANT = "llm_tester"
+
 OFFICE_KEY = "test-office-key"
 
 
@@ -485,3 +489,235 @@ class TestHistory:
         ).json()
         assert history["events"][0]["action"] == "add"
         assert history["events"][0]["client"] == "office-gemini"
+
+
+class TestNumbering:
+    """core-api T-7: `number` is allocated per tenant, never globally."""
+
+    def test_numbering_starts_at_one_per_tenant_and_increments(self, client):
+        first = client.post(
+            "/book/records", json={"body": "a"}, headers=office_headers(tenant=VALID_TENANT)
+        ).json()
+        second = client.post(
+            "/book/records", json={"body": "b"}, headers=office_headers(tenant=VALID_TENANT)
+        ).json()
+        other_tenant_first = client.post(
+            "/book/records", json={"body": "c"}, headers=office_headers(tenant=OTHER_TENANT)
+        ).json()
+
+        assert first["number"] == 1
+        assert second["number"] == 2
+        # A different tenant starts its own sequence at 1 — it can never
+        # infer VALID_TENANT's activity from its own numbers.
+        assert other_tenant_first["number"] == 1
+
+    def test_list_number_finds_the_record(self, client):
+        record = client.post(
+            "/book/records", json={"body": "find me"}, headers=office_headers()
+        ).json()
+        client.post("/book/records", json={"body": "noise"}, headers=office_headers())
+
+        found = client.get(
+            "/book/records", params={"number": record["number"]}, headers=office_headers()
+        ).json()
+        assert found["total"] == 1
+        assert found["records"][0]["id"] == record["id"]
+
+    @pytest.mark.asyncio
+    async def test_two_concurrent_creates_in_one_tenant_get_distinct_numbers(
+        self, clean_book_tables, test_database_url
+    ):
+        """Direct service calls on two independent sessions, run
+        concurrently via asyncio.gather, rather than through TestClient
+        (whose calls are issued one at a time and so would never actually
+        overlap at the database). Exercises BookService._allocate_number's
+        FOR UPDATE lock for real."""
+        from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+        from sqlalchemy.pool import NullPool
+
+        from src.domains.auth.callers import Caller
+        from src.domains.book.schemas import BookRecordCreate
+        from src.domains.book.service import get_book_service
+
+        asyncpg_url = test_database_url.replace("postgresql://", "postgresql+asyncpg://", 1)
+        service = get_book_service()
+        caller = Caller(kind="office", identity="jeroen", consumer="office-claude", tenant=VALID_TENANT)
+
+        async def _create(body: str) -> int:
+            engine = create_async_engine(asyncpg_url, poolclass=NullPool)
+            session_factory = async_sessionmaker(bind=engine, expire_on_commit=False)
+            try:
+                async with session_factory() as session:
+                    record, _ = await service.add_record(session, BookRecordCreate(body=body), caller)
+                    return record.number
+            finally:
+                await engine.dispose()
+
+        numbers = await asyncio.gather(_create("one"), _create("two"))
+        assert sorted(numbers) == [1, 2]
+
+
+class TestTenantIsolation:
+    """core-api T-7 / workspace D-73: a record in another tenant answers
+    exactly like one that does not exist, checked in both directions."""
+
+    def test_llm_tester_cannot_list_jpmschweitzer_records(self, client):
+        client.post(
+            "/book/records", json={"body": "secret"}, headers=office_headers(tenant=VALID_TENANT)
+        )
+        as_other = client.get("/book/records", headers=office_headers(tenant=OTHER_TENANT)).json()
+        assert as_other["total"] == 0
+
+    def test_llm_tester_cannot_show_a_jpmschweitzer_record(self, client):
+        record = client.post(
+            "/book/records", json={"body": "secret"}, headers=office_headers(tenant=VALID_TENANT)
+        ).json()
+        response = client.get(
+            f"/book/records/{record['id']}", headers=office_headers(tenant=OTHER_TENANT)
+        )
+        assert response.status_code == 404
+        assert response.json()["detail"]["code"] == "not_found"
+
+    def test_jpmschweitzer_cannot_show_an_llm_tester_record(self, client):
+        """The reverse direction of the previous test — the boundary is
+        symmetric, not a one-way allowlist for the production tenant."""
+        record = client.post(
+            "/book/records", json={"body": "secret"}, headers=office_headers(tenant=OTHER_TENANT)
+        ).json()
+        response = client.get(
+            f"/book/records/{record['id']}", headers=office_headers(tenant=VALID_TENANT)
+        )
+        assert response.status_code == 404
+        assert response.json()["detail"]["code"] == "not_found"
+
+    def test_llm_tester_cannot_change_a_jpmschweitzer_record(self, client):
+        record = client.post(
+            "/book/records", json={"body": "secret"}, headers=office_headers(tenant=VALID_TENANT)
+        ).json()
+        response = client.patch(
+            f"/book/records/{record['id']}",
+            json={"priority": 1},
+            headers=office_headers(tenant=OTHER_TENANT),
+        )
+        assert response.status_code == 404
+        assert response.json()["detail"]["code"] == "not_found"
+
+    def test_llm_tester_cannot_close_a_jpmschweitzer_record(self, client):
+        task = client.post(
+            "/book/records",
+            json={"kind": "task", "body": "secret"},
+            headers=office_headers(tenant=VALID_TENANT),
+        ).json()
+        response = client.post(
+            f"/book/records/{task['id']}/close",
+            json={"outcome": "done"},
+            headers=office_headers(tenant=OTHER_TENANT),
+        )
+        assert response.status_code == 404
+        assert response.json()["detail"]["code"] == "not_found"
+
+    def test_llm_tester_cannot_tag_a_jpmschweitzer_record(self, client):
+        record = client.post(
+            "/book/records", json={"body": "secret"}, headers=office_headers(tenant=VALID_TENANT)
+        ).json()
+        response = client.post(
+            f"/book/records/{record['id']}/tag",
+            json={"tag": "chore"},
+            headers=office_headers(tenant=OTHER_TENANT),
+        )
+        assert response.status_code == 404
+        assert response.json()["detail"]["code"] == "not_found"
+
+    def test_llm_tester_cannot_link_to_a_jpmschweitzer_record(self, client):
+        """`link` refuses a target in another tenant the same way (dossier)."""
+        a = client.post(
+            "/book/records", json={"body": "a"}, headers=office_headers(tenant=OTHER_TENANT)
+        ).json()
+        b = client.post(
+            "/book/records", json={"body": "b"}, headers=office_headers(tenant=VALID_TENANT)
+        ).json()
+        response = client.post(
+            f"/book/records/{a['id']}/link",
+            json={"to_id": b["id"], "relation": "related"},
+            headers=office_headers(tenant=OTHER_TENANT),
+        )
+        assert response.status_code == 404
+        assert response.json()["detail"]["code"] == "not_found"
+
+    def test_llm_tester_cannot_read_history_of_a_jpmschweitzer_record(self, client):
+        record = client.post(
+            "/book/records", json={"body": "secret"}, headers=office_headers(tenant=VALID_TENANT)
+        ).json()
+        response = client.get(
+            f"/book/records/{record['id']}/history", headers=office_headers(tenant=OTHER_TENANT)
+        )
+        assert response.status_code == 404
+        assert response.json()["detail"]["code"] == "not_found"
+
+    def test_add_with_a_parent_in_another_tenant_is_404_and_nothing_persists(self, client):
+        """The `parent` convenience link in `add` is a link target too, so
+        it is refused the same way `link` is. The attempted child must not
+        persist either — the whole add is one transaction."""
+        parent = client.post(
+            "/book/records", json={"body": "parent"}, headers=office_headers(tenant=OTHER_TENANT)
+        ).json()
+        response = client.post(
+            "/book/records",
+            json={"body": "child", "parent": parent["id"]},
+            headers=office_headers(tenant=VALID_TENANT),
+        )
+        assert response.status_code == 404
+        assert response.json()["detail"]["code"] == "not_found"
+
+        listing = client.get("/book/records", headers=office_headers(tenant=VALID_TENANT)).json()
+        assert listing["total"] == 0
+
+    def test_find_does_not_leak_a_jpmschweitzer_record_to_llm_tester(self, client):
+        client.post(
+            "/book/records",
+            json={"body": "unique-secret-garden-term"},
+            headers=office_headers(tenant=VALID_TENANT),
+        )
+        found = client.get(
+            "/book/records/find",
+            params={"q": "unique-secret-garden-term"},
+            headers=office_headers(tenant=OTHER_TENANT),
+        ).json()
+        assert found["total"] == 0
+
+
+class TestUserCallerTenantRequired:
+    """A user-kind caller without a `tatlock_user` claim gets 400
+    tenant_required — the Office caller gets this for free from
+    callers.py, but a "user"-kind caller's tenant is optional there, so
+    Book's own `_require_tenant` is what actually enforces it (dossier
+    verification #2)."""
+
+    def test_user_without_tenant_claim_is_400_on_read(self, client):
+        with patch("src.domains.auth.callers.get_current_user") as mock_get_current_user:
+            mock_get_current_user.return_value = {"preferred_username": "jeroen"}
+            response = client.get("/book/records", headers={"Authorization": "Bearer some-jwt"})
+        assert response.status_code == 400
+        assert response.json()["detail"]["code"] == "tenant_required"
+
+    def test_user_without_tenant_claim_is_400_on_write(self, client):
+        with patch("src.domains.auth.callers.get_current_user") as mock_get_current_user:
+            mock_get_current_user.return_value = {"preferred_username": "jeroen"}
+            response = client.post(
+                "/book/records", json={"body": "hi"}, headers={"Authorization": "Bearer some-jwt"}
+            )
+        assert response.status_code == 400
+        assert response.json()["detail"]["code"] == "tenant_required"
+
+    def test_user_with_tenant_claim_succeeds(self, client):
+        with patch("src.domains.auth.callers.get_current_user") as mock_get_current_user:
+            mock_get_current_user.return_value = {
+                "preferred_username": "jeroen",
+                "tatlock_user": VALID_TENANT,
+            }
+            response = client.post(
+                "/book/records", json={"body": "hi"}, headers={"Authorization": "Bearer some-jwt"}
+            )
+        assert response.status_code == 201
+        assert response.json()["created_by"] == "jeroen"
+        assert response.json()["number"] == 1

tests/test_book_migration.py

@@ -28,6 +28,17 @@ async def _table_exists(url: str, table: str) -> bool:
         await engine.dispose()
 
 
+async def _scalar(url: str, query: str, **params) -> object:
+    asyncpg_url = url.replace("postgresql://", "postgresql+asyncpg://", 1)
+    engine = create_async_engine(asyncpg_url, poolclass=NullPool)
+    try:
+        async with engine.connect() as conn:
+            result = await conn.execute(text(query), params)
+            return result.scalar()
+    finally:
+        await engine.dispose()
+
+
 class TestMigration005RoundTrip:
     """Migration 005 must upgrade and downgrade cleanly (dossier verification #2)."""
 
@@ -36,7 +47,10 @@ class TestMigration005RoundTrip:
         result = run_alembic(test_database_url, "upgrade", "head")
         assert result.returncode == 0, result.stderr
 
-        for table in ("book_records", "book_tags", "book_record_links", "book_attachments", "book_events"):
+        for table in (
+            "book_records", "book_tags", "book_record_links", "book_attachments",
+            "book_events", "book_tenant_counters",
+        ):
             assert await _table_exists(test_database_url, table), f"{table} missing after upgrade"
 
     @pytest.mark.asyncio
@@ -49,12 +63,48 @@ class TestMigration005RoundTrip:
         assert down.returncode == 0, down.stderr
         assert not await _table_exists(test_database_url, "book_records")
         assert not await _table_exists(test_database_url, "book_events")
+        assert not await _table_exists(test_database_url, "book_tenant_counters")
 
         # Leave the schema at head for the rest of the suite.
         up = run_alembic(test_database_url, "upgrade", "005")
         assert up.returncode == 0, up.stderr
         assert await _table_exists(test_database_url, "book_records")
         assert await _table_exists(test_database_url, "book_events")
+        assert await _table_exists(test_database_url, "book_tenant_counters")
+
+
+class TestMigration005TenantAndNumber:
+    """core-api T-7: owner_tenant/number amended onto 005 before its first release."""
+
+    @pytest.mark.asyncio
+    async def test_book_records_owner_tenant_and_number_are_not_null(self, test_database_url):
+        run_alembic(test_database_url, "upgrade", "head")
+        for column in ("owner_tenant", "number"):
+            nullable = await _scalar(
+                test_database_url,
+                "SELECT is_nullable FROM information_schema.columns "
+                "WHERE table_name = 'book_records' AND column_name = :column",
+                column=column,
+            )
+            assert nullable == "NO", f"book_records.{column} must be NOT NULL, got {nullable!r}"
+
+    @pytest.mark.asyncio
+    async def test_book_records_has_tenant_number_unique_constraint(self, test_database_url):
+        run_alembic(test_database_url, "upgrade", "head")
+        found = await _scalar(
+            test_database_url,
+            "SELECT conname FROM pg_constraint WHERE conname = 'uq_book_records_tenant_number'",
+        )
+        assert found == "uq_book_records_tenant_number"
+
+    @pytest.mark.asyncio
+    async def test_book_records_has_tenant_kind_status_index(self, test_database_url):
+        run_alembic(test_database_url, "upgrade", "head")
+        found = await _scalar(
+            test_database_url,
+            "SELECT indexname FROM pg_indexes WHERE indexname = 'ix_book_records_tenant_kind_status'",
+        )
+        assert found == "ix_book_records_tenant_kind_status"
 
 
 class TestBookEventsAppendOnly:
@@ -65,7 +115,10 @@ class TestBookEventsAppendOnly:
         from src.domains.book.models import BookEvent, BookRecord
 
         session = clean_book_tables
-        record = BookRecord(kind="note", body="trigger test", status="active", created_by="test-harness")
+        record = BookRecord(
+            owner_tenant="jpmschweitzer", number=1,
+            kind="note", body="trigger test", status="active", created_by="test-harness",
+        )
         session.add(record)
         await session.flush()
         event = BookEvent(record_id=record.id, identity="test-harness", client=None, action="add", diff={})
@@ -85,7 +138,10 @@ class TestBookEventsAppendOnly:
         from src.domains.book.models import BookEvent, BookRecord
 
         session = clean_book_tables
-        record = BookRecord(kind="note", body="trigger test", status="active", created_by="test-harness")
+        record = BookRecord(
+            owner_tenant="jpmschweitzer", number=1,
+            kind="note", body="trigger test", status="active", created_by="test-harness",
+        )
         session.add(record)
         await session.flush()
         event = BookEvent(record_id=record.id, identity="test-harness", client=None, action="add", diff={})