fix: create workspace_entities/workspace_documents registry and provision Workspace rows

backend/alembic/versions/0001_initial_control_plane.py

@@ -9,7 +9,7 @@
 from sqlalchemy import text
 
 from db.models import Base
-from scripts.bootstrap_db import schema_backfill_sql
+from scripts.bootstrap_db import execute_schema_backfill
 
 revision = "0001_initial_control_plane"
 down_revision = None
@@ -19,8 +19,7 @@ def upgrade() -> None:
     connection = op.get_bind()
     connection.execute(text("CREATE EXTENSION IF NOT EXISTS vector"))
     Base.metadata.create_all(connection)
-    for statement in schema_backfill_sql():
-        connection.execute(statement)
+    execute_schema_backfill(connection)
 
 
 def downgrade() -> None:

backend/alembic/versions/0011_email_read_state.py

@@ -1,6 +1,14 @@
-"""Add is_read to emails (IMAP \\Seen read state).
+"""Add is_read to email_records (IMAP \\Seen read state).
 
 Existing rows default to read so historical/file imports do not surface as unread.
+
+Checks both "email_records" (the real, current table -- a database whose own
+0001_initial_control_plane ran before ``is_read`` was added to the ``Email``
+model has this table without the column, and needs it added) and "emails"
+(a legacy name that, per 0011_email_model_reconciliation's docstring, no
+migration in this repo's history ever actually created for a real managed
+database, but is checked defensively in case one somehow exists). Guarded by
+column existence, not just table existence, so it is safely idempotent.
 """
 
 from alembic import op
@@ -12,18 +20,40 @@
 branch_labels = None
 depends_on = None
 
+_CANDIDATE_TABLES = ("email_records", "emails")
+
 
 def upgrade() -> None:
-    op.add_column(
-        "emails",
-        sa.Column(
-            "is_read",
-            sa.Boolean(),
-            nullable=False,
-            server_default=sa.text("true"),
-        ),
-    )
+    inspector = sa.inspect(op.get_bind())
+    for table_name in _CANDIDATE_TABLES:
+        if inspector.has_table(table_name) and not _has_column(
+            inspector, table_name, "is_read"
+        ):
+            op.add_column(
+                table_name,
+                sa.Column(
+                    "is_read",
+                    sa.Boolean(),
+                    nullable=False,
+                    server_default=sa.text("true"),
+                ),
+            )
 
 
 def downgrade() -> None:
-    op.drop_column("emails", "is_read")
+    # Same ownership-ambiguity problem as 0018_workspace_registry's downgrade:
+    # a fresh database gets email_records.is_read from 0001's live
+    # Base.metadata.create_all, not from this revision, so there is no way to
+    # tell "this revision added the column" apart from "the baseline already
+    # had it" -- and is_read holds real per-message read/unread state, not
+    # rebuildable derived data. As with 0001_initial_control_plane and
+    # 0018_workspace_registry: production rollbacks should restore from
+    # backup or a later explicit down revision rather than dropping
+    # customer-owned data.
+    return None
+
+
+def _has_column(inspector, table_name: str, column_name: str) -> bool:
+    return any(
+        column["name"] == column_name for column in inspector.get_columns(table_name)
+    )

backend/alembic/versions/0016_document_org_scope.py

@@ -8,6 +8,13 @@
 recognition worker can resolve the owning organization's provider without
 joining through the (organization-less) workspace entity. Nullable and additive
 so existing rows and personal-scope documents are unaffected.
+
+A database that has never had ``workspace_documents`` at all (one that ran
+``0001_initial_control_plane`` before ``Workspace``/``Document`` existed in
+``db/models.py``, and has only applied incremental migrations since) has no
+table for this revision to alter. This revision is a no-op for that case;
+``0018_workspace_registry`` creates the table later in the chain, already
+including this column.
 """
 
 from alembic import op
@@ -24,6 +31,8 @@
 def upgrade() -> None:
     connection = op.get_bind()
     inspector = sa.inspect(connection)
