diff --git a/docs/research/coordination.md b/docs/research/coordination.md index e700933860..183412d5bc 100644 --- a/docs/research/coordination.md +++ b/docs/research/coordination.md @@ -88,6 +88,11 @@ Joined path results show `head`/`base` for the file snapshot, `body_head`/`body_base` for PR metadata, and separate `head_matches` and `base_matches` flags. Check both flags and `truncated`. Coverage also lists `unverified_heads` and `unverified_bases`, even when no path matches the query. +It also lists PRs with unavailable file lists or advertised-count shortfalls +in `truncated_prs`, with a warning even when those PRs do not match +the requested path or fall outside `--limit`. A true `complete_pagination` +means the list paging finished; it does not mean every PR's file inventory +is complete. Term-and-path intersections retain each match's `truncated` flag. A missing legacy base remains unknown; it is never copied from another snapshot. Online path queries refresh endpoint-incomplete sources. Offline or failed refreshes keep the old evidence and explicitly report the missing binding. diff --git a/scripts/find_overlap.py b/scripts/find_overlap.py index 3e64e98477..bfc534e8b3 100644 --- a/scripts/find_overlap.py +++ b/scripts/find_overlap.py @@ -421,6 +421,11 @@ def main(): coverage["pr_files"] = {k: source[k] for k in ("fetched_at", "complete_pagination")} if source else None if source: coverage["pr_files"]["records"] = len(source["records"]) + truncated = sorted({row["number"] for row in source["records"] if row.get("truncated", False)}) + coverage["pr_files"]["truncated_prs"] = truncated + if truncated: + warnings.append(f"pr_files: incomplete file inventories for {len(truncated)} PRs; " + "paths may be missing. Check truncated_prs in coverage, even when no path matches.") pulls = snapshot["sources"].get("open_prs", {"records": []})["records"] heads = {pull["number"]: pull.get("head") for pull in pulls} file_heads = {row["number"]: row.get("head") for row in source["records"]} @@ -445,7 +450,7 @@ def main(): hits = {m["number"]: m for m in path_matches} matches = [dict(m, **{key: hits[m["number"]][key] for key in ("matched_paths", "head", "body_head", "head_matches", - "base", "body_base", "base_matches")}) for m in matches + "base", "body_base", "base_matches", "truncated")}) for m in matches if m.get("kind") == "pr" and m.get("number") in hits] if patterns and not any(t.strip() for t in terms): matches = [] @@ -462,7 +467,9 @@ def main(): report["path_patterns"] = patterns report["path_limits"] = ("Path matching covers the changed file paths GitHub reports for open PRs, " "not diff contents; all available file pages are followed. Missing lists or " - "advertised-count shortfalls are flagged truncated. Head/base identify the file " + "advertised-count shortfalls are flagged truncated and listed in coverage.pr_files.truncated_prs " + "independently of matches or --limit. complete_pagination describes list paging, " + "not completeness of each PR's file inventory. Head/base identify the file " "snapshot; body_head/body_base identify PR metadata. Both endpoints must match. " "This is not an atomic server snapshot. Unverified joins are flagged, " "including inventories with no matching path.") @@ -485,7 +492,8 @@ def main(): print("WARNING: " + warning) print(f"{len(matches)} possible matches; showing {min(args.limit, len(matches))}") for match in matches[:args.limit]: - print(f"\n[{match['kind']}] {match['title']}\n{match.get('url', match.get('path'))}") + flags = " (file list truncated)" if match.get("truncated") else "" + print(f"\n[{match['kind']}] {match['title']}{flags}\n{match.get('url', match.get('path'))}") if match.get("head"): print("Head: " + match["head"]) if match.get("head_matches") is False: diff --git a/tests/test_research_pr_inventory_coverage.py b/tests/test_research_pr_inventory_coverage.py new file mode 100644 index 0000000000..16e4a40cba --- /dev/null +++ b/tests/test_research_pr_inventory_coverage.py @@ -0,0 +1,92 @@ +"""Known file-inventory gaps must survive filtering and result limits.""" +from datetime import datetime, timezone +import json +import os +from pathlib import Path +import subprocess +import sys +import tempfile +import unittest + + +ROOT = Path(__file__).resolve().parents[1] +SCRIPT = ROOT / 'scripts/find_overlap.py' + + +class InventoryCoverageTests(unittest.TestCase): + def run_cli(self, inventories, *arguments): + with tempfile.TemporaryDirectory() as directory: + cache = Path(directory) / 'overlap.json' + bodies = [dict(kind='pr', number=row['number'], title='needle', body='', + head=row['head'], base=row['base'], state='open', + url=f"https://example.invalid/{row['number']}", + updated_at='2026-10-18T00:00:00Z') for row in inventories] + fetched_at = datetime.now(timezone.utc).isoformat() + sources = {name: dict(fetched_at=fetched_at, complete_pagination=True, records=rows) + for name, rows in [('open_prs', bodies), ('discussions', []), + ('pr_files', inventories)]} + original = json.dumps(dict(schema_version=1, repository='CrocSwap/integer-mult-bounds', + sources=sources)).encode('utf-8') + cache.write_bytes(original) + # A fresh synthetic snapshot must not require gh or any network call. + result = subprocess.run([sys.executable, '-B', str(SCRIPT), '--cache', str(cache), + '--max-age', '3600', *arguments], capture_output=True, + encoding='utf-8', timeout=30, env={**os.environ, 'PATH': ''}) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual(cache.read_bytes(), original) + return result.stdout + + @staticmethod + def row(number, paths, truncated): + return dict(number=number, head='a' * 40, base='b' * 40, + paths=paths, truncated=truncated) + + def test_unavailable_and_short_inventories_warn_even_without_a_hit(self): + rows = [self.row(376, [], True), self.row(377, ['elsewhere.py'], True)] + for options in ([], ['--offline']): + with self.subTest(options=options): + report = json.loads(self.run_cli(rows, '--path', 'missing.py', '--json', *options)) + coverage = report['coverage']['pr_files'] + self.assertTrue(coverage['complete_pagination']) + self.assertEqual(coverage['truncated_prs'], [376, 377]) + self.assertEqual(coverage['unverified_heads'], []) + self.assertEqual(coverage['unverified_bases'], []) + self.assertEqual(report['path_matches_total'], 0) + self.assertTrue(any('incomplete file inventories for 2 PRs' in warning + for warning in report['warnings'])) + + def test_result_limit_does_not_hide_incomplete_inventories(self): + rows = [self.row(98, ['elsewhere.py'], True), self.row(99, [], True), + self.row(100, ['target.py'], True), self.row(101, ['target.py'], False), + self.row(99, [], True)] + report = json.loads(self.run_cli(rows, '--path', 'target.py', '--limit', '1', '--json')) + self.assertEqual(report['path_matches_total'], 2) + self.assertEqual([row['number'] for row in report['path_matches']], [101]) + self.assertFalse(report['path_matches'][0]['truncated']) + self.assertEqual(report['coverage']['pr_files']['truncated_prs'], [98, 99, 100]) + self.assertTrue(any('incomplete file inventories for 3 PRs' in warning + for warning in report['warnings'])) + + def test_combined_term_and_path_matches_preserve_truncation(self): + rows = [self.row(100, ['target.py'], True), self.row(101, ['target.py'], False)] + report = json.loads(self.run_cli(rows, 'needle', '--path', 'target.py', '--json')) + self.assertEqual({row['number']: row['truncated'] for row in report['matches']}, + {100: True, 101: False}) + + def test_text_report_surfaces_unmatched_gaps_and_combined_match_flags(self): + rows = [self.row(376, [], True)] + output = self.run_cli(rows, '--path', 'missing.py') + self.assertIn('"truncated_prs": [376]', output) + self.assertIn('WARNING: pr_files: incomplete file inventories for 1 PRs', output) + output = self.run_cli([self.row(100, ['target.py'], True)], 'needle', '--path', 'target.py') + self.assertIn('[pr] needle (file list truncated)', output) + + def test_complete_empty_inventory_is_not_reported_as_a_gap(self): + report = json.loads(self.run_cli([self.row(376, [], False)], '--path', 'missing.py', '--json')) + self.assertEqual(report['coverage']['pr_files']['truncated_prs'], []) + self.assertEqual(report['warnings'], []) + self.assertEqual(report['path_matches_total'], 0) + + +if __name__ == '__main__': + unittest.main()