fix(security): stop the fail-closed check conflating same-named tables across schemas

The previous commit matched a protected table against a query's tables by
adding both the qualified name and its bare part to each side, then
intersecting. That collapsed public.customer_profiles and
marts.customer_profiles to the same key, so protecting one refused queries
against the other -- with a message naming a table the user never referenced.

This product added multi-schema support in #55 and an acme_erp fixture with
crm/sales/finance/hr/inventory in #65, so same-named tables across schemas are
the expected shape here, not a corner case. Over-blocking is the safe
direction, which is exactly why it would have survived review and surfaced
later as unexplained refusals.

Matching is now asymmetric, because the two sides carry different information.
ConnectionChatAccessPolicyService.qualifyTable() drops the schema when it is
"public", so a bare PROTECTED name means public.<table> -- it is not unknown. A
bare REFERENCE in a query is genuinely unknown: it resolves through the session
search_path and could be any schema.

    reference unqualified -> match on bare name (ambiguous, so block)
    protected public      -> a qualified reference must actually say public
    both qualified        -> exact match

Every bypass stays closed: an unqualified reference to a protected table is
still refused, and hr.salaries still matches a bare "salaries".

Also replaces a characterization test asserting the opposite. It was written
before qualifyTable's public-collapsing was discovered and encoded the wrong
belief that a bare protected name is ambiguous; the case genuinely worth
pinning is a bare reference, which it now covers.

Verified: 17/17 in UserDataAccessPolicyServiceTest. Regression baseline on the
same suite selection unchanged at 8 failures / 16 errors (414 run vs 412
before, the delta being these two tests passing).

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: geekypunk

backend/src/main/java/com/dbaagent/service/UserDataAccessPolicyService.java

