feat: fiscal_series, fiscal_certificates, fiscal_documents models + migrations

Ports the three tables that change owner per the design spec (decision #2):
FiscalSeries (was tenants.FiscalDocumentSeries), FiscalCertificate,
FiscalDocument (both from fiscal.models) — organization_id/branch_id
replaced by product_id (FK products) + tenant_ref/branch_ref (opaque
strings) per the porte table. sale_id/service_order_id dropped (no Sale
concept here). document_model is now a plain String(2) + Python enum,
never a Postgres enum (Global Constraints: zero enum PG — the auto's own
version of this column was a real PG enum, a documented debt not repeated
here).

Constraints preserved: UNIQUE chave_acesso, partial-unique
cert-vivo-per-(product_id, branch_ref) (product-scoped in addition to the
auto's branch_id, since branch_ref is an opaque string two different
products could coincidentally share), UNIQUE (product_id, tenant_ref,
branch_ref, document_model, serie) on the series.

documents.service.allocate_fiscal_number ported verbatim (mechanism +
contract): SELECT ... FOR UPDATE + populate_existing=True, no-commit
contract (caller commits together with the FiscalDocument insert, Task 5).
The next_number regression guard is deliberately deferred to Task 4's
PATCH /v1/series endpoint (needs FiscalDocument, which now exists).

AST guard (tests/shared/test_for_update_populate_existing.py) ported and
adapted to scan src/fiscal_svc/, plus two new self-tests proving the
detection logic itself in both directions (flags a missing fix, does not
false-positive on a correctly fixed multi-line chain) — the ported guard
alone only proves "currently green", not "actually detects".

33 tests green via `make k8s-test` (real-migration round trips + unique
constraint violations, N=10 concurrency, identity-map staleness repro,
tenancy-scoping not-found across product/tenant_ref/branch_ref).

alembic/env.py

@@ -13,7 +13,7 @@ from fiscal_svc.core.db import Base
 # Uncomment as modules gain SQLAlchemy models, so autogenerate can see them
 # (mirrors auto/backend/alembic/env.py's own convention):
 from fiscal_svc.tenancy import models as tenancy_models  # noqa: F401
-# from fiscal_svc.documents import models as documents_models  # noqa: F401
+from fiscal_svc.documents import models as documents_models  # noqa: F401
 
 # this is the Alembic Config object, which provides
 # access to the values within the .ini file in use.

alembic/versions/523235d06bd5_fiscal_series_certificates_documents.py

@@ -0,0 +1,126 @@
+"""fiscal_series + fiscal_certificates + fiscal_documents (Task 3): the
+three tables that "change owner" per the design spec (decision #2) --
+ported from the auto's `fiscal_document_series` (tenants), `fiscal_
+certificates`/`fiscal_documents` (fiscal), with product_id/tenant_ref/
+branch_ref replacing organization_id/branch_id (porte table).
+
+Revision ID: 523235d06bd5
+Revises: fd0ffe65c993
+Create Date: 2026-07-22 00:00:00.000000
+
+"""
+from __future__ import annotations
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "523235d06bd5"
+down_revision: Union[str, Sequence[str], None] = "fd0ffe65c993"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "fiscal_series",
+        sa.Column("id", sa.UUID(), nullable=False),
+        sa.Column("product_id", sa.UUID(), nullable=False),
+        sa.Column("tenant_ref", sa.String(length=64), nullable=False),
+        sa.Column("branch_ref", sa.String(length=64), nullable=False),
+        sa.Column("document_model", sa.String(length=2), nullable=False),
+        sa.Column("serie", sa.Integer(), nullable=False),
+        sa.Column("next_number", sa.Integer(), nullable=False),
+        sa.Column(
+            "created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False
+        ),
+        sa.Column(
+            "updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False
+        ),
+        sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True),
+        sa.ForeignKeyConstraint(["product_id"], ["products.id"]),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint(
+            "product_id", "tenant_ref", "branch_ref", "document_model", "serie",
+            name="uq_fiscal_series_product_tenant_branch_model_serie",
+        ),
+    )
+
+    op.create_table(
+        "fiscal_certificates",
+        sa.Column("id", sa.UUID(), nullable=False),
+        sa.Column("product_id", sa.UUID(), nullable=False),
+        sa.Column("tenant_ref", sa.String(length=64), nullable=False),
+        sa.Column("branch_ref", sa.String(length=64), nullable=False),
+        sa.Column("cnpj", sa.String(length=14), nullable=False),
+        sa.Column("pfx_encrypted", sa.LargeBinary(), nullable=False),
+        sa.Column("password_encrypted", sa.LargeBinary(), nullable=False),
+        sa.Column("subject_cn", sa.String(length=255), nullable=False),
+        sa.Column("cnpj_certificado", sa.String(length=14), nullable=False),
+        sa.Column("not_valid_before", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("not_valid_after", sa.DateTime(timezone=True), nullable=False),
+        sa.Column(
+            "created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False
+        ),
+        sa.Column(
+            "updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False
+        ),
+        sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True),
+        sa.ForeignKeyConstraint(["product_id"], ["products.id"]),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index(
+        "ix_fiscal_certificates_product_branch_live",
+        "fiscal_certificates",
+        ["product_id", "branch_ref"],
+        unique=True,
+        postgresql_where=sa.text("deleted_at IS NULL"),
+    )
+
+    op.create_table(
+        "fiscal_documents",
+        sa.Column("id", sa.UUID(), nullable=False),
+        sa.Column("product_id", sa.UUID(), nullable=False),
+        sa.Column("tenant_ref", sa.String(length=64), nullable=False),
+        sa.Column("branch_ref", sa.String(length=64), nullable=False),
+        sa.Column("series_id", sa.UUID(), nullable=False),
+        sa.Column("document_model", sa.String(length=2), nullable=False),
+        sa.Column("serie", sa.Integer(), nullable=False),
+        sa.Column("numero", sa.Integer(), nullable=False),
+        sa.Column("chave_acesso", sa.String(length=44), nullable=False),
+        sa.Column("codigo_numerico", sa.String(length=8), nullable=False),
+        sa.Column("status", sa.String(length=20), nullable=False),
+        sa.Column("ambiente", sa.String(length=12), nullable=False),
+        sa.Column("xml_assinado", sa.Text(), nullable=False),
+        sa.Column("rejeicao_codigo", sa.String(length=10), nullable=True),
+        sa.Column("rejeicao_motivo", sa.String(length=500), nullable=True),
+        sa.Column("protocolo", sa.String(length=20), nullable=True),
+        sa.Column("autorizada_em", sa.DateTime(timezone=True), nullable=True),
+        sa.Column(
+            "created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False
+        ),
+        sa.Column(
+            "updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False
+        ),
+        sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True),
+        sa.ForeignKeyConstraint(["product_id"], ["products.id"]),
+        sa.ForeignKeyConstraint(["series_id"], ["fiscal_series.id"]),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("chave_acesso", name="uq_fiscal_documents_chave_acesso"),
+    )
+    op.create_index(
+        "ix_fiscal_documents_product_tenant_branch_status",
+        "fiscal_documents",
+        ["product_id", "tenant_ref", "branch_ref", "status"],
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_fiscal_documents_product_tenant_branch_status", table_name="fiscal_documents")
+    op.drop_table("fiscal_documents")
+
+    op.drop_index("ix_fiscal_certificates_product_branch_live", table_name="fiscal_certificates")
+    op.drop_table("fiscal_certificates")
+
+    op.drop_table("fiscal_series")

src/fiscal_svc/documents/models.py

@@ -0,0 +1,217 @@
+"""Task 3: the three fiscal tables that "change owner" per the design spec
+(decision #2) -- they used to live in the auto (`app/modules/tenants/
+models.py`'s `FiscalDocumentSeries`, `app/modules/fiscal/models.py`'s
+`FiscalCertificate`/`FiscalDocument`) and now live HERE, as the source of
+truth this service was built to be. Ported with the porte table
+(2026-07-17-fiscal-svc-f2-servico.md) applied throughout:
+
+    | No auto                          | No serviço                        |
+    |-----------------------------------|-----------------------------------|
+    | organization_id (FK organizations)| product_id (FK products) +        |
+    |                                    | tenant_ref: String(64)            |
+    | branch_id (FK branches)           | branch_ref: String(64) + cnpj     |
+    |                                    | (the strong link)                 |
+    | Sale/rotas por venda              | não existem (sale_id/             |
+    |                                    | service_order_id DROPPED --       |
+    |                                    | emissão recebe DadosEmissao       |
+    |                                    | completo, sem correlação a uma    |
+    |                                    | linha de venda deste lado)        |
+
+Global Constraints (zero enum PG): unlike the auto's `FiscalDocumentSeries.
+document_model` (a REAL Postgres `Enum` -- a documented, deliberately
+un-repeated debt there), `document_model` here is plain `String(2)` +
+Python enum at the Pydantic border ONLY, same convention the auto's own
+`FiscalDocument.document_model`/`.status` already used (see their
+docstrings) -- this port does not re-introduce the PG enum anywhere."""
+import enum
+import uuid
+from datetime import datetime
+
+from sqlalchemy import (
+    DateTime,
+    ForeignKey,
+    Index,
+    Integer,
+    LargeBinary,
+    String,
+    Text,
+    UniqueConstraint,
+    text,
+)
+from sqlalchemy.orm import Mapped, mapped_column
+
+from fiscal_svc.core.db import Base
+from fiscal_svc.shared.base_model import SoftDeleteMixin, TimestampMixin, UUIDPKMixin
+
+
+class FiscalDocumentModel(str, enum.Enum):
+    """Ported from `auto/backend/app/modules/tenants/models.py`. Python
+    enum only -- see this module's docstring for why the auto's Postgres
+    `Enum` column for this same enum is NOT repeated here."""
+
+    NFE_55 = "55"
+    NFCE_65 = "65"
+
+
+class FiscalDocumentStatus(str, enum.Enum):
+    """Ported verbatim from `auto/backend/app/modules/fiscal/models.py` --
+    already `String` + Python enum there (ZERO Postgres enum), so no
+    adaptation needed. State machine: `ASSINADO -> TRANSMITINDO ->
+    AUTORIZADA | REJEITADA | PENDENTE_CONSULTA` (+ `CANCELADA`/`DENEGADA`).
+    Task 3 (this port) only ever WRITES `ASSINADO` -- transmission (F4)
+    writes the rest; the full vocabulary exists now so a later phase never
+    needs a column-shape migration for a new status."""
+
+    ASSINADO = "ASSINADO"
+    TRANSMITINDO = "TRANSMITINDO"
+    AUTORIZADA = "AUTORIZADA"
+    REJEITADA = "REJEITADA"
+    PENDENTE_CONSULTA = "PENDENTE_CONSULTA"
+    CANCELADA = "CANCELADA"
+    DENEGADA = "DENEGADA"
+
+
+class FiscalSeries(Base, UUIDPKMixin, TimestampMixin, SoftDeleteMixin):
+    """Ported from the auto's `FiscalDocumentSeries` (`tenants/models.py`) --
+    per the design spec this table is the NEW owner of fiscal numbering
+    (decision #2: "mudam de dono"). `product_id`/`tenant_ref`/`branch_ref`
+    replace `organization_id`/`branch_id` per the porte table (this
+    module's docstring).
+
+    Atomic allocation (`SELECT ... FOR UPDATE` + `populate_existing`) is
+    `documents.service.allocate_fiscal_number` (Task 3, ported with its
+    exact no-commit contract -- see that function's docstring). The
+    "guard retroativo" (`next_number` can never regress below the highest
+    `numero` this series has already emitted, per `FiscalDocument.numero`)
+    is deliberately NOT implemented in this task -- it belongs to the
+    `PATCH /v1/series` endpoint (Task 4), which needs this exact model
+    plus `FiscalDocument` (both now exist, Task 3) to check against. Same
+    "documented now, wired later" shape as the auto's own
+    `FiscalDocumentSeries` docstring ("a emissão em si nasce no módulo
+    fiscal da 1b")."""
+
+    __tablename__ = "fiscal_series"
+    __table_args__ = (
+        UniqueConstraint(
+            "product_id", "tenant_ref", "branch_ref", "document_model", "serie",
+            name="uq_fiscal_series_product_tenant_branch_model_serie",
+        ),
+    )
+
+    product_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("products.id"), nullable=False)
+    tenant_ref: Mapped[str] = mapped_column(String(64), nullable=False)
+    branch_ref: Mapped[str] = mapped_column(String(64), nullable=False)
+    document_model: Mapped[str] = mapped_column(String(2), nullable=False)
+    serie: Mapped[int] = mapped_column(Integer, nullable=False)
+    next_number: Mapped[int] = mapped_column(Integer, nullable=False)
+
+
+class FiscalCertificate(Base, UUIDPKMixin, TimestampMixin, SoftDeleteMixin):
+    """Ported from the auto's `FiscalCertificate` (`app/modules/fiscal/
+    models.py`) -- A1 certificate (.pfx) for a `branch_ref`. Per the porte
+    table, the emitter is `(product_id, branch_ref)`, with `cnpj` carried
+    on the row itself (the "vínculo forte", design spec decision #4: the
+    CNPJ is what's validated against the certificate at upload and against
+    the emitente at emission -- `branch_ref` alone is an opaque string this
+    service never interprets). No fallback to a tenant-level or
+    product-level certificate -- fail-closed by construction, same as the
+    auto: a branch_ref without its own live certificate cannot emit.
+
+    One VIVO (`deleted_at IS NULL`) row per `(product_id, branch_ref)`: a
+    second upload soft-deletes the previous live row (Task 4's
+    `certificates.service`). `ix_fiscal_certificates_product_branch_live`
+    is a PARTIAL UNIQUE index enforcing that at the database level (same
+    "two concurrent uploads must not both land a live row" reasoning as the
+    auto's own migration `a1b2c3d4e5f6`'s fix) -- scoped by `product_id` in
+    ADDITION to `branch_ref` (the auto's version only needed `branch_id`,
+    already product-scoped by being a real FK; here `branch_ref` is an
+    OPAQUE string owned by the calling product, so two DIFFERENT products
+    could coincidentally pick the identical string for two DIFFERENT real
+    branches -- scoping the uniqueness by `product_id` too is what keeps
+    that from cross-contaminating one product's certificate slot with
+    another's).
+
+    `pfx_encrypted`/`password_encrypted` are Fernet ciphertext (Task 4,
+    `FISCAL_CERT_ENCRYPTION_KEY` env var, own key -- never
+    `app.core.config.Settings`, same "secrets never live in Settings"
+    convention as the auto)."""
+
+    __tablename__ = "fiscal_certificates"
+    __table_args__ = (
+        Index(
+            "ix_fiscal_certificates_product_branch_live",
+            "product_id",
+            "branch_ref",
+            unique=True,
+            postgresql_where=text("deleted_at IS NULL"),
+        ),
+    )
+
+    product_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("products.id"), nullable=False)
+    tenant_ref: Mapped[str] = mapped_column(String(64), nullable=False)
+    branch_ref: Mapped[str] = mapped_column(String(64), nullable=False)
+    cnpj: Mapped[str] = mapped_column(String(14), nullable=False)
+    pfx_encrypted: Mapped[bytes] = mapped_column(LargeBinary, nullable=False)
+    password_encrypted: Mapped[bytes] = mapped_column(LargeBinary, nullable=False)
+    subject_cn: Mapped[str] = mapped_column(String(255), nullable=False)
+    cnpj_certificado: Mapped[str] = mapped_column(String(14), nullable=False)
+    not_valid_before: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
+    not_valid_after: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
+
+
+class FiscalDocument(Base, UUIDPKMixin, TimestampMixin, SoftDeleteMixin):
+    """Ported from the auto's `FiscalDocument` (`app/modules/fiscal/
+    models.py`). Per the porte table, `sale_id`/`service_order_id` are
+    DROPPED -- this service has no notion of a Sale/ServiceOrder (decision
+    #3 of the design spec: it receives a complete `DadosEmissao`, never
+    resolves or references the caller's own domain objects). A product's
+    own correlation to ITS sale/OS is the product's problem, carried via
+    the `Idempotency-Key` (Task 5) it sends, not a column here.
+
+    Invariants preserved from the auto (contract, not mere convention):
+    (a) a row is only inserted in the SAME commit that allocates `numero`
+        via `documents.service.allocate_fiscal_number` (which does not
+        commit by itself -- see its docstring) -- outbox transactional,
+        exactly as ported;
+    (b) documento é IMUTÁVEL após `ASSINADO` -- correção é cancelar (evento
+        futuro) e emitir outro, nunca um UPDATE no XML/chave já assinados;
+    (c) `chave_acesso` is UNIQUE GLOBALLY (not just per product) -- it
+        already embeds the emitente's CNPJ in its own digits, so two
+        DIFFERENT products (or tenants) can never legitimately produce the
+        same one; a collision here can only be a generation bug, never a
+        false positive between tenants OR between products.
+
+    `ix_fiscal_documents_product_tenant_branch_status` replaces the auto's
+    `ix_fiscal_documents_org_branch_status` (porte table: anti-oracle 404
+    is now by `(product_id, tenant_ref)`, and `GET /v1/documentos` filters
+    by `tenant_ref`/`branch_ref`/`status` per the design spec's API
+    table) -- the index covers exactly that filter shape."""
+
+    __tablename__ = "fiscal_documents"
+    __table_args__ = (
+        Index(
+            "ix_fiscal_documents_product_tenant_branch_status",
+            "product_id",
+            "tenant_ref",
+            "branch_ref",
+            "status",
+        ),
+        UniqueConstraint("chave_acesso", name="uq_fiscal_documents_chave_acesso"),
+    )
+
+    product_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("products.id"), nullable=False)
+    tenant_ref: Mapped[str] = mapped_column(String(64), nullable=False)
+    branch_ref: Mapped[str] = mapped_column(String(64), nullable=False)
+    series_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("fiscal_series.id"), nullable=False)
+    document_model: Mapped[str] = mapped_column(String(2), nullable=False)
+    serie: Mapped[int] = mapped_column(Integer, nullable=False)
+    numero: Mapped[int] = mapped_column(Integer, nullable=False)
+    chave_acesso: Mapped[str] = mapped_column(String(44), nullable=False)
+    codigo_numerico: Mapped[str] = mapped_column(String(8), nullable=False)
+    status: Mapped[str] = mapped_column(String(20), nullable=False)
+    ambiente: Mapped[str] = mapped_column(String(12), nullable=False)
+    xml_assinado: Mapped[str] = mapped_column(Text, nullable=False)
+    rejeicao_codigo: Mapped[str | None] = mapped_column(String(10), nullable=True)
+    rejeicao_motivo: Mapped[str | None] = mapped_column(String(500), nullable=True)
+    protocolo: Mapped[str | None] = mapped_column(String(20), nullable=True)
+    autorizada_em: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)

