Merge branch 'master' into fix/issue-685-side-effect-free-resolver

Resolve gitea_resolve_task_capability conflict by keeping both #685
runtime_reconnect_required blocker and #702 stale_binding_recovery
annotation. Keep resolve-path stale-binding assessment report-only
(auto_recover=False) so capability resolution never mutates env or
session state. Preserve #725 merger-lease and #728 PR-sync code from
master. Regression: resolve uses report-only binding assessment.

gitea_mcp_server.py

@@ -1524,6 +1524,7 @@ import review_merge_state_machine  # noqa: E402
 import pr_work_lease  # noqa: E402
 import workflow_skill  # noqa: E402
 import conflict_fix_classification  # noqa: E402
+import pr_sync_status  # noqa: E402  # PR sync / update-by-merge lifecycle
 import native_mcp_preference  # noqa: E402
 import branch_cleanup_guard  # noqa: E402
 import thread_state_ledger_validator  # noqa: E402
@@ -14205,6 +14206,777 @@ def gitea_assess_conflict_fix_classification(
     return result
 
 
+def _count_commits_behind(
+    base_url: str,
+    auth: dict,
+    *,
+    pr_head_sha: str,
+    base_head_sha: str,
+) -> int | None:
+    """Return how many base commits are missing from the PR head, if knowable."""
+    # compare old...new returns commits reachable from new not from old.
+    # pr_head...base_head ≈ commits on base not in PR (behind count).
+    try:
+        cmp_url = (
+            f"{base_url}/compare/{pr_head_sha}...{base_head_sha}"
+        )
+        data = api_request("GET", cmp_url, auth) or {}
+        total = data.get("total_commits")
+        if isinstance(total, int) and total >= 0:
+            return total
+        commits = data.get("commits")
+        if isinstance(commits, list):
+            return len(commits)
+    except Exception:
+        pass
+    return None
+
+
+def _branch_protection_requires_current_base(
+    base_url: str,
+    auth: dict,
+    *,
+    base_branch: str,
+) -> bool | None:
+    """Read Gitea branch protection ``block_on_outdated_branch`` when present."""
+    branch = (base_branch or "").strip()
+    if not branch:
+        return None
+    try:
+        # Prefer the named protection rule matching the base branch.
+        protections = api_request(
+            "GET", f"{base_url}/branch_protections", auth
+        ) or []
+        if isinstance(protections, list):
+            for rule in protections:
+                if not isinstance(rule, dict):
+                    continue
+                name = (rule.get("branch_name") or rule.get("rule_name") or "").strip()
+                # Exact match or glob-ish contains for common patterns.
+                if name == branch or name in ("*", f"{branch}"):
+                    if "block_on_outdated_branch" in rule:
+                        return bool(rule.get("block_on_outdated_branch"))
+            # Fallback: any protection that mentions the base branch.
+            for rule in protections:
+                if not isinstance(rule, dict):
+                    continue
+                name = (rule.get("branch_name") or rule.get("rule_name") or "").strip()
+                if branch in name or name.endswith(branch):
+                    if "block_on_outdated_branch" in rule:
+                        return bool(rule.get("block_on_outdated_branch"))
+        # Branch payload may embed effective protection.
+        br = api_request("GET", f"{base_url}/branches/{branch}", auth) or {}
+        prot = br.get("protection") or br.get("effective_branch_protection") or {}
+        if isinstance(prot, dict) and "block_on_outdated_branch" in prot:
+            return bool(prot.get("block_on_outdated_branch"))
+    except Exception:
+        return None
+    return None
+
+
+def _prove_author_ownership_for_pr(
+    *,
+    pr_number: int,
+    pr_title: str | None,
+    pr_body: str | None,
+    source_branch: str | None,
+    remote: str,
+    host: str | None,
+    org: str,
+    repo: str,
+    worktree_path: str | None = None,
+) -> dict:
+    """Prove author ownership for PR mutations without requiring issue==pr.
+
+    Issue #727 / PR #728: tracking issues and PR numbers often differ. Ownership
+    is proven via linked issues (title/body/branch), a live session lock for one
+    of those issues (or the PR number itself), a durable issue lock for a linked
+    issue, and optional branch/worktree binding on the lock. Unrelated issue
+    locks fail closed.
+    """
+    reasons: list[str] = []
+    text = f"{pr_title or ''}\n{pr_body or ''}"
+    linked = list(
+        extract_linked_issue_numbers(text, branch_name=source_branch)
+    )
+    # Always allow legacy same-number ownership when the lock is for the PR index.
+    candidates = sorted(set(linked + [int(pr_number)]))
+    remote_key = remote if remote in REMOTES else (host or "unknown")
+    matched_issue: int | None = None
+    matched_via: str | None = None
+    lock_record: dict | None = None
+
+    # 1) Session lock: active issue work lease for this session.
+    try:
+        session = issue_lock_store.read_session_issue_lock()
+    except Exception:
+        session = None
+    if session and issue_lock_store.is_lease_live(session):
+        try:
+            sess_issue = int(session.get("issue_number") or 0)
+        except (TypeError, ValueError):
+            sess_issue = 0
+        if sess_issue in candidates:
+            # Optional branch binding: when both sides declare a branch, they must match.
+            lock_branch = str(session.get("branch_name") or "").strip()
+            src = (source_branch or "").strip()
+            if lock_branch and src and lock_branch != src:
+                reasons.append(
+                    f"session lock for issue #{sess_issue} is bound to branch "
+                    f"'{lock_branch}', not PR source branch '{src}' (fail closed)"
+                )
+            else:
+                wt = (worktree_path or "").strip()
+                lock_wt = str(session.get("worktree_path") or "").strip()
+                if wt and lock_wt:
+                    try:
+                        same_wt = os.path.realpath(wt) == os.path.realpath(lock_wt)
+                    except Exception:
+                        same_wt = wt == lock_wt
+                    if not same_wt:
+                        reasons.append(
+                            f"session lock for issue #{sess_issue} is bound to a "
+                            "different worktree (fail closed)"
+                        )
+                    else:
+                        matched_issue = sess_issue
+                        matched_via = "session_lock"
+                        lock_record = dict(session)
+                else:
+                    matched_issue = sess_issue
+                    matched_via = "session_lock"
+                    lock_record = dict(session)
+        elif sess_issue:
+            reasons.append(
+                f"session lock issue #{sess_issue} is not linked to PR #{pr_number} "
+                f"(linked={linked or 'none'}; fail closed)"
+            )
+
+    # 2) Durable per-issue locks for each ownership candidate.
+    if matched_issue is None:
+        for issue_n in candidates:
+            try:
+                durable = issue_lock_store.load_issue_lock(
+                    remote=remote_key,
+                    org=org,
+                    repo=repo,
+                    issue_number=issue_n,
+                )
+            except Exception:
+                durable = None
+            if not durable or not issue_lock_store.is_lease_live(durable):
+                continue
+            lock_branch = str(durable.get("branch_name") or "").strip()
+            src = (source_branch or "").strip()
+            if lock_branch and src and lock_branch != src:
+                reasons.append(
+                    f"durable lock for issue #{issue_n} is bound to branch "
+                    f"'{lock_branch}', not PR source '{src}'"
+                )
+                continue
+            matched_issue = issue_n
+            matched_via = "durable_lock"
+            lock_record = dict(durable)
+            break
+
+    # 3) Branch-owned live lock (any issue) matching the PR source branch.
+    if matched_issue is None and (source_branch or "").strip():
+        try:
+            by_branch = issue_lock_store.find_live_lock_for_branch(
+                str(source_branch).strip()
+            )
+        except Exception:
+            by_branch = None
+        if by_branch:
+            try:
+                branch_issue = int(by_branch.get("issue_number") or 0)
+            except (TypeError, ValueError):
+                branch_issue = 0
+            if branch_issue in candidates:
+                matched_issue = branch_issue
+                matched_via = "branch_lock"
+                lock_record = dict(by_branch)
+            elif branch_issue:
+                reasons.append(
+                    f"live branch lock for '{source_branch}' belongs to unrelated "
+                    f"issue #{branch_issue} (not linked to PR #{pr_number}; fail closed)"
+                )
+
+    proven = matched_issue is not None and lock_record is not None
+    if not proven and not reasons:
+        reasons.append(
+            "no live author issue lock proven for PR "
+            f"#{pr_number} via linked issues {linked or '[]'}, session lock, "
+            "durable lock, or branch lock (fail closed; issue_number need not "
+            "equal pr_number when the tracking issue is linked)"
+        )
+    return {
+        "proven": proven,
+        "has_author_lock": proven,
+        "linked_issues": linked,
+        "ownership_candidates": candidates,
+        "matched_issue": matched_issue,
+        "matched_via": matched_via,
+        "lock_record": lock_record,
+        "reasons": reasons,
+    }
+
+
+@mcp.tool()
+def gitea_assess_pr_sync_status(
+    pr_number: int,
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    prepared_verdict_head_sha: str | None = None,
+    branch_protection_requires_current_base: bool | None = None,
+    checks_status: str | None = None,
+) -> dict:
+    """Read-only: assess whether an approved/open PR is merge-ready or needs sync.
+
+    Distinguishes ``merge_now``, ``update_branch_by_merge``,
+    ``author_conflict_remediation``, ``fresh_review_required``, and ``blocked``.
+
+    Pins exact PR head and live base head. Never mutates. Never returns tokens.
+    """
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "success": False,
+            "performed": False,
+            "recommended_next_action": pr_sync_status.ACTION_BLOCKED,
+            "reasons": read_block,
+            "permission_report": _permission_block_report("gitea.read"),
+        }
+
+    try:
+        h, o, r = _resolve(remote, host, org, repo)
+    except Exception as exc:
+        return {
+            "success": False,
+            "performed": False,
+            "recommended_next_action": pr_sync_status.ACTION_BLOCKED,
+            "reasons": [f"repository identity resolve failed: {_redact(str(exc))}"],
+        }
+
+    auth = _auth(h)
+    if not auth:
+        return {
+            "success": False,
+            "performed": False,
+            "recommended_next_action": pr_sync_status.ACTION_BLOCKED,
+            "reasons": ["credentials unavailable (fail closed)"],
+            "host": h,
+            "org": o,
+            "repo": r,
+        }
+
+    base = repo_api_url(h, o, r)
+    try:
+        pr = api_request("GET", f"{base}/pulls/{pr_number}", auth) or {}
+    except Exception as exc:
+        return {
+            "success": False,
+            "performed": False,
+            "recommended_next_action": pr_sync_status.ACTION_BLOCKED,
+            "reasons": [f"PR details could not be retrieved: {_redact(str(exc))}"],
+            "host": h,
+            "org": o,
+            "repo": r,
+            "pr_number": pr_number,
+        }
+
+    pr_state = (pr.get("state") or "").strip().lower() or None
+    head_obj = pr.get("head") or {}
+    base_obj = pr.get("base") or {}
+    pr_head_sha = (head_obj.get("sha") or "").strip() or None
+    source_branch = (head_obj.get("ref") or "").strip() or None
+    base_branch = (base_obj.get("ref") or "").strip() or "master"
+    base_head_sha = (base_obj.get("sha") or "").strip() or None
+    mergeable = pr.get("mergeable")
+    # Gitea sometimes exposes conflict via mergeable=false only.
+    has_conflicts = False if mergeable is True else (True if mergeable is False else None)
+
+    # Prefer live base branch tip over the PR's stored base SHA (may lag).
+    try:
+        br = api_request("GET", f"{base}/branches/{base_branch}", auth) or {}
+        live_base = ((br.get("commit") or {}).get("id") or "").strip()
+        if live_base:
+            base_head_sha = live_base
+    except Exception:
+        pass
+
+    commits_behind = None
+    if pr_head_sha and base_head_sha:
+        commits_behind = _count_commits_behind(
+            base, auth, pr_head_sha=pr_head_sha, base_head_sha=base_head_sha
+        )
+        if commits_behind is None:
+            # Fail closed on unknown behind-count only when SHAs differ.
+            commits_behind = 0 if pr_head_sha == base_head_sha else None
+
+    if branch_protection_requires_current_base is None:
+        branch_protection_requires_current_base = (
+            _branch_protection_requires_current_base(
+                base, auth, base_branch=base_branch
+            )
+        )
+    # When protection cannot be read, fail closed by requiring current base
+    # whenever the PR is behind (safer for protected repos). Callers may pass
+    # an explicit False when policy is known not to require it.
+    if branch_protection_requires_current_base is None:
+        if commits_behind is not None and commits_behind > 0:
+            branch_protection_requires_current_base = True
+        else:
+            branch_protection_requires_current_base = False
+
+    approval_at_current_head = None
+    try:
+        feedback = gitea_get_pr_review_feedback(
+            pr_number=pr_number, remote=remote, host=host, org=org, repo=repo,
+        )
+        if feedback.get("success"):
+            approval_at_current_head = bool(feedback.get("approval_at_current_head"))
+        else:
+            approval_at_current_head = False
+    except Exception:
+        approval_at_current_head = None
+
+    # Locks / leases (best-effort; absence is reported as None/False).
+    # Ownership uses linked tracking issue / branch / session lock — not only
+    # issue_number == pr_number (#727 / PR #728).
+    active_author_lock = None
+    try:
+        ownership = _prove_author_ownership_for_pr(
+            pr_number=pr_number,
+            pr_title=pr.get("title"),
+            pr_body=pr.get("body"),
+            source_branch=source_branch,
+            remote=remote,
+            host=host,
+            org=o,
+            repo=r,
+            worktree_path=None,
+        )
+        active_author_lock = bool(ownership.get("has_author_lock"))
+    except Exception:
+        active_author_lock = None
+
+    active_reviewer_lease = None
+    active_merger_lease = None
+    try:
+        comments = api_request(
+            "GET", f"{base}/issues/{pr_number}/comments", auth
+        ) or []
+        if isinstance(comments, list):
+            rev = pr_work_lease.find_active_reviewer_lease(
+                comments, pr_number=pr_number
+            )
+            active_reviewer_lease = bool(rev)
+            # Merger lease comments share reviewer_pr_lease session store when present.
+            try:
+                sess = reviewer_pr_lease.get_session_lease() or {}
+                active_merger_lease = bool(
+                    sess.get("active")
+                    and int(sess.get("pr_number") or 0) == int(pr_number)
+                    and (sess.get("phase") or "").startswith("merger")
+                )
+            except Exception:
+                active_merger_lease = False
+    except Exception:
+        pass
+
+    if checks_status is None:
+        checks_status = "unknown"
+        # Combined status on PR head when Actions/status API is available.
+        if pr_head_sha:
+            try:
+                st = api_request(
+                    "GET",
+                    f"{base}/commits/{pr_head_sha}/status",
+                    auth,
+                ) or {}
+                state = (st.get("state") or "").strip().lower()
+                if state:
+                    checks_status = state
+                elif not st:
+                    checks_status = "none"
+            except Exception:
+                checks_status = "unknown"
+
+    assessment = pr_sync_status.assess_pr_sync_status(
+        host=h,
+        org=o,
+        repo=r,
+        pr_number=pr_number,
+        pr_state=pr_state,
+        source_branch=source_branch,
+        pr_head_sha=pr_head_sha,
+        base_head_sha=base_head_sha,
+        commits_behind=commits_behind,
+        mergeable=mergeable,
+        has_conflicts=has_conflicts,
+        branch_protection_requires_current_base=branch_protection_requires_current_base,
+        approval_at_current_head=approval_at_current_head,
+        checks_status=checks_status,
+        active_author_lock=active_author_lock,
+        active_reviewer_lease=active_reviewer_lease,
+        active_merger_lease=active_merger_lease,
+        prepared_verdict_head_sha=prepared_verdict_head_sha,
+    )
+    assessment["remote"] = remote if remote in REMOTES else None
+    assessment["base_branch"] = base_branch
+    assessment["success"] = True
+    assessment["performed"] = False
+    return assessment
+
+
+@mcp.tool()
+def gitea_update_pr_branch_by_merge(
+    pr_number: int,
+    expected_pr_head_sha: str,
+    expected_base_head_sha: str,
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    worktree_path: str | None = None,
+) -> dict:
+    """Author-only: merge live base into the PR branch (Gitea Update-by-merge).
+
+    Pins both expected PR head and expected base head; fails closed on either
+    race. Never rebases or force-pushes. On conflicts, returns a structured
+    author-remediation handoff without partial remote mutation.
+
+    Requires author profile, existing issue/PR lock, PR worktree under
+    branches/, clean control checkout, and master parity.
+    """
+    profile = get_profile()
+    allowed = profile.get("allowed_operations") or []
+    forbidden = profile.get("forbidden_operations") or []
+    role = _role_kind(allowed, forbidden)
+
+    # Permission: author branch push / PR mutation surface.
+    push_block = _profile_operation_gate("gitea.branch.push")
+    if push_block:
+        return {
+            "success": False,
+            "performed": False,
+            "mutation_allowed": False,
+            "reasons": push_block,
+            "permission_report": _permission_block_report("gitea.branch.push"),
+            "role_kind": role,
+        }
+
+    if role != "author":
+        pre = pr_sync_status.assess_update_pr_branch_preflight(
+            role_kind=role,
+            expected_pr_head_sha=expected_pr_head_sha,
+            live_pr_head_sha=expected_pr_head_sha,
+            expected_base_head_sha=expected_base_head_sha,
+            live_base_head_sha=expected_base_head_sha,
+            style="merge",
+        )
+        pre["success"] = False
+        return pre
+
+    try:
+        h, o, r = _resolve(remote, host, org, repo)
+    except Exception as exc:
+        return {
+            "success": False,
+            "performed": False,
+            "mutation_allowed": False,
+            "reasons": [f"repository identity resolve failed: {_redact(str(exc))}"],
+            "role_kind": role,
+        }
+
+    # Workspace / control / parity gates.
+    try:
+        _verify_role_mutation_workspace(
+            remote, worktree_path=worktree_path, task="push_branch"
+        )
+    except Exception as exc:
+        return {
+            "success": False,
+            "performed": False,
+            "mutation_allowed": False,
+            "reasons": [f"workspace gate failed: {_redact(str(exc))}"],
+            "role_kind": role,
+            "host": h,
+            "org": o,
+            "repo": r,
+            "pr_number": pr_number,
+        }
+
+    control_clean = True
+    try:
+        porcelain = _get_workspace_porcelain(PROJECT_ROOT)
+        # Untracked-only dirt is ignored; tracked edits contaminate control.
+        tracked_dirty = [
+            line for line in (porcelain or "").splitlines()
+            if line.strip() and not line.startswith("??")
+        ]
+        control_clean = not bool(tracked_dirty)
+    except Exception:
+        control_clean = False
+
+    master_ok = True
+    try:
+        stale = _master_parity_block("gitea.branch.push")
+        master_ok = not bool(stale)
+    except Exception:
+        master_ok = False
+
+    wt = (worktree_path or "").strip() or (
+        os.environ.get("GITEA_AUTHOR_WORKTREE")
+        or os.environ.get("GITEA_ACTIVE_WORKTREE")
+        or ""
+    ).strip()
+
+    auth = _auth(h)
+    if not auth:
+        return {
+            "success": False,
+            "performed": False,
+            "mutation_allowed": False,
+            "reasons": ["credentials unavailable (fail closed)"],
+            "role_kind": role,
+            "host": h,
+            "org": o,
+            "repo": r,
+            "pr_number": pr_number,
+        }
+
+    base = repo_api_url(h, o, r)
+    try:
+        pr = api_request("GET", f"{base}/pulls/{pr_number}", auth) or {}
+    except Exception as exc:
+        return {
+            "success": False,
+            "performed": False,
+            "mutation_allowed": False,
+            "reasons": [f"PR details could not be retrieved: {_redact(str(exc))}"],
+            "role_kind": role,
+            "host": h,
+            "org": o,
+            "repo": r,
+            "pr_number": pr_number,
+        }
+
+    if (pr.get("state") or "").strip().lower() != "open":
+        return {
+            "success": False,
+            "performed": False,
+            "mutation_allowed": False,
+            "reasons": [
+                f"PR is not open (state={pr.get('state')}); refuse update"
+            ],
+            "role_kind": role,
+            "host": h,
+            "org": o,
+            "repo": r,
+            "pr_number": pr_number,
+        }
+
+    head_obj = pr.get("head") or {}
+    base_obj = pr.get("base") or {}
+    live_pr_head = (head_obj.get("sha") or "").strip() or None
+    source_branch = (head_obj.get("ref") or "").strip() or None
+    base_branch = (base_obj.get("ref") or "").strip() or "master"
+    live_base_head = (base_obj.get("sha") or "").strip() or None
+    try:
+        br = api_request("GET", f"{base}/branches/{base_branch}", auth) or {}
+        tip = ((br.get("commit") or {}).get("id") or "").strip()
+        if tip:
+            live_base_head = tip
+    except Exception:
+        pass
+
+    mergeable = pr.get("mergeable")
+    has_conflicts = False if mergeable is True else (True if mergeable is False else None)
+
+    owns_branch = True
+    if wt and source_branch:
+        try:
+            # Best-effort: worktree branch should match PR source branch.
+            import subprocess as _sp
+            out = _sp.check_output(
+                ["git", "-C", wt, "rev-parse", "--abbrev-ref", "HEAD"],
+                text=True,
+                stderr=_sp.DEVNULL,
+            ).strip()
+            if out and out != source_branch:
+                owns_branch = False
+        except Exception:
+            owns_branch = None  # unknown — do not hard-fail solely on probe
+
+    # Prove ownership via linked issue lock / session / branch — not issue==pr.
+    ownership = _prove_author_ownership_for_pr(
+        pr_number=pr_number,
+        pr_title=pr.get("title"),
+        pr_body=pr.get("body"),
+        source_branch=source_branch,
+        remote=remote,
+        host=host,
+        org=o,
+        repo=r,
+        worktree_path=wt or None,
+    )
+    has_lock = bool(ownership.get("has_author_lock"))
+
+    preflight = pr_sync_status.assess_update_pr_branch_preflight(
+        role_kind=role,
+        expected_pr_head_sha=expected_pr_head_sha,
+        live_pr_head_sha=live_pr_head,
+        expected_base_head_sha=expected_base_head_sha,
+        live_base_head_sha=live_base_head,
+        has_conflicts=has_conflicts,
+        mergeable=mergeable,
+        style="merge",
+        has_author_lock=has_lock,
+        worktree_path=wt or None,
+        worktree_owns_source_branch=owns_branch,
+        control_checkout_clean=control_clean,
+        master_parity_ok=master_ok,
+        force_push=False,
+        rebase=False,
+    )
+    if not has_lock and ownership.get("reasons"):
+        # Surface ownership diagnostics beyond the generic preflight line.
+        preflight.setdefault("reasons", [])
+        for reason in ownership["reasons"]:
+            if reason not in preflight["reasons"]:
+                preflight["reasons"].append(reason)
+    preflight["ownership"] = {
+        "linked_issues": ownership.get("linked_issues"),
+        "matched_issue": ownership.get("matched_issue"),
+        "matched_via": ownership.get("matched_via"),
+    }
+    preflight["host"] = h
+    preflight["org"] = o
+    preflight["repo"] = r
+    preflight["pr_number"] = pr_number
+    preflight["source_branch"] = source_branch
+    preflight["base_branch"] = base_branch
+    preflight["worktree_path"] = wt or None
+    preflight["profile_name"] = profile.get("profile_name")
+
+    if not preflight.get("mutation_allowed"):
+        preflight["success"] = False
+        preflight["performed"] = False
+        if preflight.get("author_remediation_handoff"):
+            preflight["recommended_next_action"] = (
+                pr_sync_status.ACTION_AUTHOR_CONFLICT_REMEDIATION
+            )
+        return preflight
+
+    # Native Gitea update: POST .../pulls/{index}/update?style=merge
+    update_url = f"{base}/pulls/{pr_number}/update?style=merge"
+    try:
+        api_request("POST", update_url, auth)
+    except Exception as exc:
+        msg = _redact(str(exc)).lower()
+        conflictish = any(
+            token in msg
+            for token in ("conflict", "409", "merge conflict", "not mergeable")
+        )
+        if conflictish:
+            return {
+                "success": False,
+                "performed": False,
+                "mutation_allowed": False,
+                "author_remediation_handoff": True,
+                "recommended_next_action": (
+                    pr_sync_status.ACTION_AUTHOR_CONFLICT_REMEDIATION
+                ),
+                "reasons": [
+                    "Gitea refused update-by-merge due to conflicts; "
+                    "no partial remote update applied",
+                    _redact(str(exc)),
+                ],
+                "host": h,
+                "org": o,
+                "repo": r,
+                "pr_number": pr_number,
+                "expected_pr_head_sha": expected_pr_head_sha,
+                "expected_base_head_sha": expected_base_head_sha,
+                "live_pr_head_sha_before": live_pr_head,
+                "style": "merge",
+                "role_kind": role,
+            }
+        return {
+            "success": False,
+            "performed": False,
+            "mutation_allowed": False,
+            "reasons": [
+                f"update-by-merge API failed (fail closed): {_redact(str(exc))}"
+            ],
+            "host": h,
+            "org": o,
+            "repo": r,
+            "pr_number": pr_number,
+            "style": "merge",
+            "role_kind": role,
+        }
+
+    # Re-read new head after successful update.
+    new_head = None
+    try:
+        pr_after = api_request("GET", f"{base}/pulls/{pr_number}", auth) or {}
+        new_head = ((pr_after.get("head") or {}).get("sha") or "").strip() or None
+        mergeable_after = pr_after.get("mergeable")
+    except Exception:
+        pr_after = {}
+        mergeable_after = None
+
+    transition = pr_sync_status.assess_post_update_head_transition(
+        former_pr_head_sha=live_pr_head,
+        new_pr_head_sha=new_head,
+        former_approval_head_sha=live_pr_head,
+        prepared_verdict_head_sha=live_pr_head,
+    )
+
+    return {
+        "success": True,
+        "performed": True,
+        "mutation_allowed": True,
+        "style": "merge",
+        "force_push": False,
+        "rebase": False,
+        "host": h,
+        "org": o,
+        "repo": r,
+        "pr_number": pr_number,
+        "source_branch": source_branch,
+        "base_branch": base_branch,
+        "former_pr_head_sha": live_pr_head,
+        "new_pr_head_sha": new_head,
+        "base_head_sha": live_base_head,
+        "mergeable_after": mergeable_after,
+        "approval_invalidated_for_former_head": transition.get(
+            "approval_invalidated", True
+        ),
+        "reviewer_lease_superseded": transition.get("reviewer_lease_superseded"),
+        "merger_lease_superseded": transition.get("merger_lease_superseded"),
+        "prepared_verdict_invalidated": transition.get(
+            "prepared_verdict_invalidated"
+        ),
+        "recommended_next_action": transition.get(
+            "recommended_next_action",
+            pr_sync_status.ACTION_FRESH_REVIEW_REQUIRED,
+        ),
+        "transition": transition,
+        "role_kind": role,
+        "profile_name": profile.get("profile_name"),
+        "worktree_path": wt or None,
+        "reasons": list(transition.get("reasons") or []) + [
+            "update-by-merge completed via native Gitea API (style=merge only)"
+        ],
+    }
+
+
 @mcp.tool()
 def gitea_assess_conflict_fix_push(
     pr_number: int,
@@ -14701,36 +15473,18 @@ def gitea_route_task_session(
     )
 
 
-_restart_triggered = False
-
-
-def _trigger_mcp_auto_restart():
-    global _restart_triggered
-    if _restart_triggered or _preflight_in_test_mode():
-        return
-    _restart_triggered = True
-
-    config_path = os.environ.get(
-        "MCP_CONFIG_PATH",
-        os.path.expanduser("~/.gemini/config/mcp_config.json")
-    )
-
-    try:
-        if os.path.exists(config_path):
-            os.utime(config_path, None)
-    except Exception:
-        pass
-
-    import threading
-    import time
-    def delayed_exit():
-        time.sleep(1.0)
-        os._exit(0)
-    threading.Thread(target=delayed_exit, daemon=True).start()
+# #685: resolver stale-runtime detection is report-only. Config touch / os._exit
+# self-recovery was removed from the read-only path (was _trigger_mcp_auto_restart).
+# Recovery is owned exclusively by the IDE/client reconnect path.
 
 
 def _check_mcp_runtimes_diagnostics(task: str, matching_profiles: list[str]) -> list[str]:
-    """Check running runtimes and return errors if they are missing or stale."""
+    """Read-only: report missing or stale MCP runtimes (no config or process mutation).
+
+    #685: Never touches MCP client config, never spawns recovery threads, never
+    calls ``os._exit``. Stale detection remains fail-closed via returned reasons
+    only; the IDE/client owns reconnect/reload.
+    """
     import subprocess
     import re
     from datetime import datetime
@@ -14810,11 +15564,13 @@ def _check_mcp_runtimes_diagnostics(task: str, matching_profiles: list[str]) ->
             }
 
     if self_stale:
