diff --git a/anti_stomp_preflight.py b/anti_stomp_preflight.py index 532db08..6c99e3d 100644 --- a/anti_stomp_preflight.py +++ b/anti_stomp_preflight.py @@ -355,6 +355,10 @@ def assess_anti_stomp_preflight( root_head_sha: str | None = None, root_porcelain: str | None = None, remote_master_sha: str | None = None, + # #983: the tracking integration ref the SHA above came from, so root-checkout + # contamination names the ref actually compared instead of a hardcoded + # 'prgs/master'. None preserves the previous generic wording. + remote_master_ref: str | None = None, check_root_checkout: bool = True, check_worktree: bool = True, create_issue_bootstrap_assessment: dict[str, Any] | None = None, @@ -553,6 +557,7 @@ def assess_anti_stomp_preflight( remote_master_sha=remote_master_sha, resolved_role=req_role or role, actual_role=role, + remote_master_ref=remote_master_ref, ) checks["root_checkout"] = { "block": bool(root_assessment.get("block")), diff --git a/canonical_repository_root.py b/canonical_repository_root.py index 885821c..9983c2f 100644 --- a/canonical_repository_root.py +++ b/canonical_repository_root.py @@ -37,6 +37,34 @@ CANONICAL_ROOT_ENV = "GITEA_CANONICAL_REPOSITORY_ROOT" # Candidate git remote names probed when deriving repository identity. _IDENTITY_REMOTE_CANDIDATES = ("prgs", "origin", "dadeschools", "mdcps") +# Fallback integration-branch names, probed only when the checkout declares no +# configured upstream (#983). Mirrors the stable base branches recognised +# elsewhere in the workflow (``stacked_pr_support``, ``root_checkout_guard``). +# The fallback is deliberately *not* ordered-first-wins: when more than one of +# these refs exists and the checkout records no upstream, the integration branch +# is genuinely ambiguous and resolution fails closed instead of guessing. +INTEGRATION_BRANCH_CANDIDATES: tuple[str, ...] = ("master", "main", "dev") + +# Sources for a *proven* base-ref derivation (#983 B1). +# +# ``refs/remotes//HEAD`` is deliberately absent from this list. It is a +# local symbolic-ref *cache* written once at clone time and refreshed only by an +# explicit ``git remote set-head``; an ordinary fetch never updates it. When the +# upstream default branch changes afterwards the cache keeps naming the old +# branch, so trusting it derives the wrong integration branch for a checkout +# that is sitting exactly on its tip. The cache is still read, but only as a +# corroborating observation reported back to the caller — never as an authority, +# and never as a tie-breaker between otherwise ambiguous candidates. +BASE_REF_SOURCE_CONFIGURED_UPSTREAM = "configured_branch_upstream" +BASE_REF_SOURCE_UNIQUE_CANDIDATE = "unique_integration_branch_ref" + +# Reason codes for base-ref derivation outcomes (#983), so callers and tests can +# assert the refusal cause instead of string-matching prose. +DENY_NO_IDENTITY_REMOTE = "no_identity_remote" +DENY_AMBIGUOUS_REMOTE = "ambiguous_identity_remote" +DENY_AMBIGUOUS_BASE_BRANCH = "ambiguous_integration_branch" +DENY_NO_BASE_BRANCH = "no_integration_branch_ref" + # Repository-authority modes (#973 B10). Exactly two values are supported. # ``mode`` selects how repository authority is established, so an unrecognised # value must never be normalised onto one of these: aliasing a trusted mode is @@ -120,17 +148,15 @@ def resolve_repo_toplevel(path: str) -> str | None: return os.path.realpath(top) if top else None -def repository_identity_slug(path: str, *, remote: str | None = None) -> str | None: - """``owner/repository`` derived from a git remote configured at *path*. +def _identity_remote_candidates(path: str, remote: str | None) -> list[str]: + """Ordered remote names to probe for identity at *path*. - Tries the caller-named remote first, then a small set of known remote names, - then whatever remote the repository actually has. Returns None when no remote - URL is parseable (identity cannot be proven). + Names are used verbatim — never case-folded. Git config subsection names are + case-sensitive, so a repository whose remote is ``MDCPS`` is reached only by + the exact string ``MDCPS``; the lowercase entry in + :data:`_IDENTITY_REMOTE_CANDIDATES` simply does not resolve, and the exact + name arrives from ``git remote`` below (#983). """ - text = (path or "").strip() - if not text: - return None - ordered: list[str] = [] for name in (remote, *_IDENTITY_REMOTE_CANDIDATES): clean = (name or "").strip() @@ -139,7 +165,7 @@ def repository_identity_slug(path: str, *, remote: str | None = None) -> str | N try: listed = subprocess.run( - ["git", "-C", text, "remote"], + ["git", "-C", path, "remote"], capture_output=True, text=True, check=True, @@ -149,11 +175,20 @@ def repository_identity_slug(path: str, *, remote: str | None = None) -> str | N for name in listed: if name and name not in ordered: ordered.append(name) + return ordered - for name in ordered: + +def _configured_remote_identities(path: str, remote: str | None) -> list[tuple[str, str]]: + """``(remote_name, owner/repository)`` for every probe name that resolves. + + Remote names are returned exactly as configured so downstream tracking refs + (``refs/remotes//``) address the real ref (#983). + """ + found: list[tuple[str, str]] = [] + for name in _identity_remote_candidates(path, remote): try: url = subprocess.run( - ["git", "-C", text, "remote", "get-url", name], + ["git", "-C", path, "remote", "get-url", name], capture_output=True, text=True, check=True, @@ -162,8 +197,415 @@ def repository_identity_slug(path: str, *, remote: str | None = None) -> str | N continue parsed = remote_repo_guard.parse_org_repo_from_remote_url(url) if parsed: - return f"{parsed[0]}/{parsed[1]}" - return None + found.append((name, f"{parsed[0]}/{parsed[1]}")) + return found + + +def _configured_branch_upstream(path: str) -> tuple[str | None, str | None]: + """``(remote_name, branch)`` the checked-out branch is configured to track. + + Reads ``branch..remote`` and ``branch..merge`` — the + checkout's own explicitly configured integration target, equivalent to + ``@{upstream}``. Unlike ``refs/remotes//HEAD`` this is not a + clone-time cache: it is written when the branch is set up to track an + upstream and rewritten whenever that tracking changes, so it states what the + checkout actually integrates onto today (#983 B1). + + Returns ``(None, None)`` on a detached HEAD or an untracked branch. Names are + returned verbatim; git remote names are case-sensitive. + """ + text = (path or "").strip() + if not text: + return None, None + + branch = _git_read(text, "symbolic-ref", "--quiet", "--short", "HEAD") + if not branch: + return None, None + + remote = _git_read(text, "config", "--get", f"branch.{branch}.remote") + merge = _git_read(text, "config", "--get", f"branch.{branch}.merge") + if not remote or not merge: + return None, None + + prefix = "refs/heads/" + upstream_branch = merge[len(prefix):].strip() if merge.startswith(prefix) else merge.strip() + if not upstream_branch: + return None, None + return remote, upstream_branch + + +def _cached_remote_head_branch(path: str, remote: str) -> str | None: + """Branch named by the *cached* ``refs/remotes//HEAD`` symref. + + Read for observability only. This value is never authoritative (see + :data:`BASE_REF_SOURCE_CONFIGURED_UPSTREAM`); it is surfaced so an operator + can see that the local cache disagrees with the configured upstream, and so + a refusal can name the misleading signal explicitly. + """ + prefix = f"refs/remotes/{remote}/" + symref = _git_read(path, "symbolic-ref", "--quiet", f"{prefix}HEAD") + if not symref or not symref.startswith(prefix): + return None + branch = symref[len(prefix):].strip() + return branch or None + + +def assess_identity_remote( + path: str, *, explicit_remote: str | None = None +) -> dict: + """Resolve the identity remote, failing closed when the target is ambiguous. + + This is the ambiguity-aware counterpart to :func:`resolve_identity_remote`, + and the reason gating and reporting can no longer disagree (#983 B2). + + *explicit_remote* is a **caller-supplied disambiguation** and nothing else. + It must come from an operator or an explicitly sanctioned repository + context; a value this module inferred while probing must never be handed + back in through it, because doing so re-labels an internal first-wins guess + as deliberate caller intent and silently suppresses the ambiguity gate. + + Decision table: + + * No remote yields a parseable identity -> :data:`DENY_NO_IDENTITY_REMOTE`. + * *explicit_remote* names one of the resolving remotes -> that remote is + authoritative, ``explicit`` True. + * Otherwise, when every resolving remote claims the **same** repository the + target is unambiguous and is accepted, ``explicit`` False. The remote + named by the checkout's configured upstream is preferred among equals so + the choice is deterministic rather than probe-order dependent. + * Otherwise distinct remotes claim different repositories and there is no + sanctioned disambiguation -> :data:`DENY_AMBIGUOUS_REMOTE`. + + Returns a dict with ``remote``, ``slug``, ``identities``, ``ambiguous``, + ``explicit``, ``reason_code`` and ``reasons``. ``remote``/``slug`` are None + on any refusal, so a caller cannot report a repository the gate refuses. + """ + text = (path or "").strip() + result: dict = { + "remote": None, + "slug": None, + "identities": [], + "ambiguous": False, + "explicit": False, + "reason_code": None, + "reasons": [], + } + if not text: + result["reason_code"] = DENY_NO_IDENTITY_REMOTE + result["reasons"].append( + "no repository path supplied for identity-remote resolution (fail closed)" + ) + return result + + named = (explicit_remote or "").strip() or None + identities = _configured_remote_identities(text, named) + result["identities"] = list(identities) + if not identities: + result["reason_code"] = DENY_NO_IDENTITY_REMOTE + result["reasons"].append( + f"no git remote at '{text}' yields a parseable repository identity " + "(fail closed)" + ) + return result + + if named: + for name, slug in identities: + if name == named: + result["remote"] = name + result["slug"] = slug + result["explicit"] = True + return result + + distinct = {slug for _, slug in identities} + if len(distinct) > 1: + listed = ", ".join(f"{name} -> {slug}" for name, slug in identities) + result["ambiguous"] = True + result["reason_code"] = DENY_AMBIGUOUS_REMOTE + detail = ( + f"explicitly named remote '{named}' does not resolve a repository identity " + "there, so it cannot disambiguate; " + if named + else "" + ) + result["reasons"].append( + f"ambiguous repository identity at '{text}': {detail}remotes resolve to " + f"different repositories ({listed}); no single authoritative target can " + "be established (fail closed)" + ) + return result + + # One repository, possibly reachable through several remote names (a mirror). + # Prefer the remote the checkout is actually configured to track so the + # choice is deterministic instead of probe-order dependent. + upstream_remote, _ = _configured_branch_upstream(text) + chosen = identities[0] + if upstream_remote: + for entry in identities: + if entry[0] == upstream_remote: + chosen = entry + break + result["remote"], result["slug"] = chosen + return result + + +def resolve_identity_remote( + path: str, *, remote: str | None = None +) -> tuple[str | None, str | None]: + """``(remote_name, owner/repository)`` for the remote that proves identity. + + The remote *name* is the piece historically thrown away by + :func:`repository_identity_slug`, even though resolving the slug already + required discovering it. Cross-repository base-ref derivation needs that + name to build ``refs/remotes//``, so it is now returned + rather than discarded (#983). Returns ``(None, None)`` when no remote URL is + parseable (identity cannot be proven). + + First-wins by design: this is the identity lookup behind + :func:`repository_identity_slug` and the #706/#973 canonical-root + validation, which compare an observed slug against an independently trusted + expected slug and therefore do not need an ambiguity verdict. Callers that + *derive* a target rather than validate one — the mutation guard and the + parity report — must use :func:`assess_identity_remote`, which fails closed + on ambiguity (#983 B2). + """ + text = (path or "").strip() + if not text: + return None, None + for name, slug in _configured_remote_identities(text, remote): + return name, slug + return None, None + + +def repository_identity_slug(path: str, *, remote: str | None = None) -> str | None: + """``owner/repository`` derived from a git remote configured at *path*. + + Tries the caller-named remote first, then a small set of known remote names, + then whatever remote the repository actually has. Returns None when no remote + URL is parseable (identity cannot be proven). + """ + return resolve_identity_remote(path, remote=remote)[1] + + +def _base_ref_result( + *, + proven: bool, + reasons: list[str], + remote: str | None = None, + branch: str | None = None, + repository_slug: str | None = None, + source: str | None = None, + reason_code: str | None = None, + identity_explicit: bool = False, + configured_upstream_remote: str | None = None, + configured_upstream_branch: str | None = None, + cached_remote_head_branch: str | None = None, +) -> dict: + """Build the base-ref derivation payload. + + ``tracking_refs`` is the ordered probe tuple downstream guards hand to + ``git rev-parse``: the ``/`` shorthand first, then the fully + qualified ``refs/remotes//``. It is empty whenever the + derivation is not ``proven``, so an unresolved target can never be probed + against some other repository's ref. + + ``cached_remote_head_branch`` reports what the local + ``refs/remotes//HEAD`` cache claims, and + ``cached_remote_head_conflicts`` whether that claim disagrees with the branch + actually derived. Both are observability only: the cache never decides the + outcome (#983 B1). + """ + tracking_ref = f"refs/remotes/{remote}/{branch}" if proven and remote and branch else None + tracking_refs: tuple[str, ...] = ( + (f"{remote}/{branch}", tracking_ref) if tracking_ref else () + ) + return { + "proven": proven, + "block": not proven, + "remote": remote, + "branch": branch, + "repository_slug": repository_slug, + "tracking_ref": tracking_ref, + "tracking_refs": tracking_refs, + "source": source, + "reason_code": reason_code, + "identity_explicit": identity_explicit, + "configured_upstream_remote": configured_upstream_remote, + "configured_upstream_branch": configured_upstream_branch, + "cached_remote_head_branch": cached_remote_head_branch, + "cached_remote_head_conflicts": bool( + cached_remote_head_branch and branch and cached_remote_head_branch != branch + ), + "reasons": list(reasons), + } + + +def _git_read(path: str, *args: str) -> str | None: + """Run a read-only git command in *path*; ``None`` on any failure.""" + try: + res = subprocess.run( + ["git", "-C", path, *args], + capture_output=True, + text=True, + check=False, + ) + except Exception: + return None + if res.returncode != 0: + return None + return (res.stdout or "").strip() or None + + +def _ref_exists(path: str, ref: str) -> bool: + """Whether *ref* resolves in the checkout at *path*.""" + try: + res = subprocess.run( + ["git", "-C", path, "rev-parse", "--verify", "--quiet", ref], + capture_output=True, + text=True, + check=False, + ) + except Exception: + return False + return res.returncode == 0 and bool((res.stdout or "").strip()) + + +def resolve_target_base_ref(path: str, *, explicit_remote: str | None = None) -> dict: + """Derive the authoritative integration base ref for the checkout at *path*. + + This is the single resolved target shared by cross-repository mutation + gating and parity reporting, so the two can never disagree about which + commit a checkout is supposed to match (#983). + + *explicit_remote* is a caller-supplied disambiguation only; see + :func:`assess_identity_remote` for why an internally inferred remote must + never be passed back in here (#983 B2). + + Resolution order: + + 1. **Identity remote** — via :func:`assess_identity_remote`, with its exact + configured case preserved (``MDCPS`` stays ``MDCPS``). Ambiguous identity + fails closed here rather than resolving to whichever remote probed first. + 2. **The checkout's configured upstream** — ``branch..remote`` plus + ``branch..merge``, accepted when it names the identity remote + and its remote-tracking ref actually exists. This is the checkout's own + declaration of what it integrates onto, and unlike the remote-HEAD cache + it is rewritten whenever that tracking changes. + 3. **Fallback** — only when the checkout declares no usable upstream, + exactly one of :data:`INTEGRATION_BRANCH_CANDIDATES` present as a + remote-tracking ref. + + ``refs/remotes//HEAD`` is **not** a step. It is read for reporting + (``cached_remote_head_branch`` / ``cached_remote_head_conflicts``) and never + decides the branch, because it is a clone-time cache that an ordinary fetch + does not refresh: a checkout whose upstream default moved on still has the + old branch cached, and trusting it gates that checkout against a ref it does + not integrate onto (#983 B1). + + Fails closed — ``proven`` False, empty ``tracking_refs``, and a + ``reason_code`` — when identity is unprovable, when distinct remotes claim + different repositories, when several candidate branches exist with no + configured upstream, or when no candidate exists at all. Nothing here + invents a branch, writes a ref, runs a fetch, or falls back to another + repository's base. Every ``proven`` result names a tracking ref that + resolves in this checkout. + """ + text = (path or "").strip() + if not text: + return _base_ref_result( + proven=False, + reasons=["no repository path supplied for base-ref derivation (fail closed)"], + reason_code=DENY_NO_IDENTITY_REMOTE, + ) + + identity = assess_identity_remote(text, explicit_remote=explicit_remote) + remote_name, slug = identity["remote"], identity["slug"] + if not remote_name: + return _base_ref_result( + proven=False, + reasons=list(identity["reasons"]), + reason_code=identity["reason_code"], + identity_explicit=bool(identity["explicit"]), + ) + + prefix = f"refs/remotes/{remote_name}/" + cached = _cached_remote_head_branch(text, remote_name) + upstream_remote, upstream_branch = _configured_branch_upstream(text) + common = { + "repository_slug": slug, + "identity_explicit": bool(identity["explicit"]), + "configured_upstream_remote": upstream_remote, + "configured_upstream_branch": upstream_branch, + "cached_remote_head_branch": cached, + } + + # 2. The checkout's configured upstream, when it belongs to the identity + # remote and its tracking ref is actually present. Requiring the ref to + # exist keeps 'proven' honest: a proven target is always resolvable. + if ( + upstream_remote == remote_name + and upstream_branch + and _ref_exists(text, f"{prefix}{upstream_branch}") + ): + return _base_ref_result( + proven=True, + reasons=[], + remote=remote_name, + branch=upstream_branch, + source=BASE_REF_SOURCE_CONFIGURED_UPSTREAM, + **common, + ) + + # 3. Exactly one known integration branch present as a remote-tracking ref. + present = [ + candidate + for candidate in INTEGRATION_BRANCH_CANDIDATES + if _ref_exists(text, f"{prefix}{candidate}") + ] + if len(present) == 1: + return _base_ref_result( + proven=True, + reasons=[], + remote=remote_name, + branch=present[0], + source=BASE_REF_SOURCE_UNIQUE_CANDIDATE, + **common, + ) + + # The cached remote HEAD is named in the refusal so the operator can see the + # signal that looks authoritative but is not, and is told the read-only fix. + cache_note = ( + f" the cached '{prefix}HEAD' names '{cached}', but that cache is written at " + "clone time and is not refreshed by fetch, so it cannot break the tie;" + if cached + else "" + ) + remedy = ( + f" Configure the checkout's upstream (git branch --set-upstream-to={remote_name}/" + ") so the integration target is declared rather than guessed." + ) + if len(present) > 1: + return _base_ref_result( + proven=False, + reasons=[ + f"the checkout at '{text}' declares no upstream on remote " + f"'{remote_name}' and several integration branches exist " + f"({', '.join(present)});{cache_note} the integration base ref is " + f"ambiguous (fail closed).{remedy}" + ], + reason_code=DENY_AMBIGUOUS_BASE_BRANCH, + **common, + ) + return _base_ref_result( + proven=False, + reasons=[ + f"the checkout at '{text}' declares no upstream on remote '{remote_name}' " + f"and none of {'/'.join(INTEGRATION_BRANCH_CANDIDATES)} exists as a " + f"remote-tracking ref under '{prefix}';{cache_note} the integration base " + f"ref cannot be derived (fail closed; no fetch is performed here).{remedy}" + ], + reason_code=DENY_NO_BASE_BRANCH, + **common, + ) def assess_canonical_repository_root( diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index e365da7..c4f29f4 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -1052,9 +1052,16 @@ def _create_issue_bootstrap_assessment( git_state = issue_lock_worktree.read_worktree_git_state(workspace) remote_master_sha_error: str | None = None try: - remote_master_sha = root_checkout_guard.resolve_remote_master_sha( + # #983: consume the resolved-target state so an *underivable* base ref + # reaches the assessor as named missing evidence rather than a bare + # None, which the bootstrap would otherwise report only as "live + # master tip is unknown" with no cause. + _base_state = root_checkout_guard.resolve_remote_master_ref_state( ctx["canonical_repo_root"] ) + remote_master_sha = _base_state["sha"] + if not remote_master_sha and _base_state.get("reasons"): + remote_master_sha_error = "; ".join(_base_state["reasons"]) except Exception as exc: remote_master_sha = None remote_master_sha_error = ( @@ -1082,9 +1089,14 @@ def _create_issue_bootstrap_assessment( # closed instead of proceeding without base-equivalence proof. remote_master_sha_error: str | None = None try: - remote_master_sha = root_checkout_guard.resolve_remote_master_sha( + # #983: same resolved-target consumption as the author bootstrap above — + # a target whose base ref cannot be derived must say why. + _base_state = root_checkout_guard.resolve_remote_master_ref_state( ctx["canonical_repo_root"] ) + remote_master_sha = _base_state["sha"] + if not remote_master_sha and _base_state.get("reasons"): + remote_master_sha_error = "; ".join(_base_state["reasons"]) except Exception as exc: remote_master_sha = None remote_master_sha_error = f"{type(exc).__name__}: {exc}".strip() or "resolver failed" @@ -1303,7 +1315,12 @@ def _run_anti_stomp_preflight( workspace = ctx["workspace_path"] canonical_root = ctx["canonical_repo_root"] git_state = issue_lock_worktree.read_worktree_git_state(canonical_root) - remote_master_sha = root_checkout_guard.resolve_remote_master_sha(canonical_root) + # #983: derive the target base ref for this repository and carry the ref + # itself alongside the SHA, so the anti-stomp root-checkout check reports the + # ref it actually compared. + _root_base_state = root_checkout_guard.resolve_remote_master_ref_state(canonical_root) + remote_master_sha = _root_base_state["sha"] + remote_master_ref = _root_base_state.get("ref") # Repo/org facts (best-effort; explicit org/repo when provided). resolved_org = org @@ -1432,6 +1449,7 @@ def _run_anti_stomp_preflight( root_head_sha=git_state.get("head_sha"), root_porcelain=git_state.get("porcelain_status") or "", remote_master_sha=remote_master_sha, + remote_master_ref=remote_master_ref, startup_head=startup_head, current_code_head=current_code_head, lease_required=lease_required, @@ -1987,7 +2005,11 @@ def _enforce_root_checkout_guard(worktree_path: str | None = None) -> None: canonical_root = ctx["canonical_repo_root"] workspace = ctx["workspace_path"] git_state = issue_lock_worktree.read_worktree_git_state(canonical_root) - remote_master_sha = root_checkout_guard.resolve_remote_master_sha(canonical_root) + # #983: consume the resolved target so the contamination message names the + # ref that was actually compared (refs/remotes//) instead of + # a hardcoded 'prgs/master' the target repository may not have. + base_state = root_checkout_guard.resolve_remote_master_ref_state(canonical_root) + remote_master_sha = base_state["sha"] assessment = root_checkout_guard.assess_root_checkout_guard( workspace_path=workspace, canonical_repo_root=canonical_root, @@ -1997,6 +2019,7 @@ def _enforce_root_checkout_guard(worktree_path: str | None = None) -> None: remote_master_sha=remote_master_sha, resolved_role=_preflight_resolved_role, actual_role=_actual_profile_role(), + remote_master_ref=base_state.get("ref"), ) if assessment["block"]: raise RuntimeError(root_checkout_guard.format_root_checkout_guard_error(assessment)) diff --git a/master_parity_gate.py b/master_parity_gate.py index 2e38c66..a55faf3 100644 --- a/master_parity_gate.py +++ b/master_parity_gate.py @@ -378,6 +378,11 @@ def format_parity(assessment: dict) -> str: # feeds the mutation gate and never changes startup_head/current_head. # --------------------------------------------------------------------------- +# Legacy hardcoded target tracking ref. Retained for callers that still pass an +# explicit ref, but no longer the default: assuming a remote named ``origin`` +# read an unrelated (often orphaned) remote-tracking ref in any checkout whose +# remote is named something else, and reported the target stale against a commit +# from a remote that may no longer even be configured (#983). DEFAULT_TARGET_TRACKING_REF = "refs/remotes/origin/master" @@ -408,7 +413,7 @@ def assess_target_repository_parity( *, canonical_root: str | None, source: str | None, - tracking_ref: str = DEFAULT_TARGET_TRACKING_REF, + tracking_ref: str | None = None, ) -> dict: """Assess the configured cross-repository target checkout. @@ -421,6 +426,11 @@ def assess_target_repository_parity( An unconfigured namespace is ``configured=False`` and never ``stale`` — the single-repository default has no second dimension to be stale about. A configured root that cannot be read is ``determinable=False`` with reasons. + + *tracking_ref* defaults to None, meaning **derive the target from the + repository itself** through the same resolver the mutation guard uses, so + gating and reporting can never disagree about which ref is authoritative + (#983). Passing an explicit ref preserves the previous behaviour verbatim. """ result = { "configured": bool(canonical_root), @@ -429,9 +439,13 @@ def assess_target_repository_parity( "repository_slug": None, "checkout_head": None, "tracking_ref": tracking_ref, + "base_remote": None, + "base_branch": None, + "tracking_ref_source": "explicit_tracking_ref" if tracking_ref else None, "remote_tracking_head": None, "determinable": False, "stale": False, + "reason_code": None, "reasons": [], } if not canonical_root: @@ -463,18 +477,53 @@ def assess_target_repository_parity( result["checkout_head"] = head result["determinable"] = True - remote_url = _git_capture(canonical_root, "remote", "get-url", "origin") - if remote_url: - # Local import keeps this module dependency-light for its startup role. - import remote_repo_guard + # Local import keeps this module dependency-light for its startup role. + import canonical_repository_root as _crr - parsed = remote_repo_guard.parse_org_repo_from_remote_url(remote_url) - if parsed: - result["repository_slug"] = f"{parsed[0]}/{parsed[1]}" - if not result["repository_slug"]: - result["reasons"].append( - "target repository identity could not be derived from its git remote" + # #983: identity comes from whichever remote actually proves it, not from a + # remote assumed to be named 'origin'. In a checkout whose only remote is + # 'prgs', the old lookup failed outright and reported the identity as + # underivable while a leftover refs/remotes/origin/master still resolved. + # + # #983 B2: this must be the *ambiguity-aware* resolver. The first-wins + # `resolve_identity_remote` picks whichever remote probes first, so on a + # target where distinct remotes claim different repositories the report + # confidently named one of them while the mutation gate refused the same + # target — gating and reporting evaluating different repositories, which is + # precisely the divergence this issue exists to end. + identity = _crr.assess_identity_remote(canonical_root) + result["repository_slug"] = identity["slug"] + if not identity["slug"]: + result["reason_code"] = identity["reason_code"] + result["reasons"].extend( + identity["reasons"] + or ["target repository identity could not be derived from its git remote"] ) + if identity["ambiguous"]: + # An ambiguous target has no single authoritative base, so reporting one + # would be a guess. Fail closed here exactly as the gate does. + return result + + # Identity is otherwise resolved independently of the base ref: a target that + # has never been fetched still has a provable repository identity, and + # reporting it as unidentifiable would lose real information over an + # unrelated missing ref. + if not tracking_ref: + # No remote argument. The identity remote resolved just above was + # *inferred here*, and feeding it back in would tell the resolver a + # caller had explicitly disambiguated the repository, suppressing its + # ambiguity gate (#983 B2). Only an operator-supplied remote may do that, + # and this call site has none. + base = _crr.resolve_target_base_ref(canonical_root) + if not base.get("proven"): + result["reason_code"] = base.get("reason_code") + result["reasons"].extend(base.get("reasons") or []) + return result + tracking_ref = base["tracking_ref"] + result["tracking_ref"] = tracking_ref + result["tracking_ref_source"] = base.get("source") + result["base_remote"] = base.get("remote") + result["base_branch"] = base.get("branch") tracking_head = _git_capture(canonical_root, "rev-parse", tracking_ref) if not tracking_head: diff --git a/root_checkout_guard.py b/root_checkout_guard.py index 03af9e7..bac50e9 100644 --- a/root_checkout_guard.py +++ b/root_checkout_guard.py @@ -1,9 +1,16 @@ """Root checkout guard (#475). -The project root checkout is the stable control checkout on master/prgs/master. -Author/reviewer/merge flows must fail closed when the control checkout is -contaminated (wrong branch, detached HEAD, dirty, or HEAD behind/ahead of -prgs/master). Isolated ``branches/...`` worktrees remain allowed. +The project root checkout is the stable control checkout on its integration +branch. Author/reviewer/merge flows must fail closed when the control checkout +is contaminated (wrong branch, detached HEAD, dirty, or HEAD behind/ahead of the +tracking integration ref). Isolated ``branches/...`` worktrees remain allowed. + +The tracking ref is *derived per repository* (#983) rather than assumed to be +``prgs/master``: a namespace bound to another repository — say remote ``MDCPS`` +on branch ``dev`` — is gated against ``refs/remotes/MDCPS/dev``. Derivation is +delegated to :mod:`canonical_repository_root`, the authoritative +repository/context resolver, so gating and parity reporting share one resolved +target instead of maintaining two disagreeing hardcoded defaults. """ from __future__ import annotations @@ -11,6 +18,7 @@ from __future__ import annotations import os import subprocess +import canonical_repository_root from author_mutation_worktree import is_path_under_branches from reviewer_worktree import parse_dirty_tracked_files @@ -20,19 +28,65 @@ REMEDIATION = ( ) BASE_BRANCHES = frozenset({"master", "main", "dev"}) + +# Legacy PRGS-specific probe order. Retained only for callers that pass an +# explicit ``remote_refs`` override; it is no longer the silent default, because +# inheriting it in a non-PRGS checkout compared that checkout against a ref it +# can never have (#983). REMOTE_MASTER_REFS = ("prgs/master", "refs/remotes/prgs/master") +def _derive_probe_refs(root: str, explicit_remote: str | None) -> dict: + """Derive the ordered tracking refs to probe for *root*. + + Returns the derivation payload from :mod:`canonical_repository_root` plus a + ``refs`` tuple, which is empty when the target is not provable. + """ + derived = canonical_repository_root.resolve_target_base_ref( + root, explicit_remote=explicit_remote + ) + return { + "refs": tuple(derived.get("tracking_refs") or ()), + "remote": derived.get("remote"), + "branch": derived.get("branch"), + "source": derived.get("source"), + "proven": bool(derived.get("proven")), + "reason_code": derived.get("reason_code"), + "reasons": list(derived.get("reasons") or []), + "cached_remote_head_branch": derived.get("cached_remote_head_branch"), + "cached_remote_head_conflicts": bool(derived.get("cached_remote_head_conflicts")), + } + + def resolve_remote_master_sha( canonical_repo_root: str, *, remote_refs: tuple[str, ...] | None = None, + explicit_remote: str | None = None, ) -> str | None: - """Return the commit SHA for the tracking master ref when available.""" + """Return the commit SHA for the tracking integration ref when available. + + This remains the single place that turns a ref into a SHA. Returns None when + the target cannot be derived or resolved — exactly what this function already + returned when ``rev-parse`` failed. Callers that must fail closed on missing + evidence (the #749/#757 bootstrap path) surface that None as *missing + evidence*, never as "no constraint". + + *explicit_remote* is a caller-supplied disambiguation and is named that way + deliberately: an internally inferred remote handed back in would suppress the + resolver's ambiguity gate (#983 B2). No production caller supplies it. + """ root = (canonical_repo_root or "").strip() if not root: return None - for ref in remote_refs or REMOTE_MASTER_REFS: + if remote_refs: + probe: tuple[str, ...] = tuple(remote_refs) + else: + derived = _derive_probe_refs(root, explicit_remote) + if not derived["proven"]: + return None + probe = derived["refs"] + for ref in probe: res = subprocess.run( ["git", "-C", root, "rev-parse", "--verify", ref], capture_output=True, @@ -46,6 +100,84 @@ def resolve_remote_master_sha( return None +def resolve_remote_master_ref_state( + canonical_repo_root: str, + *, + remote_refs: tuple[str, ...] | None = None, + explicit_remote: str | None = None, +) -> dict: + """Resolve the tracking integration ref together with its commit SHA. + + Returns ``sha``, the ``ref`` it came from, the derived ``remote`` / + ``branch``, a machine-checkable ``reason_code``, and ``reasons``. ``sha`` is + None whenever the target cannot be resolved — never a fallback to some other + repository's commit. + + The SHA itself is obtained through :func:`resolve_remote_master_sha` rather + than by re-probing here, so one public function stays authoritative for + ref-to-SHA resolution. An explicit *remote_refs* keeps the historical + behaviour exactly: those refs are probed in order and no derivation happens. + """ + root = (canonical_repo_root or "").strip() + state: dict = { + "sha": None, + "ref": None, + "remote": None, + "branch": None, + "source": None, + "reason_code": None, + "reasons": [], + "cached_remote_head_branch": None, + "cached_remote_head_conflicts": False, + } + if not root: + state["reasons"].append("no canonical repository root supplied (fail closed)") + return state + + if remote_refs: + probe: tuple[str, ...] = tuple(remote_refs) + state["source"] = "explicit_remote_refs" + else: + derived = _derive_probe_refs(root, explicit_remote) + state["remote"] = derived["remote"] + state["branch"] = derived["branch"] + state["source"] = derived["source"] + state["cached_remote_head_branch"] = derived["cached_remote_head_branch"] + state["cached_remote_head_conflicts"] = derived["cached_remote_head_conflicts"] + if not derived["proven"]: + state["reason_code"] = derived["reason_code"] + state["reasons"] = derived["reasons"] + return state + probe = derived["refs"] + + sha = resolve_remote_master_sha(root, remote_refs=probe, explicit_remote=explicit_remote) + if not sha: + state["reasons"].append( + "tracking integration ref " + f"{' / '.join(probe) if probe else '(none derived)'} does not resolve in " + f"'{root}' (fail closed)" + ) + return state + + state["sha"] = sha + # Name the ref that actually carries this commit. When the SHA comes from a + # test double no probe will match, so fall back to the first derived ref, + # which is the one the guard is conceptually comparing against. + for ref in probe: + res = subprocess.run( + ["git", "-C", root, "rev-parse", "--verify", ref], + capture_output=True, + text=True, + check=False, + ) + if res.returncode == 0 and (res.stdout or "").strip() == sha: + state["ref"] = ref + break + else: + state["ref"] = probe[0] if probe else None + return state + + resolve_tracking_master_sha = resolve_remote_master_sha @@ -59,8 +191,9 @@ def assess_root_checkout_guard( remote_master_sha: str | None, resolved_role: str | None = None, actual_role: str | None = None, + remote_master_ref: str | None = None, ) -> dict: - """Fail closed when the control checkout is not clean master/prgs/master. + """Fail closed when the control checkout is not clean on its integration ref. ``resolved_role`` is the preflight-resolved *task* role and ``actual_role`` is the *active profile* role (#540). The reconciler exemption honours either @@ -102,9 +235,13 @@ def assess_root_checkout_guard( ) if remote_master_sha and head_sha and head_sha != remote_master_sha: + # #983: name the ref that was actually compared. Reporting a literal + # 'prgs/master' in a checkout gated against refs/remotes/MDCPS/dev sends + # the operator to inspect a ref that repository does not have. + ref_label = (remote_master_ref or "").strip() or "the tracking integration ref" reasons.append( - "control checkout HEAD does not match prgs/master " - f"(HEAD {head_sha[:12]}, prgs/master {remote_master_sha[:12]})" + f"control checkout HEAD does not match {ref_label} " + f"(HEAD {head_sha[:12]}, {ref_label} {remote_master_sha[:12]})" ) proven = not reasons diff --git a/tests/test_issue_983_cross_repo_base_ref.py b/tests/test_issue_983_cross_repo_base_ref.py new file mode 100644 index 0000000..27c3665 --- /dev/null +++ b/tests/test_issue_983_cross_repo_base_ref.py @@ -0,0 +1,692 @@ +"""Regression tests for Issue #983: derived target base ref for cross-repository checkouts. + +The mutation guard previously assumed ``prgs/master`` and the parity report +assumed ``origin/master``. Any repository using neither — for example remote +``MDCPS`` on integration branch ``dev`` — could not prove base equivalence, so +every gated mutation failed closed with no reachable remedy. + +Two further defects were found by review at head ``2d5d5c9d`` and are covered +here: + +* **B1** — the first fix derived the integration branch from + ``refs/remotes//HEAD``. That symref is a *local cache* written at clone + time and never refreshed by fetch, so on the real Weekly Briefings target it + still named ``main`` while the checkout tracked and sat exactly on ``dev``. The + authoritative signal is the checkout's own configured upstream. The fixtures + below therefore reproduce the **disagreement**: the cache says ``main``, the + configured upstream says ``dev``, and ``dev`` must win. +* **B2** — the parity report passed its internally inferred identity remote back + into the resolver, which reads a caller-supplied remote as "the caller already + disambiguated" and skips its ambiguity gate. Gating and reporting could then + evaluate different repositories. An inferred remote is now never laundered into + explicit caller intent, and ambiguity fails closed on both sides. + +These tests build hermetic git repositories on disk (no network, no fetch) and +assert the derived target end to end: identity remote, integration branch, +tracking ref, fail-closed refusals, and agreement between the mutation guard and +the parity report. +""" + +from __future__ import annotations + +import inspect +import os +import subprocess +import tempfile +import unittest + +import anti_stomp_preflight +import canonical_repository_root as crr +import master_parity_gate +import root_checkout_guard + +MDCPS_URL = "https://gitea.example.net/MDCPS/WeeklyBriefings-Meta.git" +MDCPS_SLUG = "MDCPS/WeeklyBriefings-Meta" +PRGS_URL = "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git" +PRGS_SLUG = "Scaled-Tech-Consulting/Gitea-Tools" + + +def _git(root: str, *args: str) -> str: + res = subprocess.run( + ["git", "-C", root, *args], + capture_output=True, + text=True, + check=True, + ) + return (res.stdout or "").strip() + + +def _make_repo(root: str, *, remote: str | None, url: str | None) -> str: + """Initialise a repository with one commit and an optional named remote.""" + os.makedirs(root, exist_ok=True) + _git(root, "init", "--quiet") + _git(root, "config", "user.email", "test@example.invalid") + _git(root, "config", "user.name", "Issue983 Test") + _git(root, "config", "commit.gpgsign", "false") + with open(os.path.join(root, "seed.txt"), "w", encoding="utf-8") as fh: + fh.write("seed\n") + _git(root, "add", "seed.txt") + _git(root, "commit", "--quiet", "-m", "seed") + if remote and url: + _git(root, "remote", "add", remote, url) + return _git(root, "rev-parse", "HEAD") + + +def _set_remote_branch(root: str, remote: str, branch: str, sha: str) -> None: + """Create refs/remotes// without contacting a network.""" + _git(root, "update-ref", f"refs/remotes/{remote}/{branch}", sha) + + +def _set_remote_head(root: str, remote: str, branch: str) -> None: + """Write the *cached* refs/remotes//HEAD symref. + + This is the signal B1 proved untrustworthy. Fixtures use it to reproduce a + stale cache, never to manufacture the answer under test. + """ + _git( + root, + "symbolic-ref", + f"refs/remotes/{remote}/HEAD", + f"refs/remotes/{remote}/{branch}", + ) + + +def _set_upstream(root: str, remote: str, branch: str) -> None: + """Configure the current branch's upstream, exactly as git tracking does. + + Writes ``branch..remote`` / ``branch..merge`` directly + rather than via ``--set-upstream-to`` so no ref is required to pre-exist and + no network is touched. + """ + current = _git(root, "symbolic-ref", "--short", "HEAD") + _git(root, "config", f"branch.{current}.remote", remote) + _git(root, "config", f"branch.{current}.merge", f"refs/heads/{branch}") + + +def _checkout_new_branch(root: str, branch: str) -> None: + _git(root, "checkout", "--quiet", "-b", branch) + + +def _advance(root: str, message: str) -> str: + with open(os.path.join(root, "seed.txt"), "a", encoding="utf-8") as fh: + fh.write(message + "\n") + _git(root, "add", "seed.txt") + _git(root, "commit", "--quiet", "-m", message) + return _git(root, "rev-parse", "HEAD") + + +class _RepoCase(unittest.TestCase): + def setUp(self) -> None: + self._tmp = tempfile.TemporaryDirectory() + self.addCleanup(self._tmp.cleanup) + self.root = os.path.join(self._tmp.name, "repo") + + def _weekly_briefings_shape(self) -> str: + """The real Weekly Briefings target, including its stale cache. + + Remote ``MDCPS``; checked out on ``dev``; upstream configured to + ``MDCPS/dev``; both ``dev`` and ``main`` present as tracking refs; and + ``refs/remotes/MDCPS/HEAD`` still cached at ``main`` from clone time. + """ + head = _make_repo(self.root, remote="MDCPS", url=MDCPS_URL) + _checkout_new_branch(self.root, "dev") + _set_remote_branch(self.root, "MDCPS", "dev", head) + stale = _advance(self.root, "main diverged long ago") + _set_remote_branch(self.root, "MDCPS", "main", stale) + _git(self.root, "reset", "--hard", "--quiet", head) + _set_remote_head(self.root, "MDCPS", "main") # stale clone-time cache + _set_upstream(self.root, "MDCPS", "dev") # authoritative + return head + + +class TestPrgsBehaviourPreserved(_RepoCase): + """Required coverage 1: PRGS prgs/master compatibility.""" + + def _prgs(self) -> str: + head = _make_repo(self.root, remote="prgs", url=PRGS_URL) + _set_remote_branch(self.root, "prgs", "master", head) + _set_remote_head(self.root, "prgs", "master") + _set_upstream(self.root, "prgs", "master") + return head + + def test_prgs_master_resolves_unchanged(self): + head = self._prgs() + + got = crr.resolve_target_base_ref(self.root) + self.assertTrue(got["proven"], got["reasons"]) + self.assertEqual(got["remote"], "prgs") + self.assertEqual(got["branch"], "master") + self.assertEqual(got["tracking_ref"], "refs/remotes/prgs/master") + self.assertEqual(got["repository_slug"], PRGS_SLUG) + self.assertEqual(got["source"], crr.BASE_REF_SOURCE_CONFIGURED_UPSTREAM) + # Cache and upstream agree here, which is the ordinary PRGS state. + self.assertFalse(got["cached_remote_head_conflicts"]) + self.assertEqual(root_checkout_guard.resolve_remote_master_sha(self.root), head) + + def test_prgs_resolves_without_a_configured_upstream(self): + """A PRGS checkout with no tracking config still resolves master.""" + head = _make_repo(self.root, remote="prgs", url=PRGS_URL) + _set_remote_branch(self.root, "prgs", "master", head) + + got = crr.resolve_target_base_ref(self.root) + self.assertTrue(got["proven"], got["reasons"]) + self.assertEqual(got["tracking_ref"], "refs/remotes/prgs/master") + self.assertEqual(got["source"], crr.BASE_REF_SOURCE_UNIQUE_CANDIDATE) + + def test_legacy_explicit_remote_refs_path_is_untouched(self): + """An explicit remote_refs override still short-circuits derivation.""" + head = _make_repo(self.root, remote="prgs", url=PRGS_URL) + _set_remote_branch(self.root, "prgs", "master", head) + + state = root_checkout_guard.resolve_remote_master_ref_state( + self.root, remote_refs=root_checkout_guard.REMOTE_MASTER_REFS + ) + self.assertEqual(state["sha"], head) + self.assertEqual(state["source"], "explicit_remote_refs") + + +class TestStaleCachedRemoteHead(_RepoCase): + """Required coverage 3: configured upstream MDCPS/dev vs stale cache -> main. + + This is B1. The fixture deliberately does **not** point + ``refs/remotes/MDCPS/HEAD`` at ``dev``; it reproduces the disagreement that + was live on ``/Users/jasonwalker/Development/weekly-briefings``. + """ + + def test_configured_upstream_beats_stale_cached_remote_head(self): + head = self._weekly_briefings_shape() + + # Precondition: the fixture really is in the defective state. + self.assertEqual( + _git(self.root, "symbolic-ref", "refs/remotes/MDCPS/HEAD"), + "refs/remotes/MDCPS/main", + ) + self.assertEqual(_git(self.root, "config", "--get", "branch.dev.merge"), "refs/heads/dev") + self.assertNotEqual( + _git(self.root, "rev-parse", "refs/remotes/MDCPS/main"), + _git(self.root, "rev-parse", "refs/remotes/MDCPS/dev"), + ) + + got = crr.resolve_target_base_ref(self.root) + self.assertTrue(got["proven"], got["reasons"]) + self.assertEqual(got["branch"], "dev") + self.assertEqual(got["tracking_ref"], "refs/remotes/MDCPS/dev") + self.assertEqual(got["source"], crr.BASE_REF_SOURCE_CONFIGURED_UPSTREAM) + # The stale cache is reported, never obeyed. + self.assertEqual(got["cached_remote_head_branch"], "main") + self.assertTrue(got["cached_remote_head_conflicts"]) + self.assertEqual(root_checkout_guard.resolve_remote_master_sha(self.root), head) + + def test_checkout_on_its_integration_tip_is_not_blocked(self): + """The live symptom: a checkout exactly on its tip was reported stale.""" + head = self._weekly_briefings_shape() + + state = root_checkout_guard.resolve_remote_master_ref_state(self.root) + self.assertEqual(state["sha"], head) + self.assertTrue(state["cached_remote_head_conflicts"]) + + assessment = root_checkout_guard.assess_root_checkout_guard( + workspace_path=self.root, + canonical_repo_root=self.root, + current_branch="dev", + head_sha=head, + porcelain_status="", + remote_master_sha=state["sha"], + remote_master_ref=state["ref"], + ) + self.assertTrue(assessment["proven"], assessment["reasons"]) + + report = master_parity_gate.assess_target_repository_parity( + canonical_root=self.root, source="test" + ) + self.assertFalse(report["stale"]) + self.assertEqual(report["base_branch"], "dev") + self.assertEqual(report["reasons"], []) + + def test_cached_remote_head_alone_never_proves_a_target(self): + """With no configured upstream, the cache cannot supply the branch.""" + head = _make_repo(self.root, remote="MDCPS", url=MDCPS_URL) + # Only a non-candidate branch exists, and only the cache names it. + _set_remote_branch(self.root, "MDCPS", "trunk", head) + _set_remote_head(self.root, "MDCPS", "trunk") + + got = crr.resolve_target_base_ref(self.root) + self.assertFalse(got["proven"]) + self.assertEqual(got["reason_code"], crr.DENY_NO_BASE_BRANCH) + self.assertEqual(got["tracking_refs"], ()) + self.assertEqual(got["cached_remote_head_branch"], "trunk") + # The refusal names the misleading signal so an operator is not sent + # chasing a ref that looks authoritative. + self.assertIn("trunk", " ".join(got["reasons"])) + + def test_cached_remote_head_never_breaks_a_tie(self): + """Two candidates, no upstream: the cache must not decide.""" + head = _make_repo(self.root, remote="MDCPS", url=MDCPS_URL) + _set_remote_branch(self.root, "MDCPS", "dev", head) + _set_remote_branch(self.root, "MDCPS", "main", head) + _set_remote_head(self.root, "MDCPS", "dev") + + got = crr.resolve_target_base_ref(self.root) + self.assertFalse(got["proven"]) + self.assertEqual(got["reason_code"], crr.DENY_AMBIGUOUS_BASE_BRANCH) + self.assertIsNone(root_checkout_guard.resolve_remote_master_sha(self.root)) + + def test_no_proven_source_is_the_remote_head_cache(self): + """Structural guard: the cache is not in the set of proving sources.""" + sources = { + crr.BASE_REF_SOURCE_CONFIGURED_UPSTREAM, + crr.BASE_REF_SOURCE_UNIQUE_CANDIDATE, + } + self.assertNotIn("remote_head_symref", sources) + self.assertFalse(hasattr(crr, "BASE_REF_SOURCE_REMOTE_HEAD")) + + +class TestCrossRepositoryTarget(_RepoCase): + """Required coverage 2/4/5: MDCPS/dev, no origin remote, exact case.""" + + def test_mdcps_dev_resolves(self): + head = self._weekly_briefings_shape() + got = crr.resolve_target_base_ref(self.root) + self.assertTrue(got["proven"], got["reasons"]) + self.assertEqual(got["remote"], "MDCPS") + self.assertEqual(got["branch"], "dev") + self.assertEqual(got["tracking_ref"], "refs/remotes/MDCPS/dev") + self.assertEqual(got["repository_slug"], MDCPS_SLUG) + self.assertEqual(root_checkout_guard.resolve_remote_master_sha(self.root), head) + + def test_no_remote_named_origin(self): + self._weekly_briefings_shape() + self.assertEqual(_git(self.root, "remote"), "MDCPS") + got = crr.resolve_target_base_ref(self.root) + self.assertTrue(got["proven"], got["reasons"]) + self.assertNotIn("origin", got["tracking_ref"]) + + def test_remote_name_case_is_preserved_exactly(self): + self._weekly_briefings_shape() + got = crr.resolve_target_base_ref(self.root) + self.assertEqual(got["remote"], "MDCPS") + self.assertNotEqual(got["remote"], "mdcps") + # The tracking ref must address the real ref, which is case-sensitive. + self.assertEqual(got["tracking_ref"], "refs/remotes/MDCPS/dev") + self.assertTrue( + _git(self.root, "rev-parse", "--verify", got["tracking_ref"]), + "case-preserved tracking ref must resolve", + ) + + def test_lowercase_candidate_never_supplies_the_remote_name(self): + """Guards against silently case-folding MDCPS to the candidate 'mdcps'. + + ``_IDENTITY_REMOTE_CANDIDATES`` contains a lowercase ``mdcps`` entry and + is probed *before* the repository's own remote listing. Git remote names + live in case-sensitive config subsections on every platform, so the + lowercase probe cannot resolve and the exact-case name must arrive from + ``git remote``. Asserted through config rather than ref lookup because a + case-insensitive filesystem (macOS) resolves loose refs either way, which + would make a ref-based assertion test the filesystem instead of the code. + """ + self._weekly_briefings_shape() + res = subprocess.run( + ["git", "-C", self.root, "remote", "get-url", "mdcps"], + capture_output=True, + text=True, + check=False, + ) + self.assertNotEqual(res.returncode, 0, "git remote names are case-sensitive") + + # A caller naming the wrong case cannot disambiguate, but the target is + # unambiguous anyway, so the exact-case name is still resolved. + got = crr.resolve_target_base_ref(self.root, explicit_remote="mdcps") + self.assertTrue(got["proven"], got["reasons"]) + self.assertEqual(got["remote"], "MDCPS") + self.assertEqual(got["tracking_ref"], "refs/remotes/MDCPS/dev") + self.assertFalse(got["identity_explicit"], "a non-matching name is not explicit intent") + + name, slug = crr.resolve_identity_remote(self.root) + self.assertEqual(name, "MDCPS") + self.assertEqual(slug, MDCPS_SLUG) + + +class TestTargetStaleness(_RepoCase): + """Required coverage 6/7: matching tip, and behind or divergent checkout.""" + + def test_local_equal_to_resolved_tip_is_not_stale(self): + self._weekly_briefings_shape() + got = master_parity_gate.assess_target_repository_parity( + canonical_root=self.root, source="test" + ) + self.assertTrue(got["determinable"]) + self.assertFalse(got["stale"]) + self.assertEqual(got["tracking_ref"], "refs/remotes/MDCPS/dev") + self.assertEqual(got["base_remote"], "MDCPS") + self.assertEqual(got["base_branch"], "dev") + self.assertIsNone(got["reason_code"]) + + def test_local_behind_resolved_tip_is_stale(self): + head = self._weekly_briefings_shape() + advanced = _advance(self.root, "remote moved on") + _set_remote_branch(self.root, "MDCPS", "dev", advanced) + _git(self.root, "reset", "--hard", "--quiet", head) + + got = master_parity_gate.assess_target_repository_parity( + canonical_root=self.root, source="test" + ) + self.assertTrue(got["determinable"]) + self.assertTrue(got["stale"]) + self.assertEqual(got["checkout_head"], head) + self.assertEqual(got["remote_tracking_head"], advanced) + + def test_local_divergent_from_resolved_tip_is_stale(self): + head = self._weekly_briefings_shape() + remote_side = _advance(self.root, "remote side") + _set_remote_branch(self.root, "MDCPS", "dev", remote_side) + _git(self.root, "reset", "--hard", "--quiet", head) + local_side = _advance(self.root, "local side") + + got = master_parity_gate.assess_target_repository_parity( + canonical_root=self.root, source="test" + ) + self.assertTrue(got["stale"]) + self.assertEqual(got["checkout_head"], local_side) + self.assertNotEqual(local_side, remote_side) + + +class TestFailClosed(_RepoCase): + """Required coverage 8/9: missing remote or ref, and ambiguous targets.""" + + def test_missing_remote_fails_closed(self): + _make_repo(self.root, remote=None, url=None) + got = crr.resolve_target_base_ref(self.root) + self.assertFalse(got["proven"]) + self.assertEqual(got["reason_code"], crr.DENY_NO_IDENTITY_REMOTE) + self.assertEqual(got["tracking_refs"], ()) + self.assertIsNone(root_checkout_guard.resolve_remote_master_sha(self.root)) + + def test_missing_tracking_ref_fails_closed(self): + _make_repo(self.root, remote="MDCPS", url=MDCPS_URL) + # Remote configured, but nothing has ever been fetched. + got = crr.resolve_target_base_ref(self.root) + self.assertFalse(got["proven"]) + self.assertEqual(got["reason_code"], crr.DENY_NO_BASE_BRANCH) + self.assertIsNone(root_checkout_guard.resolve_remote_master_sha(self.root)) + + def test_configured_upstream_without_a_tracking_ref_fails_closed(self): + """A proven target always names a ref that resolves.""" + _make_repo(self.root, remote="MDCPS", url=MDCPS_URL) + _set_upstream(self.root, "MDCPS", "dev") # declared, never fetched + + got = crr.resolve_target_base_ref(self.root) + self.assertFalse(got["proven"]) + self.assertEqual(got["reason_code"], crr.DENY_NO_BASE_BRANCH) + self.assertIsNone(got["tracking_ref"]) + + def test_every_proven_target_resolves(self): + """No 'proven' result may name an unresolvable tracking ref.""" + head = self._weekly_briefings_shape() + got = crr.resolve_target_base_ref(self.root) + self.assertTrue(got["proven"]) + self.assertEqual(_git(self.root, "rev-parse", got["tracking_ref"]), head) + + def test_ambiguous_integration_branch_fails_closed(self): + head = _make_repo(self.root, remote="MDCPS", url=MDCPS_URL) + # Two candidate integration branches and no configured upstream. + _set_remote_branch(self.root, "MDCPS", "dev", head) + _set_remote_branch(self.root, "MDCPS", "main", head) + + got = crr.resolve_target_base_ref(self.root) + self.assertFalse(got["proven"]) + self.assertEqual(got["reason_code"], crr.DENY_AMBIGUOUS_BASE_BRANCH) + self.assertIsNone(root_checkout_guard.resolve_remote_master_sha(self.root)) + + def test_ambiguous_identity_remote_fails_closed(self): + head = _make_repo(self.root, remote="MDCPS", url=MDCPS_URL) + _git(self.root, "remote", "add", "prgs", PRGS_URL) + _set_remote_branch(self.root, "MDCPS", "dev", head) + _set_remote_branch(self.root, "prgs", "master", head) + + got = crr.resolve_target_base_ref(self.root) + self.assertFalse(got["proven"]) + self.assertEqual(got["reason_code"], crr.DENY_AMBIGUOUS_REMOTE) + self.assertEqual(got["tracking_refs"], ()) + + def test_orphan_tracking_ref_from_removed_remote_is_ignored(self): + """The live Gitea-Tools symptom: refs/remotes/origin/* outlives its remote.""" + head = _make_repo(self.root, remote="prgs", url=PRGS_URL) + _set_remote_branch(self.root, "prgs", "master", head) + _set_upstream(self.root, "prgs", "master") + # An abandoned ref left behind by a remote that no longer exists. + _set_remote_branch(self.root, "origin", "master", head) + _advance(self.root, "orphan must not be consulted") + + got = master_parity_gate.assess_target_repository_parity( + canonical_root=self.root, source="test" + ) + self.assertEqual(got["tracking_ref"], "refs/remotes/prgs/master") + self.assertEqual(got["repository_slug"], PRGS_SLUG) + self.assertNotIn( + "target repository identity could not be derived from its git remote", + got["reasons"], + ) + + +class TestExplicitVersusInferredRemote(_RepoCase): + """Required coverage 11: explicit disambiguation stays distinct from inference. + + This is B2. A remote the module inferred while probing must never re-enter + the resolver as though an operator had named it. + """ + + def _two_remotes(self) -> str: + head = _make_repo(self.root, remote="MDCPS", url=MDCPS_URL) + _git(self.root, "remote", "add", "prgs", PRGS_URL) + _set_remote_branch(self.root, "MDCPS", "dev", head) + _set_remote_branch(self.root, "prgs", "master", head) + return head + + def test_explicit_remote_disambiguates(self): + self._two_remotes() + got = crr.resolve_target_base_ref(self.root, explicit_remote="MDCPS") + self.assertTrue(got["proven"], got["reasons"]) + self.assertEqual(got["remote"], "MDCPS") + self.assertEqual(got["branch"], "dev") + self.assertTrue(got["identity_explicit"]) + + def test_inferred_remote_does_not_disambiguate(self): + """Feeding the inferred remote back in must not unlock the target.""" + self._two_remotes() + inferred, _ = crr.resolve_identity_remote(self.root) + self.assertIsNotNone(inferred, "the first-wins probe still returns a name") + + # The report infers internally and must still refuse. + report = master_parity_gate.assess_target_repository_parity( + canonical_root=self.root, source="test" + ) + self.assertEqual(report["reason_code"], crr.DENY_AMBIGUOUS_REMOTE) + self.assertIsNone(report["repository_slug"]) + self.assertIsNone(report["tracking_ref"]) + self.assertFalse(report["stale"]) + self.assertTrue(report["reasons"]) + + def test_report_never_passes_a_remote_into_the_resolver(self): + """Structural guard against the exact B2 regression.""" + src = inspect.getsource(master_parity_gate.assess_target_repository_parity) + self.assertIn("resolve_target_base_ref(canonical_root)", src) + self.assertNotIn("resolve_target_base_ref(canonical_root, remote=", src) + self.assertNotIn("explicit_remote=identity", src) + # Reporting must use the ambiguity-aware identity resolver. + self.assertIn("assess_identity_remote(canonical_root)", src) + + def test_explicit_parameter_is_named_for_its_meaning(self): + for fn in ( + crr.resolve_target_base_ref, + root_checkout_guard.resolve_remote_master_sha, + root_checkout_guard.resolve_remote_master_ref_state, + ): + params = inspect.signature(fn).parameters + self.assertIn("explicit_remote", params, fn.__name__) + self.assertNotIn("remote", params, fn.__name__) + + def test_unmatched_explicit_remote_cannot_unlock_an_ambiguous_target(self): + self._two_remotes() + got = crr.resolve_target_base_ref(self.root, explicit_remote="nonexistent") + self.assertFalse(got["proven"]) + self.assertEqual(got["reason_code"], crr.DENY_AMBIGUOUS_REMOTE) + + +class TestGatingAndReportingAgree(_RepoCase): + """Required coverage 10: guard and parity report make identical decisions.""" + + def test_same_resolved_target_for_gate_and_report(self): + self._weekly_briefings_shape() + + gate = root_checkout_guard.resolve_remote_master_ref_state(self.root) + report = master_parity_gate.assess_target_repository_parity( + canonical_root=self.root, source="test" + ) + + self.assertEqual(gate["remote"], report["base_remote"]) + self.assertEqual(gate["branch"], report["base_branch"]) + self.assertEqual(gate["sha"], report["remote_tracking_head"]) + self.assertIn( + gate["ref"], + (report["tracking_ref"], f"{gate['remote']}/{gate['branch']}"), + ) + + def test_both_sides_refuse_the_same_unresolvable_target(self): + _make_repo(self.root, remote=None, url=None) + + gate = root_checkout_guard.resolve_remote_master_ref_state(self.root) + report = master_parity_gate.assess_target_repository_parity( + canonical_root=self.root, source="test" + ) + self.assertIsNone(gate["sha"]) + self.assertIsNone(report["remote_tracking_head"]) + self.assertFalse(report["stale"]) + self.assertTrue(report["reasons"]) + self.assertEqual(gate["reason_code"], report["reason_code"]) + + def test_both_sides_refuse_the_same_ambiguous_target(self): + head = _make_repo(self.root, remote="MDCPS", url=MDCPS_URL) + _git(self.root, "remote", "add", "prgs", PRGS_URL) + _set_remote_branch(self.root, "MDCPS", "dev", head) + _set_remote_branch(self.root, "prgs", "master", head) + + gate = root_checkout_guard.resolve_remote_master_ref_state(self.root) + report = master_parity_gate.assess_target_repository_parity( + canonical_root=self.root, source="test" + ) + + self.assertIsNone(gate["sha"]) + self.assertEqual(gate["reason_code"], crr.DENY_AMBIGUOUS_REMOTE) + # The report must not name a repository the gate refuses to act on. + self.assertEqual(report["reason_code"], gate["reason_code"]) + self.assertIsNone(report["repository_slug"]) + self.assertIsNone(report["base_remote"]) + self.assertIsNone(report["base_branch"]) + self.assertFalse(report["stale"]) + + def test_both_sides_refuse_the_same_ambiguous_branch(self): + head = _make_repo(self.root, remote="MDCPS", url=MDCPS_URL) + _set_remote_branch(self.root, "MDCPS", "dev", head) + _set_remote_branch(self.root, "MDCPS", "main", head) + + gate = root_checkout_guard.resolve_remote_master_ref_state(self.root) + report = master_parity_gate.assess_target_repository_parity( + canonical_root=self.root, source="test" + ) + self.assertEqual(gate["reason_code"], crr.DENY_AMBIGUOUS_BASE_BRANCH) + self.assertEqual(report["reason_code"], gate["reason_code"]) + self.assertIsNone(report["tracking_ref"]) + + +class TestProductionCallers(unittest.TestCase): + """Required coverage 13: every production caller consumes the same target.""" + + def test_guard_reports_the_ref_it_actually_compared(self): + assessment = root_checkout_guard.assess_root_checkout_guard( + workspace_path="/tmp/nonexistent-workspace-983", + canonical_repo_root="/tmp/nonexistent-root-983", + current_branch="dev", + head_sha="a" * 40, + porcelain_status="", + remote_master_sha="b" * 40, + remote_master_ref="refs/remotes/MDCPS/dev", + ) + self.assertTrue(assessment["block"]) + joined = " ".join(assessment["reasons"]) + self.assertIn("refs/remotes/MDCPS/dev", joined) + self.assertNotIn("prgs/master", joined) + + def test_guard_message_without_a_ref_stays_generic(self): + assessment = root_checkout_guard.assess_root_checkout_guard( + workspace_path="/tmp/nonexistent-workspace-983", + canonical_repo_root="/tmp/nonexistent-root-983", + current_branch="master", + head_sha="a" * 40, + porcelain_status="", + remote_master_sha="b" * 40, + ) + joined = " ".join(assessment["reasons"]) + self.assertIn("the tracking integration ref", joined) + self.assertNotIn("prgs/master", joined) + + def test_anti_stomp_preflight_forwards_the_resolved_ref(self): + sig = inspect.signature(anti_stomp_preflight.assess_anti_stomp_preflight) + self.assertIn("remote_master_ref", sig.parameters) + src = inspect.getsource(anti_stomp_preflight.assess_anti_stomp_preflight) + self.assertIn("remote_master_ref=remote_master_ref", src) + + def test_no_production_caller_inherits_the_prgs_default(self): + """Every resolve site must derive, or pass remote_refs explicitly.""" + import gitea_mcp_server + + src = inspect.getsource(gitea_mcp_server) + # The four historical call sites now consume the resolved-target state. + self.assertGreaterEqual(src.count("resolve_remote_master_ref_state("), 4) + self.assertNotIn("resolve_remote_master_sha(canonical_root)", src) + + def test_no_production_caller_supplies_an_explicit_remote(self): + """Nothing in production may suppress the ambiguity gate (#983 B2).""" + import gitea_mcp_server + + for module in (gitea_mcp_server, anti_stomp_preflight, master_parity_gate): + src = inspect.getsource(module) + self.assertNotIn("explicit_remote=", src, module.__name__) + + +class TestRepositoryStructureUntouched(_RepoCase): + """Required coverage 12: resolution mutates no ref, remote, config, or checkout.""" + + def test_resolution_creates_no_refs_or_branches(self): + self._weekly_briefings_shape() + + fmt = "--format=%(refname) %(objectname)" + before_refs = _git(self.root, "for-each-ref", fmt) + before_remotes = _git(self.root, "remote") + before_config = _git(self.root, "config", "--local", "--list") + before_head = _git(self.root, "rev-parse", "HEAD") + before_branch = _git(self.root, "symbolic-ref", "--short", "HEAD") + before_status = _git(self.root, "status", "--porcelain", "--untracked-files=all") + before_symref = _git(self.root, "symbolic-ref", "refs/remotes/MDCPS/HEAD") + + crr.resolve_target_base_ref(self.root) + crr.assess_identity_remote(self.root) + root_checkout_guard.resolve_remote_master_sha(self.root) + root_checkout_guard.resolve_remote_master_ref_state(self.root) + master_parity_gate.assess_target_repository_parity( + canonical_root=self.root, source="test" + ) + + self.assertEqual(_git(self.root, "for-each-ref", fmt), before_refs) + self.assertEqual(_git(self.root, "remote"), before_remotes) + self.assertEqual(_git(self.root, "config", "--local", "--list"), before_config) + self.assertEqual(_git(self.root, "rev-parse", "HEAD"), before_head) + self.assertEqual(_git(self.root, "symbolic-ref", "--short", "HEAD"), before_branch) + self.assertEqual( + _git(self.root, "status", "--porcelain", "--untracked-files=all"), before_status + ) + # The stale cache is specifically NOT repaired: that would be a mutation. + self.assertEqual(_git(self.root, "symbolic-ref", "refs/remotes/MDCPS/HEAD"), before_symref) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_root_checkout_guard.py b/tests/test_root_checkout_guard.py index 8fdf9c0..5c3d2b3 100644 --- a/tests/test_root_checkout_guard.py +++ b/tests/test_root_checkout_guard.py @@ -87,13 +87,27 @@ class TestAssessRootCheckoutGuard(unittest.TestCase): self.assertTrue(result["block"]) self.assertIn("tracked local edits", result["reasons"][0]) - def test_head_behind_prgs_master_blocked(self): + def test_head_behind_tracking_base_ref_blocked(self): + """#983: the base ref is derived, so the message no longer hardcodes PRGS.""" result = self._assess( head_sha=OTHER_SHA, remote_master_sha=MASTER_SHA, ) self.assertTrue(result["block"]) - self.assertIn("does not match prgs/master", result["reasons"][0]) + self.assertIn( + "does not match the tracking integration ref", result["reasons"][0] + ) + + def test_head_behind_named_base_ref_reports_that_ref(self): + """The resolved ref is named, so a non-PRGS target is reported accurately.""" + result = self._assess( + head_sha=OTHER_SHA, + remote_master_sha=MASTER_SHA, + remote_master_ref="refs/remotes/MDCPS/dev", + ) + self.assertTrue(result["block"]) + self.assertIn("does not match refs/remotes/MDCPS/dev", result["reasons"][0]) + self.assertNotIn("prgs/master", result["reasons"][0]) def test_merger_requires_clean_control_checkout(self): result = self._assess(