ACE-072: cross-datasource bridge as a validated model type

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:

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

@@ -604,6 +604,83 @@ 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."""
+        # 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,
+            tuple(self.from_columns),
+            self.to_datasource,
+            self.to_dataset,
+            tuple(self.to_columns),
+        )
+
+
 # ---------------------------------------------------------------------------
 # Subject Area (the primary semantic unit)
 # ---------------------------------------------------------------------------
@@ -713,6 +790,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 +833,7 @@ def _fy_month(cls, v: Optional[int]) -> Optional[int]:
     "Metric",
     "Relationship",
     "CrossSubjectAreaRelationship",
+    "CrossDatasourceRelationship",
     "SubjectArea",
     "Datasource",
     "DisplayConventions",

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,117 @@ def narrative_path(artifacts_dir: str | Path) -> Path:
 def load_org_record(artifacts_dir: str | Path) -> Optional[OrgRecord]:
     """Return the ``OrgRecord`` at ``<artifacts_dir>/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 _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
+    ``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():
+        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 _bridge_entries(doc):
+            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
+    # 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():
+        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 _bridge_entries(doc):
+            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
+
+
+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: