feat: add --linkouts/-lo to build-kg for EdgeLinkouts edge deep links

docs/cli.md

@@ -315,6 +315,7 @@ The positional `GRAPH-CONFIGURATION-FILE` (also `--configuration-file`, `-f`) is
 | `--qc`, `-q` | Flag | No | `False` | Audit resolved mappings (exact -> fuzzy -> abbreviation -> SapBERT) so low-confidence edges are flagged; requires the `[qc]` extra, checked before the build starts. Also runs a final study stage that asserts over the emitted NDJSON: no duplicate node ids, every node has a non-empty `id` and `name`, every edge has a non-empty `subject`, `predicate`, and `object`, no undeclared or isolated nodes, no malformed lines, no null or empty values in any field (checked recursively), no stray whitespace (verbatim `original_*` fields excepted from the whitespace check, since they are faithful source copies), and, when any section declares a `category_override`, no edge demoted to bare `biolink:Association` (a row escaped its pin and shipped without the class-specific slots `prune_to_class` nulled into `has_supporting_studies`; the assertion is graph-wide, so one pinned section also gates edges from unpinned sections) and fails the build (non-zero exit) on any violation |
 | `--log`, `-l` | Flag | No | `False` | Enable verbose per-section logging; requires the `[log]` extra, and without it Tablassert produces no logs |
 | `--head`, `-hd` | Flag | No | `False` | Fast output-shape preview: <=5 random rows/section, cached to `.head.parquet`, never clobbers a full build |
