fix(OMN-16239): resolve ProjectionTableTarget schema through physical mapping at SQL emission

docker/domain-adapter-proof/prove.py

@@ -57,7 +57,7 @@ def _parse_args() -> argparse.Namespace:
 ROLE_PASSWORD = "domain-adapter-proof-only"  # pragma: allowlist secret
 TENANT_A = UUID("aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa")
 TENANT_B = UUID("bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb")
-TENANT_TABLE = "delegation_events"
+TENANT_TABLE = "future_tenant_projection"
 INTERNAL_TABLE = "future_internal_projection"
 CATALOG_TABLE = "plan_tiers"
 
@@ -305,7 +305,7 @@ def _prove_real_rls_with_check() -> None:
             _raises(
                 psycopg2.errors.InsufficientPrivilege,
                 lambda: cursor.execute(
-                    "INSERT INTO tenant.delegation_events VALUES (%s, %s, %s)",
+                    "INSERT INTO tenant.future_tenant_projection VALUES (%s, %s, %s)",
                     (uuid4(), "unset-context", TENANT_A),
                 ),
             )
@@ -316,7 +316,7 @@ def _prove_real_rls_with_check() -> None:
             _raises(
                 psycopg2.errors.InsufficientPrivilege,
                 lambda: cursor.execute(
-                    "INSERT INTO tenant.delegation_events VALUES (%s, %s, %s)",
+                    "INSERT INTO tenant.future_tenant_projection VALUES (%s, %s, %s)",
                     (uuid4(), "wrong-insert", TENANT_B),
                 ),
             )
@@ -326,7 +326,7 @@ def _prove_real_rls_with_check() -> None:
         with conn.cursor() as cursor:
             cursor.execute("SET LOCAL app.tenant_id = %s", (str(TENANT_A),))
             cursor.execute(
-                "INSERT INTO tenant.delegation_events VALUES (%s, %s, %s)",
+                "INSERT INTO tenant.future_tenant_projection VALUES (%s, %s, %s)",
                 (correlation_id, "valid-a", TENANT_A),
             )
         conn.commit()
