[verified] Separate rework intake from publication counter checks

harness/agents/impl/skills/gitea-issue-rework/references/input-gate.md

@@ -26,7 +26,15 @@ The target file is exactly schema `gitea-agent.rework-input/v1`:
 
 Use the independently verified current PR head and SHA-256 of the exact fresh Source Issue UTF-8 body. Names/positive integer IDs/digests/field sets are strict; Issue and PR differ. No selected review ID, final rework head, rendered comment, or `expected.json` is required. The schema and explicit provider contract are validated before credentials or network access.
 
-Two authenticated GET-only captures reuse identity-first native metadata, repository/PR/Issue/author binding, complete bounded pagination, Link provenance, per-review inline/count capability fallback and drift checks. The initial phase also requires the repository's native default branch to match PR base ref, the exact discovery-draft/v2 first line, and exact Source Issue digest. Discovery structural approval, unresolved-question checks and edit authorization are independent upstream requirements, not claims made by this gate.
+Two authenticated GET-only captures reuse identity-first native metadata, repository/PR/Issue/author binding, complete bounded pagination, Link provenance, per-review inline record counts and drift checks. Initial intake requires every review's real inline endpoint; 404/405 counter-only fallback is not allowed, even with a `per_review_or_pr_count` provider declaration. The initial phase also requires the repository's native default branch to match PR base ref, the exact discovery-draft/v2 first line, and exact Source Issue digest. Discovery structural approval, unresolved-question checks and edit authorization are independent upstream requirements, not claims made by this gate.
+
+## Initial collection checks versus publication counter checks
+
+Initial intake validates collected rows, endpoint/pagination provenance, duplicate IDs (including cross-review inline IDs), review-level `comments_count`, record cardinality and the Source Issue comment collection. These remain blocking checks. PR-wide `comments` and `review_comments` projection crosschecks are separate diagnostics, not evidence that a complete collection is missing records and not initial decision blockers. Source Issue `comments` crosschecking is unchanged.
+
+The receipt adds `counter_diagnostics`, with keys `inline`, `pr` and `pr_issue`. Each entry records the declared `field`/`components`, unmodified `observed` value, collection-derived `expected` value, and `status`: `match`, `mismatch`, or `unverified` (missing, negative or non-integer observed counter, including booleans). A match means only that the explicitly declared formula matches this observation; it does not attest deployed-provider semantics or counter freshness. Diagnostics never auto-select a projection, modify raw counts, or grant edit/publication permission. Missing/malformed PR counters do not bypass malformed native records, failed GETs, incomplete lists or drift checks.
+
+Provider schema/version binding and explicit declarations are still required. Rework/no-op/blocked decisions retain their role/head/contract/state rules regardless of diagnostics. `snapshot`/`verify-live` publication readback continues to require strict provider counter crosschecks and its existing explicitly declared fallback; an initial receipt cannot substitute for that verification.
 
 ## Derived decisions
 

harness/agents/impl/skills/gitea-issue-rework/references/readback.md

@@ -69,8 +69,13 @@ automatically detect a mode or choose one to fit observed counters. Both
 policy; per-review counters, complete collection pagination, duplicate IDs and
 record cardinality remain checked.
 
-`per_review` requires every exact-review comment endpoint. Only the explicit
+`per_review` requires every exact-review comment endpoint. Only publication
-`per_review_or_pr_count` capability permits endpoint-unavailable 404/405 fallback.
+`snapshot`/`verify-live` with an explicit `per_review_or_pr_count` capability
+permits endpoint-unavailable 404/405 fallback. Initial `input-gate` instead requires
+real inline records and rejects this fallback; its PR-wide counter projections are
+recorded as non-blocking `counter_diagnostics`, not used to establish collection
+completeness. See [input-gate.md](input-gate.md) for the phase boundary. Publication
+counter crosschecks, Source Issue protection and review decision rules are unchanged.
 Permission errors, redirects, malformed JSON and conflicting metadata never
 trigger fallback. Fallback uses a fresh PR count, the declared review/inline
 components and complete review `comments_count` records; disagreement is BLOCK.