+| `--linkouts`, `-lo` | Flag | No | `False` | Append the EdgeLinkouts per-edge record URL (`https://linkouts.skyelanegoetz.com/edges/{edge_id}`, resolved to the final edge id after dedup) to the graph infores entry (`rig.source_info.infores_id`) in every emitted edge's `sources` list, so linkouts.skyelanegoetz.com can deep-link each edge; sections declaring `provenance.override` are untouched; cached to `.linkouts.parquet`, never clobbers a full build |
 
 ```bash
 tablassert build-kg graph.yaml --qc --log

src/tablassert/cli.py

@@ -60,29 +60,31 @@
 TAXON_ALLOWLIST_PATH: Path = Path(__file__).parent / "data" / "experimental_taxa.yaml"
 
 
-def _section_store_path(h: str, head: bool = False, release: bool = False, qc: bool = False) -> Path:
+def _section_store_path(h: str, head: bool = False, release: bool = False, qc: bool = False, linkouts: bool = False) -> Path:
     """Name a section's cached subgraph parquet so build-mode flags never share a cache file.
 
     The cached parquet's content depends on the build mode: ``--head`` samples
-    rows, ``--release`` applies the release-mode significance filters, and
-    ``--qc`` drops fullmap-audit rejects. ``Tcode.collect`` quick-exits on any
-    existing store file, skipping the whole op chain, so each mode combination
-    must cache under its own suffix or one mode's build is silently reused as
-    another's.
+    rows, ``--release`` applies the release-mode significance filters, ``--qc``
+    drops fullmap-audit rejects, and ``--linkouts`` appends the EdgeLinkouts
+    record URL to the primary ``sources`` entry. ``Tcode.collect`` quick-exits
+    on any existing store file, skipping the whole op chain, so each mode
+    combination must cache under its own suffix or one mode's build is silently
+    reused as another's.
 
     Args:
         h: Section config hash (``mkhash`` of the section dict).
         head: ``--head`` preview build flag.
         release: ``--release`` build flag.
         qc: ``--qc`` build flag.
+        linkouts: ``--linkouts`` build flag.
 
     Returns:
-        Store path like ``<h>.parquet`` with ``.head`` / ``.release`` / ``.qc``
-        inserted before the extension for each set flag.
+        Store path like ``<h>.parquet`` with ``.head`` / ``.release`` / ``.qc`` /
+        ``.linkouts`` inserted before the extension for each set flag.
     """
     from tablassert.utils import STORE
 
-    suffix: str = (".head" if head else "") + (".release" if release else "") + (".qc" if qc else "")
+    suffix: str = (".head" if head else "") + (".release" if release else "") + (".qc" if qc else "") + (".linkouts" if linkouts else "")
     return STORE / f"{h}{suffix}.parquet"
 
 
@@ -272,11 +274,17 @@ def snapshot(position: int, lf: pl.LazyFrame) -> pl.LazyFrame | None:
 
 
 def build_pipeline(
-    configuration_file: Path, progress: PipelineProgress, release: bool = False, qc: bool = False, log: bool = False, head: bool = False
+    configuration_file: Path,
+    progress: PipelineProgress,
+    release: bool = False,
+    qc: bool = False,
+    log: bool = False,
+    head: bool = False,
+    linkouts: bool = False,
 ) -> None:
     """Load a graph YAML and build it through the shared in-process core."""
     graph: Graph = _load_graph(configuration_file)
-    build_graph_pipeline(graph, configuration_file, progress, release=release, qc=qc, log=log, head=head)
+    build_graph_pipeline(graph, configuration_file, progress, release=release, qc=qc, log=log, head=head, linkouts=linkouts)
 
 
 def build_graph_pipeline(
@@ -287,6 +295,7 @@ def build_graph_pipeline(
     qc: bool = False,
     log: bool = False,
     head: bool = False,
+    linkouts: bool = False,
     audit_sources: bool = True,
 ) -> None:
     """Build a validated :class:`Graph` without loading another graph YAML.
@@ -304,6 +313,7 @@ def build_graph_pipeline(
         qc: When ``True``, run quality-control audits and final study assertions.
         log: When ``True``, enable per-section verbose logging.
         head: When ``True``, build a random sample of up to five rows per section.
+        linkouts: When ``True``, append the EdgeLinkouts record URL to the graph infores entry in every edge's ``sources`` list.
     """
     from tablassert.fullmap import fullmap_db_path
     from tablassert.lib import Tcode, compile_graph, compile_subgraph
@@ -355,7 +365,7 @@ def build_graph_pipeline(
         start(f"{Path(str(s['config'])).stem} · {h[:8]}")
         # Mode flags change the cached parquet's content, so each combination caches
         # to a distinct file and can never quick-exit another mode's build.
-        store: Path = _section_store_path(h, head=head, release=release, qc=qc)
+        store: Path = _section_store_path(h, head=head, release=release, qc=qc, linkouts=linkouts)
         try:
             tcode.append(
                 Tcode.model_validate(
@@ -366,6 +376,7 @@ def build_graph_pipeline(
                         "qc": qc,
                         "release": release,
                         "head": head,
+                        "linkouts": linkouts,
                         "name": g.name,
                         "infores": g.rig.source_info.infores_id,
                         "qc_similarity_threshold": g.qc_similarity_threshold,
@@ -893,6 +904,7 @@ def build_kg(
     qc: Annotated[bool, cyclopts.Parameter(name=["--qc", "-q"], negative="")] = False,
     log: Annotated[bool, cyclopts.Parameter(name=["--log", "-l"], negative="")] = False,
     head: Annotated[bool, cyclopts.Parameter(name=["--head", "-hd"], negative="")] = False,
+    linkouts: Annotated[bool, cyclopts.Parameter(name=["--linkouts", "-lo"], negative="")] = False,
 ) -> None:
     """Build a knowledge graph from a YAML configuration file.
 
@@ -926,10 +938,17 @@ def build_kg(
             Tablassert produces no logs).
         head: Fast output-shape preview: a random sample of up to five rows per section,
             cached under its own ``.head`` store, so it never reuses or replaces a full build.
+        linkouts: Append the EdgeLinkouts per-edge record URL
+            (``https://linkouts.skyelanegoetz.com/edges/{edge_id}``, resolved to the final
+            edge id after dedup) to the graph infores entry (``rig.source_info.infores_id``)
+            in every emitted edge's ``sources`` list, so linkouts.skyelanegoetz.com can
+            deep-link each edge. Sections declaring ``provenance.override`` are untouched
+            (their ``sources`` emission is author-controlled). Cached under its own
+            ``.linkouts`` store, so it never reuses or replaces a full build.
     """
     if qc:
         extras.require("qc", required_by="--qc")
-    run(7 if qc else 6, build_pipeline, graph_configuration_file, release=release, qc=qc, log=log, head=head)
+    run(7 if qc else 6, build_pipeline, graph_configuration_file, release=release, qc=qc, log=log, head=head, linkouts=linkouts)
 
 
 @APP.command(name="validate")

src/tablassert/lib.py

@@ -583,6 +583,13 @@ def inline_supporting_study(lf: pl.LazyFrame, study_id: str, study_name: str | N
     return out.drop(drop)
 
 
+#: Per-edge record URL carried by the graph-infores primary ``sources`` entry when
+#: ``build-kg --linkouts`` is set - the EdgeLinkouts app (linkouts.skyelanegoetz.com)
+#: deep-links each edge by its own id. ``{edge_id}`` stays literal through the polars
+#: build and is resolved by the post-dedup sweep (``_resolve_edge_id_placeholders``).
+EDGE_LINKOUT_URL_TEMPLATE: str = "https://linkouts.skyelanegoetz.com/edges/{edge_id}"
+
+
 def retrieval_sources(
     lf: pl.LazyFrame,
     primary: str,
@@ -1242,6 +1249,7 @@ class Tcode(Section):
     head: bool = Field(False)
     name: str | None = Field(None)
     infores: str | None = Field(None)
+    linkouts: bool = Field(False)
     qc_similarity_threshold: float = Field(0.5)
     qc_fuzzy_ratio_threshold: float = Field(70.0)
     qc_fuzzy_partial_threshold: float = Field(80.0)
@@ -1446,6 +1454,13 @@ def _provenance_ops(self: Self) -> list[Any]:
         # (rig.source_info.infores_id); there is no implicit derivation from the graph
         # name, so a RIG and its edges can never disagree about the source identity.
         primary_knowledge_source: str | None = self.infores
+        # Per-edge record URLs for the primary entry: the section's declared source URLs,
+        # plus the EdgeLinkouts template when --linkouts is set. A section with ANY
+        # provenance.override is author-controlled (DAKP hand-writes the same template
+        # into override.sources), so the flag injects nothing there.
+        primary_urls: list[str] = [str(u) for u in self.source.url]
+        if self.linkouts and override is None:
+            primary_urls.append(EDGE_LINKOUT_URL_TEMPLATE)
         upstream_ids = override.upstream_resource_ids if override else upstream_resource_ids(self.provenance.repo)
         upstream_urls: dict[str, list[str]] | None = (
             {key: [str(u) for u in urls] for key, urls in override.upstream_source_record_urls.items()}
@@ -1489,7 +1504,7 @@ def _provenance_ops(self: Self) -> list[Any]:
             # mapping (override.upstream_source_record_urls) re-homes the record 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))
+            (retrieval_sources, (primary_knowledge_source, upstream_ids, primary_urls, upstream_urls, explicit_sources))
             if primary_knowledge_source
             else None,
             (publications, (publication_values,)) if publication_values else None,

tests/test_cli_help_friendly.py

@@ -103,6 +103,7 @@ def test_build_kg_help_documents_the_build_mode_flags() -> None:
     assert "five rows" in text  # --head's sample size
     assert "verbose per-section logging" in text
     assert "[qc]" in text
+    assert "linkouts.skyelanegoetz.com" in text  # --linkouts' concrete effect
 
 
 def test_validate_help_documents_the_schema_choice() -> None:

tests/test_e2e_smoke.py

@@ -113,6 +113,63 @@ def test_build_pipeline_against_real_redb(tmp_path: Path, monkeypatch: pytest.Mo
     assert "HGNC:6871" in edge_text
 
 
+def test_build_pipeline_with_linkouts_resolves_edge_ids(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, rig_factory: Any) -> None:
+    """SMOKE: ``build_pipeline(linkouts=True)`` emits the resolved EdgeLinkouts record URL.
+
+    Why: the flag's contract spans three stages - the template rides the op list, the
+    deduper assigns ids AFTER subgraphs are written, and the post-dedup sweep substitutes
+    them - so only a real build proves each edge's graph-infores primary entry deep-links
+    at ``https://linkouts.skyelanegoetz.com/edges/<its own id>`` with no placeholder left.
+    Mirrors smoke (i) except for ``linkouts=True``.
+    """
+    monkeypatch.chdir(tmp_path)
+    (tmp_path / ".tablassert" / "store").mkdir(parents=True)
+
+    fullmap: Path = _build_real_redb(tmp_path / "fullmap")
+
+    data: Path = tmp_path / "data.tsv"
+    data.write_text("brca1\tmapk1\nbrca1\tmapk1\n")
+
+    table: Path = tmp_path / "table.yaml"
+    table_config: dict[str, Any] = {
+        "template": {
+            "source": {"kind": "text", "local": str(data), "url": ["https://example.com/data.tsv"], "delimiter": "\t"},
+            "statement": {
+                "subject": {"method": "column", "encoding": "A"},
+                "predicate": "associated_with",
+                "object": {"method": "column", "encoding": "B"},
+            },
+            "provenance": {"repo": "PMC", "publication": "PMC0000000"},
+        }
+    }
+    to_yaml(table, table_config)
+
+    graph: Path = tmp_path / "graph.yaml"
+    graph_config: dict[str, Any] = {
+        "name": "LINKOUTS_KG",
+        "version": "1.0.0",
+        "tables": [str(table)],
+        "fullmap": str(fullmap),
+        "rig": rig_factory(tmp_path, infores_id="infores:linkouts-kg", source_info={"description": "linkouts smoke graph"}),
+    }
+    to_yaml(graph, graph_config)
+
+    build_pipeline(graph, PipelineProgress(total_stages=6), linkouts=True)
+
+    edges: Path = tmp_path / "LINKOUTS_KG_1.0.0.edges.ndjson"
+    assert edges.is_file()
+    records: list[dict[str, Any]] = [json.loads(line) for line in edges.read_text().splitlines() if line.strip()]
+    assert records
+    for record in records:
+        primary: dict[str, Any] = next(s for s in record["sources"] if s["resource_role"] == "primary_knowledge_source")
+        assert primary["resource_id"] == "infores:linkouts-kg"
+        urls: list[str] = primary["source_record_urls"]
+        # The section's own record URL survives and the linkout resolves to THIS edge's id.
+        assert urls == ["https://example.com/data.tsv", f"https://linkouts.skyelanegoetz.com/edges/{record['id']}"]
+    # No unresolved placeholder may survive anywhere in the emitted file.
+    assert "{edge_id}" not in edges.read_text()
+
+
 def test_build_pipeline_over_parquet_source(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, rig_factory: Any) -> None:
     """SMOKE (iii): the same six-stage build reads a ``kind: parquet`` source end to end.
 

tests/test_lib.py

@@ -357,8 +357,83 @@ def test_section_store_path_suffixes_every_build_mode() -> None:
     assert cli._section_store_path("abc123", head=True).name == "abc123.head.parquet"
     assert cli._section_store_path("abc123", release=True).name == "abc123.release.parquet"
     assert cli._section_store_path("abc123", qc=True).name == "abc123.qc.parquet"
+    assert cli._section_store_path("abc123", linkouts=True).name == "abc123.linkouts.parquet"
     # Modes compose instead of overwriting one another's suffix.
-    assert cli._section_store_path("abc123", head=True, release=True, qc=True).name == "abc123.head.release.qc.parquet"
+    assert cli._section_store_path("abc123", head=True, release=True, qc=True, linkouts=True).name == "abc123.head.release.qc.linkouts.parquet"
+
+
+def test_tcode_collect_appends_edge_linkout_when_linkouts(fixtures_path: Path) -> None:
+    """--linkouts appends the EdgeLinkouts template to the primary entry's record URLs.
+
+    Why: the flag's whole contract is that the graph-infores primary ``sources`` entry
+    gains ``https://linkouts.skyelanegoetz.com/edges/{edge_id}`` AFTER the section's own
+    source URLs; the template must reach ``retrieval_sources`` unresolved so the
+    post-dedup sweep can substitute each edge's own id.
+    """
+    data: Any = from_yaml(fixtures_path / "minimal_section.yaml")
+    tcode_model: Tcode = Tcode.model_validate(  # pyright: ignore
+        {
+            **data,
+            "config": fixtures_path / "minimal_section.yaml",
+            "store": Path("/tmp/sectionhash_linkouts.parquet"),
+            "infores": "infores:test-kg",
+            "linkouts": True,
+        }
+    )
+
+    collected: list[tuple[Any, tuple[Any, ...]]] = tcode_model.collect(Path("/tmp/fullmap.redb"))  # pyright: ignore
+    linkout_ops: list[tuple[Any, tuple[Any, ...]]] = [op for op in collected if op[0].__name__ == "retrieval_sources"]
+
+    assert len(linkout_ops) == 1
+    urls: list[str] = linkout_ops[0][1][2]
+    assert urls == ["https://example.com/test.tsv", "https://linkouts.skyelanegoetz.com/edges/{edge_id}"]
+
+
+def test_tcode_collect_omits_edge_linkout_by_default(fixtures_path: Path) -> None:
+    """Without --linkouts the primary entry's record URLs are exactly the section's source URLs.
+
+    Why: the flag is opt-in; default-off emission must be byte-identical to a build
+    without the flag, so the op list must carry no linkout template whatsoever.
+    """
+    data: Any = from_yaml(fixtures_path / "minimal_section.yaml")
+    tcode_model: Tcode = Tcode.model_validate(  # pyright: ignore
+        {**data, "config": fixtures_path / "minimal_section.yaml", "store": Path("/tmp/sectionhash_nolinkouts.parquet"), "infores": "infores:test-kg"}
+    )
+
+    collected: list[tuple[Any, tuple[Any, ...]]] = tcode_model.collect(Path("/tmp/fullmap.redb"))  # pyright: ignore
+    urls: list[str] = next(op for op in collected if op[0].__name__ == "retrieval_sources")[1][2]
+
+    assert urls == ["https://example.com/test.tsv"]
+
+
+def test_tcode_collect_omits_edge_linkout_when_override(fixtures_path: Path) -> None:
+    """--linkouts never injects into a section that authors its own provenance.override.
+
+    Why: override emission is author-controlled (DAKP hand-writes the same EdgeLinkouts
+    template into ``override.sources``); the flag silently adding a second one would
+    duplicate URLs and contradict the section's explicit ``sources`` contract.
+    """
+    data: Any = from_yaml(fixtures_path / "minimal_section.yaml")
+    # `publication` and `override` are mutually exclusive; the override owns provenance.
+    del data["provenance"]["publication"]
+    data["provenance"]["override"] = {"sources": [{"resource_id": "infores:my-source", "resource_role": "primary_knowledge_source"}]}
+    tcode_model: Tcode = Tcode.model_validate(  # pyright: ignore
+        {
+            **data,
+            "config": fixtures_path / "minimal_section.yaml",
+            "store": Path("/tmp/sectionhash_override_lo.parquet"),
+            "infores": "infores:test-kg",
+            "linkouts": True,
+        }
+    )
+
+    collected: list[tuple[Any, tuple[Any, ...]]] = tcode_model.collect(Path("/tmp/fullmap.redb"))  # pyright: ignore
+    retrieval_op: tuple[Any, tuple[Any, ...]] = next(op for op in collected if op[0].__name__ == "retrieval_sources")
+
+    # The explicit template is passed verbatim (args[4]); the section's own source URLs
+    # ride along for the RIG (args[2]) but the linkout template must NOT be appended.
+    assert retrieval_op[1][4] == [{"resource_id": "infores:my-source", "resource_role": "primary_knowledge_source"}]
+    assert "https://linkouts.skyelanegoetz.com/edges/{edge_id}" not in retrieval_op[1][2]
 
 
 def test_tcode_collect_quick_exit_never_serves_another_mode_cached_parquet(monkeypatch: Any, fixtures_path: Path, tmp_path: Path) -> None: