feat: judge tag-pinned profiles on their own revision

scripts/profile_sync.py

@@ -541,6 +541,7 @@ class ProfileReport:
     name: str
     repo: str | None = None
     repos: list[str] = None
+    pinned_tag: str | None = None
     host: str | None = None
     pin: str | None = None
     pin_origin: str | None = None
@@ -607,6 +608,71 @@ def select_views(
     return views
 
 
+SELF_CHECK_CONTEXT = 2
+
+
+def verify_at_pin(part: RefPart, pin_lines, tokens) -> PartResult:
+    """Check a ref against its own revision instead of against HEAD.
+
+    A profile pinned to a superseded tag documents a program HEAD no longer
+    contains, so comparing the two says nothing. What can still be checked is
+    self-consistency: does the cited range carry the value the entry declares?
+    """
+    if pin_lines is None:
+        return PartResult(
+            part, "GONE", None, None, None, [], "absent at the pinned revision"
+        )
+    if part.start is None:
+        return PartResult(part, "ANCHORED", None, None, None, [])
+    if part.start > len(pin_lines):
+        return PartResult(
+            part, "GONE", None, None, None, [], "beyond the end of the file"
+        )
+    if not tokens:
+        return PartResult(part, "ANCHORED", None, None, None, [])
+    lo = max(0, part.start - 1 - SELF_CHECK_CONTEXT)
+    hi = min(len(pin_lines), (part.end or part.start) + SELF_CHECK_CONTEXT)
+    window = "\n".join(pin_lines[lo:hi]).lower()
+    if any(token in window for token in tokens):
+        return PartResult(part, "ANCHORED", None, None, None, [])
+    return PartResult(
+        part, "CHANGED", None, None, None, [],
+        "declared value not found at the pinned revision",
+    )
+
+
+def version_tag_candidates(core_version: str) -> list[str]:
+    """Tag spellings a declared core_version might use."""
+    version = str(core_version or "").strip()
+    if not version or " " in version:
+        return []
+    bare = version.lstrip("vV")
+    return list(dict.fromkeys([version, f"v{bare}", bare]))
+
+
+def detect_pinned_tag(
+    profile: dict, repo, pin: str, head: str, cache_dir: str, offline: bool
+) -> str | None:
+    """Tag the profile is pinned to, when the repository has moved past it.
+
+    A profile can document a frozen release line that lives as a tag inside a
+    still-developed repository. HEAD is then a different program: the refs
+    describe the tag and must never be recaled onto modern code. `pcsx2-legacy`
+    is pinned to v1.6.0 of PCSX2, where a plugin-era DEV9 ref resolves onto the
+    modern built-in DEV9, a different code path.
+
+    The declared `core_version` names the tag, which is looked up by name so
+    that a repository publishing nightly tags cannot hide an old release behind
+    pagination.
+    """
+    if pin == head:
+        return None
+    for tag in version_tag_candidates(profile.get("core_version")):
+        if upstream.tag_commit(repo, tag, cache_dir, offline) == pin:
+            return tag
+    return None
+
+
 def build_report(
     name: str, profile: dict, cache_dir: str, offline: bool = False
 ) -> ProfileReport:
@@ -641,6 +707,9 @@ def build_report(
     report.repos = [v.repo.slug for v in views]
     report.pin, report.pin_origin = primary.pin, primary.origin
     report.head = primary.head
+    report.pinned_tag = detect_pinned_tag(
+        profile, primary.repo, primary.pin, primary.head, cache_dir, offline
+    )
 
     # A profile carrying no source_ref still has a pin worth writing and a
     # version worth checking, so the revisions above are resolved first.
@@ -727,13 +796,22 @@ def build_report(
         slug = view.repo.slug if view is not primary else None
         return slug, upstream.raw_url(view.repo, view.head, actual), actual
 
-    staged = [
-        (entry_name, ref, [
-            anchor_part(part, fetch, rename_getter, describe, tokens)
-            for part in split_source_ref(ref)
-        ])
-        for entry_name, ref, tokens in refs
-    ]
+    if report.pinned_tag:
+        staged = [
+            (entry_name, ref, [
+                verify_at_pin(part, fetch(PIN, part.path), tokens)
+                for part in split_source_ref(ref)
+            ])
+            for entry_name, ref, tokens in refs
+        ]
+    else:
+        staged = [
+            (entry_name, ref, [
+                anchor_part(part, fetch, rename_getter, describe, tokens)
+                for part in split_source_ref(ref)
+            ])
+            for entry_name, ref, tokens in refs
+        ]
 
     shifts = dominant_shifts([p for _, _, parts in staged for p in parts])
     for entry_name, ref, parts in staged:
@@ -798,6 +876,11 @@ def format_report(report: ProfileReport, changed_only: bool = False) -> str:
         lines.append(
             f"  {report.pin[:7]} -> {report.head[:7]}   pin: {report.pin_origin}"
         )
+    if report.pinned_tag:
+        lines.append(
+            f"  pinned to tag {report.pinned_tag}: checked against its own "
+            "revision, not against HEAD"
+        )
     if report.skipped:
         lines.append(f"  skipped: {report.skipped}")
         return "\n".join(lines)
@@ -1139,6 +1222,8 @@ def rebase_refs(
     and the next comparison would read the pinned revision at line numbers
     that only make sense at HEAD.
     """
+    if report.pinned_tag:
+        return []
     statuses = REBASE_STATUSES + (("CHANGED",) if accept_changed else ())
     blocking = [s for s in REVIEW_STATUSES if s not in statuses]
     if any((report.counts or {}).get(s) for s in blocking):
@@ -1212,7 +1297,7 @@ def bump_commit(
     path: Path, report: ProfileReport, accept_changed: bool = False
 ) -> bool:
     """Advance source_commit to HEAD when nothing needs a read again."""
-    if report.skipped or not report.head:
+    if report.skipped or not report.head or report.pinned_tag:
         return False
     blocking = REVIEW_STATUSES if not accept_changed else (
         s for s in REVIEW_STATUSES if s != "CHANGED"

scripts/upstream.py

@@ -297,8 +297,11 @@ def resolve_commit_at(
 
 def _tags_url(repo: Repo) -> str:
     if repo.family == "gitlab":
-        return f"{repo.api_base}/projects/{_project(repo)}/repository/tags"
-    return f"{repo.api_base}/repos/{repo.slug}/tags"
+        return (
+            f"{repo.api_base}/projects/{_project(repo)}"
+            f"/repository/tags?per_page=100"
+        )
+    return f"{repo.api_base}/repos/{repo.slug}/tags?per_page=100"
 
 
 def list_tags(repo: Repo, cache_dir: str, offline: bool = False) -> list[str]:
@@ -325,6 +328,39 @@ def resolve_tag_commit(
     return None
 
 
+def tag_commit(
+    repo: Repo, tag: str, cache_dir: str, offline: bool = False
+) -> str | None:
+    """Commit a named tag points at, looked up directly.
+
+    Listing tags is paginated, and a repository publishing nightly tags pushes
+    an old release far past the first page, so the tag is asked for by name.
+    An annotated tag points at a tag object, which is dereferenced.
+    """
+    if repo.family != "github":
+        return resolve_tag_commit(repo, tag, cache_dir, offline)
+    quoted = urllib.parse.quote(tag)
+    payload = _api(
+        f"{repo.api_base}/repos/{repo.slug}/git/ref/tags/{quoted}",
+        cache_dir,
+        offline,
+    )
+    if not isinstance(payload, dict):
+        return None
+    obj = payload.get("object")
+    if not isinstance(obj, dict):
+        return None
+    if obj.get("type") != "tag":
+        return obj.get("sha")
+    annotated = _api(
+        f"{repo.api_base}/repos/{repo.slug}/git/tags/{obj.get('sha')}",
+        cache_dir,
+        offline,
+    )
+    target = annotated.get("object") if isinstance(annotated, dict) else None
+    return target.get("sha") if isinstance(target, dict) else None
+
+
 def _releases_url(repo: Repo) -> str:
     if repo.family == "gitlab":
         return f"{repo.api_base}/projects/{_project(repo)}/releases"

tests/test_profile_sync.py

@@ -285,6 +285,57 @@ def _ambiguous(path, old, candidates):
     )
 
 
+class TestVerifyAtPin(unittest.TestCase):
+    """A tag-pinned profile is judged on self-consistency, not on HEAD."""
+
+    LINES = ["pad", "pad", 'ROM_LOAD("bios.bin", CRC(deadbeef))', "pad"]
+
+    def test_declared_value_present(self):
+        part = RefPart("a.c", 3, 3, "a.c:3")
+        self.assertEqual(
+            profile_sync.verify_at_pin(part, self.LINES, ["deadbeef"]).status,
+            "ANCHORED",
+        )
+
+    def test_value_found_within_the_context_window(self):
+        part = RefPart("a.c", 1, 1, "a.c:1")
+        self.assertEqual(
+            profile_sync.verify_at_pin(part, self.LINES, ["deadbeef"]).status,
+            "ANCHORED",
+        )
+
+    def test_declared_value_absent(self):
+        part = RefPart("a.c", 1, 1, "a.c:1")
+        result = profile_sync.verify_at_pin(part, self.LINES, ["cafebabe"])
+        self.assertEqual(result.status, "CHANGED")
+        self.assertIn("pinned revision", result.reason)
+
+    def test_missing_file(self):
+        part = RefPart("a.c", 1, 1, "a.c:1")
+        self.assertEqual(
+            profile_sync.verify_at_pin(part, None, ["deadbeef"]).status, "GONE"
+        )
+
+    def test_line_beyond_the_file(self):
+        part = RefPart("a.c", 99, 99, "a.c:99")
+        self.assertEqual(
+            profile_sync.verify_at_pin(part, self.LINES, ["deadbeef"]).status, "GONE"
+        )
+
+    def test_a_ref_without_a_line_is_accepted(self):
+        part = RefPart("a.c", None, None, "a.c")
+        self.assertEqual(
+            profile_sync.verify_at_pin(part, self.LINES, ["deadbeef"]).status,
+            "ANCHORED",
+        )
+
+    def test_an_entry_declaring_nothing_is_accepted(self):
+        part = RefPart("a.c", 1, 1, "a.c:1")
+        self.assertEqual(
+            profile_sync.verify_at_pin(part, self.LINES, []).status, "ANCHORED"
+        )
+
+
 class TestExternalCitation(unittest.TestCase):
     def test_project_name_then_file(self):
         self.assertTrue(profile_sync.is_external_citation("munt ROMInfo.cpp"))
@@ -624,15 +675,29 @@ class TestBuildReport(unittest.TestCase):
             profile_sync.upstream.resolve_commit_at,
             profile_sync.upstream.compare,
             profile_sync.upstream.list_tree,
+            profile_sync.upstream.list_tags,
+            profile_sync.upstream.resolve_tag_commit,
+            profile_sync.upstream.tag_commit,
             profile_sync.upstream._http_json,
             profile_sync.upstream._http_text,
         )
+        self.tags: list[str] = []
+        self.tag_commits: dict[str, str] = {}
         # Any code path reaching the HTTP layer is a leak, not a slow test.
         def _no_network(url):
             raise AssertionError(f"test reached the network: {url}")
 
         profile_sync.upstream._http_json = _no_network
         profile_sync.upstream._http_text = _no_network
+        profile_sync.upstream.list_tags = (
+            lambda repo, cache, offline=False: self.tags
+        )
+        profile_sync.upstream.resolve_tag_commit = (
+            lambda repo, tag, cache, offline=False: self.tag_commits.get(tag)
+        )
+        profile_sync.upstream.tag_commit = (
+            lambda repo, tag, cache, offline=False: self.tag_commits.get(tag)
+        )
         profile_sync.upstream.list_tree = (
             lambda repo, sha, cache_dir, offline=False: (
                 sorted({path for _, path in self.files}), False
@@ -660,6 +725,9 @@ class TestBuildReport(unittest.TestCase):
             profile_sync.upstream.resolve_commit_at,
             profile_sync.upstream.compare,
             profile_sync.upstream.list_tree,
+            profile_sync.upstream.list_tags,
+            profile_sync.upstream.resolve_tag_commit,
+            profile_sync.upstream.tag_commit,
             profile_sync.upstream._http_json,
             profile_sync.upstream._http_text,
         ) = self._orig
@@ -813,6 +881,45 @@ class TestBuildReport(unittest.TestCase):
         self.assertEqual(part.status, "RENAMED")
         self.assertEqual(part.new_path, "deep/src/stv.c")
 
+    def _versioned(self, version):
+        profile = self._profile(["a.c:2"])
+        profile["core_version"] = version
+        return profile
+
+    def test_pin_on_the_declared_version_tag_is_flagged(self):
+        self.files[("pinsha", "a.c")] = ["x", "hit"]
+        self.files[("headsha", "a.c")] = ["x", "hit"]
+        self.tag_commits = {"v1.6.0": "pinsha"}
+        report = build_report("test", self._versioned("1.6.0"), self.dir)
+        self.assertEqual(report.pinned_tag, "v1.6.0")
+
+    def test_a_bare_version_spelling_is_tried(self):
+        self.files[("pinsha", "a.c")] = ["x", "hit"]
+        self.files[("headsha", "a.c")] = ["x", "hit"]
+        self.tag_commits = {"0.78": "pinsha"}
+        report = build_report("test", self._versioned("v0.78"), self.dir)
+        self.assertEqual(report.pinned_tag, "0.78")
+
+    def test_a_tag_pointing_elsewhere_is_not_flagged(self):
+        self.files[("pinsha", "a.c")] = ["x", "hit"]
+        self.files[("headsha", "a.c")] = ["x", "hit"]
+        self.tag_commits = {"v1.6.0": "somethingelse"}
+        report = build_report("test", self._versioned("1.6.0"), self.dir)
+        self.assertIsNone(report.pinned_tag)
+
+    def test_prose_version_is_not_looked_up(self):
+        self.assertEqual(profile_sync.version_tag_candidates("SVN (2015 snapshot)"), [])
+        self.assertEqual(profile_sync.version_tag_candidates(""), [])
+
+    def test_pin_equal_to_head_is_never_flagged(self):
+        self.files[("pinsha", "a.c")] = ["x", "hit"]
+        self.files[("headsha", "a.c")] = ["x", "hit"]
+        self.tag_commits = {"v1.6.0": "pinsha"}
+        profile = self._versioned("1.6.0")
+        profile["source_commit"] = "headsha"
+        report = build_report("test", profile, self.dir)
+        self.assertIsNone(report.pinned_tag)
+
     def test_missing_date_and_commit_is_skipped(self):
         report = build_report(
             "test",
@@ -1213,6 +1320,19 @@ class TestRebaseRefs(unittest.TestCase):
             [],
         )
 
+    def test_a_pin_on_a_superseded_tag_is_never_rebased(self):
+        part = PartResult(
+            RefPart("a.c", 10, 12, "a.c:10-12"), "RENAMED", "src/a.c", 20, 22, []
+        )
+        report = self._report(
+            [EntryReport("a.bin", "a.c:10-12", "RENAMED", [part])]
+        )
+        report.counts = {"RENAMED": 1}
+        report.pinned_tag = "v1.6.0"
+        self.assertEqual(rebase_refs(self.path, report), [])
+        self.assertEqual(rebase_refs(self.path, report, accept_changed=True), [])
+        self.assertIn('source_ref: "a.c:10-12"', self.path.read_text())
+
     def test_nothing_is_rebased_while_a_ref_needs_review(self):
         shifted = PartResult(RefPart("a.c", 10, 12), "SHIFTED", None, 20, 22, [])
         report = self._report(
@@ -1336,6 +1456,13 @@ class TestBumpCommit(unittest.TestCase):
         self.assertEqual(profile_sync.pending_recale(report), 0)
         self.assertTrue(bump_commit(self.path, report))
 
+    def test_refused_when_pinned_to_a_superseded_tag(self):
+        report = ProfileReport(
+            name="p", repo="o/n", pin="pin", head="newhead",
+            entries=[], counts={"ANCHORED": 3}, pinned_tag="v1.6.0",
+        )
+        self.assertFalse(bump_commit(self.path, report))
+
     def test_refused_while_a_changed_remains(self):
         report = ProfileReport(
             name="p", repo="o/n", pin="pin", head="newhead",
@@ -1363,6 +1490,7 @@ class TestCheckVersion(unittest.TestCase):
             profile_sync.upstream.latest_release,
             profile_sync.upstream.list_tags,
             profile_sync.upstream.resolve_tag_commit,
+            profile_sync.upstream.tag_commit,
         )
         profile_sync.upstream.latest_release = (
             lambda repo, cache, offline=False: profile_sync.upstream.Release(
@@ -1383,6 +1511,7 @@ class TestCheckVersion(unittest.TestCase):
             profile_sync.upstream.latest_release,
             profile_sync.upstream.list_tags,
             profile_sync.upstream.resolve_tag_commit,
+            profile_sync.upstream.tag_commit,
         ) = self._orig
         self.tmp.cleanup()