Disambiguate colliding reflect ledger tokens

desloppify/app/commands/plan/triage/runner/orchestrator_codex_pipeline_execution.py

@@ -18,6 +18,7 @@
 from ..validation.organize_policy import validate_organize_against_reflect_ledger
 from ..validation.reflect_accounting import (
     analyze_reflect_issue_accounting,
+    display_reflect_issue_tokens,
     validate_reflect_accounting,
 )
 from .codex_runner import TriageStageRunResult, run_triage_stage
@@ -288,10 +289,8 @@ def build_reflect_repair_prompt(
     stages_data: Mapping[str, Any] | None = None,
 ) -> str:
     """Build a targeted retry prompt for a reflect report that failed accounting."""
-    missing_short = ", ".join(issue_id.rsplit("::", 1)[-1] for issue_id in missing_ids) or "none"
-    duplicate_short = (
-        ", ".join(issue_id.rsplit("::", 1)[-1] for issue_id in duplicate_ids) or "none"
-    )
+    missing_tokens = ", ".join(display_reflect_issue_tokens(missing_ids)) or "none"
+    duplicate_tokens = ", ".join(display_reflect_issue_tokens(duplicate_ids)) or "none"
     base_prompt = build_stage_prompt_fn(
         "reflect",
         triage_input,
@@ -305,16 +304,16 @@ def build_reflect_repair_prompt(
         [
             base_prompt,
             "## Repair Pass",
-            "Your previous reflect report failed the exact-hash accounting check.",
-            f"Missing hashes: {missing_short}",
-            f"Duplicated hashes: {duplicate_short}",
+            "Your previous reflect report failed the exact-token accounting check.",
+            f"Missing tokens: {missing_tokens}",
+            f"Duplicated tokens: {duplicate_tokens}",
             "Rewrite the FULL reflect report so it passes validation.",
             "Requirements for this repair:",
             "- Start with a `## Coverage Ledger` section.",
-            '- Use one ledger line per issue hash: `- abcd1234 -> cluster "name"` or `- abcd1234 -> skip "reason"`.',
-            "- Mention every required hash exactly once in that ledger.",
-            "- Do not mention hashes anywhere else in the report.",
-            "- Preserve the same strategy unless fixing the missing/duplicate hashes forces a small adjustment.",
+            '- Use one ledger line per issue token: `- <token> -> cluster "name"` or `- <token> -> skip "reason"`.',
+            "- Mention every required token exactly once in that ledger.",
+            "- Do not mention those tokens anywhere else in the report.",
+            "- Preserve the same strategy unless fixing the missing/duplicate tokens forces a small adjustment.",
             "- Output only the corrected reflect report.",
             "## Previous Reflect Report",
             original_report,

desloppify/app/commands/plan/triage/runner/stage_prompts.py

@@ -16,6 +16,7 @@
 )
 
 from ..services import TriageServices, default_triage_services
+from ..validation.reflect_accounting import required_reflect_issue_tokens
 from .stage_prompts_instruction_blocks import _STAGE_INSTRUCTIONS
 from .stage_prompts_instruction_shared import (
     _STAGES,
@@ -36,10 +37,10 @@
 from .stage_prompts_validation import _validation_requirements
 
 
-def _required_issue_hashes(triage_input: TriageInput) -> list[str]:
-    """Return sorted short hashes for open review issues."""
+def _required_issue_tokens(triage_input: TriageInput) -> list[str]:
+    """Return the exact ledger token required for each open review issue."""
     review_issues = getattr(triage_input, "review_issues", getattr(triage_input, "open_issues", {}))
-    return sorted(issue_id.rsplit("::", 1)[-1] for issue_id in review_issues)
+    return required_reflect_issue_tokens(set(review_issues))
 
 
 def _compact_issue_summary(triage_input: TriageInput) -> str:
@@ -98,22 +99,24 @@ def _issue_context_for_stage(
     if stage in {"observe", "reflect"}:
         parts = ["## Issue Data\n\n" + build_triage_prompt(triage_input)]
         if stage == "reflect":
-            short_ids = _required_issue_hashes(triage_input)
+            issue_tokens = _required_issue_tokens(triage_input)
             parts.append(
-                "## Required Issue Hashes\n"
-                f"Total open review issues: {len(short_ids)}\n"
-                "Every one of these hashes must appear exactly once in your cluster/skip blueprint.\n"
-                "Do not repeat hashes outside that blueprint.\n"
-                + ", ".join(short_ids)
+                "## Required Issue Tokens\n"
+                f"Total open review issues: {len(issue_tokens)}\n"
+                "Every one of these tokens must appear exactly once in your cluster/skip blueprint.\n"
+                "For collided short IDs, the required token is the full issue ID shown below.\n"
+                "Do not repeat these tokens outside that blueprint.\n"
+                + ", ".join(issue_tokens)
             )
             parts.append(
                 "## Coverage Ledger Template\n"
                 "Your final report MUST contain a `## Coverage Ledger` section with one line per issue.\n"
                 "Allowed forms:\n"
-                '- `- abcd1234 -> cluster "cluster-name"`\n'
-                '- `- abcd1234 -> skip "specific-reason-tag"`\n'
-                "Do not mention hashes outside the `## Coverage Ledger` section.\n"
-                + "\n".join(f"- {short_id} -> TODO" for short_id in short_ids)
+                '- `- <token> -> cluster "cluster-name"`\n'
+                '- `- <token> -> skip "specific-reason-tag"`\n'
+                "Use the exact required token for each issue.\n"
+                "Do not mention those tokens outside the `## Coverage Ledger` section.\n"
+                + "\n".join(f"- {issue_token} -> TODO" for issue_token in issue_tokens)
             )
         return "\n\n".join(parts)
     summary = _compact_issue_summary(triage_input)

desloppify/app/commands/plan/triage/validation/reflect_accounting.py

@@ -47,18 +47,25 @@ class _IdResolutionMaps:
     short_id_buckets: dict[str, list[str]]
     short_hex_map: dict[str, str]
     slug_prefix_map: dict[str, str]
+    issue_tokens: dict[str, str]
 
 
 def _build_id_resolution_maps(valid_ids: set[str]) -> _IdResolutionMaps:
     short_id_buckets: dict[str, list[str]] = {}
     short_hex_map: dict[str, str] = {}
     slug_prefix_map: dict[str, str] = {}
+    short_id_counts: Counter[str] = Counter(
+        issue_id.rsplit("::", 1)[-1]
+        for issue_id in valid_ids
+    )
+    issue_tokens: dict[str, str] = {}
     ambiguous_slugs: set[str] = set()
     for issue_id in sorted(valid_ids):
         parts = issue_id.rsplit("::", 1)
         short_id = parts[-1]
         slug = parts[0] if len(parts) == 2 else ""
         short_id_buckets.setdefault(short_id, []).append(issue_id)
+        issue_tokens[issue_id] = short_id if short_id_counts[short_id] == 1 else issue_id
         if re.fullmatch(r"[0-9a-f]{8,}", short_id):
             existing = short_hex_map.get(short_id)
             if existing is None:
@@ -78,9 +85,22 @@ def _build_id_resolution_maps(valid_ids: set[str]) -> _IdResolutionMaps:
         short_id_buckets=short_id_buckets,
         short_hex_map=short_hex_map,
         slug_prefix_map=slug_prefix_map,
+        issue_tokens=issue_tokens,
     )
 
 
+def required_reflect_issue_tokens(valid_ids: set[str]) -> list[str]:
+    """Return the exact ledger token required for each reflect issue."""
+    maps = _build_id_resolution_maps(valid_ids)
+    return [maps.issue_tokens[issue_id] for issue_id in sorted(valid_ids)]
+
+
+def display_reflect_issue_tokens(issue_ids: list[str]) -> list[str]:
+    """Return stable display tokens for issue IDs in user-facing messages."""
+    maps = _build_id_resolution_maps(set(issue_ids))
+    return [maps.issue_tokens.get(issue_id, issue_id) for issue_id in issue_ids]
+
+
 def _clean_ledger_token(raw: str) -> str:
     token = raw.strip().strip("`").strip()
     if token.startswith("[") and token.endswith("]"):
@@ -142,16 +162,12 @@ def _resolve_token_to_id(
     token: str,
     valid_ids: set[str],
     maps: _IdResolutionMaps,
-    short_id_usage: Counter[str],
 ) -> str | None:
     if token in valid_ids:
         return token
     bucket = maps.short_id_buckets.get(token)
-    if bucket:
-        bucket_index = short_id_usage[token]
-        resolved = bucket[bucket_index] if bucket_index < len(bucket) else bucket[-1]
-        short_id_usage[token] += 1
-        return resolved
+    if bucket and len(bucket) == 1:
+        return bucket[0]
     for hex_token in re.findall(r"[0-9a-f]{8,}", token):
         resolved = maps.short_hex_map.get(hex_token)
         if resolved:
@@ -215,9 +231,8 @@ def _resolve_ledger_issue_id(
     line: str,
     valid_ids: set[str],
     maps: _IdResolutionMaps,
-    short_id_usage: Counter[str],
 ) -> str | None:
-    issue_id = _resolve_token_to_id(token, valid_ids, maps, short_id_usage)
+    issue_id = _resolve_token_to_id(token, valid_ids, maps)
     if issue_id:
         return issue_id
     for hex_token in re.findall(r"[0-9a-f]{8,}", line):
@@ -253,7 +268,6 @@ def _walk_coverage_ledger(
     maps = _build_id_resolution_maps(valid_ids)
     hits: Counter[str] = Counter()
     dispositions: list[ReflectDisposition] = []
-    short_id_usage: Counter[str] = Counter()
     found_section, ledger_lines = _iter_coverage_ledger_lines(report)
 
     for line in ledger_lines:
@@ -266,7 +280,6 @@ def _walk_coverage_ledger(
             line=line,
             valid_ids=valid_ids,
             maps=maps,
-            short_id_usage=short_id_usage,
         )
         if not issue_id:
             continue
@@ -345,24 +358,26 @@ def validate_reflect_accounting(
     if not missing and not duplicates:
         return True, cited, missing, duplicates
 
+    maps = _build_id_resolution_maps(valid_ids)
+
     print(
         colorize(
             "  Reflect report must account for every open review issue exactly once.",
             "red",
         )
     )
     if missing:
-        missing_short = ", ".join(issue_id.rsplit("::", 1)[-1] for issue_id in missing[:10])
-        print(colorize(f"    Missing: {missing_short}", "yellow"))
+        missing_tokens = ", ".join(maps.issue_tokens.get(issue_id, issue_id) for issue_id in missing[:10])
+        print(colorize(f"    Missing: {missing_tokens}", "yellow"))
     if duplicates:
-        duplicate_short = ", ".join(issue_id.rsplit("::", 1)[-1] for issue_id in duplicates[:10])
-        print(colorize(f"    Duplicated: {duplicate_short}", "yellow"))
+        duplicate_tokens = ", ".join(maps.issue_tokens.get(issue_id, issue_id) for issue_id in duplicates[:10])
+        print(colorize(f"    Duplicated: {duplicate_tokens}", "yellow"))
     print(colorize("  Fix the reflect blueprint before running organize.", "dim"))
     if missing:
         print(colorize("  Expected format — include a ## Coverage Ledger section:", "dim"))
-        print(colorize('    - <hash> -> cluster "cluster-name"', "dim"))
-        print(colorize('    - <hash> -> skip "reason"', "dim"))
-        print(colorize("  Also accepted: bare hashes, colon-separated, comma-separated.", "dim"))
+        print(colorize('    - <token> -> cluster "cluster-name"', "dim"))
+        print(colorize('    - <token> -> skip "reason"', "dim"))
+        print(colorize("  Use the exact required ledger token for each issue.", "dim"))
     return False, cited, missing, duplicates
 
 

desloppify/tests/commands/plan/test_reflect_disposition_ledger.py

@@ -130,6 +130,43 @@ def test_bracket_wrapped_ids(self):
         result = parse_reflect_dispositions(report, valid_ids)
         assert len(result) == 1
 
+    def test_ambiguous_short_ids_require_disambiguated_tokens(self):
+        report = (
+            "## Coverage Ledger\n"
+            '- review::src/alpha.py::arch::shared_util -> cluster "alpha-fixes"\n'
+            '- review::src/beta.py::arch::shared_util -> cluster "beta-fixes"\n'
+        )
+        valid_ids = {
+            "review::src/beta.py::arch::shared_util",
+            "review::src/alpha.py::arch::shared_util",
+        }
+        result = parse_reflect_dispositions(report, valid_ids)
+        assert result == [
+            ReflectDisposition(
+                issue_id="review::src/alpha.py::arch::shared_util",
+                decision="cluster",
+                target="alpha-fixes",
+            ),
+            ReflectDisposition(
+                issue_id="review::src/beta.py::arch::shared_util",
+                decision="cluster",
+                target="beta-fixes",
+            ),
+        ]
+
+    def test_ambiguous_short_ids_are_not_resolved_by_order(self):
+        report = (
+            "## Coverage Ledger\n"
+            '- shared_util -> cluster "alpha-fixes"\n'
+            '- shared_util -> cluster "beta-fixes"\n'
+        )
+        valid_ids = {
+            "review::src/beta.py::arch::shared_util",
+            "review::src/alpha.py::arch::shared_util",
+        }
+        result = parse_reflect_dispositions(report, valid_ids)
+        assert result == []
+
 
 # ---------------------------------------------------------------------------
 # validate_organize_against_reflect_ledger

desloppify/tests/commands/plan/test_triage_runner.py

@@ -62,7 +62,7 @@ def test_build_reflect_prompt_includes_prior(tmp_path: Path) -> None:
     prompt = build_stage_prompt("reflect", si, prior, repo_root=tmp_path)
     assert "REFLECT" in prompt
     assert "My observation report" in prompt
-    assert "## Required Issue Hashes" in prompt
+    assert "## Required Issue Tokens" in prompt
     assert "## Coverage Ledger Template" in prompt
     assert "-> TODO" in prompt
     assert "exactly once" in prompt
@@ -188,8 +188,8 @@ def test_validate_reflect_issue_accounting_handles_short_id_collisions() -> None
     }
     report = """
 ## Coverage Ledger
-- review_packet_ownership_split -> cluster "review-packet-lifecycle-ownership"
-- review_packet_ownership_split -> cluster "review-packet-lifecycle-ownership"
+- review::src/a.py::cross_module_architecture::review_packet_ownership_split -> cluster "review-packet-lifecycle-ownership"
+- review::src/b.py::high_level_elegance::review_packet_ownership_split -> cluster "review-packet-lifecycle-ownership"
 
 ## Cluster Blueprint
 Cluster "review-packet-lifecycle-ownership" owns packet lifecycle policy.
@@ -204,6 +204,63 @@ def test_validate_reflect_issue_accounting_handles_short_id_collisions() -> None
     assert duplicates == []
 
 
+def test_validate_reflect_issue_accounting_rejects_ambiguous_short_id_collisions() -> None:
+    valid_ids = {
+        "review::src/a.py::cross_module_architecture::review_packet_ownership_split",
+        "review::src/b.py::high_level_elegance::review_packet_ownership_split",
+    }
+    report = """
+## Coverage Ledger
+- review_packet_ownership_split -> cluster "review-packet-lifecycle-ownership"
+- review_packet_ownership_split -> cluster "review-packet-lifecycle-ownership"
+"""
+    ok, cited, missing, duplicates = _validate_reflect_issue_accounting(
+        report=report,
+        valid_ids=valid_ids,
+    )
+    assert ok is False
+    assert cited == set()
+    assert missing == sorted(valid_ids)
+    assert duplicates == []
+
+
+def test_build_reflect_prompt_uses_full_ids_for_colliding_short_ids(tmp_path: Path) -> None:
+    issues = {
+        "review::src/a.py::cross_module_architecture::review_packet_ownership_split": {
+            "status": "open",
+            "detector": "review",
+            "file": "src/a.py",
+            "summary": "Issue A summary",
+            "detail": {"dimension": "cross_module_architecture", "suggestion": "Fix it"},
+        },
+        "review::src/b.py::high_level_elegance::review_packet_ownership_split": {
+            "status": "open",
+            "detector": "review",
+            "file": "src/b.py",
+            "summary": "Issue B summary",
+            "detail": {"dimension": "high_level_elegance", "suggestion": "Fix it"},
+        },
+    }
+    si = TriageInput(
+        review_issues=issues,
+        objective_backlog_issues={},
+        existing_clusters={},
+        dimension_scores={},
+        new_since_last=set(),
+        resolved_since_last=set(),
+        previously_dismissed=[],
+        triage_version=1,
+        resolved_issues={},
+        completed_clusters=[],
+    )
+
+    prompt = build_stage_prompt("reflect", si, {"observe": "obs"}, repo_root=tmp_path)
+
+    assert "## Required Issue Tokens" in prompt
+    assert "- review::src/a.py::cross_module_architecture::review_packet_ownership_split -> TODO" in prompt
+    assert "- review::src/b.py::high_level_elegance::review_packet_ownership_split -> TODO" in prompt
+
+
 def test_build_organize_prompt(tmp_path: Path) -> None:
     si = _make_triage_input()
     prior = {"observe": "obs", "reflect": "ref"}

desloppify/tests/commands/plan/test_triage_split_modules_direct.py

@@ -1617,8 +1617,8 @@ def test_pipeline_execution_helpers_cover_leaf_paths(monkeypatch, tmp_path: Path
         duplicate_ids=["review::src/b.py::facefeed"],
         build_stage_prompt_fn=lambda *_a, **_k: "base prompt",
     )
-    assert "Missing hashes: deadbeef" in prompt
-    assert "Duplicated hashes: facefeed" in prompt
+    assert "Missing tokens: deadbeef" in prompt
+    assert "Duplicated tokens: facefeed" in prompt
     assert "Previous Reflect Report" in prompt
 
     ok, reason = orchestrator_pipeline_execution_mod.preflight_stage(