fix(ingestion): reconcile dbt companion evidence against its physical…

canonic/config.py

@@ -47,6 +47,14 @@ class Connection(BaseModel):
     from ``pg_stats`` into the returned schema (used to improve LLM grain inference).
     Requesting ``fetch_column_stats`` on SQLite/DuckDB connections is a no-op (logged as
     a warning) — neither engine exposes queryable planner statistics without a data scan.
+    A ``dbt`` connection additionally recognizes ``manifest_path`` (path to the compiled
+    ``manifest.json``, default ``manifest.json``) and ``target_connection`` (the id of the
+    physical, queryable connection whose tables this manifest describes — its
+    ``RelationSchema`` evidence is stamped with that id, not this connection's own, so it
+    reconciles against that connection's live-introspected sources instead of proposing
+    independent ones; defaults to this connection's own id when omitted, matching a
+    standalone dbt-only project with no paired physical connection). Validated against the
+    configured connection ids by :meth:`CanonicConfig._validate_target_connections`.
     """
 
     id: str
@@ -305,6 +313,24 @@ def _enforce_air_gapped(self) -> CanonicConfig:
                 )
         return self
 
+    @model_validator(mode="after")
+    def _validate_target_connections(self) -> CanonicConfig:
+        """A connection's ``params.target_connection`` must name a configured connection.
+
+        Catches a typo'd or dangling reference at load time (fail fast) rather than
+        letting it silently fall back to the connection's own id or misattribute a
+        proposed source to a connection that doesn't exist.
+        """
+        ids = {c.id for c in self.connections}
+        for conn in self.connections:
+            target = conn.params.get("target_connection")
+            if target is not None and target not in ids:
+                raise ValueError(
+                    f"connections[{conn.id}].params.target_connection={target!r} does not "
+                    f"match any configured connection id; configured: {sorted(ids)}"
+                )
+        return self
+
     @classmethod
     def settings_customise_sources(
         cls,

canonic/connectors/factory.py

@@ -32,7 +32,14 @@
 
 def _make_dbt(conn: Connection) -> DbtConnector:
     manifest_path = conn.params.get("manifest_path", "manifest.json")
-    return DbtConnector(manifest_path, source=conn.id)
+    # A dbt connection is definitions-only (no run_read_only_sql); its RelationSchemas
+    # must be stamped with the *physical* connection they describe, not this connection's
+    # own id, so the builder proposes them at the same semantics/<connection>/<name>.yaml
+    # target as that connection's live introspection and reconciliation's existing
+    # tier-preference (_resolve_group/_CURATION_RANK) actually engages instead of the two
+    # colliding on the global source-name uniqueness check.
+    physical_connection = conn.params.get("target_connection", conn.id)
+    return DbtConnector(manifest_path, source=physical_connection)
 
 
 def _make_notion(conn: Connection) -> GenericEvidenceConnector:

docs/concepts/connectors.mdx

@@ -39,6 +39,8 @@ Implement `extract_definitions`: feed the semantic layer and canonical-binding c
 | --- | --- |
 | `dbt` | Parses a compiled `manifest.json`: models → typed columns/grain, measures, joins, descriptions. |
 
+A definition connector is its own connection entry (own `id`) but describes tables that live in a *different*, queryable connection. `params.target_connection` names that connection's id, so the evidence is attributed to it rather than to the definition connector's own id — this is what lets its `RelationSchema` proposals land on the same `semantics/<connection>/<name>.yaml` target as that connection's live introspection and reconcile against it (see the [worked example](/concepts/ingestion-and-reconciliation#example-adding-a-definition-connector-after-an-initial-apply)), instead of colliding on the project-wide unique source `name`. Omitting it falls back to the definition connector's own id — fine for a definitions-only project with no paired physical connection, but such sources can never actually be queried (no `run_read_only_sql`).
+
 ### Evidence
 
 Implement `extract_evidence`: feed [knowledge pages](/concepts/knowledge-layer) and reconciliation signal from docs and BI usage. Also never queryable.

docs/concepts/ingestion-and-reconciliation.mdx

@@ -61,7 +61,7 @@ A conflict is never resolved by silent overwrite. The worst case is a flagged co
 
 Say a connection starts out queryable-only. `canonic ingest` runs [live introspection](/concepts/connectors#schema-acquisition-ladder) and proposes semantic sources at provenance `inferred`, the tier every new proposal starts at. `canonic apply` accepts them into `semantics/*.yaml`. Accepting a file this way commits it, but doesn't curate it: it's still `inferred` — promoting it to `human_curated` takes an explicit [`canonic review curate`](/cli-reference/review-apply).
 
-Now a `dbt` connector is added to the same connection and `canonic ingest` runs again. `extract_definitions` produces definition evidence from the compiled manifest, an acquisition tier above raw introspection, but it still enters reconciliation at provenance `inferred`, same as the existing fact. Provenance and acquisition tier are independent: the two proposals disagreeing on the same relation is not a "existing tier higher" case and not a silent win either way. Instead, the modeling-code evidence is preferred as the tie-break within the `inferred` band, so the run resolves to the "conflicts, tier ≤ proposal" row above: propose an **edit**. That diff lands under `.canonic/pending-diffs/<run-id>/` and goes through [`canonic review`](/cli-reference/review-apply) / [`canonic apply`](/cli-reference/review-apply) exactly like the first round. If the edit touches `grain`, `joins`, or `measures`, it's ineligible for auto-apply no matter how confident the proposal is, per the `never` list under [Propose-only by default](#propose-only-by-default) below.
+Now a `dbt` connector is added — its own connection entry in `canonic.yaml` (its own `id`, `type: dbt`), with `params.target_connection` naming the physical connection above (see [Connectors](/concepts/connectors#schema-acquisition-ladder)). That's what makes this example work: `target_connection` stamps the dbt connector's evidence with the physical connection's id rather than its own, so it lands on the same `semantics/<connection>/<name>.yaml` target instead of proposing an independent source under its own connection id (which would collide on the project-wide unique source `name` instead of reconciling). With that link in place, `canonic ingest` runs again and `extract_definitions` produces definition evidence from the compiled manifest, an acquisition tier above raw introspection, but it still enters reconciliation at provenance `inferred`, same as the existing fact. Provenance and acquisition tier are independent: the two proposals disagreeing on the same relation is not a "existing tier higher" case and not a silent win either way. Instead, the modeling-code evidence is preferred as the tie-break within the `inferred` band, so the run resolves to the "conflicts, tier ≤ proposal" row above: propose an **edit**. That diff lands under `.canonic/pending-diffs/<run-id>/` and goes through [`canonic review`](/cli-reference/review-apply) / [`canonic apply`](/cli-reference/review-apply) exactly like the first round. If the edit touches `grain`, `joins`, or `measures`, it's ineligible for auto-apply no matter how confident the proposal is, per the `never` list under [Propose-only by default](#propose-only-by-default) below.
 
 Had the file instead been `freeze`d during the first review, the dbt proposal would only flag a contradiction: the existing fact is left untouched regardless of its acquisition tier or confidence, per the "existing frozen" row above.
 

docs/guides/jaffle-shop.mdx

@@ -45,7 +45,17 @@ canonic mcp start                                        # expose metrics to any
 
 ## What canonic extracts from the dbt manifest
 
-The `jaffle_dbt` connection parses `manifest.json` (schema v11, dbt 1.7, MetricFlow) as modeling-tier evidence, which ranks higher than live DuckDB introspection during reconciliation. From it, canonic extracts:
+The `jaffle_dbt` connection parses `manifest.json` (schema v11, dbt 1.7, MetricFlow) as modeling-tier evidence, which ranks higher than live DuckDB introspection during reconciliation:
+
+```yaml
+  - id: jaffle_dbt
+    type: dbt
+    params:
+      manifest_path: dbt/manifest.json
+      target_connection: jaffle_duckdb   # the physical connection these models describe
+```
+
+`target_connection` is what makes this a companion to `jaffle_duckdb` rather than an independent source: it attributes the dbt-derived evidence to `jaffle_duckdb`'s connection id, so it enriches `semantics/jaffle_duckdb/*.yaml` instead of proposing separate, same-named files under `semantics/jaffle_dbt/` (see [Connectors](/concepts/connectors#definition)). From the manifest, canonic extracts:
 
 - **5 model nodes** → `RelationSchema` with named columns, types, primary keys, and foreign-key paths.
 - **2 semantic models** → entity (grain), foreign-join paths, named measure and dimension definitions.

docs/reference/config-schema.mdx

@@ -33,6 +33,13 @@ connections:
 
 - `credentials_ref` (and `llm.api_key_ref`) must be a reference, one of `env:`, `keyring:`, or `file:`, never a literal secret; config validation rejects a literal outright.
 - Postgres/Redshift connections additionally recognize `schema`/`schemas` (search path) and `tables` (glob patterns narrowing introspection), plus `fetch_column_stats: true` to merge zero-scan cardinality/null-ratio stats into the returned schema (a no-op on SQLite/DuckDB).
+- `dbt` connections recognize `manifest_path` (path to the compiled `manifest.json`, default `manifest.json`) and `target_connection` (the id of the physical, queryable connection whose tables this manifest describes — see [Connectors](/concepts/connectors#definition)). `target_connection` must name a connection actually declared in `connections[]`; an unknown id fails config validation at load time. Omitting it falls back to the dbt connection's own id.
+
+```yaml
+  - id: jaffle_dbt
+    type: dbt
+    params: { manifest_path: dbt/manifest.json, target_connection: warehouse_pg }
+```
 
 ## `llm`
 

examples/jaffle-shop/canonic.yaml

@@ -12,11 +12,14 @@ connections:
 
   # E3 definition connector: compiled dbt manifest.json — modeling-tier evidence
   # (semantic models, metrics, grain, joins) that outranks raw introspection during
-  # reconciliation. No live database required for this connector.
+  # reconciliation. No live database required for this connector. target_connection
+  # names the physical connection these models describe, so its evidence reconciles
+  # against jaffle_duckdb's sources instead of proposing independent ones.
   - id: jaffle_dbt
     type: dbt
     params:
       manifest_path: dbt/manifest.json   # relative to this canonic.yaml
+      target_connection: jaffle_duckdb
 
 llm:
   provider: openai_compatible

tests/connectors/test_factory.py

@@ -0,0 +1,32 @@
+"""Tests for the connector factory (SPEC-E2 §2.2a) — dbt's target_connection wiring."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from canonic.config import Connection
+from canonic.connectors.dbt import DbtConnector
+from canonic.connectors.factory import default_factory
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+
+class TestMakeDbt:
+    def test_target_connection_stamps_physical_connection_id(self, dbt_manifest_path: Path) -> None:
+        conn = Connection(
+            id="jaffle_dbt",
+            type="dbt",
+            params={"manifest_path": str(dbt_manifest_path), "target_connection": "jaffle_duckdb"},
+        )
+        connector = default_factory.create(conn)
+        assert isinstance(connector, DbtConnector)
+        assert connector._source == "jaffle_duckdb"  # noqa: SLF001 — verifying the wiring itself
+
+    def test_missing_target_connection_falls_back_to_own_id(self, dbt_manifest_path: Path) -> None:
+        conn = Connection(
+            id="jaffle_dbt", type="dbt", params={"manifest_path": str(dbt_manifest_path)}
+        )
+        connector = default_factory.create(conn)
+        assert isinstance(connector, DbtConnector)
+        assert connector._source == "jaffle_dbt"  # noqa: SLF001 — today's default behavior