@@ -363,17 +363,18 @@ private void assertProtectedTablesAreInspectable(
         }
         Set<String> referenced = new LinkedHashSet<>();
         for (String name : new TablesNamesFinder<>().getTables(statement)) {
-            addNameForms(name, referenced);
+            if (name != null && !name.isBlank()) {
+                referenced.add(normalizeName(name));
+            }
         }
         Set<String> inspected = new LinkedHashSet<>();
         for (PlainSelect branch : inspectedBranches) {
             collectDirectTables(branch, inspected);
         }
         for (ConnectionChatAccessPolicyService.ProtectionDescriptor descriptor : protectedObjects.values()) {
-            Set<String> forms = new LinkedHashSet<>();
-            addNameForms(descriptor.qualifiedTableName(), forms);
-            boolean isReferenced = forms.stream().anyMatch(referenced::contains);
-            boolean wasInspected = forms.stream().anyMatch(inspected::contains);
+            String protectedName = descriptor.qualifiedTableName();
+            boolean isReferenced = referenced.stream().anyMatch(name -> namesMatch(protectedName, name));
+            boolean wasInspected = inspected.stream().anyMatch(name -> namesMatch(protectedName, name));
             if (isReferenced && !wasInspected) {
                 throw new UserDataAccessPolicyException(
                     "This query reaches restricted data through a nested query DeepSQL cannot fully verify, so it was blocked before execution.",
@@ -383,28 +384,55 @@ private void assertProtectedTablesAreInspectable(
         }
     }
 
-    /** Adds both the qualified name and its bare table part, so schema.t matches t. */
-    private void addNameForms(String name, Set<String> out) {
-        if (name == null || name.isBlank()) {
-            return;
+    /**
+     * Does a query's table reference name the protected table?
+     *
+     * Asymmetric on purpose, because the two sides carry different information.
+     * ConnectionChatAccessPolicyService.qualifyTable() drops the schema when it is
+     * "public", so a bare PROTECTED name means public.<table> -- it is not unknown.
+     * A bare REFERENCE in a query is genuinely unknown: it resolves through the
+     * session search_path and could be any schema.
+     *
+     *   reference unqualified -> match on bare name. Ambiguous, so block; the
+     *                            search_path may well point at the protected table.
+     *   protected public      -> the qualified reference must actually say public.
+     *                            marts.customer_profiles is a different table, and
+     *                            treating it as protected refused every other
+     *                            schema's copy -- which this product's own
+     *                            multi-schema fixtures (crm/sales/finance/hr) hit.
+     *   both qualified        -> exact match.
+     */
+    private boolean namesMatch(String protectedName, String referencedName) {
+        String protectedNorm = normalizeName(protectedName);
+        String referencedNorm = normalizeName(referencedName);
+        if (protectedNorm.isEmpty() || referencedNorm.isEmpty()) {
+            return false;
+        }
+        if (!referencedNorm.contains(".")) {
+            return bareName(protectedNorm).equals(referencedNorm);
         }
-        String normalized = normalizeName(name);
-        out.add(normalized);
-        int dot = normalized.lastIndexOf('.');
-        if (dot > 0 && dot < normalized.length() - 1) {
-            out.add(normalized.substring(dot + 1));
+        if (!protectedNorm.contains(".")) {
+            return referencedNorm.equals("public." + protectedNorm);
         }
+        return protectedNorm.equals(referencedNorm);
+    }
+
+    private String bareName(String normalizedName) {
+        int dot = normalizedName.lastIndexOf('.');
+        return dot > 0 && dot < normalizedName.length() - 1
+            ? normalizedName.substring(dot + 1)
+            : normalizedName;
     }
 
     /** Tables named directly in this branch's FROM/JOIN -- what inspection actually saw. */
     private void collectDirectTables(PlainSelect select, Set<String> out) {
         if (select.getFromItem() instanceof Table table) {
-            addNameForms(table.getFullyQualifiedName(), out);
+            out.add(normalizeName(table.getFullyQualifiedName()));
         }
         if (select.getJoins() != null) {
             for (Join join : select.getJoins()) {
                 if (join.getRightItem() instanceof Table table) {
-                    addNameForms(table.getFullyQualifiedName(), out);
+                    out.add(normalizeName(table.getFullyQualifiedName()));
                 }
             }
         }

backend/src/test/java/com/dbaagent/service/UserDataAccessPolicyServiceTest.java

@@ -201,6 +201,42 @@ void enforcePreExecution_blocksProtectedTableInsideWhereSubquery() {
         assertThat(exception.getErrorCode()).isEqualTo("POLICY_SQL_BLOCKED");
     }
 
+    // A qualified protection names exactly one table. marts.customer_profiles is a
+    // different table from public.customer_profiles and must not be caught by it.
+    @Test
+    void enforcePreExecution_allowsSameNamedTableInAnotherSchemaWhenProtectionIsQualified() {
+        when(policyService.resolveEffectivePolicy("conn-1", "analyst", false)).thenReturn(policy());
+        when(policyService.buildProtectionDescriptors(any(ConnectionChatAccessPolicyService.EffectivePolicy.class)))
+            .thenReturn(Map.of("public.customer_profiles",
+                descriptor("public", "customer_profiles", false, "email")));
+
+        service.enforcePreExecution(
+            "conn-1",
+            new QueryRequest("SELECT id FROM (SELECT id FROM marts.customer_profiles) t", null, null),
+            new QueryExecutionContext(QueryExecutionOrigin.CHAT, QueryExecutionContext.MutationMode.READ_ONLY_ONLY, "analyst", false, false)
+        );
+    }
+
+    // The genuinely ambiguous direction is an unqualified REFERENCE, not an
+    // unqualified protection: qualifyTable() stores public.<t> as bare <t>, so a
+    // bare protected name means public, while a bare reference in a query
+    // resolves through search_path and could be any schema. Block that one.
+    @Test
+    void enforcePreExecution_blocksUnqualifiedReferenceBecauseSearchPathIsUnknown() {
+        when(policyService.resolveEffectivePolicy("conn-1", "analyst", false)).thenReturn(policy());
+
+        UserDataAccessPolicyException exception = assertThrows(
+            UserDataAccessPolicyException.class,
+            () -> service.enforcePreExecution(
+                "conn-1",
+                new QueryRequest("SELECT id FROM (SELECT id FROM customer_profiles) t", null, null),
+                new QueryExecutionContext(QueryExecutionOrigin.CHAT, QueryExecutionContext.MutationMode.READ_ONLY_ONLY, "analyst", false, false)
+            )
+        );
+
+        assertThat(exception.getErrorCode()).isEqualTo("POLICY_SQL_BLOCKED");
+    }
+
     @Test
     void enforcePreExecution_blocksForbiddenSchemaInsideWhereSubquery() {
         assertThat(assertSchemaScopeBlocks(