Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions docs/research/coordination.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
14 changes: 11 additions & 3 deletions scripts/find_overlap.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"]}
Expand All @@ -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 = []
Expand All @@ -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.")
Expand All @@ -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:
Expand Down
92 changes: 92 additions & 0 deletions tests/test_research_pr_inventory_coverage.py
Original file line number Diff line number Diff line change
@@ -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()