tests/ingestion/test_pipeline.py

@@ -21,9 +21,13 @@
     scaffold_project,
 )
 from canonic.connectors.base import (
+    AcquisitionTier,
     Capability,
     ColumnInfo,
     ConnectorBase,
+    DefinitionEntityType,
+    DefinitionEvidence,
+    DefinitionExtract,
     ForeignKey,
     ForeignKeyRef,
     Health,
@@ -33,9 +37,10 @@
 from canonic.exc import ValidationFailed
 from canonic.ingestion.models import DraftedBy, ProposalOp, ReconciliationDecision
 from canonic.ingestion.pipeline import IngestionPipeline, first_run_auto_acceptable
-from canonic.ingestion.source import evidence_from_introspection
+from canonic.ingestion.source import evidence_from_definitions, evidence_from_introspection
 from canonic.runtime.drafter import make_drafter
 from canonic.semantic.loader import load_semantic_source
+from canonic.semantic.models import Additivity
 
 if TYPE_CHECKING:
     from pathlib import Path
@@ -465,6 +470,129 @@ async def draft_schema_joins(
     assert not first_run_auto_acceptable(customers_diff)
 
 
+# ---------------------------------------------------------------------------
+# Companion definitions connector (e.g. dbt) reconciling against its physical
+# connection, once configured with target_connection (canonic/connectors/factory.py).
+# ---------------------------------------------------------------------------
+
+_DBT_CONN = "jaffle_dbt"
+
+
+class FakeDefinitionConnector(ConnectorBase):
+    """A connector whose ``extract_definitions`` returns canned modeling-tier evidence.
+
+    Stands in for a dbt manifest connector once its factory-constructed ``RelationSchema``
+    is stamped with the *physical* connection's id (mirrors ``factory._make_dbt`` reading
+    ``target_connection``), rather than its own connection id.
+    """
+
+    def __init__(
+        self, relations: list[RelationSchema], definitions: list[DefinitionEvidence] | None = None
+    ) -> None:
+        self._relations = relations
+        self._definitions = definitions or []
+
+    def capabilities(self) -> list[Capability]:
+        return [Capability.EXTRACT_DEFINITIONS, Capability.TEST_CONNECTION]
+
+    async def test_connection(self) -> Health:
+        return Health(status="ok")
+
+    async def extract_definitions(self) -> DefinitionExtract:
+        return DefinitionExtract(
+            relations=list(self._relations), definitions=list(self._definitions)
+        )
+
+
+def _dbt_orders(*, connection: str) -> RelationSchema:
+    """A dbt-shaped ``RelationSchema`` describing the same table as ``_orders()``."""
+    columns = [
+        ColumnInfo(name="order_id", type="int", nullable=True),
+        ColumnInfo(name="customer_id", type="int", nullable=True),
+        ColumnInfo(name="amount", type="decimal", nullable=True),
+    ]
+    return RelationSchema(
+        connection=connection,
+        relation="analytics.fct_orders",
+        kind="table",
+        columns=columns,
+        primary_key=["order_id"],
+        foreign_keys=[],
+        acquisition_tier=AcquisitionTier.MODELING,
+        source_fingerprint=compute_fingerprint(columns, ["order_id"], []),
+    )
+
+
+def _revenue_measure() -> DefinitionEvidence:
+    return DefinitionEvidence(
+        source=_DBT_CONN,
+        entity="revenue",
+        entity_type=DefinitionEntityType.MEASURE,
+        expr="sum(amount)",
+        additivity=Additivity.ADDITIVE,
+        references=["analytics.fct_orders"],
+        native_ref="model.jaffle_shop.orders#revenue",
+        acquisition_tier=AcquisitionTier.MODELING,
+        source_fingerprint="sha256:revenue-measure-v1",
+    )
+
+
+async def test_companion_evidence_edits_existing_source_instead_of_colliding(
+    tmp_path: Path,
+) -> None:
+    """The reported bug: re-ingesting a definitions-only connector after the physical
+    connection is already accepted must reconcile against the same file (EDIT), not
+    propose a same-named source under a different connection (which the global
+    source-name uniqueness gate would then reject for the whole batch)."""
+    pipeline = _pipeline(tmp_path, [_orders()])
+    await pipeline.bootstrap(_CONN)
+    assert (tmp_path / "semantics" / _CONN / "fct_orders.yaml").exists()
+
+    dbt_connector = FakeDefinitionConnector([_dbt_orders(connection=_CONN)], [_revenue_measure()])
+    evidence = await evidence_from_definitions(dbt_connector, _DBT_CONN)
+
+    pipeline2 = _pipeline_no_scaffold(tmp_path, [_orders()])
+    result = await pipeline2.run(evidence)
+
+    assert not (tmp_path / "semantics" / _DBT_CONN).exists()
+    (diff,) = result.emission.diffs
+    assert diff.op is ProposalOp.EDIT
+    assert diff.target == f"semantics/{_CONN}/fct_orders.yaml"
+    entry = next(e for e in result.report.entries if e.target == diff.target)
+    assert any(m["name"] == "revenue" for m in entry.proposal.content["measures"])
+    # measures is a structural "never" field — never silently auto-applied.
+    assert entry.auto_apply is False
+
+
+async def test_bootstrap_folds_companion_evidence_into_the_physical_source(
+    tmp_path: Path,
+) -> None:
+    """`bootstrap()`'s companion-connector inclusion (pipeline.py:186-234) must resolve to a
+    single proposal per table via the existing modeling-tier tie-break, never a second
+    semantics/<dbt-connection>/<name>.yaml file."""
+    scaffold_project(tmp_path)
+    pipeline = IngestionPipeline(
+        tmp_path,
+        {
+            _CONN: FakeConnector([_orders()]),
+            _DBT_CONN: FakeDefinitionConnector(
+                [_dbt_orders(connection=_CONN)], [_revenue_measure()]
+            ),
+        },
+        ReconcileConfig(),
+    )
+
+    result = await pipeline.bootstrap(_CONN)
+
+    assert (tmp_path / "semantics" / _CONN / "fct_orders.yaml").exists()
+    assert not (tmp_path / "semantics" / _DBT_CONN).exists()
+    (diff,) = result.emission.diffs
+    assert diff.target == f"semantics/{_CONN}/fct_orders.yaml"
+    entry = next(e for e in result.report.entries if e.target == diff.target)
+    assert entry.proposal.acquisition_tier is AcquisitionTier.MODELING
+    assert any(m["name"] == "revenue" for m in entry.proposal.content["measures"])
+
+
 async def test_no_models_configured_headless_is_deterministic(tmp_path: Path) -> None:
     """llm=None + headless → NullLLMDrafter, zero model calls, deterministic emission (GH-68 S7)."""
     drafter = make_drafter(None, RuntimeConfig(), headless=True)

tests/test_config.py

@@ -155,6 +155,38 @@ def test_default_connection_parsed(self, tmp_path: Path) -> None:
         assert cfg.project.default_connection == "warehouse_pg"
 
 
+class TestTargetConnectionValidation:
+    """``connections[].params.target_connection`` must name a configured connection id."""
+
+    def _with_dbt_connection(self, params_line: str) -> str:
+        dbt_connection = (
+            "  - id: jaffle_dbt\n"
+            "    type: dbt\n"
+            "    params:\n"
+            "      manifest_path: manifest.json\n"
+            f"{params_line}"
+        )
+        return _VALID.replace(
+            "    credentials_ref: env:CANONIC_PG_DSN\n",
+            f"    credentials_ref: env:CANONIC_PG_DSN\n{dbt_connection}",
+        )
+
+    def test_target_connection_matching_configured_id_loads(self, tmp_path: Path) -> None:
+        content = self._with_dbt_connection("      target_connection: warehouse_pg\n")
+        cfg = load_config(_canonic_yaml(tmp_path, content))
+        assert cfg.connections[1].params["target_connection"] == "warehouse_pg"
+
+    def test_target_connection_naming_unknown_id_raises(self, tmp_path: Path) -> None:
+        content = self._with_dbt_connection("      target_connection: does_not_exist\n")
+        with pytest.raises(ConfigError, match="target_connection"):
+            load_config(_canonic_yaml(tmp_path, content))
+
+    def test_missing_target_connection_loads_unaffected(self, tmp_path: Path) -> None:
+        content = self._with_dbt_connection("")
+        cfg = load_config(_canonic_yaml(tmp_path, content))
+        assert "target_connection" not in cfg.connections[1].params
+
+
 class TestMcpAuthConfig:
     """``mcp.auth`` block (AMENDMENT-remote-mcp-transport.md)."""