feat(book): add the Book domain — notes, tasks, decisions, questions, rejected (T-1)

The Office (workspace D-64..D-72) needs a store for Notes, Tasks,
Decisions, Questions and Rejected options. D-68 puts it here, in
core-api's own database, behind its own auth.

Migration 005 lands the whole schema now, by hand (down_revision="004";
the chain has a hash-named revision between 003 and 004, confirmed
before trusting the numbering): book_records (kind, status validated
per kind in the service, scope work/personal, a generated tsvector
search column + GIN index), book_tags, book_record_links, book_attachments
(table only, no file handling yet), and book_events — append-only via a
database trigger that raises on UPDATE and DELETE for every role,
including the owning application.

Routes this round (D-70's core verbs), every one behind
require_caller("office", "user"), writes with identity_required=True,
every change writing a book_events row: add, show, list (the nine
views), find (full-text), change, close (task done/dropped; question
resolved via a `resolves` link into `settled`; decision superseded via
a `supersedes` link into `superseded`), link/unlink, tag/untag, history.

Append, remind, reschedule, undo, purge, redact, pin, check, snooze,
archive/trash/restore, pause/resume, due, next, batch calls and
idempotency keys are explicitly not this round — the schema already
carries what they need.

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

alembic/env.py

@@ -19,6 +19,9 @@ from src.shared.database import Base
 # Import all models to ensure they're registered with Base.metadata
 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,
+)
 
 # Alembic Config object
 config = context.config

alembic/versions/20261019_1400_005_create_book_tables.py

