diff --git a/CHANGELOG.md b/CHANGELOG.md
index c8d2626..912df26 100644
--- a/CHANGELOG.md
+++ b/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
 
diff --git a/docs/cli.md b/docs/cli.md
index 8609b16..c284665 100644
--- a/docs/cli.md
+++ b/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
diff --git a/rust/src/ndjson.rs b/rust/src/ndjson.rs
index 6e7e972..2aaa389 100644
--- a/rust/src/ndjson.rs
+++ b/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
diff --git a/src/tablassert/cli.py b/src/tablassert/cli.py
index 0db1473..75ad809 100644
--- a/src/tablassert/cli.py
+++ b/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")
diff --git a/src/tablassert/lib.py b/src/tablassert/lib.py
index 135efc3..4756478 100644
--- a/src/tablassert/lib.py
+++ b/src/tablassert/lib.py
@@ -1592,7 +1592,9 @@ def dedup_stream(p_in: Path, is_edges: bool, domain: str = "TABLASSERT", uuid_fi
         uuid_fields: Optional edge fields that constitute edge identity
             (``Graph.uuid_fields``). ``None`` hashes the whole record.
         on_collision: ``error`` aborts on two different edges deriving one id;
-            ``merge`` folds them into one edge (``Graph.uuid_on_collision``).
+            ``merge`` folds them into one edge (``Graph.uuid_on_collision``). During a merge,
+            distinct non-empty scalar ``original_*`` fields are sorted and rendered as
+            ``A``, ``A|B``, or ``A|B|C``.
 
     Notes:
         Also adds UUIDs to edges. Edges deduplicate on their derived id, so the
@@ -1714,11 +1716,7 @@ def fold_unknown_to_supporting_text(lf: pl.LazyFrame) -> pl.LazyFrame:
 
 
 def _collect_subframes(
-    subgraphs: list[Path],
-    on_phase: Callable[[str], None] | None = None,
-    on_subgraph: Callable[[], None] | None = None,
-    infores_id: str | None = None,
-    no_original: bool = False,
+    subgraphs: list[Path], on_phase: Callable[[str], None] | None = None, on_subgraph: Callable[[], None] | None = None, infores_id: str | None = None
 ) -> tuple[list[pl.LazyFrame], list[pl.LazyFrame]]:
     """Scan and normalize subgraph parquets into node/edge subframes.
 
@@ -1733,8 +1731,6 @@ def _collect_subframes(
         on_subgraph: Optional callback fired once after each subgraph is
             processed, used to tick the progress bar.
         infores_id: Graph-level infores CURIE recorded as node ``provided_by``.
-        no_original: When True, also drop the verbatim ``original_*``
-            source-cell copies from the final edge frames.
 
     Returns:
         Tuple of ``(subnodes, subedges)``: per-section node and edge
@@ -1759,9 +1755,6 @@ def _collect_subframes(
             subnodes.append(partial)
         # Drop internal pre-resolution snapshot columns from final edges.
         lf = lf.drop([c for c in lf.collect_schema().names() if c.endswith("_pre_resolution")])
-        # --no-original: drop the verbatim source-cell copies from final edges too.
-        if no_original:
-            lf = lf.drop([c for c in lf.collect_schema().names() if c.startswith("original_")])
         lf = fold_unknown_to_supporting_text(lf)
         subedges.append(lf)
         if on_subgraph is not None:
@@ -1853,7 +1846,6 @@ def compile_graph(
     section_sources: list[dict[str, object]] | None = None,
     on_phase: Callable[[str], None] | None = None,
     on_subgraph: Callable[[], None] | None = None,
-    no_original: bool = False,
     uuid_fields: list[str] | None = None,
     uuid_domain: str | None = None,
     uuid_on_collision: str = "error",
@@ -1875,8 +1867,6 @@ def compile_graph(
             ``write-edges`` / ``dedup`` / ``rig``), used to drive progress UX.
         on_subgraph: Optional callback fired once per processed subgraph,
             used to tick the progress bar.
-        no_original: When ``True``, omit the verbatim ``original_*``
-            source-cell copies from the final edge NDJSON.
         uuid_fields: Optional edge fields that constitute edge identity
             (``Graph.uuid_fields``). ``None`` hashes the whole edge record, so any
             change to any field re-mints the id.
@@ -1916,7 +1906,7 @@ def compile_graph(
 
     subnodes: list[pl.LazyFrame]
     subedges: list[pl.LazyFrame]
-    subnodes, subedges = _collect_subframes(subgraphs, on_phase, on_subgraph, rig_cfg.source_info.infores_id, no_original)
+    subnodes, subedges = _collect_subframes(subgraphs, on_phase, on_subgraph, rig_cfg.source_info.infores_id)
     _write_ndjson(subnodes, subedges, n, e, name, version, rig_cfg, section_sources, on_phase, domain, uuid_fields, uuid_on_collision)
 
 
diff --git a/tests/test_cover_cli.py b/tests/test_cover_cli.py
index dd3e65e..b5f320c 100644
--- a/tests/test_cover_cli.py
+++ b/tests/test_cover_cli.py
@@ -514,10 +514,6 @@ def parse(argv: list[str]) -> dict[str, Any]:
     # gone, cyclopts rejects the cluster's leading ``-t`` as an unknown option.
     with pytest.raises(UnknownOptionError):
         parse(["build-kg", str(config), "-tc"])
-    # --no-original / -no bind the no_original flag.
-    assert parse(["build-kg", str(config), "--no-original"])["no_original"] is True
-    assert parse(["build-kg", str(config), "-no"])["no_original"] is True
-    assert parse(["build-kg", str(config)]).get("no_original", False) is False
 
 
 def test_build_fullmap_pipeline_reports_download_progress(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
diff --git a/tests/test_lib.py b/tests/test_lib.py
index c28552b..d2ed861 100644
--- a/tests/test_lib.py
+++ b/tests/test_lib.py
@@ -1722,8 +1722,8 @@ def test_compile_graph_keeps_qualifiers_and_publications_on_edges(monkeypatch: A
     assert "PMID:123" not in nodes
 
 
-def test_compile_graph_no_original_drops_original_columns(monkeypatch: Any, tmp_path: Path, rig_factory: Any) -> None:
-    """compile_graph drops ``original_*`` edge columns only when ``no_original`` is set."""
+def test_compile_graph_preserves_original_columns_by_default(monkeypatch: Any, tmp_path: Path, rig_factory: Any) -> None:
+    """compile_graph preserves verbatim ``original_*`` edge columns by default."""
 
     def write_subgraph(p: Path) -> None:
         pl.DataFrame(
@@ -1765,14 +1765,6 @@ def write_subgraph(p: Path) -> None:
     assert '"original_subject":"ALPHA"' in default_edges
     assert '"original_object":"X-RAY"' in default_edges
 
-    stripped_sub: Path = tmp_path / "stripped.parquet"
-    write_subgraph(stripped_sub)
-    lib.compile_graph([stripped_sub], "stripped", "1.0.0", rig_factory(tmp_path, infores_id="infores:no-orig-kg"), no_original=True)
-    stripped_edges: str = (tmp_path / "stripped_1.0.0.edges.ndjson").read_text()
-    assert "original_" not in stripped_edges
-    assert '"subject":"A"' in stripped_edges
-    assert '"object":"X"' in stripped_edges
-
 
 def test_dedup_stream_nodes(tmp_path: Path) -> None:
     """dedup_stream deduplicates and strips null like values from node streams."""
diff --git a/tests/test_rs.py b/tests/test_rs.py
index c703a30..f0dcbd6 100644
--- a/tests/test_rs.py
+++ b/tests/test_rs.py
@@ -233,13 +233,20 @@ def _merge_records(stored: dict[str, Any], incoming: dict[str, Any]) -> int:
     List fields union by canonical bytes and sort ONLY when both sides carry an array
     (a list copied from a later record keeps its order); stored-side duplicates survive,
     incoming-side ones collapse. Scalars are first-wins, each difference counts one
-    conflict; a field only on `incoming` is copied, not a conflict. `id` is untouched.
+    conflict, except non-empty `original_*` strings, which are deduplicated and rendered
+    in sorted `A`, `A|B`, or `A|B|C` form. A field only on `incoming` is copied, not a
+    conflict. `id` is untouched.
     `number_of_cases` is recomputed to the union length of `supporting_case_ids` when the
     merged record carries the carrier list and either side carried a count, and that
     superseded divergence is decremented out of the conflict total (both sides present,
     unequal, not both arrays -- the exact mirror of the rust loop condition).
     """
     conflicts = 0
+    original_values: dict[str, set[str]] = {}
+    for record in (stored, incoming):
+        for key, value in record.items():
+            if key.startswith("original_") and isinstance(value, str):
+                original_values.setdefault(key, set()).update(part for part in value.split("|") if part)
     stored_cases: Any = stored.get("number_of_cases", _MISSING)
     incoming_cases: Any = incoming.get("number_of_cases", _MISSING)
     for key, incoming_value in incoming.items():
@@ -257,8 +264,14 @@ def _merge_records(stored: dict[str, Any], incoming: dict[str, Any]) -> int:
                     seen.append(item_bytes)
                     stored_value.append(item)
             stored_value.sort(key=_canonical_json_bytes)
+        elif key.startswith("original_") and isinstance(stored_value, str) and isinstance(incoming_value, str):
+            pass
         elif stored_value != incoming_value:
             conflicts += 1
+    for key, values in original_values.items():
+        if values:
+            ordered = sorted(values, key=lambda value: value.encode("utf-8"))
+            stored[key] = "|".join(ordered)
     union = stored.get("supporting_case_ids")
     if isinstance(union, list) and (stored_cases is not _MISSING or incoming_cases is not _MISSING):
         if (