From ccdbc44d6767aa6800f14e6ae74c48acd75ea272 Mon Sep 17 00:00:00 2001 From: Jeroen Schweitzer Date: Mon, 19 Oct 2026 15:18:52 +0200 Subject: [PATCH] =?UTF-8?q?feat(scribe):=20add=20the=20Scribe=20domain=20?= =?UTF-8?q?=E2=80=94=20addresses,=20send,=20mailbox,=20history,=20find,=20?= =?UTF-8?q?mark,=20tag=20(T-5)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Office (workspace D-66, D-68, D-73) needs a message bus: an address is a topic named for an identity, owned by a tenant and a consumer. D-68 puts it here, in core-api's own database, behind its own auth, following src/domains/book/. Migration 006 lands the whole schema by hand (down_revision="005"): scribe_addresses (owner_tenant + name primary key — no surrogate id, since no route ever references an address except by name), scribe_messages (retained; a generated tsvector search column + GIN index on subject/body, and a GIN index on the topics array for history's lookup), and scribe_mailbox_entries (one row per recipient address per message, unique on address+message_id). No data: seeding `jeroen` and `tower-control` is a deploy step through `register`, not this migration. Routes, every one behind require_caller("office", "user"), tenant from the caller (T-4): - register/list addresses — register is idempotent for the same consumer, 409 address_taken for another consumer in the tenant; list marks which addresses the caller's consumer owns. - send — `from` is always the caller's own identity, stamped server-side, never read from the body; 403 identity_not_owned if that identity is not a registered address owned by the caller's consumer; 404 unknown_address (naming the missing ones) if any recipient does not exist in the tenant; writes one mailbox entry per recipient. - mailbox / find — the caller's own identity's entries (mailbox: unread first; find: full text via websearch_to_tsquery), both requiring the identity be owned by the caller's consumer. - history — the retained messages sent to an address, oldest first; reading never consumes or marks, and reads scribe_messages directly rather than joining through mailbox entries. - mark / tag / untag — share one PATCH route; an entry outside the caller's tenant, or addressed to an identity the caller isn't reading as, answers 404 not_found — indistinguishable from a nonexistent entry. Every query filters on the caller's tenant (D-73): an address or message in another tenant answers exactly like one that does not exist, so no id is ever confirmed across that boundary (test_scribe.py::TestTenantIsolation, including two tests proving the tenant filter is load-bearing even when the same address *name* legitimately exists, differently owned, in two tenants — not just when the caller's own identity is unregistered there). Messaging only this round: groups, subscriptions, the notification matrix and delivery are explicitly not built (workspace T-176) — the schema already carries what `topics`/`tags` need for them. Co-Authored-By: Claude Sonnet 5 --- alembic/env.py | 3 + .../20261019_1500_006_create_scribe_tables.py | 113 ++++ src/domains/scribe/__init__.py | 9 + src/domains/scribe/controller.py | 245 ++++++++ src/domains/scribe/models.py | 136 ++++ src/domains/scribe/schemas.py | 85 +++ src/domains/scribe/service.py | 335 ++++++++++ src/main.py | 2 + tests/conftest.py | 43 ++ tests/test_scribe.py | 581 ++++++++++++++++++ tests/test_scribe_migration.py | 60 ++ 11 files changed, 1612 insertions(+) create mode 100644 alembic/versions/20261019_1500_006_create_scribe_tables.py create mode 100644 src/domains/scribe/__init__.py create mode 100644 src/domains/scribe/controller.py create mode 100644 src/domains/scribe/models.py create mode 100644 src/domains/scribe/schemas.py create mode 100644 src/domains/scribe/service.py create mode 100644 tests/test_scribe.py create mode 100644 tests/test_scribe_migration.py diff --git a/alembic/env.py b/alembic/env.py index a47d22f..5ce08fe 100644 --- a/alembic/env.py +++ b/alembic/env.py @@ -22,6 +22,9 @@ from src.domains.dashboard.models import QuickLink, DashboardWidget # noqa: F40 from src.domains.book.models import ( # noqa: F401 BookRecord, BookTag, BookRecordLink, BookAttachment, BookEvent, ) +from src.domains.scribe.models import ( # noqa: F401 + ScribeAddress, ScribeMessage, ScribeMailboxEntry, +) # Alembic Config object config = context.config diff --git a/alembic/versions/20261019_1500_006_create_scribe_tables.py b/alembic/versions/20261019_1500_006_create_scribe_tables.py new file mode 100644 index 0000000..d625149 --- /dev/null +++ b/alembic/versions/20261019_1500_006_create_scribe_tables.py @@ -0,0 +1,113 @@ +"""Create Scribe tables (dossier core-api T-5, D-66/D-68/D-73) + +Revision ID: 006 +Revises: 005 +Create Date: 2026-10-19 + +scribe_addresses, scribe_messages, scribe_mailbox_entries — messaging only +(no groups, subscriptions, notification matrix or delivery this round; +workspace T-176). No data: seeding `jeroen` and `tower-control` is a +deploy step through `register`, not this migration. +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects.postgresql import ARRAY, TSVECTOR + + +# revision identifiers, used by Alembic. +revision: str = "006" +down_revision: Union[str, None] = "005" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + """Create the Scribe tables, their search/topics GIN indexes.""" + + op.create_table( + "scribe_addresses", + sa.Column("owner_tenant", sa.String(length=63), nullable=False), + sa.Column("name", sa.String(length=63), nullable=False), + sa.Column("owner_consumer", sa.String(length=63), 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.PrimaryKeyConstraint("owner_tenant", "name"), + sa.UniqueConstraint("owner_tenant", "name", name="uq_scribe_addresses_tenant_name"), + ) + + op.create_table( + "scribe_messages", + sa.Column("id", sa.Integer(), sa.Identity(always=False), nullable=False), + sa.Column("owner_tenant", sa.String(length=63), nullable=False), + sa.Column("topics", ARRAY(sa.Text()), nullable=False, server_default=sa.text("'{}'::text[]")), + sa.Column("from_identity", sa.String(length=63), nullable=False), + sa.Column("from_consumer", sa.String(length=63), nullable=True), + sa.Column("subject", sa.String(length=500), nullable=True), + sa.Column("body", sa.Text(), nullable=False), + sa.Column("tags", ARRAY(sa.Text()), nullable=False, server_default=sa.text("'{}'::text[]")), + sa.Column("reply_to", sa.Integer(), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), + sa.PrimaryKeyConstraint("id"), + sa.ForeignKeyConstraint(["reply_to"], ["scribe_messages.id"], name="fk_scribe_messages_reply_to"), + ) + op.create_index("ix_scribe_messages_owner_tenant", "scribe_messages", ["owner_tenant"]) + op.create_index( + "ix_scribe_messages_topics", "scribe_messages", ["topics"], postgresql_using="gin" + ) + + # Generated column — 'simple' because the household writes mixed Dutch + # and English, matching book_records.search (migration 005). Added via + # raw SQL: Alembic's op.add_column has no first-class GENERATED ALWAYS + # AS support. + op.execute( + """ + ALTER TABLE scribe_messages + ADD COLUMN search tsvector + GENERATED ALWAYS AS (to_tsvector('simple', coalesce(subject, '') || ' ' || body)) STORED + """ + ) + op.create_index( + "ix_scribe_messages_search", "scribe_messages", ["search"], postgresql_using="gin" + ) + + op.create_table( + "scribe_mailbox_entries", + sa.Column("id", sa.Integer(), sa.Identity(always=False), nullable=False), + sa.Column("address", sa.String(length=63), nullable=False), + sa.Column("owner_tenant", sa.String(length=63), nullable=False), + sa.Column("message_id", sa.Integer(), nullable=False), + sa.Column("read_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("tags", ARRAY(sa.Text()), nullable=False, server_default=sa.text("'{}'::text[]")), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), + sa.PrimaryKeyConstraint("id"), + sa.ForeignKeyConstraint( + ["message_id"], ["scribe_messages.id"], name="fk_scribe_mailbox_entries_message_id" + ), + sa.UniqueConstraint("address", "message_id", name="uq_scribe_mailbox_entries_address_message"), + ) + op.create_index( + "ix_scribe_mailbox_entries_owner_tenant_address", + "scribe_mailbox_entries", + ["owner_tenant", "address"], + ) + op.create_index( + "ix_scribe_mailbox_entries_message_id", "scribe_mailbox_entries", ["message_id"] + ) + + +def downgrade() -> None: + """Drop the Scribe tables, in dependency order.""" + op.drop_index("ix_scribe_mailbox_entries_message_id", table_name="scribe_mailbox_entries") + op.drop_index( + "ix_scribe_mailbox_entries_owner_tenant_address", table_name="scribe_mailbox_entries" + ) + op.drop_table("scribe_mailbox_entries") + + op.drop_index("ix_scribe_messages_search", table_name="scribe_messages") + op.drop_index("ix_scribe_messages_topics", table_name="scribe_messages") + op.drop_index("ix_scribe_messages_owner_tenant", table_name="scribe_messages") + op.drop_table("scribe_messages") + + op.drop_table("scribe_addresses") diff --git a/src/domains/scribe/__init__.py b/src/domains/scribe/__init__.py new file mode 100644 index 0000000..d17e519 --- /dev/null +++ b/src/domains/scribe/__init__.py @@ -0,0 +1,9 @@ +""" +Scribe Domain + +The household's message bus (workspace D-66, D-68, D-73): addresses, send, +mailbox, history, find, mark, tag. Served under /scribe/*. +""" +from src.domains.scribe.controller import scribe_controller + +__all__ = ["scribe_controller"] diff --git a/src/domains/scribe/controller.py b/src/domains/scribe/controller.py new file mode 100644 index 0000000..d19026d --- /dev/null +++ b/src/domains/scribe/controller.py @@ -0,0 +1,245 @@ +""" +Scribe Controller + +The Scribe's messaging verbs (dossier core-api T-5): register, list +(addresses), send, mailbox, history, find, mark, tag/untag. Every route is +behind an Office or user caller (T-1/T-4); every query filters on the +caller's tenant (D-73) — an address or message in another tenant answers +exactly like one that does not exist. +""" +from typing import Optional + +from fastapi import APIRouter, Depends, HTTPException, Query + +from src.domains.auth.callers import Caller, require_caller +from src.domains.scribe.schemas import ( + AddressListResponse, + AddressRegisterRequest, + AddressResponse, + HistoryResponse, + MailboxEntryResponse, + MailboxListResponse, + MailboxPatchRequest, + MessageResponse, + MessageSendRequest, +) +from src.domains.scribe.service import ScribeError, get_scribe_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_scribe_reader = require_caller("office", "user") +require_scribe_writer = require_caller("office", "user", identity_required=True) + + +def _address_response(address, owned: bool) -> AddressResponse: + return AddressResponse( + name=address.name, + owner_consumer=address.owner_consumer, + created_by=address.created_by, + created_at=address.created_at, + owned=owned, + ) + + +def _message_response(message) -> MessageResponse: + return MessageResponse.model_validate(message, from_attributes=True) + + +def _entry_response(entry, message) -> MailboxEntryResponse: + return MailboxEntryResponse( + id=entry.id, + address=entry.address, + read_at=entry.read_at, + tags=entry.tags, + created_at=entry.created_at, + message=_message_response(message), + ) + + +class ScribeController(BaseController): + """Controller for the Scribe domain.""" + + def __init__(self): + super().__init__(prefix="/scribe", tags=["Scribe"]) + + def create_router(self) -> APIRouter: + router = APIRouter(prefix=self.prefix, tags=self.tags) + service = get_scribe_service() + + # ===================================================================== + # register / list addresses + # ===================================================================== + + @router.post( + "/addresses", + response_model=AddressResponse, + summary="Register an address (idempotent for the same consumer)", + ) + async def register_address( + data: AddressRegisterRequest, + caller: Caller = Depends(require_scribe_writer), + session: AsyncSession = Depends(get_async_session), + ): + try: + address, owned = await service.register_address(session, data.name, caller) + except ScribeError as error: + raise HTTPException( + status_code=error.status_code, detail={"code": error.code, **error.extra} + ) from error + return _address_response(address, owned) + + @router.get( + "/addresses", + response_model=AddressListResponse, + summary="List addresses in the caller's tenant", + ) + async def list_addresses( + caller: Caller = Depends(require_scribe_reader), + session: AsyncSession = Depends(get_async_session), + ): + try: + results = await service.list_addresses(session, caller) + except ScribeError as error: + raise HTTPException( + status_code=error.status_code, detail={"code": error.code, **error.extra} + ) from error + return AddressListResponse( + addresses=[_address_response(address, owned) for address, owned in results] + ) + + # ===================================================================== + # send + # ===================================================================== + + @router.post( + "/messages", + response_model=MessageResponse, + status_code=201, + summary="Send a message to one or more addresses", + ) + async def send_message( + data: MessageSendRequest, + caller: Caller = Depends(require_scribe_writer), + session: AsyncSession = Depends(get_async_session), + ): + try: + message = await service.send_message(session, data, caller) + except ScribeError as error: + raise HTTPException( + status_code=error.status_code, detail={"code": error.code, **error.extra} + ) from error + return _message_response(message) + + # ===================================================================== + # find + # ===================================================================== + + @router.get( + "/messages", + response_model=MailboxListResponse, + summary="Full text search over messages in the caller identity's mailbox", + ) + async def find_messages( + q: str = Query(..., min_length=1), + limit: int = Query(50, ge=1, le=200), + caller: Caller = Depends(require_scribe_reader), + session: AsyncSession = Depends(get_async_session), + ): + try: + results, total = await service.find_messages(session, caller, q, limit) + except ScribeError as error: + raise HTTPException( + status_code=error.status_code, detail={"code": error.code, **error.extra} + ) from error + return MailboxListResponse( + entries=[_entry_response(entry, message) for entry, message in results], + total=total, + limit=limit, + ) + + # ===================================================================== + # mailbox + # ===================================================================== + + @router.get( + "/mailbox", + response_model=MailboxListResponse, + summary="The caller identity's mailbox, unread first", + ) + async def get_mailbox( + unread: bool = Query(False), + tag: Optional[str] = Query(None), + limit: int = Query(50, ge=1, le=200), + before_id: Optional[int] = Query(None), + caller: Caller = Depends(require_scribe_reader), + session: AsyncSession = Depends(get_async_session), + ): + try: + results, total = await service.get_mailbox( + session, caller, unread=unread, tag=tag, limit=limit, before_id=before_id + ) + except ScribeError as error: + raise HTTPException( + status_code=error.status_code, detail={"code": error.code, **error.extra} + ) from error + return MailboxListResponse( + entries=[_entry_response(entry, message) for entry, message in results], + total=total, + limit=limit, + ) + + @router.patch( + "/mailbox/{entry_id}", + response_model=MailboxEntryResponse, + summary="Mark read/unread, or tag/untag a mailbox entry", + ) + async def patch_mailbox_entry( + entry_id: int, + data: MailboxPatchRequest, + caller: Caller = Depends(require_scribe_writer), + session: AsyncSession = Depends(get_async_session), + ): + try: + entry, message = await service.patch_mailbox_entry(session, entry_id, data, caller) + except ScribeError as error: + raise HTTPException( + status_code=error.status_code, detail={"code": error.code, **error.extra} + ) from error + return _entry_response(entry, message) + + # ===================================================================== + # history + # ===================================================================== + + @router.get( + "/addresses/{name}/history", + response_model=HistoryResponse, + summary="The retained messages sent to an address, oldest first", + ) + async def address_history( + name: str, + since_id: Optional[int] = Query(None), + limit: int = Query(100, ge=1, le=500), + caller: Caller = Depends(require_scribe_reader), + session: AsyncSession = Depends(get_async_session), + ): + try: + messages = await service.get_history(session, name, caller, since_id, limit) + except ScribeError as error: + raise HTTPException( + status_code=error.status_code, detail={"code": error.code, **error.extra} + ) from error + return HistoryResponse( + address=name, + messages=[_message_response(message) for message in messages], + ) + + return router + + +# Create controller instance +scribe_controller = ScribeController() diff --git a/src/domains/scribe/models.py b/src/domains/scribe/models.py new file mode 100644 index 0000000..c5af4ba --- /dev/null +++ b/src/domains/scribe/models.py @@ -0,0 +1,136 @@ +""" +Scribe Domain Models + +The Scribe is the household's message bus (workspace D-66, D-68, D-73): a +retained pub/sub log where an address is a topic. This round ships the +primitives migration 006 creates — addresses, messages, mailbox entries — +for direct messaging only; groups, subscriptions, the notification matrix +and delivery are later work (workspace T-176). + +SQLAlchemy 2.0 Mapped/mapped_column style, timestamptz, following +src/domains/book/models.py. Every table carries `owner_tenant` and every +query in the service layer filters on it (D-73's hard tenant boundary) — +there is no database-level RLS here, so that filter is load-bearing in +Python, not just in SQL. +""" +from datetime import datetime +from typing import List, Optional + +from sqlalchemy import ( + Computed, + DateTime, + ForeignKey, + Identity, + Integer, + String, + Text, + UniqueConstraint, + func, +) +from sqlalchemy.dialects.postgresql import ARRAY, TSVECTOR +from sqlalchemy.orm import Mapped, mapped_column + +from src.shared.database import Base + +# Address name shape, identical to src/domains/auth/callers.py's identity +# rule (an address is named for an identity, D-66) — not imported from there +# because that module's constants are private to it; kept here so a drift +# between the two never happens silently, the regex is asserted equal in +# tests/test_scribe.py instead. +ADDRESS_NAME_PATTERN = r"^[a-z0-9][a-z0-9-]{0,62}$" +RESERVED_ADDRESS_NAMES = {"book", "office", "scheduler", "tatlock"} + + +class ScribeAddress(Base): + """ + An address: a topic named for an identity. `register` creates it once + (D-66) — the pair (owner_tenant, name) is the natural key, so no + surrogate id: no route ever references an address except by name. + """ + + __tablename__ = "scribe_addresses" + __table_args__ = ( + UniqueConstraint("owner_tenant", "name", name="uq_scribe_addresses_tenant_name"), + ) + + owner_tenant: Mapped[str] = mapped_column(String(63), primary_key=True) + name: Mapped[str] = mapped_column(String(63), primary_key=True) + # Nullable: a "user"-kind caller (direct OIDC, not through the Office) + # carries no consumer (src/domains/auth/callers.py), so an address it + # registers has none either. + owner_consumer: Mapped[Optional[str]] = mapped_column(String(63), 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, + ) + + def __repr__(self) -> str: + return f"" + + +class ScribeMessage(Base): + """ + A retained message. `topics` is the set of addresses (and, later, + groups) it was published to — an integer array is denormalized from + the mailbox entries so `history` can read the retained log directly + without joining through them (D-66: "topics are retained and + re-readable"). Nothing deletes a message. + """ + + __tablename__ = "scribe_messages" + + id: Mapped[int] = mapped_column(Integer, Identity(always=False), primary_key=True) + owner_tenant: Mapped[str] = mapped_column(String(63), nullable=False, index=True) + topics: Mapped[List[str]] = mapped_column(ARRAY(Text), nullable=False, default=list) + from_identity: Mapped[str] = mapped_column(String(63), nullable=False) + from_consumer: Mapped[Optional[str]] = mapped_column(String(63), nullable=True) + subject: Mapped[Optional[str]] = mapped_column(String(500), nullable=True) + body: Mapped[str] = mapped_column(Text, nullable=False) + tags: Mapped[List[str]] = mapped_column(ARRAY(Text), nullable=False, default=list) + reply_to: Mapped[Optional[int]] = mapped_column( + Integer, ForeignKey("scribe_messages.id"), nullable=True + ) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), + server_default=func.now(), + nullable=False, + ) + # Generated by migration 006 (`to_tsvector('simple', ...)`, GIN-indexed), + # 'simple' for the same reason as book_records.search: mixed Dutch/ + # English content. `Computed` mirrors the migration's SQL. + search: Mapped[Optional[str]] = mapped_column( + TSVECTOR, + Computed("to_tsvector('simple', coalesce(subject, '') || ' ' || body)", persisted=True), + nullable=True, + ) + + def __repr__(self) -> str: + return f"" + + +class ScribeMailboxEntry(Base): + """One row per recipient address per message — what lands in a mailbox.""" + + __tablename__ = "scribe_mailbox_entries" + __table_args__ = ( + UniqueConstraint("address", "message_id", name="uq_scribe_mailbox_entries_address_message"), + ) + + id: Mapped[int] = mapped_column(Integer, Identity(always=False), primary_key=True) + address: Mapped[str] = mapped_column(String(63), nullable=False) + owner_tenant: Mapped[str] = mapped_column(String(63), nullable=False) + message_id: Mapped[int] = mapped_column( + Integer, ForeignKey("scribe_messages.id"), nullable=False + ) + read_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True) + tags: Mapped[List[str]] = mapped_column(ARRAY(Text), nullable=False, default=list) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), + server_default=func.now(), + nullable=False, + ) + + def __repr__(self) -> str: + return f"" diff --git a/src/domains/scribe/schemas.py b/src/domains/scribe/schemas.py new file mode 100644 index 0000000..5892bf8 --- /dev/null +++ b/src/domains/scribe/schemas.py @@ -0,0 +1,85 @@ +""" +Scribe Domain Schemas + +Pydantic request/response payloads for the Scribe (D-66, D-68, D-73). This +round ships `register`, `list` (addresses), `send`, `mailbox`, `history`, +`find`, `mark`, `tag`/`untag` — groups, subscriptions, the notification +matrix and delivery are later work (dossier core-api T-5). +""" +from datetime import datetime +from typing import List, Optional + +from pydantic import Field + +from src.domains.scribe.models import ADDRESS_NAME_PATTERN +from src.shared.base import BaseSchema + + +class AddressRegisterRequest(BaseSchema): + """`register` — the name only; owner (tenant, consumer) comes from the caller.""" + + name: str = Field(..., pattern=ADDRESS_NAME_PATTERN) + + +class AddressResponse(BaseSchema): + name: str + owner_consumer: Optional[str] + created_by: str + created_at: datetime + owned: bool = Field(..., description="True when the caller's consumer registered this address") + + +class AddressListResponse(BaseSchema): + addresses: List[AddressResponse] + + +class MessageSendRequest(BaseSchema): + """`send` — `from` is never in the body; it is the caller's own identity.""" + + to: List[str] = Field(..., min_length=1) + subject: Optional[str] = Field(None, max_length=500) + body: str = Field(..., min_length=1) + tags: List[str] = Field(default_factory=list) + reply_to: Optional[int] = None + + +class MessageResponse(BaseSchema): + id: int + topics: List[str] + from_identity: str + from_consumer: Optional[str] + subject: Optional[str] + body: str + tags: List[str] + reply_to: Optional[int] + created_at: datetime + + +class MailboxEntryResponse(BaseSchema): + """`mailbox` / `find` rows, and the result of `mark`/`tag`/`untag`.""" + + id: int + address: str + read_at: Optional[datetime] + tags: List[str] + created_at: datetime + message: MessageResponse + + +class MailboxListResponse(BaseSchema): + entries: List[MailboxEntryResponse] + total: int + limit: int + + +class HistoryResponse(BaseSchema): + address: str + messages: List[MessageResponse] + + +class MailboxPatchRequest(BaseSchema): + """`mark` (`read`) and `tag`/`untag` (`add_tags`/`remove_tags`) share one route.""" + + read: Optional[bool] = None + add_tags: Optional[List[str]] = None + remove_tags: Optional[List[str]] = None diff --git a/src/domains/scribe/service.py b/src/domains/scribe/service.py new file mode 100644 index 0000000..17cd114 --- /dev/null +++ b/src/domains/scribe/service.py @@ -0,0 +1,335 @@ +""" +Scribe Domain Service + +Business logic for the Scribe — the household's message bus (D-66, D-68, +D-73). Every method starts by requiring a tenant on the caller and ends by +filtering its query on it: **an address or message in another tenant +answers exactly like one that does not exist** (dossier core-api T-5), so +no method here ever confirms an id across that boundary. + +Methods take `session: AsyncSession` per call rather than storing it on +`self`, following src/domains/book/service.py. +""" +from datetime import datetime, timezone +from typing import List, Optional, Tuple + +from sqlalchemy import func, select +from sqlalchemy.ext.asyncio import AsyncSession + +from src.domains.auth.callers import Caller +from src.domains.scribe.models import ( + RESERVED_ADDRESS_NAMES, + ScribeAddress, + ScribeMailboxEntry, + ScribeMessage, +) +from src.domains.scribe.schemas import MailboxPatchRequest, MessageSendRequest +from src.shared.logging import get_logger + +logger = get_logger(__name__) + + +class ScribeError(Exception): + """Raised for a Scribe-domain failure; the controller maps it to a response.""" + + def __init__(self, code: str, status_code: int = 400, **extra): + self.code = code + self.status_code = status_code + self.extra = extra + super().__init__(code) + + +class ScribeService: + """Service for Scribe operations.""" + + # ========================================================================= + # Internal helpers + # ========================================================================= + + def _require_tenant(self, caller: Caller) -> str: + """Every Scribe operation needs a tenant; a caller without one gets + a clear error rather than a query that silently matches nothing.""" + if not caller.tenant: + raise ScribeError("tenant_required", 400) + return caller.tenant + + def _validate_address_name(self, name: str) -> None: + """Beyond the schema's regex (shape only): not `group-*`, not reserved. + Mirrors src/domains/auth/callers.py's `_validate_identity` rule — + an address is named for an identity (D-66).""" + if name.startswith("group-") or name in RESERVED_ADDRESS_NAMES: + raise ScribeError("address_invalid", 400) + + async def _get_address(self, session: AsyncSession, tenant: str, name: str) -> Optional[ScribeAddress]: + return await session.get(ScribeAddress, (tenant, name)) + + async def _require_owned_identity(self, session: AsyncSession, caller: Caller, tenant: str) -> None: + """mailbox/find read the caller's own identity's mailbox: the identity + must be a registered address owned by the caller's consumer, else + 403 identity_not_owned — the same refusal `send` uses for `from`.""" + address = await self._get_address(session, tenant, caller.identity) + if address is None or address.owner_consumer != caller.consumer: + raise ScribeError("identity_not_owned", 403) + + async def _messages_for_entries( + self, session: AsyncSession, entries: List[ScribeMailboxEntry] + ) -> dict: + if not entries: + return {} + ids = [entry.message_id for entry in entries] + stmt = select(ScribeMessage).where(ScribeMessage.id.in_(ids)) + messages = (await session.execute(stmt)).scalars().all() + return {message.id: message for message in messages} + + # ========================================================================= + # register / list addresses + # ========================================================================= + + async def register_address( + self, session: AsyncSession, name: str, caller: Caller + ) -> Tuple[ScribeAddress, bool]: + """`register` — idempotent for the same consumer; 409 address_taken + for another consumer in the tenant (D-66).""" + tenant = self._require_tenant(caller) + self._validate_address_name(name) + + existing = await self._get_address(session, tenant, name) + if existing is not None: + if existing.owner_consumer == caller.consumer: + return existing, True + raise ScribeError("address_taken", 409) + + address = ScribeAddress( + owner_tenant=tenant, + name=name, + owner_consumer=caller.consumer, + created_by=caller.identity, + ) + session.add(address) + await session.commit() + await session.refresh(address) + + logger.info(f"Scribe address registered: {tenant}/{name} by={caller.identity}") + return address, True + + async def list_addresses( + self, session: AsyncSession, caller: Caller + ) -> List[Tuple[ScribeAddress, bool]]: + """`list` — the caller's tenant; mark which the caller's consumer owns.""" + tenant = self._require_tenant(caller) + + stmt = ( + select(ScribeAddress) + .where(ScribeAddress.owner_tenant == tenant) + .order_by(ScribeAddress.name) + ) + addresses = (await session.execute(stmt)).scalars().all() + return [(a, a.owner_consumer == caller.consumer) for a in addresses] + + # ========================================================================= + # send + # ========================================================================= + + async def send_message( + self, session: AsyncSession, data: MessageSendRequest, caller: Caller + ) -> ScribeMessage: + """`send` — `from` is the caller's own identity, never the body.""" + tenant = self._require_tenant(caller) + + from_address = await self._get_address(session, tenant, caller.identity) + if from_address is None or from_address.owner_consumer != caller.consumer: + raise ScribeError("identity_not_owned", 403) + + recipients = list(dict.fromkeys(data.to)) # de-dup, keep order + found_stmt = select(ScribeAddress.name).where( + ScribeAddress.owner_tenant == tenant, ScribeAddress.name.in_(recipients) + ) + found = set((await session.execute(found_stmt)).scalars().all()) + missing = [name for name in recipients if name not in found] + if missing: + raise ScribeError("unknown_address", 404, names=missing) + + if data.reply_to is not None: + reply_target = await session.get(ScribeMessage, data.reply_to) + if reply_target is None or reply_target.owner_tenant != tenant: + raise ScribeError("not_found", 404) + + tags = list(dict.fromkeys(data.tags)) + + message = ScribeMessage( + owner_tenant=tenant, + topics=recipients, + from_identity=caller.identity, + from_consumer=caller.consumer, + subject=data.subject, + body=data.body, + tags=tags, + reply_to=data.reply_to, + ) + session.add(message) + await session.flush() # assign message.id + + for recipient in recipients: + session.add( + ScribeMailboxEntry(address=recipient, owner_tenant=tenant, message_id=message.id) + ) + + await session.commit() + await session.refresh(message) + + logger.info( + f"Scribe message sent: id={message.id} from={caller.identity} to={recipients}" + ) + return message + + # ========================================================================= + # mailbox / find + # ========================================================================= + + async def get_mailbox( + self, + session: AsyncSession, + caller: Caller, + unread: Optional[bool] = None, + tag: Optional[str] = None, + limit: int = 50, + before_id: Optional[int] = None, + ) -> Tuple[List[Tuple[ScribeMailboxEntry, ScribeMessage]], int]: + """`mailbox` — the caller's identity's entries, unread first.""" + tenant = self._require_tenant(caller) + await self._require_owned_identity(session, caller, tenant) + + conditions = [ + ScribeMailboxEntry.owner_tenant == tenant, + ScribeMailboxEntry.address == caller.identity, + ] + if unread: + conditions.append(ScribeMailboxEntry.read_at.is_(None)) + if tag is not None: + conditions.append(ScribeMailboxEntry.tags.contains([tag])) + if before_id is not None: + conditions.append(ScribeMailboxEntry.id < before_id) + + base_query = select(ScribeMailboxEntry).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(ScribeMailboxEntry.read_at.is_(None).desc(), ScribeMailboxEntry.id.desc()) + .limit(limit) + ) + entries = list((await session.execute(query)).scalars().all()) + messages_by_id = await self._messages_for_entries(session, entries) + + return [(entry, messages_by_id[entry.message_id]) for entry in entries], total + + async def find_messages( + self, session: AsyncSession, caller: Caller, q: str, limit: int = 50 + ) -> Tuple[List[Tuple[ScribeMailboxEntry, ScribeMessage]], int]: + """`find` — full text over messages in the caller identity's mailbox.""" + tenant = self._require_tenant(caller) + await self._require_owned_identity(session, caller, tenant) + + tsquery = func.websearch_to_tsquery("simple", q) + base_query = ( + select(ScribeMailboxEntry) + .join(ScribeMessage, ScribeMailboxEntry.message_id == ScribeMessage.id) + .where( + ScribeMailboxEntry.owner_tenant == tenant, + ScribeMailboxEntry.address == caller.identity, + ScribeMessage.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(ScribeMailboxEntry.id.desc()).limit(limit) + entries = list((await session.execute(query)).scalars().all()) + messages_by_id = await self._messages_for_entries(session, entries) + + return [(entry, messages_by_id[entry.message_id]) for entry in entries], total + + # ========================================================================= + # history + # ========================================================================= + + async def get_history( + self, + session: AsyncSession, + name: str, + caller: Caller, + since_id: Optional[int] = None, + limit: int = 100, + ) -> List[ScribeMessage]: + """`history` — the retained messages sent to that address, oldest + first. Reading never consumes or marks: this reads scribe_messages + directly, never scribe_mailbox_entries.""" + tenant = self._require_tenant(caller) + + address = await self._get_address(session, tenant, name) + if address is None: + raise ScribeError("unknown_address", 404) + + conditions = [ + ScribeMessage.owner_tenant == tenant, + ScribeMessage.topics.contains([name]), + ] + if since_id is not None: + conditions.append(ScribeMessage.id > since_id) + + stmt = ( + select(ScribeMessage) + .where(*conditions) + .order_by(ScribeMessage.id.asc()) + .limit(limit) + ) + return list((await session.execute(stmt)).scalars().all()) + + # ========================================================================= + # mark / tag / untag + # ========================================================================= + + async def patch_mailbox_entry( + self, session: AsyncSession, entry_id: int, data: MailboxPatchRequest, caller: Caller + ) -> Tuple[ScribeMailboxEntry, ScribeMessage]: + """`mark` (`read`) and `tag`/`untag` (`add_tags`/`remove_tags`) share + this route. An entry outside the caller's tenant, or addressed to + an identity the caller is not reading as, answers 404 not_found — + indistinguishable from a nonexistent entry, so an id never confirms + someone else's mail exists.""" + tenant = self._require_tenant(caller) + + entry = await session.get(ScribeMailboxEntry, entry_id) + if entry is None or entry.owner_tenant != tenant or entry.address != caller.identity: + raise ScribeError("not_found", 404) + + if data.read is True and entry.read_at is None: + entry.read_at = datetime.now(timezone.utc) + elif data.read is False: + entry.read_at = None + + if data.add_tags: + current = set(entry.tags) + entry.tags = list(entry.tags) + [t for t in data.add_tags if t not in current] + if data.remove_tags: + remove = set(data.remove_tags) + entry.tags = [t for t in entry.tags if t not in remove] + + await session.commit() + await session.refresh(entry) + + message = await session.get(ScribeMessage, entry.message_id) + return entry, message + + +# Singleton instance +_scribe_service: Optional[ScribeService] = None + + +def get_scribe_service() -> ScribeService: + """Get singleton Scribe service instance""" + global _scribe_service + if _scribe_service is None: + _scribe_service = ScribeService() + return _scribe_service diff --git a/src/main.py b/src/main.py index 0a2ee2d..d40a5fb 100644 --- a/src/main.py +++ b/src/main.py @@ -20,6 +20,7 @@ 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 +from src.domains.scribe import scribe_controller # Initialize settings settings = get_settings() @@ -108,6 +109,7 @@ 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/* +app.include_router(scribe_controller.router) # /scribe/* # Global exception handler diff --git a/tests/conftest.py b/tests/conftest.py index 970cf8e..83cbb74 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -118,3 +118,46 @@ async def clean_book_tables(book_session): ) await book_session.commit() yield book_session + + +@pytest.fixture(scope="session") +def scribe_schema(test_database_url: str) -> str: + """Ensure migration 006 (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 scribe_session(scribe_schema: str): + """An AsyncSession bound to core_api_test — same per-test engine pattern + as book_session (see its docstring for why: TestClient drives the app + on a different event loop than pytest-asyncio's fixture setup).""" + from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + from sqlalchemy.pool import NullPool + + asyncpg_url = scribe_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_scribe_tables(scribe_session): + """Scribe service methods commit internally (same reason as Book's + clean_book_tables). TRUNCATE all three tables together so Postgres + resolves the FK order (mailbox_entries -> messages, messages' + self-referencing reply_to) across the one statement.""" + from sqlalchemy import text + + await scribe_session.execute( + text("TRUNCATE scribe_mailbox_entries, scribe_messages, scribe_addresses") + ) + await scribe_session.commit() + yield scribe_session diff --git a/tests/test_scribe.py b/tests/test_scribe.py new file mode 100644 index 0000000..7379ccb --- /dev/null +++ b/tests/test_scribe.py @@ -0,0 +1,581 @@ +""" +Tests for the Scribe domain (dossier core-api T-5 / D-66, D-68, D-73) +against `core_api_test`. + +Follows tests/test_book.py's TestClient pattern: `get_async_session` is +overridden to the test database for every test, building a fresh +engine/session per call because TestClient drives the ASGI app on a +different event loop than pytest-asyncio's fixture setup. +""" +import pytest +from fastapi.testclient import TestClient + +from src.domains.auth.callers import _IDENTITY_RE, RESERVED_IDENTITIES +from src.domains.scribe.models import ADDRESS_NAME_PATTERN, RESERVED_ADDRESS_NAMES +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" +TENANT = "jpmschweitzer" +OTHER_TENANT = "llm_tester" + + +def test_address_name_rule_matches_callers_identity_rule(): + """models.py deliberately duplicates callers.py's identity shape rather + than importing its private constants (see models.py's comment) — assert + the two never drift apart instead.""" + assert ADDRESS_NAME_PATTERN == _IDENTITY_RE.pattern + assert RESERVED_ADDRESS_NAMES == RESERVED_IDENTITIES + + +@pytest.fixture(autouse=True) +def office_key(): + """Every Scribe 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_scribe_tables, test_database_url): + 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, consumer: str, tenant: str = TENANT) -> dict: + return { + "Authorization": f"Bearer {OFFICE_KEY}", + "X-Office-Identity": identity, + "X-Office-Consumer": consumer, + "X-Office-Tenant": tenant, + } + + +def register(client, identity: str, consumer: str, tenant: str = TENANT): + return client.post( + "/scribe/addresses", + json={"name": identity}, + headers=office_headers(identity, consumer, tenant), + ) + + +# ============================================================================= +# Caller matrix — a representative slice; the full matrix is test_callers.py +# and test_book.py's TestCallerMatrixOnBookRoutes already exercises +# require_caller itself end to end on a sibling domain. +# ============================================================================= + + +class TestCallerMatrixOnScribeRoutes: + def test_no_auth_is_401_on_scribe_routes(self, client): + from src.domains.auth.oidc import oidc_config + + oidc_config.configure(enabled=True, issuers=[], audiences=[]) + try: + response = client.post("/scribe/addresses", json={"name": "jeroen"}) + 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( + "/scribe/addresses", + json={"name": "jeroen"}, + headers={ + "Authorization": f"Bearer {OFFICE_KEY}", + "X-Office-Consumer": "office-claude", + "X-Office-Tenant": TENANT, + }, + ) + assert response.status_code == 400 + assert response.json()["detail"]["code"] == "identity_required" + + def test_office_write_without_tenant_is_400(self, client): + response = client.post( + "/scribe/addresses", + json={"name": "jeroen"}, + headers={ + "Authorization": f"Bearer {OFFICE_KEY}", + "X-Office-Identity": "jeroen", + "X-Office-Consumer": "office-claude", + }, + ) + assert response.status_code == 400 + assert response.json()["detail"]["code"] == "tenant_required" + + def test_scheduler_key_is_403_on_scribe_addresses(self, client): + settings.scheduler_service_key = "test-scheduler-key" + try: + response = client.post( + "/scribe/addresses", + json={"name": "jeroen"}, + headers={"Authorization": "Bearer test-scheduler-key"}, + ) + finally: + settings.scheduler_service_key = "" + assert response.status_code == 403 + assert response.json()["detail"]["code"] == "forbidden" + + +# ============================================================================= +# register / list addresses +# ============================================================================= + + +class TestRegisterAndListAddresses: + def test_register_creates_address_owned_by_caller(self, client): + response = register(client, "jeroen", "office-claude") + assert response.status_code == 200 + data = response.json() + assert data["name"] == "jeroen" + assert data["owner_consumer"] == "office-claude" + assert data["created_by"] == "jeroen" + assert data["owned"] is True + + def test_register_is_idempotent_for_the_same_consumer(self, client): + first = register(client, "jeroen", "office-claude") + second = register(client, "jeroen", "office-claude") + assert first.status_code == 200 + assert second.status_code == 200 + assert second.json()["owned"] is True + + def test_register_by_another_consumer_is_409_address_taken(self, client): + register(client, "jeroen", "office-claude") + response = register(client, "jeroen", "office-gemini") + assert response.status_code == 409 + assert response.json()["detail"]["code"] == "address_taken" + + @pytest.mark.parametrize("name", ["book", "office", "scheduler", "tatlock", "group-home"]) + def test_register_reserved_or_group_name_is_400(self, client, name): + """The registering identity ("jeroen") is valid and distinct from + the name being registered, so this exercises the service's own + reserved/group- check, not the caller header's identity check.""" + response = client.post( + "/scribe/addresses", + json={"name": name}, + headers=office_headers("jeroen", "office-claude"), + ) + assert response.status_code == 400 + assert response.json()["detail"]["code"] == "address_invalid" + + @pytest.mark.parametrize("name", ["Jeroen", "-abc", "a_b", ""]) + def test_register_malformed_name_is_422(self, client, name): + """Shape alone is a pydantic concern (422), not a ScribeError (400).""" + response = client.post( + "/scribe/addresses", + json={"name": name}, + headers=office_headers("jeroen", "office-claude"), + ) + assert response.status_code == 422 + + def test_list_addresses_marks_ownership(self, client): + register(client, "jeroen", "office-claude") + register(client, "tower-control", "office-gemini") + + as_claude = client.get( + "/scribe/addresses", headers=office_headers("jeroen", "office-claude") + ).json() + owned = {a["name"]: a["owned"] for a in as_claude["addresses"]} + assert owned == {"jeroen": True, "tower-control": False} + + def test_list_addresses_is_scoped_to_the_callers_tenant(self, client): + register(client, "jeroen", "office-claude", tenant=TENANT) + register(client, "jeroen", "office-claude", tenant=OTHER_TENANT) + + as_other_tenant = client.get( + "/scribe/addresses", headers=office_headers("jeroen", "office-claude", OTHER_TENANT) + ).json() + assert len(as_other_tenant["addresses"]) == 1 + + +# ============================================================================= +# The flow (dossier verification #2): register, send, mailbox, mark, +# history, find, from shows identity and consumer. +# ============================================================================= + + +class TestMessagingFlow: + def test_full_flow(self, client): + register(client, "jeroen", "office-claude") + register(client, "tower-control", "office-gemini") + + sent = client.post( + "/scribe/messages", + json={"to": ["tower-control"], "subject": "status", "body": "all green"}, + headers=office_headers("jeroen", "office-claude"), + ) + assert sent.status_code == 201 + message = sent.json() + assert message["from_identity"] == "jeroen" + assert message["from_consumer"] == "office-claude" + assert message["topics"] == ["tower-control"] + + mailbox = client.get( + "/scribe/mailbox", headers=office_headers("tower-control", "office-gemini") + ).json() + assert mailbox["total"] == 1 + entry = mailbox["entries"][0] + assert entry["read_at"] is None + assert entry["message"]["id"] == message["id"] + assert entry["message"]["body"] == "all green" + + marked = client.patch( + f"/scribe/mailbox/{entry['id']}", + json={"read": True}, + headers=office_headers("tower-control", "office-gemini"), + ) + assert marked.status_code == 200 + assert marked.json()["read_at"] is not None + + mailbox_after = client.get( + "/scribe/mailbox", headers=office_headers("tower-control", "office-gemini") + ).json() + assert mailbox_after["entries"][0]["read_at"] is not None + + history = client.get( + "/scribe/addresses/tower-control/history", + headers=office_headers("tower-control", "office-gemini"), + ).json() + assert history["address"] == "tower-control" + assert len(history["messages"]) == 1 + assert history["messages"][0]["id"] == message["id"] + + found = client.get( + "/scribe/messages", + params={"q": "green"}, + headers=office_headers("tower-control", "office-gemini"), + ).json() + assert found["total"] == 1 + assert found["entries"][0]["message"]["id"] == message["id"] + + def test_history_reading_never_marks_or_consumes(self, client): + register(client, "jeroen", "office-claude") + register(client, "tower-control", "office-gemini") + client.post( + "/scribe/messages", + json={"to": ["tower-control"], "body": "hi"}, + headers=office_headers("jeroen", "office-claude"), + ) + + client.get( + "/scribe/addresses/tower-control/history", + headers=office_headers("tower-control", "office-gemini"), + ) + mailbox = client.get( + "/scribe/mailbox", headers=office_headers("tower-control", "office-gemini") + ).json() + assert mailbox["entries"][0]["read_at"] is None + + def test_mailbox_lists_unread_first(self, client): + register(client, "jeroen", "office-claude") + register(client, "tower-control", "office-gemini") + for body in ("first", "second", "third"): + client.post( + "/scribe/messages", + json={"to": ["tower-control"], "body": body}, + headers=office_headers("jeroen", "office-claude"), + ) + + mailbox = client.get( + "/scribe/mailbox", headers=office_headers("tower-control", "office-gemini") + ).json() + first_entry_id = mailbox["entries"][0]["id"] + client.patch( + f"/scribe/mailbox/{first_entry_id}", + json={"read": True}, + headers=office_headers("tower-control", "office-gemini"), + ) + + mailbox_after = client.get( + "/scribe/mailbox", headers=office_headers("tower-control", "office-gemini") + ).json() + read_states = [e["read_at"] is None for e in mailbox_after["entries"]] + # unread (True) entries sort before the one read (False) entry + assert read_states == sorted(read_states, reverse=True) + + def test_tag_then_untag_mailbox_entry(self, client): + register(client, "jeroen", "office-claude") + register(client, "tower-control", "office-gemini") + client.post( + "/scribe/messages", + json={"to": ["tower-control"], "body": "hi"}, + headers=office_headers("jeroen", "office-claude"), + ) + entry_id = client.get( + "/scribe/mailbox", headers=office_headers("tower-control", "office-gemini") + ).json()["entries"][0]["id"] + + tagged = client.patch( + f"/scribe/mailbox/{entry_id}", + json={"add_tags": ["urgent"]}, + headers=office_headers("tower-control", "office-gemini"), + ) + assert tagged.json()["tags"] == ["urgent"] + + untagged = client.patch( + f"/scribe/mailbox/{entry_id}", + json={"remove_tags": ["urgent"]}, + headers=office_headers("tower-control", "office-gemini"), + ) + assert untagged.json()["tags"] == [] + + +# ============================================================================= +# send validation +# ============================================================================= + + +class TestSendValidation: + def test_send_to_unknown_address_is_404_naming_missing_ones(self, client): + register(client, "jeroen", "office-claude") + response = client.post( + "/scribe/messages", + json={"to": ["nobody", "tower-control"], "body": "hi"}, + headers=office_headers("jeroen", "office-claude"), + ) + assert response.status_code == 404 + detail = response.json()["detail"] + assert detail["code"] == "unknown_address" + assert set(detail["names"]) == {"nobody", "tower-control"} + + def test_send_from_identity_not_registered_is_403(self, client): + register(client, "tower-control", "office-gemini") + response = client.post( + "/scribe/messages", + json={"to": ["tower-control"], "body": "hi"}, + headers=office_headers("jeroen", "office-claude"), # jeroen never registered + ) + assert response.status_code == 403 + assert response.json()["detail"]["code"] == "identity_not_owned" + + def test_cannot_send_as_an_address_another_consumer_owns(self, client): + """dossier verification #3: a consumer cannot send *as* an address + another consumer owns.""" + register(client, "jeroen", "office-claude") + register(client, "tower-control", "office-gemini") + response = client.post( + "/scribe/messages", + json={"to": ["jeroen"], "body": "impersonating"}, + # office-gemini claiming to BE jeroen, which office-claude owns + headers=office_headers("jeroen", "office-gemini"), + ) + assert response.status_code == 403 + assert response.json()["detail"]["code"] == "identity_not_owned" + + def test_send_with_reply_to_in_another_tenant_is_404(self, client): + register(client, "jeroen", "office-claude", tenant=TENANT) + register(client, "tower-control", "office-gemini", tenant=TENANT) + sent = client.post( + "/scribe/messages", + json={"to": ["tower-control"], "body": "original"}, + headers=office_headers("jeroen", "office-claude", tenant=TENANT), + ).json() + + register(client, "jeroen", "office-claude", tenant=OTHER_TENANT) + register(client, "tower-control", "office-gemini", tenant=OTHER_TENANT) + reply = client.post( + "/scribe/messages", + json={ + "to": ["tower-control"], + "body": "cross-tenant reply", + "reply_to": sent["id"], + }, + headers=office_headers("jeroen", "office-claude", tenant=OTHER_TENANT), + ) + assert reply.status_code == 404 + assert reply.json()["detail"]["code"] == "not_found" + + +# ============================================================================= +# Isolation (dossier verification #3): the hard tenant boundary, both +# directions — every cross-tenant answer is indistinguishable from "does +# not exist". +# ============================================================================= + + +class TestTenantIsolation: + def test_llm_tester_cannot_list_jpmschweitzer_addresses(self, client): + register(client, "jeroen", "office-claude", tenant=TENANT) + as_other = client.get( + "/scribe/addresses", headers=office_headers("jeroen", "office-claude", OTHER_TENANT) + ).json() + assert as_other["addresses"] == [] + + def test_llm_tester_cannot_send_to_a_jpmschweitzer_address(self, client): + register(client, "tower-control", "office-gemini", tenant=TENANT) # TENANT only + register(client, "jeroen", "office-claude", tenant=OTHER_TENANT) + response = client.post( + "/scribe/messages", + json={"to": ["tower-control"], "body": "leaking?"}, + headers=office_headers("jeroen", "office-claude", OTHER_TENANT), + ) + # From OTHER_TENANT's point of view, "tower-control" does not exist. + assert response.status_code == 404 + assert response.json()["detail"]["code"] == "unknown_address" + + def test_llm_tester_cannot_read_mailbox_of_jpmschweitzer_address(self, client): + register(client, "jeroen", "office-claude", tenant=TENANT) + register(client, "tower-control", "office-gemini", tenant=TENANT) + client.post( + "/scribe/messages", + json={"to": ["tower-control"], "body": "secret"}, + headers=office_headers("jeroen", "office-claude", tenant=TENANT), + ) + + # llm_tester has never registered tower-control, so it cannot even + # bind the identity -> 403, indistinguishable from "no such address". + response = client.get( + "/scribe/mailbox", + headers=office_headers("tower-control", "office-gemini", OTHER_TENANT), + ) + assert response.status_code == 403 + assert response.json()["detail"]["code"] == "identity_not_owned" + + def test_llm_tester_cannot_see_history_of_a_jpmschweitzer_address(self, client): + register(client, "jeroen", "office-claude", tenant=TENANT) + register(client, "tower-control", "office-gemini", tenant=TENANT) + client.post( + "/scribe/messages", + json={"to": ["tower-control"], "body": "secret"}, + headers=office_headers("jeroen", "office-claude", tenant=TENANT), + ) + + response = client.get( + "/scribe/addresses/tower-control/history", + headers=office_headers("jeroen", "office-claude", OTHER_TENANT), + ) + assert response.status_code == 404 + assert response.json()["detail"]["code"] == "unknown_address" + + def test_llm_tester_cannot_find_a_message_of_a_jpmschweitzer_address(self, client): + register(client, "jeroen", "office-claude", tenant=TENANT) + register(client, "tower-control", "office-gemini", tenant=TENANT) + client.post( + "/scribe/messages", + json={"to": ["tower-control"], "body": "unique-secret-word"}, + headers=office_headers("jeroen", "office-claude", tenant=TENANT), + ) + + response = client.get( + "/scribe/messages", + params={"q": "unique-secret-word"}, + headers=office_headers("tower-control", "office-gemini", OTHER_TENANT), + ) + assert response.status_code == 403 + assert response.json()["detail"]["code"] == "identity_not_owned" + + def test_mark_on_an_entry_in_another_tenant_is_404(self, client): + register(client, "jeroen", "office-claude", tenant=TENANT) + register(client, "tower-control", "office-gemini", tenant=TENANT) + client.post( + "/scribe/messages", + json={"to": ["tower-control"], "body": "secret"}, + headers=office_headers("jeroen", "office-claude", tenant=TENANT), + ) + entry_id = client.get( + "/scribe/mailbox", headers=office_headers("tower-control", "office-gemini", TENANT) + ).json()["entries"][0]["id"] + + register(client, "tower-control", "office-gemini", tenant=OTHER_TENANT) + response = client.patch( + f"/scribe/mailbox/{entry_id}", + json={"read": True}, + headers=office_headers("tower-control", "office-gemini", OTHER_TENANT), + ) + assert response.status_code == 404 + assert response.json()["detail"]["code"] == "not_found" + + def test_mark_on_someone_elses_entry_within_the_same_tenant_is_404(self, client): + """Not a tenant crossing, but the same non-disclosure principle: + an entry addressed to another identity answers like it doesn't + exist, rather than confirming it belongs to someone else.""" + register(client, "jeroen", "office-claude", tenant=TENANT) + register(client, "tower-control", "office-gemini", tenant=TENANT) + client.post( + "/scribe/messages", + json={"to": ["tower-control"], "body": "secret"}, + headers=office_headers("jeroen", "office-claude", tenant=TENANT), + ) + entry_id = client.get( + "/scribe/mailbox", headers=office_headers("tower-control", "office-gemini", TENANT) + ).json()["entries"][0]["id"] + + response = client.patch( + f"/scribe/mailbox/{entry_id}", + json={"read": True}, + headers=office_headers("jeroen", "office-claude", TENANT), # not tower-control + ) + assert response.status_code == 404 + assert response.json()["detail"]["code"] == "not_found" + + def test_mailbox_does_not_leak_entries_from_another_tenants_same_named_address(self, client): + """The identity-ownership precheck alone is not the isolation + guarantee: a caller legitimately owns "jeroen" in their own tenant, + and a *different* "jeroen" (same name, different owner) exists in + another tenant with mail of its own. The mailbox query's own tenant + filter is what keeps that mail out — this is what mutation check #1 + (dossier) targets.""" + register(client, "jeroen", "office-claude", tenant=TENANT) + register(client, "tower-control", "office-gemini", tenant=TENANT) + client.post( + "/scribe/messages", + json={"to": ["jeroen"], "body": "for the TENANT jeroen"}, + headers=office_headers("tower-control", "office-gemini", tenant=TENANT), + ) + + register(client, "jeroen", "office-claude", tenant=OTHER_TENANT) + register(client, "tower-control", "office-gemini", tenant=OTHER_TENANT) + client.post( + "/scribe/messages", + json={"to": ["jeroen"], "body": "for the OTHER_TENANT jeroen"}, + headers=office_headers("tower-control", "office-gemini", tenant=OTHER_TENANT), + ) + + mailbox = client.get( + "/scribe/mailbox", headers=office_headers("jeroen", "office-claude", TENANT) + ).json() + assert mailbox["total"] == 1 + assert mailbox["entries"][0]["message"]["body"] == "for the TENANT jeroen" + + def test_find_does_not_leak_messages_from_another_tenants_same_named_address(self, client): + """Same concern as above, for `find` (mutation check #1's other half).""" + register(client, "jeroen", "office-claude", tenant=TENANT) + register(client, "tower-control", "office-gemini", tenant=TENANT) + client.post( + "/scribe/messages", + json={"to": ["jeroen"], "body": "shared-search-term in TENANT"}, + headers=office_headers("tower-control", "office-gemini", tenant=TENANT), + ) + + register(client, "jeroen", "office-claude", tenant=OTHER_TENANT) + register(client, "tower-control", "office-gemini", tenant=OTHER_TENANT) + client.post( + "/scribe/messages", + json={"to": ["jeroen"], "body": "shared-search-term in OTHER_TENANT"}, + headers=office_headers("tower-control", "office-gemini", tenant=OTHER_TENANT), + ) + + found = client.get( + "/scribe/messages", + params={"q": "shared-search-term"}, + headers=office_headers("jeroen", "office-claude", TENANT), + ).json() + assert found["total"] == 1 + assert found["entries"][0]["message"]["body"] == "shared-search-term in TENANT" diff --git a/tests/test_scribe_migration.py b/tests/test_scribe_migration.py new file mode 100644 index 0000000..1a2e15f --- /dev/null +++ b/tests/test_scribe_migration.py @@ -0,0 +1,60 @@ +""" +Tests for Scribe migration 006. + +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. +""" +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 TestMigration006RoundTrip: + """Migration 006 must upgrade and downgrade cleanly (dossier verification #1).""" + + @pytest.mark.asyncio + async def test_upgrade_creates_scribe_tables(self, test_database_url): + result = run_alembic(test_database_url, "upgrade", "head") + assert result.returncode == 0, result.stderr + + for table in ("scribe_addresses", "scribe_messages", "scribe_mailbox_entries"): + 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 = run_alembic(test_database_url, "upgrade", "head") + assert baseline.returncode == 0, baseline.stderr + + down = run_alembic(test_database_url, "downgrade", "005") + assert down.returncode == 0, down.stderr + assert not await _table_exists(test_database_url, "scribe_addresses") + assert not await _table_exists(test_database_url, "scribe_messages") + assert not await _table_exists(test_database_url, "scribe_mailbox_entries") + # Book tables, created by the migration below this one, must survive + # a downgrade of 006 — proves 006 only touches its own tables. + assert await _table_exists(test_database_url, "book_records") + + # Leave the schema at head for the rest of the suite. + up = run_alembic(test_database_url, "upgrade", "006") + assert up.returncode == 0, up.stderr + assert await _table_exists(test_database_url, "scribe_addresses") + assert await _table_exists(test_database_url, "scribe_messages") + assert await _table_exists(test_database_url, "scribe_mailbox_entries")