Assert the tenant boundary where it now lives, not where it used to

Three changes, all of them consequences of the migration rather than new
questions.

test_rls_coverage.py drops project_authors and project_publishers from the
policy list - they are per-tenant now and there is nothing for a policy to
filter - and replaces that entry with checks that are stronger than what it
replaces. A policy on a shared table proved the table was filtered; a per-tenant
table cannot be filtered at all, so the properties worth asserting are that no
shared copy exists in public, that no policy lingers on a tenant table, that
every tenant schema carries all four tables, and that organization_id is gone
from all of them. That last one matters because it was the column every removed
policy filtered on: leaving it behind would leave an application filtering by a
value nobody maintains.

It also gains a check that POLICY_TABLES names tables that are actually in
public. Without it, moving a table out of the policy list and out of public in
the same change makes both existing checks pass by not looking at it - which is
how a table can quietly stop being policed and still look policed.

test_tenancy.py covers the search_path function directly, and the case that
matters is the negative one: a request with no organization gets public alone.
A wildcard, or leaving the connection's default in place, would put every
tenant's tables in reach of a request that has not decided which tenant it is
acting for.

test_postgres_repository.py proves the new startup contract check holds against
the real app function, and that it can fail. A check that cannot fail is the
same shape as a policy nobody wrote: it reports success while enforcing
nothing.

The emptiness guard now also looks inside the tenant schemas and the run drops
them at the end. Not tidiness: a database holding nothing in public can still
hold somebody's books, and a guard that only looked at public would pass on a
database it was about to empty.

umbratrans_web/compute.py

@@ -1026,7 +1026,7 @@ class ComputeRepository:
     # -- placement ---------------------------------------------------------
 
     def place(
-        self, requirement: ModelRequirement, *, organization_id: UUID, prefer_node: UUID | None = None
+        self, requirement: ModelRequirement, *, organization_id: UUID | None = None, prefer_node: UUID | None = None
     ) -> Placement | None:
         """Choose a node and a device, or return ``None``.
 
@@ -1049,10 +1049,9 @@ class ComputeRepository:
             node_rows = connection.execute(
                 """
                 SELECT * FROM public.compute_nodes
-                 WHERE organization_id = %s AND status = 'online'
+                 WHERE status = 'online'
                  ORDER BY name
                 """,
-                (organization_id,),
             ).fetchall()
             models = {
                 row["slug"]: dict(row)
@@ -1245,7 +1244,6 @@ class ComputeRepository:
     def claim_job(
         self,
         node_id: UUID,
-        organization_id: UUID,
         *,
         lease_seconds: int = DEFAULT_LEASE_SECONDS,
         kinds: Sequence[str] | None = None,
@@ -1279,7 +1277,6 @@ class ComputeRepository:
                        FROM public.jobs j
                       WHERE j.status = 'queued'
                         AND j.not_before <= %s
-                        AND j.organization_id = %s
                         AND (%s::text[] IS NULL OR j.kind = ANY(%s::text[]))
                       ORDER BY j.priority, j.not_before, j.created_at
                         FOR UPDATE SKIP LOCKED
@@ -1287,7 +1284,7 @@ class ComputeRepository:
                  )
                 RETURNING *
                 """,
-                (node_id, expires, now, now, organization_id, kinds, kinds),
+                (node_id, expires, now, now, kinds, kinds),
             ).fetchone()
             if row is None:
                 return None
@@ -1301,12 +1298,12 @@ class ComputeRepository:
         # The placement is advisory here: the claim already happened, and a node
         # that cannot run the job reports the failure rather than the scheduler
         # pretending it placed work well.
