merge: #703 — elements keep a confirmed_by cross-anchor (cpt confirms fr); review rows carry their source (refs #703)

notebooks/code_families.py

@@ -343,12 +343,12 @@ def _(code, mo, not_built, pl, q):
 def _(code, mo, not_built, q):
     # ── 2. What the extractor wrote ──
     _els = q(
-        "SELECT type, value, detail, source, item_key, p_id, page FROM pfs.code_element "
-        "WHERE code = ? ORDER BY type, value",
+        "SELECT type, value, detail, source, confirmed_by, item_key, p_id, page "
+        "FROM pfs.code_element WHERE code = ? ORDER BY type, value",
         (code,),
     )
     _rev = q(
-        "SELECT text, proposed_value, item_key, p_id FROM pfs.code_element_review "
+        "SELECT text, proposed_value, source, item_key, p_id FROM pfs.code_element_review "
         "WHERE code = ? ORDER BY p_id",
         (code,),
     )
@@ -367,7 +367,9 @@ def _(code, mo, not_built, q):
                     "the closed list, or `none`**; whatever neither pass can place is queued for "
                     "human review rather than guessed. Nothing enters `pfs.code_element` unless "
                     "it is a member of the closed vocabulary, and every row keeps the exact "
-                    "Federal Register paragraph it came from."
+                    "Federal Register paragraph it came from. When a second source (the CPT "
+                    "manual, HCPCS, RVU) independently transcribes the same element, that source "
+                    "isn't a second row — it's recorded in `confirmed_by` on the winning row."
                 ),
                 mo.ui.table(
                     _els, label=f"pfs.code_element — {code} ({_els.height} rows)"

src/cli/pfs.py

@@ -28,10 +28,12 @@ from pfs.codetables import (
     is_missing_table_error,
     read_all_elements,
     read_all_events,
+    read_all_reviews,
     read_cpt_codes,
     read_cpt_instructions,
     read_cpt_sections,
     read_guidance,
+    read_reviews,
     write_elements,
     write_events,
     write_families,
@@ -113,7 +115,9 @@ def _run_elements(
         x = extract_code(store, con, c, classify=classify)
         if write:
             write_elements(con, c, x.rows, x.reviews)
-        typer.echo(f"{c}: {len(x.rows)} elements, {len(x.reviews)} for review")
+        confirmed = sum(1 for r in x.rows if r.confirmed_by)
+        note = f", {confirmed} confirmed by a second source" if confirmed else ""
+        typer.echo(f"{c}: {len(x.rows)} elements, {len(x.reviews)} for review{note}")
 
 
 def _codes_for(
@@ -642,17 +646,13 @@ def reaction(
 def review(code: str = typer.Option("", "--code")) -> None:
     """Element lines the classifier could not place."""
     con = _read()
-    sql = (
-        "SELECT code, text, proposed_value, item_key, p_id FROM pfs.code_element_review"
-    )
-    params: list[Any] = []
-    if code:
-        sql += " WHERE code = ?"
-        params.append(code.upper())
-    for c, text, proposed, key, p_id in con.execute(
-        sql + " ORDER BY code, p_id", params
-    ).fetchall():
-        typer.echo(f"{c}  {key} ¶{p_id}  [{proposed or '?'}]  {text[:120]}")
+    rows = read_reviews(con, code.upper()) if code else read_all_reviews(con)
+    for r in rows:
+        src = f"  ({r.source})" if r.source else ""
+        typer.echo(
+            f"{r.code}  {r.item_key} ¶{r.p_id}{src}  [{r.proposed_value or '?'}]  "
+            f"{r.text[:120]}"
+        )
 
 
 def _print_cpt_counts(counts: dict[int, dict[str, int]]) -> None:

src/pfs/codetables.py

@@ -47,6 +47,16 @@ class ElementRow:
     p_id: int
     page: int
     source: str  # "fr" | "cpt" | "hcpcs" | "rvu"
+    #: Comma-separated ``source:item_key`` tokens (e.g. ``"cpt:GQGTPGYV"``)
+    #: — one per additional source whose own independent transcription
+    #: yielded the same ``(type, value, detail)`` as this row (#703). The
+    #: winning row (FR over CPT over HCPCS/RVU, unchanged) is the one kept;
+    #: this column is the cross-check the dedupe used to discard entirely.
+    #: ``""`` when no second source confirmed this element. Last field,
+    #: defaulted, so existing callers that build an ``ElementRow`` without
+    #: it keep working (Ruling C1 pattern), and a replica read before this
+    #: column existed tolerates the short tuple the same way.
+    confirmed_by: str = ""
 
 
 @dataclass(frozen=True)
@@ -57,6 +67,11 @@ class ReviewRow:
     proposed_value: str
     item_key: str
     p_id: int
+    #: "fr" | "cpt" | "hcpcs" | "rvu" — which source's pass produced this
+    #: unplaced line (#703). Last field, defaulted, so existing callers
+    #: that build a ``ReviewRow`` without it keep working, and a replica
+    #: read before this column existed tolerates the short tuple.
+    source: str = ""
 
 
 @dataclass(frozen=True)
@@ -233,10 +248,11 @@ _DDL = """
 CREATE SCHEMA IF NOT EXISTS pfs;
 CREATE TABLE IF NOT EXISTS pfs.code_element (
     code VARCHAR, year INTEGER, type VARCHAR, value VARCHAR, detail VARCHAR,
-    text VARCHAR, item_key VARCHAR, p_id INTEGER, page INTEGER, source VARCHAR);
+    text VARCHAR, item_key VARCHAR, p_id INTEGER, page INTEGER, source VARCHAR,
+    confirmed_by VARCHAR);
 CREATE TABLE IF NOT EXISTS pfs.code_element_review (
     code VARCHAR, text VARCHAR, proposed_type VARCHAR, proposed_value VARCHAR,
-    item_key VARCHAR, p_id INTEGER);
+    item_key VARCHAR, p_id INTEGER, source VARCHAR);
 CREATE TABLE IF NOT EXISTS pfs.code_event (
     code VARCHAR, year INTEGER, kind VARCHAR, from_codes VARCHAR, to_codes VARCHAR,
     item_key VARCHAR, p_id INTEGER, page INTEGER, source VARCHAR, anchored BOOLEAN, note VARCHAR);
@@ -295,6 +311,15 @@ def ensure_tables(con: Any) -> None:
     # `note` existed) needs the column added — CREATE TABLE IF NOT EXISTS
     # above is a no-op once the table already exists.
     con.execute("ALTER TABLE pfs.code_family ADD COLUMN IF NOT EXISTS note VARCHAR")
+    # #703, same pattern: pre-existing pfs.code_element/code_element_review
+    # tables (created before confirmed_by/source existed) need the columns
+    # added in place.
+    con.execute(
+        "ALTER TABLE pfs.code_element ADD COLUMN IF NOT EXISTS confirmed_by VARCHAR"
+    )
+    con.execute(
+        "ALTER TABLE pfs.code_element_review ADD COLUMN IF NOT EXISTS source VARCHAR"
+    )
 
 
 def _insert(con: Any, table: str, rows: Sequence[Any]) -> int:
@@ -336,6 +361,20 @@ def read_elements(con: Any, code: str) -> list[ElementRow]:
     return [ElementRow(*r) for r in rows]
 
 
+def read_reviews(con: Any, code: str) -> list[ReviewRow]:
+    rows = con.execute(
+        "SELECT * FROM pfs.code_element_review WHERE code = ? ORDER BY p_id", [code]
+    ).fetchall()
+    return [ReviewRow(*r) for r in rows]
+
+
+def read_all_reviews(con: Any) -> list[ReviewRow]:
+    rows = con.execute(
+        "SELECT * FROM pfs.code_element_review ORDER BY code, p_id"
+    ).fetchall()
+    return [ReviewRow(*r) for r in rows]
+
+
 def read_events(con: Any, code: str) -> list[EventRow]:
     rows = con.execute(
         "SELECT * FROM pfs.code_event WHERE code = ? ORDER BY year, kind", [code]

src/pfs/extract.py

@@ -7,13 +7,16 @@ description, RVU short description); ``extract_code`` gathers every
 source for a code — FR, then the newest CPT codebook edition's own
 ``pfs.cpt_code`` row (stem + tail, then each required-elements list
 item), then HCPCS, then RVU — with FR rows winning over CPT and CPT
-winning over HCPCS/RVU on a duplicate ``(type, value, detail)``. No I/O
-here beyond what the caller hands in.
+winning over HCPCS/RVU on a duplicate ``(type, value, detail)``. The
+losing source isn't just dropped: it's recorded on the winning row's
+``confirmed_by`` as a ``source:item_key`` token (#703) — an independent
+transcription of the same element cross-checks the winner without
+doubling the row count. No I/O here beyond what the caller hands in.
 """
 
 from __future__ import annotations
 
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 from typing import Any, Callable, Sequence
 
 from pfs.codetables import ElementRow, ReviewRow, is_missing_table_error
@@ -41,6 +44,35 @@ class Extraction:
     reviews: tuple[ReviewRow, ...]
 
 
+def _merge_or_confirm(
+    merged: dict[tuple[str, str, str], ElementRow], r: ElementRow
+) -> None:
+    """Add *r* to *merged*, keyed by ``(type, value, detail)``. A first
+    sighting of a key is just kept. A later row for a key already held by
+    a *different* source (#703) doesn't replace the winner — FR still
+    beats CPT, CPT still beats HCPCS/RVU, and the row count doesn't grow
+    — it cross-checks it: ``r``'s ``source:item_key`` is appended to the
+    kept row's ``confirmed_by`` (deduped, so re-running the same source
+    twice, or a code with more than one CPT hit, doesn't repeat a token).
+    A later row from the *same* source as the one already kept is a
+    same-source duplicate (e.g. two FR rule years naming the same
+    element), not a second, independent source — it's dropped exactly as
+    before, uncounted as confirmation."""
+    key = (r.type, r.value, r.detail)
+    existing = merged.get(key)
+    if existing is None:
+        merged[key] = r
+        return
+    if r.source == existing.source:
+        return
+    token = f"{r.source}:{r.item_key}"
+    tokens = [t for t in existing.confirmed_by.split(",") if t]
+    if token in tokens:
+        return
+    tokens.append(token)
+    merged[key] = replace(existing, confirmed_by=",".join(tokens))
+
+
 def _row(code: str, e: Element, para: Para, *, year: int, source: str) -> ElementRow:
     return ElementRow(
         code,
@@ -75,7 +107,7 @@ def extract_run(
             rows.setdefault(e, _row(run.code, e, para, year=run.rule_year, source="fr"))
         else:
             reviews.append(
-                ReviewRow(run.code, para.text, "", "", para.item_key, para.p_id)
+                ReviewRow(run.code, para.text, "", "", para.item_key, para.p_id, "fr")
             )
     # 3. deterministic on the stem (and anything the whole text reveals)
     for e in parse_descriptor(run.text):
@@ -149,7 +181,7 @@ def _cpt_elements(
             e = Element(_TYPE_OF[choice], choice)
             rows.append(_cpt_row(code, edition_year, item_key, e, text))
         else:
-            reviews.append(ReviewRow(code, text, "", "", item_key, 0))
+            reviews.append(ReviewRow(code, text, "", "", item_key, 0, "cpt"))
     return rows, reviews
 
 
@@ -165,20 +197,20 @@ def extract_code(
     for run in descriptor_runs(store, code):
         x = extract_run(run, classify=classify)
         for r in x.rows:
-            merged.setdefault((r.type, r.value, r.detail), r)
+            _merge_or_confirm(merged, r)
         reviews.extend(x.reviews)
     cpt_rows, cpt_reviews = _cpt_elements(con, code, classify=classify)
     for r in cpt_rows:
-        merged.setdefault((r.type, r.value, r.detail), r)
+        _merge_or_confirm(merged, r)
     reviews.extend(cpt_reviews)
     long_desc = hcpcs_long_description(con, code)
     years = rvu_descriptions(con, code)
     if long_desc:
         y = years[-1][0] if years else 0
         for r in extract_text(code, long_desc, year=y, source="hcpcs").rows:
-            merged.setdefault((r.type, r.value, r.detail), r)
+            _merge_or_confirm(merged, r)
     for year, _status, desc in years:
         for r in extract_text(code, desc, year=year, source="rvu").rows:
-            merged.setdefault((r.type, r.value, r.detail), r)
+            _merge_or_confirm(merged, r)
     ordered = sorted(merged.values(), key=lambda r: (r.type, r.value, r.detail))
     return Extraction(code, tuple(ordered), tuple(reviews))

tests/cli/test_pfs_cli.py

@@ -128,6 +128,59 @@ class TestElements:
         assert [r.value for r in read_elements(con, "G0556")] == ["consent"]
         assert con.published == [True]
 
+    def test_summary_line_reports_confirmed_count_when_nonzero(self, con, monkeypatch):
+        # #703: the elements summary surfaces confirmed_by, but only when
+        # there's something to report.
+        def fake_extract(store, con_, code, *, classify=None):
+            return Extraction(
+                code,
+                (
+                    ElementRow(
+                        code,
+                        2026,
+                        "activity",
+                        "consent",
+                        "",
+                        "Consent;",
+                        "K",
+                        1,
+                        2,
+                        "fr",
+                        "cpt:GQGTPGYV",
+                    ),
+                    ElementRow(
+                        code,
+                        2026,
+                        "activity",
+                        "24-7-access",
+                        "",
+                        "24/7;",
+                        "K",
+                        3,
+                        2,
+                        "fr",
+                    ),
+                ),
+                (),
+            )
+
+        monkeypatch.setattr(pfs_cli, "extract_code", fake_extract)
+        res = runner.invoke(app, ["pfs", "elements", "--code", "g0556", "--no-llm"])
+        assert res.exit_code == 0, res.output
+        assert "G0556: 2 elements, 0 for review, 1 confirmed by a second source" in (
+            res.output
+        )
+
+    def test_summary_line_omits_confirmed_when_none(self, con, monkeypatch):
+        monkeypatch.setattr(
+            pfs_cli,
+            "extract_code",
+            lambda s, c, code, *, classify=None: Extraction(code, (), ()),
+        )
+        res = runner.invoke(app, ["pfs", "elements", "--code", "g0556", "--no-llm"])
+        assert res.exit_code == 0, res.output
+        assert "confirmed" not in res.output
+
     def test_dry_run_does_not_write(self, con, monkeypatch):
         monkeypatch.setattr(
             pfs_cli,
@@ -930,3 +983,33 @@ class TestReview:
         )
         res = runner.invoke(app, ["pfs", "review"])
         assert res.exit_code == 0 and "G0556" in res.output and "Odd line" in res.output
+
+    def test_source_shown_when_present(self, con):
+        # #703: CPT-sourced review lines are distinguishable from FR ones.
+        from pfs.codetables import ReviewRow
+
+        write_elements(
+            con,
+            "G0556",
+            [],
+            [
+                ReviewRow("G0556", "Odd fr line;", "", "", "K", 9, "fr"),
+                ReviewRow("G0556", "Odd cpt line;", "", "", "GQGTPGYV", 0, "cpt"),
+            ],
+        )
+        res = runner.invoke(app, ["pfs", "review"])
+        assert res.exit_code == 0, res.output
+        assert "(fr)" in res.output and "(cpt)" in res.output
+
+    def test_code_filter_scopes_the_query(self, con):
+        from pfs.codetables import ReviewRow
+
+        write_elements(
+            con, "G0556", [], [ReviewRow("G0556", "Odd line;", "", "", "K", 9, "fr")]
+        )
+        write_elements(
+            con, "99490", [], [ReviewRow("99490", "Other line;", "", "", "K", 3, "fr")]
+        )
+        res = runner.invoke(app, ["pfs", "review", "--code", "g0556"])
+        assert res.exit_code == 0, res.output
+        assert "Odd line" in res.output and "Other line" not in res.output

tests/pfs/test_codetables.py

@@ -15,12 +15,14 @@ from pfs.codetables import (
     is_missing_table_error,
     read_all_elements,
     read_all_events,
+    read_all_reviews,
     read_cpt_codes,
     read_cpt_instructions,
     read_cpt_sections,
     read_elements,
     read_events,
     read_families,
+    read_reviews,
     write_cpt_edition,
     write_elements,
     write_events,
@@ -96,6 +98,106 @@ class TestDDL:
             "cpt_code_alt",
         } <= names
 
+    def test_ensure_tables_upgrades_a_pre_existing_element_table_without_confirmed_by(
+        self,
+    ):
+        # #703, same pattern as code_family/note (Ruling C1): a replica
+        # created before confirmed_by/source existed has a 10-column
+        # pfs.code_element and a 6-column pfs.code_element_review —
+        # ensure_tables must add the columns in place.
+        con = duckdb.connect(":memory:")
+        try:
+            con.execute("CREATE SCHEMA IF NOT EXISTS pfs;")
+            con.execute(
+                "CREATE TABLE pfs.code_element ("
+                "code VARCHAR, year INTEGER, type VARCHAR, value VARCHAR, "
+                "detail VARCHAR, text VARCHAR, item_key VARCHAR, p_id INTEGER, "
+                "page INTEGER, source VARCHAR)"
+            )
+            con.execute(
+                "CREATE TABLE pfs.code_element_review ("
+                "code VARCHAR, text VARCHAR, proposed_type VARCHAR, "
+                "proposed_value VARCHAR, item_key VARCHAR, p_id INTEGER)"
+            )
+            con.execute(
+                "INSERT INTO pfs.code_element VALUES "
+                "('99490', 2015, 'activity', 'consent', '', 'Consent;', "
+                "'DE2VH9PD', 1245, 67716, 'fr')"
+            )
+            con.execute(
+                "INSERT INTO pfs.code_element_review VALUES "
+                "('99490', 'Odd line;', '', '', 'DE2VH9PD', 1246)"
+            )
+            ensure_tables(con)
+            el_cols = {
+                r[0]
+                for r in con.execute(
+                    "SELECT column_name FROM information_schema.columns "
+                    "WHERE table_schema='pfs' AND table_name='code_element'"
+                ).fetchall()
+            }
+            rev_cols = {
+                r[0]
+                for r in con.execute(
+                    "SELECT column_name FROM information_schema.columns "
+                    "WHERE table_schema='pfs' AND table_name='code_element_review'"
+                ).fetchall()
+            }
+            assert "confirmed_by" in el_cols
+            assert "source" in rev_cols
+            # The pre-existing row survives the upgrade with confirmed_by
+            # = NULL (mirrors the note-field precedent, C1) — read_elements
+            # does not crash on it.
+            row = read_elements(con, "99490")[0]
+            assert row.confirmed_by is None
+            # A fresh write still round-trips a real value.
+            write_elements(
+                con,
+                "99491",
+                [
+                    ElementRow(
+                        "99491",
+                        2015,
+                        "activity",
+                        "consent",
+                        "",
+                        "Consent;",
+                        "DE2VH9PD",
+                        1245,
+                        67716,
+                        "fr",
+                        "cpt:GQGTPGYV",
+                    )
+                ],
+                [],
+            )
+            assert read_elements(con, "99491")[0].confirmed_by == "cpt:GQGTPGYV"
+        finally:
+            con.close()
+
+    def test_read_elements_tolerates_a_table_with_no_confirmed_by_column(self):
+        # I4-style tolerance: a read-only path against an old replica must
+        # not call ensure_tables (that would need a write connection) — it
+        # must cope with the short tuple SELECT * returns.
+        con = duckdb.connect(":memory:")
+        try:
+            con.execute("CREATE SCHEMA IF NOT EXISTS pfs;")
+            con.execute(
+                "CREATE TABLE pfs.code_element ("
+                "code VARCHAR, year INTEGER, type VARCHAR, value VARCHAR, "
+                "detail VARCHAR, text VARCHAR, item_key VARCHAR, p_id INTEGER, "
+                "page INTEGER, source VARCHAR)"
+            )
+            con.execute(
+                "INSERT INTO pfs.code_element VALUES "
+                "('99490', 2015, 'activity', 'consent', '', 'Consent;', "
+                "'DE2VH9PD', 1245, 67716, 'fr')"
+            )
+            rows = read_elements(con, "99490")
+            assert len(rows) == 1 and rows[0].confirmed_by == ""
+        finally:
+            con.close()
+
 
 class TestElements:
     def test_write_replaces_per_code(self, con):
@@ -125,6 +227,65 @@ class TestElements:
         assert read_elements(con, "99490") == []
         assert len(read_elements(con, "G0556")) == 1
 
+    def test_confirmed_by_round_trips(self, con):
+        # #703: confirmed_by is a plain column — it round-trips like any
+        # other field, and doesn't change the row count.
+        row = ElementRow(
+            "99490",
+            2015,
+            "activity",
+            "comprehensive-care-plan",
+            "",
+            "Comprehensive care plan;",
+            "DE2VH9PD",
+            1245,
+            67716,
+            "fr",
+            "cpt:GQGTPGYV",
+        )
+        assert write_elements(con, "99490", [row], []) == 1
+        got = read_elements(con, "99490")
+        assert len(got) == 1
+        assert got[0].confirmed_by == "cpt:GQGTPGYV"
+
+    def test_confirmed_by_defaults_to_empty_string(self, con):
+        # Ruling C1 pattern: last field, defaulted, so a caller that omits
+        # it (as `_el` above does) still works.
+        write_elements(con, "99490", [_el()], [])
+        assert read_elements(con, "99490")[0].confirmed_by == ""
+
+
+class TestReviews:
+    def test_write_and_read_round_trip_source(self, con):
+        write_elements(
+            con,
+            "99490",
+            [],
+            [
+                ReviewRow("99490", "Odd fr line;", "", "", "DE2VH9PD", 1246, "fr"),
+                ReviewRow("99490", "Odd cpt line;", "", "", "GQGTPGYV", 1247, "cpt"),
+            ],
+        )
+        got = read_reviews(con, "99490")
+        assert [r.source for r in got] == ["fr", "cpt"]
+
+    def test_source_defaults_to_empty_string(self, con):
+        write_elements(
+            con, "99490", [], [ReviewRow("99490", "Odd line;", "", "", "K", 9)]
+        )
+        assert read_reviews(con, "99490")[0].source == ""
+
+    def test_read_all_reviews_matches_per_code_reader(self, con):
+        write_elements(
+            con, "99490", [], [ReviewRow("99490", "Odd line;", "", "", "K", 9, "fr")]
+        )
+        write_elements(
+            con, "G0556", [], [ReviewRow("G0556", "Other line;", "", "", "K", 3, "cpt")]
+        )
+        out = read_all_reviews(con)
+        assert {r.code for r in out} == {"99490", "G0556"}
+        assert out == read_reviews(con, "99490") + read_reviews(con, "G0556")
+
 
 class TestReadAll:
     def test_read_all_elements_matches_per_code_reader(self, con):

tests/pfs/test_extract.py

@@ -7,10 +7,12 @@ import sqlite3
 import duckdb
 import pytest
 
+from pfs.codetables import ElementRow
 from pfs.descriptors import DescriptorRun, Para, descriptor_runs
 from pfs.extract import (
     Extraction,
     _cpt_elements,
+    _merge_or_confirm,
     extract_code,
     extract_run,
     extract_text,
@@ -74,6 +76,12 @@ class TestDeterministic:
         assert [r.p_id for r in x.reviews] == [1199]
         assert x.reviews[0].proposed_value == ""
 
+    def test_review_rows_carry_the_fr_source(self):
+        # #703: ReviewRow.source distinguishes FR-sourced review lines
+        # from CPT-sourced ones.
+        x = extract_run(RUN)
+        assert x.reviews and all(r.source == "fr" for r in x.reviews)
+
     def test_home_setting_from_element_line(self):
         assert "home" in _vals(extract_run(RUN), "setting")
 
@@ -279,11 +287,12 @@ class TestExtractCodeCpt:
         stem_row = next(r for r in cpt_rows if r.value == "calendar-month")
         assert stem_row.text == "Chronic care management services"
         assert stem_row.p_id == 0 and stem_row.page == 0
-        assert [(r.text, r.item_key, r.p_id) for r in x.reviews] == [
+        assert [(r.text, r.item_key, r.p_id, r.source) for r in x.reviews] == [
             (
                 "Something the vocabulary does not know about at all",
                 "GQGTPGYV",
                 0,
+                "cpt",
             )
         ]
 
@@ -338,6 +347,10 @@ class TestExtractCodeCpt:
             consent_rows = [r for r in x.rows if r.value == "consent"]
             assert len(consent_rows) == 1
             assert consent_rows[0].source == "fr" and consent_rows[0].p_id == 901
+            # #703: the CPT manual's independent transcription of the same
+            # element cross-checks the winning FR row instead of being
+            # dropped outright — recorded as a source:item_key token.
+            assert consent_rows[0].confirmed_by == "cpt:GQGTPGYV"
         finally:
             store_con.close()
 
@@ -359,6 +372,10 @@ class TestExtractCodeCpt:
         consent_rows = [r for r in x.rows if r.value == "consent"]
         assert len(consent_rows) == 1
         assert consent_rows[0].source == "cpt"
+        # #703: HCPCS's independent transcription confirms the winning CPT
+        # row (HCPCS rows carry no item_key, so the token's second half is
+        # empty).
+        assert consent_rows[0].confirmed_by == "hcpcs:"
 
     def test_classifier_places_an_unmatched_cpt_element_line(self, store, con):
         # Mirrors TestClassifier's FR-side coverage, but for a required-
@@ -463,3 +480,57 @@ class TestText:
         assert all(
             r.item_key == "" and r.p_id == 0 and r.source == "hcpcs" for r in x.rows
         )
+
+
+def _el(*, source, item_key="", value="consent"):
+    return ElementRow(
+        "99490", 2024, "activity", value, "", "Consent;", item_key, 0, 0, source
+    )
+
+
+class TestMergeOrConfirm:
+    """#703: a second source's independent transcription of the same
+    ``(type, value, detail)`` cross-checks the winning row instead of
+    being dropped — the row count doesn't grow."""
+
+    def test_first_sighting_is_kept_unconfirmed(self):
+        merged: dict = {}
+        _merge_or_confirm(merged, _el(source="fr"))
+        (row,) = merged.values()
+        assert row.source == "fr" and row.confirmed_by == ""
+
+    def test_second_source_confirms_without_replacing_the_winner(self):
+        merged: dict = {}
+        _merge_or_confirm(merged, _el(source="fr", item_key="K1"))
+        _merge_or_confirm(merged, _el(source="cpt", item_key="GQGTPGYV"))
+        (row,) = merged.values()
+        assert row.source == "fr"  # the first source still wins
+        assert row.confirmed_by == "cpt:GQGTPGYV"
+
+    def test_same_source_duplicate_does_not_confirm(self):
+        # Two rows from the same source (e.g. two FR rule years naming the
+        # same element) is a same-source duplicate, not a second,
+        # independent source — dropped exactly as before, uncounted.
+        merged: dict = {}
+        _merge_or_confirm(merged, _el(source="fr", item_key="K1"))
+        _merge_or_confirm(merged, _el(source="fr", item_key="K2"))
+        (row,) = merged.values()
+        assert row.item_key == "K1" and row.confirmed_by == ""
+
+    def test_repeated_same_token_is_not_duplicated(self):
+        # Re-running the same source twice (or a code with more than one
+        # CPT hit for the same element) must not repeat a token.
+        merged: dict = {}
+        _merge_or_confirm(merged, _el(source="fr", item_key="K1"))
+        _merge_or_confirm(merged, _el(source="cpt", item_key="GQGTPGYV"))
+        _merge_or_confirm(merged, _el(source="cpt", item_key="GQGTPGYV"))
+        (row,) = merged.values()
+        assert row.confirmed_by == "cpt:GQGTPGYV"
+
+    def test_a_third_source_appends_a_second_token(self):
+        merged: dict = {}
+        _merge_or_confirm(merged, _el(source="fr", item_key="K1"))
+        _merge_or_confirm(merged, _el(source="cpt", item_key="GQGTPGYV"))
+        _merge_or_confirm(merged, _el(source="hcpcs"))
+        (row,) = merged.values()
+        assert row.confirmed_by == "cpt:GQGTPGYV,hcpcs:"