README.md

@@ -53,6 +53,7 @@ Any MCP-compatible agent (Antigravity, Claude Code, etc.) can call these tools n
 | `gitea_whoami` | Read-only: identify the authenticated Gitea account (safe metadata only) |
 | `gitea_get_profile` | Read-only: describe the active runtime execution profile (safe metadata only) |
 | `gitea_check_pr_eligibility` | Read-only: check if the current identity/profile may review/approve/request_changes/merge a PR |
+| `gitea_assess_conflict_fix_classification` | Read-only: classify conflict-fix need from a live PR head re-fetch before creating a conflict-fix worktree |
 | `gitea_submit_pr_review` | Gated review mutation: comment/approve/request_changes, only after identity+profile+eligibility gates pass (no merge, no self-approval) |
 | `gitea_mark_issue` | Claim/release an issue (start/done) |
 | `gitea_list_labels` | List all available labels in a repository |

conflict_fix_classification.py

@@ -0,0 +1,266 @@
+"""Live PR head re-pin before conflict-fix classification (#522).
+
+Open PR inventory fields (mergeable, head_sha) can be stale. Author sessions
+must re-fetch live PR state and pin the live head before classifying a PR as
+conflicted or creating a conflict-fix worktree.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+_FULL_SHA = re.compile(r"^[0-9a-f]{40}$", re.IGNORECASE)
+
+_INVENTORY_HEAD_RE = re.compile(
+    r"(?:inventory head sha|stale inventory head sha)\s*:\s*([0-9a-f]{40})",
+    re.IGNORECASE,
+)
+_LIVE_HEAD_RE = re.compile(
+    r"(?:live head sha|pinned head sha|live pr head sha)\s*:\s*([0-9a-f]{40})",
+    re.IGNORECASE,
+)
+_REPINS_TOOL_RE = re.compile(
+    r"gitea_assess_conflict_fix_classification",
+    re.IGNORECASE,
+)
+_CLASSIFICATION_RE = re.compile(
+    r"conflict[- ]fix classification\s*:\s*(\S+)",
+    re.IGNORECASE,
+)
+
+CLASSIFICATION_STALE_INVENTORY_SKIP = "stale_inventory_skip"
+CLASSIFICATION_CONFLICT_FIX_NEEDED = "conflict_fix_needed"
+CLASSIFICATION_LIVE_MERGEABLE = "live_mergeable_skip"
+CLASSIFICATION_INCOMPLETE = "incomplete_live_repin"
+
+_VALID_CLASSIFICATIONS = frozenset({
+    CLASSIFICATION_STALE_INVENTORY_SKIP,
+    CLASSIFICATION_CONFLICT_FIX_NEEDED,
+    CLASSIFICATION_LIVE_MERGEABLE,
+    CLASSIFICATION_INCOMPLETE,
+})
+
+
+def _normalize_sha(value: str | None) -> str | None:
+    text = (value or "").strip().lower()
+    if not text:
+        return None
+    return text if _FULL_SHA.match(text) else None
+
+
+def assess_conflict_fix_classification(
+    *,
+    pr_number: int,
+    inventory_head_sha: str | None = None,
+    inventory_mergeable: bool | None = None,
+    live_head_sha: str | None = None,
+    live_mergeable: bool | None = None,
+) -> dict[str, Any]:
+    """Classify whether conflict-fix work is justified from live PR state.
+
+    Inventory fields are advisory only. Live head + live mergeable are required
+    before any conflict-fix worktree may be created.
+    """
+    inv_head = _normalize_sha(inventory_head_sha)
+    live_head = _normalize_sha(live_head_sha)
+    reasons: list[str] = []
+
+    if not isinstance(pr_number, int) or pr_number <= 0:
+        return {
+            "classification": CLASSIFICATION_INCOMPLETE,
+            "worktree_allowed": False,
+            "skip_author_mutation": True,
+            "inventory_stale_head": False,
+            "inventory_stale_mergeable": False,
+            "pinned_head_sha": None,
+            "inventory_head_sha": inv_head,
+            "live_head_sha": live_head,
+            "inventory_mergeable": inventory_mergeable,
+            "live_mergeable": live_mergeable,
+            "pr_number": pr_number,
+            "reasons": ["pr_number must be a positive integer (fail closed)"],
+        }
+
+    if live_head is None:
+        reasons.append(
+            "live PR head SHA missing or not a full 40-char hex SHA; "
+            "re-fetch the PR before conflict-fix classification (fail closed)"
+        )
+    if live_mergeable is None:
+        reasons.append(
+            "live PR mergeable value missing; re-fetch the PR before "
+            "conflict-fix classification (fail closed)"
+        )
+
+    if reasons:
+        return {
+            "classification": CLASSIFICATION_INCOMPLETE,
+            "worktree_allowed": False,
+            "skip_author_mutation": True,
+            "inventory_stale_head": bool(
+                inv_head and live_head and inv_head != live_head
+            ),
+            "inventory_stale_mergeable": (
+                inventory_mergeable is not None
+                and live_mergeable is not None
+                and bool(inventory_mergeable) != bool(live_mergeable)
+            ),
+            "pinned_head_sha": live_head,
+            "inventory_head_sha": inv_head,
+            "live_head_sha": live_head,
+            "inventory_mergeable": inventory_mergeable,
+            "live_mergeable": live_mergeable,
+            "pr_number": pr_number,
+            "reasons": reasons,
+        }
+
+    inventory_stale_head = bool(inv_head and inv_head != live_head)
+    inventory_stale_mergeable = (
+        inventory_mergeable is not None
+        and bool(inventory_mergeable) != bool(live_mergeable)
+    )
+
+    # Live mergeable wins: do not start conflict-fix work.
+    if live_mergeable is True:
+        classification = (
+            CLASSIFICATION_STALE_INVENTORY_SKIP
+            if (
+                inventory_mergeable is False
+                or inventory_stale_head
+                or inventory_stale_mergeable
+            )
+            else CLASSIFICATION_LIVE_MERGEABLE
+        )
+        note = []
+        if inventory_stale_head:
+            note.append(
+                f"inventory head {inv_head} differs from live head {live_head}; "
+                "use live head only"
+            )
+        if inventory_mergeable is False:
+            note.append(
+                "inventory reported mergeable:false but live PR is mergeable:true; "
+                "skip author conflict-fix mutation"
+            )
+        return {
+            "classification": classification,
+            "worktree_allowed": False,
+            "skip_author_mutation": True,
+            "inventory_stale_head": inventory_stale_head,
+            "inventory_stale_mergeable": inventory_stale_mergeable,
+            "pinned_head_sha": live_head,
+            "inventory_head_sha": inv_head,
+            "live_head_sha": live_head,
+            "inventory_mergeable": inventory_mergeable,
+            "live_mergeable": live_mergeable,
+            "pr_number": pr_number,
+            "reasons": note,
+        }
+
+    # live_mergeable is False → conflict-fix may proceed on the pinned live head.
+    note = []
+    if inventory_stale_head:
+        note.append(
+            f"inventory head {inv_head} differs from live head {live_head}; "
+            "pin and use live head only for conflict-fix work"
+        )
+    if inventory_mergeable is True:
+        note.append(
+            "inventory reported mergeable:true but live PR is mergeable:false; "
+            "trust live mergeable and proceed only with live head pin"
+        )
+    note.append(
+        f"live PR #{pr_number} is mergeable:false at pinned head {live_head}; "
+        "conflict-fix worktree allowed for that head only"
+    )
+    return {
+        "classification": CLASSIFICATION_CONFLICT_FIX_NEEDED,
+        "worktree_allowed": True,
+        "skip_author_mutation": False,
+        "inventory_stale_head": inventory_stale_head,
+        "inventory_stale_mergeable": inventory_stale_mergeable,
+        "pinned_head_sha": live_head,
+        "inventory_head_sha": inv_head,
+        "live_head_sha": live_head,
+        "inventory_mergeable": inventory_mergeable,
+        "live_mergeable": live_mergeable,
+        "pr_number": pr_number,
+        "reasons": note,
+    }
+
+
+def assess_conflict_fix_classification_final_report(
+    report_text: str,
+    **_kwargs: Any,
+) -> dict[str, Any]:
+    """Require live-head re-pin proof when a report claims conflict-fix work (#522)."""
+    text = report_text or ""
+    lower = text.lower()
+    mentions_conflict_fix = (
+        "conflict-fix" in lower
+        or "conflict fix" in lower
+        or "conflict_fix" in lower
+    )
+    if not mentions_conflict_fix:
+        return {"proven": True, "reasons": [], "applicable": False}
+
+    reasons: list[str] = []
+    if not _REPINS_TOOL_RE.search(text):
+        reasons.append(
+            "conflict-fix report must cite gitea_assess_conflict_fix_classification "
+            "(live head re-pin tool)"
+        )
+
+    live_match = _LIVE_HEAD_RE.search(text)
+    if not live_match:
+        reasons.append(
+            "conflict-fix report must state Live head SHA: <40-char hex> "
+            "(or Pinned head SHA / Live PR head SHA)"
+        )
+    else:
+        live_sha = _normalize_sha(live_match.group(1))
+        if live_sha is None:
+            reasons.append("live head SHA is not a full 40-char hex digest")
+
+    inv_match = _INVENTORY_HEAD_RE.search(text)
+    # Inventory head is optional but recommended when classification is stale skip.
+    class_match = _CLASSIFICATION_RE.search(text)
+    if not class_match:
+        reasons.append(
+            "conflict-fix report must state Conflict-fix classification: "
+            f"<{'|'.join(sorted(_VALID_CLASSIFICATIONS))}>"
+        )
+    else:
+        classification = class_match.group(1).strip().lower().replace(" ", "_")
+        # allow hyphenated forms
+        classification = classification.replace("-", "_")
+        if classification not in _VALID_CLASSIFICATIONS:
+            reasons.append(
+                f"unknown conflict-fix classification {class_match.group(1)!r}; "
+                f"expected one of {sorted(_VALID_CLASSIFICATIONS)}"
+            )
+
+    if inv_match and live_match:
+        inv_sha = _normalize_sha(inv_match.group(1))
+        live_sha = _normalize_sha(live_match.group(1))
+        if inv_sha and live_sha and inv_sha != live_sha:
+            # Require explicit note that live head was used.
+            if "use live head" not in lower and "live head only" not in lower:
+                reasons.append(
+                    "inventory head differs from live head; report must state that "
+                    "the live head was used exclusively"
+                )
+
+    return {
+        "proven": not reasons,
+        "reasons": reasons,
+        "applicable": True,
+        "inventory_head_sha": _normalize_sha(inv_match.group(1)) if inv_match else None,
+        "live_head_sha": _normalize_sha(live_match.group(1)) if live_match else None,
+        "classification": (
+            class_match.group(1).strip().lower().replace("-", "_").replace(" ", "_")
+            if class_match
+            else None
+        ),
+    }

final_report_validator.py

@@ -598,6 +598,27 @@ def _rule_reviewer_stale_head_proof(report_text: str) -> list[dict[str, str]]:
     )
 
 