harness/agents/impl/skills/gitea-issue-rework/scripts/rework_input_gate.py

@@ -120,8 +120,8 @@ def _validate_raw(adapter, raw, inp):
         raise InputGateError('incomplete review comment endpoints')
     for rid, collected in raw['inline'].items():
         adapter.validate_collection(collected, prefix+'/pulls/'+str(target['pr'])+'/reviews/'+rid+'/comments',
-                                    allow_missing=adapter.provider['inline']['mode'] == 'per_review_or_pr_count')
+                                    allow_missing=False)
-    adapter.counts(raw)
+    adapter.collection_counts(raw)
     source = raw['source_issue']
     if source['body'].splitlines()[:1] != ['<!-- gitea-agent:discovery-draft/v2 -->']:
         raise InputGateError('invalid discovery draft marker')
@@ -225,11 +225,12 @@ def derive_decision(raw, inp, impl_actor_id):
     if len(json.dumps(decision,ensure_ascii=False,sort_keys=True).encode()) > 1_000_000: raise InputGateError('decision artifact too large')
     return decision
 
-def build_receipt(raw, inp, decision):
+def build_receipt(raw, inp, decision, counter_diagnostics):
     receipt = {'schema':RECEIPT_SCHEMA,'status':decision['status'],'reason_codes':decision['reason_codes'],
             'target':decision['target'],'head':decision['head'],'contract_sha256':decision['contract_sha256'],
             'actor':decision['actor'],'input_sha256':digest(inp),'raw_input_snapshot':copy.deepcopy(raw),
             'raw_input_sha256':digest(raw),'fresh_GET_provenance':{'kind':'fresh_authenticated_GET','read_only':True},
+            'counter_diagnostics':copy.deepcopy(counter_diagnostics),
             'decision':decision}
     if len(json.dumps(receipt, ensure_ascii=False, sort_keys=True, allow_nan=False).encode('utf-8')) > 1_000_000:
         raise InputGateError('receipt artifact too large')

harness/agents/impl/skills/gitea-issue-rework/scripts/rework_readback.py

@@ -332,13 +332,15 @@ class ReworkReadbackAdapter:
                 or review['user']['login'] not in ('gitea-review-agent', 'gitea-spec-agent')):
             raise ReadbackError('source review author or state mismatch')
 
-    def counts(self, raw):
+    def _collection_counts(self, raw, *, allow_counter_fallback=False):
         inline = sum(integer(review.get('comments_count')) for review in raw['reviews']['rows'])
         fallback = False
         inline_ids = set()
         for review in raw['reviews']['rows']:
             collected = raw['inline'][str(review['id'])]
             if collected['pages'][0]['status'] in (404,405):
+                if not allow_counter_fallback:
+                    raise ReadbackError('inline collection unavailable for input gate')
                 if self.provider['inline']['mode'] != 'per_review_or_pr_count':
                     raise ReadbackError('inline fallback not explicitly authorized')
                 fallback = True
@@ -347,29 +349,54 @@ class ReworkReadbackAdapter:
             for row in collected['rows']:
                 if row['id'] in inline_ids: raise ReadbackError('duplicate cross-review inline ID')
                 inline_ids.add(row['id'])
-        parts = self.provider['inline']['components']
-        computed = ((inline if 'inline' in parts else 0)
-                    + (len(raw['reviews']['rows']) if 'reviews' in parts else 0))
-        field = self.provider['inline']['field']
-        if field not in raw['pr']:
-            if (self.provider['inline']['mode'] != 'per_review' or fallback or computed != 0
-                    or len(inline_ids) != 0):
-                raise ReadbackError('missing inline provider counter')
-        elif integer(raw['pr'].get(field)) != computed:
-            raise ReadbackError('inline provider counter crosscheck failed')
         if not fallback and len(inline_ids) != inline:
             raise ReadbackError('inline record cardinality mismatch')
         counts = {'top_level':len(raw['comments']['rows']),
                   'reviews':len(raw['reviews']['rows']), 'inline':inline}
