feat: explicit retrieval-sources override with {edge_id} record-URL templating

docs/configuration/table.md

@@ -428,7 +428,7 @@ provenance:
 
 #### Manual provenance override
 
-Use `provenance.override` when a table comes from another knowledge graph or source system whose Translator provenance cannot be derived from a PMC/PMID publication. The override is wired like the other Tablassert model classes and wins over the repo/publication auto-generation for that section's upstream sources, publications, and KL/AT. The primary entry of the edge `sources` list (`resource_role: primary_knowledge_source`) is **not** overridable per section; it always derives from the graph-level `infores` (see [Graph](graph.md)); put manual infores CURIEs in `upstream_resource_ids`.
+Use `provenance.override` when a table comes from another knowledge graph or source system whose Translator provenance cannot be derived from a PMC/PMID publication. The override is wired like the other Tablassert model classes and wins over the repo/publication auto-generation for that section's upstream sources, publications, and KL/AT. The primary entry of the edge `sources` list (`resource_role: primary_knowledge_source`) is **not** re-targetable per section via `upstream_resource_ids`; it derives from the graph-level `infores` (see [Graph](graph.md)) unless an explicit [`sources` template](#explicit-sources-template) is given. Put manual infores CURIEs in `upstream_resource_ids`.
 
 ```yaml
 provenance:
@@ -445,13 +445,49 @@ Override fields:
 
 | Field | Type | Required | Description |
 |-------|------|----------|-------------|
+| `sources` | List[Object] | No | Explicit retrieval-`sources` entry templates replacing the derived primary/upstream emission entirely. Mutually exclusive with `upstream_resource_ids` and `upstream_source_record_urls` (the template subsumes both). See [Explicit sources template](#explicit-sources-template). |
 | `upstream_resource_ids` | List[String] | No | Manual upstream source infores CURIEs replacing the repo-derived `PMC`/`PMID` source map, the sanctioned place for manual infores. Each entry must start with `infores:`. |
 | `upstream_source_record_urls` | Map[String, List[URL]] | No | Per-upstream source record URLs keyed by infores CURIE; every key must appear in `upstream_resource_ids`. When set, the section's `source.url` values serve the RIG only and are NOT emitted on the primary `sources` entry; each listed upstream supporting entry carries its own `source_record_urls` instead. |
 | `publications` | List[String] | No | Manual publication CURIEs. Entries must currently start with `PMCID:`; PMID compatibility for manual overrides is intentionally deferred. |
 | `knowledge_level` | String | No | Override-specific KL value. Defaults to `statistical_association`. |
 | `agent_type` | String | No | Override-specific AT value. Defaults to `data_analysis_pipeline`. |
 
-Tablassert emits the graph-level infores (or `infores:<graph-name>` when unset) as the primary entry of the Biolink `sources` list on each edge, `{resource_id: "infores:multiomics-kg", resource_role: "primary_knowledge_source", upstream_resource_ids: [...], source_record_urls: [...]}`, with one additional `supporting_data_source` entry per upstream. When `override.upstream_source_record_urls` is set, the primary entry emits no `source_record_urls` and each mapped supporting entry carries its own instead. No flat `primary_knowledge_source` scalar is emitted: current translator-ingests practice carries retrieval provenance only in `sources`, and the Biolink `RetrievalSource` class is where `resource_id` / `upstream_resource_ids` / `source_record_urls` are defined. Each retrieval-source entry uses `resource_id` as its sole identifier. The override cannot set a per-section primary source; manual infores CURIEs belong in `upstream_resource_ids`. Older flat `resource_id` / `primary_knowledge_source` output has been removed so generated KGX matches the Biolink edge contract.
+Tablassert emits the graph-level infores (or `infores:<graph-name>` when unset) as the primary entry of the Biolink `sources` list on each edge, `{resource_id: "infores:multiomics-kg", resource_role: "primary_knowledge_source", upstream_resource_ids: [...], source_record_urls: [...]}`, with one additional `supporting_data_source` entry per upstream. When `override.upstream_source_record_urls` is set, the primary entry emits no `source_record_urls` and each mapped supporting entry carries its own instead. When `override.sources` is set, that derivation is skipped entirely and the template entries are emitted verbatim, in order. No flat `primary_knowledge_source` scalar is emitted: current translator-ingests practice carries retrieval provenance only in `sources`, and the Biolink `RetrievalSource` class is where `resource_id` / `upstream_resource_ids` / `source_record_urls` are defined. Each retrieval-source entry uses `resource_id` as its sole identifier. Without `override.sources`, the override cannot set a per-section primary source; manual infores CURIEs belong in `upstream_resource_ids`. Older flat `resource_id` / `primary_knowledge_source` output has been removed so generated KGX matches the Biolink edge contract.
+
+#### Explicit sources template
+
+`override.sources` replaces the derived `sources` list with an explicit, ordered template — one Biolink `RetrievalSource` struct per entry. Use it when the emitted provenance must differ structurally from the "graph infores as primary + one supporting entry per upstream" shape, e.g. to mark the graph as an `aggregator_knowledge_source` with the real primary elsewhere.
+
+```yaml
+provenance:
+  override:
+    sources:
+      - resource_id: infores:multiomics-drugapprovals
+        resource_role: aggregator_knowledge_source
+        upstream_resource_ids: [infores:dailymed, infores:faers]
+        source_record_urls:
+          - "https://db.systemsbiology.net/gestalt/cgi-pub/KGinfo.pl?id={edge_id}"
+      - resource_id: infores:faers
+        resource_role: primary_knowledge_source
+      - resource_id: infores:dailymed
+        resource_role: supporting_data_source
+```
+
+Entry fields:
+
+| Field | Type | Required | Description |
+|-------|------|----------|-------------|
+| `resource_id` | String | Yes | Infores CURIE of this entry (its sole identifier on output). Must start with `infores:`, and must be unique within the template. |
+| `resource_role` | String | Yes | One of the Biolink `ResourceRoleEnum` values: `primary_knowledge_source`, `aggregator_knowledge_source`, or `supporting_data_source`. Anything else fails KGX validation downstream, so it is rejected at config time. |
+| `upstream_resource_ids` | List[String] | No | Upstream infores CURIEs carried by this entry; each must start with `infores:`. |
+| `source_record_urls` | List[String] | No | Source record URLs carried by this entry. Each entry must be an absolute `http(s)://` URL, optionally containing the `{edge_id}` placeholder (see below). |
+
+Validation rules (all reported with error code `override-bad-sources`):
+
+- `sources` is **mutually exclusive** with `upstream_resource_ids` and `upstream_source_record_urls` — the template subsumes both, so combining them fails loudly.
+- When set, the list must be **non-empty**, `resource_id` values must be **unique**, and at least one entry must carry role `primary_knowledge_source` or `aggregator_knowledge_source`.
+
+**The `{edge_id}` placeholder.** A `source_record_urls` entry may embed the literal string `{edge_id}` to build per-edge URLs (e.g. a drill-down page keyed by edge id). The edge `id` is a deterministic content hash assigned at the final dedup stage — *after* subgraphs are written — so the placeholder cannot be interpolated during the table build. Instead, the literal placeholder is emitted into the subgraph (and is what the content hash covers, keeping ids deterministic), and a post-dedup sweep of the final `*.edges.ndjson` replaces `{edge_id}` with each record's own `id` in every `sources[].source_record_urls` string. Files without the marker are left byte-identical.
 
 ### Annotations
 