+def _rule_conflict_fix_classification_proof(report_text: str) -> list[dict[str, str]]:
+    from conflict_fix_classification import (
+        assess_conflict_fix_classification_final_report,
+    )
+
+    text = report_text or ""
+    result = assess_conflict_fix_classification_final_report(text)
+    if result.get("proven"):
+        return []
+    return _findings_from_reasons(
+        "author.conflict_fix_classification_proof",
+        result.get("reasons") or [],
+        field="Conflict-fix classification",
+        severity="block",
+        safe_next_action=(
+            "call gitea_assess_conflict_fix_classification, state the live head "
+            "SHA, and state the classification before creating a conflict-fix worktree"
+        ),
+    )
+
+
 def _rule_conflict_fix_push_proof(report_text: str) -> list[dict[str, str]]:
     from pr_work_lease import assess_conflict_fix_final_report
 
@@ -1393,6 +1414,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
         *_SHARED_ISSUE_LOCK_RULES,
         _rule_shared_issue_acceptance_gate,
         _rule_reviewer_vague_mutations_none,
+        _rule_conflict_fix_classification_proof,
         _rule_conflict_fix_push_proof,
         _rule_worktree_cleanup_audit_proof,
     ],

gitea_mcp_server.py

@@ -827,6 +827,7 @@ import reconciliation_workflow  # noqa: E402
 import audit_reconciliation_mode  # noqa: E402
 import review_merge_state_machine  # noqa: E402
 import pr_work_lease  # noqa: E402
