feat: config-driven rig.source_files and indented RIG YAML output

docs/configuration/graph.md

@@ -38,6 +38,7 @@ The `rig:` section carries every human-authored RIG fact. Its shape mirrors the
 | `rig.ingest_info` | Object | Yes | Rationale and scope of the ingest (see below) |
 | `rig.target_info` | Object | No | Target-level `future_considerations` and `additional_notes` (edge/node type summaries are always generated) |
 | `rig.ui_explanation` | String | No | Per-edge-type UI explanation **prefix**; the built-in Tablassert explanation is always appended after it |
+| `rig.source_files` | List[String] | No | Upstream source file names/URLs listed verbatim as `source_files` on every generated `edge_type_info` entry; never scraped from edge `source_record_urls` |
 | `rig.provenance_info` | Object | Yes | Contributor statements and provenance artifacts |
 | `rig.supporting_data_source_info` | List[Object] | No | Upstream data sources for data-derived graphs (each needs `infores_id`, `terms_of_use_info`, and `relevant_files`) |
 | `rig.artifact_base_url` | String | Yes | Public URL prefix for the generated KGX artifacts; each `.nodes.ndjson`/`.edges.ndjson` name is appended to build RIG `relevant_files` locations |
@@ -85,7 +86,7 @@ The generator **prepends** two `relevant_files` entries and two `included_conten
 
 Everything under `target_info.edge_type_info` and `target_info.node_type_info` is computed from the **final emitted KGX files** after deduplication:
 
-- **Edge types** (one per observed predicate): subject/object categories resolved from the emitted nodes, list-valued `knowledge_level`/`agent_type`, role-separated `primary_knowledge_sources` / `supporting_data_sources` / `aggregator_knowledge_sources` from each edge's `sources` retrieval provenance, observed `edge_properties`, qualifier shapes (enumerated literal values or identifier prefixes for CURIE-valued qualifiers), and `source_files` taken from the upstream `source_record_urls` (never the output filenames).
+- **Edge types** (one per observed predicate): subject/object categories resolved from the emitted nodes, list-valued `knowledge_level`/`agent_type`, role-separated `primary_knowledge_sources` / `supporting_data_sources` / `aggregator_knowledge_sources` from each edge's `sources` retrieval provenance, observed `edge_properties`, qualifier shapes (enumerated literal values or identifier prefixes for CURIE-valued qualifiers), and `source_files` from the configured `rig.source_files` (never scraped from edge `source_record_urls`).
 - **Node types**: observed categories and the identifier prefixes actually emitted (`source_identifier_types`); categories with prefix-less identifiers get a factual free-text entry.
 - **UI explanation**: `rig.ui_explanation` (when set) followed by the built-in Tablassert explanation; the default text is always present.
 

src/tablassert/ingests.py

@@ -58,15 +58,28 @@ def from_yaml(p: Path) -> object:
         return yaml.load(f, Loader=CSafeLoader)
 
 
+class _IndentedSafeDumper(yaml.SafeDumper):
+    """SafeDumper that indents block sequences under their parent mapping key."""
+
+    def increase_indent(self, flow: bool = False, indentless: bool = False) -> Any:
+        return super().increase_indent(flow, False)
+
+
 def to_yaml(p: Path, data: object) -> None:
     """Write dict-like data to YAML preserving declared key order.
 
+    Block sequences are indented under their parent mapping key (PyYAML's
+    default indentless style makes nested RIG entries hard to scan), and the
+    line width is relaxed so long prose fields are not chopped mid-sentence.
+
     Args:
         p: Destination path.
         data: Object to serialize.
     """
     with p.open("w") as f:
-        yaml.safe_dump(data, f, sort_keys=False)
+        # `yaml.dump` with an explicit SafeDumper subclass: `safe_dump` accepts no
+        # Dumper argument, and the subclass still refuses unsafe object construction.
+        yaml.dump(data, f, Dumper=_IndentedSafeDumper, sort_keys=False, default_flow_style=False, allow_unicode=True, width=120)
 
 
 def to_sections(instructions: dict[str, Any], table: Path) -> list[list[dict[str, Any]]]:

src/tablassert/models.py

@@ -1067,6 +1067,13 @@ class RIGConfig(TablaBase):
             "after it, so the generated text is this value followed by the default provenance explanation."
         ),
     )
+    source_files: list[str] | None = Field(
+        None,
+        description=(
+            "Upstream source file names/URLs listed as `source_files` on every generated `edge_type_info` entry. "
+            "Authoritative config value; never scraped from edge `source_record_urls`."
+        ),
+    )
     provenance_info: RIGProvenanceInfo = Field(..., description="Who contributed to the ingest and how.")
     artifact_base_url: str = Field(
         ...,

src/tablassert/rig.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import json
-from pathlib import Path, PurePosixPath
+from pathlib import Path
 from typing import Any
 
 from tablassert.biolink import AgentTypes, KnowledgeLevels
@@ -252,8 +252,7 @@ def rig_edge_type_info(
 
     One entry per observed predicate, aggregating the subject/object categories
     resolved from the emitted node file, KL/AT values, qualifier shapes, edge
-    properties, role-separated knowledge sources, and the upstream source files
-    recorded on each edge's retrieval provenance. Grouping by predicate matches
+    properties, and role-separated knowledge sources. Grouping by predicate matches
     the upstream RIG convention that one edge type may list several subject or
     object categories without implying a full cross-product.
 
@@ -291,7 +290,6 @@ def rig_edge_type_info(
                     "aggregator": set(),
                     "properties": set(),
                     "qualifiers": {},
-                    "files": set(),
                 },
             )
             group["subjects"].update(categories.get(str(edge.get("subject") or ""), []))
@@ -320,9 +318,6 @@ def rig_edge_type_info(
                         group["supporting"].add(resource)
                     elif role == "aggregator_knowledge_source":
                         group["aggregator"].add(resource)
-                for url in clean_values(as_list(source_entry.get("source_record_urls"))):
-                    source_file: str = PurePosixPath(url).name
-                    group["files"].add(source_file or url)
 
     info: list[dict[str, object]] = []
     for predicate in sorted(groups):
@@ -344,8 +339,6 @@ def rig_edge_type_info(
         if group["properties"]:
             entry["edge_properties"] = sorted(group["properties"])
         entry["ui_explanation"] = ui_explanation
-        if group["files"]:
-            entry["source_files"] = sorted(group["files"])
         info.append(entry)
     return info, sorted(fields), count, predicates
 
@@ -670,6 +663,11 @@ def compile_rig(name: str, version: str, rig: RIGConfig, section_sources: list[d
     node_fields: list[str] = sorted({key for node in node_rows for key, value in node.items() if value is not None})
     ui_explanation: str = compose_ui_explanation(rig.ui_explanation)
     edge_type_info, edge_fields, edge_count, observed_predicates = rig_edge_type_info(edges_path, categories, ui_explanation)
+    # `source_files` is an authored config fact (rig.source_files), never scraped
+    # from edge `source_record_urls`; when configured it applies to every edge type.
+    if rig.source_files:
+        for entry in edge_type_info:
+            entry["source_files"] = sorted(set(rig.source_files))
     observed_node_categories: set[str] = {str(category) for entry in node_type_info for category in as_list(entry.get("node_category"))}
 
     document: dict[str, object] = build_rig_document(

tests/test_lib.py

@@ -1422,7 +1422,13 @@ def test_compile_graph_emits_ndjson(monkeypatch: Any, tmp_path: Path, rig_factor
             "p_value": ["1.0000e-08", "5.0000e-02"],
         }
     ).write_parquet(sub)
-    rig = rig_factory(tmp_path, infores_id="infores:smoke", source_info={"description": "Smoke graph"}, ui_explanation="Custom UI explanation.")
+    rig = rig_factory(
+        tmp_path,
+        infores_id="infores:smoke",
+        source_info={"description": "Smoke graph"},
+        ui_explanation="Custom UI explanation.",
+        source_files=["table1.xlsx"],
+    )
     lib.compile_graph([sub], "smoke", "1.0.0", rig)
     edges: list[str] = (tmp_path / "smoke_1.0.0.edges.ndjson").read_text().strip().splitlines()
     nodes: list[str] = (tmp_path / "smoke_1.0.0.nodes.ndjson").read_text().strip().splitlines()
@@ -1454,7 +1460,7 @@ def test_compile_graph_emits_ndjson(monkeypatch: Any, tmp_path: Path, rig_factor
     assert all(entry["included_records"] for entry in included)
 
     # Edge summaries come from the FINAL graph: role-separated sources, list KL/AT,
-    # observed properties, upstream source files (not the output NDJSON names).
+    # observed properties; source files come from the configured rig.source_files.
     edge_type: dict[str, Any] = rig_doc["target_info"]["edge_type_info"][0]  # pyright: ignore
     assert edge_type["subject_categories"] == ["biolink:gene"]
     assert edge_type["predicates"] == ["biolink:related_to"]
@@ -3165,7 +3171,9 @@ def test_build_pipeline_e2e_smoke_with_monkeypatched_fullmap(monkeypatch: Any, t
             "version": "0.1.0",
             "tables": [str(table_path)],
             "fullmap": str(tmp_path / "fullmap.redb"),
-            "rig": rig_factory(tmp_path, infores_id="infores:pipeline-kg", source_info={"description": "Pipeline smoke graph."}),
+            "rig": rig_factory(
+                tmp_path, infores_id="infores:pipeline-kg", source_info={"description": "Pipeline smoke graph."}, source_files=["pipeline_table.tsv"]
+            ),
         },
     )
     progress: DummyProgress = DummyProgress()
@@ -3203,7 +3211,7 @@ def test_build_pipeline_e2e_smoke_with_monkeypatched_fullmap(monkeypatch: Any, t
     # Role separation: the graph infores is the primary source; PMC is supporting data.
     assert edge_type["primary_knowledge_sources"] == ["infores:pipeline-kg"]
     assert edge_type["supporting_data_sources"] == ["infores:pubmed-central"]
-    # Source files name the upstream table URL, not the generated NDJSON outputs.
+    # Source files come from the configured rig.source_files, not scraped from edges.
     assert edge_type["source_files"] == ["pipeline_table.tsv"]
 
     # The gate: every emitted record must construct as its own Biolink class. Without

tests/test_rig.py

@@ -18,7 +18,7 @@
 from tablassert import lib
 from tablassert.errors import TablassertError
 from tablassert.models import DEFAULT_RIG_UI_EXPLANATION, RIGConfig
-from tablassert.rig import audit_rig, build_rig_document, compose_ui_explanation, rig_edge_type_info, rig_node_type_info
+from tablassert.rig import audit_rig, build_rig_document, compile_rig, compose_ui_explanation, rig_edge_type_info, rig_node_type_info
 
 
 def _source_entry(primary: str, url: str, upstream: list[str] | None = None) -> list[dict[str, Any]]:
@@ -75,8 +75,8 @@ def test_rig_node_type_info_reports_emitted_prefixes_and_factual_fallback() -> N
     }
 
 
-def test_rig_edge_type_info_separates_roles_properties_qualifiers_and_files(tmp_path: Path) -> None:
-    """Edge summaries carry role-separated sources, observed properties, qualifier shapes, upstream file names."""
+def test_rig_edge_type_info_separates_roles_properties_and_qualifiers(tmp_path: Path) -> None:
+    """Edge summaries carry role-separated sources, observed properties, and qualifier shapes."""
     edges: Path = _write_edges(
         tmp_path,
         [
@@ -123,8 +123,9 @@ def test_rig_edge_type_info_separates_roles_properties_qualifiers_and_files(tmp_
     assert entry["aggregator_knowledge_sources"] == ["infores:aggregator"]
     assert entry["edge_properties"] == ["biolink:p_value", "biolink:publications"]
     assert entry["ui_explanation"] == "UI TEXT"
-    # Source files come from the upstream source_record_urls, never the output NDJSON names.
-    assert entry["source_files"] == ["table.tsv", "table1.xlsx"]
+    # Source files are an authored config fact (rig.source_files), never scraped
+    # from edge source_record_urls.
+    assert "source_files" not in entry
     # Literal-valued qualifiers enumerate observed values under their biolink property.
     assert entry["qualifiers"] == [{"property": "biolink:object_aspect_qualifier", "value_enumeration": ["decreased", "increased"]}]
 
@@ -316,7 +317,7 @@ def test_compile_graph_multi_source_keeps_configured_relevant_files(tmp_path: Pa
         }
     ).write_parquet(sub)
 
-    rig_dict: dict[str, Any] = rig_factory(tmp_path, infores_id="infores:multi-kg")
+    rig_dict: dict[str, Any] = rig_factory(tmp_path, infores_id="infores:multi-kg", source_files=["table1.xlsx"])
     rig_dict["ingest_info"]["relevant_files"] = [
         {"file_name": "table1.xlsx", "location": "https://pmc.ncbi.nlm.nih.gov/bin/table1.xlsx", "description": "Upstream PMC table."}
     ]
@@ -327,11 +328,41 @@ def test_compile_graph_multi_source_keeps_configured_relevant_files(tmp_path: Pa
     names: list[str] = [entry["file_name"] for entry in document["ingest_info"]["relevant_files"]]
     assert names == ["multi_1.nodes.ndjson", "multi_1.edges.ndjson", "table1.xlsx"]
     edge_type: dict[str, Any] = document["target_info"]["edge_type_info"][0]
+    # source_files come from the configured rig.source_files, never scraped from
+    # the edge's source_record_urls.
     assert edge_type["source_files"] == ["table1.xlsx"]
     # The default explanation is always present even without a configured prefix.
     assert edge_type["ui_explanation"] == DEFAULT_RIG_UI_EXPLANATION
 
 
+def test_compile_rig_applies_configured_source_files_to_every_edge_type(tmp_path: Path, rig_factory: Any) -> None:
+    """rig.source_files is listed verbatim (sorted) on every edge type entry."""
+    nodes_path: Path = tmp_path / "kg_1.nodes.ndjson"
+    nodes_path.write_text('{"id":"HGNC:1","category":["biolink:Gene"]}\n{"id":"MONDO:1","category":["biolink:Disease"]}\n')
+    edges_path: Path = _write_edges(tmp_path, [_edge("HGNC:1", "MONDO:1")]).rename(tmp_path / "kg_1.edges.ndjson")
+
+    rig = RIGConfig.model_validate(rig_factory(tmp_path, source_files=["b.tsv", "a.tsv"]))
+    compile_rig("kg", "1", rig, None, nodes_path, edges_path)
+
+    document: dict[str, Any] = yaml.safe_load((tmp_path / "kg_1.RIG.yaml").read_text())
+    for entry in document["target_info"]["edge_type_info"]:
+        assert entry["source_files"] == ["a.tsv", "b.tsv"]
+
+
+def test_compile_rig_writes_indented_block_sequences(tmp_path: Path, rig_factory: Any) -> None:
+    """Generated RIG YAML indents block sequences under their parent key."""
+    nodes_path: Path = tmp_path / "kg_1.nodes.ndjson"
+    nodes_path.write_text('{"id":"HGNC:1","category":["biolink:Gene"]}\n{"id":"MONDO:1","category":["biolink:Disease"]}\n')
+    edges_path: Path = _write_edges(tmp_path, [_edge("HGNC:1", "MONDO:1")]).rename(tmp_path / "kg_1.edges.ndjson")
+
+    rig = RIGConfig.model_validate(rig_factory(tmp_path))
+    compile_rig("kg", "1", rig, None, nodes_path, edges_path)
+
+    text: str = (tmp_path / "kg_1.RIG.yaml").read_text()
+    assert "relevant_files:\n    - file_name:" in text
+    assert "\n- file_name:" not in text  # PyYAML's default indentless style
+
+
 def test_compile_graph_rejects_stale_configured_relevant_files(tmp_path: Path, rig_factory: Any) -> None:
     """A configured relevant file that matches no table source is stale documentation and fails."""
     sub: Path = tmp_path / "sub.parquet"