fix: judge each recaled citation at its own place

A sentence written in two fields was located over the whole file, found
twice and left in both: the writer now searches the lines of its own
scalar. A structured ref was judged pending when its old text appeared
anywhere, so a chain of shifts (tr1x cut2 508 to 531, cut3 531 to 552)
held the pin on the values the previous entries took: it is now looked
up at its own field.
This commit is contained in:
Abdessamad Derraz committed 2026-10-23 16:21:04 +02:00
1 parent 97c960484c
commit 2d48cde8d6
2 files changed
+139 -41

No files matched your search

+71 -20
View File
@@ -2627,14 +2627,57 @@ def _run_after_moves(
return f"{new_path or parts[0].path}:{','.join(ranges)}"
def _unique_line(lines: list[str], target: str) -> int | None:
"""The one physical line carrying target, None when not exactly one."""
hits = [i for i, line in enumerate(lines) if target and target in line]
def _unique_line(
lines: list[str], target: str, window: tuple[int, int]
) -> int | None:
"""The one line of the window carrying target, None when not exactly one."""
hits = [
i for i in range(window[0], min(window[1], len(lines) - 1) + 1)
if target and target in lines[i]
]
return hits[0] if len(hits) == 1 else None
def scalar_windows(
text: str, document: object
) -> dict[tuple[int, object], tuple[int, int]]:
"""Physical lines each scalar spans, keyed by its holder and key.
The same sentence can sit in two fields: emucorex cites
CMakeLists.txt:706 in notes and in exclusion_note, on identical lines.
Searched over the whole file neither occurrence was unique, and only
the node's own lines tell them apart.
"""
try:
root = yaml.compose(text, Loader=yaml.SafeLoader)
except yaml.YAMLError:
return {}
windows: dict[tuple[int, object], tuple[int, int]] = {}
def walk(value: object, node: yaml.Node | None) -> None:
if isinstance(value, dict) and isinstance(node, yaml.MappingNode):
by_key = {str(k.value): v for k, v in node.value}
for key, item in value.items():
child = by_key.get(str(key))
if child is None:
continue
windows[(id(value), key)] = (
child.start_mark.line, child.end_mark.line
)
walk(item, child)
elif isinstance(value, list) and isinstance(node, yaml.SequenceNode):
for index, (item, child) in enumerate(zip(value, node.value)):
windows[(id(value), index)] = (
child.start_mark.line, child.end_mark.line
)
walk(item, child)
walk(document, root)
return windows
def _apply_prose_edits(
text: str, jobs: list[tuple[Citation, dict]]
text: str, jobs: list[tuple[Citation, dict]], document: object = None
) -> tuple[str, list[str], list[str]]:
"""Rewrite prose citation tokens in place, nothing else.
@@ -2652,6 +2695,8 @@ def _apply_prose_edits(
holders: dict[tuple[int, object], tuple] = {}
applied: list[str] = []
left: list[str] = []
windows = scalar_windows(text, document) if document is not None else {}
whole = (0, len(lines) - 1)
for citation, moves in jobs:
replacements = []
@@ -2701,8 +2746,10 @@ def _apply_prose_edits(
if straddles:
left.append(f"{citation.field}: {citation.ref} (a token spans the line break)")
continue
window = windows.get((id(citation.holder), citation.key), whole)
located = {
number: _unique_line(lines, value_lines[number]) for number in by_line
number: _unique_line(lines, value_lines[number], window)
for number in by_line
}
if citation.wrapped and (
None in located.values()
@@ -2731,14 +2778,15 @@ def _apply_prose_edits(
rf"(?<![\w/.-]){re.escape(citation.ref)}(?![\w-])"
)
run_hits = [
i for i, line in enumerate(lines) if pattern.search(line)
i for i in range(window[0], min(window[1], len(lines) - 1) + 1)
if pattern.search(lines[i])
]
if (
len(run_hits) != 1
or len(pattern.findall(lines[run_hits[0]])) != 1
):
carriers = sum(
1 for line in lines
1 for line in lines[window[0]:window[1] + 1]
if value_lines[citation.line] and value_lines[citation.line] in line
)
left.append(
@@ -2981,7 +3029,7 @@ def rebase_refs(
citation.holder[citation.key] = new_ref
applied.append(f"{entry.source_ref} -> {new_ref}")
text, prose_applied, left = _apply_prose_edits(text, prose_jobs)
text, prose_applied, left = _apply_prose_edits(text, prose_jobs, document)
refused.extend(left)
if refused:
# A citation left on the pinned lines holds the pin, so writing the
@@ -3029,21 +3077,24 @@ def pending_recale(
document = yaml.safe_load(text) if text is not None else None
# A ref the writer regenerates is pending only while the file still
# carries it as written: a bump run without the recale left lrps2's
# SHIFTED refs on the old lines under the new pin.
refs_left = (
{c.ref for c in collect_citations(document) if c.kind == "ref"}
if document is not None else set()
)
# SHIFTED refs on the old lines under the new pin. It is looked up at
# its own field, never by its text alone: tr1x moved cut2 508->531 and
# cut3 531->552, and cut3 read as pending on cut2's new value.
refs_left: dict[tuple[str, str], int] = {}
if document is not None:
for citation in collect_citations(document):
if citation.kind == "ref":
for slot in ((citation.field, citation.ref), ("", citation.ref)):
refs_left[slot] = refs_left.get(slot, 0) + 1
pending = 0
for entry in report.entries or []:
if entry.kind == "prose":
continue
if (
_is_rewritable(entry.source_ref)
and not entry.name.endswith("]")
and entry.source_ref not in refs_left
):
continue
if _is_rewritable(entry.source_ref) and not entry.name.endswith("]"):
slot = (entry.field, entry.source_ref)
if not refs_left.get(slot):
continue
refs_left[slot] -= 1
pending += sum(
1
for part in entry.parts
@@ -3637,7 +3688,7 @@ def realign_prose(
for citation, moves in jobs
)
return messages
new_text, applied, left = _apply_prose_edits(text, jobs)
new_text, applied, left = _apply_prose_edits(text, jobs, document)
messages.extend(applied)
messages.extend(f"left in place: {item}" for item in left)
if applied:
+68 -21
View File
@@ -2984,6 +2984,34 @@ class TestRebaseWaitsForThePin(unittest.TestCase):
self.assertFalse(profile_sync.pending_recale(report))
self.assertEqual(rebase_refs(self.path, report), ["a.c:10-12 -> a.c:30-32"])
def test_chained_shifts_are_judged_at_their_own_entry(self):
# tr1x: cut2 508->531, cut3 531->552, cut4 552->575. Once written,
# cut3 and cut4 read as pending on the values cut2 and cut3 took.
text = (
'emulator: Test\nsource: "https://github.com/o/n"\n'
'source_commit: "pin"\nfiles:\n'
+ "".join(
f' - name: "cut{n}.mp3"\n source_ref: "g.json5:{line}"\n'
for n, line in ((2, 508), (3, 531), (4, 552))
)
)
self.path.write_text(text, encoding="utf-8")
entries = [
EntryReport(
f"cut{n}.mp3", f"g.json5:{old}", "SHIFTED",
[PartResult(RefPart("g.json5", old, old), "SHIFTED", None, new, new, [])],
"ref", f"files[cut{n}.mp3].source_ref",
)
for n, old, new in ((2, 508, 531), (3, 531, 552), (4, 552, 575))
]
report = self._report(entries, {"SHIFTED": 3})
self.assertEqual(len(rebase_refs(self.path, report)), 3)
written = self.path.read_text()
self.assertIn('source_ref: "g.json5:575"', written)
self.assertEqual(
profile_sync.pending_recale(report, text=written), 0
)
class TestBumpCommit(unittest.TestCase):
def setUp(self):
@@ -3722,6 +3750,13 @@ files:
source_ref: "a.c:10-12"
'''
# One run written twice in a folded scalar: no physical line carries the
# folded sentence, and the run alone is not unique.
FOLDED_TWICE = PROSE_SAMPLE.replace(
"files:",
"twice: >\n Read at (q.c:7) first\n and again at (q.c:7) later.\nfiles:",
)
class TestRebaseProse(unittest.TestCase):
def setUp(self):
@@ -3789,6 +3824,18 @@ class TestRebaseProse(unittest.TestCase):
self.assertIn("(a.c:10)", self.path.read_text())
def test_ambiguous_sentence_location_is_left_alone(self):
self.path.write_text(FOLDED_TWICE, encoding="utf-8")
part = PartResult(RefPart("q.c", 7, 7, "7"), "SHIFTED", None, 9, 9, [])
applied = rebase_refs(
self.path, self._report([self._prose_entry("q.c:7", [part], "twice")])
)
self.assertEqual(applied, [])
self.assertEqual(self.path.read_text(), FOLDED_TWICE)
def test_one_sentence_in_two_fields_moves_in_both(self):
# emucorex cites CMakeLists.txt:706 in notes and in exclusion_note on
# identical lines: neither was unique in the file, one was left, and
# the pin refused to follow.
text = PROSE_SAMPLE.replace(
"files:",
'quirks: |\n The loader appends dc/ to the system directory '
@@ -3796,35 +3843,37 @@ class TestRebaseProse(unittest.TestCase):
)
self.path.write_text(text, encoding="utf-8")
part = PartResult(RefPart("a.c", 10, 10, "10"), "SHIFTED", None, 14, 14, [])
applied = rebase_refs(
self.path, self._report([self._prose_entry("a.c:10", [part])])
report = self._report([
self._prose_entry("a.c:10", [part]),
self._prose_entry("a.c:10", [part], "quirks"),
])
self.assertEqual(
rebase_refs(self.path, report),
["notes: a.c:10 -> a.c:14", "quirks: a.c:10 -> a.c:14"],
)
self.assertEqual(applied, [])
self.assertEqual(self.path.read_text(), text)
written = self.path.read_text()
self.assertEqual(written, text.replace("(a.c:10)", "(a.c:14)"))
self.assertEqual(profile_sync.pending_recale(report, text=written), 0)
self.assertTrue(bump_commit(self.path, report))
def test_a_refused_recale_names_the_citation_and_writes_nothing(self):
# The structured ref could move, the prose could not: the run
# printed nothing, wrote nothing, and the pin stayed.
text = PROSE_SAMPLE.replace(
"files:",
'quirks: |\n The loader appends dc/ to the system directory '
'(a.c:10) and reads\n nothing else.\nfiles:',
)
self.path.write_text(text, encoding="utf-8")
prose = PartResult(RefPart("a.c", 10, 10, "10"), "SHIFTED", None, 14, 14, [])
self.path.write_text(FOLDED_TWICE, encoding="utf-8")
prose = PartResult(RefPart("q.c", 7, 7, "7"), "SHIFTED", None, 9, 9, [])
structured = PartResult(RefPart("a.c", 10, 12), "SHIFTED", None, 14, 16, [])
report = self._report([
self._prose_entry("a.c:10", [prose]),
self._prose_entry("q.c:7", [prose], "twice"),
EntryReport("a.bin", "a.c:10-12", "SHIFTED", [structured]),
])
err = io.StringIO()
with contextlib.redirect_stderr(err):
self.assertEqual(rebase_refs(self.path, report), [])
self.assertIn(
"p: not recaled, cannot move notes: a.c:10 (2 lines carry its sentence)",
"p: not recaled, cannot move twice: q.c:7 (0 lines carry its sentence)",
err.getvalue(),
)
self.assertEqual(self.path.read_text(), text)
self.assertEqual(self.path.read_text(), FOLDED_TWICE)
def test_folded_scalar_moves_through_its_run_token(self):
text = PROSE_SAMPLE.replace(
@@ -3911,14 +3960,12 @@ class TestPendingProse(unittest.TestCase):
self.assertTrue(bump_commit(self.path, report))
def test_unwritable_prose_blocks_the_pin(self):
text = PROSE_SAMPLE.replace(
"files:",
'quirks: |\n The loader appends dc/ to the system directory '
'(a.c:10) and reads\n nothing else.\nfiles:',
)
self.path.write_text(text, encoding="utf-8")
part = PartResult(RefPart("a.c", 10, 10, "10"), "SHIFTED", None, 14, 14, [])
self.path.write_text(FOLDED_TWICE, encoding="utf-8")
part = PartResult(RefPart("q.c", 7, 7, "7"), "SHIFTED", None, 9, 9, [])
report = self._report([part])
report.entries[0] = EntryReport(
"twice", "q.c:7", "SHIFTED", [part], "prose", "twice"
)
self.assertEqual(rebase_refs(self.path, report), [])
self.assertFalse(bump_commit(self.path, report))