+        if integer(raw['source_issue'].get('comments')) != len(raw['source_comments']['rows']):
+            raise ReadbackError('source collection count mismatch')
+        return counts, fallback
+
+    def collection_counts(self, raw):
+        """Initial intake requires real inline records, never a counter-only fallback."""
+        return self._collection_counts(raw)[0]
+
+    def counter_diagnostics(self, raw):
+        """Declared PR projections are diagnostic only for complete initial intake."""
+        counts = self.collection_counts(raw)
+        diagnostics = {}
+        projections = {'inline': ('pr', self.provider['inline'])}
+        projections.update({scope: (scope, self.provider['aggregate'][scope])
+                            for scope in ('pr', 'pr_issue')})
+        for name, (scope, spec) in projections.items():
+            observed = raw[scope].get(spec['field'])
+            expected = sum(counts[part] for part in spec['components'])
+            valid = type(observed) is int and observed >= 0
+            diagnostics[name] = {'field':spec['field'], 'components':list(spec['components']),
+                                 'observed':observed, 'expected':expected,
+                                 'status':('match' if observed == expected else 'mismatch')
+                                           if valid else 'unverified'}
+        return diagnostics
+
+    def counts(self, raw):
+        """Publication/readback retains strict provider counter crosschecks."""
+        counts, fallback = self._collection_counts(raw, allow_counter_fallback=True)
+        parts = self.provider['inline']['components']
+        computed = sum(counts[part] for part in parts)
+        field = self.provider['inline']['field']
+        if field not in raw['pr']:
+            if (self.provider['inline']['mode'] != 'per_review' or fallback or computed != 0
+                    or counts['inline'] != 0):
+                raise ReadbackError('missing inline provider counter')
+        elif integer(raw['pr'].get(field)) != computed:
+            raise ReadbackError('inline provider counter crosscheck failed')
         for scope in ('pr','pr_issue'):
             spec = self.provider['aggregate'][scope]
             if integer(raw[scope].get(spec['field'])) != sum(counts[name] for name in spec['components']):
                 raise ReadbackError('provider aggregate count mismatch')
-        if integer(raw['source_issue'].get('comments')) != len(raw['source_comments']['rows']):
-            raise ReadbackError('source collection count mismatch')
         return counts
 
-    def capture_input(self, target, *, require_default_branch=True):
+    def _capture_raw(self, target, *, require_default_branch):
         identity, _ = self.get('/user')
         if integer(identity.get('id'), 1) != self.actor_id or identity.get('login') != 'gitea-impl-agent':
             raise ReadbackError('authenticated identity mismatch')
@@ -389,11 +416,16 @@ class ReworkReadbackAdapter:
         raw['inline'] = {str(review['id']):self.collection(prefix+'/pulls/'+str(target['pr'])+
             '/reviews/'+str(review['id'])+'/comments', allow_missing=self.provider['inline']['mode'] == 'per_review_or_pr_count') for review in raw['reviews']['rows']}
         self.bind_input(target, raw, require_default_branch=require_default_branch)
-        self.counts(raw)
+        return raw
+
+    def capture_input(self, target, *, require_default_branch=True):
+        raw = self._capture_raw(target, require_default_branch=require_default_branch)
+        self.collection_counts(raw)
         return raw
 
     def capture(self, target):
-        raw = self.capture_input(target, require_default_branch=False)
+        raw = self._capture_raw(target, require_default_branch=False)
+        self.counts(raw)
         self.bind(target, raw)
         return raw
 
