runtime: the pre-flight resolves the edge the join wrote (ACE-133)

CHANGELOG.md

@@ -172,6 +172,15 @@ below corresponds to one such version.
 
 ### Fixed
 
+- **A join on the key the model declares one-to-one is no longer reported as a fan trap because a
+  second edge exists between the same two tables.** The fan and chasm pre-flight matched a declared
+  edge to a join by table pair and never read the columns the join wrote, so a subclass view joined
+  to its base table on its id fanned whenever the model also declared a many-to-one between the pair
+  on another key, and the same receipt's joins section, which does read the key, called the join
+  one-to-one. The pre-flight now keeps only the edges whose declared columns the written join
+  matches; a join on a key the model does not declare, or two tables in scope with no join between
+  them, keep every edge and today's verdict. (ACE-133)
+
 - **A database failure nobody could read is no longer reported as a syntax error.** Every engine
   raises its execution failure with the same exit code, and the classifier read that code back as
   `syntax` whenever none of its rules matched the message, so a connection dropping mid-statement

packages/agami-core/src/semantic_model/runtime.py

@@ -1770,6 +1770,13 @@ def _preflight_select(tree: "exp.Select", org: Datasource,
         # edge to it would leak this query's CTE into the next one's analysis.
         rels = rels + cte_rels
     table_set = set(tables_in_scope.values())
+    # Only the edges the statement's own joins could have traversed count below (ACE-133). The fan
+    # and chasm detectors match a declared edge to a join by TABLE PAIR, and a model can declare two
+    # edges between one pair: a subclass view joined to its base table on the key the model declares
+    # one-to-one was reported as a fan trap because a sibling many-to-one between the same two
+    # tables was also in the list, while the joins section of the same receipt, which reads the
+    # written key, called the join one-to-one. The written key names the edge.
+    rels = _edges_as_written(rels, tree, tables_in_scope, _dialect_of(org)[0])
 
     sites = _aggregate_sites(tree, tables_in_scope, scope, visible)
     # The set the detectors read, derived from the sites rather than walked again — the two must
@@ -4664,6 +4671,75 @@ def _joined_table_pairs(tree: "exp.Select", scope_map: dict[str, str]) -> frozen
     return frozenset(pairs)
 
 
+def _written_join_pairs(
+    tree: "exp.Select", scope_map: dict[str, str]
+) -> dict[frozenset[str], frozenset[frozenset[tuple[str, str]]]]:
+    """THIS SELECT's own explicit joins as the column pairs each one wrote, keyed by the unordered
+    pair of tables it connects: the same reading as `_joined_table_pairs`, keeping the columns. The
+    same pinning rule too: only an ON that names its own join's two relations counts, and a pair an
+    unqualified column kept this layer from resolving contributes nothing."""
+    out: dict[frozenset[str], set[frozenset[tuple[str, str]]]] = {}
+    for join in tree.args.get("joins") or ():
+        on = join.args.get("on")
+        if on is None:
+            continue
+        right = scope_map.get(join.this.alias_or_name, _relation_name(join.this))
+        names = {scope_map.get(col.table, col.table) for col in on.find_all(exp.Column) if col.table}
+        if right not in names or len(names - {right}) > 1:
+            continue
+        for pair in _predicate_pairs(on, scope_map):
+            tables = frozenset(table for table, _column in pair)
+            if len(tables) == 2:
+                out.setdefault(tables, set()).add(pair)
+    return {tables: frozenset(pairs) for tables, pairs in out.items()}
+
+
+def _edges_as_written(
+    rels: list[Relationship], tree: "exp.Select", scope_map: dict[str, str], dialect: "str | None"
+) -> list[Relationship]:
+    """The declared edges the fan and chasm detectors may lean on, given the joins the statement
+    wrote.
+
+    Both detectors match a declared edge to a join by TABLE PAIR, which is right when the model
+    declares one edge between two tables and wrong when it declares two: a subclass view joined to
+    its base table on the key the model declares `one_to_one` was reported as a fan trap because a
+    sibling `many_to_one` between the same pair was also in the list, and the identity edge cannot
+    win, shadow or suppress because nothing consulted the columns. The joins section of the same
+    receipt reads the written key (`_declared_pairs(rel) <= js.pairs`) and called the same join
+    one-to-one.
+
+    So: for a pair of tables the statement joined with a readable key, when at least one declared
+    edge between them matches that key, only the matching edges stay in the list. When the written
+    key matches no declared edge (a join on a key the model does not know) or the two tables meet
+    without a join between them (a CROSS JOIN, a chain through a third table), every edge stays and
+    today's pair-level rule stands, which is the conservative side: over-reporting a fan is a receipt
+    that says more than it had to; clearing one on a key nobody declared would say something false.
+    The declared side is reduced with `_declared_pairs`, whose None means "cannot be compared" and
+    is treated here as "does not match", never as a wildcard.
+    """
+    written = _written_join_pairs(tree, scope_map)
+    if not written:
+        return rels
+    by_pair: dict[frozenset[str], list[Relationship]] = {}
+    for rel in rels:
+        by_pair.setdefault(_rel_tables(rel), []).append(rel)
+    kept: list[Relationship] = []
+    for rel in rels:
+        tables = _rel_tables(rel)
+        pairs_written = written.get(tables)
+        if pairs_written is None:
+            kept.append(rel)
+            continue
+        matching = [
+            candidate for candidate in by_pair[tables]
+            if (declared := _declared_pairs(candidate, dialect)) is not None
+            and declared <= pairs_written
+        ]
+        if not matching or rel in matching:
+            kept.append(rel)
+    return kept
+
+
 # `apply_default_filters` was deleted here by ACE-042: declared filters are business logic, not a
 # disclosure control, so nothing justified this module authoring SQL. Reporting which filters a
 # statement applied is ACE-099; do not re-add an injector.

tests/test_ace060_trap_free_aggregates.py

@@ -474,3 +474,101 @@ def test_an_undetermined_aggregate_says_which_blindness_it_hit(org):
     assert "names no column" in count["reason"]
     items = _items(org, "SELECT SUM(o.total) FROM orders o")
     assert [(i["status"], i["reason"]) for i in items] == [(rt.NOT_MULTIPLIED, None)]
+
+
+# --- ACE-133: the pre-flight resolves the edge the join wrote --------------------------------------
+
+import yaml  # noqa: E402
+
+
+def _write_two_edge_model(root: Path) -> None:
+    """A base table and its subclass view, joined by TWO declared edges: the identity edge on `id`
+    (`one_to_one`) and a sibling many-to-one on `parent_id` (many premium rows can point at one
+    widget). Which edge the statement traverses is written in its ON, and only that edge can say
+    whether the join multiplies the base table's rows."""
+    (root / "subject_areas" / AREA / "tables").mkdir(parents=True)
+    (root / "datasource.yaml").write_text(
+        yaml.safe_dump({"datasource": "Widgets", "version": 1,
+                        "storage_connections": [{"name": "c", "storage_type": "SQLite"}],
+                        "subject_areas": [f"subject_areas/{AREA}"]})
+    )
+    (root / "subject_areas" / AREA / "subject_area.yaml").write_text(
+        yaml.safe_dump({
+            "name": AREA,
+            "tables": [{"storage_connection": "c", "schema": "public", "table": t}
+                       for t in ("widget", "widget_premium_v")],
+        })
+    )
+
+    def _table(name, columns):
+        (root / "subject_areas" / AREA / "tables" / f"{name}.yaml").write_text(
+            yaml.safe_dump({"name": name, "schema": "public", "storage_connection": "c",
+                            "grain": ["id"], "description": name, "columns": columns})
+        )
+
+    _table("widget", [
+        {"name": "id", "type": "integer", "primary_key": True},
+        {"name": "amount", "type": "decimal", "aggregation": "additive"},
+    ])
+    _table("widget_premium_v", [
+        {"name": "id", "type": "integer", "primary_key": True},
+        {"name": "parent_id", "type": "integer"},
+        {"name": "tier", "type": "string"},
+    ])
+    edge = {"from_schema": "public", "to_schema": "public", "confidence": "confirmed",
+            "review_state": "approved", "signed_off_by": "you@example.com",
+            "signed_off_role": "data_owner", "signed_off_at": "2026-01-01T00:00:00Z"}
+    (root / "subject_areas" / AREA / "relationships.yaml").write_text(
+        yaml.safe_dump({"relationships": [
+            # The sibling edge FIRST, so a fix that only reordered the list could not pass.
+            dict(edge, from_table="widget_premium_v", from_column="parent_id",
+                 to_table="widget", to_column="id", relationship="many_to_one"),
+            dict(edge, from_table="widget_premium_v", from_column="id",
+                 to_table="widget", to_column="id", relationship="one_to_one"),
+        ]})
+    )
+
+
+@pytest.fixture()
+def two_edge_org(tmp_path):
+    root = tmp_path / "widgets"
+    root.mkdir(parents=True)
+    _write_two_edge_model(root)
+    return L.load_datasource(root)
+
+
+IDENTITY_JOIN = "SELECT SUM(w.amount) FROM widget w JOIN widget_premium_v p ON p.id = w.id"
+SIBLING_JOIN = "SELECT SUM(w.amount) FROM widget w JOIN widget_premium_v p ON p.parent_id = w.id"
+UNDECLARED_KEY_JOIN = "SELECT SUM(w.amount) FROM widget w JOIN widget_premium_v p ON p.tier = w.id"
+
+
+def test_a_join_on_the_identity_edge_does_not_fan_when_a_sibling_edge_also_exists(two_edge_org):
+    """Sandeep's finding: a subclass view joined to its base table on the key the model declares
+    one-to-one was reported as a fan trap, because the pre-flight collected every multiplying edge
+    between the two tables and never read which key the join wrote. The written key names the edge."""
+    receipt = rt.assemble_receipt(two_edge_org, IDENTITY_JOIN)
+    (item,) = receipt["aggregates"]["items"]
+    assert item["status"] == rt.NOT_MULTIPLIED and item["findings"] == [], item
+    # And the two sections of one receipt agree about the same join.
+    (join,) = receipt["joins"]["items"]
+    assert join["status"] == rt.DECLARED and join["cardinality"] == "one_to_one"
+
+
+def test_a_join_on_the_sibling_edge_still_fans(two_edge_org):
+    (item,) = _items(two_edge_org, SIBLING_JOIN)
+    assert item["status"] == rt.MULTIPLIED
+    assert [f["risk"] for f in item["findings"]] == ["fan_trap"]
+    assert item["joins"] == ["widget (1) <- widget_premium_v (N)"]
+
+
+def test_a_join_on_a_key_the_model_does_not_declare_keeps_the_conservative_verdict(two_edge_org):
+    """The written key matches no declared edge, so nothing says which edge the statement meant;
+    every edge between the pair still counts and the finding stands, as it did before."""
+    (item,) = _items(two_edge_org, UNDECLARED_KEY_JOIN)
+    assert item["status"] == rt.MULTIPLIED
+
+
+def test_two_tables_in_scope_without_a_join_between_them_keep_every_edge(two_edge_org):
+    """A cross join writes no key at all, so the pair-level rule stands and the sibling edge fans."""
+    (item,) = _items(two_edge_org, "SELECT SUM(w.amount) FROM widget w CROSS JOIN widget_premium_v p")
+    assert item["status"] == rt.MULTIPLIED