@@ -335,7 +335,7 @@ def _prove_real_rls_with_check() -> None:
             _raises(
                 psycopg2.errors.InsufficientPrivilege,
                 lambda: cursor.execute(
-                    "UPDATE tenant.delegation_events SET tenant_id = %s "
+                    "UPDATE tenant.future_tenant_projection SET tenant_id = %s "
                     "WHERE correlation_id = %s",
                     (TENANT_B, correlation_id),
                 ),
@@ -539,7 +539,7 @@ def main() -> None:
 
     for dsn, sql in (
         (TENANT_DSN, f"SELECT * FROM omninode_internal.{INTERNAL_TABLE}"),
-        (INTERNAL_DSN, "SELECT * FROM tenant.delegation_events"),
+        (INTERNAL_DSN, "SELECT * FROM tenant.future_tenant_projection"),
         (
             CATALOG_DSN,
             "INSERT INTO platform_catalog.plan_tiers VALUES ('x', 'X')",
@@ -571,7 +571,7 @@ def main() -> None:
     )
 
     rows = _admin_rows(
-        "SELECT tenant_id FROM tenant.delegation_events ORDER BY tenant_id"
+        "SELECT tenant_id FROM tenant.future_tenant_projection ORDER BY tenant_id"
     )
     assert TENANT_A in {row[0] for row in rows}
     assert TENANT_B in {row[0] for row in rows}

src/omnibase_infra/runtime/auto_wiring/handler_wiring.py

@@ -1934,12 +1934,21 @@ class ProjectionCatalogBindingPolicy:
 
 @dataclass(frozen=True)
 class ProjectionTableTarget:
-    """Topology-resolved location for one typed table declaration."""
+    """Topology-resolved location for one typed table declaration.
+
+    Every field beside ``table`` is a *resolution*; ``table`` alone carries the
+    raw declaration. ``physical_schema`` is the schema resolution exactly as
+    ``physical_database`` is the database resolution -- the schema PostgreSQL
+    actually holds the relation in, which during the OMN-15359 migration window
+    differs from the logically-declared ``table.schema``. Emitted SQL must
+    qualify with this field; anything reasoning about the declaration (domain,
+    contract-facing errors) reads ``table.schema``.
+    """
 
     table: ModelDbTableDeclaration
     database_ref: str
     physical_database: str
-    schema: str
+    physical_schema: str
     domain: EnumDatabaseSchemaDomain
     read_binding: ProjectionDatabaseBindingTarget | None
     write_binding: ProjectionDatabaseBindingTarget | None
@@ -1959,9 +1968,9 @@ def database_refs(self) -> tuple[str, ...]:
         return tuple(sorted({target.database_ref for target in self.table_targets}))
 
     @property
-    def schemas(self) -> tuple[str, ...]:
-        """Return the declared schemas in stable order."""
-        return tuple(sorted({target.schema for target in self.table_targets}))
+    def physical_schemas(self) -> tuple[str, ...]:
+        """Return the schemas SQL is actually issued against, in stable order."""
+        return tuple(sorted({target.physical_schema for target in self.table_targets}))
 
     @property
     def domains(self) -> tuple[EnumDatabaseSchemaDomain, ...]:
@@ -2035,10 +2044,16 @@ def _require_projection_binding_privileges(
     table: ModelDbTableDeclaration,
     *,
     operation: str,
+    grant_schema: str,
 ) -> None:
-    """Prove the selected topology principal can perform the exact operation."""
+    """Prove the selected topology principal can perform the exact operation.
+
+    ``grant_schema`` is the caller's already-resolved physical schema, not a
+    second resolution of its own: the schema whose ACLs are checked here and
+    the schema the emitted SQL qualifies with are the same value by
+    construction (OMN-16239).
+    """
     principal = database.principals[binding.principal]
-    grant_schema = physical_grant_schema_for_table(table.schema, table.name)
     has_schema_usage = any(
         grant.object_type is EnumDatabaseGrantObjectType.SCHEMA
         and grant.schema == grant_schema
@@ -2087,6 +2102,7 @@ def _projection_operation_bindings(
     table: ModelDbTableDeclaration,
     database: ModelDeploymentTopologyDatabase,
     domain: EnumDatabaseSchemaDomain,
+    physical_schema: str,
     catalog_read_binding: str | None,
     catalog_write_binding: str | None,
 ) -> tuple[
@@ -2134,13 +2150,15 @@ def _projection_operation_bindings(
             read_binding,
             table,
             operation="read",
+            grant_schema=physical_schema,
         )
     if write_binding is not None:
         _require_projection_binding_privileges(
             database,
             write_binding,
             table,
             operation="write",
+            grant_schema=physical_schema,
         )
     return read_binding, write_binding
 
@@ -2171,10 +2189,16 @@ def _resolve_projection_database_target(
         domain = topology.schema_domain(table.database_ref, table.schema)
         physical_database = database.physical_name
         databases_by_physical_name[physical_database] = database
+        # OMN-16239: resolve the physical schema exactly once, here, and feed
+        # both the grant check and the emitted SQL from it. Domain resolution
+        # above deliberately keeps using the declared schema -- the domain is a
+        # governance fact about the contract, not about physical placement.
+        physical_schema = physical_grant_schema_for_table(table.schema, table.name)
         read_binding, write_binding = _projection_operation_bindings(
             table=table,
             database=database,
             domain=domain,
+            physical_schema=physical_schema,
             catalog_read_binding=catalog_read_binding,
             catalog_write_binding=catalog_write_binding,
         )
@@ -2183,7 +2207,7 @@ def _resolve_projection_database_target(
                 table=table,
                 database_ref=table.database_ref,
                 physical_database=physical_database,
-                schema=table.schema,
+                physical_schema=physical_schema,
                 domain=domain,
                 read_binding=read_binding,
                 write_binding=write_binding,
@@ -2419,17 +2443,19 @@ def __init__(
         self._adapter = adapter
         self._target = target
 
+    # Both refusals name the DECLARED schema: they report what the contract
+    # says, so they must read the way the contract reads.
     def _assert_write_declared(self) -> None:
         if self._target.table.access not in {"write", "read_write"}:
             raise PermissionError(
-                f"{self._target.schema}.{self._target.table.name} declares "
+                f"{self._target.table.schema}.{self._target.table.name} declares "
                 f"access={self._target.table.access!r}; write refused"
             )
 
     def _assert_read_declared(self) -> None:
         if self._target.table.access not in {"read", "read_write"}:
             raise PermissionError(
-                f"{self._target.schema}.{self._target.table.name} declares "
+                f"{self._target.table.schema}.{self._target.table.name} declares "
                 f"access={self._target.table.access!r}; read refused"
             )
 
@@ -2747,7 +2773,7 @@ def _execute_upsert(
         action = f"DO UPDATE SET {updates}" if updates else "DO NOTHING"
         insert_sql = " ".join(
             (
-                f'INSERT INTO "{target.schema}"."{target.table.name}" ({quoted_cols})',
+                f'INSERT INTO "{target.physical_schema}"."{target.table.name}" ({quoted_cols})',
                 f"VALUES ({placeholders})",
                 f"ON CONFLICT ({conflict_columns}) {action}",
             )
@@ -2771,7 +2797,7 @@ def _execute_query(
         tenant_context: VerifiedProjectionTenantAuthority | None,
     ) -> list[dict[str, object]]:
         # Schema/table originate in validated typed declarations, never request data.
-        select_sql = f'SELECT * FROM "{target.schema}"."{target.table.name}"'  # noqa: S608
+        select_sql = f'SELECT * FROM "{target.physical_schema}"."{target.table.name}"'  # noqa: S608
         params: list[object] = []
         if filters:
             bad_keys = [
@@ -3202,10 +3228,10 @@ def _build_projection_db_adapter(
 
     logger.debug(
         "Selecting projection adapter: database_refs=%s physical_database=%s "
-        "schemas=%s domains=%s",
+        "physical_schemas=%s domains=%s",
         target.database_refs,
         target.physical_database,
-        target.schemas,
+        target.physical_schemas,
         [domain.value for domain in target.domains],
     )
     return ProjectionDatabaseOperations(

tests/fixtures/application_relation_ownership/topology.yaml

@@ -69,6 +69,10 @@ databases:
             schema: public
             objects: [delegation_events]
             privileges: [SELECT, INSERT, UPDATE]
+          - object_type: TABLE
+            schema: tenant
+            objects: [future_tenant_projection]
+            privileges: [SELECT, INSERT, UPDATE]
     bindings:
       app_dashboard:
         database_ref: application

tests/integration/runtime/test_live_events_projection_write_path_omn15359.py

@@ -260,7 +260,7 @@ def test_live_events_write_lands_in_and_reads_back_from_omninode_internal(
     )
     target = _resolve_projection_database_target((declaration,), application_topology())
 
-    assert target.table_targets[0].schema == "omninode_internal"
+    assert target.table_targets[0].physical_schema == "omninode_internal"
     assert target.table_targets[0].write_binding is not None
     assert (
         target.table_targets[0].write_binding.binding_ref == "omninode_runtime_service"
@@ -310,7 +310,7 @@ def test_grant_derivation_schema_agrees_with_the_insert_target_schema() -> None:
         role="live_events",
     )
     target = _resolve_projection_database_target((declaration,), application_topology())
-    insert_target_schema = target.table_targets[0].schema
+    insert_target_schema = target.table_targets[0].physical_schema
 
     grant_check_schema = physical_grant_schema_for_table(
         "omninode_internal", "live_events"

tests/integration/runtime/test_projection_dispatch_tenant_authority_context.py

@@ -156,7 +156,9 @@ async def test_dispatch_engine_keeps_verified_authority_out_of_band(
         "SELECT set_config(%s, %s, true)",
         ("app.tenant_id", str(tenant_id)),
     )
-    assert 'INSERT INTO "tenant"."delegation_events"' in calls[2][0]
+    # OMN-16239: physical schema, not the declared one -- delegation_events is
+    # still bridged to public until OMN-15359 relocates the tenant family.
+    assert 'INSERT INTO "public"."delegation_events"' in calls[2][0]
     assert calls[2][1]["correlation_id"] == envelope.envelope_id  # type: ignore[index]
     assert calls[2][1]["tenant_id"] == tenant_id  # type: ignore[index]
     assert connection.close_calls == 1
@@ -222,4 +224,8 @@ async def test_mixed_target_internal_operation_does_not_resolve_tenant_authority
 
     connect.assert_called_once_with("postgresql://internal")
     assert all("set_config" not in sql for sql, _params in calls)
-    assert any('"omninode_internal"."generation_events"' in sql for sql, _ in calls)
+    # OMN-16239: generation_events is still under the OMN-15359 bridge, so the
+    # internal write resolves to public. The point of this assertion is that the
+    # internal table was written at all on the internal binding -- the schema it
+    # is qualified with must be the physical one.
+    assert any('"public"."generation_events"' in sql for sql, _ in calls)

tests/unit/runtime/auto_wiring/test_projection_domain_adapters.py

@@ -247,7 +247,10 @@ def test_red_control_nonlocal_tenant_guc() -> None:
         "SELECT set_config(%s, %s, true)",
         ("app.tenant_id", str(tenant_id)),
     )
-    assert 'INSERT INTO "tenant"."delegation_events"' in calls[2].args[0]
+    # OMN-16239: qualified with the PHYSICAL schema. delegation_events is still
+    # under the OMN-15359 bridge and lives in public; the pre-fix assertion here
+    # named "tenant", a schema the analytics database does not even have.
+    assert 'INSERT INTO "public"."delegation_events"' in calls[2].args[0]
     assert calls[2].args[1]["tenant_id"] == tenant_id
     assert isinstance(calls[2].args[1]["tenant_id"], UUID)
     conn.commit.assert_called_once_with()