@@ -416,7 +448,7 @@ class ReworkReadbackAdapter:
         if digest(first) != digest(second):
             raise ReadbackError('concurrent provider drift')
         decision = derive_decision(first, value, self.actor_id)
-        return build_receipt(first, value, decision)
+        return build_receipt(first, value, decision, self.counter_diagnostics(first))
 
     def snapshot(self, expected):
         target = self.target(expected)

harness/tests/impl/test_rework_input_gate.py

@@ -317,5 +317,130 @@ class OrderedDecisionTests(unittest.TestCase):
                 self.assertEqual(cmd.main(),2)
             self.assertGreater(len(requests),1);self.assertEqual(output.read_bytes(),b'created concurrently');self.assertEqual(output.stat().st_mode&0o777,0o600)
 
+
+    def test_pr_comment_counter_mismatch_preserves_no_op_and_diagnostics(self):
+        self.assertEqual(self.snapshot()['status'], 'no_op')
+        self.rb.pr['comments'] += 4
+        receipt = self.snapshot()
+        self.assertEqual(receipt['status'], 'no_op')
+        diagnostics = receipt['counter_diagnostics']
+        self.assertEqual(set(diagnostics), {'inline', 'pr', 'pr_issue'})
+        self.assertEqual(diagnostics['pr']['status'], 'mismatch')
+
+    def test_review_counter_mismatch_preserves_rework(self):
+        import copy
+        rows = copy.deepcopy(self.rows)
+        rows[1].update(state='REQUEST_CHANGES', body=self.code_body('request_changes'))
+        self.set_rows(rows)
+        self.rb.pr['review_comments'] += 3
+        receipt = self.snapshot()
+        self.assertEqual(receipt['status'], 'rework')
+        self.assertEqual(receipt['counter_diagnostics']['inline']['status'], 'mismatch')
+        self.assertEqual(receipt['decision']['rework_review_ids'], [302])
+
+    def test_malformed_pr_counters_are_unverified_diagnostics(self):
+        self.rb.pr['comments'] = 'not-a-count'
+        receipt = self.snapshot()
+        self.assertEqual(receipt['status'], 'no_op')
+        self.assertEqual(receipt['counter_diagnostics']['pr']['status'], 'unverified')
+        self.assertIn('field', receipt['counter_diagnostics']['pr'])
+
+    def test_unavailable_inline_endpoint_is_rejected_even_with_pr_count_fallback(self):
+        from urllib.parse import urlsplit
+        original = self.rb.request
+        def unavailable(path):
+            if '/pulls/9/reviews/' in urlsplit(path).path and urlsplit(path).path.endswith('/comments'):
+                self.rb.calls.append(path)
+                return {'message': 'not available'}, {}, 404
+            return original(path)
+        self.rb.request = unavailable
+        self.rb.contract['inline']['mode'] = 'per_review_or_pr_count'
+        with self.assertRaisesRegex(ValueError, 'inline collection unavailable'):
+            self.snapshot()
+
+    def test_inline_endpoint_405_is_rejected_even_with_pr_count_fallback(self):
+        from urllib.parse import urlsplit
+        original = self.rb.request
+        def unavailable(path):
+            if '/pulls/9/reviews/' in urlsplit(path).path and urlsplit(path).path.endswith('/comments'):
+                self.rb.calls.append(path)
+                return {'message': 'method not allowed'}, {}, 405
+            return original(path)
+        self.rb.request = unavailable
+        self.rb.contract['inline']['mode'] = 'per_review_or_pr_count'
+        with self.assertRaisesRegex(ValueError, 'inline collection unavailable'):
+            self.snapshot()
+
+    def test_publication_snapshot_remains_strict_on_pr_counter_mismatch(self):
+        from harness.tests.impl.test_rework_readback import ReworkReadbackTests
+        publication = ReworkReadbackTests()
+        publication.setUp()
+        self.addCleanup(publication.doCleanups)
+        publication.adapter().snapshot(publication.expected)
+        publication.pr['comments'] += 1
+        with self.assertRaisesRegex(ValueError, 'provider aggregate count mismatch'):
+            publication.adapter().snapshot(publication.expected)
+
+    def test_complete_lists_with_zero_pr_issue_counter_preserve_no_op(self):
+        self.rb.pr['comments'] = self.rb.underlying['comments'] = 0
+        receipt = self.snapshot()
+        self.assertEqual(receipt['status'], 'no_op')
+        for scope in ('pr', 'pr_issue'):
+            diagnostic = receipt['counter_diagnostics'][scope]
+            self.assertEqual(diagnostic, {'field':'comments',
+                'components':['top_level','reviews','inline'], 'observed':0,
+                'expected':2, 'status':'mismatch'})
+            self.assertEqual(receipt['raw_input_snapshot'][scope]['comments'], 0)
+
+    def test_missing_and_invalid_counters_do_not_replace_record_validation(self):
+        import copy
+        saved_pr, saved_issue = copy.deepcopy(self.rb.pr), copy.deepcopy(self.rb.underlying)
+        for scope, field in (('inline','review_comments'), ('pr','comments'), ('pr_issue','comments')):
+            for value in (None, True, -1, '0', 0.5, 'missing'):
+                with self.subTest(scope=scope, value=value):
+                    self.rb.pr, self.rb.underlying = copy.deepcopy(saved_pr), copy.deepcopy(saved_issue)
+                    record = self.rb.underlying if scope == 'pr_issue' else self.rb.pr
+                    if value == 'missing': record.pop(field)
+                    else: record[field] = value
+                    receipt = self.snapshot()
+                    self.assertEqual(receipt['status'], 'no_op')
+                    self.assertEqual(receipt['counter_diagnostics'][scope]['status'], 'unverified')
+                    raw_record = receipt['raw_input_snapshot']['pr_issue' if scope == 'pr_issue' else 'pr']
+                    if value == 'missing': self.assertNotIn(field, raw_record)
+                    else: self.assertEqual(raw_record[field], value)
+
+    def test_counter_mismatch_does_not_unblock_missing_code_review(self):
+        self.set_rows(self.rows[:1])
+        self.rb.pr['comments'] = self.rb.underlying['comments'] = 0
+        receipt = self.snapshot()
+        self.assertEqual(receipt['status'], 'blocked')
+        self.assertEqual(receipt['counter_diagnostics']['pr']['status'], 'mismatch')
+        self.assertEqual(receipt['decision']['rework_review_ids'], [])
+
+    def test_counter_mismatch_does_not_hide_per_review_inline_count_mismatch(self):
+        self.rb.reviews[0]['comments_count'] = 1
+        self.rb.pr['comments'] = self.rb.underlying['comments'] = 0
+        with self.assertRaisesRegex(ValueError, 'per-review inline count mismatch'):
+            self.snapshot()
+
+    def test_counter_mismatch_does_not_hide_duplicate_review_records(self):
+        import copy
+        self.rb.reviews.append(copy.deepcopy(self.rb.reviews[0]))
+        self.rb.pr['comments'] = self.rb.underlying['comments'] = 0
+        with self.assertRaisesRegex(ValueError, 'duplicate'):
+            self.snapshot()
+
+    def test_counter_mismatch_does_not_hide_incomplete_review_pagination(self):
+        original = self.rb.request
+        def truncated(path):
+            if '/pulls/9/reviews?' in path:
+                payload, headers, status = original(path)
+                return payload, dict(headers, **{'X-Total-Count':'3'}), status
+            return original(path)
+        self.rb.request = truncated
+        self.rb.pr['comments'] = self.rb.underlying['comments'] = 0
+        with self.assertRaisesRegex(ValueError, 'truncated collection'):
+            self.snapshot()
+
 if __name__ == '__main__':
     unittest.main()