Reconcile page-split timestamps for ten-pull transactions

src/nte_history_exporter/decoder/arc.py

@@ -266,6 +266,26 @@ def annotate_arc_groups(rows: list[dict[str, Any]]) -> None:
     # Pages are anchored at page 1, so ordinal 0 of every group is captured and
     # all UIDs are stable -- even a partially captured oldest 10-pull, whose
     # unseen rows can only append after the captured ones. All rows are exported.
+    # Arc transactions are always ten records. A transaction can cross response
+    # pages whose raw timestamps occasionally differ while displaying the same
+    # second. Re-key each consecutive batch to its first timestamp before UIDs
+    # are assigned, preserving prefix UIDs and preventing an ordinal restart.
+    run_start = 0
+    while run_start < len(rows):
+        displayed = rows[run_start].get("timestamp_decoded")
+        run_end = run_start + 1
+        while run_end < len(rows) and rows[run_end].get("timestamp_decoded") == displayed:
+            run_end += 1
+        for batch_start in range(run_start, run_end, 10):
+            batch = rows[batch_start : min(batch_start + 10, run_end)]
+            canonical = batch[0].get("timestamp_raw_hex", "")
+            for row in batch:
+                original = row.get("timestamp_raw_hex", "")
+                if original != canonical:
+                    row["timestamp_reconciled_from_raw_hex"] = original
+                    row["timestamp_raw_hex"] = canonical
+        run_start = run_end
+
     groups: dict[str, list[int]] = defaultdict(list)
     for index, row in enumerate(rows):
         groups[row["timestamp_raw_hex"]].append(index)

src/nte_history_exporter/decoder/boundary.py

@@ -97,6 +97,59 @@ def is_dice_record(row: dict[str, Any]) -> bool:
         return False
 
 
+def reconcile_monopoly_page_split_timestamps(rows: list[dict[str, Any]]) -> None:
+    """Join timestamp fragments only when they form a page-split 10-pull.
+
+    Monopoly also supports singles, so equal display timestamps alone are not a
+    sufficient reason to join groups. A reconciliation is allowed only when
+    adjacent raw-timestamp groups cross a history-page boundary and contain ten
+    dice results in total. Ancillary rewards travel with their timestamp group
+    but do not count toward the ten pulls.
+    """
+    groups: list[list[dict[str, Any]]] = []
+    for row in rows:
+        raw = row.get("timestamp_raw_hex", "")
+        if groups and groups[-1][0].get("timestamp_raw_hex", "") == raw:
+            groups[-1].append(row)
+        else:
+            groups.append([row])
+
+    index = 0
+    while index < len(groups):
+        first = groups[index]
+        first_dice = sum(is_dice_record(row) for row in first)
+        if not 0 < first_dice < 10:
+            index += 1
+            continue
+
+        displayed = first[0].get("timestamp_decoded")
+        pages = {row.get("page") for row in first}
+        dice_count = first_dice
+        end = index + 1
+        while end < len(groups) and groups[end][0].get("timestamp_decoded") == displayed:
+            candidate = groups[end]
+            candidate_dice = sum(is_dice_record(row) for row in candidate)
+            if dice_count + candidate_dice > 10:
+                break
+            dice_count += candidate_dice
+            pages.update(row.get("page") for row in candidate)
+            end += 1
+            if dice_count == 10:
+                break
+
+        if dice_count == 10 and len(pages) > 1 and end > index + 1:
+            canonical = first[0].get("timestamp_raw_hex", "")
+            for group in groups[index:end]:
+                for row in group:
+                    original = row.get("timestamp_raw_hex", "")
+                    if original != canonical:
+                        row["timestamp_reconciled_from_raw_hex"] = original
+                        row["timestamp_raw_hex"] = canonical
+            index = end
+        else:
+            index += 1
+
+
 def annotate_groups(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
     """Assign timestamp-group ordinals and stable UIDs to every decoded row.
 
@@ -109,6 +162,8 @@ def annotate_groups(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
     if not rows:
         return rows
 
+    reconcile_monopoly_page_split_timestamps(rows)
+
     groups: list[list[dict[str, Any]]] = []
     current: list[dict[str, Any]] = []
     prev_ts = None

tests/test_arc_decoding.py

@@ -1,7 +1,24 @@
 from tests.support import *  # noqa: F401,F403
+from nte_history_exporter.decoder.arc import annotate_arc_groups
 
 
 class ArcDecodingTests(unittest.TestCase):
+    def test_page_split_timestamp_variants_form_one_ten_pull(self):
+        rows = [
+            {
+                "page": 1 if index < 5 else 2,
+                "timestamp_raw_hex": "aa" if index < 5 else "bb",
+                "timestamp_decoded": "2026-07-11 07:21:22",
+            }
+            for index in range(10)
+        ]
+
+        annotate_arc_groups(rows)
+
+        self.assertEqual([row["timestamp_group_ordinal"] for row in rows], list(range(10)))
+        self.assertEqual({row["timestamp_raw_hex"] for row in rows}, {"aa"})
+        self.assertEqual(rows[5]["timestamp_reconciled_from_raw_hex"], "bb")
+
     def test_arc_key_timestamp_and_uid_match_fixture(self):
         fixture = load_network_fixture()
         row = fixture_session().build_rows("arc_miracle_box")[0]

tests/test_uid_compatibility.py

@@ -2,6 +2,27 @@ from tests.support import *  # noqa: F401,F403
 
 
 class UidCompatibilityTests(unittest.TestCase):
+    def test_monopoly_page_split_timestamp_variants_complete_ten_pull(self):
+        rows = [self._synthetic_row(1, "aa", "dice") for _ in range(5)]
+        rows += [self._synthetic_row(2, "bb", "dice") for _ in range(5)]
+        for row in rows:
+            row["timestamp_decoded"] = "2026-07-11 07:21:22"
+
+        annotated = annotate_groups(rows)
+
+        self.assertEqual([row["timestamp_group_ordinal"] for row in annotated], list(range(10)))
+        self.assertEqual({row["timestamp_raw_hex"] for row in annotated}, {"aa"})
+
+    def test_monopoly_singles_with_same_second_remain_separate(self):
+        rows = [self._synthetic_row(1, "aa", "dice"), self._synthetic_row(2, "bb", "dice")]
+        for row in rows:
+            row["timestamp_decoded"] = "2026-07-11 07:21:22"
+
+        annotated = annotate_groups(rows)
+
+        self.assertEqual([row["timestamp_group_ordinal"] for row in annotated], [0, 0])
+        self.assertEqual([row["timestamp_raw_hex"] for row in annotated], ["aa", "bb"])
+
     def test_uid_source_matches_committed_network_fixture(self):
         fixture = load_network_fixture()
         rows = annotate_groups(fixture_session().build_rows("permanent"))