Preserve optional intake metadata in ledger verification

scripts/verify_pr_reconciliation.py

@@ -97,14 +97,33 @@ def check_batch932_file(item):
             'Changed integrated file: ' + name)
 
 
+def project_family_decision(previous, decision):
+    """Carry optional metadata through incremental decisions, as the ledger does.
+
+    Keep mandatory fields from the latest decision: an incomplete later row
+    must not inherit an earlier head or disposition and silently pass.
+    """
+    projected = dict(decision)
+    for field in ('closure', 'review_completion', 'verification_evidence'):
+        if field not in projected and field in previous:
+            projected[field] = previous[field]
+    return projected
+
+
 def check_family_decision(item, decision):
     """Bind projected ledger state to its authoritative, pinned family decision."""
     fields = ('number', 'reviewed_head', 'current_head', 'disposition',
-              'github_state', 'integration_location', 'unresolved_question', 'closure')
-    for field in fields + ('review_completion', 'verification_evidence'):
+              'github_state', 'integration_location', 'unresolved_question')
+    if 'closure' not in decision:
+        # New intake rows may omit an unrecorded closure. No receipt may be
+        # invented; an earlier explicit null or receipt is carried above and
+        # still requires the corresponding ledger key below.
+        require(item.get('closure') is None,
+                f'#{decision["number"]}: ledger differs from family decision: closure')
+    for field in fields + ('closure', 'review_completion', 'verification_evidence'):
         if field in fields or field in decision:
             # JSON equality must distinguish true from 1, including nested flags.
