feat: retain and merge original source values

CHANGELOG.md

@@ -4,6 +4,9 @@ All notable changes to this project are documented in this file.
 
 ## Unreleased
 
+### Changed
+- **Original source values are always retained and collision-merged deterministically.** `original_*` edge fields now combine distinct values using sorted `A`, `A|B`, or `A|B|C` formatting; the deprecated `build-kg --no-original` option and API parameters were removed.
+
 ### Documentation
 - **`README.md` no longer presents a scalar `source.url` where the live schema requires a list, and its documentation index now includes the shipped Fullmap, Development, and Changelog pages.** The quick-start configuration uses the valid list form, and the navigation links now cover the corresponding site surfaces.
 - **`docs/installation.md` no longer omits the `distill` extra or blur the distinction between recording and exporting distillation data.** The extras table and preflight guidance now document `distill`, while `agent --distill` is identified as zero-additional-dependency recording and `distill-export` as the step requiring the extra.
@@ -132,7 +135,6 @@ All notable changes to this project are documented in this file.
 ### Added
 - **Release mode drops `applied_to_treat` edges with a `number_of_cases` below 25.** The release pipeline already shed non-significant rows and zero effect sizes, but a treatment edge backed by a single-digit cohort still shipped beside them, because nothing checked how many cases stood behind the association. The new `drop_low_number_of_cases` filter closes that gap: it is wired into `Tcode` op construction gated on `--release` *and* `statement.predicate == "applied_to_treat"`, so other predicates never see it; null case counts are kept (no count was detected, not a small one); sections without a `number_of_cases` column are untouched; and like the other release filters it runs in the `significance` phase before `resolve_batch`, so dropped rows never pay for entity resolution. Non-release builds are unchanged. ([#123](https://github.com/SkyeAv/Tablassert/pull/123))
 
-- **`build-kg --no-original` omits the verbatim source-cell copies from final edges.** The final edge NDJSON carried `original_subject`, `original_object`, and any other `original_*` columns — faithful copies of the raw source cells — with no way to ship a graph without them. The new `--no-original` / `-no` flag drops every `original_*` field from the emitted edges: the columns are still produced mid-pipeline, because entity resolution and the `--qc` audits read them, and the filter sits in `_collect_subframes` beside the existing `*_pre_resolution` drop, ahead of dedup and hashing, so builds stay deterministic. The default is unchanged — full-fidelity output remains standard — and node output is unaffected, since nodes never carry `original_*` fields. ([#124](https://github.com/SkyeAv/Tablassert/pull/124))
 
 ## 16.0.0 - 2026-08-25
 

docs/cli.md

@@ -172,7 +172,6 @@ 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), and no stray whitespace (verbatim `original_*` fields excepted from the whitespace check, since they are faithful source copies) and fails the build (non-zero exit) on any violation |
 | `--log`, `-l` | Flag | No | `False` | Enable verbose per-section logging |
 | `--head`, `-hd` | Flag | No | `False` | Fast output-shape preview: ≤5 random rows/section, cached to `.head.parquet`, never clobbers a full build |
-| `--no-original`, `-no` | Flag | No | `False` | Omit the verbatim source-cell copies (`original_subject`, `original_object`, and any other `original_*` fields) from the final edge NDJSON |
 
 ```bash
 tablassert build-kg graph.yaml --qc --log

rust/src/ndjson.rs

@@ -208,6 +208,34 @@ struct MergeIndex {
     scalar_conflicts: u64,
 }
 
+type OriginalValues = FxHashMap<String, FxHashSet<String>>;
+
+fn collect_original_values(value: &Value) -> OriginalValues {
+    let mut values: OriginalValues = FxHashMap::default();
+    let Some(map) = value.as_object() else {
+        return values;
+    };
+    for (key, value) in map {
+        if key.starts_with("original_") {
+            if let Value::String(text) = value {
+                for part in text.split('|').filter(|part| !part.is_empty()) {
+                    values
+                        .entry(key.clone())
+                        .or_default()
+                        .insert(part.to_string());
+                }
+            }
+        }
+    }
+    values
+}
+
+fn format_original_values(values: &FxHashSet<String>) -> String {
+    let mut sorted: Vec<&str> = values.iter().map(String::as_str).collect();
+    sorted.sort_unstable_by(|left, right| left.as_bytes().cmp(right.as_bytes()));
+    sorted.join("|")
+}
+
 /// The canonical bytes of one list item, shared by every structure that needs them.
 ///
 /// ONE heap allocation per distinct item instead of one per consumer: `Rc<T>`'s `Hash`,
@@ -281,6 +309,8 @@ struct MergedRecord {
     hashes: FxHashSet<u64>,
     /// Union state for every field whose current value is an array.
     lists: FxHashMap<String, ListState>,
+    /// Distinct non-empty scalar source values for each `original_*` field.
+    original_values: OriginalValues,
 }
 
 impl MergedRecord {
@@ -298,6 +328,7 @@ impl MergedRecord {
         let mut hashes: FxHashSet<u64> = FxHashSet::default();
         hashes.insert(content);
         Ok(Self {
+            original_values: collect_original_values(&value),
             value,
             hashes,
             lists,
@@ -312,6 +343,11 @@ impl MergedRecord {
         let Some(map) = self.value.as_object_mut() else {
             return Err(runtime_error("expected JSON object"));
         };
+        for (key, values) in &self.original_values {
+            if !values.is_empty() {
+                map.insert(key.clone(), Value::String(format_original_values(values)));
+            }
+        }
         for (key, state) in &mut self.lists {
             if !state.unioned {
                 continue;
@@ -362,7 +398,9 @@ impl MergedRecord {
 ///   checked against the field's `ListState` set in O(1); stored items are never
 ///   re-canonicalized, and the sort by canonical bytes is deferred to write-out
 ///   (`MergedRecord::finish`), where it runs only for fields that saw a real union;
-/// - scalar fields: first-wins on conflict, counted;
+/// - scalar fields: first-wins on conflict, counted, except `original_*` fields;
+/// - `original_*` fields collect distinct non-empty strings and render sorted values as
+///   `A`, `A|B`, or `A|B|C`;
 /// - fields only on `incoming`: copied over (only a conflict when both sides disagree);
 /// - `id` is never touched: both sides carry the same one by construction.
 ///
@@ -399,6 +437,15 @@ fn merge_records(stored: &mut MergedRecord, incoming: &Value) -> PyResult<u64> {
                     stored
                         .lists
                         .insert(key.clone(), ListState::from_items(items)?);
+                } else if key.starts_with("original_") {
+                    if let Value::String(text) = incoming_value {
+                        let values = stored.original_values.entry(key.clone()).or_default();
+                        values.extend(
+                            text.split('|')
+                                .filter(|part| !part.is_empty())
+                                .map(str::to_owned),
+                        );
+                    }
                 }
                 stored_map.insert(key.clone(), incoming_value.clone());
             }
@@ -422,6 +469,26 @@ fn merge_records(stored: &mut MergedRecord, incoming: &Value) -> PyResult<u64> {
                             stored_items.push(item.clone());
                         }
                     }
+                } else if key.starts_with("original_") {
+                    if let (Value::String(stored_text), Value::String(incoming_text)) =
+                        (&*stored_value, incoming_value)
+                    {
+                        let values = stored.original_values.entry(key.clone()).or_default();
+                        values.extend(
+                            stored_text
+                                .split('|')
+                                .filter(|part| !part.is_empty())
+                                .map(str::to_owned),
+                        );
+                        values.extend(
+                            incoming_text
+                                .split('|')
+                                .filter(|part| !part.is_empty())
+                                .map(str::to_owned),
+                        );
+                    } else if stored_value != incoming_value {
+                        conflicts += 1;
+                    }
                 } else if stored_value != incoming_value {
                     conflicts += 1;
                 }
@@ -529,6 +596,10 @@ fn merge_records_reference(stored: &mut Value, incoming: &Value) -> PyResult<u64
         return Err(runtime_error("expected JSON object"));
     };
     let mut conflicts: u64 = 0;
+    let mut original_values: OriginalValues = collect_original_values(stored);
+    for (key, values) in collect_original_values(incoming) {
+        original_values.entry(key).or_default().extend(values);
+    }
     // Read both counts BEFORE the fold: the fold may copy incoming's over a stored side
     // that lacks it, and the recompute rule below needs to know each side contributed one.
     let stored_cases: Option<Value> = stored.get("number_of_cases").cloned();
@@ -565,12 +636,21 @@ fn merge_records_reference(stored: &mut Value, incoming: &Value) -> PyResult<u64
                     }
                     keyed.sort_by(|left, right| left.0.cmp(&right.0));
                     stored_items.extend(keyed.into_iter().map(|(_, item)| item));
+                } else if key.starts_with("original_") {
+                    // Aggregated and rendered after this fold, once all values are known.
                 } else if stored_value != incoming_value {
                     conflicts += 1;
                 }
             }
         }
     }
+    if let Some(map) = stored.as_object_mut() {
+        for (key, values) in &original_values {
+            if !values.is_empty() {
+                map.insert(key.clone(), Value::String(format_original_values(values)));
+            }
+        }
+    }
     // WHY: exact-unique `number_of_cases` semantics (see the docstring). Guarded on the
     // merged record actually carrying the ID list: a one-sided carrier is fine (the union
     // is just that side's list), while a carrier-less merge never recomputes.
@@ -759,8 +839,9 @@ pub fn dedup_ndjson(
 
 #[cfg(test)]
 mod tests {
-    use super::{dedup_ndjson, differing_keys, record_if_new};
+    use super::{dedup_ndjson, differing_keys, format_original_values, record_if_new};
     use rustc_hash::FxHashMap;
+    use rustc_hash::FxHashSet;
     use serde_json::Value;
     use std::fs;
     use tempfile::tempdir;
@@ -1465,6 +1546,82 @@ mod tests {
         assert_eq!(sources[1]["resource_id"], serde_json::json!("infores:b"));
     }
 
+    #[test]
+    fn format_original_values_uses_sorted_pipe_joining() {
+        let empty: FxHashSet<String> = FxHashSet::default();
+        assert_eq!(format_original_values(&empty), "");
+
+        let one = FxHashSet::from_iter([String::from("only")]);
+        assert_eq!(format_original_values(&one), "only");
+
+        let two = FxHashSet::from_iter([String::from("z"), String::from("a")]);
+        assert_eq!(format_original_values(&two), "a|z");
+
+        let three = FxHashSet::from_iter([String::from("z"), String::from("a"), String::from("m")]);
+        assert_eq!(format_original_values(&three), "a|m|z");
+    }
+
+    #[test]
+    fn merge_mode_aggregates_original_scalars_across_all_records() {
+        let dir = tempdir().expect("tempdir");
+        let (input, output) = write_edges(
+            dir.path(),
+            concat!(
+                "{\"subject\":\"A\",\"predicate\":\"r\",\"object\":\"B\",\"original_subject\":\"z\",\"original_object\":\"\"}\n",
+                "{\"subject\":\"A\",\"predicate\":\"r\",\"object\":\"B\",\"original_subject\":\"a\",\"original_object\":\"x\"}\n",
+                "{\"subject\":\"A\",\"predicate\":\"r\",\"object\":\"B\",\"original_subject\":\"m\",\"original_object\":\"y\"}\n",
+                "{\"subject\":\"A\",\"predicate\":\"r\",\"object\":\"B\",\"original_subject\":\"a\",\"original_object\":\"x\"}\n"
+            ),
+        );
+        let (merged, conflicts) = dedup_ndjson(
+            input,
+            output.clone(),
+            true,
+            None,
+            spo_fields(),
+            Some("merge".to_string()),
+        )
+        .expect("merge dedup");
+
+        assert_eq!((merged, conflicts), (2, 0));
+        let edge = merged_edge(&output);
+        assert_eq!(edge["original_subject"], serde_json::json!("a|m|z"));
+        assert_eq!(edge["original_object"], serde_json::json!("x|y"));
+    }
+
+    #[test]
+    fn merge_mode_ignores_empty_originals_and_is_order_independent() {
+        let write = |dir: &std::path::Path, rows: &str| {
+            let (input, output) = write_edges(dir, rows);
+            dedup_ndjson(
+                input,
+                output.clone(),
+                true,
+                None,
+                spo_fields(),
+                Some("merge".to_string()),
+            )
+            .expect("merge dedup");
+            merged_edge(&output)["original_subject"].clone()
+        };
+        let first = write(
+            tempdir().expect("tempdir").path(),
+            concat!(
+                "{\"subject\":\"A\",\"predicate\":\"r\",\"object\":\"B\",\"original_subject\":\"\"}\n",
+                "{\"subject\":\"A\",\"predicate\":\"r\",\"object\":\"B\",\"original_subject\":\"only\"}\n"
+            ),
+        );
+        let second = write(
+            tempdir().expect("tempdir").path(),
+            concat!(
+                "{\"subject\":\"A\",\"predicate\":\"r\",\"object\":\"B\",\"original_subject\":\"only\"}\n",
+                "{\"subject\":\"A\",\"predicate\":\"r\",\"object\":\"B\",\"original_subject\":\"\"}\n"
+            ),
+        );
+        assert_eq!(first, serde_json::json!("only"));
+        assert_eq!(first, second);
+    }
+
     #[test]
     fn merge_mode_keeps_fields_only_the_second_record_carries() {
         // WHY: first-wins arbitrates CONFLICTS; a field absent from the first record is

src/tablassert/cli.py

@@ -177,17 +177,11 @@ def _load_graph(configuration_file: Path) -> Graph:
 
 
 def build_pipeline(
-    configuration_file: Path,
-    progress: PipelineProgress,
-    release: bool = False,
-    qc: bool = False,
-    log: bool = False,
-    head: bool = False,
-    no_original: bool = False,
+    configuration_file: Path, progress: PipelineProgress, release: bool = False, qc: bool = False, log: bool = False, head: 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, no_original=no_original)
+    build_graph_pipeline(graph, configuration_file, progress, release=release, qc=qc, log=log, head=head)
 
 
 def build_graph_pipeline(
@@ -198,7 +192,6 @@ def build_graph_pipeline(
     qc: bool = False,
     log: bool = False,
     head: bool = False,
-    no_original: bool = False,
     audit_sources: bool = True,
 ) -> None:
     """Build a validated :class:`Graph` without loading another graph YAML.
@@ -216,8 +209,6 @@ 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.
-        no_original: When ``True``, omit the verbatim ``original_*`` source-cell
-            copies from the final edge NDJSON.
     """
     from tablassert.fullmap import fullmap_db_path
     from tablassert.lib import Tcode, compile_graph, compile_subgraph
@@ -324,7 +315,6 @@ def build_graph_pipeline(
         section_sources if audit_sources else None,
         on_phase=sub_step,
         on_subgraph=advance,
-        no_original=no_original,
         uuid_fields=g.uuid_fields,
         uuid_domain=g.uuid_namespace,
         uuid_on_collision=g.uuid_on_collision,
@@ -731,16 +721,13 @@ 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,
-    no_original: Annotated[bool, cyclopts.Parameter(name=["--no-original", "-no"], negative="")] = False,
 ) -> None:
     """Build a knowledge graph from a YAML configuration file.
 
     The positional config is a Graph YAML that orchestrates one or more table
     configs into a single knowledge-graph build.
 
-    ``--no-original`` omits the verbatim source-cell copies (``original_subject``,
-    ``original_object``, and any other ``original_*`` fields) from the final edge
-    NDJSON. ``--qc`` requires the ``[qc]`` extra (``pip install
+    ``--qc`` requires the ``[qc]`` extra (``pip install
     "tablassert[qc]"``); it is checked before the build starts, because the audit stage
     runs LAST and a missing extra would otherwise surface only after entity resolution
     has finished. It also runs a final study stage that asserts over the emitted NDJSON
@@ -751,7 +738,7 @@ def build_kg(
     """
     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, no_original=no_original)
+    run(7 if qc else 6, build_pipeline, graph_configuration_file, release=release, qc=qc, log=log, head=head)
 
 
 @APP.command(name="validate")