+    if not inspector.has_table(_DOCUMENTS_TABLE):
+        return
     columns = {column["name"] for column in inspector.get_columns(_DOCUMENTS_TABLE)}
     if _ORG_COLUMN not in columns:
         op.add_column(
@@ -39,9 +48,7 @@ def upgrade() -> None:
 
 
 def downgrade() -> None:
-    connection = op.get_bind()
-    inspector = sa.inspect(connection)
-    columns = {column["name"] for column in inspector.get_columns(_DOCUMENTS_TABLE)}
-    op.drop_index(_ORG_INDEX, table_name=_DOCUMENTS_TABLE, if_exists=True)
-    if _ORG_COLUMN in columns:
-        op.drop_column(_DOCUMENTS_TABLE, _ORG_COLUMN)
+    # 0018 can create this table and column after 0016 was a no-op. Alembic
+    # cannot distinguish that case from a table altered by this revision, so
+    # dropping the column here could destroy later organization assignments.
+    return None

backend/alembic/versions/0018_workspace_registry.py

@@ -0,0 +1,81 @@
+"""create workspace registry and workspace document tables
+
+Revision ID: 0018_workspace_registry
+Revises: 0017_merge_newsdom_carddav_heads
+Create Date: 2026-09-01 00:00:00.000000
+
+``Workspace``/``Document`` (``workspace_entities``/``workspace_documents``) have
+been declared in ``db/models.py`` since before this repository's incremental
+migration history begins tracking them explicitly. A database that ran
+``0001_initial_control_plane``'s ``Base.metadata.create_all`` after these
+models existed already has both tables; a database that ran ``0001`` earlier
+and has only applied incremental migrations since never got them, so
+``/api/data/documents`` fails with an undefined-relation error the first time
+it is hit. This revision is idempotent (``has_table`` guarded) so it is a
+no-op for a database that already has the tables and a real fix for one that
+does not.
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+revision = "0018_workspace_registry"
+down_revision = "0017_merge_newsdom_carddav_heads"
+
+_ENTITIES_TABLE = "workspace_entities"
+_DOCUMENTS_TABLE = "workspace_documents"
+
+
+def upgrade() -> None:
+    connection = op.get_bind()
+    inspector = sa.inspect(connection)
+
+    if not inspector.has_table(_ENTITIES_TABLE):
+        op.create_table(
+            _ENTITIES_TABLE,
+            sa.Column("workspace_id", sa.String(), nullable=False),
+            sa.Column("workspace_name", sa.String(), nullable=False),
+            sa.Column("workspace_domain", sa.String(), nullable=True),
+            sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+            sa.PrimaryKeyConstraint("workspace_id"),
+        )
+
+    if not inspector.has_table(_DOCUMENTS_TABLE):
+        op.create_table(
+            _DOCUMENTS_TABLE,
+            sa.Column("document_id", sa.String(), nullable=False),
+            sa.Column("workspace_id", sa.String(), nullable=False),
+            sa.Column("organization_id", sa.String(), nullable=True),
+            sa.Column("document_name", sa.String(), nullable=False),
+            sa.Column("document_type", sa.String(), nullable=False),
+            sa.Column("document_content", sa.Text(), nullable=True),
+            sa.Column("document_status", sa.String(), nullable=False),
+            sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+            sa.ForeignKeyConstraint(
+                ["workspace_id"], [f"{_ENTITIES_TABLE}.workspace_id"]
+            ),
+            sa.PrimaryKeyConstraint("document_id"),
+        )
+
+    for index_name, column_names in (
+        ("ix_workspace_documents_workspace_id", ["workspace_id"]),
+        ("ix_workspace_documents_organization_id", ["organization_id"]),
+    ):
+        op.create_index(
+            index_name,
+            _DOCUMENTS_TABLE,
+            column_names,
+            if_not_exists=True,
+        )
+
+
+def downgrade() -> None:
+    # This revision's upgrade is a no-op whenever the tables already exist
+    # (e.g. created by 0001's create_all), so a downgrade cannot tell "this
+    # revision created these tables" apart from "they predate it" -- and
+    # workspace_documents.document_content holds real uploaded content, not
+    # rebuildable derived state. Unconditionally dropping it risks destroying
+    # data this revision never created. As with 0001_initial_control_plane:
+    # production rollbacks should restore from backup or a later explicit
+    # down revision rather than dropping customer-owned data.
+    return None

backend/alembic/versions/0019_email_read_state_repair.py

@@ -0,0 +1,57 @@
+"""idempotently ensure email_records has is_read
+
+Revision ID: 0019_email_read_state_repair
+Revises: 0018_workspace_registry
+Create Date: 2026-09-01 00:00:00.000000
+
+Alembic never re-runs a revision's ``upgrade()`` once that revision id is
+recorded as applied for a database -- editing ``0011_email_read_state.py``'s
+content cannot repair a database that already has "0011_email_read_state"
+in its ``alembic_version`` history but never actually got
+``email_records.is_read`` (whatever the reason: an earlier version of that
+revision that targeted the wrong table, a partial/interrupted apply, manual
+intervention). This revision is the real repair path for such a database:
+appended after the current head, so it runs regardless of what 0011 already
+did or didn't do. Idempotent (has_table/has_column guarded) so it is a
+no-op for every database that already has the column, from any path.
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+revision = "0019_email_read_state_repair"
+down_revision = "0018_workspace_registry"
+
+_EMAIL_TABLE = "email_records"
+
+
+def upgrade() -> None:
+    inspector = sa.inspect(op.get_bind())
+    if inspector.has_table(_EMAIL_TABLE) and not _has_column(
+        inspector, _EMAIL_TABLE, "is_read"
+    ):
+        op.add_column(
+            _EMAIL_TABLE,
+            sa.Column(
+                "is_read",
+                sa.Boolean(),
+                nullable=False,
+                server_default=sa.text("true"),
+            ),
+        )
+
+
+def downgrade() -> None:
+    # Same ownership-ambiguity reasoning as 0011_email_read_state and
+    # 0018_workspace_registry: this revision cannot tell whether it was the
+    # one that added the column (repairing a stamped-but-incomplete
+    # database) or the column already existed from another path, and
+    # is_read holds real read/unread state, not rebuildable derived data.
+    # No-op; production rollbacks should restore from backup.
+    return None
+
+
+def _has_column(inspector, table_name: str, column_name: str) -> bool:
+    return any(
+        column["name"] == column_name for column in inspector.get_columns(table_name)
+    )

backend/api/data.py

@@ -36,6 +36,7 @@
 )
 from services.ontology_service import ontology_service
 from services.webdav_service import webdav_service
