From 618d57b0c813e5637650974d0a043621112a8baa Mon Sep 17 00:00:00 2001 From: Ashwin Ramachandran Date: Tue, 28 Jul 2026 09:58:00 +0530 Subject: [PATCH 1/5] feat(core): cross-datasource bridge as a validated model type (ACE-072) Add CrossDatasourceRelationship (deployment-level bridge) to the semantic model, merge it into the OrgRecord from inline/sidecar/legacy sources on load (deduped, idempotent), and validate its endpoints in a new validate_deployment pass wired into sm validate. Legacy cross_profile_relationships.yaml is migrated in-memory. --- packages/agami-core/src/semantic_model/cli.py | 10 +- .../agami-core/src/semantic_model/models.py | 79 ++++++ .../src/semantic_model/org_record.py | 86 +++++- .../src/semantic_model/validator.py | 120 ++++++++ tests/test_cross_datasource_bridge.py | 257 ++++++++++++++++++ 5 files changed, 548 insertions(+), 4 deletions(-) create mode 100644 tests/test_cross_datasource_bridge.py diff --git a/packages/agami-core/src/semantic_model/cli.py b/packages/agami-core/src/semantic_model/cli.py index 833b1aa9..e3fcfe3c 100644 --- a/packages/agami-core/src/semantic_model/cli.py +++ b/packages/agami-core/src/semantic_model/cli.py @@ -72,7 +72,15 @@ def cmd_validate(args) -> int: org = L.load_datasource(args.root, include_rejected=True) res = V.validate(org) print(V.format_result(res)) - return 0 if res.ok else 1 + ok = res.ok + # Deployment-level pass (F16 / ACE-072): when the profile sits under a deployment with an + # OrgRecord, ALSO validate the cross-datasource bridges — they resolve endpoints ACROSS profiles, + # so a single-profile validate can't see them. A no-op (empty, ok) on a pre-F16 layout with no + # record, so this stays a superset of the old behaviour. The artifacts dir is the profile's parent. + dres = V.validate_deployment(Path(args.root).parent) + if dres.findings: + print(V.format_result(dres)) + return 0 if (ok and dres.ok) else 1 def cmd_snapshot(args) -> int: diff --git a/packages/agami-core/src/semantic_model/models.py b/packages/agami-core/src/semantic_model/models.py index 5432b4c6..38a36a9b 100644 --- a/packages/agami-core/src/semantic_model/models.py +++ b/packages/agami-core/src/semantic_model/models.py @@ -604,6 +604,81 @@ class CrossSubjectAreaRelationship(Relationship): to_subject_area: str +# --------------------------------------------------------------------------- +# Cross-datasource relationship (deployment-level bridge, F16 / ACE-072) +# --------------------------------------------------------------------------- + + +class CrossDatasourceRelationship(_Base): + """A bridge between two SEPARATE datasources: "this key here is the same entity as + that key there" (e.g. ``accounts.account_key`` in a CRM = ``customers.account_key`` in + an ERP). It lives on the deployment-level ``OrgRecord`` — a cross-datasource edge can't + belong to a single per-profile ``Datasource`` — and is what F16's cross-datasource metrics + resolve through. + + A standalone ``_Base`` rather than a subclass of ``Relationship``: that class hard-requires a + single ``from_table``/``from_column`` + a *required* ``Cardinality`` and runs a completeness + validator that has no meaning across two engines. Here the endpoints are ``schema.table`` + datasets with (possibly composite) key columns, and cardinality is optional.""" + + # endpoints — each names a datasource, a ``schema.table`` dataset, and the join key column(s) + from_datasource: str + to_datasource: str + from_dataset: str # a "schema.table" string, resolved against the from_datasource's model + to_dataset: str + from_columns: list[str] + to_columns: list[str] + + # A cross-datasource edge crosses two engines by definition, so it is never ``same_engine`` + # (rejected below). Reuses the existing ``Executable`` literal — no ``federated`` value (ACE-073). + executable: Executable = "split" + # Cardinality is optional here (unlike Relationship): a bridge is often a bare identity link. + relationship: Optional[Cardinality] = None + description: str = "" + # Carried from a legacy migration when present; otherwise anonymous (identified by its endpoints). + name: Optional[str] = None + + # trust block (flattened for ergonomic YAML authoring) — copied from Relationship for parity. + confidence: Confidence = "proposed" + review_state: ReviewState = "unreviewed" + signed_off_by: Optional[str] = None + signed_off_at: Optional[str] = None + signed_off_role: Optional[str] = None + migrated_from: Optional[MigratedFrom] = None + + @model_validator(mode="after") + def _endpoints(self) -> "CrossDatasourceRelationship": + # same_engine is structurally impossible for a cross-datasource edge — the deployment + # validator re-asserts this with the whole model in view, but reject it here too so a + # hand-authored bridge can't even be constructed with it. + if self.executable == "same_engine": + raise ValueError( + "cross-datasource relationship cannot be executable='same_engine' " + "(two datasources are two engines); use 'split' or 'informational'" + ) + if not self.from_columns or not self.to_columns: + raise ValueError("cross-datasource relationship requires non-empty from_columns and to_columns") + if len(self.from_columns) != len(self.to_columns): + raise ValueError( + "cross-datasource relationship requires from_columns and to_columns of equal length " + f"(got {len(self.from_columns)} vs {len(self.to_columns)})" + ) + return self + + @property + def endpoint_key(self) -> tuple: + """The identity tuple used to de-duplicate bridges merged from several sources + (inline + sidecar + legacy migration), so a re-load never duplicates an edge.""" + return ( + self.from_datasource, + self.from_dataset, + tuple(self.from_columns), + self.to_datasource, + self.to_dataset, + tuple(self.to_columns), + ) + + # --------------------------------------------------------------------------- # Subject Area (the primary semantic unit) # --------------------------------------------------------------------------- @@ -713,6 +788,9 @@ class OrgRecord(_Base): # The datasources (profile names) attached under this org. Auto-maintained: rebuilt from the profile # directories present on disk on every onboard/deploy, never hand-edited, so it cannot drift. datasources: list[str] = Field(default_factory=list) + # Deployment-level bridges linking a key in one datasource to the same entity in another (F16 / ACE-072). + # The deployment record is their home: a cross-datasource edge can't belong to a single per-profile model. + cross_datasource_relationships: list[CrossDatasourceRelationship] = Field(default_factory=list) @field_validator("fiscal_year_start_month") @classmethod @@ -753,6 +831,7 @@ def _fy_month(cls, v: Optional[int]) -> Optional[int]: "Metric", "Relationship", "CrossSubjectAreaRelationship", + "CrossDatasourceRelationship", "SubjectArea", "Datasource", "DisplayConventions", diff --git a/packages/agami-core/src/semantic_model/org_record.py b/packages/agami-core/src/semantic_model/org_record.py index 986f19eb..5b5699b5 100644 --- a/packages/agami-core/src/semantic_model/org_record.py +++ b/packages/agami-core/src/semantic_model/org_record.py @@ -24,13 +24,20 @@ import yaml -from .models import OrgRecord +from .models import CrossDatasourceRelationship, MigratedFrom, OrgRecord # The record and the company narrative both sit at the artifacts-dir ROOT (one deployment = one company), # NOT under a profile dir — that is the whole point: written once, shared by every datasource. RECORD_FILENAME = "organization.yaml" NARRATIVE_FILENAME = "organization.md" +# Deployment-level cross-datasource bridges (F16 / ACE-072) merge from three sources on load: +# 1. inline on organization.yaml (parsed by model_validate); +# 2. a sidecar at the artifacts-dir root (this file); +# 3. the legacy skill-side file the connect flow used before bridges were a model type. +BRIDGES_FILENAME = "cross_datasource_relationships.yaml" +LEGACY_BRIDGES_PATH = ("local", "cross_profile_relationships.yaml") + def record_path(artifacts_dir: str | Path) -> Path: return Path(artifacts_dir) / RECORD_FILENAME @@ -43,13 +50,86 @@ def narrative_path(artifacts_dir: str | Path) -> Path: def load_org_record(artifacts_dir: str | Path) -> Optional[OrgRecord]: """Return the ``OrgRecord`` at ``/organization.yaml``, or ``None`` if the deployment has no record yet. Read-only and lenient (never raises on a missing file) so a pre-F15 deployment - degrades to today's per-profile behaviour rather than erroring.""" + degrades to today's per-profile behaviour rather than erroring. + + The record's ``cross_datasource_relationships`` (F16 / ACE-072) are the MERGE of three sources — + inline, a root sidecar, and a legacy skill-side file — de-duplicated by endpoint so re-loading is + idempotent. The merge is in-memory only; ``load_org_record`` writes nothing (a legacy migration + surfaces the edges without mutating disk).""" path = record_path(artifacts_dir) if not path.exists(): return None with path.open(encoding="utf-8") as fh: doc = yaml.safe_load(fh) or {} - return OrgRecord.model_validate(doc) + record = OrgRecord.model_validate(doc) + merged = _merge_bridges(record.cross_datasource_relationships + _load_bridges(artifacts_dir)) + if merged != record.cross_datasource_relationships: + record = record.model_copy(update={"cross_datasource_relationships": merged}) + return record + + +def _merge_bridges( + bridges: list[CrossDatasourceRelationship], +) -> list[CrossDatasourceRelationship]: + """De-duplicate bridges by their endpoint identity, keeping the FIRST occurrence (inline wins + over sidecar wins over legacy — the order they are concatenated). Idempotent: a re-load that + picks up the same legacy file again collapses back to one edge per endpoint tuple.""" + out: list[CrossDatasourceRelationship] = [] + seen: set[tuple] = set() + for b in bridges: + if b.endpoint_key in seen: + continue + seen.add(b.endpoint_key) + out.append(b) + return out + + +def _load_bridges(artifacts_dir: str | Path) -> list[CrossDatasourceRelationship]: + """Load the non-inline bridge sources: the root sidecar (source 2) and the legacy skill-side + file (source 3). Both are optional; a deployment with neither yields an empty list. Kept out of + ``load_org_record`` so that function stays a thin read + merge.""" + art = Path(artifacts_dir) + out: list[CrossDatasourceRelationship] = [] + + # Source 2 — sidecar at the artifacts-dir root; key `relationships:` or a bare list (mirrors + # loader._load_cross_rels). Entries are already in the model's field shape. + sidecar = art / BRIDGES_FILENAME + if sidecar.exists(): + doc = yaml.safe_load(sidecar.read_text(encoding="utf-8")) or {} + for r in doc.get("relationships", doc if isinstance(doc, list) else []): + out.append(CrossDatasourceRelationship(**r)) + + # Source 3 — legacy migration: the skill-side file the connect flow wrote before bridges were a + # model type. Its `from_profile`/`to_profile` become `from_datasource`/`to_datasource`; the edge is + # a `split`/`unreviewed` bridge stamped with a `migrated_from` marker (so nothing is lost, and a + # future write is idempotent). Read-only here — the migration is surfaced, not persisted. + legacy = art / LEGACY_BRIDGES_PATH[0] / LEGACY_BRIDGES_PATH[1] + if legacy.exists(): + doc = yaml.safe_load(legacy.read_text(encoding="utf-8")) or {} + for r in doc.get("relationships", doc if isinstance(doc, list) else []): + out.append(_migrate_legacy_bridge(r, str(legacy))) + return out + + +def _migrate_legacy_bridge(entry: dict, source_file: str) -> CrossDatasourceRelationship: + """Map one legacy ``cross_profile_relationships.yaml`` entry into a ``CrossDatasourceRelationship``. + The legacy shape keyed the endpoints on ``from_profile``/``to_profile`` (a profile = a datasource); + everything else carries. The edge is unreviewed by construction — a migrated guess a human still + signs off — and split (a cross-datasource edge spans two engines).""" + return CrossDatasourceRelationship( + from_datasource=entry["from_profile"], + to_datasource=entry["to_profile"], + from_dataset=entry["from_dataset"], + to_dataset=entry["to_dataset"], + from_columns=entry["from_columns"], + to_columns=entry["to_columns"], + executable="split", + relationship=entry.get("relationship"), + description=entry.get("description", ""), + name=entry.get("name"), + review_state="unreviewed", + migrated_from=MigratedFrom(source_file=source_file), + ) def ensure_org_record(artifacts_dir: str | Path) -> OrgRecord: diff --git a/packages/agami-core/src/semantic_model/validator.py b/packages/agami-core/src/semantic_model/validator.py index 8b45cd30..d6500ccd 100644 --- a/packages/agami-core/src/semantic_model/validator.py +++ b/packages/agami-core/src/semantic_model/validator.py @@ -39,6 +39,7 @@ import hashlib from dataclasses import dataclass, field +from pathlib import Path from typing import Optional try: @@ -50,6 +51,7 @@ _HAVE_SQLGLOT = False from .models import ( + CrossDatasourceRelationship, CrossSubjectAreaRelationship, Datasource, Entity, @@ -228,6 +230,123 @@ def validate(org: Datasource, *, cache: "ValidationCache | None" = None) -> Vali return res +# --------------------------------------------------------------------------- +# Deployment-level pass (F16 / ACE-072): cross-datasource bridges +# --------------------------------------------------------------------------- + + +def validate_deployment(artifacts_dir: str | Path) -> ValidationResult: + """Validate the deployment-level cross-datasource bridges on the ``OrgRecord``. + + A bridge links a key in one datasource to the same entity in another, so it can only be checked + with EVERY datasource's model in view — more than the per-profile ``validate(org)`` sees. This + loads the record (with its merged bridges) plus each attached datasource model and, for each + bridge, rejects ``same_engine`` (structurally impossible across two engines) and any endpoint + whose datasource / dataset (``schema.table``) / column doesn't resolve in that datasource's model. + + Returns an empty (``ok``) result when there is no record or no bridges — a pre-F16 deployment is a + no-op, never an error.""" + from . import loader as _loader + from . import org_record as _org_record + + res = ValidationResult() + record = _org_record.load_org_record(artifacts_dir) + if record is None or not record.cross_datasource_relationships: + return res + + art = Path(artifacts_dir) + models: dict[str, Datasource] = {} + for name in record.datasources: + try: + models[name] = _loader.load_datasource(art / name) + except Exception: + # A profile listed on the record but unreadable on disk is the per-profile validate's + # problem, not this pass's — skip it so an endpoint pointing at it just reports unresolved. + continue + + for bridge in record.cross_datasource_relationships: + _check_bridge(bridge, models, res) + return res + + +def _bridge_label(bridge: CrossDatasourceRelationship) -> str: + """Readable identity for a bridge in a finding message (bridges are usually anonymous — + identified by their endpoints, not a name).""" + return (f"{bridge.from_datasource}.{bridge.from_dataset} -> " + f"{bridge.to_datasource}.{bridge.to_dataset}") + + +def _check_bridge( + bridge: CrossDatasourceRelationship, + models: dict[str, Datasource], + res: ValidationResult, +) -> None: + # same_engine is impossible across two datasources (two engines). The model validator already + # refuses to construct one; re-assert here for a bridge that reached the deployment some other way. + if bridge.executable == "same_engine": + res.error( + "cross_datasource_executable_mismatch", + f"cross-datasource bridge {_bridge_label(bridge)}: executable='same_engine' is invalid " + "across two datasources; use 'split' or 'informational'", + ) + return + _check_bridge_endpoint( + res, bridge, models, bridge.from_datasource, bridge.from_dataset, bridge.from_columns) + _check_bridge_endpoint( + res, bridge, models, bridge.to_datasource, bridge.to_dataset, bridge.to_columns) + + +def _check_bridge_endpoint( + res: ValidationResult, + bridge: CrossDatasourceRelationship, + models: dict[str, Datasource], + datasource: str, + dataset: str, + columns: list[str], +) -> None: + """Resolve one endpoint (datasource -> dataset -> column(s)) against the loaded models, emitting + a ``cross_datasource_endpoint_unresolved`` error at the first level that doesn't resolve.""" + label = _bridge_label(bridge) + model = models.get(datasource) + if model is None: + res.error( + "cross_datasource_endpoint_unresolved", + f"cross-datasource bridge {label}: datasource {datasource!r} is not an attached " + "datasource of this deployment", + ) + return + table = _resolve_dataset(model, dataset) + if table is None: + res.error( + "cross_datasource_endpoint_unresolved", + f"cross-datasource bridge {label}: dataset {dataset!r} not found in datasource " + f"{datasource!r}", + ) + return + present = {c.name for c in table.columns} + missing = [c for c in columns if c not in present] + if missing: + res.error( + "cross_datasource_endpoint_unresolved", + f"cross-datasource bridge {label}: column(s) {missing} not found in dataset " + f"{dataset!r} of datasource {datasource!r}", + ) + + +def _resolve_dataset(model: Datasource, dataset: str) -> Optional[Table]: + """Find the ``Table`` a ``schema.table`` dataset names in a datasource's model. The schema half + is matched leniently — a schema-less model (SQLite) or a bare ``table`` dataset resolves on the + table name alone — so the bridge author isn't forced to schema-qualify when it's unambiguous.""" + target_schema, target_table = dataset.rsplit(".", 1) if "." in dataset else (None, dataset) + for sa in model.subject_areas: + for t in sa.tables_defined: + if t.name != target_table: + continue + if target_schema is None or t.schema_name is None or t.schema_name == target_schema: + return t + return None + + # --------------------------------------------------------------------------- # Individual rules # --------------------------------------------------------------------------- @@ -849,6 +968,7 @@ def format_result(res: ValidationResult) -> str: "Finding", "ValidationResult", "validate", + "validate_deployment", "recommended_confidence_cap", "format_result", "SIZING_WARN", diff --git a/tests/test_cross_datasource_bridge.py b/tests/test_cross_datasource_bridge.py new file mode 100644 index 00000000..81a3f1f4 --- /dev/null +++ b/tests/test_cross_datasource_bridge.py @@ -0,0 +1,257 @@ +"""F16 / ACE-072: the cross-datasource bridge as a first-class, validated model type. + +A bridge records that a key in ONE datasource is the same entity as a key in ANOTHER (e.g. +``accounts.account_key`` in a CRM = ``customers.account_key`` in an ERP). Before this it lived only +as a skill-side file — unvalidated, and lost on a machine re-onboard. These tests pin: the model +type (load / validate / round-trip + its endpoint validator), the deployment-level validation pass +(same_engine and unresolved endpoints rejected, a good bridge accepted), and the legacy migration +(the old file is surfaced idempotently on load). + +Synthetic fixtures only (``acme`` datasources keyed on ``account_key``); no network is touched +(file I/O + pydantic only), so ``tests/test_privacy_no_network.py`` stays the separate egress gate. +""" + +from __future__ import annotations + +import sys +from pathlib import Path + +import pytest + +pytest.importorskip("pydantic") +pytest.importorskip("yaml") + +PKG_SRC = Path(__file__).resolve().parent.parent / "packages" / "agami-core" / "src" +if str(PKG_SRC) not in sys.path: + sys.path.insert(0, str(PKG_SRC)) + +import yaml # noqa: E402 +from semantic_model import build, validator # noqa: E402 +from semantic_model import org_record as OR # noqa: E402 +from semantic_model.models import ( # noqa: E402 + Column, + CrossDatasourceRelationship, + Datasource, + StorageConnection, + SubjectArea, + Table, + TableRef, +) + + +def _datasource(name: str, schema: str, table: str, key: str) -> Datasource: + """A minimal one-table datasource whose table carries `key` (the bridge join column).""" + tbl = Table( + name=table, + schema=schema, + storage_connection="db", + grain=[key], + columns=[Column(name=key, type="string"), Column(name="amount", type="decimal")], + ) + sa = SubjectArea( + name="core", + tables_defined=[tbl], + tables=[TableRef(table=table, schema=schema, storage_connection="db")], + ) + return Datasource( + datasource=name, + storage_connections=[StorageConnection(name="db", storage_type="PostgreSQL")], + subject_areas=[sa], + ) + + +def _bridge(**over) -> CrossDatasourceRelationship: + base = dict( + from_datasource="acme_crm", + to_datasource="acme_erp", + from_dataset="crm.accounts", + to_dataset="erp.customers", + from_columns=["account_key"], + to_columns=["account_key"], + ) + base.update(over) + return CrossDatasourceRelationship(**base) + + +def _deployment(tmp_path: Path) -> Path: + """A 2-datasource deployment on disk (acme_crm + acme_erp, both keyed on account_key), with an + auto-maintained OrgRecord. Returns the artifacts dir.""" + build.write_tree( + _datasource("acme_crm", "crm", "accounts", "account_key"), tmp_path / "acme_crm" + ) + build.write_tree( + _datasource("acme_erp", "erp", "customers", "account_key"), tmp_path / "acme_erp" + ) + return tmp_path + + +def _set_bridges(art: Path, bridges: list[CrossDatasourceRelationship]) -> None: + record = OR.load_org_record(art) + OR.write_org_record(art, record.model_copy(update={"cross_datasource_relationships": bridges})) + + +# --------------------------------------------------------------------------- model + + +def test_bridge_loads_validates_and_round_trips(): + b = _bridge(description="same customer across CRM and ERP") + assert b.executable == "split" # default: never same_engine + assert b.review_state == "unreviewed" # trust-block default copied from Relationship + # Lossless YAML round-trip through the model. + reloaded = CrossDatasourceRelationship( + **yaml.safe_load(yaml.safe_dump(b.model_dump(mode="json"))) + ) + assert reloaded == b + + +def test_endpoint_key_ignores_description_and_trust(): + # De-dup identity is the endpoints only — two edges that differ just in prose collapse to one. + assert _bridge(description="a").endpoint_key == _bridge(description="b").endpoint_key + + +def test_model_rejects_same_engine(): + with pytest.raises(ValueError, match="same_engine"): + _bridge(executable="same_engine") + + +def test_model_rejects_empty_and_mismatched_columns(): + with pytest.raises(ValueError, match="non-empty"): + _bridge(from_columns=[], to_columns=[]) + with pytest.raises(ValueError, match="equal length"): + _bridge(from_columns=["account_key"], to_columns=["account_key", "region"]) + + +# --------------------------------------------------------------------------- deployment validation + + +def test_deployment_accepts_a_valid_bridge(tmp_path): + art = _deployment(tmp_path) + _set_bridges(art, [_bridge()]) + res = validator.validate_deployment(art) + assert res.ok, res.errors + + +def test_deployment_no_record_is_a_noop(tmp_path): + # A pre-F16 layout (no organization.yaml) validates as an empty, ok result — never an error. + res = validator.validate_deployment(tmp_path) + assert res.ok and not res.findings + + +def test_deployment_rejects_unknown_datasource(tmp_path): + art = _deployment(tmp_path) + _set_bridges(art, [_bridge(to_datasource="acme_missing")]) + res = validator.validate_deployment(art) + assert not res.ok + assert any(f.code == "cross_datasource_endpoint_unresolved" for f in res.findings) + + +def test_deployment_rejects_unknown_dataset(tmp_path): + art = _deployment(tmp_path) + _set_bridges(art, [_bridge(from_dataset="crm.nonexistent")]) + res = validator.validate_deployment(art) + assert not res.ok + assert any(f.code == "cross_datasource_endpoint_unresolved" for f in res.findings) + + +def test_deployment_rejects_unknown_column(tmp_path): + art = _deployment(tmp_path) + _set_bridges(art, [_bridge(to_columns=["not_a_column"])]) + res = validator.validate_deployment(art) + assert not res.ok + assert any(f.code == "cross_datasource_endpoint_unresolved" for f in res.findings) + + +def test_deployment_tolerates_an_unreadable_profile(tmp_path): + # A profile still listed on the record but broken on disk (its datasource.yaml gone) is skipped, + # not crashed on — the per-profile validate owns that failure. A bridge into it then reports the + # endpoint unresolved rather than raising. + art = _deployment(tmp_path) + (art / "acme_erp" / "datasource.yaml").unlink() # record still lists acme_erp + _set_bridges(art, [_bridge()]) + res = validator.validate_deployment(art) + assert not res.ok + assert any(f.code == "cross_datasource_endpoint_unresolved" for f in res.findings) + + +def test_deployment_rejects_same_engine_bridge(tmp_path): + # same_engine can't be CONSTRUCTED (the model validator refuses), so it reaches the deployment + # only via a hand-built object; model_construct bypasses validation to exercise that branch. + from semantic_model import loader + + art = _deployment(tmp_path) + models = {n: loader.load_datasource(art / n) for n in ("acme_crm", "acme_erp")} + bad = CrossDatasourceRelationship.model_construct( + from_datasource="acme_crm", + to_datasource="acme_erp", + from_dataset="crm.accounts", + to_dataset="erp.customers", + from_columns=["account_key"], + to_columns=["account_key"], + executable="same_engine", + ) + res = validator.ValidationResult() + validator._check_bridge(bad, models, res) + assert not res.ok + assert any(f.code == "cross_datasource_executable_mismatch" for f in res.findings) + + +def test_validate_cli_runs_the_deployment_pass(tmp_path): + from semantic_model import cli + + art = _deployment(tmp_path) + _set_bridges(art, [_bridge(from_columns=["not_a_column"])]) + # `sm validate ` now folds in the deployment pass — a broken bridge fails the profile's + # validate even though the profile itself is fine. + assert cli.main(["validate", str(art / "acme_crm")]) == 1 + + +# --------------------------------------------------------------------------- sidecar + legacy migration + + +def test_sidecar_bridges_merge_on_load(tmp_path): + art = _deployment(tmp_path) + (art / OR.BRIDGES_FILENAME).write_text( + yaml.safe_dump({"relationships": [_bridge().model_dump(mode="json")]}), encoding="utf-8" + ) + record = OR.load_org_record(art) + assert len(record.cross_datasource_relationships) == 1 + assert record.cross_datasource_relationships[0].from_dataset == "crm.accounts" + + +def test_legacy_file_migrates_idempotently(tmp_path): + art = _deployment(tmp_path) + legacy = art / "local" / "cross_profile_relationships.yaml" + legacy.parent.mkdir(parents=True, exist_ok=True) + legacy.write_text( + yaml.safe_dump( + { + "relationships": [ + { + "name": "customer_bridge", + "from_profile": "acme_crm", + "to_profile": "acme_erp", + "from_dataset": "crm.accounts", + "to_dataset": "erp.customers", + "from_columns": ["account_key"], + "to_columns": ["account_key"], + } + ] + } + ), + encoding="utf-8", + ) + + record = OR.load_org_record(art) + assert len(record.cross_datasource_relationships) == 1 + migrated = record.cross_datasource_relationships[0] + assert migrated.from_datasource == "acme_crm" # from_profile -> from_datasource + assert migrated.executable == "split" and migrated.review_state == "unreviewed" + assert migrated.name == "customer_bridge" + assert migrated.migrated_from is not None # provenance stamped so a future write is idempotent + + # Re-loading (or an inline copy of the same edge) never duplicates — deduped by endpoint. + assert len(OR.load_org_record(art).cross_datasource_relationships) == 1 + _set_bridges( + art, [migrated.model_copy(update={"migrated_from": None})] + ) # same endpoints, inline + assert len(OR.load_org_record(art).cross_datasource_relationships) == 1 From 288aeff9e718328c13b8e8c46a52f7560745b242 Mon Sep 17 00:00:00 2001 From: Ashwin Ramachandran Date: Tue, 28 Jul 2026 10:15:22 +0530 Subject: [PATCH 2/5] =?UTF-8?q?fix(core):=20address=20ACE-072=20review=20?= =?UTF-8?q?=E2=80=94=20leniency,=20endpoint=20message,=20dataset=20schema?= =?UTF-8?q?=20match=20(ACE-072)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../agami-core/src/semantic_model/models.py | 2 + .../src/semantic_model/org_record.py | 14 ++- .../src/semantic_model/validator.py | 35 ++++-- tests/test_cross_datasource_bridge.py | 117 +++++++++++++++++- 4 files changed, 154 insertions(+), 14 deletions(-) diff --git a/packages/agami-core/src/semantic_model/models.py b/packages/agami-core/src/semantic_model/models.py index 38a36a9b..f548acd2 100644 --- a/packages/agami-core/src/semantic_model/models.py +++ b/packages/agami-core/src/semantic_model/models.py @@ -669,6 +669,8 @@ def _endpoints(self) -> "CrossDatasourceRelationship": def endpoint_key(self) -> tuple: """The identity tuple used to de-duplicate bridges merged from several sources (inline + sidecar + legacy migration), so a re-load never duplicates an edge.""" + # Direction-sensitive BY DESIGN: a reverse B->A edge is a distinct declared bridge, not a + # duplicate of A->B — so the from/to halves are NOT order-normalized here. return ( self.from_datasource, self.from_dataset, diff --git a/packages/agami-core/src/semantic_model/org_record.py b/packages/agami-core/src/semantic_model/org_record.py index 5b5699b5..64fe0b2b 100644 --- a/packages/agami-core/src/semantic_model/org_record.py +++ b/packages/agami-core/src/semantic_model/org_record.py @@ -97,7 +97,12 @@ def _load_bridges(artifacts_dir: str | Path) -> list[CrossDatasourceRelationship if sidecar.exists(): doc = yaml.safe_load(sidecar.read_text(encoding="utf-8")) or {} for r in doc.get("relationships", doc if isinstance(doc, list) else []): - out.append(CrossDatasourceRelationship(**r)) + try: + out.append(CrossDatasourceRelationship(**r)) + except Exception: + # load_org_record is lenient by contract (never raises — it's on runtime paths). A + # malformed entry is skipped so its valid siblings in the same file still load. + continue # Source 3 — legacy migration: the skill-side file the connect flow wrote before bridges were a # model type. Its `from_profile`/`to_profile` become `from_datasource`/`to_datasource`; the edge is @@ -107,7 +112,12 @@ def _load_bridges(artifacts_dir: str | Path) -> list[CrossDatasourceRelationship if legacy.exists(): doc = yaml.safe_load(legacy.read_text(encoding="utf-8")) or {} for r in doc.get("relationships", doc if isinstance(doc, list) else []): - out.append(_migrate_legacy_bridge(r, str(legacy))) + try: + out.append(_migrate_legacy_bridge(r, str(legacy))) + except Exception: + # Same leniency: a legacy entry missing a required key is skipped, not raised, so a + # single malformed row can't take down a runtime load_org_record call. + continue return out diff --git a/packages/agami-core/src/semantic_model/validator.py b/packages/agami-core/src/semantic_model/validator.py index d6500ccd..d2dc767a 100644 --- a/packages/agami-core/src/semantic_model/validator.py +++ b/packages/agami-core/src/semantic_model/validator.py @@ -256,16 +256,20 @@ def validate_deployment(artifacts_dir: str | Path) -> ValidationResult: art = Path(artifacts_dir) models: dict[str, Datasource] = {} + # Datasources LISTED on the record but that failed to load — distinct from a name never attached + # at all, so the endpoint message can tell "attached but broken" from "not attached" (fix #2). + failed_to_load: set[str] = set() for name in record.datasources: try: models[name] = _loader.load_datasource(art / name) except Exception: # A profile listed on the record but unreadable on disk is the per-profile validate's # problem, not this pass's — skip it so an endpoint pointing at it just reports unresolved. + failed_to_load.add(name) continue for bridge in record.cross_datasource_relationships: - _check_bridge(bridge, models, res) + _check_bridge(bridge, models, failed_to_load, res) return res @@ -279,6 +283,7 @@ def _bridge_label(bridge: CrossDatasourceRelationship) -> str: def _check_bridge( bridge: CrossDatasourceRelationship, models: dict[str, Datasource], + failed_to_load: set[str], res: ValidationResult, ) -> None: # same_engine is impossible across two datasources (two engines). The model validator already @@ -291,15 +296,18 @@ def _check_bridge( ) return _check_bridge_endpoint( - res, bridge, models, bridge.from_datasource, bridge.from_dataset, bridge.from_columns) + res, bridge, models, failed_to_load, + bridge.from_datasource, bridge.from_dataset, bridge.from_columns) _check_bridge_endpoint( - res, bridge, models, bridge.to_datasource, bridge.to_dataset, bridge.to_columns) + res, bridge, models, failed_to_load, + bridge.to_datasource, bridge.to_dataset, bridge.to_columns) def _check_bridge_endpoint( res: ValidationResult, bridge: CrossDatasourceRelationship, models: dict[str, Datasource], + failed_to_load: set[str], datasource: str, dataset: str, columns: list[str], @@ -309,10 +317,16 @@ def _check_bridge_endpoint( label = _bridge_label(bridge) model = models.get(datasource) if model is None: + # Fail-closed either way, but distinguish the two causes: a datasource the record lists but + # couldn't load is a broken-model problem to fix (not a bad bridge), whereas an unlisted name + # is a genuinely wrong endpoint. + if datasource in failed_to_load: + reason = (f"datasource {datasource!r} is attached but its model failed to load") + else: + reason = (f"datasource {datasource!r} is not an attached datasource of this deployment") res.error( "cross_datasource_endpoint_unresolved", - f"cross-datasource bridge {label}: datasource {datasource!r} is not an attached " - "datasource of this deployment", + f"cross-datasource bridge {label}: {reason}", ) return table = _resolve_dataset(model, dataset) @@ -338,13 +352,18 @@ def _resolve_dataset(model: Datasource, dataset: str) -> Optional[Table]: is matched leniently — a schema-less model (SQLite) or a bare ``table`` dataset resolves on the table name alone — so the bridge author isn't forced to schema-qualify when it's unambiguous.""" target_schema, target_table = dataset.rsplit(".", 1) if "." in dataset else (None, dataset) + lenient: Optional[Table] = None for sa in model.subject_areas: for t in sa.tables_defined: if t.name != target_table: continue - if target_schema is None or t.schema_name is None or t.schema_name == target_schema: - return t - return None + if target_schema is not None and t.schema_name == target_schema: + return t # exact schema match wins outright + if target_schema is None or t.schema_name is None: + # schema-less model (SQLite) or bare-table dataset: remember it, but keep scanning in + # case an exact-schema table exists further on (fix #3 — prefer the exact match). + lenient = lenient or t + return lenient # --------------------------------------------------------------------------- diff --git a/tests/test_cross_datasource_bridge.py b/tests/test_cross_datasource_bridge.py index 81a3f1f4..68dd4032 100644 --- a/tests/test_cross_datasource_bridge.py +++ b/tests/test_cross_datasource_bridge.py @@ -104,9 +104,15 @@ def test_bridge_loads_validates_and_round_trips(): assert reloaded == b -def test_endpoint_key_ignores_description_and_trust(): - # De-dup identity is the endpoints only — two edges that differ just in prose collapse to one. +def test_endpoint_key_ignores_description_and_trust_fields(): + # De-dup identity is the endpoints only — two edges that differ just in prose OR in a trust + # field (review_state here) collapse to one, so a re-load never keeps both. assert _bridge(description="a").endpoint_key == _bridge(description="b").endpoint_key + assert ( + _bridge(review_state="unreviewed").endpoint_key + == _bridge(review_state="approved", signed_off_by="you@example.com", + signed_off_at="2026-01-01", signed_off_role="admin").endpoint_key + ) def test_model_rejects_same_engine(): @@ -173,6 +179,62 @@ def test_deployment_tolerates_an_unreadable_profile(tmp_path): assert any(f.code == "cross_datasource_endpoint_unresolved" for f in res.findings) +def test_deployment_distinguishes_failed_load_from_unattached(tmp_path): + # Two ways an endpoint's datasource can be unresolvable, same fail-closed code but distinct text: + # (a) a datasource the record LISTS but whose model won't load -> "attached but ... failed to load" + # (b) a name that was never attached at all -> "not an attached datasource" + art = _deployment(tmp_path) + (art / "acme_erp" / "datasource.yaml").unlink() # record still lists acme_erp, now unreadable + _set_bridges(art, [_bridge(), _bridge(to_datasource="acme_missing")]) + res = validator.validate_deployment(art) + assert not res.ok + msgs = [f.message for f in res.findings if f.code == "cross_datasource_endpoint_unresolved"] + assert any("attached but its model failed to load" in m for m in msgs) + assert any("is not an attached datasource" in m for m in msgs) + + +def _table(name: str, schema, key: str = "account_key") -> Table: + return Table( + name=name, + schema=schema, + storage_connection="db", + grain=[key], + columns=[Column(name=key, type="string")], + ) + + +def test_resolve_dataset_prefers_exact_schema_over_lenient(): + # Two same-named tables — one schema-qualified `crm`, one schema-less (SQLite-style). A + # `crm.accounts` dataset must bind the exact-schema table, not the lenient (schema=None) one. + exact = _table("accounts", "crm") + lenient = _table("accounts", None) + model = Datasource( + datasource="acme", + storage_connections=[StorageConnection(name="db", storage_type="PostgreSQL")], + subject_areas=[ + SubjectArea(name="a", tables_defined=[lenient], + tables=[TableRef(table="accounts", schema=None, storage_connection="db")]), + SubjectArea(name="b", tables_defined=[exact], + tables=[TableRef(table="accounts", schema="crm", storage_connection="db")]), + ], + ) + assert validator._resolve_dataset(model, "crm.accounts") is exact + # Keep the leniency: a bare table (no schema) still resolves on the name alone. + assert validator._resolve_dataset(model, "accounts") is not None + + +def test_resolve_dataset_lenient_when_model_is_schema_less(): + # A schema-less model (SQLite) resolves a schema-qualified dataset on the table name alone. + model = Datasource( + datasource="acme", + storage_connections=[StorageConnection(name="db", storage_type="SQLite")], + subject_areas=[SubjectArea( + name="a", tables_defined=[_table("accounts", None)], + tables=[TableRef(table="accounts", schema=None, storage_connection="db")])], + ) + assert validator._resolve_dataset(model, "crm.accounts") is not None + + def test_deployment_rejects_same_engine_bridge(tmp_path): # same_engine can't be CONSTRUCTED (the model validator refuses), so it reaches the deployment # only via a hand-built object; model_construct bypasses validation to exercise that branch. @@ -190,7 +252,7 @@ def test_deployment_rejects_same_engine_bridge(tmp_path): executable="same_engine", ) res = validator.ValidationResult() - validator._check_bridge(bad, models, res) + validator._check_bridge(bad, models, set(), res) assert not res.ok assert any(f.code == "cross_datasource_executable_mismatch" for f in res.findings) @@ -218,6 +280,49 @@ def test_sidecar_bridges_merge_on_load(tmp_path): assert record.cross_datasource_relationships[0].from_dataset == "crm.accounts" +def test_malformed_sidecar_entry_does_not_raise_and_valid_sibling_loads(tmp_path): + # load_org_record is on runtime paths and is lenient by contract — a malformed sidecar entry + # (empty columns fail the model validator) must be skipped, not propagated, and a valid sibling + # in the same file still loads. + art = _deployment(tmp_path) + (art / OR.BRIDGES_FILENAME).write_text( + yaml.safe_dump( + {"relationships": [ + {"from_datasource": "acme_crm", "to_datasource": "acme_erp", + "from_dataset": "crm.accounts", "to_dataset": "erp.customers", + "from_columns": [], "to_columns": []}, # malformed: empty columns + _bridge().model_dump(mode="json"), # valid sibling + ]} + ), + encoding="utf-8", + ) + record = OR.load_org_record(art) # must not raise + assert len(record.cross_datasource_relationships) == 1 + assert record.cross_datasource_relationships[0].from_dataset == "crm.accounts" + + +def test_malformed_legacy_entry_does_not_raise_and_valid_sibling_loads(tmp_path): + # Same leniency for the legacy migration: an entry missing a required key (KeyError in the + # migrator) is skipped, and its valid sibling in the same file still migrates. + art = _deployment(tmp_path) + legacy = art / "local" / "cross_profile_relationships.yaml" + legacy.parent.mkdir(parents=True, exist_ok=True) + legacy.write_text( + yaml.safe_dump( + {"relationships": [ + {"from_profile": "acme_crm", "to_profile": "acme_erp"}, # malformed: missing datasets/columns + {"from_profile": "acme_crm", "to_profile": "acme_erp", + "from_dataset": "crm.accounts", "to_dataset": "erp.customers", + "from_columns": ["account_key"], "to_columns": ["account_key"]}, # valid sibling + ]} + ), + encoding="utf-8", + ) + record = OR.load_org_record(art) # must not raise + assert len(record.cross_datasource_relationships) == 1 + assert record.cross_datasource_relationships[0].from_datasource == "acme_crm" + + def test_legacy_file_migrates_idempotently(tmp_path): art = _deployment(tmp_path) legacy = art / "local" / "cross_profile_relationships.yaml" @@ -254,4 +359,8 @@ def test_legacy_file_migrates_idempotently(tmp_path): _set_bridges( art, [migrated.model_copy(update={"migrated_from": None})] ) # same endpoints, inline - assert len(OR.load_org_record(art).cross_datasource_relationships) == 1 + survivors = OR.load_org_record(art).cross_datasource_relationships + assert len(survivors) == 1 + # Precedence, not just count: the inline edge (migrated_from is None) is concatenated ahead of + # the legacy one, and the merge keeps the FIRST occurrence — so the inline copy is the survivor. + assert survivors[0].migrated_from is None From bc87c500bcad1f3f9b0e470b4689df0b0b2e43da Mon Sep 17 00:00:00 2001 From: Ashwin Ramachandran Date: Tue, 28 Jul 2026 10:23:42 +0530 Subject: [PATCH 3/5] fix(core): tolerate a syntactically corrupt bridge file in load_org_record (ACE-072) The per-entry parse was already lenient; wrap the sidecar/legacy yaml.safe_load too so a corrupt bridge FILE degrades to no bridges rather than raising out of the runtime-path load_org_record. Adds test_corrupt_bridge_yaml_file_does_not_raise. --- .../agami-core/src/semantic_model/org_record.py | 14 ++++++++++++-- tests/test_cross_datasource_bridge.py | 12 ++++++++++++ 2 files changed, 24 insertions(+), 2 deletions(-) diff --git a/packages/agami-core/src/semantic_model/org_record.py b/packages/agami-core/src/semantic_model/org_record.py index 64fe0b2b..9213f17d 100644 --- a/packages/agami-core/src/semantic_model/org_record.py +++ b/packages/agami-core/src/semantic_model/org_record.py @@ -95,7 +95,12 @@ def _load_bridges(artifacts_dir: str | Path) -> list[CrossDatasourceRelationship # loader._load_cross_rels). Entries are already in the model's field shape. sidecar = art / BRIDGES_FILENAME if sidecar.exists(): - doc = yaml.safe_load(sidecar.read_text(encoding="utf-8")) or {} + try: + doc = yaml.safe_load(sidecar.read_text(encoding="utf-8")) or {} + except Exception: + # A syntactically corrupt sidecar file degrades to no bridges — same never-raise + # contract as the per-entry skip below (load_org_record is on runtime paths). + doc = {} for r in doc.get("relationships", doc if isinstance(doc, list) else []): try: out.append(CrossDatasourceRelationship(**r)) @@ -110,7 +115,12 @@ def _load_bridges(artifacts_dir: str | Path) -> list[CrossDatasourceRelationship # future write is idempotent). Read-only here — the migration is surfaced, not persisted. legacy = art / LEGACY_BRIDGES_PATH[0] / LEGACY_BRIDGES_PATH[1] if legacy.exists(): - doc = yaml.safe_load(legacy.read_text(encoding="utf-8")) or {} + try: + doc = yaml.safe_load(legacy.read_text(encoding="utf-8")) or {} + except Exception: + # A corrupt legacy file is likewise skipped, not raised — a stale hand-edited file + # can't take down a runtime load_org_record call. + doc = {} for r in doc.get("relationships", doc if isinstance(doc, list) else []): try: out.append(_migrate_legacy_bridge(r, str(legacy))) diff --git a/tests/test_cross_datasource_bridge.py b/tests/test_cross_datasource_bridge.py index 68dd4032..c7552ba9 100644 --- a/tests/test_cross_datasource_bridge.py +++ b/tests/test_cross_datasource_bridge.py @@ -301,6 +301,18 @@ def test_malformed_sidecar_entry_does_not_raise_and_valid_sibling_loads(tmp_path assert record.cross_datasource_relationships[0].from_dataset == "crm.accounts" +def test_corrupt_bridge_yaml_file_does_not_raise(tmp_path): + # Beyond a bad ENTRY: a syntactically corrupt bridge FILE (unparseable YAML) must also degrade + # to no bridges rather than propagate out of the runtime-path load_org_record. + art = _deployment(tmp_path) + (art / OR.BRIDGES_FILENAME).write_text("relationships: [ : broken", encoding="utf-8") + legacy = art / "local" / "cross_profile_relationships.yaml" + legacy.parent.mkdir(parents=True, exist_ok=True) + legacy.write_text("::: not : valid : yaml", encoding="utf-8") + record = OR.load_org_record(art) # must not raise + assert record.cross_datasource_relationships == [] + + def test_malformed_legacy_entry_does_not_raise_and_valid_sibling_loads(tmp_path): # Same leniency for the legacy migration: an entry missing a required key (KeyError in the # migrator) is skipped, and its valid sibling in the same file still migrates. From 2ade01aae29d7dc92eb87a11673f20cd504cdd5e Mon Sep 17 00:00:00 2001 From: Ashwin Ramachandran Date: Tue, 28 Jul 2026 10:28:54 +0530 Subject: [PATCH 4/5] fix(core): accept a bare-list bridge file (Copilot review) (ACE-072) _load_bridges claimed to support a mapping OR a bare YAML list, but doc.get(...) raises AttributeError on a bare list before the default is used. Extract entries via an explicit shape check (_bridge_entries); add a bare-list test. --- .../agami-core/src/semantic_model/org_record.py | 15 +++++++++++++-- tests/test_cross_datasource_bridge.py | 12 ++++++++++++ 2 files changed, 25 insertions(+), 2 deletions(-) diff --git a/packages/agami-core/src/semantic_model/org_record.py b/packages/agami-core/src/semantic_model/org_record.py index 9213f17d..fb140e9d 100644 --- a/packages/agami-core/src/semantic_model/org_record.py +++ b/packages/agami-core/src/semantic_model/org_record.py @@ -84,6 +84,17 @@ def _merge_bridges( return out +def _bridge_entries(doc) -> list: + """The entry list from a parsed bridge doc — a ``relationships:`` mapping OR a bare YAML list + (both accepted, mirroring ``loader._load_cross_rels``); anything else yields ``[]``. Explicit + rather than ``doc.get(...)`` because ``.get`` on a bare list raises ``AttributeError``.""" + if isinstance(doc, dict): + return doc.get("relationships") or [] + if isinstance(doc, list): + return doc + return [] + + def _load_bridges(artifacts_dir: str | Path) -> list[CrossDatasourceRelationship]: """Load the non-inline bridge sources: the root sidecar (source 2) and the legacy skill-side file (source 3). Both are optional; a deployment with neither yields an empty list. Kept out of @@ -101,7 +112,7 @@ def _load_bridges(artifacts_dir: str | Path) -> list[CrossDatasourceRelationship # A syntactically corrupt sidecar file degrades to no bridges — same never-raise # contract as the per-entry skip below (load_org_record is on runtime paths). doc = {} - for r in doc.get("relationships", doc if isinstance(doc, list) else []): + for r in _bridge_entries(doc): try: out.append(CrossDatasourceRelationship(**r)) except Exception: @@ -121,7 +132,7 @@ def _load_bridges(artifacts_dir: str | Path) -> list[CrossDatasourceRelationship # A corrupt legacy file is likewise skipped, not raised — a stale hand-edited file # can't take down a runtime load_org_record call. doc = {} - for r in doc.get("relationships", doc if isinstance(doc, list) else []): + for r in _bridge_entries(doc): try: out.append(_migrate_legacy_bridge(r, str(legacy))) except Exception: diff --git a/tests/test_cross_datasource_bridge.py b/tests/test_cross_datasource_bridge.py index c7552ba9..43d7d4d6 100644 --- a/tests/test_cross_datasource_bridge.py +++ b/tests/test_cross_datasource_bridge.py @@ -313,6 +313,18 @@ def test_corrupt_bridge_yaml_file_does_not_raise(tmp_path): assert record.cross_datasource_relationships == [] +def test_bare_list_sidecar_loads(tmp_path): + # The sidecar accepts a bare YAML list (not only a `relationships:` mapping). This used to crash + # because `.get` was called on a list before the shape was checked; it must now load the entries. + art = _deployment(tmp_path) + (art / OR.BRIDGES_FILENAME).write_text( + yaml.safe_dump([_bridge().model_dump(mode="json")]), # bare list, no `relationships:` key + encoding="utf-8", + ) + record = OR.load_org_record(art) # must not raise + assert len(record.cross_datasource_relationships) == 1 + + def test_malformed_legacy_entry_does_not_raise_and_valid_sibling_loads(tmp_path): # Same leniency for the legacy migration: an entry missing a required key (KeyError in the # migrator) is skipped, and its valid sibling in the same file still migrates. From 23099e6db8002d2e18014a4923ae6cd33c98938d Mon Sep 17 00:00:00 2001 From: Ashwin Ramachandran Date: Tue, 28 Jul 2026 19:05:33 +0530 Subject: [PATCH 5/5] =?UTF-8?q?fix(core):=20OrgRecord=20Postgres=20storage?= =?UTF-8?q?=20is=20lossless=20=E2=80=94=20persist=20all=20buckets=20(ACE-0?= =?UTF-8?q?72)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit write/load_organization_record hand-listed 4 fields, silently dropping new OrgRecord buckets on a hosted deploy — the cross_datasource_relationships bridge among them. Dump/validate the whole record (its own stated contract), so the bridge (and future buckets) round-trip. Adds a persistence regression test. --- packages/agami-core/src/model_store.py | 26 +++++++++------------ tests/test_org_record_deploy.py | 31 +++++++++++++++++++++++++- 2 files changed, 40 insertions(+), 17 deletions(-) diff --git a/packages/agami-core/src/model_store.py b/packages/agami-core/src/model_store.py index 66b33cfe..296379b5 100644 --- a/packages/agami-core/src/model_store.py +++ b/packages/agami-core/src/model_store.py @@ -259,14 +259,12 @@ def write_organization_record(store: Store, record: OrgRecord, org_id: str = DEF UPDATE` rewrites only the content columns and preserves any hosted-owned `org_name`/`created_at`, so hosted onboarding (which INSERTs the row first) and core deploy (which upserts content) coexist on one row. Portable across SQLite (>=3.24) and Postgres. Idempotent — a redeploy replaces, never duplicates.""" - doc = json.dumps( - { - "fiscal_year_start_month": record.fiscal_year_start_month, - "display_conventions": record.display_conventions.model_dump(mode="json"), - "glossary": record.glossary, - "datasources": record.datasources, - } - ) + # Lossless: dump EVERY non-columned field into `doc` so any OrgRecord bucket (cross-datasource + # relationships/metrics, and whatever is added later) round-trips without editing this writer — + # the module's own stated contract. org_id/name/description are their own columns; the rest rides + # the JSON blob. (The prior hand-listed dict silently dropped new buckets on deploy — ACE-072's + # bridge among them.) + doc = json.dumps(record.model_dump(mode="json", exclude={"org_id", "name", "description"})) store.execute( "INSERT INTO organization (org_id, org_name, description, doc) VALUES (?, ?, ?, ?) " "ON CONFLICT (org_id) DO UPDATE SET " @@ -289,14 +287,10 @@ def load_organization_record(store: Store, org_id: str = DEFAULT_ORG) -> OrgReco return None row = rows[0] doc = json.loads(row["doc"]) if row["doc"] else {} - return OrgRecord( - org_id=org_id, - name=row["org_name"], - description=row["description"], - fiscal_year_start_month=doc.get("fiscal_year_start_month"), - display_conventions=doc.get("display_conventions") or {}, - glossary=doc.get("glossary") or {}, - datasources=doc.get("datasources") or [], + # Lossless rebuild: the columned fields + every field the writer put in `doc` (the same + # `model_validate` path the on-disk loader uses), so new buckets ride back automatically. + return OrgRecord.model_validate( + {"org_id": org_id, "name": row["org_name"], "description": row["description"], **doc} ) diff --git a/tests/test_org_record_deploy.py b/tests/test_org_record_deploy.py index da853e2c..82b6ebc8 100644 --- a/tests/test_org_record_deploy.py +++ b/tests/test_org_record_deploy.py @@ -26,7 +26,11 @@ import model_store as MS # noqa: E402 import tools # noqa: E402 from semantic_model import org_record as OR # noqa: E402 -from semantic_model.models import DisplayConventions, OrgRecord # noqa: E402 +from semantic_model.models import ( # noqa: E402 + CrossDatasourceRelationship, + DisplayConventions, + OrgRecord, +) from store import Store # noqa: E402 @@ -57,6 +61,31 @@ def test_write_then_load_round_trips_losslessly(): s.close() +def test_cross_datasource_relationships_survive_the_round_trip(): + # Regression: the OrgRecord writer used to hand-list its fields and silently DROP new buckets — so + # the ACE-072 bridge vanished on a Postgres deploy. The lossless dump/validate must round-trip it. + s = _store() + rec = _full_record().model_copy( + update={ + "cross_datasource_relationships": [ + CrossDatasourceRelationship( + from_datasource="acme_crm", + to_datasource="acme_erp", + from_dataset="crm.accounts", + to_dataset="erp.customers", + from_columns=["account_key"], + to_columns=["account_key"], + ) + ] + } + ) + MS.write_organization_record(s, rec, org_id="org1") + loaded = MS.load_organization_record(s, "org1") + assert loaded == rec # the bridge survived — not silently dropped + assert loaded.cross_datasource_relationships[0].from_columns == ["account_key"] + s.close() + + def test_second_write_replaces_not_duplicates(): s = _store() MS.write_organization_record(s, _full_record(), org_id="org1")