-        _trigger_mcp_auto_restart()
+        # #685: report-only — no config utime, no thread, no os._exit.
         reasons.append(
-            "stale-runtime: The active Gitea MCP server process is stale (running code from before changes were merged). "
-            "Auto-restart has been triggered: touched mcp_config.json to reload the daemon. "
-            "The current process will cleanly exit shortly."
+            "stale-runtime: The active Gitea MCP server process is stale "
+            "(running code from before changes were merged). "
+            "Reconnect the IDE/client-managed MCP namespace for this profile "
+            "so it reloads current master. The resolver does not touch "
+            "mcp_config.json, spawn recovery threads, or terminate this process."
         )
 
     if matching_profiles:
@@ -14846,10 +15602,15 @@ def gitea_resolve_task_capability(
     remote: str = "dadeschools",
     host: str | None = None,
 ) -> dict:
-    """Read-only: Resolve which capability, profile, and namespace is required for a Gitea task.
+    """Read-only / side-effect free: resolve capability, profile, and namespace for a task.
 
-    Helps the client or LLM determine the correct namespace or profile before acting,
-    and returns exact next action instructions if the current session is not authorized.
+    Does **not** mutate MCP client configuration, spawn recovery threads, kill
+    processes, or trigger daemon reloads (#685). Stale-runtime detection remains
+    fail-closed: when the serving process is stale the result includes
+    ``blocker_kind=runtime_reconnect_required``, ``restart_required=true``,
+    ``stop_required=true``, and ``mutation_performed=false`` with a precise
+    ``exact_safe_next_action`` pointing at IDE/client reconnect. Recovery is
+    owned by the client reconnect path — never by this resolver.
 
     Args:
         task: The task/action to check (e.g. review_pr, create_issue).
@@ -14926,6 +15687,8 @@ def gitea_resolve_task_capability(
         "commit_files",
         "gitea_commit_files",
         "address_pr_change_requests",
+        "update_pr_branch_by_merge",
+        "gitea_update_pr_branch_by_merge",
         "delete_branch",
         "cleanup_merged_pr_branch",
         "reconciliation_cleanup",
@@ -14968,15 +15731,15 @@ def gitea_resolve_task_capability(
                 "exact_safe_next_action": next_safe_action,
             }
 
-    # #702 F1: validate (and, when provably stale, recover) the inherited
-    # GITEA_ACTIVE_WORKTREE binding BEFORE the terminal launcher probe. The
-    # probe uses that binding as its cwd; a worktree removed mid-session
-    # would otherwise wedge every mutation-task resolution on 'missing cwd'
-    # before the sanctioned recovery could ever run. Recovery stays
-    # fail-closed: only the provably-stale classes clear, with the durable
-    # audit record persisted first (F6).
+    # #702 F1 + #685: assess inherited GITEA_ACTIVE_WORKTREE binding BEFORE the
+    # terminal launcher probe so missing-cwd wedges can be diagnosed. On the
+    # read-only capability-resolve path recovery is report-only
+    # (auto_recover=False): never clear env, write session-state, or otherwise
+    # mutate during gitea_resolve_task_capability. Sanctioned recovery remains
+    # available to mutation entrypoints that call _assess_stale_active_binding
+    # with auto_recover=True.
     try:
-        stale_binding = _assess_stale_active_binding(auto_recover=True)
+        stale_binding = _assess_stale_active_binding(auto_recover=False)
     except Exception:
         stale_binding = None
 
@@ -15100,31 +15863,32 @@ def gitea_resolve_task_capability(
 
     configured = len(matching_profiles) > 0
     available_in_session = allowed_in_current_session
+    runtime_stale_blocker = False
 
     if "PYTEST_CURRENT_TEST" not in os.environ or "GITEA_FORCE_MCP_RUNTIME_CHECK" in os.environ:
         runtime_reasons = _check_mcp_runtimes_diagnostics(task, matching_profiles)
         if runtime_reasons:
             restart_required = True
+            runtime_stale_blocker = True
             reason_msg = "; ".join(runtime_reasons)
-            next_safe_action = (
-                "stale-runtime: Gitea MCP runtime conflict or missing process detected. "
-                "Please fully restart the Gitea MCP server and retry."
-            )
 
     if not allowed_in_current_session:
         if configured and switching:
             restart_required = True
             available_in_session = False
-            reason_msg = (
-                f"{required_role.capitalize()} profile exists but MCP server "
-                "was added after session startup and is not attached."
-            )
+            if not reason_msg:
+                reason_msg = (
+                    f"{required_role.capitalize()} profile exists but MCP server "
+                    "was added after session startup and is not attached."
+                )
         elif not configured:
-            reason_msg = (
-                f"No profile configured with permission '{required_permission}'."
-            )
+            if not reason_msg:
+                reason_msg = (
+                    f"No profile configured with permission '{required_permission}'."
+                )
         elif role_mismatch_reason:
-            reason_msg = role_mismatch_reason
+            if not reason_msg:
+                reason_msg = role_mismatch_reason
     different_namespace_required = False
     next_safe_action = "None; ready for operations."
 
@@ -15150,6 +15914,16 @@ def gitea_resolve_task_capability(
                     "or use the corresponding MCP namespace."
                 )
 
+    # #685: stale-runtime typed remediation wins for exact_next_action when the
+    # serving process/profile inventory is stale — even if permission is OK.
+    if runtime_stale_blocker:
+        next_safe_action = (
+            "blocker_kind=runtime_reconnect_required: reconnect/restart the "
+            "IDE-managed Gitea MCP server for this profile so it reloads current "
+            "master. Do not edit mcp_config.json by hand; the resolver does not "
+            "touch config, spawn recovery threads, or terminate the process."
+        )
+
     # Task/role alignment guards (#167): the requested task, not the
     # available credential, decides what the session may do. A review/merge
     # task under a non-reviewer profile must stop — not silently degrade
@@ -15211,12 +15985,18 @@ def gitea_resolve_task_capability(
         "configured": configured,
         "restart_required": restart_required,
         "stop_required": stop_required or restart_required,
+        # #685: resolver is always side-effect free; never claims mutations.
+        "mutation_performed": False,
         "task_role_guidance": task_role_guidance,
         "matching_configured_profile": matching_profiles,
         "runtime_switching_supported": switching,
         "different_mcp_namespace_required": different_namespace_required,
         "exact_safe_next_action": next_safe_action,
     }
+    # #685: report typed reconnect blocker without mutating config or exiting.
+    if runtime_stale_blocker:
+        result["blocker_kind"] = "runtime_reconnect_required"
+    # #702: optional stale-binding recovery guidance (read-only annotation).
     if stale_binding is not None and stale_binding.get("classification") != (
         stale_binding_recovery.CLASSIFICATION_UNBOUND
     ):

pr_sync_status.py

@@ -0,0 +1,648 @@
+"""PR synchronization and conflict-remediation lifecycle (#PR-SYNC).
+
+Pure assessment helpers for:
+
+* ``gitea_assess_pr_sync_status`` — recommend the next sanctioned action for an
+  open PR when the base branch advances, approvals go stale, or conflicts appear.
+* ``gitea_update_pr_branch_by_merge`` preflight — author-only, fail-closed pin
+  of both PR head and live base head; never rebase/force-push.
+
+All recommendation logic is hermetic (no network) so unit tests cover every
+acceptance path without Gitea credentials.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+_FULL_SHA = re.compile(r"^[0-9a-f]{40}$", re.IGNORECASE)
+
+# Recommended next actions (controller / skill vocabulary).
+ACTION_MERGE_NOW = "merge_now"
+ACTION_UPDATE_BRANCH_BY_MERGE = "update_branch_by_merge"
+ACTION_AUTHOR_CONFLICT_REMEDIATION = "author_conflict_remediation"
+ACTION_FRESH_REVIEW_REQUIRED = "fresh_review_required"
+ACTION_BLOCKED = "blocked"
+
+_VALID_ACTIONS = frozenset({
+    ACTION_MERGE_NOW,
+    ACTION_UPDATE_BRANCH_BY_MERGE,
+    ACTION_AUTHOR_CONFLICT_REMEDIATION,
+    ACTION_FRESH_REVIEW_REQUIRED,
+    ACTION_BLOCKED,
+})
+
+# Author-only mutation; reviewer/merger must never update author branches.
+_AUTHOR_UPDATE_ROLES = frozenset({"author"})
+_DENIED_UPDATE_ROLES = frozenset({"reviewer", "merger", "reconciler", "mixed", "limited"})
+
+# Allowed Gitea update style — merge only (never rebase).
+UPDATE_STYLE_MERGE = "merge"
+_FORBIDDEN_UPDATE_STYLES = frozenset({"rebase", "rebase-merge", "squash", "force"})
+
+
+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 _normalize_role(role: str | None) -> str:
+    return (role or "").strip().lower() or "unknown"
+
+
+def assess_pr_sync_status(
+    *,
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    pr_number: int,
+    pr_state: str | None = None,
+    source_branch: str | None = None,
+    pr_head_sha: str | None = None,
+    base_head_sha: str | None = None,
+    commits_behind: int | None = None,
+    mergeable: bool | None = None,
+    has_conflicts: bool | None = None,
+    branch_protection_requires_current_base: bool | None = None,
+    approval_at_current_head: bool | None = None,
+    checks_status: str | None = None,
+    active_author_lock: bool | None = None,
+    active_reviewer_lease: bool | None = None,
+    active_merger_lease: bool | None = None,
+    prepared_verdict_head_sha: str | None = None,
+    checks_required: bool = True,
+) -> dict[str, Any]:
+    """Recommend the next sanctioned PR lifecycle action.
+
+    Decision order (fail closed):
+
+    1. Incomplete identity / open state / head SHAs → ``blocked``
+    2. Conflicts (or mergeable false with conflict signal) →
+       ``author_conflict_remediation``
+    3. Head changed after approval / prepared verdict for former head →
+       ``fresh_review_required`` (unless conflicts already routed author work)
+    4. Behind live base + branch protection requires current base +
+       auto-mergeable → ``update_branch_by_merge``
+    5. Approval at current head + mergeable + checks ok (+ update not
+       required) → ``merge_now``
+    6. Otherwise ``blocked`` with structured reasons
+    """
+    reasons: list[str] = []
+    pr_head = _normalize_sha(pr_head_sha)
+    base_head = _normalize_sha(base_head_sha)
+    prepared_head = _normalize_sha(prepared_verdict_head_sha)
+    state = (pr_state or "").strip().lower()
+    checks = (checks_status or "unknown").strip().lower()
+    behind = commits_behind if isinstance(commits_behind, int) and commits_behind >= 0 else None
+    requires_current = bool(branch_protection_requires_current_base)
+    approval_ok = bool(approval_at_current_head)
+
+    # Derive conflict when caller only supplies mergeable=False.
+    if has_conflicts is None and mergeable is False:
+        has_conflicts = True
+    conflicts = bool(has_conflicts) if has_conflicts is not None else None
+
+    result: dict[str, Any] = {
+        "host": (host or "").strip() or None,
+        "org": (org or "").strip() or None,
+        "repo": (repo or "").strip() or None,
+        "pr_number": pr_number,
+        "pr_state": state or None,
+        "source_branch": (source_branch or "").strip() or None,
+        "pr_head_sha": pr_head,
+        "base_head_sha": base_head,
+        "commits_behind": behind,
+        "mergeable": mergeable,
+        "has_conflicts": conflicts,
+        "branch_protection_requires_current_base": requires_current,
+        "approval_at_current_head": approval_ok if approval_at_current_head is not None else None,
+        "checks_status": checks,
+        "active_locks_and_leases": {
+            "author_lock": bool(active_author_lock) if active_author_lock is not None else None,
+            "reviewer_lease": bool(active_reviewer_lease) if active_reviewer_lease is not None else None,
+            "merger_lease": bool(active_merger_lease) if active_merger_lease is not None else None,
+        },
+        "prepared_verdict_head_sha": prepared_head,
+        "recommended_next_action": ACTION_BLOCKED,
+        "approval_valid_for_merge": False,
+        "stale_approval": False,
+        "stale_prepared_verdict": False,
+        "reasons": reasons,
+        "success": True,
+        "performed": False,
+    }
+
+    if not isinstance(pr_number, int) or pr_number <= 0:
+        reasons.append("pr_number must be a positive integer (fail closed)")
+        return result
+
+    if state and state not in ("open",):
+        reasons.append(f"PR is not open (state={state}); cannot synchronize or merge")
+        result["recommended_next_action"] = ACTION_BLOCKED
+        return result
+    if not state:
+        reasons.append("PR state missing (fail closed)")
+        return result
+
+    if pr_head is None:
+        reasons.append(
+            "exact PR head SHA missing or not a full 40-char hex SHA (fail closed)"
+        )
+    if base_head is None:
+        reasons.append(
+            "exact live base head SHA missing or not a full 40-char hex SHA (fail closed)"
+        )
+    if behind is None:
+        reasons.append("commits_behind missing or invalid (fail closed)")
+    if mergeable is None:
+        reasons.append("mergeable signal missing (fail closed)")
+    if approval_at_current_head is None:
+        reasons.append("approval_at_current_head missing (fail closed)")
+    if branch_protection_requires_current_base is None:
+        reasons.append(
+            "branch_protection_requires_current_base missing (fail closed)"
+        )
+
+    if reasons:
+        result["recommended_next_action"] = ACTION_BLOCKED
+        return result
+
+    # Stale prepared verdict / approval across heads.
+    stale_prepared = bool(prepared_head and prepared_head != pr_head)
+    result["stale_prepared_verdict"] = stale_prepared
+    stale_approval = not approval_ok
+    result["stale_approval"] = stale_approval
+    result["approval_valid_for_merge"] = bool(approval_ok and not stale_prepared)
+
+    # ── Conflicts: author remediation in dedicated worktree ──────────────
+    if conflicts is True or mergeable is False:
+        if conflicts is True or mergeable is False:
+            # Distinguish pure non-mergeable without explicit conflict flag
+            # still routes author remediation (Gitea cannot auto-update).
+            reasons.append(
+                f"PR #{pr_number} has conflicts or is not mergeable at head "
+                f"{pr_head}; route author conflict remediation in the existing "
+                "issue/PR worktree under branches/ (never force-push or rebase)"
+            )
+            if stale_approval:
+                reasons.append(
+                    "any approval at a former head is invalid; after remediation "
+                    "require a fresh independent review at the new exact head"
+                )
+            result["recommended_next_action"] = ACTION_AUTHOR_CONFLICT_REMEDIATION
+            result["approval_valid_for_merge"] = False
+            return result
+
+    # ── Stale approval / prepared verdict at former head ─────────────────
+    if not approval_ok or stale_prepared:
+        if behind and behind > 0 and requires_current and mergeable is True:
+            # Must update first; update will also invalidate approval.
+            reasons.append(
+                f"PR is {behind} commit(s) behind live base and branch protection "
+                "requires current base; automatic merge-from-base is available"
+            )
+            reasons.append(
+                "approval is not valid at the current head (or prepared verdict "
+                "is head-scoped to a former SHA); after update require fresh review"
+            )
+            result["recommended_next_action"] = ACTION_UPDATE_BRANCH_BY_MERGE
+            result["approval_valid_for_merge"] = False
+            return result
+        reasons.append(
+            "approval is not valid at the exact current PR head "
+            f"({pr_head}); never reuse a former-head approval for merge"
+        )
+        if stale_prepared:
+            reasons.append(
+                f"prepared verdict pinned to former head {prepared_head} cannot "
+                f"authorize review/merge at current head {pr_head}"
+            )
+        if active_reviewer_lease:
+            reasons.append(
+                "active reviewer lease must be released or superseded for the "
+                "new head before a fresh independent review"
+            )
+        if active_merger_lease:
+            reasons.append(
+                "active merger lease cannot cross heads; release or re-acquire "
+                "at the new exact head"
+            )
+        result["recommended_next_action"] = ACTION_FRESH_REVIEW_REQUIRED
+        result["approval_valid_for_merge"] = False
+        return result
+
+    # ── Outdated + protection requires current base, auto-updatable ──────
+    if behind and behind > 0 and requires_current:
+        if mergeable is True and conflicts is not True:
+            reasons.append(
+                f"PR is {behind} commit(s) behind live base {base_head}; "
+                "branch protection requires the PR branch to contain current base; "
+                "Gitea can merge base into the PR branch without conflicts"
+            )
+            reasons.append(
+                "route author-only update_branch_by_merge; never rebase or force-push; "
+                "new head invalidates prior approval and requires fresh independent review"
+            )
+            result["recommended_next_action"] = ACTION_UPDATE_BRANCH_BY_MERGE
+            # Approval today is at old head that will change — not mergeable yet
+            # for final merge until re-reviewed, but assess marks update path.
+            result["approval_valid_for_merge"] = False
+            result["stale_approval"] = True  # will become stale after update
+            return result
+        reasons.append(
+            "update required by branch protection but automatic merge is not available"
+        )
+        result["recommended_next_action"] = ACTION_BLOCKED
+        return result
+
+    # ── Checks gate for merge_now ────────────────────────────────────────
+    if checks_required and checks not in ("success", "passed", "ok", "none", "skipped", "not_required"):
+        if checks in ("pending", "running", "queued"):
+            reasons.append(f"required checks are not finished (status={checks})")
+            result["recommended_next_action"] = ACTION_BLOCKED
+            return result
+        if checks in ("failure", "failed", "error", "cancelled"):
+            reasons.append(f"required checks failed (status={checks})")
+            result["recommended_next_action"] = ACTION_BLOCKED
+            return result
+        # unknown — fail closed when checks_required
+        if checks == "unknown":
+            reasons.append("checks status unknown (fail closed)")
+            result["recommended_next_action"] = ACTION_BLOCKED
+            return result
+
+    # ── Ready to merge without update ────────────────────────────────────
+    # Includes: current with approval; outdated when update is NOT required.
+    if approval_ok and mergeable is True and conflicts is not True:
+        if behind and behind > 0 and not requires_current:
+            reasons.append(
+                f"PR is {behind} commit(s) behind live base but branch protection "
+                "does not require the latest base; preserve the approved head"
+            )
+        else:
+            reasons.append(
+                "PR has a valid approval at the exact current head, is conflict-free "
+                "and mergeable; do not update the branch unnecessarily"
+            )
+        reasons.append("route directly to the sanctioned merger workflow (merge_now)")
+        result["recommended_next_action"] = ACTION_MERGE_NOW
+        result["approval_valid_for_merge"] = True
+        return result
+
+    reasons.append("no sanctioned next action matched the observed PR state (fail closed)")
+    result["recommended_next_action"] = ACTION_BLOCKED
+    return result
+
+
+def assess_update_pr_branch_preflight(
+    *,
+    role_kind: str | None,
+    expected_pr_head_sha: str | None,
+    live_pr_head_sha: str | None,
+    expected_base_head_sha: str | None,
+    live_base_head_sha: str | None,
+    has_conflicts: bool | None = None,
+    mergeable: bool | None = None,
+    style: str | None = UPDATE_STYLE_MERGE,
+    has_author_lock: bool | None = None,
+    worktree_path: str | None = None,
+    worktree_owns_source_branch: bool | None = None,
+    control_checkout_clean: bool | None = None,
+    master_parity_ok: bool | None = None,
+    force_push: bool = False,
+    rebase: bool = False,
+) -> dict[str, Any]:
+    """Author-only preflight for update-by-merge; fail closed on any race.
+
+    When conflicts exist, returns a structured author-remediation handoff and
+    sets ``mutation_allowed=False`` so no partial remote update is performed.
+    """
+    reasons: list[str] = []
+    role = _normalize_role(role_kind)
+    exp_pr = _normalize_sha(expected_pr_head_sha)
+    live_pr = _normalize_sha(live_pr_head_sha)
+    exp_base = _normalize_sha(expected_base_head_sha)
+    live_base = _normalize_sha(live_base_head_sha)
+    style_norm = (style or "").strip().lower() or UPDATE_STYLE_MERGE
+
+    conflicts = has_conflicts
+    if conflicts is None and mergeable is False:
+        conflicts = True
+
+    result: dict[str, Any] = {
+        "mutation_allowed": False,
+        "performed": False,
+        "style": style_norm,
+        "role_kind": role,
+        "expected_pr_head_sha": exp_pr,
+        "live_pr_head_sha": live_pr,
+        "expected_base_head_sha": exp_base,
+        "live_base_head_sha": live_base,
+        "head_race": False,
+        "base_race": False,
+        "has_conflicts": conflicts,
+        "author_remediation_handoff": False,
+        "approval_invalidated_for_former_head": False,
+        "force_push": bool(force_push),
+        "rebase": bool(rebase),
+        "reasons": reasons,
+        "success": True,
+    }
+
+    # Role gate — author only.
+    if role not in _AUTHOR_UPDATE_ROLES:
+        reasons.append(
+            f"role '{role}' cannot update an author PR branch "
+            "(author-only; reviewer/merger denial)"
+        )
+        return result
+
+    if force_push:
+        reasons.append("force-push is forbidden for PR branch update (fail closed)")
+        return result
+    if rebase or style_norm in _FORBIDDEN_UPDATE_STYLES:
+        reasons.append(
+            f"update style '{style_norm}' forbidden; only style=merge is allowed "
+            "(never rebase)"
+        )
+        return result
+    if style_norm != UPDATE_STYLE_MERGE:
+        reasons.append(
+            f"unknown update style '{style_norm}'; expected '{UPDATE_STYLE_MERGE}'"
+        )
+        return result
+
+    if not exp_pr or not live_pr:
+        reasons.append(
+            "expected and live PR head SHAs must be full 40-char hex (fail closed)"
+        )
+    if not exp_base or not live_base:
+        reasons.append(
+            "expected and live base head SHAs must be full 40-char hex (fail closed)"
+        )
+    if reasons:
+        return result
+
+    if exp_pr != live_pr:
+        result["head_race"] = True
+        reasons.append(
+            f"PR head race: expected {exp_pr} but live is {live_pr} "
+            "(fail closed; no partial mutation)"
+        )
+        return result
+    if exp_base != live_base:
+        result["base_race"] = True
+        reasons.append(
+            f"base head race: expected {exp_base} but live is {live_base} "
+            "(fail closed; no partial mutation)"
+        )
+        return result
+
+    if has_author_lock is not True:
+        reasons.append(
+            "existing issue/PR lock required before update_branch_by_merge (fail closed)"
+        )
+    wt = (worktree_path or "").strip()
+    if not wt:
+        reasons.append("existing PR worktree path required under branches/ (fail closed)")
+    else:
+        norm = wt.replace("\\", "/")
+        if "/branches/" not in norm and not norm.startswith("branches/"):
+            reasons.append(
+                f"worktree_path '{wt}' must be under branches/ (fail closed)"
+            )
+    if worktree_owns_source_branch is False:
+        reasons.append(
+            "worktree does not own the PR source branch (fail closed)"
+        )
+    if control_checkout_clean is False:
+        reasons.append("control checkout is not clean (fail closed)")
+    if master_parity_ok is False:
+        reasons.append("runtime/master parity failed (fail closed)")
+
+    if conflicts is True:
+        result["author_remediation_handoff"] = True
+        reasons.append(
+            "conflicts present: return structured author-remediation handoff "
+            "without creating a partial remote update"
+        )
+        reasons.append(
+            "resolve conflicts explicitly in the existing dedicated worktree, "
+            "run focused and regression tests, commit/push via sanctioned author "
+            "workflow, then route the new exact head to independent review"
+        )
+        return result
+
+    if mergeable is False and conflicts is not False:
+        result["author_remediation_handoff"] = True
+        reasons.append(
+            "PR is not mergeable; refuse automatic update and hand off to "
+            "author conflict remediation (fail closed)"
+        )
+        return result
+
+    if reasons:
+        return result
+
+    result["mutation_allowed"] = True
+    reasons.append(
+        "preflight passed: author may merge live base into the PR branch via "
+        "native MCP update (style=merge only)"
+    )
+    return result
+
+
+def assess_post_update_head_transition(
+    *,
+    former_pr_head_sha: str | None,
+    new_pr_head_sha: str | None,
+    former_approval_head_sha: str | None = None,
+    former_reviewer_lease_head_sha: str | None = None,
+    former_merger_lease_head_sha: str | None = None,
+    prepared_verdict_head_sha: str | None = None,
+) -> dict[str, Any]:
+    """After a successful update-by-merge, invalidate cross-head artifacts.
+
+    The new head never inherits approval, reviewer/merger leases, or prepared
+    verdicts from the former head.
+    """
+    former = _normalize_sha(former_pr_head_sha)
+    new = _normalize_sha(new_pr_head_sha)
+    reasons: list[str] = []
+    result: dict[str, Any] = {
+        "former_pr_head_sha": former,
+        "new_pr_head_sha": new,
+        "head_changed": bool(former and new and former != new),
+        "approval_invalidated": False,
+        "reviewer_lease_superseded": False,
+        "merger_lease_superseded": False,
+        "prepared_verdict_invalidated": False,
+        "recommended_next_action": ACTION_BLOCKED,
+        "reasons": reasons,
+        "success": True,
+    }
+
+    if not former or not new:
+        reasons.append("former and new PR head SHAs required (fail closed)")
+        return result
+    if former == new:
+        reasons.append(
+            "PR head unchanged after update; unexpected for merge-from-base"
+        )
+        result["recommended_next_action"] = ACTION_BLOCKED
+        return result
+
+    result["head_changed"] = True
+    # Approval at former head is always void at new head.
+    former_approval = _normalize_sha(former_approval_head_sha) or former
+    if former_approval != new:
+        result["approval_invalidated"] = True
+        reasons.append(
+            f"approval for former head {former_approval} is invalid at new head "
+            f"{new}; never preserve approval across a head change"
+        )
+
+    rev_lease = _normalize_sha(former_reviewer_lease_head_sha)
+    if rev_lease and rev_lease != new:
+        result["reviewer_lease_superseded"] = True
+        reasons.append(
+            f"reviewer lease for head {rev_lease} cannot cross to {new}; "
+            "release or supersede before fresh review"
+        )
+    elif rev_lease is None:
+        # Unknown lease head still must not authorize at new head.
+        result["reviewer_lease_superseded"] = True
+        reasons.append(
+            "any reviewer lease scoped to the former head is superseded at the new head"
+        )
+
+    mer_lease = _normalize_sha(former_merger_lease_head_sha)
+    if mer_lease and mer_lease != new:
+        result["merger_lease_superseded"] = True
+        reasons.append(
+            f"merger lease for head {mer_lease} cannot cross to {new}"
+        )
+    else:
+        result["merger_lease_superseded"] = True
+        reasons.append(
+            "any merger lease scoped to the former head is superseded at the new head"
+        )
+
+    prepared = _normalize_sha(prepared_verdict_head_sha)
+    if prepared and prepared != new:
+        result["prepared_verdict_invalidated"] = True
+        reasons.append(
+            f"prepared verdict for head {prepared} cannot authorize the new head {new}"
+        )
+    elif prepared is None:
+        result["prepared_verdict_invalidated"] = True
+        reasons.append(
+            "prepared verdicts associated with the former head are invalidated"
+        )
+
+    result["recommended_next_action"] = ACTION_FRESH_REVIEW_REQUIRED
+    reasons.append(
+        "route the new exact head to independent review "
+        f"({ACTION_FRESH_REVIEW_REQUIRED})"
+    )
+    return result
+
+
+def assess_sequential_queue_step(
+    *,
+    just_merged: bool,
+    master_refreshed: bool,
+    next_pr_number: int | None = None,
+) -> dict[str, Any]:
+    """Controller sequential processing: reassess only after merge + master refresh."""
+    reasons: list[str] = []
+    if just_merged and not master_refreshed:
+        reasons.append(
+            "master must be refreshed after each merge before reassessing the next PR "
+            "(do not update every PR simultaneously)"
+        )
+        return {
+            "reassess_allowed": False,
+            "process_next": False,
+            "next_pr_number": next_pr_number,
+            "reasons": reasons,
+            "success": True,
+        }
+    if not just_merged:
+        reasons.append("no merge completed; sequential step does not advance")
+        return {
+            "reassess_allowed": False,
+            "process_next": False,
+            "next_pr_number": next_pr_number,
+            "reasons": reasons,
+            "success": True,
+        }
+    if next_pr_number:
+        reasons.append(
+            "merge completed and live master refreshed; reassess the next PR "
+            f"#{next_pr_number}"
+        )
+    else:
+        reasons.append(
+            "merge completed and live master refreshed; reassess the next PR"
+        )
+    return {
+        "reassess_allowed": True,
+        "process_next": True,
+        "next_pr_number": next_pr_number,
+        "post_merge_cleanup_handoff": True,
+        "reasons": reasons,
+        "success": True,
+    }
+
+
+def merge_approval_usable_at_head(
+    *,
+    current_head_sha: str | None,
+    approved_head_sha: str | None,
+) -> dict[str, Any]:
+    """Stale approval cannot authorize merge (head-scoped only)."""
+    current = _normalize_sha(current_head_sha)
+    approved = _normalize_sha(approved_head_sha)
+    ok = bool(current and approved and current == approved)
+    reasons: list[str] = []
+    if not ok:
+        reasons.append(
+            f"stale approval: approved head '{approved or '(none)'}' does not "
+            f"match current head '{current or '(none)'}'; cannot authorize merge"
+        )
+    return {
+        "approval_usable": ok,
+        "current_head_sha": current,
+        "approved_head_sha": approved,
+        "reasons": reasons,
+    }
+
+
+def lease_usable_at_head(
+    *,
+    lease_kind: str,
+    lease_head_sha: str | None,
+    current_head_sha: str | None,
+) -> dict[str, Any]:
+    """Reviewer/merger leases cannot cross heads."""
+    current = _normalize_sha(current_head_sha)
+    lease_head = _normalize_sha(lease_head_sha)
+    kind = (lease_kind or "").strip().lower() or "unknown"
+    ok = bool(current and lease_head and current == lease_head)
+    reasons: list[str] = []
+    if not ok:
+        reasons.append(
+            f"stale {kind} lease: lease head '{lease_head or '(none)'}' cannot "
+            f"authorize work at current head '{current or '(none)'}'"
+        )
+    return {
+        "lease_usable": ok,
+        "lease_kind": kind,
+        "lease_head_sha": lease_head,
+        "current_head_sha": current,
+        "reasons": reasons,
+    }

skills/llm-project-workflow/templates/merge-pr.md

@@ -33,22 +33,34 @@ Steps:
    *If the current identity does not match the required role (or is the PR author), STOP. Relaunch/switch to the correct profile first.*
 2. Verify authenticated identity + active profile.
 3. Confirm PR #<pr>: author (not you), state open, mergeable, review approved. Check if PR body uses `Closes #N` or `Fixes #N`; if it uses `Implements #N` or `Refs #N`, manual closing will be needed in step 29.
-4. Capability evidence (#179): cite the exact gitea_resolve_task_capability
+4. **PR sync assess (required):** call `gitea_assess_pr_sync_status` with
+   explicit `remote`/`org`/`repo`. Route on `recommended_next_action`:
+   - `merge_now` → continue merger path (do **not** update the branch)
+   - `update_branch_by_merge` → STOP; hand off to **author** for
+     `gitea_update_pr_branch_by_merge` (pins expected PR head + base head)
+   - `author_conflict_remediation` → STOP; author worktree conflict fix
+   - `fresh_review_required` → STOP; independent re-review at current head
+   - `blocked` → STOP and diagnose
+   Never merge on a former-head approval after update/remediation.
+5. Capability evidence (#179): cite the exact gitea_resolve_task_capability
    output (or runtime context) proving merge_pr is allowed — a bare
    "capability checks passed" claim is downgraded.
-5. Final live-state recheck (#179), immediately before the merge mutation —
+6. Final live-state recheck (#179), immediately before the merge mutation —
    re-read the live PR and prove:
    - PR still open
    - live head SHA still equals the pinned/reviewed head SHA
    - base branch unchanged
    - no undismissed REQUEST_CHANGES / blocking review state remains
    If any recheck fails → STOP, re-pin, re-validate.
-6. If any gate fails → STOP and report.
-7. Merge with explicit confirmation (e.g. confirmation="MERGE PR <pr>"),
+7. If any gate fails → STOP and report.
+8. Merge with explicit confirmation (e.g. confirmation="MERGE PR <pr>"),
    pinning the reviewed head SHA (expected_head_sha) and, where supported,
    the changed-file set.
-8. Confirm remote master now contains the merge commit (or the expected changes if squash merged).
+9. Confirm remote master now contains the merge commit (or the expected changes if squash merged).
    *Note: Gitea PR "closed" state is NOT equivalent to "merged". Do not assume a closed PR succeeded without verifying the actual landed changes.*
+10. **Sequential queue:** after merge, refresh live `master`, run post-merge
+    cleanup handoff, then reassess the **next** PR (do not batch-update all
+    open PRs).
 
 Post-merge cleanup (#517): merger sessions must NOT perform ad hoc cleanup.
 - Record merge mutations separately from cleanup mutations in the controller handoff.

skills/llm-project-workflow/workflows/review-merge-pr.md

@@ -950,9 +950,53 @@ Final reports must state:
 * final live head SHA before merge
 * whether any push occurred during validation
 
+## 26D. PR synchronization and conflict-remediation lifecycle
+
+**Do not treat “approved” as the final readiness state.** For every approved
+open PR, call `gitea_assess_pr_sync_status` (native MCP only) and route by
+`recommended_next_action`:
+
+| Action | Meaning | Next role / tools |
+|--------|---------|-------------------|
+| `merge_now` | Valid approval at exact current head; conflict-free; update not required; checks ok | Merger: sanctioned merge workflow only — **do not** update the branch |
+| `update_branch_by_merge` | Behind live base; protection requires current base; Gitea can merge base without conflicts | **Author only:** `gitea_update_pr_branch_by_merge` with pinned `expected_pr_head_sha` + `expected_base_head_sha` |
+| `author_conflict_remediation` | Conflicts / not auto-updatable | **Author only:** existing `branches/` worktree, issue lock, merge master, resolve, test, push; never force-push/rebase |
+| `fresh_review_required` | Head changed after approval, or update/remediation produced a new head | Independent reviewer at the **new exact head**; old approvals/leases/verdicts are void |
+| `blocked` | Other gate (checks, incomplete facts, closed PR, …) | Diagnose; do not merge or update |
+
+### Hard rules
+
+* Pin both expected PR head and expected base head for every update. Either
+  race → fail closed with no partial mutation.
+* Never rebase or force-push author branches.
+* Never update an author branch from a reviewer or merger profile.
+* Never preserve approval, reviewer lease, merger lease, or prepared verdict
+  across a head change.
+* Process PRs **sequentially**: synchronize/remediate one → review new head →
+  merge → refresh live `master` → reassess the next PR. Do not update every
+  open PR at once (each merge restales the rest).
+* Conflict remediation only in the existing dedicated issue/PR worktree under
+  `branches/`. Do not delete that worktree until the updated PR is merged and
+  cleanup eligibility is proven.
+* Post-merge: canonical cleanup handoff to reconciler (section 28).
+
+### After `gitea_update_pr_branch_by_merge` succeeds
+
+1. Treat former-head approval as invalidated.
+2. Release/supersede obsolete reviewer and merger leases.
+3. Route `recommended_next_action=fresh_review_required` at the new head.
+4. Independent re-review, then merger lease/adopt + merge at the new head only.
+
+All Gitea reads/mutations use native MCP. Never substitute direct API, curl,
+tea/gh, Web UI mutation by an LLM, database changes, raw Git push, or token
+access.
+
 ## 27. Merge rules
 
-Before merge, rerun fresh live checks:
+Before merge, call `gitea_assess_pr_sync_status` when the PR is approved/open
+and may be behind base. Only proceed with merge when
+`recommended_next_action` is `merge_now` and approval remains at the current
+head. Then rerun fresh live checks:
 
 * whoami
 * active profile/runtime

task_capability_map.py

@@ -73,13 +73,23 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.branch.push",
         "role": "author",
     },
-    "acquire_reviewer_pr_lease": {
-        "permission": "gitea.pr.comment",
-        "role": "reviewer",
+    # PR synchronization lifecycle: assess is read-only (any role with gitea.read);
+    # update-by-merge is author-only and mutates the PR head via Gitea API.
+    "assess_pr_sync_status": {
+        "permission": "gitea.read",
+        "role": "author",
     },
-    "gitea_acquire_reviewer_pr_lease": {
-        "permission": "gitea.pr.comment",
-        "role": "reviewer",
+    "gitea_assess_pr_sync_status": {
+        "permission": "gitea.read",
+        "role": "author",
+    },
+    "update_pr_branch_by_merge": {
+        "permission": "gitea.branch.push",
+        "role": "author",
+    },
+    "gitea_update_pr_branch_by_merge": {
+        "permission": "gitea.branch.push",
+        "role": "author",
     },
     "review_pr": {
         "permission": "gitea.pr.review",

tests/test_issue_685_side_effect_free_resolver.py

@@ -0,0 +1,430 @@
+"""#685: gitea_resolve_task_capability must be side-effect free.
+
+Stale-runtime detection remains fail-closed, but the resolver must never:
+  * touch mcp_config.json (or any MCP client config)
+  * spawn recovery threads
+  * call os._exit / terminate the serving process
+  * claim that an auto-restart was triggered
+
+Recovery is owned by the IDE/client reconnect path only.
+"""
+
+from __future__ import annotations
+
+import os
+import tempfile
+import threading
+import unittest
+from datetime import datetime
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import sys
+
+ROOT = str(Path(__file__).resolve().parent.parent)
+if ROOT not in sys.path:
+    sys.path.insert(0, ROOT)
+
+import gitea_mcp_server as mcp_server
+
+
+ROLE_PROFILES = (
+    ("create_issue", "prgs-author", "author"),
+    ("review_pr", "prgs-reviewer", "reviewer"),
+    ("merge_pr", "prgs-merger", "merger"),
+    ("reconciliation_cleanup", "prgs-reconciler", "reconciler"),
+)
+
+
+def _stale_self_ps_mocks(profile: str = "prgs-author"):
+    """Build subprocess mocks: self PID is stale vs code mtime."""
+    mock_getpid = MagicMock(return_value=12345)
+    mock_exists = MagicMock(return_value=True)
+    code_time = datetime(2026, 7, 8, 14, 0, 0)
+    mock_getmtime = MagicMock(return_value=code_time.timestamp())
+
+    ps_output = (
+        "  PID LSTART COMMAND\n"
+        "12345 Wed Jul  8 13:00:00 2026 /path/to/python mcp_server.py\n"
+    )
+    mock_run_ps = MagicMock()
+    mock_run_ps.stdout = ps_output
+
+    mock_run_env = MagicMock()
+    mock_run_env.stdout = f"GITEA_MCP_PROFILE={profile}"
+
+    mock_run_git = MagicMock()
+    mock_run_git.stdout = "SAME"
+
+    def side_effect(args, **kwargs):
+        if args[0] == "ps" and "eww" in args:
+            return mock_run_env
+        if args[0] == "ps":
+            return mock_run_ps
+        if args[0] == "git":
+            return mock_run_git
+        raise ValueError(f"Unexpected subprocess args: {args}")
+
+    mock_run = MagicMock(side_effect=side_effect)
+    return mock_getpid, mock_exists, mock_getmtime, mock_run
+
+
+class TestIssue685DiagnosticsNoSideEffects(unittest.TestCase):
+    def setUp(self):
+        mcp_server._process_boot_head_sha = None
+
+    def tearDown(self):
+        mcp_server._process_boot_head_sha = None
+
+    @patch.dict(os.environ, {"GITEA_FORCE_MCP_RUNTIME_CHECK": "1"}, clear=False)
+    @patch("subprocess.run")
+    @patch("os.path.getmtime")
+    @patch("os.path.exists")
+    @patch("os.getpid")
+    @patch("os.utime")
+    @patch("threading.Thread")
+    @patch("os._exit")
+    def test_stale_self_does_not_touch_config_or_exit(
+        self,
+        mock_exit,
+        mock_thread,
+        mock_utime,
+        mock_getpid,
+        mock_exists,
+        mock_getmtime,
+        mock_run,
+    ):
+        mock_getpid.return_value = 12345
+        mock_exists.return_value = True
+        mock_getmtime.return_value = datetime(2026, 7, 8, 14, 0, 0).timestamp()
+        mock_run.side_effect = _stale_self_ps_mocks("prgs-author")[3].side_effect
+
+        before_threads = threading.active_count()
+        reasons = mcp_server._check_mcp_runtimes_diagnostics(
+            "create_issue", ["prgs-author"]
+        )
+        after_threads = threading.active_count()
+
+        self.assertTrue(
+            any("stale-runtime" in r and "active Gitea MCP server process is stale" in r
+                for r in reasons),
+            reasons,
+        )
+        # Must not claim auto-restart / config touch
+        blob = " ".join(reasons)
+        self.assertNotIn("Auto-restart has been triggered", blob)
+        self.assertNotIn("touched mcp_config", blob)
+        self.assertNotIn("will cleanly exit", blob)
+
+        mock_utime.assert_not_called()
+        mock_thread.assert_not_called()
+        mock_exit.assert_not_called()
+        self.assertEqual(before_threads, after_threads)
+
+    @patch.dict(os.environ, {"GITEA_FORCE_MCP_RUNTIME_CHECK": "1"}, clear=False)
+    @patch("subprocess.run")
+    @patch("os.path.getmtime")
+    @patch("os.path.exists")
+    @patch("os.getpid")
+    @patch("os.utime")
+    def test_repeated_stale_calls_do_not_trigger_restart_loop(
+        self, mock_utime, mock_getpid, mock_exists, mock_getmtime, mock_run
+    ):
+        mock_getpid.return_value = 12345
+        mock_exists.return_value = True
+        mock_getmtime.return_value = datetime(2026, 7, 8, 14, 0, 0).timestamp()
+        mock_run.side_effect = _stale_self_ps_mocks("prgs-author")[3].side_effect
+
+        for _ in range(5):
+            reasons = mcp_server._check_mcp_runtimes_diagnostics(
+                "create_issue", ["prgs-author"]
+            )
+            self.assertTrue(any("stale-runtime" in r for r in reasons))
+
+        mock_utime.assert_not_called()
+
+    def test_trigger_mcp_auto_restart_removed(self):
+        """#685 AC: auto-restart helper is removed (unreachable from read-only)."""
+        self.assertFalse(hasattr(mcp_server, "_trigger_mcp_auto_restart"))
+        self.assertFalse(hasattr(mcp_server, "_restart_triggered"))
+
+
+class TestIssue685ResolverTypedBlocker(unittest.TestCase):
+    def setUp(self):
+        mcp_server._process_boot_head_sha = None
+
+    def tearDown(self):
+        mcp_server._process_boot_head_sha = None
+        if hasattr(mcp_server, "capability_stop_terminal"):
+            mcp_server.capability_stop_terminal.clear()
+
+    def _resolve_with_stale_runtime(self, task: str, profile_name: str, role: str):
+        allowed = [
+            "gitea.read",
+            "gitea.issue.create",
+            "gitea.issue.comment",
+            "gitea.issue.close",
+            "gitea.branch.create",
+            "gitea.branch.push",
+            "gitea.branch.delete",
+            "gitea.pr.create",
+            "gitea.pr.comment",
+            "gitea.pr.review",
+            "gitea.pr.approve",
+            "gitea.pr.request_changes",
+            "gitea.pr.merge",
+            "gitea.pr.close",
+            "gitea.repo.commit",
+        ]
+        profile = {
+            "profile_name": profile_name,
+            "role": role,
+            "allowed_operations": allowed,
+            "forbidden_operations": [],
+        }
+        config = {
+            "profiles": {
+                profile_name: {
+                    "role": role,
+                    "allowed_operations": allowed,
+                    "forbidden_operations": [],
+                }
+            }
+        }
+
+        mock_getpid, mock_exists, mock_getmtime, mock_run = _stale_self_ps_mocks(
+            profile_name
+        )
+
+        with patch.dict(
+            os.environ,
+            {
+                "GITEA_FORCE_MCP_RUNTIME_CHECK": "1",
+                "GITEA_MCP_PROFILE": profile_name,
+            },
+            clear=False,
+        ), patch.object(mcp_server, "get_profile", return_value=profile), patch.object(
+            mcp_server.gitea_config, "load_config", return_value=config
+        ), patch.object(
+            mcp_server, "_authenticated_username", return_value="test-user"
+        ), patch.object(
+            mcp_server, "_ensure_matching_profile", return_value=None
+        ), patch.object(
+            mcp_server, "record_preflight_check", return_value=None
+        ), patch.object(
+            mcp_server, "record_mutation_authority", return_value=None
+        ), patch.object(
+            mcp_server, "init_review_decision_lock", return_value=None
+        ), patch(
+            "subprocess.run", mock_run
+        ), patch(
+            "os.path.getmtime", mock_getmtime
+        ), patch(
+            "os.path.exists", mock_exists
+        ), patch(
+            "os.getpid", mock_getpid
+        ), patch(
+            "os.utime"
+        ) as mock_utime, patch(
+            "threading.Thread"
+        ) as mock_thread, patch(
+            "os._exit"
+        ) as mock_exit:
+            result = mcp_server.gitea_resolve_task_capability(task=task, remote="prgs")
+            return result, mock_utime, mock_thread, mock_exit
+
+    def test_stale_returns_typed_blocker_fields(self):
+        result, mock_utime, mock_thread, mock_exit = self._resolve_with_stale_runtime(
+            "create_issue", "prgs-author", "author"
+        )
+        self.assertTrue(result.get("restart_required"), result)
+        self.assertTrue(result.get("stop_required"), result)
+        self.assertEqual(result.get("blocker_kind"), "runtime_reconnect_required")
+        self.assertIs(result.get("mutation_performed"), False)
+        action = result.get("exact_safe_next_action") or ""
+        self.assertIn("reconnect", action.lower())
+        self.assertNotIn("None; ready for operations", action)
+        reason = result.get("reason") or ""
+        self.assertIn("stale-runtime", reason)
+        self.assertNotIn("Auto-restart has been triggered", reason)
+        mock_utime.assert_not_called()
+        mock_thread.assert_not_called()
+        mock_exit.assert_not_called()
+
+    def test_all_four_role_profiles_get_same_side_effect_free_contract(self):
+        for task, profile, role in ROLE_PROFILES:
+            with self.subTest(task=task, profile=profile):
+                # Skip tasks that may be unknown on this branch
+                try:
+                    import task_capability_map as tcm
+
+                    tcm.required_permission(task)
+                except Exception:
+                    self.skipTest(f"task {task} not in capability map")
+
+                result, mock_utime, mock_thread, mock_exit = (
+                    self._resolve_with_stale_runtime(task, profile, role)
+                )
+                self.assertTrue(
+                    result.get("restart_required") or result.get("stop_required"),
+                    result,
+                )
+                self.assertEqual(
+                    result.get("blocker_kind"), "runtime_reconnect_required", result
+                )
+                self.assertIs(result.get("mutation_performed"), False, result)
+                mock_utime.assert_not_called()
+                mock_thread.assert_not_called()
+                mock_exit.assert_not_called()
+
+    def test_config_mtime_and_contents_unchanged(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            cfg = os.path.join(tmp, "mcp_config.json")
+            original = '{"servers": {"gitea-author": {}}}'
+            with open(cfg, "w", encoding="utf-8") as fh:
+                fh.write(original)
+            mtime_before = os.path.getmtime(cfg)
+
+            result, mock_utime, mock_thread, mock_exit = self._resolve_with_stale_runtime(
+                "create_issue", "prgs-author", "author"
+            )
+            # Force-path also must not use real utime when diagnostics runs
+            with open(cfg, encoding="utf-8") as fh:
+                after = fh.read()
+            self.assertEqual(after, original)
+            self.assertEqual(os.path.getmtime(cfg), mtime_before)
+            mock_utime.assert_not_called()
+            self.assertTrue(result.get("restart_required"), result)
+
+
+class TestIssue685MutationGatesStillFailClosed(unittest.TestCase):
+    def test_parity_stale_still_reports_restart_required(self):
+        """Mutation-facing parity gate remains fail-closed when heads differ."""
+        import master_parity_gate as mpg
+
+        out = mpg.assess_master_parity(
+            {"startup_head": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"},
+            "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
+        )
+        self.assertFalse(out.get("in_parity"))
+        self.assertTrue(out.get("restart_required") or out.get("stale"))
+
+
+class TestIssue685DocstringReadOnlyContract(unittest.TestCase):
+    def test_resolve_docstring_declares_side_effect_free(self):
+        doc = mcp_server.gitea_resolve_task_capability.__doc__ or ""
+        lower = doc.lower()
+        self.assertTrue(
+            "side-effect" in lower or "read-only" in lower or "does not mutate" in lower,
+            doc,
+        )
+        self.assertNotIn("auto-restart", lower)
+        self.assertNotIn("os._exit", lower)
+
+
+class TestIssue685MergeCoexistenceWithMasterAnnotations(unittest.TestCase):
+    """After merging master: #685 reconnect blocker + #702 report-only binding."""
+
+    def setUp(self):
+        mcp_server._process_boot_head_sha = None
+
+    def tearDown(self):
+        mcp_server._process_boot_head_sha = None
+        if hasattr(mcp_server, "capability_stop_terminal"):
+            mcp_server.capability_stop_terminal.clear()
+
+    def test_resolve_uses_report_only_stale_binding_assessment(self):
+        """Capability resolve must call stale-binding assess with auto_recover=False."""
+        allowed = [
+            "gitea.read",
+            "gitea.issue.create",
+            "gitea.issue.comment",
+            "gitea.branch.push",
+            "gitea.pr.create",
+            "gitea.pr.comment",
+            "gitea.pr.review",
+            "gitea.pr.merge",
+            "gitea.repo.commit",
+        ]
+        profile = {
+            "profile_name": "prgs-author",
+            "role": "author",
+            "allowed_operations": allowed,
+            "forbidden_operations": [],
+        }
+        config = {
+            "profiles": {
+                "prgs-author": {
+                    "role": "author",
+                    "allowed_operations": allowed,
+                    "forbidden_operations": [],
+                }
+            }
+        }
+        mock_getpid, mock_exists, mock_getmtime, mock_run = _stale_self_ps_mocks(
+            "prgs-author"
+        )
+        fake_binding = {
+            "classification": "missing_path",
+            "active_worktree": "/tmp/gone",
+            "reasons": ["path missing"],
+        }
+
+        with patch.dict(
+            os.environ,
+            {
+                "GITEA_FORCE_MCP_RUNTIME_CHECK": "1",
+                "GITEA_MCP_PROFILE": "prgs-author",
+            },
+            clear=False,
+        ), patch.object(mcp_server, "get_profile", return_value=profile), patch.object(
+            mcp_server.gitea_config, "load_config", return_value=config
+        ), patch.object(
+            mcp_server, "_authenticated_username", return_value="test-user"
+        ), patch.object(
+            mcp_server, "_ensure_matching_profile", return_value=None
+        ), patch.object(
+            mcp_server, "record_preflight_check", return_value=None
+        ), patch.object(
+            mcp_server, "record_mutation_authority", return_value=None
+        ), patch.object(
+            mcp_server, "init_review_decision_lock", return_value=None
+        ), patch.object(
+            mcp_server,
+            "_assess_stale_active_binding",
+            return_value=fake_binding,
+        ) as mock_assess, patch(
+            "subprocess.run", mock_run
+        ), patch(
+            "os.path.getmtime", mock_getmtime
+        ), patch(
+            "os.path.exists", mock_exists
+        ), patch(
+            "os.getpid", mock_getpid
+        ), patch(
+            "os.utime"
+        ) as mock_utime, patch(
+            "threading.Thread"
+        ) as mock_thread, patch(
+            "os._exit"
+        ) as mock_exit:
+            result = mcp_server.gitea_resolve_task_capability(
+                task="create_issue", remote="prgs"
+            )
+
+        mock_assess.assert_called()
+        # Every call must be report-only (no env clear / session-state write).
+        for call in mock_assess.call_args_list:
+            self.assertFalse(call.kwargs.get("auto_recover", True))
+        self.assertEqual(
+            result.get("blocker_kind"), "runtime_reconnect_required", result
+        )
+        self.assertEqual(result.get("stale_binding_recovery"), fake_binding, result)
+        self.assertIs(result.get("mutation_performed"), False, result)
+        mock_utime.assert_not_called()
+        mock_thread.assert_not_called()
+        mock_exit.assert_not_called()
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_mcp_stale_runtime.py

@@ -111,21 +111,13 @@ class TestMcpStaleRuntime(unittest.TestCase):
         reasons = gitea_mcp_server._check_mcp_runtimes_diagnostics("create_issue", ["prgs-author"])
         self.assertTrue(any("stale-runtime: The active Gitea MCP server process is stale" in r for r in reasons))
 
-    @patch("threading.Thread")
-    @patch("os.utime")
-    @patch("os.path.exists")
-    @patch.dict("os.environ", {"MCP_CONFIG_PATH": "/tmp/mcp_config.json"})
-    def test_auto_restart_trigger_touches_and_spawns(self, mock_exists, mock_utime, mock_thread):
-        mock_exists.return_value = True
-        gitea_mcp_server._restart_triggered = False
-
-        # Ensure we are not skipped in test mode for testing purposes
-        with patch("gitea_mcp_server._preflight_in_test_mode", return_value=False):
-            gitea_mcp_server._trigger_mcp_auto_restart()
-
-        mock_utime.assert_called_once_with("/tmp/mcp_config.json", None)
-        mock_thread.assert_called_once()
-        self.assertTrue(gitea_mcp_server._restart_triggered)
+    def test_auto_restart_helper_removed_from_read_only_path(self):
+        """#685: config-touch / os._exit self-recovery is no longer on the server."""
+        self.assertFalse(
+            hasattr(gitea_mcp_server, "_trigger_mcp_auto_restart"),
+            "_trigger_mcp_auto_restart must not remain (side-effect-free resolver)",
+        )
+        self.assertFalse(hasattr(gitea_mcp_server, "_restart_triggered"))
 
 
 if __name__ == "__main__":

tests/test_pr_ownership_issue_pr_mismatch.py

@@ -0,0 +1,196 @@
+"""#727 / PR #728: author ownership when tracking issue number != PR number.
+
+Regression for REQUEST_CHANGES: gitea_update_pr_branch_by_merge must not require
+issue_number == pr_number. Ownership is proven via linked issue + lock/branch
+context and fails closed for unrelated issues.
+"""
+
+from __future__ import annotations
+
+import os
+import tempfile
+import unittest
+from datetime import datetime, timedelta, timezone
+from unittest.mock import patch
+
+import issue_lock_store
+import gitea_mcp_server as mcp
+
+
+def _live_lock(
+    *,
+    issue_number: int,
+    branch_name: str = "feat/issue-727-pr-sync-status",
+    worktree_path: str = "/tmp/branches/issue-727-pr-sync-status",
+    remote: str = "prgs",
+    org: str = "Scaled-Tech-Consulting",
+    repo: str = "Gitea-Tools",
+) -> dict:
+    now = datetime.now(timezone.utc)
+    return {
+        "issue_number": issue_number,
+        "branch_name": branch_name,
+        "worktree_path": worktree_path,
+        "remote": remote,
+        "org": org,
+        "repo": repo,
+        "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
+        "acquired_at": now.isoformat(),
+        "expires_at": (now + timedelta(hours=2)).isoformat(),
+        "owner_pid": os.getpid(),
+        "status": "active",
+    }
+
+
+class TestAuthorOwnershipIssuePrMismatch(unittest.TestCase):
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory(prefix="gitea-issue-locks-")
+        self.lock_dir = self._tmp.name
+        os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir
+
+    def tearDown(self):
+        # Clear session pointer for this process.
+        try:
+            ptr = issue_lock_store.session_pointer_path(self.lock_dir)
+            if os.path.isfile(ptr):
+                os.unlink(ptr)
+        except Exception:
+            pass
+        os.environ.pop("GITEA_ISSUE_LOCK_DIR", None)
+        self._tmp.cleanup()
+
+    def test_issue_727_pr_728_session_lock_accepted(self):
+        """Tracking issue #727 lock is valid ownership for PR #728."""
+        lock = _live_lock(issue_number=727)
+        issue_lock_store.bind_session_lock(lock)
+        result = mcp._prove_author_ownership_for_pr(
+            pr_number=728,
+            pr_title="feat: pr sync",
+            pr_body="Closes #727\n\nNative PR sync lifecycle.",
+            source_branch="feat/issue-727-pr-sync-status",
+            remote="prgs",
+            host=None,
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            worktree_path=lock["worktree_path"],
+        )
+        self.assertTrue(result["proven"], result)
+        self.assertTrue(result["has_author_lock"])
+        self.assertEqual(result["matched_issue"], 727)
+        self.assertEqual(result["matched_via"], "session_lock")
+        self.assertIn(727, result["linked_issues"])
+
+    def test_issue_727_pr_728_durable_lock_accepted(self):
+        lock = _live_lock(issue_number=727)
+        path = issue_lock_store.lock_file_path(
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            issue_number=727,
+            lock_dir=self.lock_dir,
+        )
+        issue_lock_store.save_lock_file(path, lock)
+        result = mcp._prove_author_ownership_for_pr(
+            pr_number=728,
+            pr_title="feat: pr sync",
+            pr_body="Fixes #727",
+            source_branch="feat/issue-727-pr-sync-status",
+            remote="prgs",
+            host=None,
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+        )
+        self.assertTrue(result["proven"], result)
+        self.assertEqual(result["matched_issue"], 727)
+        self.assertEqual(result["matched_via"], "durable_lock")
+
+    def test_legacy_same_number_still_accepted(self):
+        lock = _live_lock(
+            issue_number=100,
+            branch_name="fix/issue-100",
+            worktree_path="/tmp/branches/issue-100",
+        )
+        issue_lock_store.bind_session_lock(lock)
+        result = mcp._prove_author_ownership_for_pr(
+            pr_number=100,
+            pr_title="fix something",
+            pr_body="Closes #100",
+            source_branch="fix/issue-100",
+            remote="prgs",
+            host=None,
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            worktree_path=lock["worktree_path"],
+        )
+        self.assertTrue(result["proven"], result)
+        self.assertEqual(result["matched_issue"], 100)
+
+    def test_unrelated_issue_lock_rejected(self):
+        """Lock for issue #999 must not authorize PR #728 linked only to #727."""
+        lock = _live_lock(
+            issue_number=999,
+            branch_name="feat/issue-999",
+            worktree_path="/tmp/branches/issue-999",
+        )
+        issue_lock_store.bind_session_lock(lock)
+        path = issue_lock_store.lock_file_path(
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            issue_number=999,
+            lock_dir=self.lock_dir,
+        )
+        issue_lock_store.save_lock_file(path, lock)
+        result = mcp._prove_author_ownership_for_pr(
+            pr_number=728,
+            pr_title="feat: pr sync",
+            pr_body="Closes #727",
+            source_branch="feat/issue-727-pr-sync-status",
+            remote="prgs",
+            host=None,
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            worktree_path=lock["worktree_path"],
+        )
+        self.assertFalse(result["proven"], result)
+        self.assertFalse(result["has_author_lock"])
+        self.assertIsNone(result["matched_issue"])
+        self.assertTrue(result["reasons"])
+
+    def test_branch_mismatch_on_session_lock_fail_closed(self):
+        lock = _live_lock(
+            issue_number=727,
+            branch_name="feat/other-branch",
+        )
+        issue_lock_store.bind_session_lock(lock)
+        result = mcp._prove_author_ownership_for_pr(
+            pr_number=728,
+            pr_title="feat: pr sync",
+            pr_body="Closes #727",
+            source_branch="feat/issue-727-pr-sync-status",
+            remote="prgs",
+            host=None,
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            worktree_path=lock["worktree_path"],
+        )
+        self.assertFalse(result["proven"], result)
+        self.assertTrue(any("branch" in r for r in result["reasons"]))
+
+    def test_no_lock_fail_closed(self):
+        result = mcp._prove_author_ownership_for_pr(
+            pr_number=728,
+            pr_title="feat: pr sync",
+            pr_body="Closes #727",
+            source_branch="feat/issue-727-pr-sync-status",
+            remote="prgs",
+            host=None,
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+        )
+        self.assertFalse(result["proven"], result)
+        self.assertTrue(any("no live author issue lock" in r for r in result["reasons"]))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_pr_sync_status.py

@@ -0,0 +1,338 @@
+"""Hermetic tests for PR synchronization / conflict-remediation lifecycle."""
+
+from __future__ import annotations
+
+import unittest
+
+from pr_sync_status import (
+    ACTION_AUTHOR_CONFLICT_REMEDIATION,
+    ACTION_BLOCKED,
+    ACTION_FRESH_REVIEW_REQUIRED,
+    ACTION_MERGE_NOW,
+    ACTION_UPDATE_BRANCH_BY_MERGE,
+    assess_post_update_head_transition,
+    assess_pr_sync_status,
+    assess_sequential_queue_step,
+    assess_update_pr_branch_preflight,
+    lease_usable_at_head,
+    merge_approval_usable_at_head,
+)
+
+
+def _sha(prefix: str) -> str:
+    return (prefix + "0" * 40)[:40]
+
+
+PR_HEAD = _sha("aaaaaaaa")
+BASE_HEAD = _sha("bbbbbbbb")
+NEW_HEAD = _sha("cccccccc")
+OLD_HEAD = _sha("dddddddd")
+
+
+def _base_kwargs(**overrides):
+    data = {
+        "host": "gitea.prgs.cc",
+        "org": "Scaled-Tech-Consulting",
+        "repo": "Gitea-Tools",
+        "pr_number": 100,
+        "pr_state": "open",
+        "source_branch": "fix/issue-100",
+        "pr_head_sha": PR_HEAD,
+        "base_head_sha": BASE_HEAD,
+        "commits_behind": 0,
+        "mergeable": True,
+        "has_conflicts": False,
+        "branch_protection_requires_current_base": False,
+        "approval_at_current_head": True,
+        "checks_status": "success",
+        "active_author_lock": True,
+        "active_reviewer_lease": False,
+        "active_merger_lease": False,
+    }
+    data.update(overrides)
+    return data
+
+
+class TestAssessPrSyncStatus(unittest.TestCase):
+    def test_approved_current_mergeable_merge_now(self):
+        result = assess_pr_sync_status(**_base_kwargs())
+        self.assertEqual(result["recommended_next_action"], ACTION_MERGE_NOW)
+        self.assertTrue(result["approval_valid_for_merge"])
+        self.assertFalse(result["stale_approval"])
+        self.assertEqual(result["pr_head_sha"], PR_HEAD)
+        self.assertEqual(result["base_head_sha"], BASE_HEAD)
+
+    def test_approved_outdated_update_not_required_merge_now(self):
+        result = assess_pr_sync_status(
+            **_base_kwargs(
+                commits_behind=3,
+                branch_protection_requires_current_base=False,
+            )
+        )
+        self.assertEqual(result["recommended_next_action"], ACTION_MERGE_NOW)
+        self.assertTrue(result["approval_valid_for_merge"])
+        self.assertTrue(any("does not require" in r for r in result["reasons"]))
+
+    def test_outdated_update_required_no_conflict(self):
+        result = assess_pr_sync_status(
+            **_base_kwargs(
+                commits_behind=2,
+                branch_protection_requires_current_base=True,
+                mergeable=True,
+                has_conflicts=False,
+            )
+        )
+        self.assertEqual(
+            result["recommended_next_action"], ACTION_UPDATE_BRANCH_BY_MERGE
+        )
+        self.assertFalse(result["approval_valid_for_merge"])
+        self.assertTrue(any("update_branch_by_merge" in r for r in result["reasons"]))
+
+    def test_outdated_has_conflicts_author_remediation(self):
+        result = assess_pr_sync_status(
+            **_base_kwargs(
+                commits_behind=5,
+                branch_protection_requires_current_base=True,
+                mergeable=False,
+                has_conflicts=True,
+            )
+        )
+        self.assertEqual(
+            result["recommended_next_action"], ACTION_AUTHOR_CONFLICT_REMEDIATION
+        )
+        self.assertFalse(result["approval_valid_for_merge"])
+
+    def test_stale_approval_fresh_review_required(self):
+        result = assess_pr_sync_status(
+            **_base_kwargs(
+                approval_at_current_head=False,
+                commits_behind=0,
+            )
+        )
+        self.assertEqual(
+            result["recommended_next_action"], ACTION_FRESH_REVIEW_REQUIRED
+        )
+        self.assertTrue(result["stale_approval"])
+        self.assertFalse(result["approval_valid_for_merge"])
+
+    def test_stale_prepared_verdict_cannot_cross_heads(self):
+        result = assess_pr_sync_status(
+            **_base_kwargs(
+                approval_at_current_head=True,
+                prepared_verdict_head_sha=OLD_HEAD,
+            )
+        )
+        self.assertEqual(
+            result["recommended_next_action"], ACTION_FRESH_REVIEW_REQUIRED
+        )
+        self.assertTrue(result["stale_prepared_verdict"])
+        self.assertFalse(result["approval_valid_for_merge"])
+
+    def test_missing_heads_fail_closed(self):
+        result = assess_pr_sync_status(
+            **_base_kwargs(pr_head_sha="short", base_head_sha=None)
+        )
+        self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
+        self.assertTrue(any("PR head" in r for r in result["reasons"]))
+
+    def test_closed_pr_blocked(self):
+        result = assess_pr_sync_status(**_base_kwargs(pr_state="closed"))
+        self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
+
+    def test_failed_checks_block_merge_now(self):
+        result = assess_pr_sync_status(**_base_kwargs(checks_status="failure"))
+        self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
+
+    def test_stale_approval_with_update_required_routes_update(self):
+        result = assess_pr_sync_status(
+            **_base_kwargs(
+                approval_at_current_head=False,
+                commits_behind=1,
+                branch_protection_requires_current_base=True,
+                mergeable=True,
+                has_conflicts=False,
+            )
+        )
+        self.assertEqual(
+            result["recommended_next_action"], ACTION_UPDATE_BRANCH_BY_MERGE
+        )
+
+
+class TestUpdatePrBranchPreflight(unittest.TestCase):
+    def _ok_kwargs(self, **overrides):
+        data = {
+            "role_kind": "author",
+            "expected_pr_head_sha": PR_HEAD,
+            "live_pr_head_sha": PR_HEAD,
+            "expected_base_head_sha": BASE_HEAD,
+            "live_base_head_sha": BASE_HEAD,
+            "has_conflicts": False,
+            "mergeable": True,
+            "style": "merge",
+            "has_author_lock": True,
+            "worktree_path": "/Users/x/Development/Gitea-Tools/branches/issue-100",
+            "worktree_owns_source_branch": True,
+            "control_checkout_clean": True,
+            "master_parity_ok": True,
+            "force_push": False,
+            "rebase": False,
+        }
+        data.update(overrides)
+        return data
+
+    def test_author_allowed_when_preflight_clean(self):
+        result = assess_update_pr_branch_preflight(**self._ok_kwargs())
+        self.assertTrue(result["mutation_allowed"])
+        self.assertFalse(result["head_race"])
+        self.assertFalse(result["base_race"])
+
+    def test_head_race_fail_closed(self):
+        result = assess_update_pr_branch_preflight(
+            **self._ok_kwargs(live_pr_head_sha=NEW_HEAD)
+        )
+        self.assertFalse(result["mutation_allowed"])
+        self.assertTrue(result["head_race"])
+        self.assertTrue(any("PR head race" in r for r in result["reasons"]))
+
+    def test_base_race_fail_closed(self):
+        result = assess_update_pr_branch_preflight(
+            **self._ok_kwargs(live_base_head_sha=NEW_HEAD)
+        )
+        self.assertFalse(result["mutation_allowed"])
+        self.assertTrue(result["base_race"])
+        self.assertTrue(any("base head race" in r for r in result["reasons"]))
+
+    def test_conflicts_structured_handoff_no_mutation(self):
+        result = assess_update_pr_branch_preflight(
+            **self._ok_kwargs(has_conflicts=True, mergeable=False)
+        )
+        self.assertFalse(result["mutation_allowed"])
+        self.assertTrue(result["author_remediation_handoff"])
+        self.assertTrue(any("conflicts" in r.lower() for r in result["reasons"]))
+
+    def test_reviewer_denied(self):
+        result = assess_update_pr_branch_preflight(
+            **self._ok_kwargs(role_kind="reviewer")
+        )
+        self.assertFalse(result["mutation_allowed"])
+        self.assertTrue(any("author-only" in r for r in result["reasons"]))
+
+    def test_merger_denied(self):
+        result = assess_update_pr_branch_preflight(
+            **self._ok_kwargs(role_kind="merger")
+        )
+        self.assertFalse(result["mutation_allowed"])
+        self.assertTrue(any("author-only" in r for r in result["reasons"]))
+
+    def test_no_force_push(self):
+        result = assess_update_pr_branch_preflight(
+            **self._ok_kwargs(force_push=True)
+        )
+        self.assertFalse(result["mutation_allowed"])
+        self.assertTrue(any("force-push" in r for r in result["reasons"]))
+
+    def test_no_rebase(self):
+        result = assess_update_pr_branch_preflight(
+            **self._ok_kwargs(style="rebase", rebase=True)
+        )
+        self.assertFalse(result["mutation_allowed"])
+        self.assertTrue(any("rebase" in r for r in result["reasons"]))
+
+    def test_missing_lock_fail_closed(self):
+        result = assess_update_pr_branch_preflight(
+            **self._ok_kwargs(has_author_lock=False)
+        )
+        self.assertFalse(result["mutation_allowed"])
+
+    def test_worktree_not_under_branches(self):
+        result = assess_update_pr_branch_preflight(
+            **self._ok_kwargs(worktree_path="/tmp/not-a-branches-path")
+        )
+        self.assertFalse(result["mutation_allowed"])
+        self.assertTrue(any("branches/" in r for r in result["reasons"]))
+
+
+class TestPostUpdateHeadTransition(unittest.TestCase):
+    def test_update_invalidates_approval_and_requires_fresh_review(self):
+        result = assess_post_update_head_transition(
+            former_pr_head_sha=PR_HEAD,
+            new_pr_head_sha=NEW_HEAD,
+            former_approval_head_sha=PR_HEAD,
+            former_reviewer_lease_head_sha=PR_HEAD,
+            former_merger_lease_head_sha=PR_HEAD,
+            prepared_verdict_head_sha=PR_HEAD,
+        )
+        self.assertTrue(result["head_changed"])
+        self.assertTrue(result["approval_invalidated"])
+        self.assertTrue(result["reviewer_lease_superseded"])
+        self.assertTrue(result["merger_lease_superseded"])
+        self.assertTrue(result["prepared_verdict_invalidated"])
+        self.assertEqual(
+            result["recommended_next_action"], ACTION_FRESH_REVIEW_REQUIRED
+        )
+
+    def test_same_head_unexpected(self):
+        result = assess_post_update_head_transition(
+            former_pr_head_sha=PR_HEAD,
+            new_pr_head_sha=PR_HEAD,
+        )
+        self.assertFalse(result["head_changed"])
+        self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
+
+
+class TestStaleArtifacts(unittest.TestCase):
+    def test_stale_approval_cannot_authorize_merge(self):
+        result = merge_approval_usable_at_head(
+            current_head_sha=NEW_HEAD,
+            approved_head_sha=PR_HEAD,
+        )
+        self.assertFalse(result["approval_usable"])
+
+    def test_fresh_approval_usable(self):
+        result = merge_approval_usable_at_head(
+            current_head_sha=NEW_HEAD,
+            approved_head_sha=NEW_HEAD,
+        )
+        self.assertTrue(result["approval_usable"])
+
+    def test_stale_reviewer_lease_cannot_cross_heads(self):
+        result = lease_usable_at_head(
+            lease_kind="reviewer",
+            lease_head_sha=PR_HEAD,
+            current_head_sha=NEW_HEAD,
+        )
+        self.assertFalse(result["lease_usable"])
+
+    def test_stale_merger_lease_cannot_cross_heads(self):
+        result = lease_usable_at_head(
+            lease_kind="merger",
+            lease_head_sha=PR_HEAD,
+            current_head_sha=NEW_HEAD,
+        )
+        self.assertFalse(result["lease_usable"])
+
+
+class TestSequentialQueue(unittest.TestCase):
+    def test_reassess_after_merge_and_master_refresh(self):
+        result = assess_sequential_queue_step(
+            just_merged=True,
+            master_refreshed=True,
+            next_pr_number=101,
+        )
+        self.assertTrue(result["reassess_allowed"])
+        self.assertTrue(result["process_next"])
+        self.assertTrue(result["post_merge_cleanup_handoff"])
+        self.assertEqual(result["next_pr_number"], 101)
+
+    def test_block_reassess_without_master_refresh(self):
+        result = assess_sequential_queue_step(
+            just_merged=True,
+            master_refreshed=False,
+            next_pr_number=101,
+        )
+        self.assertFalse(result["reassess_allowed"])
+        self.assertTrue(any("master must be refreshed" in r for r in result["reasons"]))
+
+
+if __name__ == "__main__":
+    unittest.main()