src/fiscal_svc/documents/service.py

@@ -0,0 +1,124 @@
+import uuid
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from fiscal_svc.documents.models import FiscalDocumentModel, FiscalSeries
+
+
+class FiscalSeriesNotFoundError(Exception):
+    """Ported from the auto's `tenants.service.FiscalSeriesNotFoundError`.
+    Raised when a `(product_id, tenant_ref, branch_ref, document_model,
+    serie)` tuple does not resolve to a live row -- deliberately raised
+    both when the series truly does not exist and when it belongs to
+    another product/tenant/branch, so a cross-tenant lookup can't be used
+    to distinguish "not found" from "not yours" (anti-oracle, porte table:
+    "por org" -> "por (product_id, tenant_ref)")."""
+
+    def __init__(
+        self,
+        product_id: uuid.UUID,
+        tenant_ref: str,
+        branch_ref: str,
+        document_model: FiscalDocumentModel | str,
+        serie: int,
+    ):
+        self.product_id = product_id
+        self.tenant_ref = tenant_ref
+        self.branch_ref = branch_ref
+        self.document_model = document_model
+        self.serie = serie
+        super().__init__(
+            f"Série fiscal não encontrada para product_id={product_id}, "
+            f"tenant_ref={tenant_ref}, branch_ref={branch_ref}, "
+            f"modelo={document_model}, série={serie}"
+        )
+
+
+async def allocate_fiscal_number(
+    session: AsyncSession,
+    product_id: uuid.UUID,
+    tenant_ref: str,
+    branch_ref: str,
+    document_model: FiscalDocumentModel | str,
+    serie: int,
+) -> int:
+    """Ported verbatim (mechanism + contract) from the auto's
+    `tenants.service.allocate_fiscal_number`, with the porte table's
+    tenancy adaptation applied to the lookup (`organization_id`/`branch_id`
+    -> `product_id`/`tenant_ref`/`branch_ref`). Every invariant from the
+    original is preserved:
+
+    `SELECT ... FOR UPDATE` locks the SERIES ROW: two concurrent emissions
+    for the same `(product_id, tenant_ref, branch_ref, document_model,
+    serie)` never receive the same number -- the second waits for the
+    first's lock and reads the already-incremented `next_number`.
+
+    `.execution_options(populate_existing=True)` is OBRIGATÓRIO: the lock
+    happens in Postgres, but if this `FiscalSeries` object is already in
+    THIS session's identity map (e.g. an earlier unlocked read of the same
+    row, same session), SQLAlchemy's default behavior hands back the
+    CACHED object without repopulating it from the freshly-(re)locked row
+    -- even though the `FOR UPDATE` really did lock the real row. Without
+    this, `next_number` can stay pinned to a stale value and two callers
+    can allocate the SAME number. `fiscal_svc.core.db`'s
+    `expire_on_commit=False` makes this worse (the object never expires on
+    its own after a commit either). See
+    `auto/backend/app/modules/tenants/service.py::allocate_fiscal_number`'s
+    docstring and `tests/modules/tenants/test_fiscal_series.py::
+    test_allocate_fiscal_number_repopulates_identity_mapped_object` for the
+    original reproduction this ports the fix (and the test) for.
+
+    CONTRATO -- ESTA FUNÇÃO NÃO COMMITA. Only `flush()`s: the number is
+    allocated in memory (and the row stays LOCKED by this transaction's
+    `FOR UPDATE`) but nothing is durable until the CALLER commits. This is
+    deliberate: the real emission flow (Task 5) is "alocar número -> montar
+    o XML -> assinar -> persistir o `FiscalDocument` (com esse número)",
+    and allocation + the document INSERT must commit TOGETHER, atomically,
+    in the SAME transaction -- otherwise a failure between allocating and
+    inserting the document would burn a number with no record of what it
+    was for (and, worse, no `FiscalDocument` row to even know it needs
+    inutilização at SEFAZ later). A caller that never commits (a test that
+    only calls this and rolls back, for instance) leaves the allocation
+    with zero durable effect -- expected, not a bug.
+
+    Not an endpoint -- an INTERNAL function the emission service (Task 5)
+    calls. Each caller must pass its OWN `AsyncSession` (one transaction per
+    call): it's one session's `FOR UPDATE` blocking another session's
+    `SELECT` that actually serializes two concurrent allocations; two calls
+    sharing one session/transaction would have no real DB concurrency to
+    serialize."""
+    result = await session.execute(
+        select(FiscalSeries)
+        .where(
+            FiscalSeries.product_id == product_id,
+            FiscalSeries.tenant_ref == tenant_ref,
+            FiscalSeries.branch_ref == branch_ref,
+            FiscalSeries.document_model == (
+                document_model.value
+                if isinstance(document_model, FiscalDocumentModel)
+                else document_model
+            ),
+            FiscalSeries.serie == serie,
+            FiscalSeries.deleted_at.is_(None),
+        )
+        .with_for_update()
+        .execution_options(populate_existing=True)
+    )
+    series = result.scalar_one_or_none()
+    if series is None:
+        raise FiscalSeriesNotFoundError(
+            product_id=product_id,
+            tenant_ref=tenant_ref,
+            branch_ref=branch_ref,
+            document_model=document_model,
+            serie=serie,
+        )
+    allocated = series.next_number
+    series.next_number = allocated + 1
+    # NÃO commita -- ver a seção "CONTRATO" do docstring. O caller commita,
+    # idealmente junto com o INSERT do FiscalDocument que consome este
+    # número. `flush()` garante que o UPDATE já foi enviado ao Postgres (o
+    # lock do FOR UPDATE segue travado até o commit do caller).
+    await session.flush()
+    return allocated

tests/documents/test_allocate_fiscal_number.py

@@ -0,0 +1,274 @@
+"""Task 3: `documents.service.allocate_fiscal_number` -- ported from the
+auto's `tenants.service.allocate_fiscal_number`
+(`tests/modules/tenants/test_fiscal_series.py`), same mechanism/contract,
+tenancy adapted to `(product_id, tenant_ref, branch_ref)`. Covers: sequential
+allocation, the no-commit contract (Fix 1's two directions), the genuine
+N=10 concurrency guarantee, the identity-map staleness repro
+`populate_existing` exists to fix, and not-found (missing/soft-deleted/
+wrong product/wrong tenant/wrong branch -- every dimension of the new
+tenancy, not just product)."""
+import asyncio
+import uuid
+
+import pytest
+from sqlalchemy.ext.asyncio import async_sessionmaker
+
+from fiscal_svc.documents.models import FiscalDocumentModel, FiscalSeries
+from fiscal_svc.documents.service import FiscalSeriesNotFoundError, allocate_fiscal_number
+from fiscal_svc.tenancy.service import create_product
+
+
+async def _product(db_session, name="auto"):
+    return await create_product(db_session, name=name, api_key=f"k-{uuid.uuid4().hex}")
+
+
+async def _series(db_session, product, *, tenant_ref="tenant-1", branch_ref="branch-1", next_number=1014):
+    series = FiscalSeries(
+        product_id=product.id,
+        tenant_ref=tenant_ref,
+        branch_ref=branch_ref,
+        document_model=FiscalDocumentModel.NFE_55.value,
+        serie=1,
+        next_number=next_number,
+    )
+    db_session.add(series)
+    await db_session.commit()
+    await db_session.refresh(series)
+    return series
+
+
+@pytest.mark.asyncio
+async def test_allocate_fiscal_number_sequential_calls_increment(db_session):
+    product = await _product(db_session)
+    series = await _series(db_session, product)
+
+    first = await allocate_fiscal_number(
+        db_session, product.id, series.tenant_ref, series.branch_ref, FiscalDocumentModel.NFE_55, 1
+    )
+    await db_session.commit()
+    second = await allocate_fiscal_number(
+        db_session, product.id, series.tenant_ref, series.branch_ref, FiscalDocumentModel.NFE_55, 1
+    )
+    await db_session.commit()
+
+    assert first == 1014
+    assert second == 1015
+    await db_session.refresh(series)
+    assert series.next_number == 1016
+
+
+@pytest.mark.asyncio
+async def test_allocate_fiscal_number_accepts_the_raw_string_document_model(db_session):
+    """`document_model` is a plain `String(2)` column (zero enum PG) -- the
+    function must accept either the `FiscalDocumentModel` enum member or
+    its raw `.value` string, since callers (Task 4/5) may hold either."""
+    product = await _product(db_session)
+    series = await _series(db_session, product)
+
+    allocated = await allocate_fiscal_number(
+        db_session, product.id, series.tenant_ref, series.branch_ref, "55", 1
+    )
+    await db_session.commit()
+
+    assert allocated == 1014
+
+
+@pytest.mark.asyncio
+async def test_allocate_fiscal_number_without_commit_does_not_persist(test_engine, db_session):
+    """Fix 1's contract, safe direction: a caller that allocates and then
+    rolls back (or never commits) leaves NO durable trace -- the next
+    allocation must hand out the SAME number again."""
+    product = await _product(db_session)
+    series = await _series(db_session, product, next_number=100)
+
+    session_maker = async_sessionmaker(test_engine, expire_on_commit=False)
+
+    rollback_session = session_maker()
+    try:
+        allocated = await allocate_fiscal_number(
+            rollback_session, product.id, series.tenant_ref, series.branch_ref,
+            FiscalDocumentModel.NFE_55, 1,
+        )
+        assert allocated == 100
+        await rollback_session.rollback()
+    finally:
+        await rollback_session.close()
+
+    await db_session.refresh(series)
+    assert series.next_number == 100, (
+        f"expected next_number to revert to 100 after rollback, got "
+        f"{series.next_number} -- the un-committed allocation leaked a "
+        "durable side effect"
+    )
+
+    reallocated = await allocate_fiscal_number(
+        db_session, product.id, series.tenant_ref, series.branch_ref, FiscalDocumentModel.NFE_55, 1
+    )
+    await db_session.commit()
+    assert reallocated == 100
+
+
+@pytest.mark.asyncio
+async def test_allocate_fiscal_number_missing_series_raises(db_session):
+    product = await _product(db_session)
+
+    with pytest.raises(FiscalSeriesNotFoundError):
+        await allocate_fiscal_number(
+            db_session, product.id, "tenant-1", "branch-1", FiscalDocumentModel.NFE_55, 1
+        )
+
+
+@pytest.mark.asyncio
+async def test_allocate_fiscal_number_soft_deleted_series_raises(db_session):
+    from datetime import datetime, timezone
+
+    product = await _product(db_session)
+    series = await _series(db_session, product)
+    series.deleted_at = datetime.now(timezone.utc)
+    await db_session.commit()
+
+    with pytest.raises(FiscalSeriesNotFoundError):
+        await allocate_fiscal_number(
+            db_session, product.id, series.tenant_ref, series.branch_ref, FiscalDocumentModel.NFE_55, 1
+        )
+
+
+@pytest.mark.asyncio
+async def test_allocate_fiscal_number_other_product_raises(db_session):
+    """Tenancy-scoping guard (porte adaptation): the SAME
+    tenant_ref/branch_ref/document_model/serie under a DIFFERENT product_id
+    must never allocate from a series that belongs to a different product --
+    the anti-oracle boundary is (product_id, tenant_ref), so this exercises
+    the product_id half of it."""
+    product_a = await _product(db_session, name="auto")
+    series = await _series(db_session, product_a)
+    product_b = await _product(db_session, name="crm")
+
+    with pytest.raises(FiscalSeriesNotFoundError):
+        await allocate_fiscal_number(
+            db_session, product_b.id, series.tenant_ref, series.branch_ref,
+            FiscalDocumentModel.NFE_55, 1,
+        )
+
+
+@pytest.mark.asyncio
+async def test_allocate_fiscal_number_other_tenant_ref_raises(db_session):
+    """Same tenancy boundary, tenant_ref half: two tenants of the SAME
+    product must not share a series just because branch_ref/model/serie
+    coincide."""
+    product = await _product(db_session)
+    series = await _series(db_session, product, tenant_ref="tenant-a")
+
+    with pytest.raises(FiscalSeriesNotFoundError):
+        await allocate_fiscal_number(
+            db_session, product.id, "tenant-b", series.branch_ref, FiscalDocumentModel.NFE_55, 1
+        )
+
+
+@pytest.mark.asyncio
+async def test_allocate_fiscal_number_other_branch_ref_raises(db_session):
+    """Same tenancy boundary, branch_ref half -- the auto's own guard test
+    (`test_allocate_fiscal_number_other_org_raises`) had no branch-level
+    equivalent since `branch_id` there is a real FK scoped by
+    `organization_id`; here `branch_ref` is its own opaque dimension."""
+    product = await _product(db_session)
+    series = await _series(db_session, product, branch_ref="branch-a")
+
+    with pytest.raises(FiscalSeriesNotFoundError):
+        await allocate_fiscal_number(
+            db_session, product.id, series.tenant_ref, "branch-b", FiscalDocumentModel.NFE_55, 1
+        )
+
+
+@pytest.mark.asyncio
+async def test_allocate_fiscal_number_concurrent_calls_produce_distinct_sequential_numbers(
+    test_engine, db_session
+):
+    """The regression guard this whole feature exists for, ported verbatim
+    from the auto's own concurrency test: N=10 GENUINELY concurrent
+    `allocate_fiscal_number` calls against the SAME series must produce 10
+    DISTINCT, SEQUENTIAL numbers. Each call uses its OWN
+    `AsyncSession(test_engine)` (a real, separate DB connection) so the
+    event loop can genuinely interleave their network round-trips."""
+    product = await _product(db_session)
+    series = await _series(db_session, product, next_number=100)
+
+    async def _allocate_and_commit(session):
+        allocated = await allocate_fiscal_number(
+            session, product.id, series.tenant_ref, series.branch_ref,
+            FiscalDocumentModel.NFE_55, 1,
+        )
+        await session.commit()
+        return allocated
+
+    session_maker = async_sessionmaker(test_engine, expire_on_commit=False)
+    sessions = [session_maker() for _ in range(10)]
+    try:
+        results = await asyncio.gather(
+            *(_allocate_and_commit(session) for session in sessions),
+            return_exceptions=True,
+        )
+    finally:
+        for session in sessions:
+            await session.close()
+
+    errors = [r for r in results if isinstance(r, BaseException)]
+    assert not errors, f"expected all 10 concurrent allocations to succeed, got errors={errors!r}"
+
+    allocated = sorted(results)
+    assert allocated == list(range(100, 110)), (
+        f"expected 10 distinct sequential numbers 100..109, got {allocated!r} "
+        "(a repeat means two concurrent callers received the same NF-e "
+        "number; a gap means one was skipped)"
+    )
+
+    await db_session.refresh(series)
+    assert series.next_number == 110
+
+
+@pytest.mark.asyncio
+async def test_allocate_fiscal_number_repopulates_identity_mapped_object(test_engine, db_session):
+    """`SELECT ... FOR UPDATE` locks the ROW in Postgres, but if the calling
+    session already has that row's Python object in its IDENTITY MAP, plain
+    SQLAlchemy does NOT repopulate that object's attributes from the newly
+    fetched row by default. Reproduced with ordinary sequential awaits
+    across two sessions (no `asyncio.gather` needed) -- ported from the
+    auto's own reproduction."""
+    from fiscal_svc.documents.models import FiscalSeries as _FiscalSeries
+    from sqlalchemy import select
+
+    product = await _product(db_session)
+    series = await _series(db_session, product, next_number=100)
+
+    # Step 1: put the series into S1 (db_session)'s identity map with
+    # next_number=100.
+    loaded = (
+        await db_session.execute(select(_FiscalSeries).where(_FiscalSeries.id == series.id))
+    ).scalar_one()
+    assert loaded.next_number == 100
+
+    # Step 2: a genuinely separate session allocates 100 and commits -- the
+    # DB row now holds 101.
+    session_maker = async_sessionmaker(test_engine, expire_on_commit=False)
+    other_session = session_maker()
+    try:
+        other_allocated = await allocate_fiscal_number(
+            other_session, product.id, series.tenant_ref, series.branch_ref,
+            FiscalDocumentModel.NFE_55, 1,
+        )
+        await other_session.commit()
+    finally:
+        await other_session.close()
+    assert other_allocated == 100
+
+    # Step 3: S1 allocates next -- MUST be 101 (the real, post-S2 value),
+    # never 100 again (that would be a duplicate NF-e number).
+    allocated = await allocate_fiscal_number(
+        db_session, product.id, series.tenant_ref, series.branch_ref, FiscalDocumentModel.NFE_55, 1
+    )
+    await db_session.commit()
+    assert allocated == 101, (
+        f"expected 101 (the value S2 already committed), got {allocated} -- "
+        "S1's allocate_fiscal_number returned a STALE identity-map-cached "
+        "next_number instead of repopulating from the FOR UPDATE-locked row"
+    )