+from services.workspace_scope import get_or_create_workspace
 
 router = APIRouter(prefix="/api/data", tags=["data"])
 
@@ -2484,6 +2485,36 @@ async def _count_scalar(db: AsyncSession, statement) -> int:
     return int(result.scalar_one() or 0)
 
 
+def _auth_context_owns_its_workspace(auth_context: AuthContext) -> bool:
+    """Whether ``auth_context.workspace_id`` is the canonical
+    ``workspace-<organization_id>``/``workspace-<user_id>`` derivation for
+    this session, rather than an internally inconsistent claim pairing."""
+    expected_workspace_id = (
+        f"workspace-{auth_context.organization_id}"
+        if auth_context.organization_id
+        else f"workspace-{auth_context.user_id}"
+    )
+    return auth_context.workspace_id == expected_workspace_id
+
+
+def _document_organization_filter(auth_context: AuthContext) -> ColumnElement:
+    """Scope a ``Document`` query to this session's organization.
+
+    Revision 0016_document_org_scope added ``organization_id`` without
+    backfilling existing rows, so a legacy document can have it unset. Such a
+    row is trusted to belong to whichever single organization the document's
+    (already-filtered) ``workspace_id`` canonically maps to -- but only when
+    this session's own workspace/organization pairing is that canonical
+    derivation, not an internally inconsistent claim.
+    """
+    organization_filter = Document.organization_id == auth_context.organization_id
+    if _auth_context_owns_its_workspace(auth_context):
+        organization_filter = or_(
+            organization_filter, Document.organization_id.is_(None)
+        )
+    return organization_filter
+
+
 async def _get_workspace_document(
     db: AsyncSession,
     auth_context: AuthContext,
@@ -2493,6 +2524,7 @@ async def _get_workspace_document(
         select(Document).where(
             Document.document_id == document_id,
             Document.workspace_id == auth_context.workspace_id,
+            _document_organization_filter(auth_context),
         )
     )
     document = result.scalar_one_or_none()
@@ -3166,6 +3198,7 @@ async def upload_data_document(
     auth_context: AuthContext = Depends(get_auth_context),
     db: AsyncSession = Depends(get_db),
 ) -> DataDocumentActionResponse:
+    await get_or_create_workspace(db, auth_context.workspace_id)
     document = Document(
         workspace_id=auth_context.workspace_id,
         organization_id=auth_context.organization_id,
@@ -3301,6 +3334,7 @@ async def upload_document_for_pdf_dom_recognition(
             status_code=415,
             detail="Only application/pdf uploads are supported for DOM recognition.",
         )
+    await get_or_create_workspace(db, auth_context.workspace_id)
     document = Document(
         workspace_id=auth_context.workspace_id,
         organization_id=auth_context.organization_id,
@@ -3931,7 +3965,10 @@ async def get_data_quality_surface(
     documents = await _scoped_rows(
         db,
         select(Document)
-        .where(Document.workspace_id == auth_context.workspace_id)
+        .where(
+            Document.workspace_id == auth_context.workspace_id,
+            _document_organization_filter(auth_context),
+        )
         .order_by(Document.created_at.desc(), Document.document_id.asc())
         .limit(8),
     )