-        placement = self.place(job.requires, organization_id=organization_id, prefer_node=node_id)
+        placement = self.place(job.requires, prefer_node=node_id)
         if placement is None or placement.node.id != node_id:
             fallback = Placement(
                 node=ComputeNode(
                     id=node_id,
-                    organization_id=organization_id,
+                    organization_id=None,
                     name="(claiming node)",
                     is_controller=False,
                     trust_domain="local",

umbratrans_web/routes_compute.py

@@ -1119,7 +1119,7 @@ def claim_job(
     node.require("compute:claim")
     repository = compute_repository(request)
     claimed = repository.claim_job(
-        node.node_id, node.organization_id, lease_seconds=lease_seconds, kinds=kinds
+        node.node_id, lease_seconds=lease_seconds, kinds=kinds
     )
     if claimed is None:
         return None

umbratrans_web/tests/test_compute.py

@@ -210,8 +210,8 @@ def test_two_agents_claiming_at_once_get_different_jobs(compute, organization) -
     claimed: list = []
     with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
         futures = [
-            pool.submit(compute.claim_job, first.id, organization),
-            pool.submit(compute.claim_job, second.id, organization),
+            pool.submit(compute.claim_job, first.id),
+            pool.submit(compute.claim_job, second.id),
         ]
         for future in futures:
             result = future.result()
@@ -229,13 +229,13 @@ def test_two_agents_claiming_at_once_get_different_jobs(compute, organization) -
 def test_a_claim_takes_the_job_out_of_the_queue(compute, organization) -> None:
     node = _online(compute, _node(compute, organization, "solo"), [_device(_node(compute, organization, "s"))])
     job = compute.submit_job(organization, kind="sleep", requested_by="test")
-    claimed = compute.claim_job(node.id, organization)
+    claimed = compute.claim_job(node.id)
     assert claimed is not None
     assert claimed.job.id == job.id
     assert claimed.job.status == RUNNING
     assert claimed.job.lease_owner == node.id
     # A second claim finds nothing: the job is not queued.
-    assert compute.claim_job(node.id, organization) is None
+    assert compute.claim_job(node.id) is None
 
 
 @integration
@@ -243,7 +243,7 @@ def test_a_lease_is_renewable_and_stops_when_the_job_is_not_held(compute, organi
     node = _online(compute, _node(compute, organization, "renew"), [_device(_node(compute, organization, "r"))])
     other = _online(compute, _node(compute, organization, "other"), [_device(_node(compute, organization, "o"))])
     compute.submit_job(organization, kind="sleep", requested_by="test")
-    claimed = compute.claim_job(node.id, organization, lease_seconds=30)
+    claimed = compute.claim_job(node.id, lease_seconds=30)
     assert claimed is not None
 
     # Renewed by the holder.
@@ -262,7 +262,7 @@ def test_only_the_holding_node_can_complete_a_job(compute, organization) -> None
     node = _online(compute, _node(compute, organization, "holder"), [_device(_node(compute, organization, "h"))])
     other = _online(compute, _node(compute, organization, "thief"), [_device(_node(compute, organization, "t"))])
     compute.submit_job(organization, kind="sleep", requested_by="test")
-    claimed = compute.claim_job(node.id, organization)
+    claimed = compute.claim_job(node.id)
     assert claimed is not None
     with pytest.raises(ConflictError) as error:
         compute.complete_job(claimed.job.id, other.id, result={"stolen": True})
@@ -287,9 +287,9 @@ def test_an_expired_lease_returns_the_job_to_the_queue(compute, organization) ->
 
     node = _online(compute, _node(compute, organization, "dying"), [_device(_node(compute, organization, "d"))])
     compute.submit_job(organization, kind="sleep", requested_by="test", max_attempts=3)
-    claimed = compute.claim_job(node.id, organization, lease_seconds=10)
+    claimed = compute.claim_job(node.id, lease_seconds=10)
     assert claimed is not None
-    assert compute.claim_job(node.id, organization) is None, "held, so not claimable"
+    assert compute.claim_job(node.id) is None, "held, so not claimable"
 
     from datetime import datetime, timedelta, timezone
 
@@ -311,7 +311,7 @@ def test_a_job_that_exhausts_its_attempts_fails_with_its_error(compute, organiza
 
     node = _online(compute, _node(compute, organization, "exhaust"), [_device(_node(compute, organization, "e"))])
     compute.submit_job(organization, kind="sleep", requested_by="test", max_attempts=1)
-    claimed = compute.claim_job(node.id, organization, lease_seconds=10)
+    claimed = compute.claim_job(node.id, lease_seconds=10)
     assert claimed is not None
 
     from datetime import datetime, timedelta, timezone
@@ -328,18 +328,18 @@ def test_a_failure_is_retried_with_backoff_then_given_up(compute, organization)
     node = _online(compute, _node(compute, organization, "retry"), [_device(_node(compute, organization, "r2"))])
     job = compute.submit_job(organization, kind="sleep", requested_by="test", max_attempts=2)
 
-    claimed = compute.claim_job(node.id, organization)
+    claimed = compute.claim_job(node.id)
     assert claimed is not None
     requeued = compute.fail_job(claimed.job.id, node.id, error="tts exploded", retry=True)
     assert requeued.status == QUEUED, "retried, with a delay"
     assert requeued.error == "tts exploded"
     assert requeued.attempt == 1
     # Backoff: not claimable immediately.
-    assert compute.claim_job(node.id, organization) is None
+    assert compute.claim_job(node.id) is None
 
     # Exhaust the attempts.
     compute.sweep_expired_leases()
-    second = compute.claim_job(node.id, organization)
+    second = compute.claim_job(node.id)
     if second is not None:
         final = compute.fail_job(second.job.id, node.id, error="tts exploded again", retry=True)
         assert final.status == FAILED
@@ -580,7 +580,7 @@ def test_decomposition_makes_ordinary_children_and_a_container_parent(compute, o
 
     node = _online(compute, _node(compute, organization, "splitter"), [_device(_node(compute, organization, "s3"))])
     parent = compute.submit_job(organization, kind="analyze_project", requested_by="test")
-    claimed = compute.claim_job(node.id, organization)
+    claimed = compute.claim_job(node.id)
     assert claimed is not None
 
     children = compute.submit_children(
@@ -610,12 +610,12 @@ def test_a_parent_with_a_failed_child_ends_partial_not_complete(compute, organiz
 
     node = _online(compute, _node(compute, organization, "settler"), [_device(_node(compute, organization, "s4"))])
     parent = compute.submit_job(organization, kind="analyze_project", requested_by="test")
-    compute.claim_job(node.id, organization)
+    compute.claim_job(node.id)
     children = compute.submit_children(
         parent.id, node.id, [{"kind": "analyze_utterances"}, {"kind": "analyze_utterances"}]
     )
     for child in children:
-        compute.claim_job(node.id, organization)
+        compute.claim_job(node.id)
     # One child fails permanently, one completes.
     compute.fail_job(children[0].id, node.id, error="chapter is corrupt", retry=False)
     compute.complete_job(children[1].id, node.id, result={"utterances": 40})
@@ -630,12 +630,12 @@ def test_a_parent_with_a_failed_child_ends_partial_not_complete(compute, organiz
 def test_a_parent_with_all_children_complete_ends_complete(compute, organization) -> None:
     node = _online(compute, _node(compute, organization, "allgood"), [_device(_node(compute, organization, "s5"))])
     parent = compute.submit_job(organization, kind="analyze_project", requested_by="test")
-    compute.claim_job(node.id, organization)
+    compute.claim_job(node.id)
     children = compute.submit_children(
         parent.id, node.id, [{"kind": "analyze_utterances"}, {"kind": "analyze_utterances"}]
     )
     for child in children:
-        compute.claim_job(node.id, organization)
+        compute.claim_job(node.id)
         compute.complete_job(child.id, node.id, result={"ok": True})
     settled = compute.get_job(parent.id)
     assert settled is not None
@@ -649,7 +649,7 @@ def test_only_the_holding_node_may_decompose(compute, organization) -> None:
     holder = _online(compute, _node(compute, organization, "owner"), [_device(_node(compute, organization, "o2"))])
     other = _online(compute, _node(compute, organization, "notme"), [_device(_node(compute, organization, "n2"))])
     parent = compute.submit_job(organization, kind="analyze_project", requested_by="test")
-    compute.claim_job(holder.id, organization)
+    compute.claim_job(holder.id)
     with pytest.raises(ConflictError):
         compute.submit_children(parent.id, other.id, [{"kind": "analyze_utterances"}])
 
@@ -919,7 +919,7 @@ def test_an_application_sweeps_expired_leases_with_nobody_asking(compute, organi
     node = _online(compute, _node(compute, organization, "swept"),
                    [_device(_node(compute, organization, "s"))])
     compute.submit_job(organization, kind="sleep", requested_by="test", max_attempts=3)
-    claimed = compute.claim_job(node.id, organization, lease_seconds=1)
+    claimed = compute.claim_job(node.id, lease_seconds=1)
     assert claimed is not None
 
     # Not yet expired: the loop must not rescue a lease that is still live, or it

umbratrans_web/tests/test_postgres_repository.py

@@ -739,7 +739,7 @@ integration = pytest.mark.skipif(
 
 
 @pytest.fixture(scope="session")
-def require_empty_database() -> None:
+def require_empty_database():
     """Refuse to run against a database that holds anything.
 
     This suite empties ``users``, ``organizations`` and ``audit_events`` when it
@@ -759,21 +759,18 @@ def require_empty_database() -> None:
     So the safety is not in cleaning up carefully. It is in refusing to start.
     A database with accounts in it is not a test database, whatever its name, and
     the only safe thing to do with one is decline.
+
+    Since migration 0005 the check also looks inside the tenant schemas, and the
+    teardown drops them. That is not tidiness: a project's rows now live in a
+    schema per organization, so a database that holds nothing in ``public`` can
+    still hold somebody's books. Checking only the platform tables would have made
+    this guard pass on a database it would then have emptied.
     """
 
     dsn = _integration_dsn()
     if dsn is None or not _destructive_run_acknowledged() or "umbracontrol" in dsn:
         return
-    pool = ConnectionPool(dsn, min_size=1, max_size=2, open=True, kwargs={"row_factory": dict_row})
-    try:
-        with pool.connection() as connection:
-            counts = {}
-            for table in ("users", "organizations", "audit_events", "user_roles", "sessions"):
-                row = connection.execute(f"SELECT count(*) AS total FROM public.{table}").fetchone()
-                counts[table] = int(row["total"])
-    finally:
-        pool.close()
-
+    counts = _occupancy(dsn)
     populated = {name: count for name, count in counts.items() if count}
     if populated:
         detail = ", ".join(f"{name}={count}" for name, count in sorted(populated.items()))
@@ -782,6 +779,59 @@ def require_empty_database() -> None:
             f"({detail}). This suite empties those tables when it finishes and cannot do so "
             "selectively, because audit_events is append-only. Use a scratch database."
         )
+    yield
+    _drop_tenant_schemas(dsn)
+
+
+#: The tables the suite empties, and the per-tenant tables it has to look inside.
+PLATFORM_TABLES = ("users", "organizations", "audit_events", "user_roles", "sessions")
+TENANT_TABLES = ("projects", "project_authors", "project_publishers", "invitations")
+
+
+def _occupancy(dsn: str) -> dict[str, int]:
+    """Row counts for every table this suite could destroy, platform and tenant."""
+
+    counts: dict[str, int] = {}
+    pool = ConnectionPool(dsn, min_size=1, max_size=2, open=True, kwargs={"row_factory": dict_row})
+    try:
+        with pool.connection() as connection:
+            for table in PLATFORM_TABLES:
+                row = connection.execute(f"SELECT count(*) AS total FROM public.{table}").fetchone()
+                counts[table] = int(row["total"])
+            schemas = connection.execute(
+                "SELECT nspname FROM pg_namespace WHERE nspname LIKE 'tenant\\_%' ORDER BY nspname"
+            ).fetchall()
+            for schema in schemas:
+                for table in TENANT_TABLES:
+                    row = connection.execute(
+                        f'SELECT count(*) AS total FROM "{schema["nspname"]}"."{table}"'
+                    ).fetchone()
+                    counts[f"{schema['nspname']}.{table}"] = int(row["total"])
+    finally:
+        pool.close()
+    return counts
+
+
+def _drop_tenant_schemas(dsn: str) -> None:
+    """Remove every tenant schema, so a run leaves nothing behind for the next one.
+
+    `TRUNCATE ... CASCADE` empties the tenant *tables* - each of the four has a
+    foreign key to `public.users`, so truncating users takes them with it - but the
+    schemas themselves would survive, and a hundred runs would leave a hundred
+    schemas for organizations that no longer exist. Only reached when the guard
+    above established the database held nothing to begin with.
+    """
+
+    pool = ConnectionPool(dsn, min_size=1, max_size=2, open=True, kwargs={"row_factory": dict_row})
+    try:
+        with pool.connection() as connection:
+            schemas = connection.execute(
+                "SELECT nspname FROM pg_namespace WHERE nspname LIKE 'tenant\\_%' ORDER BY nspname"
+            ).fetchall()
+            for schema in schemas:
+                connection.execute(f'DROP SCHEMA IF EXISTS "{schema["nspname"]}" CASCADE')
+    finally:
+        pool.close()
 
 
 @pytest.fixture
@@ -801,6 +851,11 @@ def postgres(require_empty_database: None) -> Any:
         # created. TRUNCATE rather than DELETE because audit_events has a trigger
         # that raises on delete; truncating does not fire it, and takes the
         # tables that reference these with it.
+        #
+        # `CASCADE` reaches the tenant schemas as well, which is the point of
+        # mentioning it: each of the four tenant tables has a foreign key to
+        # `public.users`, so a project's rows go when users go. The schemas
+        # themselves are dropped once per session, in `require_empty_database`.
         with repository._pool.connection() as connection:  # noqa: SLF001 - test teardown
             connection.execute("TRUNCATE public.users, public.organizations, public.audit_events CASCADE")
     finally:
@@ -871,6 +926,45 @@ def test_the_rls_identity_contract_holds_against_the_real_schema(
     postgres.verify_rls_contract()
 
 
+@integration
+def test_the_tenant_schema_contract_holds_against_the_real_schema(
+    postgres: PostgresAccountRepository,
+) -> None:
+    """The same self-check for the tenant schema name.
+
+    The name is derived in two languages - Python here, ``app.tenant_schema_name``
+    in the migration that created the schemas - and nothing joins them. A
+    disagreement is silent and total: every tenant query would resolve into a
+    schema that does not exist, so every project request for every organization
+    would raise "relation does not exist", and the error would name the statement
+    rather than the cause.
+
+    Asserted against the real function rather than against a constant, because a
+    constant in the test would agree with itself whatever the database said.
+    """
+
+    postgres.verify_tenant_schema_contract()
+
+
+@integration
+def test_a_tenant_schema_mismatch_is_refused_rather_than_served(
+    postgres: PostgresAccountRepository,
+) -> None:
+    """The check has to be able to fail, or passing it means nothing.
+
+    The adapter's name derivation is pointed at a prefix the database does not
+    use, and the self-check has to refuse. A check that cannot fail is the same
+    failure mode as a policy nobody wrote: it reports success while enforcing
+    nothing.
+    """
+
+    with pytest.MonkeyPatch.context() as monkeypatch:
+        monkeypatch.setattr("umbratrans_web.postgres_repository.tenant_schema_name", lambda _: "tenant_not_this_one")
+        with pytest.raises(RepositoryError) as error:
+            postgres.verify_tenant_schema_contract()
+    assert error.value.code == "tenant_schema_mismatch"
+
+
 @integration
 def test_organization_rows_round_trip(
     postgres: PostgresAccountRepository, pg_service: AccountService, platform: Principal
@@ -1091,13 +1185,29 @@ def test_the_guard_covers_every_table_the_teardown_empties() -> None:
 
     A table emptied by the teardown but not checked by the guard would be
     destroyed without the suite ever objecting.
+
+    The names are read out of the teardown's own TRUNCATE statement and then looked
+    up in the guard's list, rather than the other way round. An earlier version
+    searched the guard's source text for the string `"users"`, which stopped being
+    true the moment the guard began delegating to a constant - and a text search
+    that fails when a refactor moves a string is a test of the layout, not of the
+    property. Deriving both sides from what they actually do keeps the property and
+    drops the coincidence.
     """
 
     import inspect
+    import re
 
-    source = inspect.getsource(require_empty_database)
-    for table in ("users", "organizations", "audit_events"):
-        assert f'"{table}"' in source, f"{table} is emptied but not guarded"
+    source = inspect.getsource(postgres)
+    # Comments first. The teardown's own comment explains why it truncates rather
+    # than deletes, and the word TRUNCATE appears in that sentence too - which is
+    # how the first version of this parsed a paragraph as a table list.
+    statements = re.sub(r"#.*", "", source)
+    truncate = re.search(r"TRUNCATE\s+(public\.[^;)]+?)(?:CASCADE)?[\"']", statements)
+    assert truncate is not None, "the teardown no longer truncates anything; this test cannot check it"
+    emptied = {name.strip().split(".")[-1] for name in truncate.group(1).split(",")}
+    unguarded = emptied - set(PLATFORM_TABLES)
+    assert not unguarded, f"the teardown empties {sorted(unguarded)} and the guard does not look at them"
 
 
 @integration
@@ -1276,8 +1386,10 @@ def test_assignment_is_a_project_scoped_row_in_user_roles(
     project = pg_service.create_project(platform, name="Assigned", organization_id=organization.id)
     account = _member(pg_service, platform, organization.id, "Assignee")
 
-    assert postgres.assign_user_to_project(project.id, account) is True
-    assert postgres.assign_user_to_project(project.id, account) is False, "assigning twice is a no-op"
+    assert postgres.assign_user_to_project(project.id, account, organization_id=organization.id) is True
+    assert (
+        postgres.assign_user_to_project(project.id, account, organization_id=organization.id) is False
+    ), "assigning twice is a no-op"
     assert postgres.list_project_member_ids(project.id) == [account]
 
     with postgres._pool.connection() as connection:  # noqa: SLF001 - asserting the stored row's shape
@@ -1313,7 +1425,7 @@ def test_unassigning_leaves_the_accounts_tenant_membership_alone(
     organization = pg_service.create_organization(platform, name="Keep Org", slug=f"kp-{uuid4().hex[:8]}")
     project = pg_service.create_project(platform, name="Keep", organization_id=organization.id)
     account = _member(pg_service, platform, organization.id, "Keeper")
-    postgres.assign_user_to_project(project.id, account)
+    postgres.assign_user_to_project(project.id, account, organization_id=organization.id)
     postgres.unassign_user_from_project(project.id, account)
 
     with postgres._pool.connection() as connection:  # noqa: SLF001 - asserting nothing else was touched
@@ -1428,7 +1540,7 @@ def test_an_assigned_member_can_still_sign_in(
     with postgres._pool.connection() as connection:  # noqa: SLF001 - asserting the stored shape
         assert len(_active_tenant_rows(connection, member)) == 1
 
-    assert postgres.assign_user_to_project(project.id, member) is True
+    assert postgres.assign_user_to_project(project.id, member, organization_id=organization.id) is True
 
     with postgres._pool.connection() as connection:  # noqa: SLF001 - and now there are two rows
         tenant_rows = _active_tenant_rows(connection, member)
@@ -1461,7 +1573,7 @@ def test_assigning_to_several_projects_does_not_accumulate_tenancy_rows(
     member = _member(pg_service, platform, organization.id, "Busy")
     for index in range(3):
         project = pg_service.create_project(platform, name=f"Busy {index}", organization_id=organization.id)
-        assert postgres.assign_user_to_project(project.id, member) is True
+        assert postgres.assign_user_to_project(project.id, member, organization_id=organization.id) is True
 
     with postgres._pool.connection() as connection:  # noqa: SLF001 - asserting the stored shape
         assert len(_active_tenant_rows(connection, member)) == 1
@@ -1546,7 +1658,7 @@ def test_setup_is_required_ignores_project_assignments(
     organization = pg_service.create_organization(platform, name="Setup Org", slug=f"so-{uuid4().hex[:8]}")
     project = pg_service.create_project(platform, name="Setup Book", organization_id=organization.id)
     member = _member(pg_service, platform, organization.id, "Setup")
-    postgres.assign_user_to_project(project.id, member)
+    postgres.assign_user_to_project(project.id, member, organization_id=organization.id)
 
     before = postgres.count_users_with_role(RoleName.PLATFORM_ADMIN)
     assert pg_service.setup_is_required() is False

umbratrans_web/tests/test_rls_coverage.py

@@ -61,6 +61,12 @@ def _query(sql: str, params: tuple = ()) -> list[tuple]:
 #: Every table this project gives a policy to. Kept as a literal rather than
 #: derived from pg_policies, because deriving it is the mistake: a table with no
 #: policy would simply not appear in the list it is being checked against.
+#:
+#: `project_authors` and `project_publishers` are *not* here, and that is the
+#: schema-per-tenant migration talking: they are tenant-owned, they live one per
+#: organization, and they carry no policy because there is nothing for a policy to
+#: filter - the schema a statement runs in already *is* the tenant. They are
+#: checked by `test_the_tenant_tables_are_not_in_public_and_have_no_policy`.
 POLICY_TABLES = (
     "artifacts",
     "authors",
@@ -73,9 +79,16 @@ POLICY_TABLES = (
     "node_credentials",
     "node_heartbeats",
     "node_models",
+    "publishers",
+)
+
+#: The tables migration 0005 moved into a per-organization schema, and which must
+#: therefore have no shared copy left behind to protect.
+TENANT_TABLES = (
+    "projects",
     "project_authors",
     "project_publishers",
-    "publishers",
+    "invitations",
 )
 
 
@@ -117,6 +130,104 @@ def test_every_policy_table_actually_has_a_policy() -> None:
     assert not rows, f"{[name for (name,) in rows]} have RLS enabled and no policy at all"
 
 
+def test_every_named_policy_table_is_actually_there() -> None:
+    """`POLICY_TABLES` may not name a table that is not in `public`.
+
+    Without this, dropping a table out of the policy list and out of `public` in
+    the same change makes the two checks above pass by not looking at it - which
+    is how a table can quietly stop being policed and still look policed. The
+    literal is the contract, so the literal is checked against the database.
+    """
+
+    rows = _query(
+        "SELECT c.relname FROM pg_class c "
+        "JOIN pg_namespace n ON n.oid = c.relnamespace "
+        "WHERE n.nspname = 'public' AND c.relname = ANY(%s) ORDER BY c.relname",
+        (list(POLICY_TABLES),),
+    )
+    found = [name for (name,) in rows]
+    assert found == sorted(POLICY_TABLES), (
+        f"POLICY_TABLES names tables that are not in public: "
+        f"{sorted(set(POLICY_TABLES) - set(found))}"
+    )
+
+
+def test_the_tenant_tables_are_not_in_public_and_have_no_policy() -> None:
+    """The other half of the migration, and the property it exists to create.
+
+    There must be no shared `projects` to filter. A `projects` in `public` would
+    be the old world with the new tests around it: every policy would still be
+    written as an `organization_id` filter while the adapter had stopped supplying
+    one, and the two would disagree in the direction that shows rows.
+
+    And in the tenant schemas there must be no policy either. A policy there would
+    be redundant, and redundant is not the problem - the problem is that it would
+    be *load-bearing* if the application ever ran as a non-owner, silently
+    reinstating per-assignment visibility on top of organization-wide visibility.
+    """
+
+    shared = _query(
+        "SELECT c.relname FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace "
+        "WHERE n.nspname = 'public' AND c.relkind = 'r' AND c.relname = ANY(%s) ORDER BY 1",
+        (list(TENANT_TABLES),),
+    )
+    assert not shared, (
+        f"{[name for (name,) in shared]} are tenant-owned and must not exist in public; "
+        "a shared copy is a table two organizations' rows could both be in"
+    )
+    policed = _query(
+        "SELECT schemaname, tablename FROM pg_policies WHERE tablename = ANY(%s) ORDER BY 1, 2",
+        (list(TENANT_TABLES),),
+    )
+    assert not policed, (
+        f"{[(s, t) for s, t in policed]} carry a row-level-security policy; "
+        "the tenant schema is the boundary, and a policy on top of it is dormant "
+        "today and load-bearing if the role ever changes"
+    )
+
+
+def test_every_tenant_schema_has_its_own_copy_of_the_tenant_tables() -> None:
+    """Each organization's schema carries all four, so an unqualified query finds them.
+
+    Read from the schemas rather than assumed from the migration: a tenant schema
+    missing `projects` would make every project request for that one organization
+    raise `relation does not exist`, and an organization with a schema but no
+    tables is a state nothing else in this suite would notice.
+    """
+
+    schemas = _query(r"SELECT nspname FROM pg_namespace WHERE nspname LIKE 'tenant\_%' ORDER BY 1")
+    assert schemas, "no tenant schemas exist: migration 0005 has not been applied to this database"
+    for (schema,) in schemas:
+        present = {
+            name
+            for (name,) in _query(
+                "SELECT c.relname FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace "
+                "WHERE n.nspname = %s AND c.relkind = 'r' AND c.relname = ANY(%s)",
+                (schema, list(TENANT_TABLES)),
+            )
+        }
+        missing = set(TENANT_TABLES) - present
+        assert not missing, f"{schema} is missing {sorted(missing)}"
+
+
+def test_no_tenant_table_carries_an_organization_column() -> None:
+    """`organization_id` is gone from the tenant tables, and that is the point.
+
+    A column that is always equal to the schema's owner is a second source of
+    truth. It is also the column every removed policy filtered on, so a migration
+    that dropped the policies and kept the column would leave an application that
+    filters by a value nobody maintains - and a row written with the wrong value
+    would be invisible to a query that believed it.
+    """
+
+    rows = _query(
+        "SELECT table_schema, table_name, column_name FROM information_schema.columns "
+        "WHERE table_name = ANY(%s) AND column_name = 'organization_id' ORDER BY 1, 2",
+        (list(TENANT_TABLES),),
+    )
+    assert not rows, f"{[(s, t) for s, t, _ in rows]} still carry organization_id"
+
+
 def test_the_datastore_tables_exist_with_the_columns_the_application_reads() -> None:
     """Column presence, not table presence.
 
@@ -124,13 +235,15 @@ def test_the_datastore_tables_exist_with_the_columns_the_application_reads() ->
     `derivation` by the lineage code. A migration that created the table without
     those columns would pass every existence check in the suite and fail on the
     first artefact written.
+
+    The two project-credit tables are looked for in a tenant schema, because that
+    is where migration 0005 put them. Looking for them in `public` would keep
+    asserting a shape the database no longer has.
     """
 
     expected = {
         "publishers": {"parent_id", "owner_organization_id", "kind"},
         "authors": {"name", "display_name", "slug", "user_id", "owner_organization_id"},
-        "project_publishers": {"project_id", "publisher_id", "lang"},
-        "project_authors": {"project_id", "author_id", "credit_role", "lang"},
         "artifacts": {
             "project_id", "organization_id", "lang", "kind", "status",
             "content_hash", "byte_size", "media_type", "storage_key",
@@ -149,6 +262,24 @@ def test_the_datastore_tables_exist_with_the_columns_the_application_reads() ->
         missing = columns - present
         assert not missing, f"{table} is missing {sorted(missing)}"
 
+    tenant_columns = {
+        "project_publishers": {"project_id", "publisher_id", "lang"},
+        "project_authors": {"project_id", "author_id", "credit_role", "lang"},
+    }
+    schemas = _query(r"SELECT nspname FROM pg_namespace WHERE nspname LIKE 'tenant\_%' ORDER BY 1 LIMIT 1")
+    assert schemas, "no tenant schemas exist: migration 0005 has not been applied to this database"
+    (schema,) = schemas[0]
+    for table, columns in tenant_columns.items():
+        rows = _query(
+            "SELECT column_name FROM information_schema.columns "
+            "WHERE table_schema = %s AND table_name = %s",
+            (schema, table),
+        )
+        present = {name for (name,) in rows}
+        assert present, f"{schema}.{table} does not exist"
+        missing = columns - present
+        assert not missing, f"{schema}.{table} is missing {sorted(missing)}"
+
 
 def test_shared_reference_data_is_readable_by_any_authenticated_principal() -> None:
     """publishers and authors must keep a SELECT policy of their own.

umbratrans_web/tests/test_tenancy.py

@@ -31,7 +31,14 @@ import pytest
 
 from umbratrans_web.errors import ConflictError, ForbiddenError, InputError, StateTransitionError
 from umbratrans_web.models import Principal, ProjectStatus, RoleName
-from umbratrans_web.tenancy import require_within, scope_for
+from umbratrans_web.tenancy import (
+    PLATFORM_SCHEMA,
+    TENANT_SCHEMA_PREFIX,
+    require_within,
+    scope_for,
+    tenant_schema_name,
+    tenant_search_path,
+)
 
 from .conftest import ensure_organization, provision_local
 
@@ -430,3 +437,76 @@ def test_a_project_with_an_unusable_name_is_refused(service) -> None:
         with pytest.raises(InputError):
             service.create_project(_principal(platform), name=name, organization_id=org)
     assert service.repository.count_projects(organization_id=org) == 0
+
+
+# ---------------------------------------------------------------------------
+# The schema a tenant's tables are in
+# ---------------------------------------------------------------------------
+# Three properties, and the third is the one that matters most.
+#
+# * The name is a function of the organization id, so it cannot be stored wrong.
+# * The search path for a tenant is that tenant's schema and `public`, in that
+#   order - `public` must be reachable or the platform tables vanish, and the
+#   tenant's schema must be first or a table of the same name elsewhere would win.
+# * **A request that has not resolved an organization gets `public` alone.** That
+#   is the isolation property, and it is a property of this function rather than
+#   of the database: it is what makes a tenant table unreachable rather than
+#   merely unfound.
+
+
+def test_a_tenant_schema_name_is_derived_from_the_organization() -> None:
+    organization = uuid4()
+    assert tenant_schema_name(organization) == f"{TENANT_SCHEMA_PREFIX}{organization.hex}"
+    # The same identifier always gives the same schema, which is what lets the
+    # database create the schema and the connection find it without either side
+    # storing anything.
+    assert tenant_schema_name(organization) == tenant_schema_name(organization)
+    assert tenant_schema_name(uuid4()) != tenant_schema_name(organization)
+
+
+def test_two_organizations_never_share_a_schema() -> None:
+    identifiers = {uuid4() for _ in range(64)}
+    names = {tenant_schema_name(identifier) for identifier in identifiers}
+    assert len(names) == len(identifiers)
+
+
+def test_a_tenant_search_path_puts_the_tenant_first_and_public_second() -> None:
+    organization = uuid4()
+    path = tenant_search_path(organization)
+    assert path == f'"{tenant_schema_name(organization)}", {PLATFORM_SCHEMA}'
+
+
+def test_a_request_with_no_organization_gets_public_alone() -> None:
+    """The isolation property, in the one place that decides it.
+
+    `None` is what a caller with no resolved organization has, and it is not a
+    spelling of "every tenant". Returning a wildcard here - or leaving the
+    connection's own default in place, which may name a schema nobody chose - would
+    put every tenant's tables in reach of a request that has not decided which
+    tenant it is acting for. So this returns `public`, and only `public`.
+    """
+
+    assert tenant_search_path(None) == PLATFORM_SCHEMA
+    assert TENANT_SCHEMA_PREFIX not in tenant_search_path(None)
+
+
+def test_the_search_path_cannot_be_shaped_by_an_organization_identifier() -> None:
+    """A UUID cannot become a second schema, and the path says so structurally.
+
+    `search_path` is parsed as a list of names, so a value carrying a comma or a
+    quote would change the list rather than name a schema. That is unreachable from
+    a `UUID` - which is why the name is derived and never stored - and this test
+    pins the shape rather than the reasoning: every name the function emits is the
+    prefix followed by exactly thirty-two lowercase hex characters, so there is
+    nothing in it that could terminate the list.
+    """
+
+    for _ in range(32):
+        path = tenant_search_path(uuid4())
+        for part in path.split(", "):
+            bare = part.strip('"')
+            if bare == PLATFORM_SCHEMA:
+                continue
+            assert bare.startswith(TENANT_SCHEMA_PREFIX)
+            assert len(bare) == len(TENANT_SCHEMA_PREFIX) + 32
+            assert all(character in "0123456789abcdef" for character in bare[len(TENANT_SCHEMA_PREFIX) :])