diff --git a/canonic/config.py b/canonic/config.py index 5901533..2ec337d 100644 --- a/canonic/config.py +++ b/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, diff --git a/canonic/connectors/factory.py b/canonic/connectors/factory.py index a2b2066..3f28c21 100644 --- a/canonic/connectors/factory.py +++ b/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//.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: diff --git a/docs/concepts/connectors.mdx b/docs/concepts/connectors.mdx index 2efdabe..bd7a340 100644 --- a/docs/concepts/connectors.mdx +++ b/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//.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. diff --git a/docs/concepts/ingestion-and-reconciliation.mdx b/docs/concepts/ingestion-and-reconciliation.mdx index 6ee7e90..dd78a4c 100644 --- a/docs/concepts/ingestion-and-reconciliation.mdx +++ b/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//` 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//.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//` 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. diff --git a/docs/guides/jaffle-shop.mdx b/docs/guides/jaffle-shop.mdx index 7928f61..a3a6478 100644 --- a/docs/guides/jaffle-shop.mdx +++ b/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. diff --git a/docs/reference/config-schema.mdx b/docs/reference/config-schema.mdx index 3e41ec5..846b878 100644 --- a/docs/reference/config-schema.mdx +++ b/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` diff --git a/examples/jaffle-shop/canonic.yaml b/examples/jaffle-shop/canonic.yaml index e3aa42d..657d072 100644 --- a/examples/jaffle-shop/canonic.yaml +++ b/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 diff --git a/tests/connectors/test_factory.py b/tests/connectors/test_factory.py new file mode 100644 index 0000000..49528ac --- /dev/null +++ b/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 diff --git a/tests/ingestion/test_pipeline.py b/tests/ingestion/test_pipeline.py index b7f2964..d510bfb 100644 --- a/tests/ingestion/test_pipeline.py +++ b/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//.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) diff --git a/tests/test_config.py b/tests/test_config.py index 70de0ff..3c92680 100644 --- a/tests/test_config.py +++ b/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).""" diff --git a/uv.lock b/uv.lock index 4ccd97b..4bbe474 100644 --- a/uv.lock +++ b/uv.lock @@ -304,7 +304,7 @@ wheels = [ [[package]] name = "canonic" -version = "0.6.0" +version = "0.10.0" source = { editable = "." } dependencies = [ { name = "aiosqlite" },