feat: anchor on declared values when refs miss

scripts/profile_sync.py

@@ -171,6 +171,20 @@ def collect_tokens(entry: dict) -> list[str]:
     return tokens
 
 
+def _anchor_tokens(entry: dict) -> list[str]:
+    """Values worth searching for when a cited line misses its subject.
+
+    Archive entries are named `stvbios.zip` while the driver source writes the
+    set name alone, so the stem is searched as well as the full name.
+    """
+    tokens = collect_tokens(entry)
+    name = os.path.basename(str(entry.get("name", ""))).lower()
+    stem = name.rsplit(".", 1)[0]
+    if stem and stem not in tokens and len(stem) > 2:
+        tokens.append(stem)
+    return tokens
+
+
 def worst_status(statuses) -> str:
     """Severity of an entry is the worst severity among its parts."""
     worst = "ANCHORED"
@@ -255,6 +269,36 @@ def anchor_block(
     return _map_changed(pin, head, lo, hi)
 
 
+NUDGE_WINDOW = 60
+
+
+def nudge_to_declared(
+    pin_lines: list[str], start: int, end: int, tokens
+) -> tuple[int, int] | None:
+    """Move a cited range onto the nearest line carrying a declared value.
+
+    Some refs sit a few lines off their subject, and a few land on a blank
+    line, which anchors nothing. When the entry declares a hash or a name and
+    that value appears close by, the ref plainly meant that line.
+    """
+    if not tokens:
+        return None
+    lowered = [line.lower() for line in pin_lines]
+    hits = [
+        i + 1
+        for i, line in enumerate(lowered)
+        if any(token in line for token in tokens)
+    ]
+    if not hits:
+        return None
+    nearest = min(hits, key=lambda n: abs(n - start))
+    if abs(nearest - start) > NUDGE_WINDOW or nearest == start:
+        return None
+    if sum(1 for h in hits if abs(h - nearest) <= 1) > 1:
+        return None
+    return nearest, nearest + (end - start)
+
+
 def resolve_rename(
     result: CompareResult, path: str, head_paths=()
 ) -> tuple[str | None, list[str]]:
@@ -298,7 +342,9 @@ def _narrow(matches: list[str], path: str) -> list[str]:
     return same if len(same) == 1 else matches
 
 
-def anchor_part(part: RefPart, fetch, rename_getter, describe=None) -> PartResult:
+def anchor_part(
+    part: RefPart, fetch, rename_getter, describe=None, tokens=()
+) -> PartResult:
     """Locate one reference part at HEAD, following a rename when needed.
 
     `describe(path)` returns (repo slug, raw URL at HEAD) for the repository
@@ -364,7 +410,16 @@ def anchor_part(part: RefPart, fetch, rename_getter, describe=None) -> PartResul
             part, "GONE", None, None, None, [], "pin revision missing", slug, url
         )
 
-    anchored = anchor_block(pin_lines, head_lines, part.start, part.end or part.start)
+    end = part.end or part.start
+    anchored = anchor_block(pin_lines, head_lines, part.start, end)
+    note = anchored.reason
+    if anchored.status in REVIEW_STATUSES:
+        nudged = nudge_to_declared(pin_lines, part.start, end, tokens)
+        if nudged is not None:
+            retry = anchor_block(pin_lines, head_lines, *nudged)
+            if retry.status not in REVIEW_STATUSES:
+                anchored = retry
+                note = f"cited line was {nudged[0] - part.start:+d} off its subject"
     status = anchored.status
     if renamed and status in ("ANCHORED", "SHIFTED"):
         status = "RENAMED"
@@ -375,7 +430,7 @@ def anchor_part(part: RefPart, fetch, rename_getter, describe=None) -> PartResul
         anchored.start,
         anchored.end,
         anchored.candidates,
-        anchored.reason,
+        note,
         slug,
         url,
     )
@@ -518,7 +573,11 @@ def build_report(
     report = ProfileReport(name=name, entries=[], counts={})
 
     refs = [
-        (entry.get("name", "") + (f" [{label}]" if label else ""), value)
+        (
+            entry.get("name", "") + (f" [{label}]" if label else ""),
+            value,
+            _anchor_tokens(entry),
+        )
         for entry in (profile.get("files") or [])
         if isinstance(entry, dict) and entry.get("source_ref")
         for label, value in source_ref_values(entry.get("source_ref"))
@@ -629,10 +688,10 @@ def build_report(
 
     staged = [
         (entry_name, ref, [
-            anchor_part(part, fetch, rename_getter, describe)
+            anchor_part(part, fetch, rename_getter, describe, tokens)
             for part in split_source_ref(ref)
         ])
-        for entry_name, ref in refs
+        for entry_name, ref, tokens in refs
     ]
 
     shifts = dominant_shifts([p for _, _, parts in staged for p in parts])

tests/test_profile_sync.py

@@ -280,7 +280,64 @@ def _part(path, old, new, status):
 
 
 def _ambiguous(path, old, candidates):
-    return PartResult(RefPart(path, old, old), "AMBIGUOUS", None, None, None, candidates)
+    return PartResult(
+        RefPart(path, old, old), "AMBIGUOUS", None, None, None, candidates
+    )
+
+
+class TestAnchorTokens(unittest.TestCase):
+    def test_archive_stem_is_searched_too(self):
+        tokens = profile_sync._anchor_tokens({"name": "stvbios.zip"})
+        self.assertIn("stvbios.zip", tokens)
+        self.assertIn("stvbios", tokens)
+
+    def test_hashes_still_win(self):
+        tokens = profile_sync._anchor_tokens(
+            {"name": "a.zip", "sha1": "AABB" * 10}
+        )
+        self.assertIn("aabb" * 10, tokens)
+
+    def test_very_short_stem_is_skipped(self):
+        self.assertNotIn("ab", profile_sync._anchor_tokens({"name": "ab.zip"}))
+
+
+class TestNudgeToDeclared(unittest.TestCase):
+    LINES = ["pad"] * 20 + ["ROM_START( stvbios )"] + ["pad"] * 20
+
+    def test_moves_onto_the_declared_value(self):
+        self.assertEqual(
+            profile_sync.nudge_to_declared(self.LINES, 19, 19, ["stvbios"]),
+            (21, 21),
+        )
+
+    def test_range_length_is_kept(self):
+        self.assertEqual(
+            profile_sync.nudge_to_declared(self.LINES, 19, 22, ["stvbios"]),
+            (21, 24),
+        )
+
+    def test_nothing_when_already_on_target(self):
+        self.assertIsNone(
+            profile_sync.nudge_to_declared(self.LINES, 21, 21, ["stvbios"])
+        )
+
+    def test_nothing_beyond_the_window(self):
+        lines = ["pad"] * 500 + ["ROM_START( stvbios )"]
+        self.assertIsNone(profile_sync.nudge_to_declared(lines, 1, 1, ["stvbios"]))
+
+    def test_nothing_without_tokens(self):
+        self.assertIsNone(profile_sync.nudge_to_declared(self.LINES, 19, 19, []))
+
+    def test_nothing_when_the_value_is_absent(self):
+        self.assertIsNone(
+            profile_sync.nudge_to_declared(self.LINES, 19, 19, ["absent"])
+        )
+
+    def test_adjacent_duplicates_are_refused(self):
+        lines = ["pad"] * 20 + ["stvbios a", "stvbios b"] + ["pad"] * 20
+        self.assertIsNone(
+            profile_sync.nudge_to_declared(lines, 19, 19, ["stvbios"])
+        )
 
 
 class TestDominantShift(unittest.TestCase):
@@ -315,7 +372,9 @@ class TestDominantShift(unittest.TestCase):
                          "AMBIGUOUS")
 
     def test_range_length_is_preserved(self):
-        part = PartResult(RefPart("a.c", 100, 104), "AMBIGUOUS", None, None, None, [120])
+        part = PartResult(
+            RefPart("a.c", 100, 104), "AMBIGUOUS", None, None, None, [120]
+        )
         settled = profile_sync.resolve_by_shift(part, {"a.c": 20})
         self.assertEqual((settled.start, settled.end), (120, 124))