@@ -0,0 +1,172 @@
+"""Create Book tables (T-1, D-68/D-69/D-70)
+
+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
+rounds (append, remind, reschedule, 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.
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects.postgresql import JSONB, TSVECTOR, UUID
+
+
+# revision identifiers, used by Alembic.
+revision: str = "005"
+down_revision: Union[str, None] = "004"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Create the Book tables, its search column/index, and the append-only trigger."""
+
+    op.create_table(
+        "book_records",
+        sa.Column("id", UUID(as_uuid=True), 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=""),
+        sa.Column("checklist", JSONB(), nullable=False, server_default=sa.text("'[]'::jsonb")),
+        sa.Column("status", sa.String(length=20), nullable=False),
+        sa.Column("scope", sa.String(length=10), nullable=False, server_default="personal"),
+        sa.Column("priority", sa.Integer(), nullable=True),
+        sa.Column("pinned", sa.Boolean(), nullable=False, server_default=sa.false()),
+        sa.Column("energy", sa.String(length=10), nullable=True),
+        sa.Column("estimate_minutes", sa.Integer(), nullable=True),
+        sa.Column("next_step", sa.Text(), nullable=True),
+        sa.Column("start_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("due_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("all_day", sa.Boolean(), nullable=False, server_default=sa.false()),
+        sa.Column("remind_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("recurrence", sa.String(length=255), nullable=True),
+        sa.Column("recurrence_mode", sa.String(length=20), nullable=True),
+        sa.Column("remind_to", sa.String(length=63), nullable=True),
+        sa.Column("last_touched_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+        sa.Column("archived_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("trashed_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("paused_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("paused_statuses", JSONB(), nullable=True),
+        sa.Column("source", sa.String(length=50), nullable=True),
+        sa.Column("external_id", sa.String(length=255), nullable=True),
+        sa.Column("colour", sa.String(length=20), nullable=True),
+        sa.Column("created_by", sa.String(length=63), nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+        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"),
+    )
+    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"])
+
+    # Generated column — 'simple' because the household writes mixed Dutch
+    # and English, and a language-specific config would stem one of the two
+    # badly. Added via raw SQL: Alembic's op.add_column has no first-class
+    # GENERATED ALWAYS AS support.
+    op.execute(
+        """
+        ALTER TABLE book_records
+        ADD COLUMN search tsvector
+        GENERATED ALWAYS AS (to_tsvector('simple', coalesce(title, '') || ' ' || body)) STORED
+        """
+    )
+    op.create_index(
+        "ix_book_records_search", "book_records", ["search"], postgresql_using="gin"
+    )
+
+    op.create_table(
+        "book_tags",
+        sa.Column("record_id", UUID(as_uuid=True), sa.ForeignKey("book_records.id", ondelete="CASCADE"), nullable=False),
+        sa.Column("tag", sa.String(length=63), nullable=False),
+        sa.PrimaryKeyConstraint("record_id", "tag"),
+    )
+    op.create_index("ix_book_tags_tag", "book_tags", ["tag"])
+
+    op.create_table(
+        "book_record_links",
+        sa.Column("from_id", UUID(as_uuid=True), sa.ForeignKey("book_records.id", ondelete="CASCADE"), nullable=False),
+        sa.Column("relation", sa.String(length=20), nullable=False),
+        sa.Column("to_id", UUID(as_uuid=True), sa.ForeignKey("book_records.id", ondelete="CASCADE"), nullable=False),
+        sa.Column("created_by", sa.String(length=63), nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+        sa.PrimaryKeyConstraint("from_id", "relation", "to_id"),
+    )
+    op.create_index("ix_book_record_links_to_id", "book_record_links", ["to_id"])
+
+    op.create_table(
+        "book_attachments",
+        sa.Column("id", UUID(as_uuid=True), nullable=False),
+        sa.Column("record_id", UUID(as_uuid=True), sa.ForeignKey("book_records.id", ondelete="CASCADE"), nullable=False),
+        sa.Column("path", sa.String(length=1000), nullable=False),
+        sa.Column("mime", sa.String(length=255), nullable=True),
+        sa.Column("size", sa.Integer(), nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index("ix_book_attachments_record_id", "book_attachments", ["record_id"])
+
+    op.create_table(
+        "book_events",
+        sa.Column("id", UUID(as_uuid=True), nullable=False),
+        sa.Column("record_id", UUID(as_uuid=True), sa.ForeignKey("book_records.id", ondelete="CASCADE"), nullable=False),
+        sa.Column("at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+        sa.Column("identity", sa.String(length=63), nullable=False),
+        sa.Column("client", sa.String(length=63), nullable=True),
+        sa.Column("action", sa.String(length=20), nullable=False),
+        sa.Column("diff", JSONB(), nullable=False, server_default=sa.text("'{}'::jsonb")),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index("ix_book_events_record_id", "book_events", ["record_id"])
+
+    # Append-only, for every role including the owning application.
+    op.execute(
+        """
+        CREATE FUNCTION book_events_append_only() RETURNS trigger AS $$
+        BEGIN
+            RAISE EXCEPTION 'book_events is append-only: % not allowed', TG_OP;
+        END;
+        $$ LANGUAGE plpgsql
+        """
+    )
+    op.execute(
+        """
+        CREATE TRIGGER book_events_no_update_delete
+        BEFORE UPDATE OR DELETE ON book_events
+        FOR EACH ROW EXECUTE FUNCTION book_events_append_only()
+        """
+    )
+
+
+def downgrade() -> None:
+    """Drop the Book tables, in dependency order."""
+    op.execute("DROP TRIGGER IF EXISTS book_events_no_update_delete ON book_events")
+    op.execute("DROP FUNCTION IF EXISTS book_events_append_only()")
+
+    op.drop_index("ix_book_events_record_id", table_name="book_events")
+    op.drop_table("book_events")
+
+    op.drop_index("ix_book_attachments_record_id", table_name="book_attachments")
+    op.drop_table("book_attachments")
+
+    op.drop_index("ix_book_record_links_to_id", table_name="book_record_links")
+    op.drop_table("book_record_links")
+
+    op.drop_index("ix_book_tags_tag", table_name="book_tags")
+    op.drop_table("book_tags")
+
+    op.drop_index("ix_book_records_search", 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")
+    op.drop_table("book_records")

src/domains/book/__init__.py

@@ -0,0 +1,9 @@
+"""
+Book Domain
+
+The household's Notes, Tasks, Decisions, Questions and Rejected options
+(workspace D-68, D-69, D-70). Served under /book/*.
+"""
+from src.domains.book.controller import book_controller
+
+__all__ = ["book_controller"]

src/domains/book/controller.py

@@ -0,0 +1,316 @@
+"""
+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).
+"""
+import uuid
+from typing import Optional
+
+from fastapi import APIRouter, Depends, HTTPException, Query
+
+from src.domains.auth.callers import Caller, require_caller
+from src.domains.book.schemas import (
+    BookCloseRequest,
+    BookEventResponse,
+    BookHistoryResponse,
+    BookLinkRequest,
+    BookLinkResponse,
+    BookRecordCreate,
+    BookRecordDetailResponse,
+    BookRecordListResponse,
+    BookRecordResponse,
+    BookRecordUpdate,
+    BookTagRequest,
+    ListView,
+)
+from src.domains.book.service import BookError, get_book_service
+from src.shared.base import BaseController
+from src.shared.database import get_async_session
+from src.shared.logging import get_logger
+from sqlalchemy.ext.asyncio import AsyncSession
+
+logger = get_logger(__name__)
+
+require_book_reader = require_caller("office", "user")
+require_book_writer = require_caller("office", "user", identity_required=True)
+
+
+def _record_response(record, tags) -> BookRecordResponse:
+    schema = BookRecordResponse.model_validate(record, from_attributes=True)
+    return schema.model_copy(update={"tags": tags})
+
+
+def _link_response(link) -> BookLinkResponse:
+    return BookLinkResponse.model_validate(link, from_attributes=True)
+
+
+def _detail_response(record, tags, links, backlinks) -> BookRecordDetailResponse:
+    base = BookRecordResponse.model_validate(record, from_attributes=True).model_dump()
+    base["tags"] = tags
+    return BookRecordDetailResponse(
+        **base,
+        links=[_link_response(link) for link in links],
+        backlinks=[_link_response(link) for link in backlinks],
+    )
+
+
+def _event_response(event) -> BookEventResponse:
+    return BookEventResponse.model_validate(event, from_attributes=True)
+
+
+class BookController(BaseController):
+    """Controller for the Book domain."""
+
+    def __init__(self):
+        super().__init__(prefix="/book", tags=["Book"])
+
+    def create_router(self) -> APIRouter:
+        router = APIRouter(prefix=self.prefix, tags=self.tags)
+        service = get_book_service()
+
+        # =====================================================================
+        # add
+        # =====================================================================
+
+        @router.post(
+            "/records",
+            response_model=BookRecordResponse,
+            status_code=201,
+            summary="Add a record to the Book",
+        )
+        async def add_record(
+            data: BookRecordCreate,
+            caller: Caller = Depends(require_book_writer),
+            session: AsyncSession = Depends(get_async_session),
+        ):
+            try:
+                record, tags = await service.add_record(session, data, caller)
+            except BookError as error:
+                raise HTTPException(status_code=error.status_code, detail={"code": error.code}) from error
+            return _record_response(record, tags)
+
+        # =====================================================================
+        # list
+        # =====================================================================
+
+        @router.get(
+            "/records",
+            response_model=BookRecordListResponse,
+            summary="List records",
+        )
+        async def list_records(
+            view: Optional[ListView] = Query(None, description="inbox, today, upcoming, waiting, someday, pinned, archive, trash, done"),
+            tag: Optional[str] = Query(None),
+            scope: Optional[str] = Query(None),
+            parent: Optional[uuid.UUID] = Query(None),
+            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(
+                session, view=view, tag=tag, scope=scope, parent=parent, limit=limit, offset=offset
+            )
+            return BookRecordListResponse(
+                records=[_record_response(record, tags) for record, tags in results],
+                total=total,
+                limit=limit,
+                offset=offset,
+            )
+
+        # =====================================================================
+        # find — declared before /{record_id} so "find" is never parsed as an id
+        # =====================================================================
+
+        @router.get(
+            "/records/find",
+            response_model=BookRecordListResponse,
+            summary="Full-text search the Book",
+        )
+        async def find_records(
+            q: str = Query(..., min_length=1),
+            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.find_records(session, q=q, limit=limit, offset=offset)
+            return BookRecordListResponse(
+                records=[_record_response(record, tags) for record, tags in results],
+                total=total,
+                limit=limit,
+                offset=offset,
+            )
+
+        # =====================================================================
+        # show
+        # =====================================================================
+
+        @router.get(
+            "/records/{record_id}",
+            response_model=BookRecordDetailResponse,
+            summary="Show a record, its links and its tags",
+        )
+        async def show_record(
+            record_id: uuid.UUID,
+            caller: Caller = Depends(require_book_reader),
+            session: AsyncSession = Depends(get_async_session),
+        ):
+            try:
+                record, tags, links, backlinks = await service.show_record(session, record_id)
+            except BookError as error:
+                raise HTTPException(status_code=error.status_code, detail={"code": error.code}) from error
+            return _detail_response(record, tags, links, backlinks)
+
+        # =====================================================================
+        # change
+        # =====================================================================
+
+        @router.patch(
+            "/records/{record_id}",
+            response_model=BookRecordResponse,
+            summary="Change a record",
+        )
+        async def change_record(
+            record_id: uuid.UUID,
+            data: BookRecordUpdate,
+            caller: Caller = Depends(require_book_writer),
+            session: AsyncSession = Depends(get_async_session),
+        ):
+            try:
+                record, tags = await service.change_record(session, record_id, data, caller)
+            except BookError as error:
+                raise HTTPException(status_code=error.status_code, detail={"code": error.code}) from error
+            return _record_response(record, tags)
+
+        # =====================================================================
+        # close
+        # =====================================================================
+
+        @router.post(
+            "/records/{record_id}/close",
+            response_model=BookRecordResponse,
+            summary="Close a record (task done/dropped, question resolved, decision superseded)",
+        )
+        async def close_record(
+            record_id: uuid.UUID,
+            data: BookCloseRequest,
+            caller: Caller = Depends(require_book_writer),
+            session: AsyncSession = Depends(get_async_session),
+        ):
+            try:
+                record, tags = await service.close_record(session, record_id, data, caller)
+            except BookError as error:
+                raise HTTPException(status_code=error.status_code, detail={"code": error.code}) from error
+            return _record_response(record, tags)
+
+        # =====================================================================
+        # link / unlink
+        # =====================================================================
+
+        @router.post(
+            "/records/{record_id}/link",
+            response_model=BookRecordResponse,
+            summary="Link a record to another",
+        )
+        async def link_record(
+            record_id: uuid.UUID,
+            data: BookLinkRequest,
+            caller: Caller = Depends(require_book_writer),
+            session: AsyncSession = Depends(get_async_session),
+        ):
+            try:
+                record, tags = await service.link_records(
+                    session, record_id, data.to_id, data.relation, caller
+                )
+            except BookError as error:
+                raise HTTPException(status_code=error.status_code, detail={"code": error.code}) from error
+            return _record_response(record, tags)
+
+        @router.post(
+            "/records/{record_id}/unlink",
+            response_model=BookRecordResponse,
+            summary="Remove a link between two records",
+        )
+        async def unlink_record(
+            record_id: uuid.UUID,
+            data: BookLinkRequest,
+            caller: Caller = Depends(require_book_writer),
+            session: AsyncSession = Depends(get_async_session),
+        ):
+            try:
+                record, tags = await service.unlink_records(
+                    session, record_id, data.to_id, data.relation, caller
+                )
+            except BookError as error:
+                raise HTTPException(status_code=error.status_code, detail={"code": error.code}) from error
+            return _record_response(record, tags)
+
+        # =====================================================================
+        # tag / untag
+        # =====================================================================
+
+        @router.post(
+            "/records/{record_id}/tag",
+            response_model=BookRecordResponse,
+            summary="Tag a record",
+        )
+        async def tag_record(
+            record_id: uuid.UUID,
+            data: BookTagRequest,
+            caller: Caller = Depends(require_book_writer),
+            session: AsyncSession = Depends(get_async_session),
+        ):
+            try:
+                record, tags = await service.tag_record(session, record_id, data.tag, caller)
+            except BookError as error:
+                raise HTTPException(status_code=error.status_code, detail={"code": error.code}) from error
+            return _record_response(record, tags)
+
+        @router.post(
+            "/records/{record_id}/untag",
+            response_model=BookRecordResponse,
+            summary="Remove a tag from a record",
+        )
+        async def untag_record(
+            record_id: uuid.UUID,
+            data: BookTagRequest,
+            caller: Caller = Depends(require_book_writer),
+            session: AsyncSession = Depends(get_async_session),
+        ):
+            try:
+                record, tags = await service.untag_record(session, record_id, data.tag, caller)
+            except BookError as error:
+                raise HTTPException(status_code=error.status_code, detail={"code": error.code}) from error
+            return _record_response(record, tags)
+
+        # =====================================================================
+        # history
+        # =====================================================================
+
+        @router.get(
+            "/records/{record_id}/history",
+            response_model=BookHistoryResponse,
+            summary="The append-only event history for a record",
+        )
+        async def record_history(
+            record_id: uuid.UUID,
+            caller: Caller = Depends(require_book_reader),
+            session: AsyncSession = Depends(get_async_session),
+        ):
+            try:
+                events = await service.get_history(session, record_id)
+            except BookError as error:
+                raise HTTPException(status_code=error.status_code, detail={"code": error.code}) from error
+            return BookHistoryResponse(
+                record_id=record_id,
+                events=[_event_response(event) for event in events],
+            )
+
+        return router
+
+
+# Create controller instance
+book_controller = BookController()

src/domains/book/models.py

@@ -0,0 +1,248 @@
+"""
+Book Domain Models
+
+The Book holds Notes, Tasks, Decisions, Questions and Rejected options for
+the Office (workspace D-68, D-69, D-70). SQLAlchemy 2.0 Mapped/mapped_column
+style, timestamptz, following src/domains/auth/models.py.
+
+Table prefix is `book_` (D-68). `book_events` is append-only — migration 005
+adds a trigger that raises on UPDATE and DELETE, enforced at the database,
+not here.
+"""
+import uuid
+from datetime import datetime
+from typing import Optional
+
+from sqlalchemy import (
+    Boolean,
+    Computed,
+    DateTime,
+    ForeignKey,
+    Integer,
+    String,
+    Text,
+    UniqueConstraint,
+    func,
+)
+from sqlalchemy.dialects.postgresql import JSONB, TSVECTOR, UUID
+from sqlalchemy.orm import Mapped, mapped_column
+
+from src.shared.database import Base
+
+# Kinds and their valid statuses. Enforced in the service layer (per kind),
+# not as a database CHECK — the valid status set depends on the kind, which
+# a single-column CHECK cannot express without a trigger.
+RECORD_KINDS = ("note", "task", "decision", "question", "rejected")
+
+STATUSES_BY_KIND = {
+    "note": {"active"},
+    "task": {"inbox", "open", "waiting", "someday", "done", "dropped"},
+    "decision": {"proposed", "accepted", "superseded"},
+    "question": {"open", "settled", "dropped"},
+    "rejected": {"rejected"},
+}
+
+DEFAULT_STATUS_BY_KIND = {
+    "note": "active",
+    "task": "inbox",
+    "decision": "proposed",
+    "question": "open",
+    "rejected": "rejected",
+}
+
+SCOPES = ("work", "personal")
+DEFAULT_SCOPE = "personal"
+
+LINK_RELATIONS = ("parent", "blocks", "resolves", "supersedes", "related")
+
+
+class BookRecord(Base):
+    """A Note, Task, Decision, Question or Rejected option."""
+
+    __tablename__ = "book_records"
+    __table_args__ = (
+        UniqueConstraint("source", "external_id", name="uq_book_records_source_external_id"),
+    )
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True),
+        primary_key=True,
+        default=uuid.uuid4,
+    )
+    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="")
+    checklist: Mapped[list] = mapped_column(JSONB, nullable=False, default=list)
+    status: Mapped[str] = mapped_column(String(20), nullable=False, index=True)
+    scope: Mapped[str] = mapped_column(String(10), nullable=False, default=DEFAULT_SCOPE)
+    priority: Mapped[Optional[int]] = mapped_column(Integer, nullable=True)
+    pinned: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
+    energy: Mapped[Optional[str]] = mapped_column(String(10), nullable=True)
+    estimate_minutes: Mapped[Optional[int]] = mapped_column(Integer, nullable=True)
+    next_step: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
+    start_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True)
+    due_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True, index=True)
+    all_day: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
+    remind_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True)
+    recurrence: Mapped[Optional[str]] = mapped_column(String(255), nullable=True)
+    recurrence_mode: Mapped[Optional[str]] = mapped_column(String(20), nullable=True)
+    remind_to: Mapped[Optional[str]] = mapped_column(String(63), nullable=True)
+    last_touched_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        server_default=func.now(),
+        nullable=False,
+    )
+    archived_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True)
+    trashed_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True)
+    paused_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True)
+    paused_statuses: Mapped[Optional[dict]] = mapped_column(JSONB, nullable=True)
+    source: Mapped[Optional[str]] = mapped_column(String(50), nullable=True)
+    external_id: Mapped[Optional[str]] = mapped_column(String(255), nullable=True)
+    colour: Mapped[Optional[str]] = mapped_column(String(20), nullable=True)
+    created_by: Mapped[str] = mapped_column(String(63), nullable=False)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        server_default=func.now(),
+        nullable=False,
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        server_default=func.now(),
+        onupdate=func.now(),
+        nullable=False,
+    )
+    # Generated by migration 005 (`to_tsvector('simple', ...)`, GIN-indexed).
+    # 'simple' because the household writes mixed Dutch and English — a
+    # language-specific config would stem one of the two badly. `Computed`
+    # mirrors the migration's SQL so SQLAlchemy knows this column is
+    # database-generated and excludes it from INSERT/UPDATE — without it,
+    # the ORM tries to write NULL into a GENERATED ALWAYS column and
+    # Postgres rejects the insert outright.
+    search: Mapped[Optional[str]] = mapped_column(
+        TSVECTOR,
+        Computed("to_tsvector('simple', coalesce(title, '') || ' ' || body)", persisted=True),
+        nullable=True,
+    )
+
+    # Deliberately no `tags` relationship here. The service always queries
+    # book_tags explicitly (`BookService._tags_for`) rather than traversing
+    # an ORM relationship — partly because async SQLAlchemy can't lazy-load
+    # on an unloaded attribute without a greenlet error, but mainly because
+    # a relationship named `tags` collides with the response schema's
+    # `tags: List[str]` field: `BookRecordResponse.model_validate(record,
+    # from_attributes=True)` would read the relationship (a list of BookTag
+    # ORM objects) instead of the list of strings the controller builds
+    # separately, and fail validation. Cascade delete is handled by the
+    # column's own `ondelete="CASCADE"`, so the relationship bought nothing.
+
+    def __repr__(self) -> str:
+        return f"<BookRecord {self.kind} {self.id}>"
+
+
+class BookTag(Base):
+    """A free tag on a record. `work`/`personal` live on `scope`, not here."""
+
+    __tablename__ = "book_tags"
+
+    record_id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("book_records.id", ondelete="CASCADE"),
+        primary_key=True,
+    )
+    tag: Mapped[str] = mapped_column(String(63), primary_key=True)
+
+    def __repr__(self) -> str:
+        return f"<BookTag {self.tag} on {self.record_id}>"
+
+
+class BookRecordLink(Base):
+    """A directed relation between two records."""
+
+    __tablename__ = "book_record_links"
+
+    from_id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("book_records.id", ondelete="CASCADE"),
+        primary_key=True,
+    )
+    relation: Mapped[str] = mapped_column(String(20), primary_key=True)
+    to_id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("book_records.id", ondelete="CASCADE"),
+        primary_key=True,
+    )
+    created_by: Mapped[str] = mapped_column(String(63), nullable=False)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        server_default=func.now(),
+        nullable=False,
+    )
+
+    def __repr__(self) -> str:
+        return f"<BookRecordLink {self.from_id} {self.relation} {self.to_id}>"
+
+
+class BookAttachment(Base):
+    """A file attached to a record. Table only — no file handling yet."""
+
+    __tablename__ = "book_attachments"
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True),
+        primary_key=True,
+        default=uuid.uuid4,
+    )
+    record_id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("book_records.id", ondelete="CASCADE"),
+        nullable=False,
+        index=True,
+    )
+    path: Mapped[str] = mapped_column(String(1000), nullable=False)
+    mime: Mapped[Optional[str]] = mapped_column(String(255), nullable=True)
+    size: Mapped[Optional[int]] = mapped_column(Integer, nullable=True)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        server_default=func.now(),
+        nullable=False,
+    )
+
+    def __repr__(self) -> str:
+        return f"<BookAttachment {self.path} on {self.record_id}>"
+
+
+class BookEvent(Base):
+    """
+    An append-only audit row for every change to a record.
+
+    Migration 005 adds a trigger that raises on UPDATE and DELETE against
+    this table — enforced for every role, the owning application included.
+    `redact` (D-70 rule 4) is the one sanctioned exception, and it is a
+    later round; nothing here weakens the trigger.
+    """
+
+    __tablename__ = "book_events"
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True),
+        primary_key=True,
+        default=uuid.uuid4,
+    )
+    record_id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("book_records.id", ondelete="CASCADE"),
+        nullable=False,
+        index=True,
+    )
+    at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        server_default=func.now(),
+        nullable=False,
+    )
+    identity: Mapped[str] = mapped_column(String(63), nullable=False)
+    client: Mapped[Optional[str]] = mapped_column(String(63), nullable=True)
+    action: Mapped[str] = mapped_column(String(20), nullable=False)
+    diff: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict)
+
+    def __repr__(self) -> str:
+        return f"<BookEvent {self.action} on {self.record_id} by {self.identity}>"