+import conflict_fix_classification  # noqa: E402
 import native_mcp_preference  # noqa: E402
 import worktree_cleanup_audit  # noqa: E402
 
@@ -8247,6 +8248,40 @@ def gitea_acquire_conflict_fix_lease(
     }
 
 
+@mcp.tool()
+def gitea_assess_conflict_fix_classification(
+    pr_number: int,
+    inventory_head_sha: str | None = None,
+    inventory_mergeable: bool | None = None,
+    live_head_sha: str | None = None,
+    live_mergeable: bool | None = None,
+) -> dict:
+    """Read-only: classify conflict-fix need from live PR state (#522).
+
+    Open PR inventory can be stale. Callers must pass a live re-fetched PR head
+    and mergeability value before creating a conflict-fix worktree.
+    """
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "success": False,
+            "performed": False,
+            "reasons": read_block,
+            "permission_report": _permission_block_report("gitea.read"),
+        }
+
+    result = conflict_fix_classification.assess_conflict_fix_classification(
+        pr_number=pr_number,
+        inventory_head_sha=inventory_head_sha,
+        inventory_mergeable=inventory_mergeable,
+        live_head_sha=live_head_sha,
+        live_mergeable=live_mergeable,
+    )
+    result["success"] = True
+    result["performed"] = False
+    return result
+
+
 @mcp.tool()
 def gitea_assess_conflict_fix_push(
     pr_number: int,

review_proofs.py

@@ -5540,6 +5540,15 @@ def assess_already_landed_classification_report(report_text, **kwargs):
     return _assess(report_text, **kwargs)
 
 
+def assess_conflict_fix_classification_final_report(report_text, **kwargs):
+    """#522: require live PR head re-pin before conflict-fix classification."""
+    from conflict_fix_classification import (
+        assess_conflict_fix_classification_final_report as _assess,
+    )
+
+    return _assess(report_text, **kwargs)
+
+
 def assess_prior_blocker_skip_proof(report_text, **kwargs):
     """#318: require live blocker proof before skipping earlier open PRs."""
     from reviewer_blocker_skip import assess_prior_blocker_skip_proof as _assess

skills/llm-project-workflow/workflows/work-issue.md

@@ -615,10 +615,39 @@ After push, report:
 
 If push fails, stop and produce a recovery handoff.
 
-## 20A. Conflict-fix lease and push gate (#399)
+## 20A. Live head re-pin before conflict-fix classification (#522)
+
+Open PR inventory `mergeable` / `head_sha` fields are **advisory and may be
+stale**. Before classifying a PR as conflicted or creating a conflict-fix
+worktree:
+
+1. Re-fetch the live PR (`gitea_view_pr` or equivalent) and record:
+   * inventory head SHA (if any)
+   * inventory mergeable (if any)
+   * **live** head SHA (full 40-char)
+   * **live** mergeable
+2. Call `gitea_assess_conflict_fix_classification` with those values.
+3. Act only on the returned classification and **pinned live head**:
+   * `stale_inventory_skip` / `live_mergeable_skip` → **do not** create a
+     conflict-fix worktree; do not mutate the PR branch for conflicts.
+   * `conflict_fix_needed` → conflict-fix worktree allowed for the pinned
+     live head only.
+   * `incomplete_live_repin` → stop; re-fetch live state.
+4. If inventory head ≠ live head, report both SHAs and use the live head only.
+5. If inventory says `mergeable:false` but live says `mergeable:true`, classify
+   as stale inventory and skip author conflict-fix mutation.
+
+Conflict-fix final reports that claim conflict-fix work must include:
+
+* citation of `gitea_assess_conflict_fix_classification`
+* `Live head SHA: <40-char hex>` (or Pinned head SHA / Live PR head SHA)
+* `Conflict-fix classification: <stale_inventory_skip|conflict_fix_needed|live_mergeable_skip|incomplete_live_repin>`
+
+## 20B. Conflict-fix lease and push gate (#399)
 
 When pushing to an existing PR branch to resolve merge conflicts:
 
+0. Complete §20A live-head re-pin / classification first.
 1. Call `gitea_acquire_conflict_fix_lease` before any push.
 2. Call `gitea_assess_conflict_fix_push` immediately before `git push` with:
    * branch head before push

tests/test_conflict_fix_classification.py

@@ -0,0 +1,142 @@
+"""Tests for live PR head re-pin before conflict-fix classification (#522)."""
+
+from __future__ import annotations
+
+import unittest
+
+from conflict_fix_classification import (
+    CLASSIFICATION_CONFLICT_FIX_NEEDED,
+    CLASSIFICATION_INCOMPLETE,
+    CLASSIFICATION_LIVE_MERGEABLE,
+    CLASSIFICATION_STALE_INVENTORY_SKIP,
+    assess_conflict_fix_classification,
+    assess_conflict_fix_classification_final_report,
+)
+
+
+def _sha(prefix: str) -> str:
+    return (prefix + "0" * 40)[:40]
+
+
+class TestConflictFixClassification(unittest.TestCase):
+    def test_stale_inventory_mergeable_skip(self):
+        """PR #508 style: inventory mergeable:false/stale head, live mergeable:true."""
+        result = assess_conflict_fix_classification(
+            pr_number=508,
+            inventory_head_sha=_sha("dad1dc8"),
+            inventory_mergeable=False,
+            live_head_sha=_sha("3f3d6cb"),
+            live_mergeable=True,
+        )
+        self.assertEqual(result["classification"], CLASSIFICATION_STALE_INVENTORY_SKIP)
+        self.assertTrue(result["skip_author_mutation"])
+        self.assertFalse(result["worktree_allowed"])
+        self.assertTrue(result["inventory_stale_head"])
+        self.assertTrue(result["inventory_stale_mergeable"])
+        self.assertEqual(result["pinned_head_sha"], _sha("3f3d6cb"))
+
+    def test_stale_inventory_head_only_live_mergeable_true(self):
+        """PR #493 style: head moved; live still mergeable."""
+        result = assess_conflict_fix_classification(
+            pr_number=493,
+            inventory_head_sha=_sha("685e627"),
+            inventory_mergeable=True,
+            live_head_sha=_sha("4561d7a"),
+            live_mergeable=True,
+        )
+        self.assertEqual(result["classification"], CLASSIFICATION_STALE_INVENTORY_SKIP)
+        self.assertTrue(result["skip_author_mutation"])
+        self.assertFalse(result["worktree_allowed"])
+        self.assertTrue(result["inventory_stale_head"])
+        self.assertEqual(result["pinned_head_sha"], _sha("4561d7a"))
+
+    def test_live_conflict_allows_worktree_on_pinned_head(self):
+        result = assess_conflict_fix_classification(
+            pr_number=99,
+            inventory_head_sha=_sha("aaaaaaa"),
+            inventory_mergeable=False,
+            live_head_sha=_sha("bbbbbbb"),
+            live_mergeable=False,
+        )
+        self.assertEqual(result["classification"], CLASSIFICATION_CONFLICT_FIX_NEEDED)
+        self.assertTrue(result["worktree_allowed"])
+        self.assertFalse(result["skip_author_mutation"])
+        self.assertTrue(result["inventory_stale_head"])
+        self.assertEqual(result["pinned_head_sha"], _sha("bbbbbbb"))
+
+    def test_matching_inventory_live_mergeable_skip(self):
+        head = _sha("cccccccc")
+        result = assess_conflict_fix_classification(
+            pr_number=10,
+            inventory_head_sha=head,
+            inventory_mergeable=True,
+            live_head_sha=head,
+            live_mergeable=True,
+        )
+        self.assertEqual(result["classification"], CLASSIFICATION_LIVE_MERGEABLE)
+        self.assertFalse(result["worktree_allowed"])
+        self.assertTrue(result["skip_author_mutation"])
+        self.assertFalse(result["inventory_stale_head"])
+
+    def test_missing_live_head_incomplete(self):
+        result = assess_conflict_fix_classification(
+            pr_number=1,
+            inventory_head_sha=_sha("ddddddd"),
+            inventory_mergeable=False,
+            live_head_sha=None,
+            live_mergeable=False,
+        )
+        self.assertEqual(result["classification"], CLASSIFICATION_INCOMPLETE)
+        self.assertFalse(result["worktree_allowed"])
+        self.assertTrue(result["skip_author_mutation"])
+        self.assertTrue(any("live PR head" in r for r in result["reasons"]))
+
+    def test_short_sha_rejected(self):
+        result = assess_conflict_fix_classification(
+            pr_number=1,
+            inventory_head_sha="abc1234",
+            inventory_mergeable=False,
+            live_head_sha="def5678",
+            live_mergeable=False,
+        )
+        self.assertEqual(result["classification"], CLASSIFICATION_INCOMPLETE)
+        self.assertFalse(result["worktree_allowed"])
+
+
+class TestConflictFixClassificationFinalReport(unittest.TestCase):
+    def test_non_conflict_report_not_applicable(self):
+        result = assess_conflict_fix_classification_final_report(
+            "Implemented feature without merge issues."
+        )
+        self.assertTrue(result["proven"])
+        self.assertFalse(result["applicable"])
+
+    def test_conflict_report_requires_tool_live_head_classification(self):
+        result = assess_conflict_fix_classification_final_report(
+            "Did conflict-fix work on PR #99."
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["applicable"])
+        joined = " ".join(result["reasons"])
+        self.assertIn("gitea_assess_conflict_fix_classification", joined)
+        self.assertIn("Live head SHA", joined)
+        self.assertIn("classification", joined.lower())
+
+    def test_good_conflict_report_passes(self):
+        live = _sha("3f3d6cb")
+        inv = _sha("dad1dc8")
+        text = f"""
+Conflict-fix classification: stale_inventory_skip
+Called gitea_assess_conflict_fix_classification before worktree creation.
+Inventory head SHA: {inv}
+Live head SHA: {live}
+Use live head only; skip author mutation.
+"""
+        result = assess_conflict_fix_classification_final_report(text)
+        self.assertTrue(result["proven"], msg=result.get("reasons"))
+        self.assertEqual(result["live_head_sha"], live)
+        self.assertEqual(result["inventory_head_sha"], inv)
+
+
+if __name__ == "__main__":
+    unittest.main()