tests/migrations/test_fiscal_documents_schema.py

@@ -0,0 +1,306 @@
+"""Migration test (Task 3): `fiscal_series` + `fiscal_certificates` +
+`fiscal_documents` -- table/columns exist via `information_schema` over the
+`create_all` schema (Teste A), and each round-trips (INSERT cru via the
+real ORM) on a database built PURELY by `alembic upgrade head` (Teste B) --
+same two-test precedent as `test_products_schema.py` (Task 2) and the
+auto's own `tests/migrations/test_fiscal_*_schema.py`. Also proves the
+constraints Global Constraints calls out by name: UNIQUE `chave_acesso`,
+the partial-unique cert-vivo-per-`(product_id, branch_ref)` index, and the
+UNIQUE `(product_id, tenant_ref, branch_ref, document_model, serie)` on
+`fiscal_series`."""
+from datetime import datetime, timedelta, timezone
+
+import pytest
+import pytest_asyncio
+from sqlalchemy import select, text
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+# Import de registro: garante que os modelos relevantes estão registrados em
+# Base.metadata antes do create_all da fixture test_engine, ao rodar este
+# arquivo isolado.
+from fiscal_svc.documents import models as _documents_models  # noqa: F401
+from fiscal_svc.tenancy import models as _tenancy_models  # noqa: F401
+from tests.migrations._helpers import run_alembic, run_psql
+
+_MIGRATION_DB_NAME = "fiscal_svc_test_fiscal_documents_schema"
+_MIGRATION_DB_URL = (
+    f"postgresql+asyncpg://postgres:postgres@localhost:5432/{_MIGRATION_DB_NAME}"
+)
+
+
+@pytest.mark.asyncio
+async def test_fiscal_series_table_and_columns_exist(db_session):
+    rows = {
+        r[0]
+        for r in (
+            await db_session.execute(
+                text(
+                    "select column_name from information_schema.columns "
+                    "where table_name='fiscal_series'"
+                )
+            )
+        )
+    }
+    assert {
+        "id", "product_id", "tenant_ref", "branch_ref", "document_model",
+        "serie", "next_number", "created_at", "updated_at", "deleted_at",
+    } <= rows
+
+
+@pytest.mark.asyncio
+async def test_fiscal_certificates_table_and_columns_exist(db_session):
+    rows = {
+        r[0]: r[1]
+        for r in (
+            await db_session.execute(
+                text(
+                    "select column_name, is_nullable from information_schema.columns "
+                    "where table_name='fiscal_certificates'"
+                )
+            )
+        )
+    }
+    for required in (
+        "product_id", "tenant_ref", "branch_ref", "cnpj", "pfx_encrypted",
+        "password_encrypted", "subject_cn", "cnpj_certificado",
+        "not_valid_before", "not_valid_after", "deleted_at",
+    ):
+        assert required in rows, f"coluna {required} ausente em fiscal_certificates"
+    assert rows["pfx_encrypted"] == "NO"
+    assert rows["deleted_at"] == "YES"
+
+
+@pytest.mark.asyncio
+async def test_fiscal_documents_table_and_columns_exist(db_session):
+    rows = {
+        r[0]: r[1]
+        for r in (
+            await db_session.execute(
+                text(
+                    "select column_name, is_nullable from information_schema.columns "
+                    "where table_name='fiscal_documents'"
+                )
+            )
+        )
+    }
+    for required in (
+        "product_id", "tenant_ref", "branch_ref", "series_id", "document_model",
+        "serie", "numero", "chave_acesso", "codigo_numerico", "status",
+        "ambiente", "xml_assinado", "rejeicao_codigo", "rejeicao_motivo",
+        "protocolo", "autorizada_em", "deleted_at",
+    ):
+        assert required in rows, f"coluna {required} ausente em fiscal_documents"
+    assert rows["chave_acesso"] == "NO"
+    # sale_id/service_order_id are DELIBERATELY absent -- porte table.
+    assert "sale_id" not in rows
+    assert "service_order_id" not in rows
+
+
+@pytest_asyncio.fixture
+async def migration_database():
+    run_psql("-c", f"DROP DATABASE IF EXISTS {_MIGRATION_DB_NAME} WITH (FORCE);")
+    run_psql("-c", f"CREATE DATABASE {_MIGRATION_DB_NAME};")
+    yield
+    run_psql("-c", f"DROP DATABASE IF EXISTS {_MIGRATION_DB_NAME} WITH (FORCE);")
+
+
+async def _make_product(session, name="auto"):
+    from fiscal_svc.tenancy.service import create_product
+
+    return await create_product(session, name=name, api_key=f"key-{name}-{id(session)}")
+
+
+@pytest.mark.asyncio
+async def test_fiscal_series_and_certificate_and_document_round_trip_on_a_real_migrated_database(
+    migration_database,
+):
+    result = run_alembic(_MIGRATION_DB_URL, "upgrade", "head")
+    assert result.returncode == 0, result.stderr
+
+    from fiscal_svc.documents.models import FiscalCertificate, FiscalDocument, FiscalSeries
+
+    engine = create_async_engine(_MIGRATION_DB_URL, echo=False)
+    session_maker = async_sessionmaker(engine, expire_on_commit=False)
+    try:
+        async with session_maker() as session:
+            product = await _make_product(session)
+
+            series = FiscalSeries(
+                product_id=product.id,
+                tenant_ref="tenant-1",
+                branch_ref="branch-1",
+                document_model="55",
+                serie=1,
+                next_number=1014,
+            )
+            session.add(series)
+            await session.flush()
+
+            certificate = FiscalCertificate(
+                product_id=product.id,
+                tenant_ref="tenant-1",
+                branch_ref="branch-1",
+                cnpj="14200166000187",
+                pfx_encrypted=b"\x00\x01ciphertext-pfx",
+                password_encrypted=b"\x00\x02ciphertext-pw",
+                subject_cn="EMPRESA TESTE LTDA:14200166000187",
+                cnpj_certificado="14200166000187",
+                not_valid_before=datetime.now(timezone.utc) - timedelta(days=1),
+                not_valid_after=datetime.now(timezone.utc) + timedelta(days=365),
+            )
+            session.add(certificate)
+
+            document = FiscalDocument(
+                product_id=product.id,
+                tenant_ref="tenant-1",
+                branch_ref="branch-1",
+                series_id=series.id,
+                document_model="55",
+                serie=1,
+                numero=1014,
+                chave_acesso="4" * 44,
+                codigo_numerico="12345678",
+                status="ASSINADO",
+                ambiente="homologacao",
+                xml_assinado="<NFe/>",
+            )
+            session.add(document)
+
+            # Would crash here (UndefinedColumn/DataError) before the fix.
+            await session.commit()
+
+            series_id, certificate_id, document_id = series.id, certificate.id, document.id
+
+        async with session_maker() as session:
+            reloaded_series = await session.get(FiscalSeries, series_id)
+            assert reloaded_series.next_number == 1014
+
+            reloaded_cert = await session.get(FiscalCertificate, certificate_id)
+            assert reloaded_cert.cnpj_certificado == "14200166000187"
+            assert reloaded_cert.pfx_encrypted == b"\x00\x01ciphertext-pfx"
+
+            reloaded_doc = await session.get(FiscalDocument, document_id)
+            assert reloaded_doc.status == "ASSINADO"
+            assert reloaded_doc.chave_acesso == "4" * 44
+
+            result_q = await session.execute(
+                select(FiscalDocument).where(FiscalDocument.chave_acesso == "4" * 44)
+            )
+            assert result_q.scalar_one().id == document_id
+    finally:
+        await engine.dispose()
+
+
+@pytest.mark.asyncio
+async def test_chave_acesso_unique_constraint_holds_on_real_migration(migration_database):
+    result = run_alembic(_MIGRATION_DB_URL, "upgrade", "head")
+    assert result.returncode == 0, result.stderr
+
+    from fiscal_svc.documents.models import FiscalDocument, FiscalSeries
+
+    engine = create_async_engine(_MIGRATION_DB_URL, echo=False)
+    session_maker = async_sessionmaker(engine, expire_on_commit=False)
+    try:
+        async with session_maker() as session:
+            product = await _make_product(session, name="auto-chave")
+            series = FiscalSeries(
+                product_id=product.id, tenant_ref="t1", branch_ref="b1",
+                document_model="55", serie=1, next_number=1,
+            )
+            session.add(series)
+            await session.flush()
+
+            def _doc(numero):
+                return FiscalDocument(
+                    product_id=product.id, tenant_ref="t1", branch_ref="b1",
+                    series_id=series.id, document_model="55", serie=1, numero=numero,
+                    chave_acesso="9" * 44, codigo_numerico="12345678",
+                    status="ASSINADO", ambiente="homologacao", xml_assinado="<NFe/>",
+                )
+
+            session.add(_doc(1))
+            await session.commit()
+
+            session.add(_doc(2))
+            with pytest.raises(IntegrityError):
+                await session.commit()
+            await session.rollback()
+    finally:
+        await engine.dispose()
+
+
+@pytest.mark.asyncio
+async def test_two_live_certificates_for_same_product_branch_violate_unique_index_on_real_migration(
+    migration_database,
+):
+    """Same fix/reasoning as the auto's `a1b2c3d4e5f6` migration: a PARTIAL
+    UNIQUE index (here on `(product_id, branch_ref) WHERE deleted_at IS
+    NULL`) makes two concurrently-uploaded LIVE certificates for the same
+    slot structurally impossible, not just avoided by the service layer."""
+    result = run_alembic(_MIGRATION_DB_URL, "upgrade", "head")
+    assert result.returncode == 0, result.stderr
+
+    from fiscal_svc.documents.models import FiscalCertificate
+
+    engine = create_async_engine(_MIGRATION_DB_URL, echo=False)
+    session_maker = async_sessionmaker(engine, expire_on_commit=False)
+    try:
+        async with session_maker() as session:
+            product = await _make_product(session, name="auto-cert-corrida")
+
+            def _cert():
+                return FiscalCertificate(
+                    product_id=product.id,
+                    tenant_ref="t1",
+                    branch_ref="b1",
+                    cnpj="14200166000280",
+                    pfx_encrypted=b"\x00pfx",
+                    password_encrypted=b"\x00pw",
+                    subject_cn="EMPRESA TESTE LTDA:14200166000280",
+                    cnpj_certificado="14200166000280",
+                    not_valid_before=datetime.now(timezone.utc) - timedelta(days=1),
+                    not_valid_after=datetime.now(timezone.utc) + timedelta(days=365),
+                )
+
+            session.add(_cert())
+            await session.commit()
+
+            session.add(_cert())
+            with pytest.raises(IntegrityError):
+                await session.commit()
+            await session.rollback()
+    finally:
+        await engine.dispose()
+
+
+@pytest.mark.asyncio
+async def test_duplicate_fiscal_series_tuple_violates_unique_constraint_on_real_migration(
+    migration_database,
+):
+    result = run_alembic(_MIGRATION_DB_URL, "upgrade", "head")
+    assert result.returncode == 0, result.stderr
+
+    from fiscal_svc.documents.models import FiscalSeries
+
+    engine = create_async_engine(_MIGRATION_DB_URL, echo=False)
+    session_maker = async_sessionmaker(engine, expire_on_commit=False)
+    try:
+        async with session_maker() as session:
+            product = await _make_product(session, name="auto-serie-dup")
+
+            def _series():
+                return FiscalSeries(
+                    product_id=product.id, tenant_ref="t1", branch_ref="b1",
+                    document_model="55", serie=1, next_number=1,
+                )
+
+            session.add(_series())
+            await session.commit()
+
+            session.add(_series())
+            with pytest.raises(IntegrityError):
+                await session.commit()
+            await session.rollback()
+    finally:
+        await engine.dispose()