-            require(field in item and json.dumps(item[field], sort_keys=True) ==
+            require(field in item and field in decision and json.dumps(item[field], sort_keys=True) ==
                     json.dumps(decision[field], sort_keys=True),
                     f'#{decision["number"]}: ledger differs from family decision: {field}')
 
@@ -266,6 +285,7 @@ def main():
     snapshot = json.loads((AUDIT / 'starting-snapshot.json').read_text())
     expected = {p['number']: p for p in snapshot['pull_requests']}
     family_decisions = {}
+    family_projections = {}
     for family_path in decision_paths():
         if not family_path.exists():
             continue
@@ -276,12 +296,14 @@ def main():
             # The generator applies later rounds in this order; retain the final
             # authoritative projection without rewriting historical metadata.
             family_decisions[row['number']] = row
+            family_projections[row['number']] = project_family_decision(
+                family_projections.get(row['number'], {}), row)
             if row['number'] not in expected:
                 expected[row['number']] = row['metadata']
     actual = {p['number']: p for p in ledger['pull_requests']}
     require(len(actual) == len(ledger['pull_requests']) == len(expected), 'Duplicate or missing PR')
     require(actual.keys() == expected.keys(), 'Starting PR set changed')
-    for number, decision in family_decisions.items():
+    for number, decision in family_projections.items():
         check_family_decision(actual[number], decision)
     for number, pr in actual.items():
         require(pr['starting_head'] == expected[number]['headRefOid'], f'#{number}: starting head changed')
@@ -592,7 +614,7 @@ def main():
         if companion_path.exists():
             for companion in json.loads(companion_path.read_text())['pull_requests']:
                 item = actual[companion['number']]
-                check_companion_review(item, companion, family_decisions.get(companion['number']))
+                check_companion_review(item, companion, family_projections.get(companion['number']))
         print('PASS batch970 batch: reviewed sources, finite replays, scoped claims and closure bindings')
     receipt_path = ROOT / 'docs/research/community-batch1047-integration.json'
     if receipt_path.exists():

tests/test_pr_reconciliation.py

@@ -25,7 +25,11 @@ def setUpClass(cls):
         for path in checker.decision_paths():
             if path.exists():
                 cls.families[path.parent.name.removeprefix('community-').removesuffix('-audit')] = json.loads(path.read_text())['pull_requests']
-        cls.latest = {row['number']: row for decisions in cls.families.values() for row in decisions}
+        cls.latest = {}
+        for decisions in cls.families.values():
+            for row in decisions:
+                cls.latest[row['number']] = checker.project_family_decision(
+                    cls.latest.get(row['number'], {}), row)
 
     def setUp(self):
         self.decision = deepcopy(self.latest[626])
@@ -35,7 +39,9 @@ def test_current_family_projections_match_authoritative_decisions(self):
         final = {}
         for family, decisions in self.families.items():
             for decision in decisions:
-                final[decision['number']] = (family, decision)
+                number = decision['number']
+                previous = final.get(number, (None, {}))[1]
+                final[number] = (family, checker.project_family_decision(previous, decision))
         for number, (family, decision) in final.items():
             with self.subTest(family=family, number=number):
                 checker.check_family_decision(self.items[number], decision)
@@ -70,13 +76,96 @@ def test_changed_or_missing_projection_fields_are_rejected(self):
                     checker.check_family_decision(item, self.decision)
 
     def test_missing_closure_is_distinct_from_recorded_null(self):
-        decision = self.latest[620]
-        item = deepcopy(self.items[620])
+        # PR620 later gained a closure receipt; retain its explicit-null
+        # historical decision as this missing-key rejection fixture.
+        decision = next(row for row in self.families['k492'] if row['number'] == 620)
+        item = deepcopy(decision)
         self.assertIsNone(decision['closure'])
         item.pop('closure')
         with self.assertRaisesRegex(ValueError, '#620:.*closure'):
             checker.check_family_decision(item, decision)
 
+    def test_incremental_decision_preserves_recorded_optional_metadata(self):
+        later = deepcopy(self.decision)
+        for field in ('closure', 'review_completion', 'verification_evidence'):
+            later.pop(field)
+        projected = checker.project_family_decision(self.decision, later)
+        self.assertEqual(projected, self.decision)
+        self.assertNotIn('closure', later)
+        checker.check_family_decision(self.item, projected)
+        for field in ('closure', 'review_completion', 'verification_evidence'):
+            for operation in ('drop', 'replace'):
+                item = deepcopy(self.item)
+                if operation == 'drop':
+                    item.pop(field)
+                else:
+                    item[field] = None
+                with self.subTest(field=field, operation=operation), self.assertRaisesRegex(
+                        ValueError, '#626:.*' + field):
+                    checker.check_family_decision(item, projected)
+        item = deepcopy(self.item)
+        item['review_completion']['published'] = 1
+        with self.assertRaisesRegex(ValueError, '#626:.*review_completion'):
+            checker.check_family_decision(item, projected)
+
+    def test_incremental_explicit_closure_overrides_previous_receipt(self):
+        later = deepcopy(self.decision)
+        for closure in (None, {'url': 'https://example.invalid/new-receipt'}):
+            later['closure'] = closure
+            projected = checker.project_family_decision(self.decision, later)
+            item = deepcopy(self.item)
+            item['closure'] = closure
+            checker.check_family_decision(item, projected)
+            with self.subTest(closure=closure), self.assertRaisesRegex(ValueError, '#626:.*closure'):
+                checker.check_family_decision(self.item, projected)
+        later.pop('closure')
+        previous = deepcopy(self.decision)
+        previous['closure'] = None
+        projected = checker.project_family_decision(previous, later)
+        item = deepcopy(self.item)
+        item['closure'] = None
+        checker.check_family_decision(item, projected)
+        item.pop('closure')
+        with self.assertRaisesRegex(ValueError, '#626:.*closure'):
+            checker.check_family_decision(item, projected)
+
+    def test_unrecorded_closure_accepts_only_absent_or_explicit_null(self):
+        decision = deepcopy(self.decision)
+        decision.pop('closure')
+        for operation in ('drop', 'null'):
+            item = deepcopy(decision)
+            if operation == 'null':
+                item['closure'] = None
+            checker.check_family_decision(item, decision)
+        for closure in (False, 0, {}, [], {'url': 'https://example.invalid/forged'}):
+            item = deepcopy(decision)
+            item['closure'] = closure
+            with self.subTest(closure=closure), self.assertRaisesRegex(ValueError, '#626:.*closure'):
+                checker.check_family_decision(item, decision)
+
+    def test_incremental_decision_cannot_inherit_missing_mandatory_fields(self):
+        for field in ('reviewed_head', 'current_head', 'disposition', 'github_state',
+                      'integration_location', 'unresolved_question'):
+            later = deepcopy(self.decision)
+            later.pop(field)
+            projected = checker.project_family_decision(self.decision, later)
+            self.assertNotIn(field, projected)
+            with self.subTest(field=field), self.assertRaisesRegex(ValueError, '#626:.*' + field):
+                checker.check_family_decision(self.item, projected)
+
+    def test_companion_review_retains_inherited_whole_head_completion(self):
+        later = deepcopy(self.decision)
+        for field in ('closure', 'review_completion', 'verification_evidence'):
+            later.pop(field)
+        projected = checker.project_family_decision(self.decision, later)
+        companion = {'head': self.item['current_head'], 'remaining': 'Earlier scoped hold.'}
+        item = deepcopy(self.item)
+        item['companion_reviews'] = [companion]
+        checker.check_companion_review(item, companion, projected)
+        item['closure'] = None
+        with self.assertRaisesRegex(ValueError, '#626:.*closure'):
+            checker.check_companion_review(item, companion, projected)
+
     def test_boolean_completion_flags_cannot_be_replaced_by_integers(self):
         for number in (626, 620):
             decision = self.latest[number]