diff --git a/author_issue_bootstrap.py b/author_issue_bootstrap.py index 66d2a48..10834f1 100644 --- a/author_issue_bootstrap.py +++ b/author_issue_bootstrap.py @@ -23,6 +23,7 @@ import shutil import subprocess from typing import Any, Mapping +import author_lock_contract import author_mutation_worktree import control_plane_db import issue_lock_store @@ -278,6 +279,36 @@ def _verify_assignment_and_lease_ids( return None +def _branch_exists(canonical_repo_root: str, branch_name: str) -> bool: + """Whether *branch_name* still resolves in the canonical checkout (#953 F2). + + Used after compensating recovery to observe what survived rather than infer + it from the journal. Fails closed to ``True``: an unobservable branch is + reported as present, so the recommendation stays conservative rather than + telling an author to re-bootstrap over something that may still be there. + """ + if not branch_name: + return False + try: + res = subprocess.run( + [ + "git", + "-C", + canonical_repo_root, + "rev-parse", + "--verify", + "--quiet", + f"refs/heads/{branch_name}", + ], + capture_output=True, + text=True, + check=False, + ) + except Exception: + return True + return res.returncode == 0 + + def run_compensating_recovery( journal: dict[str, Any], canonical_repo_root: str, @@ -314,10 +345,21 @@ def run_compensating_recovery( issue_number=issue_num, session=session_id, lock_dir=journal_dir, + remote=journal.get("remote"), + # The same defaults the lock was written under, so the + # rollback targets the exact file bind_session_lock keyed. + org=journal.get("org") or "Scaled-Tech-Consulting", + repo=journal.get("repo") or "Gitea-Tools", ) rolled_back.append(f"lock:issue-{issue_num}") - except Exception: - pass + except Exception as exc: + # #953 F2: a swallowed failure here is what made the rollback + # report success while leaving an unrecoverable lock behind. + # Record it so the post-compensation classification can see the + # lock survived and recommend accordingly. + rolled_back.append( + f"lock_release_failed:issue-{issue_num}:{type(exc).__name__}" + ) artifacts["lock_created"] = False worktree_created = ( @@ -1198,26 +1240,31 @@ def bootstrap_author_issue_worktree( save_phase_journal(journal, journal_dir=lock_dir) # Phase 6: STATE_ESTABLISHED — Issue Lock Acquisition + # + # #953: this used to hand-build a thinner record — claimant at the top + # level, no work_lease, no lock_provenance, no expiry — which every + # downstream reader then refused. It now builds through the one shared + # canonical contract, so the lock bootstrap writes is the same lock + # gitea_lock_issue writes. from datetime import datetime, timezone try: - lock_data = { - "remote": remote, - "org": org or "Scaled-Tech-Consulting", - "repo": repo or "Gitea-Tools", - "issue_number": issue_number, - "branch": target_branch, - "branch_name": target_branch, - "worktree_path": target_worktree, - "owner_session": session, - "claimant": { - "username": identity, - "profile": profile, - }, - "assignment_id": assignment_id, - "lease_id": lease_id, - "expected_base_sha": live_master_sha, - "created_at": datetime.now(timezone.utc).isoformat(), - } + lock_data = author_lock_contract.build_canonical_issue_lock( + issue_number=issue_number, + branch_name=target_branch, + worktree_path=target_worktree, + remote=remote, + org=org or "Scaled-Tech-Consulting", + repo=repo or "Gitea-Tools", + identity=identity, + profile=profile, + tool="gitea_bootstrap_author_issue_worktree", + source=author_lock_contract.SOURCE_BOOTSTRAP, + owner_session=session, + assignment_id=assignment_id, + lease_id=lease_id, + expected_base_sha=live_master_sha, + ) + lock_data["created_at"] = datetime.now(timezone.utc).isoformat() journal.setdefault("pending_creations", {})["lock"] = True journal["artifacts_created"]["lock_created"] = True save_phase_journal(journal, journal_dir=lock_dir) @@ -1235,9 +1282,65 @@ def bootstrap_author_issue_worktree( "exact_next_action": "Verify lease/assignment state and retry.", } + # ── #953 AC7: verify the lock that was actually written ── + # Reporting "lock_created: true" and then directing the author to + # implement is what produced the unrecoverable state: by the time any + # reader refused the lock, the branch already carried commits and every + # sanctioned recovery path had become ineligible. The lock is therefore + # read back from disk and structurally verified *before* this function + # can report success, and a partial lock fails closed here — while the + # branch is still base-equivalent and recovery is still cheap. + written_lock = issue_lock_store.read_lock_file(lock_res) + contract = author_lock_contract.assess_lock_contract(written_lock) + if not contract["canonical"]: + journal["failure_reason"] = author_lock_contract.format_contract_refusal( + contract + ) + compensation = run_compensating_recovery( + journal, root, journal_dir=lock_dir + ) + # AC5/AC15: the recommendation must describe the state compensation + # actually left, not the state that provoked it. + # ``run_compensating_recovery`` has by now released the lock, removed + # the worktree, and deleted the branch, so recommending + # incomplete-lock recovery for those exact artifacts would refuse + # twice over. Observe what survived and answer for that. + post_state = author_lock_contract.assess_post_compensation_state( + compensation, + lock_present=bool(lock_res) and os.path.exists(lock_res), + worktree_present=os.path.isdir(target_worktree), + branch_present=_branch_exists(root, target_branch), + ) + return { + "success": False, + "reason_code": "incomplete_issue_lock_contract", + "message": author_lock_contract.format_contract_refusal(contract), + "issue_number": issue_number, + "branch_name": target_branch, + "worktree_path": target_worktree, + "lock_state": lock_res, + "lock_contract": contract, + "missing_fields": contract["missing_fields"], + "implementation_allowed": False, + "compensating_recovery": compensation, + "post_compensation_state": post_state, + # AC15: never strand a branch or worktree without a structured + # recovery recommendation — and never name an artifact the + # rollback has already deleted. + "exact_next_action": author_lock_contract.post_compensation_action( + post_state, + issue_number=issue_number, + branch_name=target_branch, + worktree_path=target_worktree, + missing_fields=contract["missing_fields"], + ), + "phase_journal": journal, + } + journal["phases"][PHASE_6_STATE_ESTABLISHED] = { "status": "completed", "lock": lock_res, + "lock_contract": contract["contract"], } journal["phases"][PHASE_7_TRANSITION_COMPLETED] = { "status": "completed", @@ -1261,9 +1364,16 @@ def bootstrap_author_issue_worktree( "assignment_id": assignment_id, "idempotency_key": key, "lock_state": lock_res, + "lock_contract": contract, + # #953 AC6: the canonical ownership token for this claim. Never null + # on a successful bootstrap — it is the fencing token every + # subsequent heartbeat and renewal is checked against. + "task_session_id": contract["task_session_id"], + "implementation_allowed": True, "phase_journal": journal, - "exact_next_action": ( - "Call gitea_whoami, then gitea_resolve_task_capability(task='work_issue') " - "and proceed with author implementation in the bootstrapped worktree." - ), + # #953 AC5: executable under the state actually returned. The lock + # has been read back and verified canonical, so proceeding to + # implementation is genuinely the correct next step here — which is + # exactly what the old unconditional wording could not promise. + "exact_next_action": author_lock_contract.recommended_action(contract), } diff --git a/author_lock_contract.py b/author_lock_contract.py new file mode 100644 index 0000000..47db9e2 --- /dev/null +++ b/author_lock_contract.py @@ -0,0 +1,623 @@ +"""One canonical author issue-lock contract shared by every writer (#953). + +Before this module, ``gitea_lock_issue`` and +``gitea_bootstrap_author_issue_worktree`` each wrote their own lock record. +``gitea_lock_issue`` wrote the canonical shape — ``work_lease`` carrying the +claimant plus a sanctioned ``lock_provenance`` — while bootstrap wrote a thinner +record with the claimant at the lock top level, ``lease_id: null``, and no +``work_lease``, ``lock_provenance``, or expiry at all. + +Every downstream reader was written against the canonical shape, so a lock that +bootstrap reported as successfully created was simultaneously: + +* un-heartbeatable — the ownership check read the claimant only from + ``work_lease.claimant``; +* un-renewable — expiry is read only from ``work_lease.expires_at``, so a + missing lease read as "never expires", and #760 exact-owner renewal only ever + assesses an *expired* lease; +* un-re-lockable — the branch had by then advanced past its base; +* and rejected by the #447 create-PR provenance guard. + +Each of those gates is individually correct. The defect was that two writers +disagreed about what a lock *is*. This module is the single definition, and both +writers now build through it. + +Nothing here weakens a guard. ``build_sanctioned_lock_provenance`` remains the +only provenance source, provenance is never accepted from a caller, and the +#447 guard is untouched — this module simply makes bootstrap satisfy it. +""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +from typing import Any, Mapping + +import issue_lock_provenance +import issue_lock_store +import lease_policy + +# Bootstrap writes through the same sanctioned source as gitea_lock_issue: the +# lock it produces *is* a canonical lock, not a second dialect that readers must +# learn. Adding a distinct source would have required widening +# SANCTIONED_LOCK_SOURCES, which is exactly the #447 weakening this issue's +# safety requirements forbid. +SOURCE_BOOTSTRAP = issue_lock_provenance.SOURCE_LOCK_ISSUE + +# Recovery of an incomplete bootstrap lock (#953 AC8-AC11) deliberately writes +# through SOURCE_LOCK_ISSUE too, and records its distinctness in +# ``lock_provenance.written_by_tool`` plus the ``bootstrap_lock_recovery`` +# transition block instead. There is no distinct recovery *source* constant, for +# the same reason bootstrap has none: minting one would require widening +# SANCTIONED_LOCK_SOURCES, which the #447 safety requirements forbid. + +#: Top-level keys every canonical author issue lock must carry. +REQUIRED_LOCK_FIELDS: tuple[str, ...] = ( + "remote", + "org", + "repo", + "issue_number", + "branch_name", + "worktree_path", + "work_lease", + "lock_provenance", +) + +#: Keys every canonical ``work_lease`` must carry. +REQUIRED_WORK_LEASE_FIELDS: tuple[str, ...] = ( + "operation_type", + "issue_number", + "branch", + "worktree_path", + "claimant", + "created_at", + "expires_at", + "last_heartbeat_at", + "task_session_id", + "lifecycle_version", +) + +# ── Explicit expiration states (AC12) ── +# The bug this replaces: a lock with no recorded expiry produced +# ``is_lease_expired() -> False``, which reads as "not yet expired" and made the +# lock permanently non-expiring *and* permanently ineligible for the renewal +# path, which only ever assesses an expired lease. "Absent" and "in the future" +# are different facts and are now named differently. +EXPIRATION_RECORDED = "recorded" +EXPIRATION_MISSING = "missing" +EXPIRATION_UNPARSEABLE = "unparseable" + +#: Structural verdicts returned by :func:`assess_lock_contract`. +CONTRACT_CANONICAL = "canonical" +CONTRACT_INCOMPLETE = "incomplete" +CONTRACT_LEGACY = "legacy" +CONTRACT_ABSENT = "absent" + + +def _text(value: Any) -> str: + return str(value or "").strip() + + +def now_utc() -> datetime: + return datetime.now(timezone.utc) + + +def format_timestamp(value: datetime) -> str: + """Serialize in the durable ``...Z`` form already used on disk.""" + return ( + value.astimezone(timezone.utc) + .replace(microsecond=0) + .isoformat() + .replace("+00:00", "Z") + ) + + +def lock_claimant(lock: Mapping[str, Any] | None) -> dict[str, str]: + """Read the claimant from either canonical or legacy placement. + + ``work_lease.claimant`` is canonical and is preferred. A top-level + ``claimant`` is the legacy/bootstrap placement and is accepted as a + fallback (AC14) — three separate readers already disagreed about this + (``issue_lock_store``, ``issue_lock_renewal``, ``issue_lock_recovery``), + which is why it now lives in one place. + + Reading a legacy placement is *not* a widening: every caller still compares + the values it returns against server-resolved identity and profile. This + only decides where to look, never whether ownership is proven. + + Delegates to ``issue_lock_store.lock_claimant`` rather than reimplementing + the rule. A second copy here would be a fourth reader that could drift from + the other three, which is the exact failure #953 exists to end. It lives in + the store because ``author_lock_contract`` imports the store, so defining it + here would make that import circular. + """ + recorded = issue_lock_store.lock_claimant(dict(lock) if isinstance(lock, Mapping) else None) + return { + "username": _text(recorded.get("username")), + "profile": _text(recorded.get("profile")), + } + + +def claimant_placement(lock: Mapping[str, Any] | None) -> str: + """Where the claimant was found: ``work_lease``, ``top_level``, or ``absent``.""" + if not isinstance(lock, Mapping): + return "absent" + lease = lock.get("work_lease") + if isinstance(lease, Mapping) and isinstance(lease.get("claimant"), Mapping): + return "work_lease" + if isinstance(lock.get("claimant"), Mapping): + return "top_level" + return "absent" + + +def build_claimant(*, username: str | None, profile: str | None) -> dict[str, str]: + """Build the canonical claimant pair from server-resolved values.""" + return {"username": _text(username), "profile": _text(profile)} + + +def build_author_issue_work_lease( + *, + issue_number: int, + branch_name: str, + worktree_path: str, + claimant: Mapping[str, Any], + task_session_id: str | None = None, + created: datetime | None = None, +) -> dict[str, Any]: + """Build the canonical author ``work_lease``. + + The single definition behind both writers. The TTL comes from the central + policy rather than a literal, and the window slides from the last valid + heartbeat (#790), so an abandoned task releases its claim within one TTL. + """ + started = created or now_utc() + policy = lease_policy.policy_for(lease_policy.TASK_CLASS_AUTHOR_ISSUE_WORK) + expires = started + timedelta(minutes=policy.initial_ttl_minutes) + session_id = _text(task_session_id) or issue_lock_store.mint_task_session_id( + issue_lock_store.AUTHOR_ISSUE_WORK_LEASE + ) + return { + "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE, + "issue_number": int(issue_number), + "pr_number": None, + "branch": branch_name, + "worktree_path": worktree_path, + "claimant": dict(claimant), + "created_at": format_timestamp(started), + "expires_at": format_timestamp(expires), + "last_heartbeat_at": format_timestamp(started), + # #790 AC-N1: the ownership key for this task, distinct from the + # recorded PID, which is the shared daemon and identifies no task. + "task_session_id": session_id, + # #790 AC-N8: the explicit lifecycle marker. Its absence — never a + # timestamp comparison — is what makes a lock legacy. + "lifecycle_version": lease_policy.LIFECYCLE_HEARTBEAT_V1, + "heartbeat_count": 1, + } + + +def build_canonical_issue_lock( + *, + issue_number: int, + branch_name: str, + worktree_path: str, + remote: str, + org: str, + repo: str, + identity: str | None, + profile: str | None, + tool: str, + source: str = issue_lock_provenance.SOURCE_LOCK_ISSUE, + owner_session: str | None = None, + assignment_id: str | None = None, + lease_id: str | None = None, + expected_base_sha: str | None = None, + task_session_id: str | None = None, + created: datetime | None = None, +) -> dict[str, Any]: + """Build a complete canonical lock record. + + ``tool`` and ``source`` are server-supplied. There is deliberately no + parameter through which a caller could inject provenance: the #953 safety + requirements forbid caller-manufactured provenance, so provenance is always + minted here from ``build_sanctioned_lock_provenance``. + """ + claimant = build_claimant(username=identity, profile=profile) + work_lease = build_author_issue_work_lease( + issue_number=issue_number, + branch_name=branch_name, + worktree_path=worktree_path, + claimant=claimant, + task_session_id=task_session_id, + created=created, + ) + record: dict[str, Any] = { + "remote": remote, + "org": org, + "repo": repo, + "issue_number": int(issue_number), + "branch": branch_name, + "branch_name": branch_name, + "worktree_path": worktree_path, + "work_lease": work_lease, + "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance( + tool=tool, + source=source, + claimant=claimant, + ), + } + if owner_session is not None: + record["owner_session"] = owner_session + if assignment_id is not None: + record["assignment_id"] = assignment_id + # #953 AC6: a null lease id is recorded only when no workflow lease was + # allocated for this bootstrap. The task-session identifier in the + # work_lease is what downstream ownership checks fence on, and it is never + # null on a canonical lock. + if lease_id is not None: + record["lease_id"] = lease_id + if expected_base_sha is not None: + record["expected_base_sha"] = expected_base_sha + return record + + +def expiration_state(lock: Mapping[str, Any] | None) -> dict[str, Any]: + """Classify a lock's recorded expiry explicitly (AC12). + + Distinguishes "no expiry was ever recorded" from "an expiry was recorded + and is still in the future". Collapsing those two into a single ``False`` + from ``is_lease_expired`` is what let a malformed lock be treated as + permanently live and simultaneously never renewable. + """ + if not isinstance(lock, Mapping): + return {"state": EXPIRATION_MISSING, "expires_at": None, "expired": None} + lease = lock.get("work_lease") + raw = lease.get("expires_at") if isinstance(lease, Mapping) else None + text = _text(raw) + if not text: + return {"state": EXPIRATION_MISSING, "expires_at": None, "expired": None} + try: + parsed = datetime.fromisoformat(text.replace("Z", "+00:00")).astimezone( + timezone.utc + ) + except ValueError: + return {"state": EXPIRATION_UNPARSEABLE, "expires_at": text, "expired": None} + return { + "state": EXPIRATION_RECORDED, + "expires_at": text, + "expired": parsed <= now_utc(), + } + + +def missing_contract_fields(lock: Mapping[str, Any] | None) -> list[str]: + """Name every canonical field a lock does not carry (AC7).""" + if not isinstance(lock, Mapping): + return [""] + missing: list[str] = [] + for field in REQUIRED_LOCK_FIELDS: + value = lock.get(field) + if value is None or (isinstance(value, str) and not value.strip()): + missing.append(field) + lease = lock.get("work_lease") + if not isinstance(lease, Mapping): + if "work_lease" not in missing: + missing.append("work_lease") + else: + for field in REQUIRED_WORK_LEASE_FIELDS: + value = lease.get(field) + if value is None or (isinstance(value, str) and not value.strip()): + missing.append(f"work_lease.{field}") + provenance = lock.get("lock_provenance") + if isinstance(provenance, Mapping): + if ( + _text(provenance.get("source")) + not in issue_lock_provenance.SANCTIONED_LOCK_SOURCES + ): + missing.append("lock_provenance.source (not sanctioned)") + if not _text(provenance.get("written_by_tool")): + missing.append("lock_provenance.written_by_tool") + claimant = lock_claimant(lock) + if not claimant["username"]: + missing.append("claimant.username") + if not claimant["profile"]: + missing.append("claimant.profile") + return missing + + +def assess_lock_contract(lock: Mapping[str, Any] | None) -> dict[str, Any]: + """Structural, read-only verdict on a durable lock record (AC7, AC16). + + Pure inspection: it reads the record it is handed and mutates nothing — + no lock, lease, branch, worktree, issue, or PR. Callers use it both to + verify a lock they just wrote and to report on one they found. + """ + if not isinstance(lock, Mapping) or not lock: + return { + "contract": CONTRACT_ABSENT, + "canonical": False, + "missing_fields": [""], + "claimant": {"username": "", "profile": ""}, + "claimant_placement": "absent", + "expiration": { + "state": EXPIRATION_MISSING, + "expires_at": None, + "expired": None, + }, + "heartbeatable": False, + "create_pr_eligible": False, + "lock_generation": None, + "task_session_id": None, + "reasons": ["no durable lock record"], + } + + missing = missing_contract_fields(lock) + claimant = lock_claimant(lock) + placement = claimant_placement(lock) + expiration = expiration_state(lock) + provenance_check = issue_lock_provenance.assess_lock_file_for_create_pr(dict(lock)) + + # Canonical means: every required field present, the claimant in the + # canonical placement, an expiry actually recorded, and the untouched #447 + # guard satisfied. + canonical = ( + not missing + and placement == "work_lease" + and expiration["state"] == EXPIRATION_RECORDED + and bool(provenance_check.get("proven")) + ) + if canonical: + contract = CONTRACT_CANONICAL + elif placement == "top_level" and claimant["username"] and claimant["profile"]: + contract = CONTRACT_LEGACY + else: + contract = CONTRACT_INCOMPLETE + + reasons: list[str] = [] + if missing: + reasons.append("missing canonical fields: " + ", ".join(missing)) + if placement == "top_level": + reasons.append( + "claimant recorded at the lock top level rather than in work_lease " + "(legacy/bootstrap placement)" + ) + if expiration["state"] == EXPIRATION_MISSING: + reasons.append( + "no expiration recorded; the lock is neither expirable nor renewable " + "until it is upgraded" + ) + elif expiration["state"] == EXPIRATION_UNPARSEABLE: + reasons.append(f"unparseable expires_at '{expiration['expires_at']}'") + if provenance_check.get("block"): + reasons.extend(provenance_check.get("reasons") or []) + + # Heartbeat needs the claimant pair (from either placement, post-fix) plus a + # task-session identifier to fence on. + lease = lock.get("work_lease") + task_session_id = ( + _text(lease.get("task_session_id")) if isinstance(lease, Mapping) else "" + ) + heartbeatable = bool( + claimant["username"] and claimant["profile"] and task_session_id + ) + + return { + "contract": contract, + "canonical": canonical, + "missing_fields": missing, + "claimant": claimant, + "claimant_placement": placement, + "expiration": expiration, + "heartbeatable": heartbeatable, + "create_pr_eligible": bool(provenance_check.get("proven")), + "lock_generation": lock.get("lock_generation"), + "task_session_id": task_session_id or None, + "reasons": reasons, + } + + +def format_contract_refusal(assessment: Mapping[str, Any]) -> str: + """Human-readable refusal naming exactly what the lock is missing.""" + missing = ", ".join(assessment.get("missing_fields") or []) or "unknown fields" + return ( + "Issue lock contract incomplete (#953): " + f"{missing}. The lock cannot be heartbeated, renewed, or accepted by " + "gitea_create_pr in this state (fail closed)" + ) + + +def recommended_action(assessment: Mapping[str, Any]) -> str: + """The one executable next step for a lock in this state (AC5, AC15).""" + contract = assessment.get("contract") + if contract == CONTRACT_CANONICAL: + return ( + "Lock is canonical. Call gitea_whoami, then " + "gitea_resolve_task_capability(task='work_issue'), then proceed with " + "author implementation in the bootstrapped worktree." + ) + if contract == CONTRACT_ABSENT: + return ( + "No durable lock exists. Call gitea_lock_issue for this issue and " + "branch before writing any implementation bytes." + ) + return ( + "Do not begin implementation. Call " + "gitea_recover_incomplete_bootstrap_lock for this exact issue, branch, " + "and worktree to upgrade the lock to the canonical contract, or " + "gitea_lock_issue while the worktree is still base-equivalent." + ) + + +# ── Post-compensation recovery guidance (#953 AC5/AC15, review 632 F2) ── +# +# ``recommended_action`` above answers "what can be done about a lock in this +# shape". That is the wrong question on the bootstrap AC7 refusal path, because +# ``run_compensating_recovery`` has already run by the time the answer is +# reported: it releases the lock, removes the worktree when clean — which it +# always is there, no implementation bytes having been written — and deletes the +# created branch. Recommending incomplete-lock recovery for those artifacts +# hands the author two refusals in a row (``no_durable_lock``, then +# ``worktree_invalid``) for a state that a plain bootstrap retry would fix. The +# advice must describe the state that actually *remains*. + +#: Compensation removed every artifact this transition created. +CLEANUP_COMPLETE = "complete" +#: Compensation removed some artifacts; others survive and are still actionable. +CLEANUP_PARTIAL = "partial" +#: Compensation itself failed or could not be observed; nothing is provable. +CLEANUP_FAILED = "failed" + + +def assess_post_compensation_state( + recovery: Mapping[str, Any] | None, + *, + lock_present: bool, + worktree_present: bool, + branch_present: bool, +) -> dict[str, Any]: + """Classify what survived compensation, from observed durable state. + + Pure. The caller observes the filesystem and git; this decides. Observation + is authoritative over the journal's ``rolled_back`` list, which records what + compensation *attempted*: ``run_compensating_recovery`` swallows a failed + lock release and appends nothing, so an absent marker proves nothing either + way. The list is still carried through as corroborating evidence. + + The three states are distinct facts, not degrees of the same one: + + * ``CLEANUP_COMPLETE`` — compensation ran and nothing it created remains. + * ``CLEANUP_PARTIAL`` — compensation ran and artifacts survive, whether by + design (a worktree dirty at rollback time, a branch carrying commits) or + because a rollback step errored. Either way the surviving set was observed + directly, so it is known and actionable; ``failed_rollback_steps`` records + which cause applies. + * ``CLEANUP_FAILED`` — compensation never ran to completion, so nothing it + would have removed can be assumed removed. + """ + rolled_back = list((recovery or {}).get("rolled_back") or []) + executed = bool((recovery or {}).get("executed")) + failed_steps = [entry for entry in rolled_back if "_failed" in entry] + + surviving: list[str] = [] + if lock_present: + surviving.append("lock") + if worktree_present: + surviving.append("worktree") + if branch_present: + surviving.append("branch") + + if not executed: + state = CLEANUP_FAILED + elif surviving: + state = CLEANUP_PARTIAL + else: + state = CLEANUP_COMPLETE + + return { + "cleanup_state": state, + "compensation_executed": executed, + "lock_present": bool(lock_present), + "worktree_present": bool(worktree_present), + "branch_present": bool(branch_present), + "surviving_artifacts": surviving, + "removed_artifacts": [ + name + for name, present in ( + ("lock", lock_present), + ("worktree", worktree_present), + ("branch", branch_present), + ) + if not present + ], + "failed_rollback_steps": failed_steps, + "rolled_back": rolled_back, + } + + +def post_compensation_action( + state: Mapping[str, Any], + *, + issue_number: int, + branch_name: str, + worktree_path: str, + missing_fields: list[str] | None = None, +) -> str: + """The one executable next step for the state compensation actually left. + + Every branch names only artifacts the classification says still exist, so no + recommendation can point at something the rollback deleted. + """ + missing = ", ".join(missing_fields or []) or "the reported missing fields" + cleanup_state = state.get("cleanup_state") + lock_present = bool(state.get("lock_present")) + worktree_present = bool(state.get("worktree_present")) + branch_present = bool(state.get("branch_present")) + + if not state.get("compensation_executed"): + # Compensation never ran, so nothing was rolled back and nothing about + # the remaining state was decided. The read-only surface is the only + # action executable under any state. + return ( + "Compensating rollback did not complete, so the remaining state is " + f"not proven. Call gitea_inspect_issue_lock_contract for issue " + f"#{issue_number} (read-only) to establish what survives before any " + "further action. Do not retry bootstrap until it is known." + ) + + prefix = "" + failed_steps = state.get("failed_rollback_steps") or [] + if failed_steps: + prefix = ( + "Compensating rollback reported a failed step " + f"({', '.join(failed_steps)}); what survives was observed directly " + "and the action below is scoped to exactly that. " + ) + + if cleanup_state == CLEANUP_COMPLETE: + return ( + "Compensating rollback removed the malformed lock, the branch, and " + f"the worktree, so nothing from this attempt remains. Resolve " + f"{missing} and re-run gitea_bootstrap_author_issue_worktree for " + f"issue #{issue_number} from the clean pre-bootstrap state. Do not " + "call gitea_recover_incomplete_bootstrap_lock: there is no lock, " + "branch, or worktree left for it to act on." + ) + + if lock_present and worktree_present and branch_present: + return prefix + ( + "The lock, branch, and worktree all survive. Call " + "gitea_recover_incomplete_bootstrap_lock for issue " + f"#{issue_number}, branch '{branch_name}', and worktree " + f"'{worktree_path}', passing the worktree's current head as " + "expected_head, to upgrade the lock to the canonical contract." + ) + + if not lock_present and worktree_present and branch_present: + return prefix + ( + "The malformed lock was released but the branch and worktree " + "survive. No implementation bytes were written, so the worktree is " + f"still base-equivalent: call gitea_lock_issue for issue " + f"#{issue_number} on branch '{branch_name}' from worktree " + f"'{worktree_path}' to acquire a canonical lock." + ) + + if lock_present and not worktree_present: + return prefix + ( + f"The worktree for issue #{issue_number} is gone but the durable " + "lock survived, so neither gitea_recover_incomplete_bootstrap_lock " + "(it would refuse worktree_invalid) nor gitea_lock_issue (it has no " + "worktree to bind) is executable. Call " + "gitea_inspect_issue_lock_contract for issue " + f"#{issue_number} (read-only) to confirm the surviving lock; it " + "must be released by its recorded owner before bootstrap is " + "retried." + ) + + # Lock gone, worktree gone, some git artifact left (a branch with commits, + # or a branch this transition did not create). + return prefix + ( + "Compensating rollback removed the lock and worktree; branch " + f"'{branch_name}' survives and was not deleted. Call " + f"gitea_inspect_issue_lock_contract for issue #{issue_number} " + "(read-only) to confirm no durable lock remains, then re-run " + "gitea_bootstrap_author_issue_worktree, which will adopt the existing " + "branch rather than recreating it." + ) diff --git a/bootstrap_lock_recovery.py b/bootstrap_lock_recovery.py new file mode 100644 index 0000000..a451c91 --- /dev/null +++ b/bootstrap_lock_recovery.py @@ -0,0 +1,304 @@ +"""Target-specific recovery for incomplete bootstrap issue locks (#953). + +The situation this exists for: ``gitea_bootstrap_author_issue_worktree`` +reported success, wrote an incomplete lock, and told the author to implement. +The author did — legitimately, following the tool's own reported next action — +and the branch now carries real committed and pushed work. At that point every +pre-existing recovery path is simultaneously ineligible: + +* heartbeat refuses, because the claimant is not where it looks; +* ``gitea_lock_issue`` refuses, because the branch is no longer base-equivalent; +* #760 exact-owner renewal never engages, because a lock with no recorded + expiry is never *expired*; +* the #447 create-PR guard refuses, because there is no provenance. + +Distinct from every neighbouring path: #753 ``issue_lock_recovery`` requires a +dead owner PID, #760 ``issue_lock_renewal`` requires an *expired* lease, and +#442 ``issue_lock_adoption`` decides branch adoption. None of them addresses a +lock that is structurally incomplete and therefore never expires at all. + +**What this will not do.** It never moves, resets, or rewinds a branch, and +never requires base-equivalence — the committed work is the thing being +preserved. It never pushes and never opens a pull request. It touches only the +one lock file named by (remote, org, repo, issue). It accepts no caller-supplied +provenance and no caller-supplied authorization flag; both are minted +server-side. It refuses a healthy foreign-owned lock outright, and a matching +username alone is never accepted as proof of ownership — the profile must match +too, and the lock's recorded binding must agree with the observed branch, +worktree, and head. +""" + +from __future__ import annotations + +import os +from typing import Any, Mapping + +import author_lock_contract +import issue_lock_store + +#: Refusal codes, so callers can branch on cause rather than parse prose. +REFUSAL_NO_LOCK = "no_durable_lock" +REFUSAL_ALREADY_CANONICAL = "already_canonical" +REFUSAL_FOREIGN_CLAIMANT = "foreign_claimant" +REFUSAL_HEALTHY_FOREIGN = "healthy_foreign_lock" +REFUSAL_IDENTITY_UNRESOLVED = "identity_unresolved" +REFUSAL_BINDING_MISMATCH = "binding_mismatch" +REFUSAL_WORKTREE_INVALID = "worktree_invalid" +REFUSAL_HEAD_MISMATCH = "head_mismatch" + + +def _text(value: Any) -> str: + return str(value or "").strip() + + +def _same_realpath(left: str | None, right: str | None) -> bool: + lhs, rhs = _text(left), _text(right) + if not lhs or not rhs: + return False + try: + return os.path.realpath(lhs) == os.path.realpath(rhs) + except OSError: + return lhs == rhs + + +def assess_bootstrap_lock_recovery( + existing_lock: Mapping[str, Any] | None, + *, + issue_number: int, + branch_name: str, + worktree_path: str, + remote: str, + org: str, + repo: str, + identity: str | None, + profile: str | None, + observed_head: str | None, + declared_head: str | None, + worktree_exists: bool, + worktree_registered: bool, + current_branch: str | None, + now: Any = None, +) -> dict[str, Any]: + """Decide whether this exact lock may be upgraded by this exact caller. + + Pure: every input is an observation the caller already made, and nothing + here reads or writes the filesystem, git, or Gitea. That is what makes the + same decision testable in isolation and reusable by the read-only + inspection surface, which must not mutate anything (AC16). + + Returns a dict with ``recovery_sanctioned`` plus the full evidence set. A + refusal never raises — it reports, so the caller can surface exactly which + piece of evidence was missing. + """ + reasons: list[str] = [] + refusal_code: str | None = None + + contract = author_lock_contract.assess_lock_contract(existing_lock) + + if not existing_lock: + return { + "recovery_sanctioned": False, + "refusal_code": REFUSAL_NO_LOCK, + "reasons": [ + f"no durable issue lock exists for issue #{issue_number}; there is " + "nothing to recover (fail closed)" + ], + "contract": contract, + "evidence": {}, + "expected_generation": None, + } + + active_identity = _text(identity) + active_profile = _text(profile) + recorded = author_lock_contract.lock_claimant(existing_lock) + freshness = issue_lock_store.assess_lock_freshness(dict(existing_lock), now=now) + generation = issue_lock_store.lock_generation(existing_lock) + + evidence: dict[str, Any] = { + "recorded_claimant": recorded, + "active_identity": active_identity, + "active_profile": active_profile, + "recorded_branch": existing_lock.get("branch_name"), + "recorded_worktree": existing_lock.get("worktree_path"), + "recorded_owner_session": existing_lock.get("owner_session"), + "recorded_generation": generation, + "recorded_remote": existing_lock.get("remote"), + "recorded_org": existing_lock.get("org"), + "recorded_repo": existing_lock.get("repo"), + "observed_head": _text(observed_head), + "declared_head": _text(declared_head), + "current_branch": _text(current_branch), + "worktree_exists": bool(worktree_exists), + "worktree_registered": bool(worktree_registered), + "freshness": freshness, + "claimant_placement": contract.get("claimant_placement"), + "expiration_state": contract.get("expiration", {}).get("state"), + } + + # ── Repository and issue identity (AC10) ── + if _text(existing_lock.get("remote")) != _text(remote): + reasons.append( + f"recorded remote '{existing_lock.get('remote')}' does not match '{remote}'" + ) + refusal_code = refusal_code or REFUSAL_BINDING_MISMATCH + if _text(existing_lock.get("org")) != _text(org): + reasons.append( + f"recorded org '{existing_lock.get('org')}' does not match '{org}'" + ) + refusal_code = refusal_code or REFUSAL_BINDING_MISMATCH + if _text(existing_lock.get("repo")) != _text(repo): + reasons.append( + f"recorded repo '{existing_lock.get('repo')}' does not match '{repo}'" + ) + refusal_code = refusal_code or REFUSAL_BINDING_MISMATCH + if existing_lock.get("issue_number") != issue_number: + reasons.append( + f"lock targets issue #{existing_lock.get('issue_number')}, not " + f"#{issue_number}" + ) + refusal_code = refusal_code or REFUSAL_BINDING_MISMATCH + + # ── Branch and worktree binding (AC10) ── + if _text(existing_lock.get("branch_name")) != _text(branch_name): + reasons.append( + f"recorded branch '{existing_lock.get('branch_name')}' does not match " + f"'{branch_name}'" + ) + refusal_code = refusal_code or REFUSAL_BINDING_MISMATCH + if not _same_realpath(existing_lock.get("worktree_path"), worktree_path): + reasons.append( + f"recorded worktree '{existing_lock.get('worktree_path')}' does not " + f"match '{worktree_path}'" + ) + refusal_code = refusal_code or REFUSAL_BINDING_MISMATCH + + # ── The worktree is real, registered, and on the branch (AC10) ── + # Deliberately no base-equivalence requirement and no constraint on how far + # the branch has advanced: the whole point is that it already carries the + # author's legitimate commits (AC9). + if not worktree_exists: + reasons.append(f"declared worktree '{worktree_path}' does not exist") + refusal_code = refusal_code or REFUSAL_WORKTREE_INVALID + if not worktree_registered: + reasons.append(f"worktree '{worktree_path}' is not a registered git worktree") + refusal_code = refusal_code or REFUSAL_WORKTREE_INVALID + if _text(current_branch) != _text(branch_name): + reasons.append( + f"worktree is on branch '{_text(current_branch) or 'unknown'}', not " + f"'{branch_name}'" + ) + refusal_code = refusal_code or REFUSAL_WORKTREE_INVALID + + # ── Current head fencing (AC10) ── + # The caller names the commit it believes it is recovering. A mismatch means + # the worktree moved under the caller, so the decision is stale. + if not _text(observed_head): + reasons.append("could not observe the worktree head") + refusal_code = refusal_code or REFUSAL_HEAD_MISMATCH + elif _text(declared_head) and _text(declared_head) != _text(observed_head): + reasons.append( + f"declared head '{_text(declared_head)}' does not match observed head " + f"'{_text(observed_head)}'" + ) + refusal_code = refusal_code or REFUSAL_HEAD_MISMATCH + + # ── Ownership (AC10, AC11) ── + # A matching username alone is never sufficient: the profile must match too, + # and both are compared against server-resolved values the caller cannot set. + if not active_identity or not active_profile: + reasons.append( + "active identity and profile could not both be resolved; ownership " + "cannot be proven" + ) + refusal_code = refusal_code or REFUSAL_IDENTITY_UNRESOLVED + if not recorded["username"] or not recorded["profile"]: + reasons.append( + "durable lock does not record both a claimant username and profile" + ) + refusal_code = refusal_code or REFUSAL_FOREIGN_CLAIMANT + elif ( + recorded["username"] != active_identity + or recorded["profile"] != active_profile + ): + # AC11: a foreign-owned lock is never recoverable through this path, + # healthy or not. The healthy case is reported distinctly so the refusal + # is legible, but both refuse. + if freshness.get("live"): + reasons.append( + f"lock is owned by a healthy foreign claimant " + f"'{recorded['username']}/{recorded['profile']}'; takeover is not " + "a recovery path" + ) + refusal_code = REFUSAL_HEALTHY_FOREIGN + else: + reasons.append( + f"lock claimant '{recorded['username']}/{recorded['profile']}' " + f"does not match active '{active_identity}/{active_profile}'" + ) + refusal_code = refusal_code or REFUSAL_FOREIGN_CLAIMANT + + # ── Nothing to recover ── + # A lock that is already canonical is left strictly alone. Rewriting it would + # mint a new task-session identifier and invalidate the heartbeat token the + # legitimate owner is already using. + if contract.get("canonical") and not reasons: + return { + "recovery_sanctioned": False, + "refusal_code": REFUSAL_ALREADY_CANONICAL, + "reasons": [ + "lock already satisfies the canonical contract; no recovery is " + "required" + ], + "contract": contract, + "evidence": evidence, + "expected_generation": generation, + } + + sanctioned = not reasons + return { + "recovery_sanctioned": sanctioned, + "refusal_code": None if sanctioned else refusal_code, + "reasons": reasons, + "contract": contract, + "evidence": evidence, + "expected_generation": generation, + } + + +def build_recovery_record( + assessment: Mapping[str, Any], + *, + recovered_at: str, + new_task_session_id: str, +) -> dict[str, Any]: + """Auditable record of the ownership and generation transition (AC10). + + A recovered lock must never read as an original claim, so both sides of the + transition are preserved: what the incomplete lock recorded, and what + replaced it. + """ + evidence = dict(assessment.get("evidence") or {}) + contract = dict(assessment.get("contract") or {}) + return { + "recovery_kind": "incomplete_bootstrap_lock", + "recovered_at": recovered_at, + "prior_contract": contract.get("contract"), + "prior_missing_fields": list(contract.get("missing_fields") or []), + "prior_claimant_placement": evidence.get("claimant_placement"), + "prior_expiration_state": evidence.get("expiration_state"), + "prior_generation": evidence.get("recorded_generation"), + "prior_owner_session": evidence.get("recorded_owner_session"), + "prior_freshness": (evidence.get("freshness") or {}).get("status"), + "replacement_task_session_id": new_task_session_id, + "preserved_head": evidence.get("observed_head"), + "branch_reset": False, + "base_equivalence_required": False, + } + + +def format_recovery_refusal(assessment: Mapping[str, Any]) -> str: + reasons = "; ".join( + assessment.get("reasons") or ["unknown bootstrap lock recovery refusal"] + ) + code = assessment.get("refusal_code") or "refused" + return f"Bootstrap lock recovery refused ({code}): {reasons} (fail closed)" diff --git a/docs/author-issue-lock-contract.md b/docs/author-issue-lock-contract.md new file mode 100644 index 0000000..0fb51c6 --- /dev/null +++ b/docs/author-issue-lock-contract.md @@ -0,0 +1,209 @@ +# The canonical author issue-lock contract (#953) + +Every author issue lock has exactly one shape. Both writers — +`gitea_bootstrap_author_issue_worktree` and `gitea_lock_issue` — build it +through `author_lock_contract.build_canonical_issue_lock`, and every reader +consumes that same shape. + +Before #953 the two writers disagreed. `gitea_lock_issue` wrote the canonical +record; bootstrap wrote a thinner one with the claimant at the lock top level, +`lease_id: null`, and no `work_lease`, `lock_provenance`, or expiry. Because +every reader was written against the canonical shape, a lock that bootstrap +reported as successfully created could not be heartbeated, renewed, re-locked, +or accepted by `gitea_create_pr`. Each of those gates was individually correct; +the defect was that two writers disagreed about what a lock *is*. + +## Required ordering + +**Finalize the lock before writing any implementation bytes.** This ordering is +what keeps recovery cheap: while the worktree is still base-equivalent, a lock +problem can be fixed by simply calling `gitea_lock_issue` again. Once the branch +carries commits, base-equivalence is gone and the ordinary re-lock path is no +longer available. + +1. `gitea_whoami` — resolve identity and profile. +2. `gitea_resolve_task_capability(task='work_issue')`. +3. `gitea_bootstrap_author_issue_worktree` — creates the branch, the registered + worktree under `branches/`, and a **canonical** lock. It reads the lock back + and verifies it structurally before reporting success; a partial lock fails + closed here, with the missing fields named, and never reports + `implementation_allowed: true`. +4. `gitea_heartbeat_issue_lock` — prove the lock is usable, using the + `task_session_id` bootstrap returned. +5. Implement, commit, push. +6. `gitea_create_pr`. + +If bootstrap returns `success: false` with +`reason_code: incomplete_issue_lock_contract`, **do not implement**. Its +`exact_next_action` names the executable recovery step. Bootstrap's reported +next action always matches the state it actually returned. + +### What that refusal leaves behind + +The AC7 refusal runs `run_compensating_recovery` *before* it reports, so the +advice has to describe the post-rollback state rather than the shape of the lock +that provoked it. Recommending incomplete-lock recovery for artifacts the +rollback already deleted would produce `no_durable_lock` and then +`worktree_invalid` — two refusals for a state a plain retry fixes. + +The refusal therefore carries `compensating_recovery` and +`post_compensation_state`, and derives `exact_next_action` from what was +observed on disk. `cleanup_state` is one of: + +| `cleanup_state` | Meaning | Next action | +| --- | --- | --- | +| `complete` | lock, branch, and worktree all removed | resolve `missing_fields` and re-run `gitea_bootstrap_author_issue_worktree` | +| `partial` | rollback ran; some artifacts survive, by design or because a step errored | scoped to exactly what survives — see below | +| `failed` | rollback never completed, so nothing is proven removed | `gitea_inspect_issue_lock_contract` (read-only) before anything else | + +Within `partial`, the surviving set decides the action: + +| Survives | Next action | +| --- | --- | +| lock + branch + worktree | `gitea_recover_incomplete_bootstrap_lock` for that exact issue, branch, and worktree | +| branch + worktree (lock released) | `gitea_lock_issue` — no implementation bytes were written, so the worktree is still base-equivalent | +| lock only (worktree removed) | `gitea_inspect_issue_lock_contract`; the surviving lock must be released by its recorded owner before bootstrap is retried | +| branch only | `gitea_inspect_issue_lock_contract`, then re-run bootstrap, which adopts the existing branch | + +`failed_rollback_steps` names any rollback step that errored, and the returned +action says so rather than presenting the surviving state as intentional. + +> The lock half of that rollback was dead code until #953 review 632 F2: +> `run_compensating_recovery` called `issue_lock_store.release_session_lock`, +> which did not exist, inside a bare `except Exception: pass`. Every rollback +> removed the branch and worktree and silently left the lock — the exact +> uninspectable, unrecoverable state this issue exists to eliminate. The +> function now exists, releases only a lock whose recorded `owner_session` +> matches, and its failures are recorded rather than swallowed. + +## The contract + +A canonical lock carries every field in +`author_lock_contract.REQUIRED_LOCK_FIELDS`: + +| Field | Meaning | +| --- | --- | +| `remote`, `org`, `repo`, `issue_number` | repository and issue identity | +| `branch_name`, `worktree_path` | the binding this claim owns | +| `work_lease` | the canonical lease block, below | +| `lock_provenance` | sanctioned source, minted server-side | +| `lock_generation` | monotonic; every write advances it | + +`work_lease` carries every field in +`author_lock_contract.REQUIRED_WORK_LEASE_FIELDS`, notably: + +| Field | Meaning | +| --- | --- | +| `claimant.{username,profile}` | **canonical** claimant placement | +| `expires_at` | sliding TTL from `lease_policy` | +| `last_heartbeat_at`, `heartbeat_count` | liveness evidence | +| `task_session_id` | the ownership fencing token — never null | +| `lifecycle_version` | `heartbeat-v1`; its absence is what makes a lock legacy | + +### Claimant placement and legacy compatibility + +`work_lease.claimant` is canonical. A top-level `claimant` is the legacy +placement written by pre-#953 bootstrap and is still **read** — through the one +shared reader, `issue_lock_store.lock_claimant` — so an existing lock is not +refused for "not recording a claimant" when it plainly records one. + +Tolerating the placement is not a widening. Every caller still compares the +values against server-resolved identity and profile, so a legacy placement +grants nothing the canonical placement would not. When both are present, the +`work_lease` copy wins: after an upgrade, a stale top-level copy must never +decide ownership. + +### Expiration is explicit + +A lock with no recorded expiry is **not** "not yet expired". `is_lease_expired` +returns `False` for it, which used to make such a lock permanently non-expiring +*and* permanently ineligible for #760 exact-owner renewal, which only ever +assesses an expired lease. `author_lock_contract.expiration_state` names the +real fact: `recorded`, `missing`, or `unparseable`. A `missing` expiry makes the +lock eligible for the recovery path below rather than stranding it. + +## Recovering an existing incomplete bootstrap lock + +For locks already written by the old bootstrap — including those whose branches +already carry legitimate committed and pushed work — use: + +```text +gitea_inspect_issue_lock_contract(issue_number, branch_name, worktree_path, remote=...) +gitea_recover_incomplete_bootstrap_lock(issue_number, branch_name, worktree_path, expected_head, remote=...) +``` + +`gitea_inspect_issue_lock_contract` is strictly read-only: it performs no lock, +lease, branch, worktree, issue, or pull-request mutation. Use it first to see +which fields are missing and what the recommended action is; pass `dry_run=True` +to the recovery tool to preview the decision without writing. + +`gitea_recover_incomplete_bootstrap_lock` upgrades that one lock to the +canonical contract. Before writing anything it verifies: + +* repository (`remote`, `org`, `repo`) and issue number +* claimant username **and** profile against the server-resolved values — a + matching username alone is never accepted +* branch, worktree path, worktree existence, and worktree registration +* the worktree is on the recorded branch +* the observed head equals the caller's `expected_head` +* the existing lock's generation and provenance state +* the absence of healthy foreign ownership + +What it deliberately does **not** do: + +* it never moves, resets, or rewinds the branch, and never requires + base-equivalence — preserving the committed work is the entire point; +* it never pushes and never creates a pull request; +* it touches only the single lock file for that exact remote/org/repo/issue; +* it accepts no caller-supplied provenance and no caller-supplied authorization + flag — both are minted server-side. + +A recovered lock records a `bootstrap_lock_recovery` block holding both sides of +the transition — prior contract, prior missing fields, prior generation, prior +owning session, the replacement `task_session_id`, and the preserved head — so a +recovered claim never reads as an original one. + +### Gates, in order + +`gitea_recover_incomplete_bootstrap_lock` is an author-only durable-lock +mutation and carries the same three gates as every comparable author operation, +in this order: + +1. `role_session_router.check_author_mutation_after_reviewer_stop` — no author + fallback after a reviewer `wrong_role_stop`. +2. `_namespace_mutation_block(task, remote=remote, author_role_exclusive=True)` — + the namespace wall. It refuses a reviewer-bound session and, because this + task's required permission is `gitea.issue.comment` (which merger, + controller, and reconciler profiles also hold), additionally requires the + active profile's derived role kind to be exactly `author`. A refusal carries + `namespace_block: true` and emits a `BLOCKED` audit record naming the + namespace and profile. +3. `_profile_permission_block` — operation, provenance, and session-context + gates. + +Exact-owner claimant matching inside `assess_bootstrap_lock_recovery` runs +*after* all three. It is a further layer, never a substitute for them: on its +own it refuses one step too late and leaves the audit trail silent about the +attempt. + +### Refusals + +| `refusal_code` | Meaning | +| --- | --- | +| `no_durable_lock` | nothing to recover | +| `already_canonical` | lock is fine; rewriting would invalidate a live heartbeat token | +| `foreign_claimant` | recorded claimant is not the active identity/profile pair | +| `healthy_foreign_lock` | a live foreign-owned lock; takeover is not a recovery path | +| `identity_unresolved` | identity or profile could not be resolved | +| `binding_mismatch` | repository, issue, branch, or worktree does not match | +| `worktree_invalid` | worktree missing, unregistered, or on another branch | +| `head_mismatch` | the worktree moved under the caller | + +## The #447 create-PR provenance guard is unchanged + +`issue_lock_provenance.assess_lock_file_for_create_pr` still requires both a +sanctioned `lock_provenance` and a `work_lease`, and the sanctioned source set +was **not** widened. Bootstrap writes through +`issue_lock_provenance.SOURCE_LOCK_ISSUE` — the lock it produces *is* a +canonical lock, not a second dialect with its own exemption. Bootstrap now +satisfies the guard rather than the guard being relaxed to admit bootstrap. diff --git a/docs/mcp-tool-inventory.md b/docs/mcp-tool-inventory.md index 799d0d0..e93d359 100644 --- a/docs/mcp-tool-inventory.md +++ b/docs/mcp-tool-inventory.md @@ -103,6 +103,7 @@ that gates each call, not which tools exist. - `gitea_get_shell_health` - `gitea_heartbeat_issue_lock` - `gitea_heartbeat_reviewer_pr_lease` +- `gitea_inspect_issue_lock_contract` - `gitea_inspect_workflow_lease` - `gitea_issue_irrecoverable_provenance_authorization` - `gitea_list_dependency_edges` @@ -134,6 +135,7 @@ that gates each call, not which tools exist. - `gitea_record_pre_review_command` - `gitea_record_shell_spawn_outcome` - `gitea_record_stable_branch_push_attempt` +- `gitea_recover_incomplete_bootstrap_lock` - `gitea_release_merger_pr_lease` - `gitea_release_reviewer_pr_lease` - `gitea_release_workflow_lease` diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 3721039..0c3aacc 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -2110,6 +2110,8 @@ import issue_lock_store # noqa: E402 import issue_lock_adoption # noqa: E402 import issue_lock_recovery # noqa: E402 import issue_lock_renewal # noqa: E402 +import author_lock_contract # noqa: E402 +import bootstrap_lock_recovery # noqa: E402 import dirty_orphan_worktree_recovery # noqa: E402 # #860 dirty orphan recovery import dirty_same_claimant_session_rebind # noqa: E402 # #864 import stacked_pr_support # noqa: E402 @@ -2658,33 +2660,17 @@ def _build_author_issue_work_lease( worktree_path: str, host: str | None, ) -> dict: - created = _work_lease_now() - # #790 Slice A: the window comes from the central policy, not a literal here. - # It is also now a *sliding* window — the lease lives ``initial_ttl_minutes`` - # past its last valid heartbeat rather than a fixed four hours past its - # creation, so an abandoned task stops holding the claim within one TTL. - policy = lease_policy.policy_for(lease_policy.TASK_CLASS_AUTHOR_ISSUE_WORK) - expires = created + timedelta(minutes=policy.initial_ttl_minutes) - return { - "operation_type": AUTHOR_ISSUE_WORK_LEASE, - "issue_number": issue_number, - "pr_number": None, - "branch": branch_name, - "worktree_path": worktree_path, - "claimant": _work_lease_claimant(host), - "created_at": _work_lease_timestamp(created), - "expires_at": _work_lease_timestamp(expires), - "last_heartbeat_at": _work_lease_timestamp(created), - # #790 AC-N1: the ownership key for this task. Distinct from the recorded - # PID, which is the shared daemon and identifies no individual task. - "task_session_id": issue_lock_store.mint_task_session_id( - AUTHOR_ISSUE_WORK_LEASE - ), - # #790 AC-N8: the explicit lifecycle marker. Its absence — never a - # timestamp comparison — is what makes a lock legacy. - "lifecycle_version": lease_policy.LIFECYCLE_HEARTBEAT_V1, - "heartbeat_count": 1, - } + # #953: the lease shape now lives in author_lock_contract so that bootstrap + # and gitea_lock_issue cannot drift apart again. The policy-derived sliding + # TTL (#790 Slice A) and the task-session ownership key (#790 AC-N1) are + # unchanged — they simply have one definition instead of two. + return author_lock_contract.build_author_issue_work_lease( + issue_number=issue_number, + branch_name=branch_name, + worktree_path=worktree_path, + claimant=_work_lease_claimant(host), + created=_work_lease_now(), + ) def _active_work_lease_block( @@ -4954,6 +4940,370 @@ def gitea_heartbeat_issue_lock( return outcome +def _observe_recovery_worktree(worktree_path: str) -> dict: + """Observe head, branch, existence, and registration for lock recovery. + + Read-only: it runs ``git`` queries and touches nothing. Kept separate from + the decision so the decision stays a pure function of observations (#953). + """ + observation = { + "worktree_exists": os.path.isdir(worktree_path), + "worktree_registered": False, + "current_branch": "", + "observed_head": "", + } + if not observation["worktree_exists"]: + return observation + try: + observation["current_branch"] = subprocess.run( + ["git", "-C", worktree_path, "rev-parse", "--abbrev-ref", "HEAD"], + capture_output=True, + text=True, + check=False, + ).stdout.strip() + observation["observed_head"] = subprocess.run( + ["git", "-C", worktree_path, "rev-parse", "HEAD"], + capture_output=True, + text=True, + check=False, + ).stdout.strip() + listing = subprocess.run( + ["git", "-C", worktree_path, "worktree", "list", "--porcelain"], + capture_output=True, + text=True, + check=False, + ).stdout + real = os.path.realpath(worktree_path) + observation["worktree_registered"] = any( + os.path.realpath(line.split(" ", 1)[1].strip()) == real + for line in listing.splitlines() + if line.startswith("worktree ") + ) + except Exception: # fail closed: unobservable is not provable + return observation + return observation + + +@mcp.tool() +def gitea_inspect_issue_lock_contract( + issue_number: int, + branch_name: str | None = None, + worktree_path: str | None = None, + remote: str = "dadeschools", + host: str | None = None, + org: str | None = None, + repo: str | None = None, +) -> dict: + """Inspect a durable author issue lock against the canonical contract (#953 AC8/AC16). + + Strictly read-only. It performs no lock, lease, branch, worktree, issue, or + pull-request mutation of any kind — it reads the durable lock record and + reports. Use it to find out *why* a lock is being refused before choosing a + recovery path, and to confirm afterwards that recovery produced a canonical + lock. + + Reports which canonical fields are missing, where the claimant is recorded + (``work_lease`` is canonical, top level is the legacy/bootstrap placement), + whether an expiration is actually recorded — as opposed to absent, which + used to masquerade as "not yet expired" — whether the lock can be + heartbeated, and whether it satisfies the untouched #447 create-PR + provenance guard. + + Args: + issue_number: The issue whose lock to inspect. + branch_name: Optional; when given, the recovery eligibility preview is + evaluated against this branch. + worktree_path: Optional; when given, the recovery eligibility preview is + evaluated against this worktree. + remote: Known instance — 'dadeschools' or 'prgs'. + host: Override the Gitea host. + org: Override the owner/organization. + repo: Override the repository name. + + Returns: + dict with 'success', 'lock_present', 'lock_contract' (the structural + verdict), 'recommended_action', and — when branch_name and + worktree_path are supplied — a non-mutating 'recovery_preview'. + """ + blocked = _profile_permission_block( + "gitea.read", + issue_number=issue_number, + remote=remote, + host=host, + org=org, + repo=repo, + org_explicit=org is not None, + repo_explicit=repo is not None, + ) + if blocked: + return blocked + + h, o, r = _resolve(remote, host, org, repo) + existing = _load_existing_issue_lock( + remote=remote, org=o, repo=r, issue_number=issue_number + ) + contract = author_lock_contract.assess_lock_contract(existing) + result = { + "success": True, + "performed": False, + "mutation_performed": False, + "read_only": True, + "issue_number": issue_number, + "lock_present": bool(existing), + "lock_contract": contract, + "lock_freshness": ( + issue_lock_store.assess_lock_freshness(dict(existing)) + if existing + else {"status": issue_lock_store.STATUS_ABSENT, "live": False} + ), + "recommended_action": author_lock_contract.recommended_action(contract), + } + + if branch_name and worktree_path: + resolved = issue_lock_worktree.resolve_author_worktree_path( + worktree_path, _canonical_local_git_root() + ) + observation = _observe_recovery_worktree(resolved) + claimant = _work_lease_claimant(h) + result["recovery_preview"] = bootstrap_lock_recovery.assess_bootstrap_lock_recovery( + existing, + issue_number=issue_number, + branch_name=branch_name, + worktree_path=resolved, + remote=remote, + org=o, + repo=r, + identity=claimant.get("username"), + profile=claimant.get("profile"), + observed_head=observation["observed_head"], + declared_head=None, + worktree_exists=observation["worktree_exists"], + worktree_registered=observation["worktree_registered"], + current_branch=observation["current_branch"], + ) + return result + + +@mcp.tool() +def gitea_recover_incomplete_bootstrap_lock( + issue_number: int, + branch_name: str, + worktree_path: str, + expected_head: str, + remote: str = "dadeschools", + host: str | None = None, + org: str | None = None, + repo: str | None = None, + dry_run: bool = False, +) -> dict: + """Upgrade an incomplete bootstrap issue lock to the canonical contract (#953 AC8-AC11). + + Explicit, target-specific recovery. It does **not** widen + ``gitea_lock_issue``, and it is not a takeover path. + + The state it repairs: ``gitea_bootstrap_author_issue_worktree`` reported + success but wrote a lock with the claimant at the top level, no + ``work_lease``, no ``lock_provenance``, and no expiry. The author then + implemented, committed, and pushed — following bootstrap's own reported next + action — after which heartbeat, re-lock, exact-owner renewal, and the #447 + create-PR guard all refuse simultaneously. + + Deliberate non-behaviours: the branch is never moved, reset, or rewound, and + base-equivalence is never required — preserving the already-committed and + pushed work is the entire point. Nothing is pushed and no pull request is + created. Only the single lock file for this exact (remote, org, repo, issue) + is written. + + Ownership is proven, never asserted. The claimant recorded on the durable + lock must match **both** the server-resolved identity and the active + profile; a matching username alone is refused. Repository, issue, branch, + worktree, registration, current branch, and head are all verified before any + write, and the declared ``expected_head`` must equal the observed head. A + healthy foreign-owned lock is refused outright. Provenance and authorization + are minted server-side — there is no parameter through which a caller can + supply either. + + Args: + issue_number: The issue whose incomplete lock is being recovered. + branch_name: The branch recorded on the lock; must match. + worktree_path: The registered worktree recorded on the lock; must match. + expected_head: Full SHA the caller believes the worktree is at. A + mismatch fails closed, so a worktree that moved underneath the + caller cannot be recovered against stale evidence. + remote: Known instance — 'dadeschools' or 'prgs'. + host: Override the Gitea host. + org: Override the owner/organization. + repo: Override the repository name. + dry_run: Report the decision and evidence, mutate nothing. + + Returns: + dict with 'success', 'performed', the resulting canonical + 'lock_contract' and 'work_lease', the auditable + 'bootstrap_lock_recovery' transition record, and 'exact_next_action'; on + refusal 'success'/'performed' False with 'refusal_code' and 'reasons' + naming exactly which evidence was missing, and no write performed. + """ + task = "recover_incomplete_bootstrap_lock" + ok, block_reasons = role_session_router.check_author_mutation_after_reviewer_stop( + task + ) + if not ok: + return _author_mutation_block(block_reasons) + + # #953 F1: the namespace/session wall every author state-creating mutation + # carries, and which this tool — the structural neighbour of + # gitea_recover_dirty_orphaned_issue_worktree, writing the same durable + # lock — was the only one to omit. Exact-owner claimant matching inside + # assess_bootstrap_lock_recovery is a later layer, not a substitute: it + # refuses one commit too late and leaves no BLOCKED audit record of the + # attempt. author_role_exclusive is required here because this task is gated + # on gitea.issue.comment, which merger, controller, and reconciler profiles + # also hold. + blocked = _namespace_mutation_block( + task, remote=remote, author_role_exclusive=True + ) + if blocked: + return blocked + + blocked = _profile_permission_block( + task_capability_map.required_permission(task), + issue_number=issue_number, + remote=remote, + host=host, + org=org, + repo=repo, + org_explicit=org is not None, + repo_explicit=repo is not None, + ) + if blocked: + return blocked + + h, o, r = _resolve(remote, host, org, repo) + resolved_worktree = issue_lock_worktree.resolve_author_worktree_path( + worktree_path, _canonical_local_git_root() + ) + existing = _load_existing_issue_lock( + remote=remote, org=o, repo=r, issue_number=issue_number + ) + observation = _observe_recovery_worktree(resolved_worktree) + + # The claimant pair is resolved server-side from the live session; the + # caller cannot influence which identity or profile recovery compares + # against. + claimant = _work_lease_claimant(h) + assessment = bootstrap_lock_recovery.assess_bootstrap_lock_recovery( + existing, + issue_number=issue_number, + branch_name=branch_name, + worktree_path=resolved_worktree, + remote=remote, + org=o, + repo=r, + identity=claimant.get("username"), + profile=claimant.get("profile"), + observed_head=observation["observed_head"], + declared_head=expected_head, + worktree_exists=observation["worktree_exists"], + worktree_registered=observation["worktree_registered"], + current_branch=observation["current_branch"], + ) + + if not assessment["recovery_sanctioned"]: + return { + "success": False, + "performed": False, + "mutation_performed": False, + "issue_number": issue_number, + "refusal_code": assessment["refusal_code"], + "reasons": assessment["reasons"], + "message": bootstrap_lock_recovery.format_recovery_refusal(assessment), + "lock_contract": assessment["contract"], + "evidence": assessment["evidence"], + } + + if dry_run: + return { + "success": True, + "performed": False, + "mutation_performed": False, + "dry_run": True, + "issue_number": issue_number, + "would_recover": True, + "lock_contract": assessment["contract"], + "evidence": assessment["evidence"], + "exact_next_action": ( + "Re-run without dry_run=True to upgrade this lock to the " + "canonical contract." + ), + } + + recovered = author_lock_contract.build_canonical_issue_lock( + issue_number=issue_number, + branch_name=branch_name, + worktree_path=resolved_worktree, + remote=remote, + org=o, + repo=r, + identity=claimant.get("username"), + profile=claimant.get("profile"), + tool="gitea_recover_incomplete_bootstrap_lock", + source=issue_lock_provenance.SOURCE_LOCK_ISSUE, + owner_session=(existing or {}).get("owner_session"), + expected_base_sha=(existing or {}).get("expected_base_sha"), + ) + # AC10: preserve both sides of the transition so a recovered lock never + # reads as an original claim. + recovered["bootstrap_lock_recovery"] = bootstrap_lock_recovery.build_recovery_record( + assessment, + recovered_at=_work_lease_timestamp(_work_lease_now()), + new_task_session_id=recovered["work_lease"]["task_session_id"], + ) + + try: + lock_path = issue_lock_store.bind_session_lock( + recovered, + expected_generation=assessment["expected_generation"], + recovery_sanctioned=True, + ) + except Exception as exc: + return { + "success": False, + "performed": False, + "mutation_performed": False, + "issue_number": issue_number, + "refusal_code": "lock_write_failed", + "reasons": [str(exc)], + "message": f"Recovered lock could not be persisted: {exc} (fail closed)", + } + + written = issue_lock_store.read_lock_file(lock_path) + contract = author_lock_contract.assess_lock_contract(written) + return { + "success": True, + "performed": True, + "mutation_performed": True, + "issue_number": issue_number, + "branch_name": branch_name, + "worktree_path": resolved_worktree, + "lock_file_path": lock_path, + "lock_contract": contract, + "work_lease": (written or {}).get("work_lease"), + "task_session_id": contract["task_session_id"], + "lock_generation": (written or {}).get("lock_generation"), + "prior_generation": assessment["expected_generation"], + "bootstrap_lock_recovery": (written or {}).get("bootstrap_lock_recovery"), + "preserved_head": observation["observed_head"], + "branch_reset": False, + "pushed": False, + "pr_created": False, + "exact_next_action": ( + "Lock is canonical. Heartbeat it with the returned task_session_id, " + "then continue the author workflow; publish and create the pull " + "request through the normal sanctioned calls." + ), + } + + @mcp.tool() def gitea_recover_dirty_orphaned_issue_worktree( issue_number: int, @@ -15155,8 +15505,21 @@ def _profile_permission_block(required_operation: str, **extra_fields) -> dict | ) -def _namespace_mutation_block(mutation_task: str, **extra_fields) -> dict | None: - """Reviewer/author namespace alignment gate (#209).""" +def _namespace_mutation_block( + mutation_task: str, + *, + author_role_exclusive: bool = False, + **extra_fields, +) -> dict | None: + """Reviewer/author namespace alignment gate (#209). + + ``author_role_exclusive`` additionally requires the active profile's derived + role kind to be exactly ``author`` (#953 F1). Off by default, so the six + pre-existing call sites are unchanged. Tools whose required permission is + ``gitea.issue.comment`` — which every configured role holds — opt in, since + the reviewer-namespace check alone would let a merger, controller, or + reconciler session through to a durable author lock write. + """ required_permission = task_capability_map.required_permission(mutation_task) required_role = task_capability_map.required_role(mutation_task) # #714: evaluate active profile only — never auto-switch. @@ -15178,6 +15541,9 @@ def _namespace_mutation_block(mutation_task: str, **extra_fields) -> dict | None } ok, reasons = role_namespace_gate.check_author_mutation_namespace( mutation_task, profile) + if ok and author_role_exclusive: + ok, reasons = role_namespace_gate.check_author_role_kind( + mutation_task, profile) if ok: return None blocked = { diff --git a/issue_lock_store.py b/issue_lock_store.py index f216656..8237305 100644 --- a/issue_lock_store.py +++ b/issue_lock_store.py @@ -299,9 +299,13 @@ def _ownership_refusals( f"lock worktree '{lock.get('worktree_path')}' does not match " f"'{worktree_path}'" ) - lease = lock.get("work_lease") if isinstance(lock, dict) else None - claimant = lease.get("claimant") if isinstance(lease, dict) else None - claimant = claimant if isinstance(claimant, dict) else {} + # #953 AC2/AC13/AC14: read through the shared claimant reader so a lock + # written by bootstrap — which records the claimant at the top level — is + # not refused for "not recording a claimant" when it plainly records one. + # This is not a widening: the values are still compared against the + # server-resolved identity and profile immediately below, so a legacy + # placement grants nothing that the canonical placement would not. + claimant = lock_claimant(lock) if isinstance(lock, dict) else {} recorded_identity = str(claimant.get("username") or "").strip() recorded_profile = str(claimant.get("profile") or "").strip() if not recorded_identity or not recorded_profile: @@ -636,6 +640,99 @@ def iter_lock_files(lock_dir: str | None = None) -> list[str]: return sorted(paths) +def release_session_lock( + *, + issue_number: int, + session: str, + lock_dir: str | None = None, + remote: str | None = None, + org: str | None = None, + repo: str | None = None, +) -> str: + """Remove exactly the durable lock *session* created for *issue_number*. + + ``author_issue_bootstrap.run_compensating_recovery`` has called this name + since #850, but it was never defined: the call raised ``AttributeError`` + into a bare ``except Exception: pass``, so the lock half of every + compensating rollback silently did nothing. The branch and worktree were + removed and the lock was left behind — a state no sanctioned tool can act + on, since recovery refuses ``worktree_invalid`` and ``gitea_lock_issue`` has + no worktree to bind (#953 review 632 F2). + + Ownership is proven, not asserted. A record is removed only when its + recorded ``owner_session`` equals *session* and its issue number matches; + ``remote``/``org``/``repo`` narrow it further when supplied. Zero matches or + more than one both raise, so a caller can never delete a lock it does not + own and an ambiguous directory is never guessed at. The ``.json.lock`` flock + sidecar is deliberately left in place — it is a zero-byte mutex another + process may hold, and removing it under contention would be a race. + + Returns the removed lock file path. + """ + target_issue = int(issue_number) + owner = str(session or "").strip() + if not owner: + raise ValueError( + "release_session_lock requires the owning session id (fail closed)" + ) + + def _is_owned_durable_lock(record: dict[str, Any] | None) -> bool: + # A durable lock, not a bootstrap phase journal or a session pointer, + # both of which can share a directory and carry the same issue number + # and owner_session. + if not record or "lock_generation" not in record: + return False + if not str(record.get("branch_name") or "").strip(): + return False + if not str(record.get("worktree_path") or "").strip(): + return False + try: + if int(record.get("issue_number") or 0) != target_issue: + return False + except (TypeError, ValueError): + return False + return str(record.get("owner_session") or "").strip() == owner + + # Prefer the exact keyed path when the caller knows the repository; scanning + # is the fallback for callers that only carry the issue number. + if remote and org and repo: + exact = lock_file_path( + remote=remote, + org=org, + repo=repo, + issue_number=target_issue, + lock_dir=lock_dir, + ) + if not _is_owned_durable_lock(read_lock_file(exact)): + raise FileNotFoundError( + f"durable issue lock '{exact}' is absent or is not owned by " + f"session '{owner}' (fail closed; nothing released)" + ) + os.remove(exact) + return exact + + matches: list[str] = [] + for path in iter_lock_files(lock_dir): + if _is_owned_durable_lock(read_lock_file(path)): + matches.append(path) + + if not matches: + raise FileNotFoundError( + f"no durable issue lock for issue #{target_issue} is owned by " + f"session '{owner}' (fail closed; nothing released)" + ) + if len(matches) > 1: + raise RuntimeError( + f"{len(matches)} durable locks for issue #{target_issue} claim " + f"session '{owner}'; refusing to guess which to release " + "(fail closed)" + ) + + path = matches[0] + os.remove(path) + return path + + def find_lock_for_branch( *, remote: str, @@ -1112,21 +1209,43 @@ def assess_same_issue_lease_conflict( ) -def _lock_claimant(lock: dict[str, Any] | None) -> dict[str, str]: +def lock_claimant(lock: dict[str, Any] | None) -> dict[str, str]: + """Read the claimant from either canonical or legacy placement (#953 AC13/AC14). + + ``work_lease.claimant`` is the canonical placement and is preferred; a + top-level ``claimant`` is the legacy/bootstrap placement and is accepted as + a fallback. This is the single definition. Before #953 the readers + disagreed: this module, ``issue_lock_renewal``, and ``issue_lock_recovery`` + tolerated both placements, while ``_ownership_refusals`` looked only in + ``work_lease`` — which is what made a bootstrap-written lock + un-heartbeatable. + + Preferring ``work_lease`` over the top level is deliberate: once a legacy + lock is upgraded, the canonical placement is authoritative and a stale + top-level copy must never win. + + This decides *where to look*, never whether ownership is proven — every + caller still compares these values against server-resolved identity and + profile. + """ if not isinstance(lock, dict): return {} - claimant = lock.get("claimant") + lease = lock.get("work_lease") + claimant = lease.get("claimant") if isinstance(lease, dict) else None if not isinstance(claimant, dict): - lease = lock.get("work_lease") - claimant = lease.get("claimant") if isinstance(lease, dict) else None + claimant = lock.get("claimant") if not isinstance(claimant, dict): return {} return { - "username": str(claimant.get("username") or ""), - "profile": str(claimant.get("profile") or ""), + "username": str(claimant.get("username") or "").strip(), + "profile": str(claimant.get("profile") or "").strip(), } +#: Back-compatible alias for the pre-#953 private name. +_lock_claimant = lock_claimant + + def assess_foreign_lock_overwrite( existing_lock: dict[str, Any] | None, incoming_lock: dict[str, Any], diff --git a/role_namespace_gate.py b/role_namespace_gate.py index 7ed1073..a5e3b25 100644 --- a/role_namespace_gate.py +++ b/role_namespace_gate.py @@ -74,6 +74,43 @@ def check_author_mutation_namespace( return True, [] +def check_author_role_kind( + mutation_task: str, + profile: dict, +) -> tuple[bool, list[str]]: + """Author-exclusive wall for durable-lock mutations (#953 F1). + + ``check_author_mutation_namespace`` walls off reviewer-bound sessions, which + is the whole gate for tasks whose required permission is itself author-only + (``gitea.pr.create``, ``gitea.repo.commit``). It is *not* sufficient for a + task gated on ``gitea.issue.comment``, which every configured role holds: a + merger, controller, or reconciler session would clear both the namespace + check and the permission gate and still reach the durable write. + + Opt-in per call site and additive. It refuses any active profile whose + derived role kind is not exactly ``author`` for a task the router declares + author-required, and grants nothing to anyone — a ``mixed`` profile is + refused rather than admitted. + """ + required_role = role_session_router.required_role_for_task(mutation_task) + if required_role != "author": + return True, [] + + allowed = profile.get("allowed_operations") or [] + forbidden = profile.get("forbidden_operations") or [] + active_role = derive_role_kind(allowed, forbidden) + if active_role == "author": + return True, [] + + profile_name = profile.get("profile_name") or "" + namespace = infer_mcp_namespace(profile_name) + return False, [ + f"author mutation '{mutation_task}' blocked: active session role kind is " + f"'{active_role}', not 'author' ({profile_name} / {namespace}); this " + "operation writes a durable author issue lock and is author-exclusive", + ] + + def mutation_audit_context(mutation_task: str, profile: dict, *, remote=None, repository=None) -> dict: """Structured mutation metadata for audit records (#209).""" diff --git a/role_session_router.py b/role_session_router.py index e557266..6b0a6de 100644 --- a/role_session_router.py +++ b/role_session_router.py @@ -75,6 +75,10 @@ AUTHOR_TASKS = frozenset({ "push_branch", "bootstrap_author_issue_worktree", "gitea_bootstrap_author_issue_worktree", + # #953: recovery of an incomplete bootstrap lock is an author-only durable + # state mutation and belongs to the same class as bootstrap itself. + "recover_incomplete_bootstrap_lock", + "gitea_recover_incomplete_bootstrap_lock", "create_pr", "comment_pr", "address_pr_change_requests", @@ -112,6 +116,12 @@ TASK_REQUIRED_ROLE = { "claim_issue": "author", "create_branch": "author", "push_branch": "author", + # #953: without this entry ``required_role_for_task`` returns None and + # ``role_namespace_gate.check_author_mutation_namespace`` short-circuits to + # "allowed" — the namespace wall on the recovery tool would be inert. The + # capability map already records the same role; both tables must agree. + "recover_incomplete_bootstrap_lock": "author", + "gitea_recover_incomplete_bootstrap_lock": "author", "create_pr": "author", "comment_pr": "author", "address_pr_change_requests": "author", diff --git a/task_capability_map.py b/task_capability_map.py index a30b7f9..393ce98 100644 --- a/task_capability_map.py +++ b/task_capability_map.py @@ -41,6 +41,27 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = { "permission": "gitea.issue.comment", "role": "author", }, + # #953: target-specific upgrade of an incomplete bootstrap lock (explicit + # operation, never a widening of lock_issue). Author-only, and the tool + # additionally proves exact-owner claimant match before writing. + "recover_incomplete_bootstrap_lock": { + "permission": "gitea.issue.comment", + "role": "author", + }, + "gitea_recover_incomplete_bootstrap_lock": { + "permission": "gitea.issue.comment", + "role": "author", + }, + # #953: read-only lock contract inspection. Read permission only — it must + # never be able to mutate. + "inspect_issue_lock_contract": { + "permission": "gitea.read", + "role": "author", + }, + "gitea_inspect_issue_lock_contract": { + "permission": "gitea.read", + "role": "author", + }, # #860: dirty orphaned same-claimant worktree recovery (explicit operation). "recover_dirty_orphaned_issue_worktree": { "permission": "gitea.issue.comment", diff --git a/tests/test_issue_953_bootstrap_lock_contract.py b/tests/test_issue_953_bootstrap_lock_contract.py new file mode 100644 index 0000000..04fc741 --- /dev/null +++ b/tests/test_issue_953_bootstrap_lock_contract.py @@ -0,0 +1,1975 @@ +"""One canonical bootstrap/lock contract and its recovery path (#953). + +Covers the defect in which ``gitea_bootstrap_author_issue_worktree`` reported a +lock as created, wrote a shape no downstream reader accepts, and then directed +the author to implement — after which heartbeat, re-lock, exact-owner renewal, +and the #447 create-PR guard all refuse simultaneously and no sanctioned +recovery path remains eligible. + +Every fixture here is synthetic and isolated: locks are written into temporary +directories and the git repositories are created per-test with ``git init``. +The #949-shaped regression reproduces that lock *shape*; it never touches the +real issue #949 branch, worktree, lock, issue, or head. +""" + +import contextlib +import os +import subprocess +import sys +import tempfile +import unittest +from datetime import datetime, timedelta, timezone +from unittest import mock + +sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) +sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent)) + +from mutation_profile_fixture import shared_mutation_env # noqa: E402 + +import author_issue_bootstrap # noqa: E402 +import author_lock_contract # noqa: E402 +import bootstrap_lock_recovery # noqa: E402 +import gitea_audit # noqa: E402 +import issue_lock_provenance # noqa: E402 +import issue_lock_store # noqa: E402 +import mcp_server # noqa: E402 +import role_namespace_gate # noqa: E402 +import role_session_router # noqa: E402 + +ISSUE = 9530 +BRANCH = f"fix/issue-{ISSUE}-canonical-contract" +IDENTITY = "example-author-user" +PROFILE = "example-author-profile" +FOREIGN_IDENTITY = "example-other-user" +FOREIGN_PROFILE = "example-other-profile" +REMOTE = "prgs" +ORG = "ExampleOrg" +REPO = "ExampleRepo" +HEAD = "a" * 40 +OTHER_HEAD = "b" * 40 + + +def _ts(delta_minutes: int = 0) -> str: + return ( + (datetime.now(timezone.utc) + timedelta(minutes=delta_minutes)) + .replace(microsecond=0) + .isoformat() + .replace("+00:00", "Z") + ) + + +def bootstrap_shaped_lock(**overrides): + """The exact malformed shape #949 was left in by the old bootstrap. + + Claimant at the lock top level, ``lease_id`` null, and no ``work_lease``, + ``lock_provenance``, or ``expires_at``. + """ + lock = { + "remote": REMOTE, + "org": ORG, + "repo": REPO, + "issue_number": ISSUE, + "branch": BRANCH, + "branch_name": BRANCH, + "worktree_path": "/scratch/wt-9530", + "owner_session": "author_issue_work-deadbeefdeadbeef", + "claimant": {"username": IDENTITY, "profile": PROFILE}, + "assignment_id": None, + "lease_id": None, + "expected_base_sha": OTHER_HEAD, + "lock_generation": 1, + } + lock.update(overrides) + return lock + + +def canonical_lock(worktree="/scratch/wt-9530", **overrides): + lock = author_lock_contract.build_canonical_issue_lock( + issue_number=ISSUE, + branch_name=BRANCH, + worktree_path=worktree, + remote=REMOTE, + org=ORG, + repo=REPO, + identity=IDENTITY, + profile=PROFILE, + tool="gitea_lock_issue", + ) + lock["lock_generation"] = 2 + lock["session_pid"] = os.getpid() + lock.update(overrides) + return lock + + +class CanonicalContractShape(unittest.TestCase): + """AC1, AC6, AC13: one contract, emitted with every required field.""" + + def test_bootstrap_builder_emits_the_full_canonical_schema(self): + lock = author_lock_contract.build_canonical_issue_lock( + issue_number=ISSUE, + branch_name=BRANCH, + worktree_path="/scratch/wt", + remote=REMOTE, + org=ORG, + repo=REPO, + identity=IDENTITY, + profile=PROFILE, + tool="gitea_bootstrap_author_issue_worktree", + source=author_lock_contract.SOURCE_BOOTSTRAP, + ) + for field in author_lock_contract.REQUIRED_LOCK_FIELDS: + self.assertIn(field, lock, f"canonical lock missing {field}") + for field in author_lock_contract.REQUIRED_WORK_LEASE_FIELDS: + self.assertIn(field, lock["work_lease"], f"work_lease missing {field}") + self.assertEqual( + lock["work_lease"]["claimant"], + {"username": IDENTITY, "profile": PROFILE}, + ) + self.assertEqual( + lock["lock_provenance"]["written_by_tool"], + "gitea_bootstrap_author_issue_worktree", + ) + + def test_bootstrap_and_lock_issue_produce_the_same_contract(self): + """AC13: the two writers must not disagree about what a lock is.""" + from_bootstrap = author_lock_contract.build_canonical_issue_lock( + issue_number=ISSUE, + branch_name=BRANCH, + worktree_path="/scratch/wt", + remote=REMOTE, + org=ORG, + repo=REPO, + identity=IDENTITY, + profile=PROFILE, + tool="gitea_bootstrap_author_issue_worktree", + source=author_lock_contract.SOURCE_BOOTSTRAP, + ) + from_lock_issue = author_lock_contract.build_canonical_issue_lock( + issue_number=ISSUE, + branch_name=BRANCH, + worktree_path="/scratch/wt", + remote=REMOTE, + org=ORG, + repo=REPO, + identity=IDENTITY, + profile=PROFILE, + tool="gitea_lock_issue", + ) + self.assertEqual(sorted(from_bootstrap.keys()), sorted(from_lock_issue.keys())) + self.assertEqual( + sorted(from_bootstrap["work_lease"].keys()), + sorted(from_lock_issue["work_lease"].keys()), + ) + for lock in (from_bootstrap, from_lock_issue): + self.assertTrue( + author_lock_contract.assess_lock_contract(lock)["canonical"] + ) + + def test_no_successful_build_returns_a_null_ownership_token(self): + """AC6: the fencing token every later check keys on is never null.""" + lock = author_lock_contract.build_canonical_issue_lock( + issue_number=ISSUE, + branch_name=BRANCH, + worktree_path="/scratch/wt", + remote=REMOTE, + org=ORG, + repo=REPO, + identity=IDENTITY, + profile=PROFILE, + tool="gitea_bootstrap_author_issue_worktree", + ) + self.assertTrue(lock["work_lease"]["task_session_id"]) + self.assertIsNotNone(lock["work_lease"]["expires_at"]) + + def test_provenance_cannot_be_supplied_by_a_caller(self): + """Safety: provenance is minted server-side, never accepted.""" + import inspect as _inspect + + params = _inspect.signature( + author_lock_contract.build_canonical_issue_lock + ).parameters + self.assertNotIn("lock_provenance", params) + self.assertNotIn("provenance", params) + + +class MalformedAndPartialLocks(unittest.TestCase): + """AC7, AC12, AC19: malformed, partial, and missing-expiration locks.""" + + def test_bootstrap_shaped_lock_is_reported_as_not_canonical(self): + assessment = author_lock_contract.assess_lock_contract(bootstrap_shaped_lock()) + self.assertFalse(assessment["canonical"]) + self.assertIn("work_lease", assessment["missing_fields"]) + self.assertIn("lock_provenance", assessment["missing_fields"]) + + def test_missing_fields_are_reported_structurally_and_by_name(self): + """AC7: the refusal names what is missing, not just that it failed.""" + assessment = author_lock_contract.assess_lock_contract(bootstrap_shaped_lock()) + message = author_lock_contract.format_contract_refusal(assessment) + self.assertIn("work_lease", message) + self.assertIn("lock_provenance", message) + self.assertIsInstance(assessment["missing_fields"], list) + + def test_missing_expiration_is_explicit_not_never_expiring(self): + """AC12: the bug — absent expiry read as 'not yet expired'.""" + lock = bootstrap_shaped_lock() + # The pre-existing reader still reports "not expired" for this lock... + self.assertFalse(issue_lock_store.is_lease_expired(lock)) + # ...so the contract states the real fact explicitly instead. + state = author_lock_contract.expiration_state(lock) + self.assertEqual(state["state"], author_lock_contract.EXPIRATION_MISSING) + self.assertIsNone(state["expired"]) + assessment = author_lock_contract.assess_lock_contract(lock) + self.assertTrue( + any("neither expirable nor renewable" in r for r in assessment["reasons"]) + ) + + def test_missing_expiration_lock_is_recoverable_rather_than_stranded(self): + """AC12: it must not be non-expiring *and* ineligible for every path.""" + assessment = bootstrap_lock_recovery.assess_bootstrap_lock_recovery( + bootstrap_shaped_lock(worktree_path="/scratch/wt-9530"), + issue_number=ISSUE, + branch_name=BRANCH, + worktree_path="/scratch/wt-9530", + remote=REMOTE, + org=ORG, + repo=REPO, + identity=IDENTITY, + profile=PROFILE, + observed_head=HEAD, + declared_head=HEAD, + worktree_exists=True, + worktree_registered=True, + current_branch=BRANCH, + ) + self.assertTrue(assessment["recovery_sanctioned"], assessment["reasons"]) + + def test_partial_lock_missing_only_provenance_is_not_canonical(self): + lock = canonical_lock() + lock.pop("lock_provenance") + assessment = author_lock_contract.assess_lock_contract(lock) + self.assertFalse(assessment["canonical"]) + self.assertFalse(assessment["create_pr_eligible"]) + + def test_unparseable_expiration_is_named_rather_than_silently_ignored(self): + lock = canonical_lock() + lock["work_lease"]["expires_at"] = "not-a-timestamp" + state = author_lock_contract.expiration_state(lock) + self.assertEqual(state["state"], author_lock_contract.EXPIRATION_UNPARSEABLE) + + def test_expired_lock_is_reported_as_expired(self): + lock = canonical_lock() + lock["work_lease"]["expires_at"] = _ts(-60) + state = author_lock_contract.expiration_state(lock) + self.assertEqual(state["state"], author_lock_contract.EXPIRATION_RECORDED) + self.assertTrue(state["expired"]) + + def test_absent_lock_reports_absent_contract(self): + assessment = author_lock_contract.assess_lock_contract(None) + self.assertEqual(assessment["contract"], author_lock_contract.CONTRACT_ABSENT) + self.assertIn( + "gitea_lock_issue", author_lock_contract.recommended_action(assessment) + ) + + +class ClaimantCompatibility(unittest.TestCase): + """AC2, AC13, AC14: legacy and canonical claimant placement.""" + + def test_claimant_is_read_from_the_legacy_top_level_placement(self): + recorded = author_lock_contract.lock_claimant(bootstrap_shaped_lock()) + self.assertEqual(recorded, {"username": IDENTITY, "profile": PROFILE}) + + def test_claimant_is_read_from_the_canonical_work_lease_placement(self): + recorded = author_lock_contract.lock_claimant(canonical_lock()) + self.assertEqual(recorded, {"username": IDENTITY, "profile": PROFILE}) + + def test_work_lease_placement_wins_over_a_stale_top_level_copy(self): + """An upgraded lock must not be re-read from its stale legacy copy.""" + lock = canonical_lock() + lock["claimant"] = {"username": FOREIGN_IDENTITY, "profile": FOREIGN_PROFILE} + self.assertEqual( + author_lock_contract.lock_claimant(lock), + {"username": IDENTITY, "profile": PROFILE}, + ) + + def test_ownership_check_accepts_the_legacy_placement(self): + """AC2: the exact refusal that made a fresh bootstrap lock un-heartbeatable.""" + refusals = issue_lock_store._ownership_refusals( + bootstrap_shaped_lock(worktree_path="/scratch/wt-9530"), + issue_number=ISSUE, + branch_name=BRANCH, + worktree_path="/scratch/wt-9530", + identity=IDENTITY, + profile=PROFILE, + ) + self.assertNotIn( + "lock does not record both a claimant username and profile", refusals + ) + self.assertEqual(refusals, []) + + def test_ownership_check_still_refuses_a_mismatched_claimant(self): + """Tolerating the placement must not tolerate the wrong owner.""" + refusals = issue_lock_store._ownership_refusals( + bootstrap_shaped_lock(worktree_path="/scratch/wt-9530"), + issue_number=ISSUE, + branch_name=BRANCH, + worktree_path="/scratch/wt-9530", + identity=FOREIGN_IDENTITY, + profile=PROFILE, + ) + self.assertTrue(any("does not match active identity" in r for r in refusals)) + + def test_ownership_check_still_refuses_a_lock_with_no_claimant_at_all(self): + lock = bootstrap_shaped_lock(worktree_path="/scratch/wt-9530") + lock.pop("claimant") + refusals = issue_lock_store._ownership_refusals( + lock, + issue_number=ISSUE, + branch_name=BRANCH, + worktree_path="/scratch/wt-9530", + identity=IDENTITY, + profile=PROFILE, + ) + self.assertIn( + "lock does not record both a claimant username and profile", refusals + ) + + +class CreatePrProvenanceGuardPreserved(unittest.TestCase): + """AC4 and the safety requirement that #447 is not weakened.""" + + def test_canonical_bootstrap_lock_passes_the_447_guard(self): + lock = author_lock_contract.build_canonical_issue_lock( + issue_number=ISSUE, + branch_name=BRANCH, + worktree_path="/scratch/wt", + remote=REMOTE, + org=ORG, + repo=REPO, + identity=IDENTITY, + profile=PROFILE, + tool="gitea_bootstrap_author_issue_worktree", + source=author_lock_contract.SOURCE_BOOTSTRAP, + ) + verdict = issue_lock_provenance.assess_lock_file_for_create_pr(lock) + self.assertTrue(verdict["proven"], verdict["reasons"]) + + def test_the_old_bootstrap_shape_is_still_rejected_by_the_447_guard(self): + """The guard must keep failing closed on a lock with no provenance.""" + verdict = issue_lock_provenance.assess_lock_file_for_create_pr( + bootstrap_shaped_lock() + ) + self.assertFalse(verdict["proven"]) + self.assertTrue(verdict["block"]) + + def test_guard_still_rejects_an_unsanctioned_provenance_source(self): + lock = canonical_lock() + lock["lock_provenance"]["source"] = "hand_written_by_caller" + verdict = issue_lock_provenance.assess_lock_file_for_create_pr(lock) + self.assertFalse(verdict["proven"]) + + def test_guard_still_rejects_provenance_without_work_lease(self): + lock = canonical_lock() + lock.pop("work_lease") + verdict = issue_lock_provenance.assess_lock_file_for_create_pr(lock) + self.assertFalse(verdict["proven"]) + + def test_sanctioned_source_set_was_not_widened(self): + """Bootstrap satisfies the guard; it does not get its own exemption.""" + self.assertEqual( + author_lock_contract.SOURCE_BOOTSTRAP, + issue_lock_provenance.SOURCE_LOCK_ISSUE, + ) + + +class RecoveryOwnershipVerification(unittest.TestCase): + """AC10, AC11: what recovery proves before it changes lock state.""" + + def _assess(self, lock=None, **overrides): + kwargs = dict( + issue_number=ISSUE, + branch_name=BRANCH, + worktree_path="/scratch/wt-9530", + remote=REMOTE, + org=ORG, + repo=REPO, + identity=IDENTITY, + profile=PROFILE, + observed_head=HEAD, + declared_head=HEAD, + worktree_exists=True, + worktree_registered=True, + current_branch=BRANCH, + ) + kwargs.update(overrides) + target = ( + lock + if lock is not None + else bootstrap_shaped_lock(worktree_path="/scratch/wt-9530") + ) + return bootstrap_lock_recovery.assess_bootstrap_lock_recovery(target, **kwargs) + + def test_exact_owner_recovery_is_sanctioned(self): + self.assertTrue(self._assess()["recovery_sanctioned"]) + + def test_mismatched_repository_is_refused(self): + result = self._assess(repo="OtherRepo") + self.assertFalse(result["recovery_sanctioned"]) + self.assertEqual( + result["refusal_code"], bootstrap_lock_recovery.REFUSAL_BINDING_MISMATCH + ) + + def test_mismatched_org_is_refused(self): + self.assertFalse(self._assess(org="OtherOrg")["recovery_sanctioned"]) + + def test_mismatched_remote_is_refused(self): + self.assertFalse(self._assess(remote="dadeschools")["recovery_sanctioned"]) + + def test_mismatched_issue_is_refused(self): + self.assertFalse(self._assess(issue_number=ISSUE + 1)["recovery_sanctioned"]) + + def test_mismatched_branch_is_refused(self): + self.assertFalse( + self._assess(branch_name="fix/issue-9530-other")["recovery_sanctioned"] + ) + + def test_mismatched_worktree_is_refused(self): + self.assertFalse( + self._assess(worktree_path="/scratch/elsewhere")["recovery_sanctioned"] + ) + + def test_missing_worktree_is_refused(self): + result = self._assess(worktree_exists=False) + self.assertFalse(result["recovery_sanctioned"]) + self.assertEqual( + result["refusal_code"], bootstrap_lock_recovery.REFUSAL_WORKTREE_INVALID + ) + + def test_unregistered_worktree_is_refused(self): + self.assertFalse(self._assess(worktree_registered=False)["recovery_sanctioned"]) + + def test_worktree_on_a_different_branch_is_refused(self): + self.assertFalse(self._assess(current_branch="master")["recovery_sanctioned"]) + + def test_head_mismatch_is_refused(self): + result = self._assess(declared_head=OTHER_HEAD) + self.assertFalse(result["recovery_sanctioned"]) + self.assertEqual( + result["refusal_code"], bootstrap_lock_recovery.REFUSAL_HEAD_MISMATCH + ) + + def test_unresolvable_identity_is_refused(self): + self.assertFalse(self._assess(identity="")["recovery_sanctioned"]) + + def test_unresolvable_profile_is_refused(self): + self.assertFalse(self._assess(profile="")["recovery_sanctioned"]) + + def test_matching_username_alone_does_not_prove_ownership(self): + """Safety: a matching username with the wrong profile is still foreign.""" + self.assertFalse(self._assess(profile=FOREIGN_PROFILE)["recovery_sanctioned"]) + + def test_healthy_foreign_owned_lock_cannot_be_recovered(self): + """AC11: the foreign-takeover refusal.""" + foreign = canonical_lock(worktree="/scratch/wt-9530") + foreign["work_lease"]["claimant"] = { + "username": FOREIGN_IDENTITY, + "profile": FOREIGN_PROFILE, + } + foreign["session_pid"] = os.getpid() # alive → healthy + result = self._assess(lock=foreign) + self.assertFalse(result["recovery_sanctioned"]) + self.assertEqual( + result["refusal_code"], bootstrap_lock_recovery.REFUSAL_HEALTHY_FOREIGN + ) + + def test_foreign_owned_incomplete_lock_is_also_refused(self): + """A foreign lock is refused whether or not it is healthy.""" + foreign = bootstrap_shaped_lock( + worktree_path="/scratch/wt-9530", + claimant={"username": FOREIGN_IDENTITY, "profile": FOREIGN_PROFILE}, + ) + result = self._assess(lock=foreign) + self.assertFalse(result["recovery_sanctioned"]) + self.assertIn( + result["refusal_code"], + { + bootstrap_lock_recovery.REFUSAL_FOREIGN_CLAIMANT, + bootstrap_lock_recovery.REFUSAL_HEALTHY_FOREIGN, + }, + ) + + def test_healthy_same_owner_canonical_lock_is_left_alone(self): + """Nothing to recover: rewriting would invalidate a live heartbeat token.""" + result = self._assess(lock=canonical_lock(worktree="/scratch/wt-9530")) + self.assertFalse(result["recovery_sanctioned"]) + self.assertEqual( + result["refusal_code"], bootstrap_lock_recovery.REFUSAL_ALREADY_CANONICAL + ) + + def test_absent_lock_is_refused(self): + result = self._assess(lock={}) + self.assertFalse(result["recovery_sanctioned"]) + self.assertEqual(result["refusal_code"], bootstrap_lock_recovery.REFUSAL_NO_LOCK) + + def test_recovery_never_requires_base_equivalence(self): + """AC9: the branch carries commits; that must not be disqualifying.""" + import inspect as _inspect + + params = _inspect.signature( + bootstrap_lock_recovery.assess_bootstrap_lock_recovery + ).parameters + self.assertNotIn("base_equivalent", params) + self.assertNotIn("expected_base_sha", params) + + def test_recovery_accepts_no_caller_supplied_authorization(self): + """Safety: no caller-manufactured provenance or authorization.""" + import inspect as _inspect + + params = _inspect.signature( + bootstrap_lock_recovery.assess_bootstrap_lock_recovery + ).parameters + for forbidden in ( + "recovery_sanctioned", + "lock_provenance", + "provenance", + "operator_override", + "authorized", + ): + self.assertNotIn(forbidden, params) + + +class RecoveryAuditTrail(unittest.TestCase): + """AC10: auditable ownership and generation transition.""" + + def _assessment(self, **overrides): + lock = bootstrap_shaped_lock(worktree_path="/scratch/wt-9530", **overrides) + return bootstrap_lock_recovery.assess_bootstrap_lock_recovery( + lock, + issue_number=ISSUE, + branch_name=BRANCH, + worktree_path="/scratch/wt-9530", + remote=REMOTE, + org=ORG, + repo=REPO, + identity=IDENTITY, + profile=PROFILE, + observed_head=HEAD, + declared_head=HEAD, + worktree_exists=True, + worktree_registered=True, + current_branch=BRANCH, + ) + + def test_recovery_record_preserves_both_sides_of_the_transition(self): + record = bootstrap_lock_recovery.build_recovery_record( + self._assessment(), recovered_at=_ts(), new_task_session_id="task-new" + ) + self.assertEqual(record["recovery_kind"], "incomplete_bootstrap_lock") + self.assertEqual(record["prior_generation"], 1) + self.assertEqual( + record["prior_owner_session"], "author_issue_work-deadbeefdeadbeef" + ) + self.assertEqual(record["replacement_task_session_id"], "task-new") + self.assertEqual(record["preserved_head"], HEAD) + self.assertFalse(record["branch_reset"]) + self.assertFalse(record["base_equivalence_required"]) + self.assertIn("work_lease", record["prior_missing_fields"]) + + def test_expected_generation_is_reported_for_compare_and_swap(self): + self.assertEqual( + self._assessment(lock_generation=7)["expected_generation"], 7 + ) + + +class RecoveryIsolationAndPersistence(unittest.TestCase): + """AC8, AC16: recovery upgrades only its target and inspection mutates nothing.""" + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.lock_dir = self.tmp.name + + def _write(self, lock, **kwargs): + return issue_lock_store.bind_session_lock( + lock, lock_dir=self.lock_dir, **kwargs + ) + + def test_recovery_upgrades_the_target_lock_to_canonical(self): + path = self._write(bootstrap_shaped_lock(worktree_path="/scratch/wt-9530")) + before = issue_lock_store.read_lock_file(path) + self.assertFalse(author_lock_contract.assess_lock_contract(before)["canonical"]) + + self._write( + author_lock_contract.build_canonical_issue_lock( + issue_number=ISSUE, + branch_name=BRANCH, + worktree_path="/scratch/wt-9530", + remote=REMOTE, + org=ORG, + repo=REPO, + identity=IDENTITY, + profile=PROFILE, + tool="gitea_recover_incomplete_bootstrap_lock", + ) + ) + after = issue_lock_store.read_lock_file(path) + self.assertTrue(author_lock_contract.assess_lock_contract(after)["canonical"]) + self.assertGreater(after["lock_generation"], before["lock_generation"]) + + def test_recovery_does_not_touch_an_unrelated_lock(self): + """A failure or success must affect only the exact target lock.""" + other_issue = ISSUE + 77 + other_path = self._write( + bootstrap_shaped_lock( + issue_number=other_issue, + branch_name=f"fix/issue-{other_issue}-unrelated", + worktree_path="/scratch/wt-other", + ) + ) + other_before = issue_lock_store.read_lock_file(other_path) + + target_path = self._write( + bootstrap_shaped_lock(worktree_path="/scratch/wt-9530") + ) + self._write( + author_lock_contract.build_canonical_issue_lock( + issue_number=ISSUE, + branch_name=BRANCH, + worktree_path="/scratch/wt-9530", + remote=REMOTE, + org=ORG, + repo=REPO, + identity=IDENTITY, + profile=PROFILE, + tool="gitea_recover_incomplete_bootstrap_lock", + ) + ) + self.assertNotEqual(target_path, other_path) + self.assertEqual(issue_lock_store.read_lock_file(other_path), other_before) + + def test_inspection_performs_no_mutation(self): + """AC16: assessing a lock must not rewrite it.""" + path = self._write(bootstrap_shaped_lock(worktree_path="/scratch/wt-9530")) + before = issue_lock_store.read_lock_file(path) + mtime_before = os.path.getmtime(path) + + author_lock_contract.assess_lock_contract(before) + bootstrap_lock_recovery.assess_bootstrap_lock_recovery( + before, + issue_number=ISSUE, + branch_name=BRANCH, + worktree_path="/scratch/wt-9530", + remote=REMOTE, + org=ORG, + repo=REPO, + identity=IDENTITY, + profile=PROFILE, + observed_head=HEAD, + declared_head=HEAD, + worktree_exists=True, + worktree_registered=True, + current_branch=BRANCH, + ) + self.assertEqual(issue_lock_store.read_lock_file(path), before) + self.assertEqual(os.path.getmtime(path), mtime_before) + + +class HeartbeatOnFreshAndRecoveredLocks(unittest.TestCase): + """AC2, AC3: heartbeat and renewal against real durable locks.""" + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.lock_dir = self.tmp.name + self.worktree = os.path.join(self.tmp.name, "wt") + os.makedirs(self.worktree, exist_ok=True) + + def _canonical(self, tool="gitea_bootstrap_author_issue_worktree"): + return author_lock_contract.build_canonical_issue_lock( + issue_number=ISSUE, + branch_name=BRANCH, + worktree_path=self.worktree, + remote=REMOTE, + org=ORG, + repo=REPO, + identity=IDENTITY, + profile=PROFILE, + tool=tool, + ) + + def _heartbeat(self, token): + return issue_lock_store.heartbeat_session_lock( + remote=REMOTE, + org=ORG, + repo=REPO, + issue_number=ISSUE, + branch_name=BRANCH, + worktree_path=self.worktree, + identity=IDENTITY, + profile=PROFILE, + task_session_id=token, + lock_dir=self.lock_dir, + ) + + def test_a_freshly_built_canonical_lock_can_be_heartbeated_immediately(self): + """AC2: the property a bootstrap lock never had.""" + lock = self._canonical() + issue_lock_store.bind_session_lock(lock, lock_dir=self.lock_dir) + outcome = self._heartbeat(lock["work_lease"]["task_session_id"]) + self.assertTrue(outcome["success"], outcome.get("reasons")) + self.assertTrue(outcome["performed"]) + + def test_heartbeat_does_not_change_ownership_or_create_a_second_lock(self): + lock = self._canonical() + path = issue_lock_store.bind_session_lock(lock, lock_dir=self.lock_dir) + self._heartbeat(lock["work_lease"]["task_session_id"]) + after = issue_lock_store.read_lock_file(path) + self.assertEqual( + author_lock_contract.lock_claimant(after), + {"username": IDENTITY, "profile": PROFILE}, + ) + # Count issue locks only: bind_session_lock also writes a + # session-.json pointer, which is pre-existing behaviour and not a + # second claim on the issue. + locks = [ + f + for f in os.listdir(self.lock_dir) + if f.endswith(".json") and not f.startswith("session-") + ] + self.assertEqual(len(locks), 1, f"expected exactly one issue lock, got {locks}") + + def test_heartbeat_refuses_a_foreign_task_session_token(self): + lock = self._canonical(tool="gitea_lock_issue") + issue_lock_store.bind_session_lock(lock, lock_dir=self.lock_dir) + outcome = self._heartbeat("author_issue_work-someoneelse") + self.assertFalse(outcome["success"]) + + +class BootstrapToCreatePrRegression(unittest.TestCase): + """AC17, AC18: the exact #949 sequence, against isolated fixtures only. + + This reproduces the *shape* of the #949 failure — bootstrap, implement, + commit, push, create PR — in a throwaway git repository. It never reads or + writes the real issue #949 branch, worktree, lock, issue, or head. + """ + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.lock_dir = os.path.join(self.tmp.name, "locks") + os.makedirs(self.lock_dir, exist_ok=True) + self.origin = os.path.join(self.tmp.name, "origin.git") + self.repo = os.path.join(self.tmp.name, "repo") + subprocess.run( + ["git", "init", "--bare", self.origin], check=True, capture_output=True + ) + subprocess.run(["git", "init", self.repo], check=True, capture_output=True) + self._git("config", "user.email", "author@example.invalid") + self._git("config", "user.name", "Example Author") + with open(os.path.join(self.repo, "README.md"), "w") as handle: + handle.write("base\n") + self._git("add", "README.md") + self._git("commit", "-m", "base commit") + self._git("branch", "-M", "master") + self._git("remote", "add", "origin", self.origin) + self._git("push", "-u", "origin", "master") + + def _git(self, *args): + return subprocess.run( + ["git", "-C", self.repo, *args], check=True, capture_output=True, text=True + ) + + def _heartbeat(self, token): + return issue_lock_store.heartbeat_session_lock( + remote=REMOTE, + org=ORG, + repo=REPO, + issue_number=ISSUE, + branch_name=BRANCH, + worktree_path=self.repo, + identity=IDENTITY, + profile=PROFILE, + task_session_id=token, + lock_dir=self.lock_dir, + ) + + def _implement_and_commit(self): + with open(os.path.join(self.repo, "feature.py"), "w") as handle: + handle.write("VALUE = 1\n") + self._git("add", "feature.py") + self._git("commit", "-m", "feat: implement the issue") + + def test_bootstrap_implement_commit_push_create_pr_completes(self): + # 1. Bootstrap: branch, worktree, and a canonical lock. + self._git("checkout", "-b", BRANCH) + lock = author_lock_contract.build_canonical_issue_lock( + issue_number=ISSUE, + branch_name=BRANCH, + worktree_path=self.repo, + remote=REMOTE, + org=ORG, + repo=REPO, + identity=IDENTITY, + profile=PROFILE, + tool="gitea_bootstrap_author_issue_worktree", + source=author_lock_contract.SOURCE_BOOTSTRAP, + ) + path = issue_lock_store.bind_session_lock(lock, lock_dir=self.lock_dir) + token = lock["work_lease"]["task_session_id"] + + # The lock is usable the moment bootstrap returns. + self.assertTrue( + author_lock_contract.assess_lock_contract( + issue_lock_store.read_lock_file(path) + )["canonical"] + ) + + # 2. Renewal is available *before* the branch diverges (AC3). + before = self._heartbeat(token) + self.assertTrue(before["success"], before.get("reasons")) + + # 3. Implement and commit — the branch now carries real work. + self._implement_and_commit() + head = self._git("rev-parse", "HEAD").stdout.strip() + + # 4. Push. + self._git("push", "-u", "origin", BRANCH) + remote_head = subprocess.run( + ["git", "-C", self.origin, "rev-parse", f"refs/heads/{BRANCH}"], + capture_output=True, + text=True, + check=True, + ).stdout.strip() + self.assertEqual(head, remote_head) + + # 5. Renewal still available *after* commits (AC3) — no base-equivalence. + after = self._heartbeat(token) + self.assertTrue(after["success"], after.get("reasons")) + + # 6. create_pr's #447 provenance guard accepts the lock (AC4). + final = issue_lock_store.read_lock_file(path) + verdict = issue_lock_provenance.assess_lock_file_for_create_pr(final) + self.assertTrue(verdict["proven"], verdict["reasons"]) + + # AC18: completed with no manual lock edit, no branch rewind, no + # fallback transport. The base commit is still an ancestor of head. + merge_base = self._git("merge-base", "master", BRANCH).stdout.strip() + master_head = self._git("rev-parse", "master").stdout.strip() + self.assertEqual(merge_base, master_head) + + def test_the_old_bootstrap_shape_reproduces_the_949_dead_end(self): + """The regression must actually fail without the fix.""" + self._git("checkout", "-b", BRANCH) + self._implement_and_commit() + + legacy = bootstrap_shaped_lock(worktree_path=self.repo) + issue_lock_store.bind_session_lock(legacy, lock_dir=self.lock_dir) + + # create_pr refuses — the #447 guard, unchanged. + self.assertFalse( + issue_lock_provenance.assess_lock_file_for_create_pr(legacy)["proven"] + ) + # And it is never classified as expired, so renewal never engages. + self.assertFalse(issue_lock_store.is_lease_expired(legacy)) + self.assertEqual( + author_lock_contract.expiration_state(legacy)["state"], + author_lock_contract.EXPIRATION_MISSING, + ) + + def test_recovery_of_a_committed_branch_preserves_the_commits(self): + """AC9: recovery must not rewind a branch that carries pushed work.""" + self._git("checkout", "-b", BRANCH) + self._implement_and_commit() + self._git("push", "-u", "origin", BRANCH) + head_before = self._git("rev-parse", "HEAD").stdout.strip() + + legacy = bootstrap_shaped_lock(worktree_path=self.repo) + path = issue_lock_store.bind_session_lock(legacy, lock_dir=self.lock_dir) + + assessment = bootstrap_lock_recovery.assess_bootstrap_lock_recovery( + issue_lock_store.read_lock_file(path), + issue_number=ISSUE, + branch_name=BRANCH, + worktree_path=self.repo, + remote=REMOTE, + org=ORG, + repo=REPO, + identity=IDENTITY, + profile=PROFILE, + observed_head=head_before, + declared_head=head_before, + worktree_exists=True, + worktree_registered=True, + current_branch=BRANCH, + ) + self.assertTrue(assessment["recovery_sanctioned"], assessment["reasons"]) + + recovered = author_lock_contract.build_canonical_issue_lock( + issue_number=ISSUE, + branch_name=BRANCH, + worktree_path=self.repo, + remote=REMOTE, + org=ORG, + repo=REPO, + identity=IDENTITY, + profile=PROFILE, + tool="gitea_recover_incomplete_bootstrap_lock", + ) + recovered["bootstrap_lock_recovery"] = ( + bootstrap_lock_recovery.build_recovery_record( + assessment, + recovered_at=_ts(), + new_task_session_id=recovered["work_lease"]["task_session_id"], + ) + ) + issue_lock_store.bind_session_lock( + recovered, + lock_dir=self.lock_dir, + expected_generation=assessment["expected_generation"], + recovery_sanctioned=True, + ) + + # The branch head is untouched and the lock is now canonical. + self.assertEqual(self._git("rev-parse", "HEAD").stdout.strip(), head_before) + final = issue_lock_store.read_lock_file(path) + self.assertTrue(author_lock_contract.assess_lock_contract(final)["canonical"]) + self.assertTrue( + issue_lock_provenance.assess_lock_file_for_create_pr(final)["proven"] + ) + self.assertEqual( + final["bootstrap_lock_recovery"]["preserved_head"], head_before + ) + self.assertFalse(final["bootstrap_lock_recovery"]["branch_reset"]) + + +class BootstrapWiring(unittest.TestCase): + """AC1, AC5, AC7, AC15: what the bootstrap tool itself now does.""" + + def _source(self): + import author_issue_bootstrap + + with open(author_issue_bootstrap.__file__) as handle: + return handle.read() + + def test_bootstrap_builds_through_the_canonical_contract(self): + source = self._source() + self.assertIn("author_lock_contract.build_canonical_issue_lock", source) + self.assertIn("author_lock_contract.assess_lock_contract", source) + + def test_bootstrap_fails_closed_on_an_incomplete_written_lock(self): + """AC7: partial lock creation stops before implementation begins.""" + source = self._source() + self.assertIn("incomplete_issue_lock_contract", source) + self.assertIn('"implementation_allowed": False', source) + + def test_next_action_for_a_canonical_lock_directs_to_implementation(self): + """AC5: executable under the state actually returned.""" + assessment = author_lock_contract.assess_lock_contract(canonical_lock()) + self.assertIn("work_issue", author_lock_contract.recommended_action(assessment)) + + def test_next_action_for_an_incomplete_lock_forbids_implementation(self): + """AC5, AC15: never direct an author into the unrecoverable state.""" + assessment = author_lock_contract.assess_lock_contract(bootstrap_shaped_lock()) + action = author_lock_contract.recommended_action(assessment) + self.assertIn("Do not begin implementation", action) + self.assertIn("gitea_recover_incomplete_bootstrap_lock", action) + + +class _NativeToolBase(unittest.TestCase): + """A real git worktree, a real durable lock, and the real registered tools. + + Review 632 F4: every previous reference to the two new tools under + ``tests/`` was a string literal, and the suite reconstructed the recovery + sequence by hand. A hand-written sequence cannot see a divergence between + itself and the tool body — which is exactly how F1 and F2, both call-site + defects, survived 61 passing cases. These call the registered functions. + """ + + TOOL_ISSUE = 9531 + TOOL_BRANCH = "fix/issue-9531-native-tool-path" + TOOL_ORG = "Example-Org" + TOOL_REPO = "Example-Repo" + AUTHOR_PROFILE = "test-author-prgs" + REVIEWER_PROFILE = "test-reviewer-prgs" + MERGER_PROFILE = "test-merger-prgs" + TOOL_IDENTITY = "example-native-user" + + def setUp(self): + self.lock_dir = tempfile.TemporaryDirectory() + self.addCleanup(self.lock_dir.cleanup) + self.repo = tempfile.mkdtemp(prefix="issue953-native-") + self.addCleanup( + lambda: subprocess.run(["rm", "-rf", self.repo], check=False) + ) + self._init_repo() + self.remotes = mock.patch.dict( + mcp_server.REMOTES, + { + "prgs": { + "host": "gitea.prgs.cc", + "org": self.TOOL_ORG, + "repo": self.TOOL_REPO, + } + }, + ) + self.remotes.start() + self.addCleanup(mock.patch.stopall) + mcp_server._IDENTITY_CACHE.clear() + # The sticky reviewer-stop route is process-global and outlives whatever + # test set it. These cases assert on the *namespace* gate, so the gate + # ahead of it must start clean or it refuses first for another reason. + role_session_router.clear_route_state() + self.addCleanup(role_session_router.clear_route_state) + + def _git(self, *args): + return subprocess.run( + ["git", "-C", self.repo, *args], + capture_output=True, + text=True, + check=True, + ) + + def _init_repo(self): + self._git("init", "-q", "-b", "master") + self._git("config", "user.email", "test@example.com") + self._git("config", "user.name", "Test") + with open(os.path.join(self.repo, "seed.txt"), "w") as handle: + handle.write("seed\n") + self._git("add", "seed.txt") + self._git("commit", "-q", "-m", "seed") + self._git("checkout", "-q", "-b", self.TOOL_BRANCH) + # The state recovery exists for: the branch already carries pushed work. + with open(os.path.join(self.repo, "impl.txt"), "w") as handle: + handle.write("implementation\n") + self._git("add", "impl.txt") + self._git("commit", "-q", "-m", "implementation") + self.head = self._git("rev-parse", "HEAD").stdout.strip() + self.worktree = os.path.realpath(self.repo) + + def _lock_path(self): + return issue_lock_store.lock_file_path( + remote=REMOTE, + org=self.TOOL_ORG, + repo=self.TOOL_REPO, + issue_number=self.TOOL_ISSUE, + lock_dir=self.lock_dir.name, + ) + + def write_incomplete_lock(self): + """The exact malformed shape the old bootstrap left behind.""" + lock = bootstrap_shaped_lock( + issue_number=self.TOOL_ISSUE, + branch=self.TOOL_BRANCH, + branch_name=self.TOOL_BRANCH, + worktree_path=self.worktree, + org=self.TOOL_ORG, + repo=self.TOOL_REPO, + claimant={ + "username": self.TOOL_IDENTITY, + "profile": self.AUTHOR_PROFILE, + }, + ) + path = self._lock_path() + lock["lock_file_path"] = path + issue_lock_store.save_lock_file(path, lock) + return path + + def _env(self, profile_name): + env = shared_mutation_env( + profile_name, + include_example_repo=True, + GITEA_ISSUE_LOCK_DIR=self.lock_dir.name, + ) + env["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name + return env + + def call_recovery( + self, + *, + profile_name=None, + identity=None, + claimant_profile=None, + expected_head=None, + audit_sink=None, + namespace_patch=None, + **kwargs, + ): + """Invoke the registered recovery tool itself, not its assessors.""" + profile_name = profile_name or self.AUTHOR_PROFILE + env = self._env(profile_name) + patches = [ + mock.patch( + "mcp_server._work_lease_claimant", + return_value={ + "username": identity or self.TOOL_IDENTITY, + "profile": claimant_profile or profile_name, + }, + ), + mock.patch("mcp_server.get_auth_header", return_value="token x"), + mock.patch( + "mcp_server._canonical_local_git_root", return_value=self.worktree + ), + mock.patch.dict(os.environ, env, clear=True), + ] + if audit_sink is not None: + patches.append( + mock.patch( + "mcp_server._audit", + side_effect=lambda *a, **kw: audit_sink.append((a, kw)), + ) + ) + if namespace_patch is not None: + patches.append(namespace_patch) + with contextlib.ExitStack() as stack: + for patch in patches: + stack.enter_context(patch) + os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name + return mcp_server.gitea_recover_incomplete_bootstrap_lock( + issue_number=kwargs.pop("issue_number", self.TOOL_ISSUE), + branch_name=kwargs.pop("branch_name", self.TOOL_BRANCH), + worktree_path=kwargs.pop("worktree_path", self.worktree), + expected_head=expected_head or self.head, + remote="prgs", + **kwargs, + ) + + def call_inspection(self, *, profile_name=None, **kwargs): + """Invoke the registered read-only inspection tool itself.""" + env = self._env(profile_name or self.AUTHOR_PROFILE) + with mock.patch( + "mcp_server._work_lease_claimant", + return_value={ + "username": self.TOOL_IDENTITY, + "profile": profile_name or self.AUTHOR_PROFILE, + }, + ), mock.patch( + "mcp_server.get_auth_header", return_value="token x" + ), mock.patch( + "mcp_server._canonical_local_git_root", return_value=self.worktree + ), mock.patch.dict( + os.environ, env, clear=True + ): + os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name + return mcp_server.gitea_inspect_issue_lock_contract( + issue_number=kwargs.pop("issue_number", self.TOOL_ISSUE), + remote="prgs", + **kwargs, + ) + + +class NamespaceMutationWallOnRecovery(_NativeToolBase): + """Review 632 F1: the namespace/session wall on the new author mutation. + + ``gitea_recover_incomplete_bootstrap_lock`` writes the same durable lock as + ``gitea_recover_dirty_orphaned_issue_worktree`` and must carry the same + third gate. Exact-owner claimant comparison inside + ``assess_bootstrap_lock_recovery`` is a later layer, not a substitute: it + refuses without a namespace evaluation and without a BLOCKED audit record. + """ + + def test_the_recovery_tool_calls_the_namespace_mutation_gate(self): + """The gate must be reached, with this task, before any write.""" + self.write_incomplete_lock() + seen = [] + + def _spy(task, **kwargs): + seen.append((task, kwargs)) + return None + + self.call_recovery( + namespace_patch=mock.patch( + "mcp_server._namespace_mutation_block", side_effect=_spy + ) + ) + + self.assertEqual(len(seen), 1, seen) + task, kwargs = seen[0] + self.assertEqual(task, "recover_incomplete_bootstrap_lock") + self.assertTrue(kwargs.get("author_role_exclusive")) + + def test_the_gate_return_value_is_consumed_and_returned(self): + """A gate refusal must abort the tool, not be computed and discarded.""" + path = self.write_incomplete_lock() + before = issue_lock_store.read_lock_file(path) + refusal = {"success": False, "performed": False, "namespace_block": True} + + result = self.call_recovery( + namespace_patch=mock.patch( + "mcp_server._namespace_mutation_block", return_value=refusal + ) + ) + + self.assertIs(result, refusal) + self.assertEqual(issue_lock_store.read_lock_file(path), before) + + def test_correct_author_namespace_succeeds(self): + path = self.write_incomplete_lock() + result = self.call_recovery() + self.assertTrue(result.get("success"), result) + self.assertTrue(result.get("performed"), result) + written = issue_lock_store.read_lock_file(path) + self.assertTrue( + author_lock_contract.assess_lock_contract(written)["canonical"], written + ) + + def test_reviewer_namespace_is_rejected_with_matching_claimant_data(self): + """Identity that would satisfy the owner check must not be a way in.""" + path = self.write_incomplete_lock() + before = issue_lock_store.read_lock_file(path) + + result = self.call_recovery( + profile_name=self.REVIEWER_PROFILE, + claimant_profile=self.AUTHOR_PROFILE, + ) + + self.assertFalse(result.get("success"), result) + self.assertFalse(result.get("performed"), result) + self.assertTrue(result.get("namespace_block"), result) + self.assertEqual(result.get("mcp_namespace"), "gitea-reviewer") + # Not the later exact-owner refusal — the namespace layer stopped it. + self.assertNotEqual(result.get("refusal_code"), "foreign_claimant") + self.assertEqual(issue_lock_store.read_lock_file(path), before) + + def test_reviewer_rejection_emits_the_standard_blocked_audit(self): + self.write_incomplete_lock() + audit = [] + self.call_recovery(profile_name=self.REVIEWER_PROFILE, audit_sink=audit) + + blocked = [ + (args, kwargs) + for args, kwargs in audit + if kwargs.get("result") == gitea_audit.BLOCKED + ] + self.assertTrue(blocked, audit) + args, kwargs = blocked[0] + self.assertEqual(args[0], "recover_incomplete_bootstrap_lock") + self.assertEqual( + kwargs.get("mutation_task"), "recover_incomplete_bootstrap_lock" + ) + + def test_merger_profile_is_rejected(self): + """gitea.issue.comment is held by every role; the wall cannot rely on it.""" + path = self.write_incomplete_lock() + before = issue_lock_store.read_lock_file(path) + result = self.call_recovery(profile_name=self.MERGER_PROFILE) + self.assertFalse(result.get("success"), result) + self.assertFalse(result.get("performed"), result) + # Specifically the namespace/role wall, not some later refusal. + self.assertTrue(result.get("namespace_block"), result) + self.assertTrue( + any( + "recover_incomplete_bootstrap_lock' blocked" in reason + for reason in result.get("reasons") or [] + ), + result, + ) + self.assertIsNone(result.get("refusal_code"), result) + self.assertEqual(issue_lock_store.read_lock_file(path), before) + + def test_wrong_profile_for_the_claimant_is_rejected(self): + """A matching username under a different profile is still foreign.""" + path = self.write_incomplete_lock() + before = issue_lock_store.read_lock_file(path) + result = self.call_recovery(claimant_profile="test-author-dadeschools") + self.assertFalse(result.get("success"), result) + self.assertFalse(result.get("performed"), result) + # The namespace wall passes here (the session *is* author-bound); this + # must be the exact-owner layer refusing the mismatched profile. + self.assertIsNone(result.get("namespace_block"), result) + self.assertIn( + result.get("refusal_code"), ("foreign_claimant", "healthy_foreign_lock"), result + ) + self.assertEqual(issue_lock_store.read_lock_file(path), before) + + def test_mismatched_head_is_rejected_without_mutation(self): + path = self.write_incomplete_lock() + before = issue_lock_store.read_lock_file(path) + result = self.call_recovery(expected_head=OTHER_HEAD) + self.assertFalse(result.get("success"), result) + self.assertFalse(result.get("mutation_performed"), result) + self.assertEqual(issue_lock_store.read_lock_file(path), before) + + def test_rejection_leaves_branch_worktree_and_unrelated_locks_untouched(self): + unrelated = issue_lock_store.lock_file_path( + remote=REMOTE, + org=self.TOOL_ORG, + repo=self.TOOL_REPO, + issue_number=self.TOOL_ISSUE + 41, + lock_dir=self.lock_dir.name, + ) + issue_lock_store.save_lock_file( + unrelated, bootstrap_shaped_lock(issue_number=self.TOOL_ISSUE + 41) + ) + unrelated_before = issue_lock_store.read_lock_file(unrelated) + + path = self.write_incomplete_lock() + before = issue_lock_store.read_lock_file(path) + head_before = self._git("rev-parse", "HEAD").stdout.strip() + + self.call_recovery(profile_name=self.REVIEWER_PROFILE) + + self.assertEqual(issue_lock_store.read_lock_file(path), before) + self.assertEqual(issue_lock_store.read_lock_file(unrelated), unrelated_before) + self.assertEqual(self._git("rev-parse", "HEAD").stdout.strip(), head_before) + self.assertTrue(os.path.isdir(self.worktree)) + self.assertEqual(self._git("status", "--porcelain").stdout.strip(), "") + + def test_the_gate_routes_this_task_as_author_required(self): + """Without a router entry the namespace check silently allows everything.""" + self.assertEqual( + role_session_router.required_role_for_task( + "recover_incomplete_bootstrap_lock" + ), + "author", + ) + ok, reasons = role_namespace_gate.check_author_mutation_namespace( + "recover_incomplete_bootstrap_lock", + { + "profile_name": "prgs-reviewer", + "allowed_operations": ["gitea.read", "gitea.pr.approve"], + "forbidden_operations": [], + }, + ) + self.assertFalse(ok, reasons) + + def test_role_kind_wall_admits_author_and_refuses_every_other_role(self): + cases = { + "author": (["gitea.pr.create", "gitea.branch.push"], True), + "reviewer": (["gitea.pr.approve"], False), + "merger": (["gitea.pr.merge"], False), + "limited": (["gitea.issue.comment", "gitea.read"], False), + "mixed": (["gitea.pr.approve", "gitea.pr.create"], False), + } + for label, (ops, expected) in cases.items(): + with self.subTest(role=label): + ok, _ = role_namespace_gate.check_author_role_kind( + "recover_incomplete_bootstrap_lock", + { + "profile_name": f"prgs-{label}", + "allowed_operations": ops, + "forbidden_operations": [], + }, + ) + self.assertEqual(ok, expected) + + +class InspectionToolExecutes(_NativeToolBase): + """AC16 proved against the registered tool, not only its assessors.""" + + def test_the_registered_inspection_tool_reports_the_contract(self): + self.write_incomplete_lock() + result = self.call_inspection() + self.assertTrue(result.get("success"), result) + self.assertTrue(result.get("read_only")) + self.assertTrue(result.get("lock_present")) + self.assertFalse(result["lock_contract"]["canonical"]) + + def test_the_registered_inspection_tool_mutates_nothing(self): + path = self.write_incomplete_lock() + before = issue_lock_store.read_lock_file(path) + mtime_before = os.path.getmtime(path) + head_before = self._git("rev-parse", "HEAD").stdout.strip() + + result = self.call_inspection( + branch_name=self.TOOL_BRANCH, worktree_path=self.worktree + ) + + self.assertFalse(result.get("mutation_performed")) + self.assertFalse(result.get("performed")) + self.assertIn("recovery_preview", result) + self.assertEqual(issue_lock_store.read_lock_file(path), before) + self.assertEqual(os.path.getmtime(path), mtime_before) + self.assertEqual(self._git("rev-parse", "HEAD").stdout.strip(), head_before) + self.assertEqual(self._git("status", "--porcelain").stdout.strip(), "") + + def test_inspection_reports_an_absent_lock_without_creating_one(self): + result = self.call_inspection() + self.assertTrue(result.get("success"), result) + self.assertFalse(result.get("lock_present")) + self.assertFalse(os.path.exists(self._lock_path())) + + +class Ac7PostCompensationGuidance(unittest.TestCase): + """Review 632 F2: the returned action must fit the post-rollback state. + + The AC7 refusal runs ``run_compensating_recovery`` first, which releases the + lock and removes the branch and worktree. Recommending incomplete-lock + recovery for those exact artifacts hands the author ``no_durable_lock`` and + then ``worktree_invalid`` — the unexecutable-guidance failure class #953 + exists to remove, reintroduced on the new fail-closed path. + """ + + ISSUE_NUMBER = 9532 + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.repo = os.path.join(self.tmp.name, "repo") + os.makedirs(self.repo) + self._git("init", "-q", "-b", "master") + self._git("config", "user.email", "test@example.com") + self._git("config", "user.name", "Test") + with open(os.path.join(self.repo, "README.md"), "w") as handle: + handle.write("# seed\n") + self._git("add", "README.md") + self._git("commit", "-q", "-m", "seed") + self.master_sha = self._git("rev-parse", "HEAD").stdout.strip() + os.makedirs(os.path.join(self.repo, "branches"), exist_ok=True) + self.lock_dir = os.path.join(self.tmp.name, "locks") + os.makedirs(self.lock_dir, exist_ok=True) + self.journal_dir = os.path.join(self.tmp.name, "journals") + os.makedirs(self.journal_dir, exist_ok=True) + self._journal_env = mock.patch.dict( + os.environ, {"GITEA_BOOTSTRAP_JOURNAL_DIR": self.journal_dir} + ) + self._journal_env.start() + self.addCleanup(self._journal_env.stop) + + def _git(self, *args): + return subprocess.run( + ["git", "-C", self.repo, *args], + capture_output=True, + text=True, + check=True, + ) + + def _branch_names(self): + out = subprocess.run( + ["git", "-C", self.repo, "branch", "--format=%(refname:short)"], + capture_output=True, + text=True, + check=False, + ) + return [line.strip() for line in out.stdout.splitlines() if line.strip()] + + def _lock_files(self): + """Durable issue locks only — not phase journals or session pointers.""" + found = [] + for path in issue_lock_store.iter_lock_files(self.lock_dir): + record = issue_lock_store.read_lock_file(path) or {} + if "lock_generation" in record: + found.append(os.path.basename(path)) + return sorted(found) + + def _run_bootstrap(self, *, key, partial=False, extra_patches=()): + """Drive the real bootstrap; optionally force a partial written lock.""" + with contextlib.ExitStack() as stack: + if partial: + real_build = author_lock_contract.build_canonical_issue_lock + + def _partial(**kwargs): + record = real_build(**kwargs) + # The #949 shape: claimant hoisted to the top level, no + # work_lease, no provenance, no expiry. + return { + "remote": record["remote"], + "org": record["org"], + "repo": record["repo"], + "issue_number": record["issue_number"], + "branch": record["branch"], + "branch_name": record["branch_name"], + "worktree_path": record["worktree_path"], + "claimant": dict(record["work_lease"]["claimant"]), + "lease_id": None, + "owner_session": record.get("owner_session"), + } + + stack.enter_context( + mock.patch.object( + author_issue_bootstrap.author_lock_contract, + "build_canonical_issue_lock", + side_effect=_partial, + ) + ) + for patch in extra_patches: + stack.enter_context(patch) + return author_issue_bootstrap.bootstrap_author_issue_worktree( + issue_number=self.ISSUE_NUMBER, + canonical_repo_root=self.repo, + expected_base_sha=self.master_sha, + idempotency_key=key, + remote="prgs", + lock_dir=self.lock_dir, + owner_session=f"session-{key}", + active_identity=IDENTITY, + active_profile=PROFILE, + ) + + def test_ac7_failure_then_complete_compensation_directs_to_retry_bootstrap(self): + result = self._run_bootstrap(key="ac7-complete", partial=True) + + self.assertFalse(result.get("success"), result) + self.assertEqual(result.get("reason_code"), "incomplete_issue_lock_contract") + self.assertFalse(result.get("implementation_allowed")) + state = result["post_compensation_state"] + self.assertEqual(state["cleanup_state"], author_lock_contract.CLEANUP_COMPLETE) + self.assertEqual(state["surviving_artifacts"], []) + + action = result["exact_next_action"] + self.assertIn("gitea_bootstrap_author_issue_worktree", action) + self.assertIn("Do not call gitea_recover_incomplete_bootstrap_lock", action) + + def test_the_returned_retry_action_is_executable(self): + """Follow the advice literally and it must succeed.""" + first = self._run_bootstrap(key="ac7-retry-1", partial=True) + self.assertIn( + "gitea_bootstrap_author_issue_worktree", first["exact_next_action"] + ) + + second = self._run_bootstrap(key="ac7-retry-2") + self.assertTrue(second.get("success"), second) + self.assertTrue(second.get("implementation_allowed")) + self.assertTrue(second["lock_contract"]["canonical"], second["lock_contract"]) + self.assertTrue(second.get("task_session_id")) + + def test_retry_after_complete_compensation_leaves_exactly_one_lock(self): + self._run_bootstrap(key="ac7-single-1", partial=True) + self.assertEqual(self._lock_files(), []) + + second = self._run_bootstrap(key="ac7-single-2") + self.assertTrue(second.get("success"), second) + locks = self._lock_files() + self.assertEqual(len(locks), 1, locks) + written = issue_lock_store.read_lock_file(second["lock_state"]) + self.assertTrue(author_lock_contract.assess_lock_contract(written)["canonical"]) + branches = [b for b in self._branch_names() if b != "master"] + self.assertEqual(len(branches), 1, branches) + + def test_no_recommendation_names_an_artifact_the_rollback_deleted(self): + result = self._run_bootstrap(key="ac7-no-ghosts", partial=True) + action = result["exact_next_action"] + state = result["post_compensation_state"] + + self.assertFalse(state["branch_present"]) + self.assertFalse(state["worktree_present"]) + self.assertFalse(state["lock_present"]) + self.assertNotIn(result["worktree_path"], action) + self.assertNotIn(f"branch '{result['branch_name']}'", action) + self.assertFalse(os.path.isdir(result["worktree_path"])) + self.assertNotIn(result["branch_name"], self._branch_names()) + + def test_partial_compensation_with_a_surviving_lock_is_not_reported_complete(self): + def _boom(**kwargs): + raise RuntimeError("lock release failed") + + outcome = self._run_bootstrap( + key="ac7-lock-survives", + partial=True, + extra_patches=[ + mock.patch.object( + author_issue_bootstrap.issue_lock_store, + "release_session_lock", + side_effect=_boom, + ) + ], + ) + + state = outcome["post_compensation_state"] + self.assertTrue(state["lock_present"], state) + self.assertEqual(state["cleanup_state"], author_lock_contract.CLEANUP_PARTIAL) + self.assertIn("lock", state["surviving_artifacts"]) + self.assertTrue(state["failed_rollback_steps"], state) + action = outcome["exact_next_action"] + self.assertIn("failed step", action) + self.assertIn("gitea_inspect_issue_lock_contract", action) + # The advice must not send the author at artifacts the rollback removed. + self.assertNotIn("re-run gitea_bootstrap_author_issue_worktree", action) + + def test_partial_compensation_with_surviving_branch_and_worktree(self): + """A worktree dirty at rollback time is preserved, and so is its branch.""" + real_assess = author_lock_contract.assess_lock_contract + + def _dirty_then_report(lock): + verdict = real_assess(lock) + path = (lock or {}).get("worktree_path") + if path and os.path.isdir(path): + with open(os.path.join(path, "uncommitted.txt"), "w") as handle: + handle.write("author bytes\n") + return verdict + + outcome = self._run_bootstrap( + key="ac7-wt-survives", + partial=True, + extra_patches=[ + mock.patch.object( + author_issue_bootstrap.author_lock_contract, + "assess_lock_contract", + side_effect=_dirty_then_report, + ) + ], + ) + + state = outcome["post_compensation_state"] + self.assertEqual(state["cleanup_state"], author_lock_contract.CLEANUP_PARTIAL) + self.assertTrue(state["worktree_present"], state) + self.assertTrue(state["branch_present"], state) + self.assertTrue(os.path.isdir(outcome["worktree_path"])) + self.assertIn(outcome["branch_name"], self._branch_names()) + self.assertIn("gitea_lock_issue", outcome["exact_next_action"]) + self.assertIn(outcome["branch_name"], outcome["exact_next_action"]) + + def test_compensation_failure_is_distinguished_from_partial_cleanup(self): + outcome = self._run_bootstrap( + key="ac7-comp-failed", + partial=True, + extra_patches=[ + mock.patch.object( + author_issue_bootstrap, + "run_compensating_recovery", + return_value={ + "executed": False, + "rolled_back": [], + "reason": "boom", + }, + ) + ], + ) + + state = outcome["post_compensation_state"] + self.assertEqual(state["cleanup_state"], author_lock_contract.CLEANUP_FAILED) + action = outcome["exact_next_action"] + self.assertIn("did not complete", action) + self.assertIn("gitea_inspect_issue_lock_contract", action) + self.assertNotIn("re-run gitea_bootstrap_author_issue_worktree", action) + + def test_a_failed_rollback_step_is_recorded_even_when_nothing_survives(self): + """A step that errored is still reported, and cleanup is still complete.""" + state = author_lock_contract.assess_post_compensation_state( + { + "executed": True, + "rolled_back": ["lease_release_failed:lease-1:RuntimeError"], + }, + lock_present=False, + worktree_present=False, + branch_present=False, + ) + self.assertEqual(state["cleanup_state"], author_lock_contract.CLEANUP_COMPLETE) + self.assertEqual( + state["failed_rollback_steps"], + ["lease_release_failed:lease-1:RuntimeError"], + ) + + def test_the_compensation_lock_release_actually_removes_the_lock(self): + """The rollback's lock half was dead code before #953 review 632 F2.""" + self.assertTrue(hasattr(issue_lock_store, "release_session_lock")) + result = self._run_bootstrap(key="ac7-release-real", partial=True) + self.assertIn( + f"lock:issue-{self.ISSUE_NUMBER}", + result["compensating_recovery"]["rolled_back"], + ) + self.assertEqual(self._lock_files(), []) + + def test_release_refuses_a_lock_owned_by_a_different_session(self): + created = self._run_bootstrap(key="ac7-foreign-release") + self.assertTrue(created.get("success"), created) + with self.assertRaises(FileNotFoundError): + issue_lock_store.release_session_lock( + issue_number=self.ISSUE_NUMBER, + session="session-somebody-else", + lock_dir=self.lock_dir, + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + ) + self.assertEqual(len(self._lock_files()), 1, self._lock_files()) + + def test_ac7_refusal_never_reports_implementation_ready(self): + result = self._run_bootstrap(key="ac7-never-ready", partial=True) + self.assertFalse(result.get("success")) + self.assertFalse(result.get("implementation_allowed")) + self.assertNotIn( + "proceed with author implementation", result["exact_next_action"] + ) + + def test_ac7_refusal_touches_no_unrelated_lock(self): + unrelated_path = issue_lock_store.lock_file_path( + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + issue_number=self.ISSUE_NUMBER + 63, + lock_dir=self.lock_dir, + ) + issue_lock_store.save_lock_file( + unrelated_path, bootstrap_shaped_lock(issue_number=self.ISSUE_NUMBER + 63) + ) + before = issue_lock_store.read_lock_file(unrelated_path) + mtime_before = os.path.getmtime(unrelated_path) + + self._run_bootstrap(key="ac7-isolation", partial=True) + + self.assertEqual(issue_lock_store.read_lock_file(unrelated_path), before) + self.assertEqual(os.path.getmtime(unrelated_path), mtime_before) + + def test_surviving_lock_and_worktree_direct_to_target_specific_recovery(self): + """The one state in which incomplete-lock recovery *is* executable.""" + state = author_lock_contract.assess_post_compensation_state( + {"executed": True, "rolled_back": []}, + lock_present=True, + worktree_present=True, + branch_present=True, + ) + action = author_lock_contract.post_compensation_action( + state, + issue_number=self.ISSUE_NUMBER, + branch_name=BRANCH, + worktree_path="/scratch/wt", + missing_fields=["work_lease"], + ) + self.assertEqual(state["cleanup_state"], author_lock_contract.CLEANUP_PARTIAL) + self.assertIn("gitea_recover_incomplete_bootstrap_lock", action) + self.assertIn(BRANCH, action) + self.assertIn("/scratch/wt", action) + + +class NativeEndToEndBootstrapToCreatePr(unittest.TestCase): + """AC17/AC18 driven through the real tools, with no gitea_lock_issue repair. + + bootstrap → inspect → heartbeat/renew → legitimate divergence → downstream + validation → create_pr provenance. The lock the real bootstrap writes must + carry the whole sequence on its own; repairing it with the older + ``gitea_lock_issue`` path would prove nothing about the new contract, so + that path is patched to fail the test if anything reaches for it. + """ + + ISSUE_NUMBER = 9533 + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.origin = os.path.join(self.tmp.name, "origin.git") + self.repo = os.path.join(self.tmp.name, "repo") + subprocess.run( + ["git", "init", "-q", "--bare", self.origin], check=True, capture_output=True + ) + subprocess.run( + ["git", "init", "-q", "-b", "master", self.repo], + check=True, + capture_output=True, + ) + self._git("config", "user.email", "author@example.invalid") + self._git("config", "user.name", "Example Author") + with open(os.path.join(self.repo, "README.md"), "w") as handle: + handle.write("base\n") + self._git("add", "README.md") + self._git("commit", "-q", "-m", "base commit") + self._git("remote", "add", "origin", self.origin) + self._git("push", "-q", "-u", "origin", "master") + self.master_sha = self._git("rev-parse", "HEAD").stdout.strip() + + self.lock_dir = os.path.join(self.tmp.name, "locks") + os.makedirs(self.lock_dir, exist_ok=True) + self.journal_dir = os.path.join(self.tmp.name, "journals") + os.makedirs(self.journal_dir, exist_ok=True) + self._journal_env = mock.patch.dict( + os.environ, {"GITEA_BOOTSTRAP_JOURNAL_DIR": self.journal_dir} + ) + self._journal_env.start() + self.addCleanup(self._journal_env.stop) + self.remotes = mock.patch.dict( + mcp_server.REMOTES, + { + "prgs": { + "host": "gitea.prgs.cc", + "org": "Scaled-Tech-Consulting", + "repo": "Gitea-Tools", + } + }, + ) + self.remotes.start() + self.addCleanup(mock.patch.stopall) + mcp_server._IDENTITY_CACHE.clear() + + def _git(self, *args, cwd=None): + return subprocess.run( + ["git", "-C", cwd or self.repo, *args], + check=True, + capture_output=True, + text=True, + ) + + def _inspect(self, worktree, branch): + env = shared_mutation_env( + "test-author-prgs", GITEA_ISSUE_LOCK_DIR=self.lock_dir + ) + env["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir + with mock.patch( + "mcp_server._work_lease_claimant", + return_value={"username": IDENTITY, "profile": PROFILE}, + ), mock.patch( + "mcp_server.get_auth_header", return_value="token x" + ), mock.patch( + "mcp_server._canonical_local_git_root", return_value=self.repo + ), mock.patch.dict( + os.environ, env, clear=True + ): + os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir + return mcp_server.gitea_inspect_issue_lock_contract( + issue_number=self.ISSUE_NUMBER, + branch_name=branch, + worktree_path=worktree, + remote="prgs", + ) + + def test_bootstrap_inspect_heartbeat_diverge_and_create_pr_provenance(self): + def _forbidden(*args, **kwargs): + raise AssertionError( + "gitea_lock_issue must not be needed to repair a bootstrap lock" + ) + + with mock.patch.object(mcp_server, "gitea_lock_issue", side_effect=_forbidden): + # 1. Bootstrap through the real function. + result = author_issue_bootstrap.bootstrap_author_issue_worktree( + issue_number=self.ISSUE_NUMBER, + canonical_repo_root=self.repo, + expected_base_sha=self.master_sha, + idempotency_key="e2e-953", + remote="prgs", + lock_dir=self.lock_dir, + owner_session="session-e2e-953", + active_identity=IDENTITY, + active_profile=PROFILE, + ) + self.assertTrue(result.get("success"), result) + self.assertTrue(result.get("implementation_allowed")) + branch = result["branch_name"] + worktree = result["worktree_path"] + token = result["task_session_id"] + self.assertTrue(token) + + # 2. Inspect through the registered read-only tool. + inspected = self._inspect(worktree, branch) + self.assertTrue(inspected.get("success"), inspected) + self.assertTrue(inspected["lock_contract"]["canonical"], inspected) + self.assertTrue(inspected["lock_contract"]["heartbeatable"]) + self.assertTrue(inspected["lock_contract"]["create_pr_eligible"]) + self.assertFalse(inspected.get("mutation_performed")) + + # 3. Heartbeat/renew while still base-equivalent. + before = issue_lock_store.heartbeat_session_lock( + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + issue_number=self.ISSUE_NUMBER, + branch_name=branch, + worktree_path=worktree, + identity=IDENTITY, + profile=PROFILE, + task_session_id=token, + lock_dir=self.lock_dir, + ) + self.assertTrue(before["success"], before.get("reasons")) + + # 4. Legitimate divergence: implement, commit, push. + with open(os.path.join(worktree, "feature.py"), "w") as handle: + handle.write("VALUE = 1\n") + self._git("add", "feature.py", cwd=worktree) + self._git("commit", "-q", "-m", "feat: implement", cwd=worktree) + self._git("push", "-q", "-u", "origin", branch, cwd=worktree) + head = self._git("rev-parse", "HEAD", cwd=worktree).stdout.strip() + remote_head = subprocess.run( + ["git", "-C", self.origin, "rev-parse", f"refs/heads/{branch}"], + capture_output=True, + text=True, + check=True, + ).stdout.strip() + self.assertEqual(head, remote_head) + + # 5. Downstream validation after divergence. + after = issue_lock_store.heartbeat_session_lock( + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + issue_number=self.ISSUE_NUMBER, + branch_name=branch, + worktree_path=worktree, + identity=IDENTITY, + profile=PROFILE, + task_session_id=token, + lock_dir=self.lock_dir, + ) + self.assertTrue(after["success"], after.get("reasons")) + # The pre-mutation ownership re-check (#438) accepts the diverged + # branch without any repair step. + diverged_lock = issue_lock_store.read_lock_file(result["lock_state"]) + proof = issue_lock_store.verify_lock_for_mutation( + diverged_lock, + issue_number=self.ISSUE_NUMBER, + branch_name=branch, + worktree_path=worktree, + ) + self.assertTrue(proof["proven"], proof["reasons"]) + + # 6. The unchanged #447 create_pr provenance guard accepts it. + final = issue_lock_store.read_lock_file(result["lock_state"]) + verdict = issue_lock_provenance.assess_lock_file_for_create_pr(final) + self.assertTrue(verdict["proven"], verdict["reasons"]) + + # The branch was never rewound to satisfy any gate. + merge_base = self._git("merge-base", "master", branch, cwd=worktree).stdout.strip() + self.assertEqual(merge_base, self.master_sha) + + +class DeadProvenanceConstantRemoved(unittest.TestCase): + """Review 632 F3: no second lock source may appear to exist.""" + + def test_no_recovery_source_constant_is_exported(self): + self.assertFalse( + hasattr(author_lock_contract, "SOURCE_BOOTSTRAP_LOCK_RECOVERY") + ) + + def test_bootstrap_source_is_still_the_sanctioned_lock_issue_source(self): + self.assertEqual( + author_lock_contract.SOURCE_BOOTSTRAP, + issue_lock_provenance.SOURCE_LOCK_ISSUE, + ) + + def test_the_sanctioned_source_set_is_still_not_widened(self): + self.assertNotIn( + "gitea_recover_incomplete_bootstrap_lock", + issue_lock_provenance.SANCTIONED_LOCK_SOURCES, + ) + + +class CapabilityRegistration(unittest.TestCase): + """The new operations are registered and role-gated.""" + + def test_recovery_is_registered_as_an_author_operation(self): + import task_capability_map + + self.assertEqual( + task_capability_map.required_permission( + "recover_incomplete_bootstrap_lock" + ), + "gitea.issue.comment", + ) + self.assertEqual( + task_capability_map.required_role("recover_incomplete_bootstrap_lock"), + "author", + ) + + def test_inspection_requires_only_read_permission(self): + """AC16: a read-only surface must not carry a mutating permission.""" + import task_capability_map + + self.assertEqual( + task_capability_map.required_permission("inspect_issue_lock_contract"), + "gitea.read", + ) + + +class WorkflowDocumentation(unittest.TestCase): + """AC20: the canonical ordering and recovery path are documented.""" + + def test_author_workflow_documents_ordering_and_recovery(self): + root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + doc = os.path.join(root, "docs", "author-issue-lock-contract.md") + self.assertTrue(os.path.exists(doc), f"missing {doc}") + with open(doc) as handle: + text = handle.read() + self.assertIn("gitea_recover_incomplete_bootstrap_lock", text) + self.assertIn("gitea_lock_issue", text) + self.assertIn("gitea_bootstrap_author_issue_worktree", text) + self.assertIn("before writing any implementation", text.lower()) + + +if __name__ == "__main__": + unittest.main()