tests/shared/test_for_update_populate_existing.py

@@ -0,0 +1,189 @@
+"""Ported verbatim (mechanism unchanged, only the scanned directory) from
+`auto/backend/tests/shared/test_for_update_populate_existing.py` -- the
+static AST guard for the exact bug `documents.service.allocate_fiscal_number`
+exists to avoid: `SELECT ... FOR UPDATE` locks the row in Postgres, but if
+the calling session already has that row's object in its identity map (an
+earlier unlocked fetch of the same object, same session), SQLAlchemy's
+default behavior hands back the CACHED object instead of repopulating it
+from the freshly-(re)locked row. The Postgres lock is real; the in-memory
+object the code decides against is a stale snapshot.
+`.execution_options(populate_existing=True)` is the fix -- see
+`fiscal_svc.documents.service.allocate_fiscal_number`'s docstring for the
+full write-up (ported from the auto's own `tenants.service.
+allocate_fiscal_number`).
+
+Scans every `.py` file under `src/fiscal_svc/` (never `tests/`) with the
+`ast` module -- deliberately not regex over the source text, because a
+fluent call chain like
+`select(...).where(...).with_for_update().execution_options(...)` routinely
+breaks across several lines/parens, and a text-based check would either
+miss those or need to reimplement a parser badly. Requires that every
+`.with_for_update()` call has a `.execution_options(populate_existing=
+True)` call somewhere in the SAME chain of method calls."""
+from __future__ import annotations
+
+import ast
+from pathlib import Path
+
+SRC_DIR = Path(__file__).resolve().parents[2] / "src" / "fiscal_svc"
+REPO_DIR = SRC_DIR.parents[1]
+
+_FIX_EXPLANATION = """
+WHY: `.with_for_update()` alone locks the row in Postgres, but if this
+session already has an object for that row in its identity map (e.g. an
+earlier unlocked lookup of the same row, same session, done for a 404/
+access check before calling into the locking code), SQLAlchemy's default
+identity-map behavior returns that CACHED object instead of repopulating it
+from the row `FOR UPDATE` just (re)read. The lock becomes decorative: the
+code holds a real Postgres lock on a row it never actually re-reads, and
+makes its decision against a stale in-memory snapshot instead.
+
+HOW TO FIX: chain `.execution_options(populate_existing=True)` onto the
+SAME query as the `.with_for_update()`, e.g.:
+
+    query = query.with_for_update().execution_options(populate_existing=True)
+
+or, split across lines/parens, as long as it's the same chain:
+
+    result = await session.execute(
+        select(Model)
+        .where(Model.id == model_id)
+        .with_for_update()
+        .execution_options(populate_existing=True)
+    )
+
+If you have found a `FOR UPDATE` that must NOT populate_existing (none exist
+today -- this would be unusual), do not just delete this check or special-
+case your file/line here. Talk to the team about adding an explicit,
+commented opt-out marker to this test first, so the next reader still gets
+an explanation instead of a silently-shrinking guard.
+""".strip()
+
+
+def _is_populate_existing_true_kwarg(call: ast.Call) -> bool:
+    """True if `call` is `.execution_options(...)` carrying a literal
+    `populate_existing=True` keyword argument."""
+    if not (isinstance(call.func, ast.Attribute) and call.func.attr == "execution_options"):
+        return False
+    return any(
+        kw.arg == "populate_existing" and isinstance(kw.value, ast.Constant) and kw.value.value is True
+        for kw in call.keywords
+    )
+
+
+def _build_parent_map(tree: ast.AST) -> dict[int, ast.AST]:
+    parents: dict[int, ast.AST] = {}
+    for parent in ast.walk(tree):
+        for child in ast.iter_child_nodes(parent):
+            parents[id(child)] = parent
+    return parents
+
+
+def _chain_root(node: ast.AST, parents: dict[int, ast.AST]) -> ast.AST:
+    """Walk up from `node` while still inside the same fluent method-call
+    chain -- i.e. while each successive parent is itself a `Call` or
+    `Attribute` node, the two node shapes chaining (`a.b().c().d()`) is
+    built from in the AST. This is what lets a chain split across many
+    lines/parens (still a single expression to the parser) be treated as
+    one unit, and stops at the first non-chain boundary (assignment,
+    statement, argument to an unrelated call, ...).
+    """
+    current = node
+    while True:
+        parent = parents.get(id(current))
+        if parent is None or not isinstance(parent, (ast.Call, ast.Attribute)):
+            return current
+        current = parent
+
+
+def _find_violations(src_dir: Path) -> list[str]:
+    """Returns one `path:line` string per `.with_for_update()` call site
+    under `src_dir` that does not have a `.execution_options(
+    populate_existing=True)` call in the same method-call chain."""
+    violations: list[str] = []
+    for path in sorted(src_dir.rglob("*.py")):
+        source = path.read_text()
+        try:
+            tree = ast.parse(source, filename=str(path))
+        except SyntaxError:
+            continue
+        parents = _build_parent_map(tree)
+        for node in ast.walk(tree):
+            if not (
+                isinstance(node, ast.Call)
+                and isinstance(node.func, ast.Attribute)
+                and node.func.attr == "with_for_update"
+            ):
+                continue
+            root = _chain_root(node, parents)
+            has_populate_existing = any(
+                isinstance(candidate, ast.Call) and _is_populate_existing_true_kwarg(candidate)
+                for candidate in ast.walk(root)
+            )
+            if not has_populate_existing:
+                # `path.relative_to(REPO_DIR)` when `src_dir` is actually
+                # inside this repo (the real guard below); falls back to the
+                # absolute path when it isn't (the synthetic-`tmp_path`
+                # self-tests further down, which scan a throwaway directory
+                # outside the repo entirely).
+                try:
+                    rel = path.relative_to(REPO_DIR)
+                except ValueError:
+                    rel = path
+                # `end_lineno`, not `lineno`: for a multi-line chain, a
+                # Call/Attribute node's `lineno` is inherited from where the
+                # WHOLE expression starts (e.g. the `select(...)` at the top
+                # of the chain), while `end_lineno` lands on the line the
+                # `.with_for_update()` text itself is on -- the line the
+                # next developer actually needs to look at.
+                violations.append(f"{rel}:{node.end_lineno}")
+    return violations
+
+
+def test_every_with_for_update_call_site_chains_populate_existing():
+    violations = _find_violations(SRC_DIR)
+    assert not violations, (
+        "Found `.with_for_update()` call site(s) missing "
+        "`.execution_options(populate_existing=True)` in the same call "
+        "chain:\n  " + "\n  ".join(violations) + "\n\n" + _FIX_EXPLANATION
+    )
+
+
+# --- Detection proven in BOTH directions (Global Constraints) --------------
+#
+# The test above only proves the guard is currently GREEN against this
+# repo's real code -- on its own that's equally consistent with "the guard
+# actually detects the bug" and "the guard is a no-op that always passes".
+# These two exercise `_find_violations` directly against synthetic files in
+# a throwaway directory, so the guard's OWN detection logic is proven both
+# ways: it flags the missing fix, and it does not false-positive on the
+# fixed shape (including a chain split across lines, the exact shape
+# `allocate_fiscal_number`'s real call site uses).
+
+
+def test_find_violations_flags_with_for_update_missing_populate_existing(tmp_path):
+    (tmp_path / "offender.py").write_text(
+        "async def f(session):\n"
+        "    return await session.execute(\n"
+        "        select(Model).where(Model.id == x).with_for_update()\n"
+        "    )\n"
+    )
+
+    violations = _find_violations(tmp_path)
+
+    assert len(violations) == 1
+    assert "offender.py" in violations[0]
+
+
+def test_find_violations_does_not_flag_a_correctly_fixed_multiline_chain(tmp_path):
+    (tmp_path / "fixed.py").write_text(
+        "async def f(session):\n"
+        "    return await session.execute(\n"
+        "        select(Model)\n"
+        "        .where(Model.id == x)\n"
+        "        .with_for_update()\n"
+        "        .execution_options(populate_existing=True)\n"
+        "    )\n"
+    )
+
+    assert _find_violations(tmp_path) == []