src/tablassert/errors.py

@@ -22,6 +22,7 @@
     "encoding-bad-remove-entry",
     "graph-bad-infores",
     "override-bad-publication",
+    "override-bad-sources",
     "override-bad-upstream-infores",
     "override-bad-upstream-urls",
     "provenance-bad-pmc-id",

src/tablassert/lib.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import json
 import math
 import operator
 import re
@@ -44,7 +45,7 @@
 from tablassert.enums import EncodingMethods, Files, InformationResources, Repositories, Tokens
 from tablassert.fullmap import ResolveSpec, fullmap_db_path, resolve, resolve_batch
 from tablassert.log import cat
-from tablassert.models import Encoding, NodeEncoding, Qualifier, RIGConfig, Section
+from tablassert.models import EDGE_ID_PLACEHOLDER, Encoding, NodeEncoding, Qualifier, RIGConfig, Section
 from tablassert.nlp import level_one, level_two
 from tablassert.qc import fullmap_audit
 from tablassert.rig import (
@@ -554,7 +555,12 @@ def inline_supporting_study(lf: pl.LazyFrame, study_id: str, study_name: str | N
 
 
 def retrieval_sources(
-    lf: pl.LazyFrame, primary: str, upstream: list[str], urls: list[str], upstream_urls: dict[str, list[str]] | None = None
+    lf: pl.LazyFrame,
+    primary: str,
+    upstream: list[str],
+    urls: list[str],
+    upstream_urls: dict[str, list[str]] | None = None,
+    explicit: list[dict[str, Any]] | None = None,
 ) -> pl.LazyFrame:
     """Add the Biolink ``sources`` retrieval-provenance column.
 
@@ -572,18 +578,32 @@ def retrieval_sources(
     its own ``source_record_urls`` and the primary entry emits none (the primary is
     the transforming resource, not a downloadable record).
 
+    When ``explicit`` is given, exactly those entries are emitted, in order, and
+    ``primary``/``upstream``/``urls``/``upstream_urls`` are ignored. Each entry
+    template carries ``resource_id``, ``resource_role``, and optional
+    ``upstream_resource_ids``/``source_record_urls``; ``source_record_urls`` values
+    may contain the literal ``{edge_id}`` placeholder, which is NOT resolved here
+    (the edge id is only assigned at the final dedup stage) but in a post-dedup
+    sweep of the final edges NDJSON.
+
     Args:
         lf: Source LazyFrame.
         primary: Infores CURIE of the primary knowledge source.
         upstream: Infores CURIEs of upstream/supporting data sources.
         urls: Source record URLs for the primary entry (ignored when ``upstream_urls`` is set).
         upstream_urls: Optional per-upstream source record URLs keyed by infores CURIE.
+        explicit: Optional explicit entry templates emitted verbatim, in order.
 
     Returns:
         LazyFrame with a ``sources`` ``list[struct]`` column appended.
     """
-    if upstream_urls is not None:
-        entries: list[pl.Expr] = [_retrieval_source(primary, "primary_knowledge_source", upstream)]
+    if explicit is not None:
+        entries: list[pl.Expr] = [
+            _retrieval_source(entry["resource_id"], entry["resource_role"], entry.get("upstream_resource_ids"), entry.get("source_record_urls"))
+            for entry in explicit
+        ]
+    elif upstream_urls is not None:
+        entries = [_retrieval_source(primary, "primary_knowledge_source", upstream)]
         entries.extend(_retrieval_source(x, "supporting_data_source", urls=upstream_urls.get(x)) for x in upstream)
     else:
         entries = [_retrieval_source(primary, "primary_knowledge_source", upstream, urls)]
@@ -1253,6 +1273,12 @@ def _provenance_ops(self: Self) -> list[Any]:
             if override and override.upstream_source_record_urls is not None
             else None
         )
+        # An explicit `sources` template replaces the derived primary/upstream
+        # emission entirely; the model already forbids combining it with
+        # `upstream_resource_ids`/`upstream_source_record_urls`.
+        explicit_sources: list[dict[str, Any]] | None = (
+            [entry.model_dump(exclude_none=True) for entry in override.sources] if override and override.sources is not None else None
+        )
         knowledge_level = override.knowledge_level if override else self.provenance.knowledge_level
         agent_type = override.agent_type if override else self.provenance.agent_type
         publication_values = override.publications if override else [publication_curie(self.provenance.repo, self.provenance.publication or "")]
@@ -1278,8 +1304,9 @@ def _provenance_ops(self: Self) -> list[Any]:
             # RetrievalSource); current translator-ingests emits no flat
             # `primary_knowledge_source` scalar, so neither do we. A per-upstream URL
             # mapping (override.upstream_source_record_urls) re-homes the record URLs
-            # from the primary entry onto the matching supporting entries.
-            (retrieval_sources, (primary_knowledge_source, upstream_ids, [str(u) for u in self.source.url], upstream_urls))
+            # from the primary entry onto the matching supporting entries; an explicit
+            # `sources` template (override.sources) replaces the whole derivation.
+            (retrieval_sources, (primary_knowledge_source, upstream_ids, [str(u) for u in self.source.url], upstream_urls, explicit_sources))
             if primary_knowledge_source
             else None,
             (publications, (publication_values,)) if publication_values else None,
@@ -1520,6 +1547,44 @@ def dedup_stream(p_in: Path, is_edges: bool) -> None:
     p_in.unlink()
 
 
+def _resolve_edge_id_placeholders(edges_path: Path) -> None:
+    """Resolve ``{edge_id}`` placeholders in a final edges NDJSON file.
+
+    The edge ``id`` is a deterministic content hash assigned by the Rust deduper
+    after subgraphs are written, so explicit ``override.sources`` record URLs
+    cannot embed it during the polars build: the literal placeholder is what
+    gets hashed, and this post-dedup sweep substitutes each record's own id into
+    every string inside every ``sources[].source_record_urls`` list. The pass is
+    skipped entirely when no line contains the marker (cheap substring precheck,
+    no full parse), leaving the file byte-identical.
+
+    Args:
+        edges_path: Path to the deduplicated ``*.edges.ndjson`` file.
+
+    Returns:
+        ``None``; rewrites ``edges_path`` in place via a temp file when any
+        placeholder was resolved.
+    """
+    tmp_path: Path = edges_path.with_name(edges_path.name + ".placeholder.tmp")
+    resolved: bool = False
+    with edges_path.open("r", encoding="utf-8") as src, tmp_path.open("w", encoding="utf-8") as dst:
+        for line in src:
+            if EDGE_ID_PLACEHOLDER not in line:
+                dst.write(line)
+                continue
+            record: dict[str, Any] = json.loads(line)
+            for source in record.get("sources") or []:
+                urls: list[str] | None = source.get("source_record_urls")
+                if urls:
+                    source["source_record_urls"] = [url.replace(EDGE_ID_PLACEHOLDER, record["id"]) for url in urls]
+            dst.write(json.dumps(record, ensure_ascii=False) + "\n")
+            resolved = True
+    if resolved:
+        tmp_path.replace(edges_path)
+    else:
+        tmp_path.unlink()
+
+
 def fold_unknown_to_supporting_text(lf: pl.LazyFrame) -> pl.LazyFrame:
     """Fold any non-Biolink edge column into ``supporting_text`` as ``col: value`` strings.
 
@@ -1631,6 +1696,12 @@ def _write_ndjson(
     ``compile_graph``; each commented phase boundary below is a hook point for
     the US-009 ``on_phase`` progress callback.
 
+    Dedup assigns each edge ``id`` as the deterministic content hash of the
+    PRE-resolution record -- the literal ``{edge_id}`` placeholder in explicit
+    ``override.sources`` record URLs is what gets hashed -- and the placeholder
+    sweep that follows substitutes the assigned id into the final edges file
+    only, keeping ids deterministic.
+
     Args:
         subnodes: Per-section node LazyFrames from ``_collect_subframes``.
         subedges: Per-section edge LazyFrames from ``_collect_subframes``.
@@ -1665,6 +1736,9 @@ def _write_ndjson(
     if on_phase is not None:
         on_phase("dedup")
     dedup_stream(edges_tmp, is_edges=True)
+    # The deduper hashes the record WITH the literal `{edge_id}` placeholder still
+    # in place, so edge ids stay deterministic regardless of this resolution pass.
+    _resolve_edge_id_placeholders(edges_tmp.with_suffix(""))
     dedup_stream(nodes_tmp, is_edges=False)
 
     # Phase: rig. Summaries come from the FINAL deduplicated KGX files, and the