src/domains/book/schemas.py

@@ -0,0 +1,190 @@
+"""
+Book Domain Schemas
+
+Pydantic request/response payloads for the Book (D-68, D-70). This round
+ships `add`, `show`, `list`, `find`, `change`, `close`, `link`/`unlink`,
+`tag`/`untag`, `history` — the verbs the Office needs next round are not
+schematized here (dossier: append, remind, reschedule, undo, purge, redact,
+pin, check, snooze, archive/trash/restore, pause/resume, due, next, batch,
+idempotency keys).
+"""
+import uuid
+from datetime import datetime
+from typing import Any, List, Literal, Optional
+
+from pydantic import Field
+
+from src.shared.base import BaseSchema
+
+RecordKind = Literal["note", "task", "decision", "question", "rejected"]
+RecordScope = Literal["work", "personal"]
+RecordEnergy = Literal["low", "medium", "high"]
+RecurrenceMode = Literal["schedule", "after_completion"]
+LinkRelation = Literal["parent", "blocks", "resolves", "supersedes", "related"]
+ListView = Literal[
+    "inbox", "today", "upcoming", "waiting", "someday",
+    "pinned", "archive", "trash", "done",
+]
+
+
+class BookChecklistItem(BaseSchema):
+    """One line of a record's checklist."""
+
+    text: str
+    checked: bool = False
+    position: int = 0
+
+
+class BookRecordCreate(BaseSchema):
+    """`add` — kind defaults to a Note so text-only capture costs nothing (D-72)."""
+
+    kind: RecordKind = "note"
+    title: Optional[str] = Field(None, max_length=500)
+    body: str = ""
+    checklist: List[BookChecklistItem] = Field(default_factory=list)
+    status: Optional[str] = None
+    scope: RecordScope = "personal"
+    priority: Optional[int] = None
+    pinned: bool = False
+    energy: Optional[RecordEnergy] = None
+    estimate_minutes: Optional[int] = None
+    next_step: Optional[str] = None
+    start_at: Optional[datetime] = None
+    due_at: Optional[datetime] = None
+    all_day: bool = False
+    remind_at: Optional[datetime] = None
+    recurrence: Optional[str] = None
+    recurrence_mode: Optional[RecurrenceMode] = None
+    remind_to: Optional[str] = None
+    source: Optional[str] = None
+    external_id: Optional[str] = None
+    colour: Optional[str] = None
+    tags: List[str] = Field(default_factory=list)
+    parent: Optional[uuid.UUID] = Field(
+        None, description="Convenience: also create a `parent` link to this record"
+    )
+
+
+class BookRecordUpdate(BaseSchema):
+    """`change` — every field optional; only what is set is applied."""
+
+    kind: Optional[RecordKind] = None
+    title: Optional[str] = Field(None, max_length=500)
+    body: Optional[str] = None
+    checklist: Optional[List[BookChecklistItem]] = None
+    status: Optional[str] = None
+    scope: Optional[RecordScope] = None
+    priority: Optional[int] = None
+    pinned: Optional[bool] = None
+    energy: Optional[RecordEnergy] = None
+    estimate_minutes: Optional[int] = None
+    next_step: Optional[str] = None
+    start_at: Optional[datetime] = None
+    due_at: Optional[datetime] = None
+    all_day: Optional[bool] = None
+    remind_at: Optional[datetime] = None
+    recurrence: Optional[str] = None
+    recurrence_mode: Optional[RecurrenceMode] = None
+    remind_to: Optional[str] = None
+    source: Optional[str] = None
+    external_id: Optional[str] = None
+    colour: Optional[str] = None
+
+
+class BookCloseRequest(BaseSchema):
+    """
+    `close` — the field that applies depends on the record's kind:
+    - task: `outcome` ("done" or "dropped")
+    - question: `resolved_by` (the Decision id that resolves it); sets `settled`
+    - decision: `superseded_by` (the new Decision id); the old one becomes `superseded`
+    """
+
+    outcome: Optional[Literal["done", "dropped"]] = None
+    resolved_by: Optional[uuid.UUID] = None
+    superseded_by: Optional[uuid.UUID] = None
+
+
+class BookLinkRequest(BaseSchema):
+    """`link` / `unlink` — direction is from the record in the path to `to_id`."""
+
+    to_id: uuid.UUID
+    relation: LinkRelation
+
+
+class BookTagRequest(BaseSchema):
+    """`tag` / `untag`."""
+
+    tag: str = Field(..., min_length=1, max_length=63)
+
+
+class BookLinkResponse(BaseSchema):
+    from_id: uuid.UUID
+    relation: str
+    to_id: uuid.UUID
+    created_by: str
+    created_at: datetime
+
+
+class BookRecordResponse(BaseSchema):
+    """Every write returns the stored record (D-70 rule 7)."""
+
+    id: uuid.UUID
+    kind: str
+    title: Optional[str]
+    body: str
+    checklist: List[dict] = Field(default_factory=list)
+    status: str
+    scope: str
+    priority: Optional[int]
+    pinned: bool
+    energy: Optional[str]
+    estimate_minutes: Optional[int]
+    next_step: Optional[str]
+    start_at: Optional[datetime]
+    due_at: Optional[datetime]
+    all_day: bool
+    remind_at: Optional[datetime]
+    recurrence: Optional[str]
+    recurrence_mode: Optional[str]
+    remind_to: Optional[str]
+    last_touched_at: datetime
+    archived_at: Optional[datetime]
+    trashed_at: Optional[datetime]
+    paused_at: Optional[datetime]
+    paused_statuses: Optional[dict]
+    source: Optional[str]
+    external_id: Optional[str]
+    colour: Optional[str]
+    created_by: str
+    created_at: datetime
+    updated_at: datetime
+    tags: List[str] = Field(default_factory=list)
+
+
+class BookRecordDetailResponse(BookRecordResponse):
+    """`show` — the record plus its outgoing links and backlinks."""
+
+    links: List[BookLinkResponse] = Field(default_factory=list)
+    backlinks: List[BookLinkResponse] = Field(default_factory=list)
+
+
+class BookRecordListResponse(BaseSchema):
+    records: List[BookRecordResponse]
+    total: int
+    limit: int
+    offset: int
+
+
+class BookEventResponse(BaseSchema):
+    id: uuid.UUID
+    record_id: uuid.UUID
+    at: datetime
+    identity: str
+    client: Optional[str]
+    action: str
+    diff: dict
+
+
+class BookHistoryResponse(BaseSchema):
+    record_id: uuid.UUID
+    events: List[BookEventResponse]

src/domains/book/service.py

@@ -0,0 +1,500 @@
+"""
+Book Domain Service
+
+Business logic for the Book — Notes, Tasks, Decisions, Questions and
+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.
+
+Methods take `session: AsyncSession` per call rather than storing it on
+`self`, following `src/domains/dashboard/service.py`.
+"""
+from datetime import datetime, timezone
+from typing import List, Optional, Tuple
+from zoneinfo import ZoneInfo
+
+from sqlalchemy import delete, func, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.domains.auth.callers import Caller
+from src.domains.book.models import (
+    DEFAULT_STATUS_BY_KIND,
+    STATUSES_BY_KIND,
+    BookEvent,
+    BookRecord,
+    BookRecordLink,
+    BookTag,
+)
+from src.domains.book.schemas import BookCloseRequest, BookRecordCreate, BookRecordUpdate
+from src.shared.config import get_settings
+from src.shared.logging import get_logger
+
+logger = get_logger(__name__)
+
+
+class BookError(Exception):
+    """Raised for a Book-domain failure; the controller maps it to a response."""
+
+    def __init__(self, code: str, status_code: int = 400):
+        self.code = code
+        self.status_code = status_code
+        super().__init__(code)
+
+
+class BookService:
+    """Service for Book operations."""
+
+    # =========================================================================
+    # Internal helpers
+    # =========================================================================
+
+    async def _get_or_404(self, session: AsyncSession, record_id) -> BookRecord:
+        record = await session.get(BookRecord, record_id)
+        if record is None:
+            raise BookError("not_found", 404)
+        return record
+
+    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)
+        return list(result.scalars().all())
+
+    async def _write_event(
+        self,
+        session: AsyncSession,
+        record_id,
+        caller: Caller,
+        action: str,
+        diff: dict,
+    ) -> None:
+        event = BookEvent(
+            record_id=record_id,
+            identity=caller.identity,
+            client=caller.client,
+            action=action,
+            diff=diff,
+        )
+        session.add(event)
+        await session.flush()
+
+    def _validate_status(self, kind: str, status: Optional[str]) -> str:
+        """Resolve and validate a status for `kind`; defaults when unset."""
+        resolved = status or DEFAULT_STATUS_BY_KIND[kind]
+        if resolved not in STATUSES_BY_KIND[kind]:
+            raise BookError("invalid_status", 400)
+        return resolved
+
+    def _touch(self, record: BookRecord) -> None:
+        record.last_touched_at = datetime.now(timezone.utc)
+
+    # =========================================================================
+    # add / show / list / find
+    # =========================================================================
+
+    async def add_record(
+        self,
+        session: AsyncSession,
+        data: BookRecordCreate,
+        caller: Caller,
+    ) -> Tuple[BookRecord, List[str]]:
+        """`add` — text-only input lands an unfiled Note (D-72)."""
+        status = self._validate_status(data.kind, data.status)
+
+        record = BookRecord(
+            kind=data.kind,
+            title=data.title,
+            body=data.body,
+            checklist=[item.model_dump() for item in data.checklist],
+            status=status,
+            scope=data.scope,
+            priority=data.priority,
+            pinned=data.pinned,
+            energy=data.energy,
+            estimate_minutes=data.estimate_minutes,
+            next_step=data.next_step,
+            start_at=data.start_at,
+            due_at=data.due_at,
+            all_day=data.all_day,
+            remind_at=data.remind_at,
+            recurrence=data.recurrence,
+            recurrence_mode=data.recurrence_mode,
+            remind_to=data.remind_to,
+            source=data.source,
+            external_id=data.external_id,
+            colour=data.colour,
+            created_by=caller.identity,
+        )
+        session.add(record)
+        await session.flush()  # assign record.id
+
+        tag_names = list(dict.fromkeys(data.tags))  # de-dup, keep order
+        for tag in tag_names:
+            session.add(BookTag(record_id=record.id, tag=tag))
+
+        if data.parent is not None:
+            session.add(BookRecordLink(
+                from_id=record.id,
+                relation="parent",
+                to_id=data.parent,
+                created_by=caller.identity,
+            ))
+
+        await self._write_event(session, record.id, caller, "add", {"kind": data.kind, "status": status})
+        await session.commit()
+        await session.refresh(record)
+
+        logger.info(f"Book record added: kind={data.kind} id={record.id} by={caller.identity}")
+        return record, tag_names
+
+    async def show_record(
+        self,
+        session: AsyncSession,
+        record_id,
+    ) -> 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)
+        tags = await self._tags_for(session, record_id)
+
+        links_stmt = select(BookRecordLink).where(BookRecordLink.from_id == record_id)
+        links = list((await session.execute(links_stmt)).scalars().all())
+
+        backlinks_stmt = select(BookRecordLink).where(BookRecordLink.to_id == record_id)
+        backlinks = list((await session.execute(backlinks_stmt)).scalars().all())
+
+        return record, tags, links, backlinks
+
+    async def list_records(
+        self,
+        session: AsyncSession,
+        view: Optional[str] = None,
+        tag: Optional[str] = None,
+        scope: Optional[str] = None,
+        parent: Optional[object] = 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`,
+        `waiting`, `someday`, `done`) filter on the matching task status and
+        simply return nothing for kinds that never carry it. `today` and
+        `upcoming` bucket on `due_at` in `book_timezone`; richer due/overdue
+        handling is `due`/`next`, both next round.
+        """
+        conditions = []
+
+        if view in ("inbox", "waiting", "someday", "done"):
+            conditions.append(BookRecord.status == view)
+        elif view == "pinned":
+            conditions.append(BookRecord.pinned.is_(True))
+        elif view == "archive":
+            conditions.append(BookRecord.archived_at.isnot(None))
+        elif view == "trash":
+            conditions.append(BookRecord.trashed_at.isnot(None))
+        elif view in ("today", "upcoming"):
+            tz = ZoneInfo(get_settings().book_timezone)
+            local_now = datetime.now(tz)
+            day_start = local_now.replace(hour=0, minute=0, second=0, microsecond=0)
+            day_end = day_start.replace(hour=23, minute=59, second=59, microsecond=999999)
+            conditions.append(BookRecord.due_at.isnot(None))
+            if view == "today":
+                conditions.append(BookRecord.due_at >= day_start)
+                conditions.append(BookRecord.due_at <= day_end)
+            else:
+                conditions.append(BookRecord.due_at > day_end)
+
+        if view not in ("archive", "trash"):
+            conditions.append(BookRecord.archived_at.is_(None))
+            conditions.append(BookRecord.trashed_at.is_(None))
+
+        if scope is not None:
+            conditions.append(BookRecord.scope == scope)
+
+        if parent is not None:
+            parent_subquery = (
+                select(BookRecordLink.from_id)
+                .where(BookRecordLink.relation == "parent", BookRecordLink.to_id == parent)
+            )
+            conditions.append(BookRecord.id.in_(parent_subquery))
+
+        if tag is not None:
+            tag_subquery = select(BookTag.record_id).where(BookTag.tag == tag)
+            conditions.append(BookRecord.id.in_(tag_subquery))
+
+        base_query = select(BookRecord).where(*conditions)
+
+        count_query = select(func.count()).select_from(base_query.subquery())
+        total = (await session.execute(count_query)).scalar() or 0
+
+        query = (
+            base_query
+            .order_by(BookRecord.last_touched_at.desc())
+            .offset(offset)
+            .limit(limit)
+        )
+        records = list((await session.execute(query)).scalars().all())
+
+        results = []
+        for record in records:
+            tags = await self._tags_for(session, record.id)
+            results.append((record, tags))
+
+        return results, total
+
+    async def find_records(
+        self,
+        session: AsyncSession,
+        q: str,
+        limit: int = 50,
+        offset: int = 0,
+    ) -> Tuple[List[Tuple[BookRecord, List[str]]], int]:
+        """`find` — full-text search over title and body."""
+        tsquery = func.websearch_to_tsquery("simple", q)
+        base_query = select(BookRecord).where(BookRecord.search.op("@@")(tsquery))
+
+        count_query = select(func.count()).select_from(base_query.subquery())
+        total = (await session.execute(count_query)).scalar() or 0
+
+        query = base_query.order_by(BookRecord.updated_at.desc()).offset(offset).limit(limit)
+        records = list((await session.execute(query)).scalars().all())
+
+        results = []
+        for record in records:
+            tags = await self._tags_for(session, record.id)
+            results.append((record, tags))
+
+        return results, total
+
+    # =========================================================================
+    # change / close
+    # =========================================================================
+
+    async def change_record(
+        self,
+        session: AsyncSession,
+        record_id,
+        data: BookRecordUpdate,
+        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)
+
+        update_data = data.model_dump(exclude_unset=True)
+        diff: dict = {}
+
+        new_kind = update_data.get("kind", record.kind)
+        if "status" in update_data or "kind" in update_data:
+            resolved_status = self._validate_status(new_kind, update_data.get("status"))
+            update_data["status"] = resolved_status
+
+        if "checklist" in update_data:
+            update_data["checklist"] = [
+                item if isinstance(item, dict) else item.model_dump()
+                for item in update_data["checklist"]
+            ]
+
+        for field, value in update_data.items():
+            before = getattr(record, field)
+            if before != value:
+                diff[field] = {"from": str(before), "to": str(value)}
+                setattr(record, field, value)
+
+        self._touch(record)
+
+        if diff:
+            await self._write_event(session, record_id, caller, "change", diff)
+
+        await session.commit()
+        await session.refresh(record)
+
+        tags = await self._tags_for(session, record_id)
+        logger.info(f"Book record changed: id={record_id} by={caller.identity} fields={list(diff)}")
+        return record, tags
+
+    async def close_record(
+        self,
+        session: AsyncSession,
+        record_id,
+        data: BookCloseRequest,
+        caller: Caller,
+    ) -> Tuple[BookRecord, List[str]]:
+        """
+        `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`).
+        """
+        record = await self._get_or_404(session, record_id)
+        diff: dict = {}
+
+        if record.kind == "task":
+            if data.outcome is None:
+                raise BookError("invalid_status", 400)
+            record.status = data.outcome
+            diff = {"status": data.outcome}
+
+        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)
+            if decision.kind != "decision":
+                raise BookError("invalid_status", 400)
+            session.add(BookRecordLink(
+                from_id=decision.id,
+                relation="resolves",
+                to_id=record.id,
+                created_by=caller.identity,
+            ))
+            record.status = "settled"
+            diff = {"status": "settled", "resolved_by": str(decision.id)}
+
+        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)
+            if new_decision.kind != "decision":
+                raise BookError("invalid_status", 400)
+            session.add(BookRecordLink(
+                from_id=new_decision.id,
+                relation="supersedes",
+                to_id=record.id,
+                created_by=caller.identity,
+            ))
+            record.status = "superseded"
+            diff = {"status": "superseded", "superseded_by": str(new_decision.id)}
+
+        else:
+            # Notes and Rejected options have no `close` transition.
+            raise BookError("invalid_status", 400)
+
+        self._touch(record)
+        await self._write_event(session, record_id, caller, "close", diff)
+        await session.commit()
+        await session.refresh(record)
+
+        tags = await self._tags_for(session, record_id)
+        logger.info(f"Book record closed: id={record_id} by={caller.identity} diff={diff}")
+        return record, tags
+
+    # =========================================================================
+    # link / unlink
+    # =========================================================================
+
+    async def link_records(
+        self,
+        session: AsyncSession,
+        record_id,
+        to_id,
+        relation: str,
+        caller: Caller,
+    ) -> Tuple[BookRecord, List[str]]:
+        record = await self._get_or_404(session, record_id)
+        await self._get_or_404(session, to_id)  # the target must exist
+
+        session.add(BookRecordLink(
+            from_id=record_id,
+            relation=relation,
+            to_id=to_id,
+            created_by=caller.identity,
+        ))
+        await self._write_event(
+            session, record_id, caller, "link", {"relation": relation, "to_id": str(to_id)}
+        )
+        await session.commit()
+        await session.refresh(record)
+
+        tags = await self._tags_for(session, record_id)
+        return record, tags
+
+    async def unlink_records(
+        self,
+        session: AsyncSession,
+        record_id,
+        to_id,
+        relation: str,
+        caller: Caller,
+    ) -> Tuple[BookRecord, List[str]]:
+        record = await self._get_or_404(session, record_id)
+
+        stmt = delete(BookRecordLink).where(
+            BookRecordLink.from_id == record_id,
+            BookRecordLink.relation == relation,
+            BookRecordLink.to_id == to_id,
+        )
+        result = await session.execute(stmt)
+        if result.rowcount == 0:
+            raise BookError("not_found", 404)
+
+        await self._write_event(
+            session, record_id, caller, "unlink", {"relation": relation, "to_id": str(to_id)}
+        )
+        await session.commit()
+        await session.refresh(record)
+
+        tags = await self._tags_for(session, record_id)
+        return record, tags
+
+    # =========================================================================
+    # tag / untag
+    # =========================================================================
+
+    async def tag_record(
+        self,
+        session: AsyncSession,
+        record_id,
+        tag: str,
+        caller: Caller,
+    ) -> Tuple[BookRecord, List[str]]:
+        record = await self._get_or_404(session, record_id)
+
+        existing = await session.get(BookTag, (record_id, tag))
+        if existing is None:
+            session.add(BookTag(record_id=record_id, tag=tag))
+            await self._write_event(session, record_id, caller, "tag", {"tag": tag})
+            await session.commit()
+
+        tags = await self._tags_for(session, record_id)
+        return record, tags
+
+    async def untag_record(
+        self,
+        session: AsyncSession,
+        record_id,
+        tag: str,
+        caller: Caller,
+    ) -> Tuple[BookRecord, List[str]]:
+        record = await self._get_or_404(session, record_id)
+
+        stmt = delete(BookTag).where(BookTag.record_id == record_id, BookTag.tag == tag)
+        result = await session.execute(stmt)
+        if result.rowcount > 0:
+            await self._write_event(session, record_id, caller, "untag", {"tag": tag})
+            await session.commit()
+
+        tags = await self._tags_for(session, record_id)
+        return record, tags
+
+    # =========================================================================
+    # history
+    # =========================================================================
+
+    async def get_history(self, session: AsyncSession, record_id, limit: int = 100) -> List[BookEvent]:
+        await self._get_or_404(session, record_id)
+
+        stmt = (
+            select(BookEvent)
+            .where(BookEvent.record_id == record_id)
+            .order_by(BookEvent.at.desc())
+            .limit(limit)
+        )
+        return list((await session.execute(stmt)).scalars().all())
+
+
+# Singleton instance
+_book_service: Optional[BookService] = None
+
+
+def get_book_service() -> BookService:
+    """Get singleton Book service instance"""
+    global _book_service
+    if _book_service is None:
+        _book_service = BookService()
+    return _book_service

src/main.py

@@ -19,6 +19,7 @@ from src.domains.infrastructure import infrastructure_controller
 from src.domains.housekeeping import housekeeping_controller
 from src.domains.static import static_controller
 from src.domains.dashboard import dashboard_controller
+from src.domains.book import book_controller
 
 # Initialize settings
 settings = get_settings()
@@ -106,6 +107,7 @@ app.include_router(infrastructure_controller.router)           # /infrastructure
 app.include_router(housekeeping_controller.router)             # /housekeeping/*
 app.include_router(static_controller.router)                   # /static/*
 app.include_router(dashboard_controller.router)                # /dashboard/*
+app.include_router(book_controller.router)                     # /book/*
 
 
 # Global exception handler

tests/conftest.py

@@ -0,0 +1,120 @@
+"""
+Shared test fixtures.
+
+Database-backed tests need `TEST_DATABASE_URL` (core-api's gitignored .env).
+They **skip** when it is unset and **refuse** (hard failure, not a skip)
+when it points at `core_api` — that is production, and the dev `.env`'s
+other `POSTGRES_*` values point there too, so this is the one guard that
+stands between a test run and the production database.
+
+Running the actual migration against the test database is done through the
+`alembic` CLI in a subprocess rather than in-process, because
+`alembic/env.py` always builds its connection URL from `get_settings()`
+(composed from `POSTGRES_*`), with no override hook of its own — and
+`get_settings()` is `lru_cache`d, so by the time a test runs, other test
+modules have almost certainly already imported `src.main` and cached a
+Settings instance pointed at production. A subprocess gets a clean
+environment and a clean cache; overriding `POSTGRES_HOST`/`USER`/
+`PASSWORD`/`DATABASE` in that child's environment, derived from
+`TEST_DATABASE_URL`, is what actually redirects the migration.
+"""
+import os
+import subprocess
+from pathlib import Path
+from urllib.parse import urlsplit
+
+import pytest
+from dotenv import load_dotenv
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+ALEMBIC_BIN = REPO_ROOT / ".venv" / "bin" / "alembic"
+
+load_dotenv(REPO_ROOT / ".env")
+
+
+def _components(url: str) -> dict:
+    """POSTGRES_* overrides for a subprocess, derived from a database URL."""
+    parts = urlsplit(url)
+    host = parts.hostname or ""
+    if parts.port:
+        host = f"{host}:{parts.port}"
+    return {
+        "POSTGRES_HOST": host,
+        "POSTGRES_USER": parts.username or "",
+        "POSTGRES_PASSWORD": parts.password or "",
+        "POSTGRES_DATABASE": parts.path.lstrip("/"),
+    }
+
+
+def run_alembic(url: str, *args: str) -> subprocess.CompletedProcess:
+    """Run `alembic <args>` against `url`, never against the repo's own .env database."""
+    env = os.environ.copy()
+    env.update(_components(url))
+    result = subprocess.run(
+        [str(ALEMBIC_BIN), *args],
+        cwd=str(REPO_ROOT),
+        env=env,
+        capture_output=True,
+        text=True,
+    )
+    return result
+
+
+@pytest.fixture(scope="session")
+def test_database_url() -> str:
+    """The core_api_test database URL, or skip/refuse per the rules above."""
+    url = os.environ.get("TEST_DATABASE_URL")
+    if not url:
+        pytest.skip("TEST_DATABASE_URL is not set — database tests skipped")
+
+    db_name = urlsplit(url).path.lstrip("/")
+    if db_name == "core_api":
+        pytest.fail(
+            "TEST_DATABASE_URL resolves to 'core_api' (production) — refusing to use it for tests"
+        )
+
+    return url
+
+
+@pytest.fixture(scope="session")
+def book_schema(test_database_url: str) -> str:
+    """Ensure migration 005 (and everything before it) is applied to core_api_test."""
+    result = run_alembic(test_database_url, "upgrade", "head")
+    if result.returncode != 0:
+        pytest.fail(f"alembic upgrade head failed:\n{result.stdout}\n{result.stderr}")
+    return test_database_url
+
+
+@pytest.fixture
+async def book_session(book_schema: str):
+    """An AsyncSession bound to core_api_test, rolled back after each test."""
+    from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+    from sqlalchemy.pool import NullPool
+
+    asyncpg_url = book_schema.replace("postgresql://", "postgresql+asyncpg://", 1)
+    engine = create_async_engine(asyncpg_url, poolclass=NullPool)
+    session_factory = async_sessionmaker(bind=engine, expire_on_commit=False)
+
+    async with session_factory() as session:
+        yield session
+        await session.rollback()
+
+    await engine.dispose()
+
+
+@pytest.fixture
+async def clean_book_tables(book_session):
+    """
+    Book service methods commit internally (same pattern as the Dashboard
+    service), so `book_session`'s post-test rollback never runs against
+    anything — the data is already durable. TRUNCATE before each test
+    instead, which the append-only trigger does not block: it is a
+    row-level BEFORE UPDATE OR DELETE trigger, and TRUNCATE fires neither.
+    """
+    from sqlalchemy import text
+
+    await book_session.execute(
+        text("TRUNCATE book_events, book_attachments, book_record_links, book_tags, book_records")
+    )
+    await book_session.commit()
+    yield book_session

tests/test_book.py

@@ -0,0 +1,373 @@
+"""
+Tests for the Book domain (T-1 / D-68, D-70) 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 uuid
+
+import pytest
+from fastapi.testclient import TestClient
+
+from src.domains.auth.oidc import oidc_config
+from src.main import app
+from src.shared.config import get_settings
+from src.shared.database import get_async_session
+
+settings = get_settings()
+
+OFFICE_KEY = "test-office-key"
+
+
+@pytest.fixture(autouse=True)
+def office_key():
+    """Every Book test authenticates as the office caller unless it says otherwise."""
+    original = settings.office_service_key
+    settings.office_service_key = OFFICE_KEY
+    yield
+    settings.office_service_key = original
+
+
+@pytest.fixture
+def client(clean_book_tables, test_database_url):
+    """
+    A TestClient whose DB dependency is overridden to core_api_test.
+
+    The override must build its own engine/session per call rather than
+    reuse `clean_book_tables`'s already-open one: TestClient drives the
+    ASGI app from a different event loop than the one pytest-asyncio ran
+    the fixture's setup in, and an asyncpg connection is not portable
+    across loops — reusing it here raised "Future ... attached to a
+    different loop" and then hung. Building fresh per call keeps
+    everything inside whichever loop is actually running the request.
+    """
+    from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+    from sqlalchemy.pool import NullPool
+
+    asyncpg_url = test_database_url.replace("postgresql://", "postgresql+asyncpg://", 1)
+
+    async def _override():
+        engine = create_async_engine(asyncpg_url, poolclass=NullPool)
+        session_factory = async_sessionmaker(bind=engine, expire_on_commit=False)
+        async with session_factory() as session:
+            yield session
+        await engine.dispose()
+
+    app.dependency_overrides[get_async_session] = _override
+    yield TestClient(app)
+    app.dependency_overrides.pop(get_async_session, None)
+
+
+def office_headers(identity: str = "jeroen", client_name: str | None = None) -> dict:
+    headers = {"Authorization": f"Bearer {OFFICE_KEY}", "X-Office-Identity": identity}
+    if client_name:
+        headers["X-Office-Client"] = client_name
+    return headers
+
+
+class TestCallerMatrixOnBookRoutes:
+    """Dossier verification #3, the Book-specific half."""
+
+    def test_no_auth_is_401_on_book_routes(self, client):
+        oidc_config.configure(enabled=True, issuers=[], audiences=[])
+        try:
+            response = client.post("/book/records", json={"body": "hi"})
+        finally:
+            oidc_config.configure(enabled=False, issuers=[], audiences=[])
+        assert response.status_code == 401
+
+    def test_office_write_without_identity_is_400(self, client):
+        response = client.post(
+            "/book/records",
+            json={"body": "hi"},
+            headers={"Authorization": f"Bearer {OFFICE_KEY}"},
+        )
+        assert response.status_code == 400
+        assert response.json()["detail"]["code"] == "identity_required"
+
+    def test_scheduler_key_is_403_on_book_records(self, client):
+        settings.scheduler_service_key = "test-scheduler-key"
+        try:
+            response = client.post(
+                "/book/records",
+                json={"body": "hi"},
+                headers={"Authorization": "Bearer test-scheduler-key"},
+            )
+        finally:
+            settings.scheduler_service_key = ""
+        assert response.status_code == 403
+        assert response.json()["detail"]["code"] == "forbidden"
+
+    def test_malformed_identity_is_400(self, client):
+        response = client.post(
+            "/book/records",
+            json={"body": "hi"},
+            headers={"Authorization": f"Bearer {OFFICE_KEY}", "X-Office-Identity": "Not Valid"},
+        )
+        assert response.status_code == 400
+        assert response.json()["detail"]["code"] == "identity_invalid"
+
+
+class TestAddShowListFind:
+    def test_add_note_defaults_to_personal_scope_and_inbox_free(self, client):
+        response = client.post(
+            "/book/records", json={"body": "buy milk"}, headers=office_headers()
+        )
+        assert response.status_code == 201
+        data = response.json()
+        assert data["kind"] == "note"
+        assert data["status"] == "active"
+        assert data["scope"] == "personal"
+        assert data["created_by"] == "jeroen"
+
+    def test_add_rejects_bad_status_for_kind(self, client):
+        response = client.post(
+            "/book/records",
+            json={"kind": "task", "body": "x", "status": "settled"},
+            headers=office_headers(),
+        )
+        assert response.status_code == 400
+        assert response.json()["detail"]["code"] == "invalid_status"
+
+    def test_every_record_carries_exactly_one_of_work_or_personal(self, client):
+        default_scope = client.post(
+            "/book/records", json={"body": "default scope"}, headers=office_headers()
+        ).json()["scope"]
+        assert default_scope in ("work", "personal")
+
+        work_scope = client.post(
+            "/book/records",
+            json={"body": "work item", "scope": "work"},
+            headers=office_headers(),
+        ).json()["scope"]
+        assert work_scope == "work"
+
+        rejected = client.post(
+            "/book/records",
+            json={"body": "bad scope", "scope": "both"},
+            headers=office_headers(),
+        )
+        assert rejected.status_code == 422  # pydantic Literal rejects a third value
+
+    def test_show_returns_record_tags_links_and_backlinks(self, client):
+        parent = client.post(
+            "/book/records", json={"kind": "task", "body": "project"}, headers=office_headers()
+        ).json()
+        child = client.post(
+            "/book/records",
+            json={"kind": "task", "body": "subtask", "parent": parent["id"], "tags": ["chore"]},
+            headers=office_headers(),
+        ).json()
+
+        child_shown = client.get(f"/book/records/{child['id']}", headers=office_headers()).json()
+        assert child_shown["tags"] == ["chore"]
+        assert len(child_shown["links"]) == 1
+        assert child_shown["links"][0]["relation"] == "parent"
+        assert child_shown["links"][0]["to_id"] == parent["id"]
+
+        parent_shown = client.get(f"/book/records/{parent['id']}", headers=office_headers()).json()
+        assert len(parent_shown["backlinks"]) == 1
+        assert parent_shown["backlinks"][0]["from_id"] == child["id"]
+
+    def test_show_missing_record_is_404(self, client):
+        response = client.get(f"/book/records/{uuid.uuid4()}", headers=office_headers())
+        assert response.status_code == 404
+        assert response.json()["detail"]["code"] == "not_found"
+
+    def test_list_filters_by_scope(self, client):
+        client.post("/book/records", json={"body": "a", "scope": "work"}, headers=office_headers())
+        client.post("/book/records", json={"body": "b", "scope": "personal"}, headers=office_headers())
+
+        work_only = client.get("/book/records", params={"scope": "work"}, headers=office_headers()).json()
+        assert work_only["total"] == 1
+        assert work_only["records"][0]["scope"] == "work"
+
+    def test_find_full_text_matches_title_and_body(self, client):
+        client.post(
+            "/book/records",
+            json={"kind": "note", "title": "Weekend plans", "body": "visit the garden centre"},
+            headers=office_headers(),
+        )
+        client.post(
+            "/book/records",
+            json={"kind": "task", "body": "renew the car insurance"},
+            headers=office_headers(),
+        )
+
+        found = client.get("/book/records/find", params={"q": "garden"}, headers=office_headers()).json()
+        assert found["total"] == 1
+        assert "garden" in found["records"][0]["body"]
+
+        found_title = client.get("/book/records/find", params={"q": "weekend"}, headers=office_headers()).json()
+        assert found_title["total"] == 1
+
+
+class TestChange:
+    def test_change_updates_only_given_fields(self, client):
+        record = client.post(
+            "/book/records", json={"kind": "task", "body": "water the plants"}, headers=office_headers()
+        ).json()
+
+        changed = client.patch(
+            f"/book/records/{record['id']}",
+            json={"priority": 1},
+            headers=office_headers(),
+        ).json()
+        assert changed["priority"] == 1
+        assert changed["body"] == "water the plants"
+
+    def test_change_without_identity_is_400(self, client):
+        record = client.post(
+            "/book/records", json={"body": "x"}, headers=office_headers()
+        ).json()
+        response = client.patch(
+            f"/book/records/{record['id']}",
+            json={"priority": 1},
+            headers={"Authorization": f"Bearer {OFFICE_KEY}"},
+        )
+        assert response.status_code == 400
+        assert response.json()["detail"]["code"] == "identity_required"
+
+
+class TestClose:
+    def test_close_task_done(self, client):
+        task = client.post(
+            "/book/records", json={"kind": "task", "body": "mow the lawn"}, headers=office_headers()
+        ).json()
+        closed = client.post(
+            f"/book/records/{task['id']}/close", json={"outcome": "done"}, headers=office_headers()
+        ).json()
+        assert closed["status"] == "done"
+
+    def test_close_question_via_resolves_sets_settled(self, client):
+        question = client.post(
+            "/book/records",
+            json={"kind": "question", "body": "what colour for the fence?"},
+            headers=office_headers(),
+        ).json()
+        decision = client.post(
+            "/book/records",
+            json={"kind": "decision", "body": "fence will be green"},
+            headers=office_headers(),
+        ).json()
+
+        closed = client.post(
+            f"/book/records/{question['id']}/close",
+            json={"resolved_by": decision["id"]},
+            headers=office_headers(),
+        ).json()
+        assert closed["status"] == "settled"
+
+        shown = client.get(f"/book/records/{question['id']}", headers=office_headers()).json()
+        assert any(b["relation"] == "resolves" and b["from_id"] == decision["id"] for b in shown["backlinks"])
+
+    def test_close_decision_via_supersedes_sets_superseded(self, client):
+        old_decision = client.post(
+            "/book/records", json={"kind": "decision", "body": "fence will be green"}, headers=office_headers()
+        ).json()
+        new_decision = client.post(
+            "/book/records", json={"kind": "decision", "body": "fence will be red"}, headers=office_headers()
+        ).json()
+
+        closed = client.post(
+            f"/book/records/{old_decision['id']}/close",
+            json={"superseded_by": new_decision["id"]},
+            headers=office_headers(),
+        ).json()
+        assert closed["status"] == "superseded"
+
+    def test_close_note_has_no_transition(self, client):
+        note = client.post("/book/records", json={"body": "just a note"}, headers=office_headers()).json()
+        response = client.post(
+            f"/book/records/{note['id']}/close", json={}, headers=office_headers()
+        )
+        assert response.status_code == 400
+        assert response.json()["detail"]["code"] == "invalid_status"
+
+
+class TestLinkUnlink:
+    def test_link_then_unlink(self, client):
+        a = client.post("/book/records", json={"body": "a"}, headers=office_headers()).json()
+        b = client.post("/book/records", json={"body": "b"}, headers=office_headers()).json()
+
+        client.post(
+            f"/book/records/{a['id']}/link",
+            json={"to_id": b["id"], "relation": "related"},
+            headers=office_headers(),
+        )
+        shown = client.get(f"/book/records/{a['id']}", headers=office_headers()).json()
+        assert len(shown["links"]) == 1
+
+        unlinked = client.post(
+            f"/book/records/{a['id']}/unlink",
+            json={"to_id": b["id"], "relation": "related"},
+            headers=office_headers(),
+        )
+        assert unlinked.status_code == 200
+        shown_after = client.get(f"/book/records/{a['id']}", headers=office_headers()).json()
+        assert len(shown_after["links"]) == 0
+
+    def test_unlink_nonexistent_link_is_404(self, client):
+        a = client.post("/book/records", json={"body": "a"}, headers=office_headers()).json()
+        b = client.post("/book/records", json={"body": "b"}, headers=office_headers()).json()
+        response = client.post(
+            f"/book/records/{a['id']}/unlink",
+            json={"to_id": b["id"], "relation": "blocks"},
+            headers=office_headers(),
+        )
+        assert response.status_code == 404
+
+
+class TestTagUntag:
+    def test_tag_then_untag(self, client):
+        record = client.post("/book/records", json={"body": "x"}, headers=office_headers()).json()
+
+        tagged = client.post(
+            f"/book/records/{record['id']}/tag", json={"tag": "chore"}, headers=office_headers()
+        ).json()
+        assert tagged["tags"] == ["chore"]
+
+        untagged = client.post(
+            f"/book/records/{record['id']}/untag", json={"tag": "chore"}, headers=office_headers()
+        ).json()
+        assert untagged["tags"] == []
+
+    def test_tagging_twice_is_idempotent(self, client):
+        record = client.post("/book/records", json={"body": "x"}, headers=office_headers()).json()
+        client.post(f"/book/records/{record['id']}/tag", json={"tag": "chore"}, headers=office_headers())
+        second = client.post(
+            f"/book/records/{record['id']}/tag", json={"tag": "chore"}, headers=office_headers()
+        )
+        assert second.status_code == 200
+        assert second.json()["tags"] == ["chore"]
+
+
+class TestHistory:
+    def test_every_write_appends_an_event_carrying_identity(self, client):
+        record = client.post(
+            "/book/records", json={"body": "x"}, headers=office_headers(identity="jeroen")
+        ).json()
+        client.patch(
+            f"/book/records/{record['id']}",
+            json={"priority": 2},
+            headers=office_headers(identity="tower-control"),
+        )
+
+        history = client.get(f"/book/records/{record['id']}/history", headers=office_headers()).json()
+        actions = [e["action"] for e in history["events"]]
+        identities = {e["identity"] for e in history["events"]}
+
+        assert "add" in actions
+        assert "change" in actions
+        assert identities == {"jeroen", "tower-control"}
+
+    def test_history_events_are_immutable_at_the_api(self, client):
+        """There is no update/delete verb for events — only `add` and `change`
+        write them, and the append-only trigger (tested in
+        test_book_migration.py) is what actually enforces it."""
+        record = client.post("/book/records", json={"body": "x"}, headers=office_headers()).json()
+        history = client.get(f"/book/records/{record['id']}/history", headers=office_headers()).json()
+        assert len(history["events"]) == 1
+        assert history["events"][0]["action"] == "add"

tests/test_book_migration.py

@@ -0,0 +1,101 @@
+"""
+Tests for Book migration 005.
+
+Runs against `core_api_test` only (see conftest.py — skip when
+TEST_DATABASE_URL is unset, hard failure if it resolves to `core_api`).
+Self-contained: starts and ends at head regardless of what ran before it,
+so test order never matters for the schema the rest of the suite needs.
+"""
+import pytest
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import create_async_engine
+from sqlalchemy.pool import NullPool
+
+from tests.conftest import run_alembic
+
+
+async def _table_exists(url: str, table: str) -> bool:
+    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("SELECT to_regclass(:name) IS NOT NULL AS present"),
+                {"name": table},
+            )
+            return bool(result.scalar())
+    finally:
+        await engine.dispose()
+
+
+class TestMigration005RoundTrip:
+    """Migration 005 must upgrade and downgrade cleanly (dossier verification #2)."""
+
+    @pytest.mark.asyncio
+    async def test_upgrade_creates_book_tables(self, test_database_url):
+        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"):
+            assert await _table_exists(test_database_url, table), f"{table} missing after upgrade"
+
+    @pytest.mark.asyncio
+    async def test_downgrade_then_upgrade_round_trips_cleanly(self, test_database_url):
+        # Baseline: make sure we start at head.
+        baseline = run_alembic(test_database_url, "upgrade", "head")
+        assert baseline.returncode == 0, baseline.stderr
+
+        down = run_alembic(test_database_url, "downgrade", "004")
+        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")
+
+        # 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")
+
+
+class TestBookEventsAppendOnly:
+    """The trigger must block UPDATE and DELETE on book_events, no exceptions."""
+
+    @pytest.mark.asyncio
+    async def test_update_on_book_events_raises(self, clean_book_tables):
+        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")
+        session.add(record)
+        await session.flush()
+        event = BookEvent(record_id=record.id, identity="test-harness", client=None, action="add", diff={})
+        session.add(event)
+        await session.commit()
+
+        with pytest.raises(Exception) as exc_info:
+            await session.execute(
+                text("UPDATE book_events SET action = 'tampered' WHERE id = :id").bindparams(id=event.id)
+            )
+            await session.commit()
+        assert "append-only" in str(exc_info.value).lower() or "book_events" in str(exc_info.value)
+        await session.rollback()
+
+    @pytest.mark.asyncio
+    async def test_delete_on_book_events_raises(self, clean_book_tables):
+        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")
+        session.add(record)
+        await session.flush()
+        event = BookEvent(record_id=record.id, identity="test-harness", client=None, action="add", diff={})
+        session.add(event)
+        await session.commit()
+
+        with pytest.raises(Exception) as exc_info:
+            await session.execute(
+                text("DELETE FROM book_events WHERE id = :id").bindparams(id=event.id)
+            )
+            await session.commit()
+        assert "append-only" in str(exc_info.value).lower() or "book_events" in str(exc_info.value)
+        await session.rollback()