Compare commits

..
Author SHA1 Message Date
sysadminandClaude Opus 4.8 ab4af23afd fix: use exact issue-number boundary in own-branch adoption (#442)
Replace substring issue-marker matching with a numeric word-boundary
regex so issue-42 adoption is not false-blocked by unrelated issue-420
branches. Add regression tests for the #42 vs #420 collision.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-07 17:31:34 -04:00
sysadmin c2330e3929 resolve conflicts for PR #461 2026-07-07 17:30:06 -04:00
sysadminandClaude Opus 4.8 ea471542da feat: add own-branch lock adoption recovery to gitea_lock_issue (Closes #442)
Closes #442.

gitea_lock_issue treated ANY remote branch containing issue-<n> as
duplicate competing work (#400) and failed closed — including the issue's
own already-pushed branch. If the in-memory lock was lost (e.g. MCP server
restart) after the branch was pushed but before the PR, there was no
non-destructive way to reacquire the lock, deadlocking PR creation. This
surfaced during recovery of #420 (feat/issue-420-server-code-parity).

Adds a safe own-branch adoption path.

Changes:
- issue_lock_adoption.py — assess_own_branch_adoption() distinguishes the
  issue's exact requested branch (adopt) from a different same-issue branch
  (competing, still fail-closed); ambiguous (own + other) fails closed.
  build_adoption_proof() emits the required proof block.
- gitea_mcp_server.py — gitea_lock_issue consults the adoption assessment in
  place of the blanket branch-match block; on adoption the result carries an
  `adoption` proof (issue, branch, head commit, reason, no-existing-PR proof,
  no-competing-live-lock proof, lock file path/status). Open-PR and active-
  lease checks already run first, so adoption implies neither exists. Adds
  _branch_entry_commit_sha helper.
- Tests: tests/test_issue_lock_adoption.py (8 cases) + 3 MCP-level cases in
  test_mcp_server.py (adopt own branch, open PR blocks adoption, create_pr
  proceeds after adopted lock). Existing competing-branch block preserved
  (still raises "already has matching branch").

Safety:
- Different same-issue branch, competing branch, existing open PR, ambiguous
  ownership all remain fail-closed. #400 duplicate-work protection intact.
- #420's feature branch was only read (git ls-remote); not modified/deleted.

Validation:
  venv/bin/python -m pytest tests/ -q  ->  1611 passed, 6 skipped

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-07 17:30:06 -04:00
sysadmin f061fee18d Merge pull request 'feat: add per-PR reviewer leases for parallel review (Closes #407)' (#424) from feat/issue-407-reviewer-pr-leases into master 2026-07-07 16:27:19 -05:00
sysadminandClaude Opus 4.8 3e4b721d60 test: align duplicate gate lock fixtures with #447 provenance (#407)
Rebase onto master (#413/#463) requires sanctioned lock_provenance in
create_pr duplicate-recheck tests.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-07 17:18:00 -04:00
sysadminandClaude Opus 4.8 1830360850 test: fix reviewer lease gate regressions in MCP integration tests (#407)
Align merge/review/audit mocks with the per-PR reviewer lease gate introduced
in #407: install owned session leases after init_review_decision_lock, stub
_fetch_pr_comments and _authenticated_username to preserve mock ordering, and
update head-SHA mismatch expectations for lease-first fail-closed paths.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-07 17:16:06 -04:00
sysadminandClaude Opus 4.8 06c2069234 feat: add per-PR reviewer leases for parallel review (Closes #407)
Structured PR-thread leases with acquire/heartbeat MCP tools and
mutation-time enforcement so reviewers cannot race the same PR queue.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-07 17:16:06 -04:00
21 changed files with 1421 additions and 1180 deletions
+6 -40
View File
@@ -274,24 +274,12 @@ is proven abandoned and the takeover is recorded.
Gitea-Tools lease gates: `gitea_lock_issue` (fail-closed before author
mutations), `status:in-progress`, and claim comments. `gitea_lock_issue`
records an `author_issue_work` lease in a keyed lock file under
`GITEA_ISSUE_LOCK_DIR` (default `~/.cache/gitea-tools/issue-locks`), one file
per `remote` + `org` + `repo` + `issue_number`. The current MCP session binds
its active lock through a per-process pointer so concurrent repos/issues never
share one overwrite-prone slot (#443).
Each lock payload includes issue number, optional PR number, branch, worktree
path, claimant identity/profile, created timestamp, expiry timestamp, and last
heartbeat timestamp. An active same-issue/same-operation lease blocks duplicate
work. An expired lease still blocks takeover until a recovery review records why
the prior work is abandoned, completed, or unsafe to continue.
**Do not manually seed `/tmp/gitea_issue_lock.json` or any lock file as a normal
recovery path.** That global slot is deprecated and can clobber unrelated live
leases (#438). After an MCP restart, call `gitea_lock_issue` again — own-branch
adoption rebinds the session when the issue's exact branch already exists (#442).
`gitea_create_pr` resolves the durable keyed lock by session pointer or by
matching `head` branch without unsafe manual seeding.
records an `author_issue_work` lease in the issue-lock payload with issue
number, optional PR number, branch, worktree path, claimant identity/profile,
created timestamp, expiry timestamp, and last heartbeat timestamp. An active
same-issue/same-operation lease blocks duplicate work. An expired lease still
blocks takeover until a recovery review records why the prior work is abandoned,
completed, or unsafe to continue.
**Issue-lock recovery (#447):** Do not manually seed, restore, or delete
`/tmp/gitea_issue_lock.json` as a normal recovery path. That file is global
@@ -308,28 +296,6 @@ metadata. Final-report validation blocks handoffs that hide lock read/write/dele
under `External-state mutations: none` or mix author PR creation with reviewer
approval in one run. See also #438 (global lock redesign).
### Non-destructive lock recovery after push (#440)
When work is pushed but the in-memory MCP session lock is lost (server restart,
crashed process, or stale session pointer):
1. **Do not delete the remote branch.** Branch deletion is not a normal recovery
step and can destroy the only copy of unmerged work.
2. **Re-run `gitea_lock_issue`** with the same `issue_number`, exact
`branch_name`, and active `branches/` worktree. Own-branch adoption
reacquires the lease when the remote branch is the caller's exact branch and
no open PR or competing same-issue branch exists.
3. **Call `gitea_create_pr`** with the same `head` branch. The server resolves
the durable keyed lock file even when the session pointer was cleared at
restart.
4. **Stop fail-closed** when another actor owns a competing branch, an open PR
already exists, or a live foreign lease blocks takeover — never adopt their
branch.
Branch ownership uses structured evidence
`(fix|feat|docs|chore)/issue-<n>-<desc>` — not broad substring matches on
`issue-<n>`.
Remote branches matching the issue number are also treated as active work unless
the recovery review proves the branch is abandoned or superseded. Never delete
or clean up a branch when it has an active lease, dirty worktree, open PR, or is
+325 -71
View File
@@ -539,13 +539,12 @@ import review_proofs # noqa: E402
import agent_temp_artifacts
import issue_lock_worktree # noqa: E402
import issue_lock_provenance # noqa: E402
import issue_lock_store # noqa: E402
import issue_lock_adoption # noqa: E402
import issue_branch_ownership # noqa: E402
import already_landed_reconcile # noqa: E402
import author_mutation_worktree # noqa: E402
import issue_claim_heartbeat # noqa: E402
import issue_work_duplicate_gate # noqa: E402
import issue_lock_adoption # noqa: E402
import reviewer_pr_lease # noqa: E402
import merged_cleanup_reconcile # noqa: E402
import reconciler_profile # noqa: E402
import reconciliation_workflow # noqa: E402
@@ -553,9 +552,8 @@ import review_merge_state_machine # noqa: E402
import native_mcp_preference # noqa: E402
# Keyed issue-lock storage (#443): per remote/org/repo/issue files under
# GITEA_ISSUE_LOCK_DIR, bound to the current MCP session via a per-PID pointer.
# Legacy global path retained only for test/doc references — do not seed manually.
# Fail-closed exact-issue-lock file (#204): written by gitea_lock_issue,
# consumed by gitea_create_pr and scripts/worktree-start.
ISSUE_LOCK_FILE = "/tmp/gitea_issue_lock.json"
WORK_LEASE_TTL_HOURS = 4
AUTHOR_ISSUE_WORK_LEASE = "author_issue_work"
@@ -587,59 +585,15 @@ def _parse_work_lease_timestamp(value: str | None) -> datetime | None:
return None
def _load_existing_issue_lock(
*,
remote: str | None = None,
org: str | None = None,
repo: str | None = None,
issue_number: int | None = None,
) -> dict | None:
if remote and org and repo and issue_number is not None:
return issue_lock_store.load_issue_lock(
remote=remote,
org=org,
repo=repo,
issue_number=issue_number,
)
return issue_lock_store.read_session_issue_lock()
def _resolve_issue_lock_for_pr(
*,
remote: str,
org: str,
repo: str,
head: str,
) -> dict:
lock_data = issue_lock_store.read_session_issue_lock()
if not lock_data:
lock_data = issue_lock_store.find_lock_for_branch(
remote=remote,
org=org,
repo=repo,
branch_name=head,
)
if not lock_data:
raise RuntimeError(
"Issue lock is missing (fail closed). Call gitea_lock_issue first."
)
return lock_data
def _save_issue_lock(data: dict) -> str:
existing = issue_lock_store.load_issue_lock(
remote=str(data.get("remote") or ""),
org=str(data.get("org") or ""),
repo=str(data.get("repo") or ""),
issue_number=int(data.get("issue_number") or 0),
)
overwrite_block = issue_lock_store.assess_foreign_lock_overwrite(existing, data)
if overwrite_block:
raise RuntimeError(overwrite_block)
def _load_existing_issue_lock() -> dict | None:
if not os.path.exists(ISSUE_LOCK_FILE):
return None
try:
return issue_lock_store.bind_session_lock(data)
except Exception as e:
raise RuntimeError(f"Could not write issue lock file: {e}") from e
with open(ISSUE_LOCK_FILE, "r", encoding="utf-8") as f:
data = json.load(f)
return data if isinstance(data, dict) else None
except Exception:
return None
def _work_lease_claimant(host: str | None) -> dict:
@@ -1287,11 +1241,11 @@ def gitea_lock_issue(
worktree_path: Author scratch-clone path to validate (defaults to
GITEA_AUTHOR_WORKTREE or the MCP server project root).
"""
# 1. Enforce canonical issue branch ownership (#440)
if not issue_branch_ownership.branch_belongs_to_issue(branch_name, issue_number):
# 1. Enforce branch name includes issue number
expected_pattern = f"issue-{issue_number}"
if expected_pattern not in branch_name:
raise ValueError(
f"Branch name '{branch_name}' must match "
f"(fix|feat|docs|chore)/issue-{issue_number}-<desc> (fail closed)"
f"Branch name '{branch_name}' must contain locked issue pattern '{expected_pattern}' (fail closed)"
)
blocked = _profile_permission_block(
@@ -1302,9 +1256,8 @@ def gitea_lock_issue(
resolved_worktree = issue_lock_worktree.resolve_author_worktree_path(
worktree_path, PROJECT_ROOT
)
h, o, r = _resolve(remote, host, org, repo)
active_lease_block = issue_lock_store.assess_same_issue_lease_conflict(
_load_existing_issue_lock(remote=remote, org=o, repo=r, issue_number=issue_number),
active_lease_block = _active_work_lease_block(
_load_existing_issue_lock(),
issue_number=issue_number,
branch_name=branch_name,
worktree_path=resolved_worktree,
@@ -1328,6 +1281,7 @@ def gitea_lock_issue(
issue_lock_worktree.format_issue_lock_worktree_error(lock_assessment)
)
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
duplicate_gate = _assess_issue_duplicate_gate(
issue_number,
@@ -1387,7 +1341,11 @@ def gitea_lock_issue(
),
}
lock_file_path = _save_issue_lock(data)
try:
with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
json.dump(data, f)
except Exception as e:
raise RuntimeError(f"Could not write issue lock file: {e}")
agent_artifacts = agent_temp_artifacts.find_agent_temp_artifacts_from_porcelain(
git_state.get("porcelain_status") or ""
@@ -1402,16 +1360,16 @@ def gitea_lock_issue(
"branch_name": branch_name,
"worktree_path": resolved_worktree,
"work_lease": work_lease,
"lock_file_path": lock_file_path,
}
if adoption["adopt"]:
# #442: recovery lock over the issue's own already-pushed branch.
result["adoption"] = issue_lock_adoption.build_adoption_proof(
issue_number=issue_number,
branch_name=branch_name,
assessment=adoption,
open_pr_checked=True,
competing_lock_checked=True,
lock_file_path=lock_file_path,
lock_file_path=ISSUE_LOCK_FILE,
lock_file_status="written",
)
result["message"] = (
@@ -1509,8 +1467,15 @@ def gitea_create_pr(
verify_preflight_purity(remote, worktree_path=worktree_path)
h, o, r = _resolve(remote, host, org, repo)
# ── Issue Lock Validation (Issue #194 / #196 / #443) ──
lock_data = _resolve_issue_lock_for_pr(remote=remote, org=o, repo=r, head=head)
# ── Issue Lock Validation (Issue #194 / #196) ──
if not os.path.exists(ISSUE_LOCK_FILE):
raise RuntimeError("Issue lock is missing (fail closed). Call gitea_lock_issue first.")
try:
with open(ISSUE_LOCK_FILE, "r", encoding="utf-8") as f:
lock_data = json.load(f)
except Exception as e:
raise RuntimeError(f"Could not read issue lock file: {e} (fail closed)")
lock_provenance_check = issue_lock_provenance.assess_lock_file_for_create_pr(
lock_data
@@ -2105,6 +2070,7 @@ def init_review_decision_lock(remote: str | None, task: str | None):
(os.environ.get(SESSION_PROFILE_LOCK_ENV) or "").strip()
or profile_name
)
reviewer_pr_lease.clear_session_lease()
_save_review_decision_lock({
"task": task,
"remote": remote,
@@ -2547,6 +2513,20 @@ def _evaluate_pr_review_submission(
result["permission_report"] = elig["permission_report"]
return result
if live:
reasons.extend(_reviewer_pr_lease_gate(
pr_number=pr_number,
remote=remote,
host=host,
org=org,
repo=repo,
mutation=action,
live_head_sha=result.get("head_sha"),
pinned_head_sha=expected_head_sha,
))
if reasons:
return result
auth_user = result["authenticated_user"]
pr_author = result["pr_author"]
if action == "approve" and auth_user and pr_author and auth_user == pr_author:
@@ -3029,7 +3009,13 @@ def _prepare_commit_payload_files(files: list[dict]) -> tuple[list[dict], list[d
processed_files = []
source_proofs = []
lock_data = issue_lock_store.read_session_issue_lock() or {}
lock_data = {}
if os.path.exists(ISSUE_LOCK_FILE):
try:
with open(ISSUE_LOCK_FILE, "r", encoding="utf-8") as f:
lock_data = json.load(f)
except Exception:
pass
locked_worktree = lock_data.get("worktree_path")
if locked_worktree:
@@ -3340,6 +3326,19 @@ def gitea_merge_pr(
result["permission_report"] = elig["permission_report"]
return result
reasons.extend(_reviewer_pr_lease_gate(
pr_number=pr_number,
remote=remote,
host=host,
org=org,
repo=repo,
mutation="merge",
live_head_sha=result.get("head_sha"),
pinned_head_sha=expected_head_sha,
))
if reasons:
return result
# Gate 4 — head SHA must match if the caller pinned a reviewed SHA.
actual_sha = result["head_sha"]
if expected_head_sha and actual_sha and expected_head_sha != actual_sha:
@@ -4639,6 +4638,261 @@ def _namespace_mutation_block(mutation_task: str, **extra_fields) -> dict | None
return blocked
def _fetch_pr_comments(
pr_number: int,
*,
remote: str,
host: str | None,
org: str | None,
repo: str | None,
) -> list[dict]:
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
api = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments"
return api_request("GET", api, auth) or []
def _reviewer_pr_lease_gate(
*,
pr_number: int,
remote: str,
host: str | None,
org: str | None,
repo: str | None,
mutation: str,
live_head_sha: str | None,
pinned_head_sha: str | None,
) -> list[str]:
"""Return block reasons when the session lacks an owned PR reviewer lease."""
session = reviewer_pr_lease.get_session_lease()
session_id = (session or {}).get("session_id")
identity = _authenticated_username(remote) or ""
try:
comments = _fetch_pr_comments(
pr_number, remote=remote, host=host, org=org, repo=repo)
except Exception as exc:
return [f"cannot fetch PR comments for lease gate: {_redact(str(exc))}"]
assessment = reviewer_pr_lease.assess_mutation_lease_gate(
pr_number=pr_number,
comments=comments,
reviewer_identity=identity,
session_id=session_id,
mutation=mutation,
live_head_sha=live_head_sha,
pinned_head_sha=pinned_head_sha,
)
return list(assessment.get("reasons") or []) if assessment.get("block") else []
@mcp.tool()
def gitea_acquire_reviewer_pr_lease(
pr_number: int,
worktree: str,
candidate_head: str | None = None,
target_branch: str = "master",
target_branch_sha: str | None = None,
issue_number: int | None = None,
session_id: str | None = None,
remote: str = "dadeschools",
host: str | None = None,
org: str | None = None,
repo: str | None = None,
) -> dict:
"""Acquire a per-PR reviewer lease before review/merge mutations (#407)."""
read_block = _profile_operation_gate("gitea.read")
if read_block:
return {
"success": False,
"acquired": False,
"reasons": read_block,
"permission_report": _permission_block_report("gitea.read"),
}
comment_block = _profile_operation_gate("gitea.pr.comment")
if comment_block:
return {
"success": False,
"acquired": False,
"reasons": comment_block,
"permission_report": _permission_block_report("gitea.pr.comment"),
}
verify_preflight_purity(remote)
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
profile = get_profile()
identity = _authenticated_username(remote) or profile.get("username") or ""
sid = (session_id or "").strip() or reviewer_pr_lease.new_session_id()
repo_label = f"{o}/{r}"
comments = _fetch_pr_comments(
pr_number, remote=remote, host=host, org=org, repo=repo)
assessment = reviewer_pr_lease.assess_acquire_lease(
comments,
pr_number=pr_number,
reviewer_identity=identity,
profile=profile.get("profile_name") or "unknown",
session_id=sid,
repo=repo_label,
issue_number=issue_number,
worktree=worktree,
candidate_head=candidate_head,
target_branch=target_branch,
target_branch_sha=target_branch_sha,
)
if not assessment.get("acquire_allowed"):
return {
"success": False,
"acquired": False,
"reasons": assessment.get("reasons") or [],
"existing_lease": assessment.get("existing_lease"),
}
body = assessment["lease_body"]
comment_url = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments"
with _audited(
"comment_pr",
host=h,
remote=remote,
org=o,
repo=r,
pr_number=pr_number,
request_metadata={"source": "acquire_reviewer_pr_lease"},
):
posted = api_request("POST", comment_url, auth, {"body": body})
session_lease = reviewer_pr_lease.record_session_lease({
"pr_number": pr_number,
"issue_number": issue_number,
"session_id": sid,
"reviewer_identity": identity,
"profile": profile.get("profile_name"),
"worktree": worktree,
"phase": "claimed",
"candidate_head": candidate_head,
"target_branch": target_branch,
"target_branch_sha": target_branch_sha,
"repo": repo_label,
"comment_id": posted.get("id"),
})
return {
"success": True,
"acquired": True,
"pr_number": pr_number,
"session_id": sid,
"comment_id": posted.get("id"),
"session_lease": session_lease,
"reasons": [],
}
@mcp.tool()
def gitea_heartbeat_reviewer_pr_lease(
pr_number: int,
phase: str,
worktree: str | None = None,
candidate_head: str | None = None,
target_branch_sha: str | None = None,
remote: str = "dadeschools",
host: str | None = None,
org: str | None = None,
repo: str | None = None,
) -> dict:
"""Post a reviewer lease heartbeat / phase update on the PR thread (#407)."""
comment_block = _profile_operation_gate("gitea.pr.comment")
if comment_block:
return {
"success": False,
"posted": False,
"reasons": comment_block,
"permission_report": _permission_block_report("gitea.pr.comment"),
}
session = reviewer_pr_lease.get_session_lease()
if not session or session.get("pr_number") != pr_number:
return {
"success": False,
"posted": False,
"reasons": [
f"no in-session lease for PR #{pr_number}; acquire first "
"(fail closed)"
],
}
verify_preflight_purity(remote)
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
body = reviewer_pr_lease.format_lease_body(
repo=f"{o}/{r}",
pr_number=pr_number,
issue_number=session.get("issue_number"),
reviewer_identity=session.get("reviewer_identity") or "",
profile=session.get("profile") or "unknown",
session_id=session.get("session_id") or reviewer_pr_lease.new_session_id(),
worktree=worktree or session.get("worktree") or "",
phase=phase,
candidate_head=candidate_head or session.get("candidate_head"),
target_branch=session.get("target_branch") or "master",
target_branch_sha=target_branch_sha or session.get("target_branch_sha"),
)
comment_url = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments"
with _audited(
"comment_pr",
host=h,
remote=remote,
org=o,
repo=r,
pr_number=pr_number,
request_metadata={"source": "heartbeat_reviewer_pr_lease", "phase": phase},
):
posted = api_request("POST", comment_url, auth, {"body": body})
updated = reviewer_pr_lease.record_session_lease({
**session,
"phase": phase,
"worktree": worktree or session.get("worktree"),
"candidate_head": candidate_head or session.get("candidate_head"),
"target_branch_sha": target_branch_sha or session.get("target_branch_sha"),
"last_comment_id": posted.get("id"),
})
return {
"success": True,
"posted": True,
"pr_number": pr_number,
"phase": phase,
"comment_id": posted.get("id"),
"session_lease": updated,
"reasons": [],
}
@mcp.tool()
def gitea_assess_reviewer_pr_lease(
pr_number: int,
remote: str = "dadeschools",
host: str | None = None,
org: str | None = None,
repo: str | None = None,
) -> dict:
"""Read-only: assess active reviewer lease state for a PR (#407)."""
read_block = _profile_operation_gate("gitea.read")
if read_block:
return {
"success": False,
"reasons": read_block,
"permission_report": _permission_block_report("gitea.read"),
}
comments = _fetch_pr_comments(
pr_number, remote=remote, host=host, org=org, repo=repo)
active = reviewer_pr_lease.find_active_reviewer_lease(
comments, pr_number=pr_number)
return {
"success": True,
"pr_number": pr_number,
"active_lease": active,
"session_lease": reviewer_pr_lease.get_session_lease(),
"reasons": [],
}
@mcp.tool()
def gitea_list_issue_comments(
issue_number: int,
-28
View File
@@ -1,28 +0,0 @@
"""Structured issue/branch ownership evidence (#440).
Author branches must follow ``(fix|feat|docs|chore)/issue-<n>-<desc>``. Duplicate-work
and lock-recovery gates use this parser instead of broad substring matching on
``issue-<n>`` so unrelated branch names cannot false-positive.
"""
from __future__ import annotations
import re
ISSUE_BRANCH_RE = re.compile(
r"^(?P<prefix>fix|feat|docs|chore)/issue-(?P<num>\d+)(?:-|$)"
)
def parse_issue_branch(branch_name: str) -> int | None:
"""Return the issue number encoded in a canonical author branch, else None."""
match = ISSUE_BRANCH_RE.match((branch_name or "").strip())
if not match:
return None
return int(match.group("num"))
def branch_belongs_to_issue(branch_name: str, issue_number: int) -> bool:
"""True when ``branch_name`` is a canonical branch for ``issue_number``."""
parsed = parse_issue_branch(branch_name)
return parsed is not None and parsed == issue_number
+45 -6
View File
@@ -1,4 +1,4 @@
"""Own-branch lock adoption / recovery for ``gitea_lock_issue`` (#440 / #442 / #443).
"""Own-branch lock adoption / recovery for ``gitea_lock_issue`` (#442).
When an issue's own already-pushed branch exists, lock reacquisition must be
allowed (adoption) instead of being treated as #400 duplicate competing work.
@@ -14,7 +14,7 @@ this module additionally records whether they passed for proof purposes.
from __future__ import annotations
import issue_branch_ownership
import re
ADOPT = "adopt_existing_branch"
BLOCK_COMPETING = "block_competing_branch"
@@ -35,24 +35,58 @@ def _branch_sha(entry) -> str | None:
return None
def _branch_carries_issue_marker(branch_name: str, issue_number: int) -> bool:
"""Return True when *branch_name* references issue *issue_number* exactly.
Uses a numeric word-boundary so ``issue-42`` does not match inside
``issue-420`` (AC6 / #440).
"""
name = (branch_name or "").strip()
if not name:
return False
pattern = rf"(?:^|/)issue-{int(issue_number)}(?![0-9])"
return re.search(pattern, name) is not None
def assess_own_branch_adoption(
*,
issue_number: int,
requested_branch: str,
existing_branches,
) -> dict:
"""Decide whether an existing matching branch is adoptable."""
"""Decide whether an existing matching branch is adoptable.
Args:
issue_number: The tracking issue number being locked.
requested_branch: The exact branch the caller wants to lock.
existing_branches: Iterable of remote branch entries — either names or
dicts with ``name`` and optional ``commit_sha``.
Returns:
dict with:
* ``outcome`` — one of ADOPT / BLOCK_COMPETING / NO_MATCH
* ``adopt`` (bool), ``block`` (bool)
* ``reason`` (str)
* ``matched_branch`` (str | None), ``matched_head_sha`` (str | None)
* ``competing_branches`` (list[str])
ADOPT: the issue's exact branch exists and no other same-issue branch does.
BLOCK_COMPETING: at least one same-issue branch is not the requested branch.
NO_MATCH: no branch carries the issue marker — normal lock path applies.
"""
requested = (requested_branch or "").strip()
matches: list[tuple[str, str | None]] = []
for entry in existing_branches or []:
name = _branch_name(entry).strip()
if issue_branch_ownership.branch_belongs_to_issue(name, issue_number):
if _branch_carries_issue_marker(name, issue_number):
matches.append((name, _branch_sha(entry)))
competing = sorted({name for name, _ in matches if name != requested})
exact = [(name, sha) for name, sha in matches if name == requested]
# Fail closed whenever any non-requested same-issue branch exists, even if
# the requested branch is also present: ownership is then ambiguous.
if competing:
return {
"outcome": BLOCK_COMPETING,
@@ -104,7 +138,12 @@ def build_adoption_proof(
lock_file_path: str,
lock_file_status: str,
) -> dict:
"""Assemble the proof block returned by ``gitea_lock_issue`` on adoption."""
"""Assemble the proof block returned by ``gitea_lock_issue`` on adoption.
Requirement #4: adoption results must carry issue number, branch name,
branch head commit, adoption reason, no-existing-PR proof, no-competing-
live-lock proof, and lock file path/status.
"""
return {
"issue_number": issue_number,
"branch_name": branch_name,
@@ -114,4 +153,4 @@ def build_adoption_proof(
"no_competing_live_lock_proof": bool(competing_lock_checked),
"lock_file_path": lock_file_path,
"lock_file_status": lock_file_status,
}
}
-385
View File
@@ -1,385 +0,0 @@
"""Keyed, persistent issue-lock storage (#443).
Replaces the single global ``/tmp/gitea_issue_lock.json`` slot with per-issue
lock files under ``GITEA_ISSUE_LOCK_DIR`` (default
``~/.cache/gitea-tools/issue-locks``). Each MCP session binds its active lock
via a per-process pointer file so concurrent repos/issues never clobber each
other.
"""
from __future__ import annotations
import json
import os
import re
import tempfile
from datetime import datetime, timedelta, timezone
from typing import Any
LOCK_DIR_ENV = "GITEA_ISSUE_LOCK_DIR"
DEFAULT_LOCK_DIR = os.path.expanduser("~/.cache/gitea-tools/issue-locks")
WORK_LEASE_TTL_HOURS = 4
AUTHOR_ISSUE_WORK_LEASE = "author_issue_work"
_SAFE_SEGMENT_RE = re.compile(r"[^A-Za-z0-9._+-]+")
def default_lock_dir() -> str:
raw = (os.environ.get(LOCK_DIR_ENV) or DEFAULT_LOCK_DIR).strip()
return raw or DEFAULT_LOCK_DIR
def _sanitize_segment(value: str) -> str:
text = (value or "").strip()
if not text:
return "_"
return _SAFE_SEGMENT_RE.sub("_", text)
def lock_key(
*,
remote: str,
org: str,
repo: str,
issue_number: int,
) -> str:
return "-".join(
_sanitize_segment(part)
for part in (remote, org, repo, str(issue_number))
)
def lock_file_path(
*,
remote: str,
org: str,
repo: str,
issue_number: int,
lock_dir: str | None = None,
) -> str:
root = (lock_dir or default_lock_dir()).strip()
return os.path.join(root, f"{lock_key(remote=remote, org=org, repo=repo, issue_number=issue_number)}.json")
def session_pointer_path(lock_dir: str | None = None) -> str:
root = (lock_dir or default_lock_dir()).strip()
return os.path.join(root, f"session-{os.getpid()}.json")
def _ensure_lock_dir(lock_dir: str | None = None) -> str:
root = (lock_dir or default_lock_dir()).strip()
os.makedirs(root, mode=0o700, exist_ok=True)
return root
def read_lock_file(path: str) -> dict[str, Any] | None:
lock_path = (path or "").strip()
if not lock_path or not os.path.exists(lock_path):
return None
try:
with open(lock_path, encoding="utf-8") as handle:
data = json.load(handle)
except (OSError, json.JSONDecodeError):
return None
return data if isinstance(data, dict) else None
def save_lock_file(path: str, data: dict[str, Any]) -> None:
lock_path = (path or "").strip()
if not lock_path:
raise ValueError("lock path is required (fail closed)")
parent = os.path.dirname(lock_path) or "."
os.makedirs(parent, mode=0o700, exist_ok=True)
payload = json.dumps(data, indent=2, sort_keys=True) + "\n"
fd, temp_path = tempfile.mkstemp(prefix=".lock-", suffix=".json", dir=parent)
try:
with os.fdopen(fd, "w", encoding="utf-8") as handle:
handle.write(payload)
handle.flush()
os.fsync(handle.fileno())
os.replace(temp_path, lock_path)
finally:
if os.path.exists(temp_path):
try:
os.remove(temp_path)
except OSError:
pass
def bind_session_lock(lock_data: dict[str, Any], lock_dir: str | None = None) -> str:
"""Persist a keyed lock and bind it to the current process session."""
remote = str(lock_data.get("remote") or "")
org = str(lock_data.get("org") or "")
repo = str(lock_data.get("repo") or "")
issue_number = int(lock_data.get("issue_number") or 0)
if not remote or not org or not repo or issue_number <= 0:
raise ValueError("lock record must include remote, org, repo, and issue_number")
root = _ensure_lock_dir(lock_dir)
path = lock_file_path(
remote=remote,
org=org,
repo=repo,
issue_number=issue_number,
lock_dir=root,
)
record = dict(lock_data)
record["lock_file_path"] = path
record["session_pid"] = os.getpid()
save_lock_file(path, record)
pointer = {
"pid": os.getpid(),
"lock_file_path": path,
"issue_number": issue_number,
"branch_name": record.get("branch_name"),
"remote": remote,
"org": org,
"repo": repo,
}
save_lock_file(session_pointer_path(root), pointer)
return path
def read_session_issue_lock(lock_dir: str | None = None) -> dict[str, Any] | None:
root = (lock_dir or default_lock_dir()).strip()
pointer = read_lock_file(session_pointer_path(root))
if not pointer:
return None
lock_path = str(pointer.get("lock_file_path") or "").strip()
if not lock_path:
return None
return read_lock_file(lock_path)
def load_issue_lock(
*,
remote: str,
org: str,
repo: str,
issue_number: int,
lock_dir: str | None = None,
) -> dict[str, Any] | None:
return read_lock_file(
lock_file_path(
remote=remote,
org=org,
repo=repo,
issue_number=issue_number,
lock_dir=lock_dir,
)
)
def iter_lock_files(lock_dir: str | None = None) -> list[str]:
root = (lock_dir or default_lock_dir()).strip()
if not os.path.isdir(root):
return []
paths: list[str] = []
for name in os.listdir(root):
if not name.endswith(".json") or name.startswith("session-"):
continue
paths.append(os.path.join(root, name))
return sorted(paths)
def find_lock_for_branch(
*,
remote: str,
org: str,
repo: str,
branch_name: str,
lock_dir: str | None = None,
) -> dict[str, Any] | None:
target = (branch_name or "").strip()
if not target:
return None
for path in iter_lock_files(lock_dir):
lock = read_lock_file(path)
if not lock:
continue
if (
str(lock.get("remote") or "") == remote
and str(lock.get("org") or "") == org
and str(lock.get("repo") or "") == repo
and str(lock.get("branch_name") or "").strip() == target
):
lock = dict(lock)
lock.setdefault("lock_file_path", path)
return lock
return None
def _lease_now(now: datetime | None = None) -> datetime:
return now or datetime.now(timezone.utc)
def _parse_lease_timestamp(value: str | None) -> datetime | None:
text = (value or "").strip()
if not text:
return None
try:
return datetime.fromisoformat(text.replace("Z", "+00:00")).astimezone(timezone.utc)
except ValueError:
return None
def lease_expires_at(lock: dict[str, Any] | None) -> datetime | None:
if not lock:
return None
lease = lock.get("work_lease")
if not isinstance(lease, dict):
return None
return _parse_lease_timestamp(lease.get("expires_at"))
def is_lease_expired(lock: dict[str, Any] | None, *, now: datetime | None = None) -> bool:
expires = lease_expires_at(lock)
if expires is None:
return False
return expires <= _lease_now(now)
def is_lease_live(lock: dict[str, Any] | None, *, now: datetime | None = None) -> bool:
if not lock:
return False
lease = lock.get("work_lease")
if not isinstance(lease, dict):
return True
expires = _parse_lease_timestamp(lease.get("expires_at"))
if expires is None:
return True
return expires > _lease_now(now)
def _same_realpath(left: str | None, right: str | None) -> bool:
if not left or not right:
return False
try:
return os.path.realpath(left) == os.path.realpath(right)
except OSError:
return left == right
def assess_same_issue_lease_conflict(
existing_lock: dict[str, Any] | None,
*,
issue_number: int,
branch_name: str,
worktree_path: str,
operation_type: str = AUTHOR_ISSUE_WORK_LEASE,
now: datetime | None = None,
) -> str | None:
"""Return a fail-closed error when a competing live lease blocks acquisition."""
if not existing_lock:
return None
existing_issue = existing_lock.get("issue_number")
lease = existing_lock.get("work_lease")
existing_operation = (
lease.get("operation_type")
if isinstance(lease, dict)
else AUTHOR_ISSUE_WORK_LEASE
)
if existing_issue != issue_number or existing_operation != operation_type:
return None
existing_branch = existing_lock.get("branch_name")
existing_worktree = existing_lock.get("worktree_path")
same_owner = (
existing_branch == branch_name
and _same_realpath(str(existing_worktree or ""), worktree_path)
)
if is_lease_expired(existing_lock, now=now):
return (
f"Issue #{issue_number} has an expired {operation_type} lease on "
f"branch '{existing_branch}' from worktree '{existing_worktree}'. "
"Recovery review is required before takeover (fail closed)"
)
if same_owner:
return None
return (
f"Issue #{issue_number} already has an active {operation_type} lease on "
f"branch '{existing_branch}' from worktree '{existing_worktree}' "
"(fail closed)"
)
def assess_foreign_lock_overwrite(
existing_lock: dict[str, Any] | None,
incoming_lock: dict[str, Any],
*,
now: datetime | None = None,
) -> str | None:
"""Block writes that would clobber an unrelated live lease on the same key."""
if not existing_lock:
return None
same_issue = existing_lock.get("issue_number") == incoming_lock.get("issue_number")
same_branch = existing_lock.get("branch_name") == incoming_lock.get("branch_name")
same_worktree = _same_realpath(
str(existing_lock.get("worktree_path") or ""),
str(incoming_lock.get("worktree_path") or ""),
)
if same_issue and same_branch and same_worktree:
return None
if not is_lease_live(existing_lock, now=now):
return None
return (
"Refusing to overwrite a live foreign issue lock "
f"(issue #{existing_lock.get('issue_number')}, "
f"branch '{existing_lock.get('branch_name')}', "
f"worktree '{existing_lock.get('worktree_path')}') (fail closed)"
)
def find_live_lock_for_branch(
branch_name: str,
lock_dir: str | None = None,
) -> dict[str, Any] | None:
target = (branch_name or "").strip()
if not target:
return None
for path in iter_lock_files(lock_dir):
lock = read_lock_file(path)
if not lock:
continue
if str(lock.get("branch_name") or "").strip() != target:
continue
if not is_lease_live(lock):
continue
record = dict(lock)
record.setdefault("lock_file_path", path)
return record
return None
def resolve_locked_branch_for_session(
branch_name: str | None = None,
lock_dir: str | None = None,
) -> str:
if branch_name:
lock = find_live_lock_for_branch(branch_name, lock_dir)
if lock:
return str(lock.get("branch_name") or "")
lock = read_session_issue_lock(lock_dir)
return str((lock or {}).get("branch_name") or "")
def has_active_issue_lock(
branch: str,
*,
lock_dir: str | None = None,
) -> bool:
target = (branch or "").strip()
if not target:
return False
for path in iter_lock_files(lock_dir):
lock = read_lock_file(path)
if not lock:
continue
if str(lock.get("branch_name") or "").strip() != target:
continue
if is_lease_live(lock):
return True
return False
+14 -10
View File
@@ -13,9 +13,9 @@ import subprocess
from typing import Any
from reviewer_worktree import parse_dirty_tracked_files
import issue_lock_store
PROTECTED_BRANCHES = frozenset({"master", "main", "dev"})
ISSUE_LOCK_FILE = os.environ.get("GITEA_ISSUE_LOCK_FILE", "/tmp/gitea_issue_lock.json")
CLOSES_FIXES_RE = re.compile(r"\b(?:closes|fixes)\s+#(\d+)\b", re.IGNORECASE)
@@ -37,18 +37,22 @@ def resolve_worktree_path(project_root: str, branch: str) -> str:
def read_issue_lock(path: str | None = None) -> dict[str, Any] | None:
if path:
return issue_lock_store.read_lock_file(path.strip())
return issue_lock_store.read_session_issue_lock()
lock_path = (path or ISSUE_LOCK_FILE).strip()
if not lock_path or not os.path.exists(lock_path):
return None
try:
with open(lock_path, encoding="utf-8") as handle:
data = json.load(handle)
except (OSError, json.JSONDecodeError):
return None
return data if isinstance(data, dict) else None
def has_active_issue_lock(branch: str, lock_path: str | None = None) -> bool:
if lock_path:
lock = issue_lock_store.read_lock_file(lock_path.strip())
if not lock:
return False
return (lock.get("branch_name") or "").strip() == (branch or "").strip()
return issue_lock_store.has_active_issue_lock(branch)
lock = read_issue_lock(lock_path)
if not lock:
return False
return (lock.get("branch_name") or "").strip() == (branch or "").strip()
def collect_open_pr_heads(open_prs: list[dict[str, Any]]) -> set[str]:
+382
View File
@@ -0,0 +1,382 @@
"""Per-PR reviewer leases for safe parallel review sessions (#407)."""
from __future__ import annotations
import os
import re
import uuid
from datetime import datetime, timedelta, timezone
from typing import Any
MARKER = "<!-- mcp-review-lease:v1 -->"
_FIELD_RE = re.compile(
r"^\s*([a-z_]+)\s*:\s*(.+?)\s*$",
re.IGNORECASE | re.MULTILINE,
)
_FULL_SHA = re.compile(r"^[0-9a-f]{40}$", re.IGNORECASE)
_TERMINAL_PHASES = frozenset({"done", "released", "blocked"})
_ACTIVE_PHASES = frozenset({
"claimed",
"validating",
"approved",
"request-changes",
"merging",
})
DEFAULT_LEASE_TTL_MINUTES = 120
STALE_WARNING_MINUTES = 30
RECLAIMABLE_MINUTES = 60
_SESSION_LEASE: dict[str, Any] | None = None
def _parse_timestamp(value: str | None) -> datetime | None:
if not value:
return None
text = value.strip()
if text.endswith("Z"):
text = text[:-1] + "+00:00"
try:
parsed = datetime.fromisoformat(text)
except ValueError:
return None
if parsed.tzinfo is None:
return parsed.replace(tzinfo=timezone.utc)
return parsed.astimezone(timezone.utc)
def _normalize_sha(value: str | None) -> str | None:
text = (value or "").strip().lower()
return text if text and _FULL_SHA.match(text) else None
def _parse_pr_ref(value: str | None) -> int | None:
digits = re.sub(r"[^\d]", "", value or "")
return int(digits) if digits.isdigit() else None
def new_session_id() -> str:
return f"{os.getpid()}-{uuid.uuid4().hex[:12]}"
def format_lease_body(
*,
repo: str,
pr_number: int,
issue_number: int | None,
reviewer_identity: str,
profile: str,
session_id: str,
worktree: str,
phase: str,
candidate_head: str | None,
target_branch: str,
target_branch_sha: str | None,
last_activity: datetime | None = None,
expires_at: datetime | None = None,
blocker: str = "none",
) -> str:
now = last_activity or datetime.now(timezone.utc)
expires = expires_at or (now + timedelta(minutes=DEFAULT_LEASE_TTL_MINUTES))
last_text = now.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace(
"+00:00", "Z"
)
expires_text = expires.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace(
"+00:00", "Z"
)
issue_text = f"#{issue_number}" if issue_number else "none"
lines = [
MARKER,
f"repo: {repo}",
f"pr: #{pr_number}",
f"issue: {issue_text}",
f"reviewer_identity: {reviewer_identity}",
f"profile: {profile}",
f"session_id: {session_id}",
f"worktree: {worktree}",
f"phase: {phase}",
f"candidate_head: {candidate_head or 'none'}",
f"target_branch: {target_branch}",
f"target_branch_sha: {target_branch_sha or 'none'}",
f"last_activity: {last_text}",
f"expires_at: {expires_text}",
f"blocker: {blocker}",
]
return "\n".join(lines)
def parse_lease_comment(body: str) -> dict[str, Any] | None:
text = body or ""
if MARKER not in text:
return None
fields: dict[str, str] = {}
for match in _FIELD_RE.finditer(text):
fields[match.group(1).strip().lower()] = match.group(2).strip()
if not fields:
return None
return {
"repo": fields.get("repo"),
"pr_number": _parse_pr_ref(fields.get("pr")),
"issue_number": _parse_pr_ref(fields.get("issue")),
"reviewer_identity": fields.get("reviewer_identity"),
"profile": fields.get("profile"),
"session_id": fields.get("session_id"),
"worktree": fields.get("worktree"),
"phase": (fields.get("phase") or "").strip().lower() or None,
"candidate_head": _normalize_sha(fields.get("candidate_head")),
"target_branch": fields.get("target_branch"),
"target_branch_sha": _normalize_sha(fields.get("target_branch_sha")),
"last_activity": fields.get("last_activity"),
"expires_at": fields.get("expires_at"),
"blocker": fields.get("blocker"),
"raw_fields": fields,
}
def _lease_entries(comments: list[dict], *, pr_number: int) -> list[dict]:
entries: list[dict] = []
for comment in comments or []:
parsed = parse_lease_comment(comment.get("body") or "")
if not parsed:
continue
if parsed.get("pr_number") not in (None, pr_number):
continue
entries.append({
**parsed,
"comment_id": comment.get("id"),
"author": (comment.get("user") or {}).get("login") or comment.get("author"),
"created_at": comment.get("created_at"),
"updated_at": comment.get("updated_at"),
})
return entries
def _lease_expired(lease: dict, *, now: datetime) -> bool:
expires_at = _parse_timestamp(lease.get("expires_at"))
return bool(expires_at and expires_at <= now)
def _minutes_since_activity(lease: dict, *, now: datetime) -> float | None:
last = _parse_timestamp(lease.get("last_activity"))
if not last:
return None
return (now - last).total_seconds() / 60.0
def classify_lease_freshness(lease: dict, *, now: datetime | None = None) -> str:
"""Return active, stale_warning, reclaimable, expired, or terminal."""
now = now or datetime.now(timezone.utc)
phase = (lease.get("phase") or "").strip().lower()
if phase in _TERMINAL_PHASES:
return "terminal"
if _lease_expired(lease, now=now):
return "expired"
minutes = _minutes_since_activity(lease, now=now)
if minutes is None:
return "active"
if minutes >= RECLAIMABLE_MINUTES:
return "reclaimable"
if minutes >= STALE_WARNING_MINUTES:
return "stale_warning"
return "active"
def find_active_reviewer_lease(
comments: list[dict],
*,
pr_number: int,
now: datetime | None = None,
) -> dict[str, Any] | None:
"""Newest non-terminal, unexpired lease for *pr_number*."""
now = now or datetime.now(timezone.utc)
for lease in reversed(_lease_entries(comments, pr_number=pr_number)):
phase = (lease.get("phase") or "").strip().lower()
if phase in _TERMINAL_PHASES:
continue
if _lease_expired(lease, now=now):
continue
if phase in _ACTIVE_PHASES or phase:
lease = dict(lease)
lease["freshness"] = classify_lease_freshness(lease, now=now)
return lease
return None
def assess_acquire_lease(
comments: list[dict],
*,
pr_number: int,
reviewer_identity: str,
profile: str,
session_id: str,
repo: str,
issue_number: int | None,
worktree: str,
candidate_head: str | None,
target_branch: str,
target_branch_sha: str | None,
now: datetime | None = None,
) -> dict[str, Any]:
"""Fail closed when another session holds an active lease."""
now = now or datetime.now(timezone.utc)
reasons: list[str] = []
existing = find_active_reviewer_lease(comments, pr_number=pr_number, now=now)
if existing:
owner_session = (existing.get("session_id") or "").strip()
freshness = existing.get("freshness") or classify_lease_freshness(existing, now=now)
if owner_session and owner_session != session_id and freshness in {
"active", "stale_warning"
}:
reasons.append(
f"PR #{pr_number} already has active reviewer lease "
f"(session_id={owner_session}, phase={existing.get('phase')})"
)
elif owner_session and owner_session != session_id and freshness == "reclaimable":
reasons.append(
f"PR #{pr_number} lease is reclaimable but still held by "
f"session_id={owner_session}; explicit reclaim not implemented "
"(fail closed)"
)
if not (reviewer_identity or "").strip():
reasons.append("reviewer identity required for lease acquisition")
if not (session_id or "").strip():
reasons.append("session_id required for lease acquisition")
if not (worktree or "").strip():
reasons.append("worktree path required for lease acquisition")
allowed = not reasons
body = None
if allowed:
body = format_lease_body(
repo=repo,
pr_number=pr_number,
issue_number=issue_number,
reviewer_identity=reviewer_identity,
profile=profile,
session_id=session_id,
worktree=worktree,
phase="claimed",
candidate_head=candidate_head,
target_branch=target_branch,
target_branch_sha=target_branch_sha,
last_activity=now,
)
return {
"acquire_allowed": allowed,
"reasons": reasons,
"existing_lease": existing,
"lease_body": body,
"session_id": session_id,
}
def record_session_lease(lease: dict[str, Any]) -> dict[str, Any]:
global _SESSION_LEASE
_SESSION_LEASE = dict(lease)
return dict(_SESSION_LEASE)
def clear_session_lease() -> None:
global _SESSION_LEASE
_SESSION_LEASE = None
def get_session_lease() -> dict[str, Any] | None:
return dict(_SESSION_LEASE) if _SESSION_LEASE else None
def assess_mutation_lease_gate(
*,
pr_number: int,
comments: list[dict],
reviewer_identity: str,
session_id: str | None,
mutation: str,
live_head_sha: str | None,
pinned_head_sha: str | None,
now: datetime | None = None,
) -> dict[str, Any]:
"""Reviewer mutations require an owned, current PR lease."""
now = now or datetime.now(timezone.utc)
reasons: list[str] = []
session = get_session_lease()
active = find_active_reviewer_lease(comments, pr_number=pr_number, now=now)
if not session:
reasons.append(
f"no in-session reviewer lease recorded; acquire via "
f"gitea_acquire_reviewer_pr_lease before {mutation}"
)
elif session.get("pr_number") != pr_number:
reasons.append(
f"session lease is for PR #{session.get('pr_number')}, not #{pr_number}"
)
elif (session.get("session_id") or "") != (session_id or session.get("session_id")):
reasons.append("session lease session_id mismatch (fail closed)")
if active:
owner = (active.get("session_id") or "").strip()
if owner and session_id and owner != session_id:
reasons.append(
f"active PR lease owned by session_id={owner}; current session "
f"cannot {mutation}"
)
pinned = _normalize_sha(pinned_head_sha)
live = _normalize_sha(live_head_sha)
lease_head = active.get("candidate_head")
if pinned and live and pinned != live:
reasons.append(
"PR head changed during lease; stop and re-validate before "
f"reviewer {mutation}"
)
if lease_head and live and lease_head != live:
reasons.append(
"live PR head differs from lease candidate_head; refresh lease "
f"before {mutation}"
)
freshness = active.get("freshness") or classify_lease_freshness(active, now=now)
if freshness in {"expired", "reclaimable"}:
reasons.append(f"reviewer lease freshness is '{freshness}' (fail closed)")
else:
reasons.append(f"no active reviewer lease found on PR #{pr_number}")
allowed = not reasons
return {
"mutation_allowed": allowed,
"block": not allowed,
"reasons": reasons,
"active_lease": active,
"session_lease": session,
}
def assess_lease_inventory(
comments_by_pr: dict[int, list[dict]],
*,
now: datetime | None = None,
) -> dict[str, Any]:
"""Summarize lease states across PR comment threads."""
now = now or datetime.now(timezone.utc)
active: list[dict] = []
stale: list[dict] = []
reclaimable: list[dict] = []
for pr_number, comments in (comments_by_pr or {}).items():
lease = find_active_reviewer_lease(comments, pr_number=pr_number, now=now)
if not lease:
continue
freshness = lease.get("freshness") or classify_lease_freshness(lease, now=now)
entry = {"pr_number": pr_number, "session_id": lease.get("session_id"), "freshness": freshness}
if freshness == "stale_warning":
stale.append(entry)
elif freshness == "reclaimable":
reclaimable.append(entry)
else:
active.append(entry)
return {
"active_review_leases": active,
"stale_review_leases": stale,
"reclaimable_review_leases": reclaimable,
}
+5 -11
View File
@@ -38,21 +38,13 @@ fi
branch="$1"
start_ref="${2:-prgs/master}"
script_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
repo_root="$(cd "$script_dir/.." && pwd)"
# Enforce issue-linked, traceable branch names (issue → branch → worktree → PR).
if [[ "$allow_unlinked" -eq 0 ]]; then
locked_branch=$(python3 -c "
import sys
sys.path.insert(0, '$repo_root')
import issue_lock_store
print(issue_lock_store.resolve_locked_branch_for_session('$branch'))
")
if [[ -z "$locked_branch" ]]; then
echo "Error: No session issue lock is bound. Call gitea_lock_issue before branch creation (fail closed)." >&2
if [[ ! -f "/tmp/gitea_issue_lock.json" ]]; then
echo "Error: Issue lock file '/tmp/gitea_issue_lock.json' is missing. You must lock exactly one issue before branch creation (fail closed)." >&2
exit 2
fi
locked_branch=$(python3 -c "import json; print(json.load(open('/tmp/gitea_issue_lock.json')).get('branch_name', ''))")
if [[ "$branch" != "$locked_branch" ]]; then
echo "Error: Requested branch '$branch' does not match locked branch '$locked_branch' (fail closed)." >&2
exit 2
@@ -76,6 +68,8 @@ EOF
fi
fi
script_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
repo_root="$(cd "$script_dir/.." && pwd)"
worktree_name="${branch//\//-}"
worktree_path="$repo_root/branches/$worktree_name"
@@ -732,6 +732,26 @@ The final report must identify:
* whether same-PR merge continuation was allowed
* whether the run stopped as required
## 26B. Per-PR reviewer lease (#407)
Parallel reviewer sessions are allowed only when each session holds a distinct,
live PR lease.
Before validation or review mutation on a selected PR:
1. Call `gitea_acquire_reviewer_pr_lease` with worktree path, candidate head SHA,
and target branch SHA.
2. Post heartbeats via `gitea_heartbeat_reviewer_pr_lease` before validation,
after validation, before review mutation, and before merge.
3. Do not approve, request changes, or merge unless the in-session lease
matches the selected PR.
If PR head or target branch advances during the lease, stop and refresh
inventory before continuing.
Final reports must include lease session id, acquisition proof, heartbeat
status, and release/blocked status.
## 27. Merge rules
Before merge, rerun fresh live checks:
+4 -5
View File
@@ -74,23 +74,22 @@ ISSUE_WRITE_ENV = {
class TestIssueLockArtifactWarning(unittest.TestCase):
def setUp(self):
self._lock_dir = tempfile.TemporaryDirectory()
env = {**ISSUE_WRITE_ENV, "GITEA_ISSUE_LOCK_DIR": self._lock_dir.name}
self._env_patcher = patch.dict(os.environ, env, clear=True)
self._env_patcher = patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True)
self._env_patcher.start()
def tearDown(self):
self._env_patcher.stop()
self._lock_dir.cleanup()
@patch(
"mcp_server.issue_duplicate_context_fetcher",
return_value=([], [], {"status": "not_claimed"}),
)
@patch("mcp_server.api_get_all", return_value=[])
@patch("mcp_server._auth", return_value="token x")
@patch("mcp_server._resolve", return_value=("h", "o", "r"))
@patch("mcp_server.ISSUE_LOCK_FILE", new_callable=lambda: tempfile.mktemp())
@patch("issue_lock_worktree.read_worktree_git_state")
def test_lock_success_includes_artifact_warning(self, mock_state, *_mocks):
def test_lock_success_includes_artifact_warning(self, mock_state, _lock_file, *_mocks):
mock_state.return_value = {
"current_branch": "master",
"porcelain_status": "?? _emit_payload.py\n",
+29 -3
View File
@@ -286,6 +286,21 @@ class TestSimpleToolAudit(_AuditWiringBase):
class TestGatedToolAudit(_AuditWiringBase):
def setUp(self):
super().setUp()
from tests.test_mcp_server import _install_owned_reviewer_lease
import reviewer_pr_lease
self._lease_patch = _install_owned_reviewer_lease(8)
self._lease_patch.start()
self._auth_identity_patch = patch(
"mcp_server._authenticated_username", return_value="reviewer-bot"
)
self._auth_identity_patch.start()
self.addCleanup(self._auth_identity_patch.stop)
self.addCleanup(self._lease_patch.stop)
self.addCleanup(reviewer_pr_lease.clear_session_lease)
def _pr(self, author, state="open", sha="abc123", mergeable=True):
return {"user": {"login": author}, "state": state,
"head": {"sha": sha}, "mergeable": mergeable}
@@ -344,11 +359,22 @@ class TestGatedToolAudit(_AuditWiringBase):
GITEA_ALLOWED_OPERATIONS="read,review,approve")
with patch.dict(os.environ, env, clear=True):
from mcp_server import init_review_decision_lock, gitea_mark_final_review_decision
from tests.test_mcp_server import _install_owned_reviewer_lease
import reviewer_pr_lease
init_review_decision_lock("prgs", "review_pr")
gitea_mark_final_review_decision(8, "approve", remote="prgs")
r = gitea_submit_pr_review(pr_number=8, action="approve",
body="LGTM", remote="prgs",
final_review_decision_ready=True)
lease_patch = _install_owned_reviewer_lease(8)
lease_patch.start()
try:
r = gitea_submit_pr_review(
pr_number=8, action="approve",
body="LGTM", remote="prgs",
final_review_decision_ready=True,
)
finally:
lease_patch.stop()
reviewer_pr_lease.clear_session_lease()
self.assertTrue(r["performed"])
recs = self._records()
self.assertEqual(len(recs), 1)
+5 -7
View File
@@ -66,12 +66,9 @@ class TestCommitPayloads(unittest.TestCase):
)
self.locked_worktree_path = os.path.realpath(self.locked_worktree_dir.name)
import issue_lock_store
self.lock_file_path = "/tmp/gitea_issue_lock.json"
import issue_lock_provenance
self._lock_dir = tempfile.TemporaryDirectory()
os.environ["GITEA_ISSUE_LOCK_DIR"] = self._lock_dir.name
work_lease = {
"operation_type": "author_issue_work",
"issue_number": 263,
@@ -92,7 +89,8 @@ class TestCommitPayloads(unittest.TestCase):
claimant=work_lease.get("claimant"),
),
}
self.lock_file_path = issue_lock_store.bind_session_lock(self.lock_data)
with open(self.lock_file_path, "w", encoding="utf-8") as fh:
fh.write(json.dumps(self.lock_data))
# Reset preflight status to bypass/pass verification in tests
self.orig_whoami_called = mcp_server._preflight_whoami_called
@@ -116,7 +114,8 @@ class TestCommitPayloads(unittest.TestCase):
self._dir.cleanup()
self.locked_worktree_dir.cleanup()
self._lock_dir.cleanup()
if os.path.exists(self.lock_file_path):
os.remove(self.lock_file_path)
def _env(self, profile: str) -> dict:
return {
@@ -125,7 +124,6 @@ class TestCommitPayloads(unittest.TestCase):
"GITEA_TOKEN_AUTHOR": "author-pass",
"GITEA_TEST_PORCELAIN": "",
"GITEA_AUTHOR_WORKTREE": self.locked_worktree_path,
"GITEA_ISSUE_LOCK_DIR": self._lock_dir.name,
}
@patch("mcp_server.api_request")
-36
View File
@@ -1,36 +0,0 @@
"""Structured issue branch ownership (#440)."""
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import issue_branch_ownership as ibo # noqa: E402
class TestIssueBranchOwnership(unittest.TestCase):
def test_canonical_branch_parses_issue_number(self):
self.assertEqual(
ibo.parse_issue_branch("feat/issue-420-server-code-parity"),
420,
)
def test_prefix_variants_supported(self):
for prefix in ("fix", "feat", "docs", "chore"):
branch = f"{prefix}/issue-440-lock-recovery"
self.assertTrue(ibo.branch_belongs_to_issue(branch, 440))
def test_substring_false_positive_rejected(self):
self.assertFalse(
ibo.branch_belongs_to_issue("feat/my-issue-420-backport", 420)
)
self.assertFalse(ibo.branch_belongs_to_issue("release/issue-420-hotfix", 420))
def test_different_issue_number_rejected(self):
self.assertFalse(
ibo.branch_belongs_to_issue("feat/issue-421-server-code-parity", 420)
)
if __name__ == "__main__":
unittest.main()
+69 -8
View File
@@ -1,4 +1,4 @@
"""Unit tests for own-branch lock adoption decision (#442 / #443)."""
"""Unit tests for own-branch lock adoption decision (#442)."""
import sys
import unittest
from pathlib import Path
@@ -25,6 +25,16 @@ class TestAssessOwnBranchAdoption(unittest.TestCase):
)
self.assertEqual(result["outcome"], ADOPT)
self.assertTrue(result["adopt"])
self.assertFalse(result["block"])
self.assertEqual(result["matched_branch"], REQ)
self.assertEqual(result["matched_head_sha"], "934688a")
def test_exact_own_branch_adopted_when_sha_missing(self):
result = assess_own_branch_adoption(
issue_number=420, requested_branch=REQ, existing_branches=[REQ]
)
self.assertEqual(result["outcome"], ADOPT)
self.assertIsNone(result["matched_head_sha"])
def test_different_branch_same_issue_blocks(self):
result = assess_own_branch_adoption(
@@ -34,6 +44,20 @@ class TestAssessOwnBranchAdoption(unittest.TestCase):
)
self.assertEqual(result["outcome"], BLOCK_COMPETING)
self.assertTrue(result["block"])
self.assertFalse(result["adopt"])
self.assertIn("feat/issue-420-other-work", result["competing_branches"])
self.assertIn("fail closed", result["reason"])
def test_own_branch_plus_competing_branch_blocks(self):
# Ambiguous ownership: fail closed even though the exact branch exists.
result = assess_own_branch_adoption(
issue_number=420,
requested_branch=REQ,
existing_branches=[{"name": REQ}, {"name": "feat/issue-420-rogue"}],
)
self.assertEqual(result["outcome"], BLOCK_COMPETING)
self.assertTrue(result["block"])
self.assertEqual(result["competing_branches"], ["feat/issue-420-rogue"])
def test_no_matching_branch_is_normal_path(self):
result = assess_own_branch_adoption(
@@ -42,19 +66,44 @@ class TestAssessOwnBranchAdoption(unittest.TestCase):
existing_branches=[{"name": "feat/issue-999-unrelated"}],
)
self.assertEqual(result["outcome"], NO_MATCH)
self.assertFalse(result["block"])
self.assertFalse(result["adopt"])
def test_substring_only_branch_name_is_ignored(self):
def test_empty_branch_list_is_normal_path(self):
result = assess_own_branch_adoption(
issue_number=420,
requested_branch=REQ,
existing_branches=[{"name": "feat/my-issue-420-backport"}],
issue_number=420, requested_branch=REQ, existing_branches=[]
)
self.assertEqual(result["outcome"], NO_MATCH)
def test_higher_issue_number_branch_does_not_block_lower_issue_adoption(self):
# issue-420 must not be treated as competing work for issue #42.
own_branch = "feat/issue-42-widget"
result = assess_own_branch_adoption(
issue_number=42,
requested_branch=own_branch,
existing_branches=[
{"name": own_branch, "commit_sha": "abc1234"},
{"name": "feat/issue-420-server-code-parity"},
],
)
self.assertEqual(result["outcome"], ADOPT)
self.assertTrue(result["adopt"])
self.assertFalse(result["block"])
self.assertEqual(result["matched_branch"], own_branch)
def test_unrelated_higher_number_branch_is_ignored_without_own_branch(self):
result = assess_own_branch_adoption(
issue_number=42,
requested_branch="feat/issue-42-thing",
existing_branches=[{"name": "feat/issue-420-server-code-parity"}],
)
self.assertEqual(result["outcome"], NO_MATCH)
self.assertFalse(result["block"])
self.assertFalse(result["adopt"])
class TestBuildAdoptionProof(unittest.TestCase):
def test_proof_has_required_fields(self):
def test_proof_has_all_required_fields(self):
assessment = assess_own_branch_adoption(
issue_number=420,
requested_branch=REQ,
@@ -66,12 +115,24 @@ class TestBuildAdoptionProof(unittest.TestCase):
assessment=assessment,
open_pr_checked=True,
competing_lock_checked=True,
lock_file_path="/tmp/example-lock.json",
lock_file_path="/tmp/gitea_issue_lock.json",
lock_file_status="written",
)
for key in (
"issue_number",
"branch_name",
"branch_head_commit",
"adoption_reason",
"no_existing_pr_proof",
"no_competing_live_lock_proof",
"lock_file_path",
"lock_file_status",
):
self.assertIn(key, proof)
self.assertEqual(proof["branch_head_commit"], "934688a")
self.assertTrue(proof["no_existing_pr_proof"])
self.assertTrue(proof["no_competing_live_lock_proof"])
if __name__ == "__main__":
unittest.main()
unittest.main()
-170
View File
@@ -1,170 +0,0 @@
"""End-to-end lock recovery scenarios for issue #440."""
import os
import sys
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import issue_lock_adoption # noqa: E402
import issue_lock_store as ils # noqa: E402
import mcp_server # noqa: E402
from mcp_server import gitea_create_pr, gitea_lock_issue # noqa: E402
PRGS_REPO = mcp_server.REMOTES["prgs"]["repo"]
FAKE_AUTH = "Basic dGVzdDp0ZXN0"
BRANCH = "feat/issue-420-server-code-parity"
ISSUE_WRITE_ENV = {
"GITEA_ALLOWED_OPERATIONS": (
"gitea.issue.create,gitea.issue.close,gitea.issue.comment"
),
}
CREATE_PR_ENV = {
"GITEA_PROFILE_NAME": "author-test",
"GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.pr.create,gitea.branch.push",
"GITEA_FORBIDDEN_OPERATIONS": "gitea.pr.approve,gitea.pr.merge,gitea.pr.review",
}
def _clean_master_git_state_for_lock():
return {
"current_branch": "master",
"porcelain_status": "",
"base_equivalent": True,
"inspected_git_root": "/tmp/repo",
"base_branch": "master",
}
def _lock_record(**overrides):
import issue_lock_provenance
work_lease = {
"operation_type": "author_issue_work",
"expires_at": "2999-01-01T00:00:00Z",
}
record = {
"issue_number": 420,
"branch_name": BRANCH,
"remote": "prgs",
"org": "Scaled-Tech-Consulting",
"repo": PRGS_REPO,
"worktree_path": os.path.realpath(os.getcwd()),
"work_lease": work_lease,
"lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
tool="gitea_lock_issue",
claimant=work_lease.get("claimant"),
),
}
record.update(overrides)
return record
class TestIssueLockRecovery(unittest.TestCase):
def setUp(self):
self._tmpdir = tempfile.TemporaryDirectory()
self.addCleanup(self._tmpdir.cleanup)
self.lock_dir = self._tmpdir.name
def test_substring_branch_does_not_trigger_adoption(self):
result = issue_lock_adoption.assess_own_branch_adoption(
issue_number=420,
requested_branch=BRANCH,
existing_branches=[{"name": "feat/my-issue-420-backport"}],
)
self.assertEqual(result["outcome"], issue_lock_adoption.NO_MATCH)
def test_competing_actor_branch_blocks_adoption(self):
result = issue_lock_adoption.assess_own_branch_adoption(
issue_number=420,
requested_branch=BRANCH,
existing_branches=[{"name": "feat/issue-420-other-work"}],
)
self.assertTrue(result["block"])
@patch(
"mcp_server.issue_lock_worktree.read_worktree_git_state",
return_value=_clean_master_git_state_for_lock(),
)
@patch("mcp_server.api_get_all")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@patch(
"mcp_server.issue_duplicate_context_fetcher",
return_value=([], [], {"status": "not_claimed"}),
)
def test_lock_recovery_after_restart_adopts_own_branch(self, _dup_fetcher, _auth, mock_api, _git):
mock_api.return_value = [{"name": BRANCH, "commit": {"id": "934688a"}}]
env = {**ISSUE_WRITE_ENV, "GITEA_ISSUE_LOCK_DIR": self.lock_dir}
with patch.dict(os.environ, env, clear=True):
with patch("os.getpid", return_value=9999):
res = gitea_lock_issue(
issue_number=420,
branch_name=BRANCH,
remote="prgs",
)
self.assertTrue(res["success"])
self.assertIn("adoption", res)
found = ils.find_lock_for_branch(
remote="prgs",
org="Scaled-Tech-Consulting",
repo=PRGS_REPO,
branch_name=BRANCH,
lock_dir=self.lock_dir,
)
self.assertEqual(found["branch_name"], BRANCH)
@patch("mcp_server.api_request")
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", return_value=(True, []))
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_pr_resolves_durable_lock_after_session_loss(self, _auth, _role, mock_api):
mock_api.return_value = {"number": 421, "html_url": "https://example/pr/421"}
worktree = os.path.realpath(os.getcwd())
path = ils.lock_file_path(
remote="prgs",
org="Scaled-Tech-Consulting",
repo=PRGS_REPO,
issue_number=420,
lock_dir=self.lock_dir,
)
ils.save_lock_file(path, _lock_record(worktree_path=worktree))
env = {**CREATE_PR_ENV, "GITEA_ISSUE_LOCK_DIR": self.lock_dir}
with patch.dict(os.environ, env, clear=True):
with patch("os.getpid", return_value=8888):
self.assertIsNone(ils.read_session_issue_lock(lock_dir=self.lock_dir))
res = gitea_create_pr(
title=f"feat: server parity Closes #420",
head=BRANCH,
remote="prgs",
worktree_path=worktree,
)
self.assertEqual(res["number"], 421)
@patch(
"mcp_server.issue_lock_worktree.read_worktree_git_state",
return_value=_clean_master_git_state_for_lock(),
)
@patch("mcp_server.api_get_all")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@patch(
"mcp_server.issue_duplicate_context_fetcher",
return_value=([{
"number": 99,
"head": {"ref": BRANCH},
"title": "WIP",
"body": "",
}], [], {"status": "not_claimed"}),
)
def test_open_pr_blocks_recovery_lock(self, _dup_fetcher, _auth, mock_api, _git):
env = {**ISSUE_WRITE_ENV, "GITEA_ISSUE_LOCK_DIR": self.lock_dir}
with patch.dict(os.environ, env, clear=True):
with self.assertRaises(ValueError) as ctx:
gitea_lock_issue(issue_number=420, branch_name=BRANCH, remote="prgs")
self.assertIn("open PR #99 already covers issue", str(ctx.exception))
if __name__ == "__main__":
unittest.main()
-181
View File
@@ -1,181 +0,0 @@
"""Unit tests for keyed issue-lock storage (#443)."""
import json
import os
import sys
import tempfile
import unittest
from datetime import datetime, timedelta, timezone
from pathlib import Path
from unittest import mock
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import issue_lock_store as ils # noqa: E402
def _lease(expires_at: str) -> dict:
return {
"operation_type": ils.AUTHOR_ISSUE_WORK_LEASE,
"expires_at": expires_at,
"created_at": "2026-01-01T00:00:00Z",
"last_heartbeat_at": "2026-01-01T00:00:00Z",
}
def _lock_record(**overrides) -> dict:
record = {
"issue_number": 420,
"branch_name": "feat/issue-420-server-code-parity",
"remote": "prgs",
"org": "Scaled-Tech-Consulting",
"repo": "Gitea-Tools",
"worktree_path": "/tmp/wt-420",
"work_lease": _lease("2999-01-01T00:00:00Z"),
}
record.update(overrides)
return record
class TestIssueLockStore(unittest.TestCase):
def setUp(self):
self._dir = tempfile.TemporaryDirectory()
self.lock_dir = self._dir.name
self._env = mock.patch.dict(os.environ, {"GITEA_ISSUE_LOCK_DIR": self.lock_dir})
self._env.start()
def tearDown(self):
self._env.stop()
self._dir.cleanup()
def test_concurrent_repo_locks_do_not_overwrite(self):
lock_a = _lock_record(
issue_number=108,
branch_name="feat/issue-108-root-menu",
repo="mcp-control-plane",
worktree_path="/tmp/wt-108",
)
lock_b = _lock_record(
issue_number=420,
branch_name="feat/issue-420-server-code-parity",
repo="Gitea-Tools",
worktree_path="/tmp/wt-420",
)
path_a = ils.bind_session_lock(lock_a)
with mock.patch("os.getpid", return_value=9999):
path_b = ils.bind_session_lock(lock_b)
self.assertNotEqual(path_a, path_b)
self.assertTrue(os.path.exists(path_a))
self.assertTrue(os.path.exists(path_b))
stored_a = ils.read_lock_file(path_a)
stored_b = ils.read_lock_file(path_b)
self.assertEqual(stored_a["issue_number"], 108)
self.assertEqual(stored_b["issue_number"], 420)
def test_concurrent_issue_locks_same_repo_do_not_overwrite(self):
lock_a = _lock_record(issue_number=427, branch_name="feat/issue-427-a")
lock_b = _lock_record(issue_number=428, branch_name="feat/issue-428-b")
path_a = ils.bind_session_lock(lock_a)
with mock.patch("os.getpid", return_value=4242):
path_b = ils.bind_session_lock(lock_b)
self.assertNotEqual(path_a, path_b)
self.assertEqual(ils.read_lock_file(path_a)["issue_number"], 427)
self.assertEqual(ils.read_lock_file(path_b)["issue_number"], 428)
def test_foreign_live_lease_blocks_overwrite(self):
existing = _lock_record(
branch_name="feat/issue-420-other",
worktree_path="/tmp/other",
work_lease=_lease("2999-01-01T00:00:00Z"),
)
path = ils.lock_file_path(
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
issue_number=420,
)
ils.save_lock_file(path, existing)
incoming = _lock_record(worktree_path="/tmp/mine")
block = ils.assess_foreign_lock_overwrite(existing, incoming)
self.assertIn("live foreign issue lock", block or "")
def test_expired_lease_allows_takeover_with_conflict_check(self):
existing = _lock_record(
branch_name="feat/issue-420-other",
worktree_path="/tmp/other",
work_lease=_lease("2000-01-01T00:00:00Z"),
)
incoming = _lock_record(worktree_path="/tmp/mine")
self.assertIsNone(ils.assess_foreign_lock_overwrite(existing, incoming))
block = ils.assess_same_issue_lease_conflict(
existing,
issue_number=420,
branch_name="feat/issue-420-server-code-parity",
worktree_path="/tmp/mine",
)
self.assertIn("Recovery review is required", block or "")
def test_same_owner_lease_conflict_allows_refresh(self):
worktree = "/tmp/wt-420"
existing = _lock_record(worktree_path=worktree)
block = ils.assess_same_issue_lease_conflict(
existing,
issue_number=420,
branch_name="feat/issue-420-server-code-parity",
worktree_path=worktree,
)
self.assertIsNone(block)
def test_find_lock_for_branch_after_restart(self):
record = _lock_record()
path = ils.lock_file_path(
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
issue_number=420,
)
ils.save_lock_file(path, record)
with mock.patch("os.getpid", return_value=5555):
self.assertIsNone(ils.read_session_issue_lock())
found = ils.find_lock_for_branch(
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
branch_name="feat/issue-420-server-code-parity",
)
self.assertEqual(found["issue_number"], 420)
def test_has_active_issue_lock_scans_keyed_store(self):
ils.bind_session_lock(_lock_record())
self.assertTrue(
ils.has_active_issue_lock("feat/issue-420-server-code-parity")
)
self.assertFalse(ils.has_active_issue_lock("feat/issue-999-other"))
def test_atomic_write_preserves_unrelated_lock(self):
path_a = ils.lock_file_path(
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
issue_number=108,
)
ils.save_lock_file(path_a, _lock_record(issue_number=108, repo="mcp-control-plane"))
path_b = ils.lock_file_path(
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
issue_number=420,
)
ils.save_lock_file(path_b, _lock_record())
self.assertTrue(os.path.exists(path_a))
self.assertTrue(os.path.exists(path_b))
self.assertEqual(ils.read_lock_file(path_a)["issue_number"], 108)
if __name__ == "__main__":
unittest.main()
+15 -1
View File
@@ -9,6 +9,7 @@ from unittest.mock import patch
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import issue_lock_provenance
import issue_work_duplicate_gate as dup_gate
import mcp_server
from issue_work_duplicate_gate import (
@@ -122,8 +123,9 @@ class TestDuplicateReportOutcome(unittest.TestCase):
class TestInjectableDuplicateFetcher(unittest.TestCase):
@patch("mcp_server.api_get_all", return_value=[])
@patch("mcp_server.get_auth_header", return_value="token x")
def test_lock_issue_uses_injected_fetcher(self, _auth):
def test_lock_issue_uses_injected_fetcher(self, _auth, _api):
seen = {}
def fetcher(h, o, r, auth, issue_number):
@@ -172,11 +174,23 @@ class TestMcpDuplicateRecheck(unittest.TestCase):
self._dir.cleanup()
def _write_lock(self, issue_number=400, branch="feat/issue-400-x"):
work_lease = {
"operation_type": "author_issue_work",
"issue_number": issue_number,
"branch": branch,
"claimant": {"username": "test-user", "profile": "test-author"},
"expires_at": "2999-01-01T00:00:00Z",
}
with open(self.lock_path, "w", encoding="utf-8") as fh:
json.dump({
"issue_number": issue_number,
"branch_name": branch,
"remote": "prgs",
"work_lease": work_lease,
"lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
tool="gitea_lock_issue",
claimant=work_lease.get("claimant"),
),
}, fh)
@patch("mcp_server._assess_issue_duplicate_gate")
+284 -189
View File
@@ -6,7 +6,6 @@ the MCP protocol) with mocked API responses.
import json
import os
import sys
import tempfile
import unittest
from unittest.mock import patch, MagicMock
@@ -46,7 +45,6 @@ from gitea_auth import get_profile # noqa: E402
import gitea_config # noqa: E402
import mcp_server
import issue_lock_store
FAKE_AUTH = "Basic dGVzdDp0ZXN0"
@@ -81,6 +79,64 @@ def _visible_approval_reviews(reviewer="reviewer-bot", sha="abc123"):
return [_formal_review(reviewer, "APPROVED", sha=sha)]
_DEFAULT_LEASE_SESSION = "mcp-test-reviewer-lease"
def _reviewer_lease_comment(
pr_number,
*,
session_id=_DEFAULT_LEASE_SESSION,
head_sha="abc123",
reviewer="reviewer-bot",
):
from datetime import datetime, timezone
import reviewer_pr_lease
body = reviewer_pr_lease.format_lease_body(
repo="Scaled-Tech-Consulting/Gitea-Tools",
pr_number=pr_number,
issue_number=407,
reviewer_identity=reviewer,
profile="gitea-reviewer",
session_id=session_id,
worktree="branches/review-test",
phase="claimed",
candidate_head=head_sha,
target_branch="master",
target_branch_sha="b" * 40,
last_activity=datetime.now(timezone.utc),
)
return {"id": 9001, "body": body, "user": {"login": reviewer}}
def _install_owned_reviewer_lease(
pr_number,
*,
session_id=_DEFAULT_LEASE_SESSION,
head_sha="abc123",
):
import reviewer_pr_lease
reviewer_pr_lease.clear_session_lease()
reviewer_pr_lease.record_session_lease({
"pr_number": pr_number,
"session_id": session_id,
"candidate_head": head_sha,
"target_branch": "master",
})
return patch(
"mcp_server._fetch_pr_comments",
return_value=[
_reviewer_lease_comment(
pr_number,
session_id=session_id,
head_sha=head_sha,
)
],
)
# Issue-write tools are profile-gated (#69).
ISSUE_WRITE_ENV = {
"GITEA_ALLOWED_OPERATIONS": (
@@ -118,7 +174,6 @@ def _sample_issue_lock(issue_number=123, branch_name="feat/x", **overrides):
"remote": "dadeschools",
"org": "Scaled-Tech-Consulting",
"repo": "Gitea-Tools",
"worktree_path": "/tmp/test-worktree",
"work_lease": work_lease,
"lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
tool="gitea_lock_issue",
@@ -134,17 +189,6 @@ def _clear_duplicate_context_fetcher(*_args, **_kwargs):
return [], [], {"status": "not_claimed"}
def _bind_test_lock(**overrides) -> str:
remote = overrides.get("remote", "dadeschools")
record = _sample_issue_lock(**overrides)
if remote in mcp_server.REMOTES:
profile = mcp_server.REMOTES[remote]
record.setdefault("org", profile["org"])
record.setdefault("repo", profile["repo"])
record["remote"] = remote
return issue_lock_store.bind_session_lock(record)
# ---------------------------------------------------------------------------
# Create Issue
# ---------------------------------------------------------------------------
@@ -207,21 +251,18 @@ class TestCreatePR(unittest.TestCase):
return_value=(True, []))
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_creates_pr(self, _auth, mock_api, _role, _dup_fetcher):
worktree = os.path.realpath(os.getcwd())
@patch("os.path.exists", return_value=True)
@patch("builtins.open")
def test_creates_pr(self, mock_open, mock_exists, _auth, mock_api, _role, _dup_fetcher):
lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x"))
mock_open.return_value.__enter__.return_value.read.return_value = lock_json
mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
with tempfile.TemporaryDirectory() as lock_dir:
env = {**CREATE_PR_ENV, "GITEA_ISSUE_LOCK_DIR": lock_dir}
with patch.dict(os.environ, env, clear=True):
_bind_test_lock(issue_number=123, branch_name="feat/x", worktree_path=worktree)
result = gitea_create_pr(
title="feat: X Closes #123",
head="feat/x",
base="main",
worktree_path=worktree,
)
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
self.assertEqual(result["number"], 3)
self.assertNotIn("url", result)
mock_exists.assert_called_with(ISSUE_LOCK_FILE)
mock_open.assert_called_with(ISSUE_LOCK_FILE, "r", encoding="utf-8")
payload = mock_api.call_args[0][3]
self.assertEqual(payload["head"], "feat/x")
self.assertEqual(payload["base"], "main")
@@ -235,42 +276,30 @@ class TestCreatePR(unittest.TestCase):
return_value=(True, []))
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_pr_reveal_opt_in_includes_url(self, _auth, mock_api, _role, _dup_fetcher):
worktree = os.path.realpath(os.getcwd())
@patch("os.path.exists", return_value=True)
@patch("builtins.open")
def test_create_pr_reveal_opt_in_includes_url(self, mock_open, mock_exists, _auth, mock_api, _role, _dup_fetcher):
lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x"))
mock_open.return_value.__enter__.return_value.read.return_value = lock_json
mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
with tempfile.TemporaryDirectory() as lock_dir:
env = {**CREATE_PR_ENV, "GITEA_ISSUE_LOCK_DIR": lock_dir, "GITEA_MCP_REVEAL_ENDPOINTS": "1"}
with patch.dict(os.environ, env, clear=True):
_bind_test_lock(issue_number=123, branch_name="feat/x", worktree_path=worktree)
result = gitea_create_pr(
title="feat: X Closes #123",
head="feat/x",
base="main",
worktree_path=worktree,
)
env = {**CREATE_PR_ENV, "GITEA_MCP_REVEAL_ENDPOINTS": "1"}
with patch.dict(os.environ, env, clear=True):
result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
self.assertIn("pulls/3", result["url"])
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_pr_locked_issue_mismatch_fails(self, _auth, _role):
worktree = os.path.realpath(os.getcwd())
with tempfile.TemporaryDirectory() as lock_dir:
env = {**CREATE_PR_ENV, "GITEA_ISSUE_LOCK_DIR": lock_dir}
with patch.dict(os.environ, env, clear=True):
_bind_test_lock(
issue_number=123,
branch_name="feat/x",
worktree_path=worktree,
)
with self.assertRaises(ValueError) as ctx:
gitea_create_pr(
title="feat: X Closes #999",
head="feat/x",
base="main",
worktree_path=worktree,
)
@patch("os.path.exists", return_value=True)
@patch("builtins.open")
def test_create_pr_locked_issue_mismatch_fails(self, mock_open, mock_exists, _auth, _role):
lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x"))
mock_open.return_value.__enter__.return_value.read.return_value = lock_json
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
with self.assertRaises(ValueError) as ctx:
gitea_create_pr(title="feat: X Closes #999", head="feat/x", base="main")
self.assertIn("Closes #123", str(ctx.exception))
mock_open.assert_called_with(ISSUE_LOCK_FILE, "r", encoding="utf-8")
# ---------------------------------------------------------------------------
@@ -568,6 +597,19 @@ class TestViewPR(unittest.TestCase):
class TestMergePR(unittest.TestCase):
"""Gated merge workflow (#16). gitea_merge_pr is the only merge path."""
def setUp(self):
import reviewer_pr_lease
self._lease_patch = _install_owned_reviewer_lease(8)
self._lease_patch.start()
self._auth_identity_patch = patch(
"mcp_server._authenticated_username", return_value="reviewer-bot"
)
self._auth_identity_patch.start()
self.addCleanup(self._auth_identity_patch.stop)
self.addCleanup(self._lease_patch.stop)
self.addCleanup(reviewer_pr_lease.clear_session_lease)
def _pr(self, author, state="open", sha="abc123", mergeable=True):
return {
"user": {"login": author},
@@ -828,9 +870,11 @@ class TestMergePR(unittest.TestCase):
pr_number=8, confirmation=self._confirm(8),
expected_head_sha="deadbeef", remote="prgs")
self.assertFalse(r["performed"])
self.assertIn(
"expected head SHA does not match current PR head (fail closed)",
r["reasons"])
self.assertTrue(any(
"expected head SHA does not match current PR head (fail closed)" in reason
or "PR head changed during lease" in reason
for reason in r["reasons"]
))
self._assert_no_merge_call(mock_api)
@patch("mcp_server.api_request")
@@ -1709,7 +1753,20 @@ class TestReviewDecisionValidationGate(unittest.TestCase):
}
def setUp(self):
import reviewer_pr_lease
init_review_decision_lock("prgs", "review_pr")
self._lease_patch = _install_owned_reviewer_lease(
self.PR, head_sha=self.SHA,
)
self._lease_patch.start()
self._auth_identity_patch = patch(
"mcp_server._authenticated_username", return_value="reviewer-bot"
)
self._auth_identity_patch.start()
self.addCleanup(self._auth_identity_patch.stop)
self.addCleanup(self._lease_patch.stop)
self.addCleanup(reviewer_pr_lease.clear_session_lease)
def _env(self):
return patch.dict(os.environ, {
@@ -1804,8 +1861,19 @@ class TestSubmitPrReview(unittest.TestCase):
"""Gated review-mutation tool (#15)."""
def setUp(self):
import reviewer_pr_lease
init_review_decision_lock("prgs", "review_pr")
gitea_mark_final_review_decision(8, "approve", remote="prgs")
self._lease_patch = _install_owned_reviewer_lease(8)
self._lease_patch.start()
self._auth_identity_patch = patch(
"mcp_server._authenticated_username", return_value="reviewer-bot"
)
self._auth_identity_patch.start()
self.addCleanup(self._auth_identity_patch.stop)
self.addCleanup(self._lease_patch.stop)
self.addCleanup(reviewer_pr_lease.clear_session_lease)
def _pr(self, author, state="open", sha="abc123", mergeable=True):
return {
@@ -2043,9 +2111,11 @@ class TestSubmitPrReview(unittest.TestCase):
final_review_decision_ready=True,
)
self.assertFalse(r["performed"])
self.assertIn(
"expected head SHA does not match current PR head (fail closed)",
r["reasons"])
self.assertTrue(any(
"expected head SHA does not match current PR head (fail closed)" in reason
or "PR head changed during lease" in reason
for reason in r["reasons"]
))
self._assert_no_mutation(mock_api)
def test_head_sha_match_allows(self):
@@ -2108,9 +2178,9 @@ class TestSubmitPrReview(unittest.TestCase):
env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
"GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
with patch.dict(os.environ, env, clear=True):
gitea_mark_final_review_decision(5, "approve", remote="prgs")
gitea_mark_final_review_decision(8, "approve", remote="prgs")
r = gitea_submit_pr_review(
pr_number=5, action="approve", remote="prgs",
pr_number=8, action="approve", remote="prgs",
final_review_decision_ready=True,
)
self.assertFalse(r["performed"])
@@ -2329,6 +2399,13 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
self.assertEqual(res["cleanup_status"].get(1), "not present")
def test_merge_pr_with_closes_removes_label(self):
import reviewer_pr_lease
lease_patch = _install_owned_reviewer_lease(1, head_sha="sha123")
lease_patch.start()
self.addCleanup(lease_patch.stop)
self.addCleanup(reviewer_pr_lease.clear_session_lease)
def api_side_effect(method, url, auth, payload=None):
if method == "GET" and "/user" in url:
return {"login": "merger"}
@@ -2363,6 +2440,13 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
self.assertEqual(res["cleanup_status"].get(123), "released")
def test_merge_pr_with_branch_name_removes_label(self):
import reviewer_pr_lease
lease_patch = _install_owned_reviewer_lease(1, head_sha="sha123")
lease_patch.start()
self.addCleanup(lease_patch.stop)
self.addCleanup(reviewer_pr_lease.clear_session_lease)
def api_side_effect(method, url, auth, payload=None):
if method == "GET" and "/user" in url:
return {"login": "merger"}
@@ -3044,10 +3128,12 @@ class TestVerifyMutationAuthority(unittest.TestCase):
# profile; the active profile resolves as reviewer — side-channel
# override rejected even with a matching in-process authority.
self._authority()
with patch.dict(os.environ,
{"GITEA_SESSION_PROFILE_LOCK": "prgs-author"}):
with self.assertRaises(RuntimeError) as ctx:
mcp_server.verify_mutation_authority("prgs")
with patch("mcp_server.gitea_config.is_runtime_switching_enabled",
return_value=False):
with patch.dict(os.environ,
{"GITEA_SESSION_PROFILE_LOCK": "prgs-author"}):
with self.assertRaises(RuntimeError) as ctx:
mcp_server.verify_mutation_authority("prgs")
self.assertIn("side-channel override rejected", str(ctx.exception))
def test_foreign_pid_authority_is_not_trusted(self):
@@ -3098,12 +3184,7 @@ class TestIssueLocking(unittest.TestCase):
"""Test issue locking and PR gating constraints."""
def setUp(self):
self._lock_dir = tempfile.TemporaryDirectory()
env = {
**ISSUE_WRITE_ENV,
"GITEA_ISSUE_LOCK_DIR": self._lock_dir.name,
}
self._env_patcher = patch.dict(os.environ, env, clear=True)
self._env_patcher = patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True)
self._env_patcher.start()
self._dup_fetcher_patcher = patch(
"mcp_server.issue_duplicate_context_fetcher",
@@ -3114,13 +3195,8 @@ class TestIssueLocking(unittest.TestCase):
def tearDown(self):
self._dup_fetcher_patcher.stop()
self._env_patcher.stop()
self._lock_dir.cleanup()
def _create_pr_env(self) -> dict:
return {
**CREATE_PR_ENV,
"GITEA_ISSUE_LOCK_DIR": self._lock_dir.name,
}
if os.path.exists(ISSUE_LOCK_FILE):
os.remove(ISSUE_LOCK_FILE)
@patch(
"mcp_server.issue_lock_worktree.read_worktree_git_state",
@@ -3138,15 +3214,16 @@ class TestIssueLocking(unittest.TestCase):
self.assertIn("expires_at", res["work_lease"])
self.assertIn("last_heartbeat_at", res["work_lease"])
self.assertEqual(res["work_lease"]["claimant"]["profile"], "gitea-default")
self.assertIn("lock_file_path", res)
lock = issue_lock_store.read_lock_file(res["lock_file_path"])
self.assertTrue(os.path.exists(ISSUE_LOCK_FILE))
with open(ISSUE_LOCK_FILE, encoding="utf-8") as f:
lock = json.load(f)
self.assertIn("worktree_path", lock)
self.assertIn("work_lease", lock)
def test_lock_issue_mismatch_branch_fails(self):
with self.assertRaises(ValueError) as ctx:
gitea_lock_issue(issue_number=196, branch_name="feat/issue-195-mutations", remote="prgs")
self.assertIn("must match", str(ctx.exception))
self.assertIn("must contain locked issue pattern", str(ctx.exception))
@patch(
"mcp_server.issue_lock_worktree.read_worktree_git_state",
@@ -3201,13 +3278,90 @@ class TestIssueLocking(unittest.TestCase):
)
@patch("mcp_server.api_get_all")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_lock_issue_adopts_exact_own_branch(self, _auth, mock_api, _git_state):
branch = "feat/issue-196-mutations"
mock_api.return_value = [{"name": branch, "commit": {"id": "abc123"}}]
res = gitea_lock_issue(issue_number=196, branch_name=branch, remote="prgs")
def test_lock_issue_adopts_own_existing_branch(self, _auth, mock_api, _git_state):
# #442: the issue's own already-pushed branch is adoptable for recovery.
sha = "934688a71dd19d9b46369e73fb3ac356ee0cc211"
def _api(url, *a, **k):
if "/pulls" in url:
return []
if "/branches" in url:
return [{"name": "feat/issue-196-mutations", "commit": {"id": sha}}]
return []
mock_api.side_effect = _api
res = gitea_lock_issue(
issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs"
)
self.assertTrue(res["success"])
self.assertIn("adoption", res)
self.assertEqual(res["adoption"]["branch_head_commit"], "abc123")
proof = res["adoption"]
self.assertEqual(proof["issue_number"], 196)
self.assertEqual(proof["branch_name"], "feat/issue-196-mutations")
self.assertEqual(proof["branch_head_commit"], sha)
self.assertTrue(proof["no_existing_pr_proof"])
self.assertTrue(proof["no_competing_live_lock_proof"])
self.assertEqual(proof["lock_file_path"], ISSUE_LOCK_FILE)
self.assertIn("adopt", proof["adoption_reason"].lower())
self.assertTrue(os.path.exists(ISSUE_LOCK_FILE))
@patch(
"mcp_server.issue_lock_worktree.read_worktree_git_state",
return_value=_clean_master_git_state_for_lock(),
)
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_lock_issue_open_pr_blocks_adoption(self, _auth, _git_state):
# Even with the exact own branch present, an open PR blocks adoption.
self.mock_dup_fetcher.return_value = ([{
"number": 200,
"head": {"ref": "feat/issue-196-mutations"},
"title": "WIP",
"body": "",
}], ["feat/issue-196-mutations"], {"status": "not_claimed"})
with self.assertRaises(ValueError) as ctx:
gitea_lock_issue(
issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs"
)
self.assertIn("open PR #200 already covers issue", str(ctx.exception))
self.assertFalse(os.path.exists(ISSUE_LOCK_FILE))
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch(
"mcp_server.issue_lock_worktree.read_worktree_git_state",
return_value=_clean_master_git_state_for_lock(),
)
@patch("mcp_server.api_request")
@patch("mcp_server.api_get_all")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_pr_proceeds_after_adopted_lock(
self, _auth, mock_api, mock_req, _git_state, _role
):
# #442 recovery goal: after adopting the own branch, create_pr proceeds.
sha = "934688a71dd19d9b46369e73fb3ac356ee0cc211"
def _api(url, *a, **k):
if "/pulls" in url:
return []
if "/branches" in url:
return [{"name": "feat/issue-196-mutations", "commit": {"id": sha}}]
return []
mock_api.side_effect = _api
lock_res = gitea_lock_issue(
issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs"
)
self.assertIn("adoption", lock_res)
mock_req.return_value = {"number": 7, "html_url": "https://gitea.prgs.cc/pulls/7"}
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
pr = gitea_create_pr(
title="feat: parity gate Closes #196",
head="feat/issue-196-mutations",
base="master",
remote="prgs",
)
self.assertEqual(pr["number"], 7)
@patch(
"mcp_server.issue_lock_worktree.read_worktree_git_state",
@@ -3216,27 +3370,16 @@ class TestIssueLocking(unittest.TestCase):
@patch("mcp_server.api_get_all", return_value=[])
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_lock_issue_blocks_active_same_operation_lease(self, _auth, _api, _git_state):
prgs_repo = mcp_server.REMOTES["prgs"]["repo"]
issue_lock_store.save_lock_file(
issue_lock_store.lock_file_path(
remote="prgs",
org="Scaled-Tech-Consulting",
repo=prgs_repo,
issue_number=196,
),
{
with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
json.dump({
"issue_number": 196,
"branch_name": "feat/issue-196-other-work",
"remote": "prgs",
"org": "Scaled-Tech-Consulting",
"repo": prgs_repo,
"worktree_path": "/tmp/other-worktree",
"work_lease": {
"operation_type": "author_issue_work",
"expires_at": "2999-01-01T00:00:00Z",
},
},
)
}, f)
with self.assertRaises(RuntimeError) as ctx:
gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
self.assertIn("already has an active author_issue_work lease", str(ctx.exception))
@@ -3248,27 +3391,16 @@ class TestIssueLocking(unittest.TestCase):
@patch("mcp_server.api_get_all", return_value=[])
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_lock_issue_blocks_expired_same_operation_lease_for_recovery(self, _auth, _api, _git_state):
prgs_repo = mcp_server.REMOTES["prgs"]["repo"]
issue_lock_store.save_lock_file(
issue_lock_store.lock_file_path(
remote="prgs",
org="Scaled-Tech-Consulting",
repo=prgs_repo,
issue_number=196,
),
{
with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
json.dump({
"issue_number": 196,
"branch_name": "feat/issue-196-other-work",
"remote": "prgs",
"org": "Scaled-Tech-Consulting",
"repo": prgs_repo,
"worktree_path": "/tmp/other-worktree",
"work_lease": {
"operation_type": "author_issue_work",
"expires_at": "2000-01-01T00:00:00Z",
},
},
)
}, f)
with self.assertRaises(RuntimeError) as ctx:
gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
self.assertIn("Recovery review is required before takeover", str(ctx.exception))
@@ -3335,7 +3467,9 @@ class TestIssueLocking(unittest.TestCase):
return_value=(True, []))
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_pr_missing_lock_fails(self, _auth, _role):
with patch.dict(os.environ, self._create_pr_env(), clear=True):
if os.path.exists(ISSUE_LOCK_FILE):
os.remove(ISSUE_LOCK_FILE)
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
with self.assertRaises(RuntimeError) as ctx:
gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-mutations", remote="prgs")
self.assertIn("Issue lock is missing", str(ctx.exception))
@@ -3344,64 +3478,37 @@ class TestIssueLocking(unittest.TestCase):
return_value=(True, []))
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_pr_branch_mismatch_fails(self, _auth, _role):
worktree = os.path.realpath(os.getcwd())
_bind_test_lock(
issue_number=196,
branch_name="feat/issue-196-mutations",
remote="prgs",
worktree_path=worktree,
)
with patch.dict(os.environ, self._create_pr_env(), clear=True):
with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
json.dump(_sample_issue_lock(
issue_number=196, branch_name="feat/issue-196-mutations"), f)
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
with self.assertRaises(ValueError) as ctx:
gitea_create_pr(
title="feat: X Closes #196",
head="feat/issue-196-different",
remote="prgs",
worktree_path=worktree,
)
gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-different", remote="prgs")
self.assertIn("does not match locked branch", str(ctx.exception))
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_pr_forbidden_terms_fails(self, _auth, _role):
worktree = os.path.realpath(os.getcwd())
_bind_test_lock(
issue_number=196,
branch_name="feat/issue-196-mutations",
remote="prgs",
worktree_path=worktree,
)
with patch.dict(os.environ, self._create_pr_env(), clear=True):
with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
json.dump(_sample_issue_lock(
issue_number=196, branch_name="feat/issue-196-mutations"), f)
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
for term in ("equivalent to #196", "related to #196", "same as #196"):
with self.assertRaises(ValueError) as ctx:
gitea_create_pr(
title=f"feat: X {term}",
head="feat/issue-196-mutations",
remote="prgs",
worktree_path=worktree,
)
gitea_create_pr(title=f"feat: X {term}", head="feat/issue-196-mutations", remote="prgs")
self.assertIn("contains forbidden term", str(ctx.exception))
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_pr_missing_closes_ref_fails(self, _auth, _role):
worktree = os.path.realpath(os.getcwd())
_bind_test_lock(
issue_number=196,
branch_name="feat/issue-196-mutations",
remote="prgs",
worktree_path=worktree,
)
with patch.dict(os.environ, self._create_pr_env(), clear=True):
with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
json.dump(_sample_issue_lock(
issue_number=196, branch_name="feat/issue-196-mutations"), f)
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
with self.assertRaises(ValueError) as ctx:
gitea_create_pr(
title="feat: X refs #196",
head="feat/issue-196-mutations",
remote="prgs",
worktree_path=worktree,
)
gitea_create_pr(title="feat: X refs #196", head="feat/issue-196-mutations", remote="prgs")
self.assertIn("must contain 'Closes #196' or 'Fixes #196' exactly", str(ctx.exception))
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
@@ -3409,13 +3516,13 @@ class TestIssueLocking(unittest.TestCase):
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_pr_worktree_mismatch_fails(self, _auth, _role):
scratch = os.path.realpath("/tmp/gitea-tools-author-scratch/issue-249-pr")
_bind_test_lock(
issue_number=249,
branch_name="feat/issue-249-issue-lock-scratch-worktree",
worktree_path=scratch,
remote="prgs",
)
with patch.dict(os.environ, self._create_pr_env(), clear=True):
with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
json.dump(_sample_issue_lock(
issue_number=249,
branch_name="feat/issue-249-issue-lock-scratch-worktree",
worktree_path=scratch,
), f)
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
with self.assertRaises(ValueError) as ctx:
gitea_create_pr(
title="feat: lock scratch worktree Closes #249",
@@ -3429,33 +3536,21 @@ class TestIssueLocking(unittest.TestCase):
return_value=(True, []))
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_pr_manual_lock_seed_blocked(self, _auth, _role):
worktree = os.path.realpath(os.getcwd())
env = self._create_pr_env()
with patch.dict(os.environ, env, clear=True):
issue_lock_store.save_lock_file(
issue_lock_store.lock_file_path(
remote="prgs",
org="Scaled-Tech-Consulting",
repo=mcp_server.REMOTES["prgs"]["repo"],
issue_number=447,
lock_dir=self._lock_dir.name,
),
with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
json.dump(
_sample_issue_lock(
issue_number=447,
branch_name="feat/issue-447-lock-provenance",
remote="prgs",
org="Scaled-Tech-Consulting",
repo=mcp_server.REMOTES["prgs"]["repo"],
worktree_path=worktree,
lock_provenance=None,
),
f,
)
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
with self.assertRaises(RuntimeError) as ctx:
gitea_create_pr(
title="feat: lock provenance Closes #447",
head="feat/issue-447-lock-provenance",
remote="prgs",
worktree_path=worktree,
)
self.assertIn("lock provenance", str(ctx.exception).lower())
@@ -3466,13 +3561,13 @@ class TestIssueLocking(unittest.TestCase):
def test_create_pr_honors_scratch_worktree_lock(self, _auth, _role, mock_api):
scratch = os.path.realpath("/tmp/gitea-tools-author-scratch/issue-249-e2e")
mock_api.return_value = {"number": 250, "html_url": "https://example/pr/250"}
_bind_test_lock(
issue_number=249,
branch_name="feat/issue-249-issue-lock-scratch-worktree",
worktree_path=scratch,
remote="prgs",
)
with patch.dict(os.environ, self._create_pr_env(), clear=True):
with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
json.dump(_sample_issue_lock(
issue_number=249,
branch_name="feat/issue-249-issue-lock-scratch-worktree",
worktree_path=scratch,
), f)
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
res = gitea_create_pr(
title="feat: issue-lock scratch worktree Closes #249",
head="feat/issue-249-issue-lock-scratch-worktree",
+16 -3
View File
@@ -156,9 +156,22 @@ class TestPRQueueInventory(unittest.TestCase):
]
from mcp_server import init_review_decision_lock, gitea_mark_final_review_decision
init_review_decision_lock("prgs", "review_pr")
gitea_mark_final_review_decision(1, "approve", remote="prgs")
result = gitea_review_pr(pr_number=1, event="APPROVE", remote="prgs", final_review_decision_ready=True)
from tests.test_mcp_server import _install_owned_reviewer_lease
import reviewer_pr_lease
with patch("mcp_server._authenticated_username", return_value="reviewer1"):
init_review_decision_lock("prgs", "review_pr")
gitea_mark_final_review_decision(1, "approve", remote="prgs")
lease_patch = _install_owned_reviewer_lease(
1, head_sha="abc1", session_id="inventory-review-lease",
)
lease_patch.start()
self.addCleanup(lease_patch.stop)
self.addCleanup(reviewer_pr_lease.clear_session_lease)
result = gitea_review_pr(
pr_number=1, event="APPROVE", remote="prgs",
final_review_decision_ready=True,
)
self.assertTrue(result["success"])
self.assertIn("=== PR Queue Inventory ===", result["message"])
self.assertIn("Repository:", result["message"])
+193
View File
@@ -0,0 +1,193 @@
"""Tests for per-PR reviewer leases (#407)."""
import sys
import unittest
from datetime import datetime, timedelta, timezone
from unittest.mock import patch
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
import reviewer_pr_lease as leases
def _lease_comment(
pr_number: int,
session_id: str,
*,
phase: str = "claimed",
minutes_ago: int = 0,
candidate_head: str = "a" * 40,
) -> dict:
now = datetime.now(timezone.utc) - timedelta(minutes=minutes_ago)
body = leases.format_lease_body(
repo="Scaled-Tech-Consulting/Gitea-Tools",
pr_number=pr_number,
issue_number=295,
reviewer_identity="rev1",
profile="prgs-reviewer",
session_id=session_id,
worktree="branches/review-pr382",
phase=phase,
candidate_head=candidate_head,
target_branch="master",
target_branch_sha="b" * 40,
last_activity=now,
)
return {"id": 1, "body": body, "user": {"login": "rev1"}}
class TestReviewerLeaseAcquire(unittest.TestCase):
def setUp(self):
leases.clear_session_lease()
def test_two_reviewers_cannot_lease_same_pr(self):
comments = [_lease_comment(382, "session-a")]
result = leases.assess_acquire_lease(
comments,
pr_number=382,
reviewer_identity="rev2",
profile="prgs-reviewer",
session_id="session-b",
repo="Scaled-Tech-Consulting/Gitea-Tools",
issue_number=295,
worktree="branches/review-pr382-b",
candidate_head="c" * 40,
target_branch="master",
target_branch_sha="d" * 40,
)
self.assertFalse(result["acquire_allowed"])
self.assertTrue(any("already has active" in r for r in result["reasons"]))
def test_two_reviewers_can_lease_different_prs(self):
comments = [_lease_comment(382, "session-a")]
result = leases.assess_acquire_lease(
comments,
pr_number=383,
reviewer_identity="rev2",
profile="prgs-reviewer",
session_id="session-b",
repo="Scaled-Tech-Consulting/Gitea-Tools",
issue_number=296,
worktree="branches/review-pr383",
candidate_head="c" * 40,
target_branch="master",
target_branch_sha="d" * 40,
)
self.assertTrue(result["acquire_allowed"])
self.assertIsNotNone(result["lease_body"])
class TestReviewerLeaseFreshness(unittest.TestCase):
def test_stale_warning_after_30_minutes(self):
lease = leases.parse_lease_comment(
_lease_comment(382, "session-a", minutes_ago=35)["body"]
)
self.assertEqual(
leases.classify_lease_freshness(lease),
"stale_warning",
)
def test_reclaimable_after_60_minutes(self):
lease = leases.parse_lease_comment(
_lease_comment(382, "session-a", minutes_ago=65)["body"]
)
self.assertEqual(
leases.classify_lease_freshness(lease),
"reclaimable",
)
class TestReviewerLeaseMutationGate(unittest.TestCase):
def setUp(self):
leases.clear_session_lease()
def test_reviewer_without_lease_cannot_mutate(self):
head = "f" * 40
comments = [_lease_comment(382, "other-session", candidate_head=head)]
result = leases.assess_mutation_lease_gate(
pr_number=382,
comments=comments,
reviewer_identity="rev1",
session_id="my-session",
mutation="approve",
live_head_sha=head,
pinned_head_sha=head,
)
self.assertTrue(result["block"])
def test_owned_lease_allows_mutation(self):
head = "f" * 40
comments = [_lease_comment(382, "my-session", candidate_head=head)]
leases.record_session_lease({
"pr_number": 382,
"session_id": "my-session",
"candidate_head": head,
"target_branch": "master",
})
result = leases.assess_mutation_lease_gate(
pr_number=382,
comments=comments,
reviewer_identity="rev1",
session_id="my-session",
mutation="approve",
live_head_sha=head,
pinned_head_sha=head,
)
self.assertFalse(result["block"])
def test_head_change_invalidates_lease(self):
reviewed = "f" * 40
live = "e" * 40
comments = [_lease_comment(382, "my-session", candidate_head=reviewed)]
leases.record_session_lease({
"pr_number": 382,
"session_id": "my-session",
"candidate_head": reviewed,
})
result = leases.assess_mutation_lease_gate(
pr_number=382,
comments=comments,
reviewer_identity="rev1",
session_id="my-session",
mutation="merge",
live_head_sha=live,
pinned_head_sha=reviewed,
)
self.assertTrue(result["block"])
self.assertTrue(any("head" in r.lower() for r in result["reasons"]))
class TestReviewerLeaseMcpGate(unittest.TestCase):
def setUp(self):
leases.clear_session_lease()
patch("mcp_server.verify_preflight_purity").start()
patch("gitea_audit.audit_enabled", return_value=False).start()
mcp_server = __import__("mcp_server")
mcp_server._IDENTITY_CACHE.clear()
mcp_server.init_review_decision_lock("prgs", "review_pr")
mcp_server.record_preflight_check("whoami")
mcp_server.record_preflight_check("capability", "reviewer")
def tearDown(self):
patch.stopall()
leases.clear_session_lease()
def test_reviewer_pr_lease_gate_helper_blocks_without_session(self):
import mcp_server
head = "a" * 40
with patch("mcp_server._fetch_pr_comments", return_value=[]):
reasons = mcp_server._reviewer_pr_lease_gate(
pr_number=382,
remote="prgs",
host=None,
org=None,
repo=None,
mutation="approve",
live_head_sha=head,
pinned_head_sha=head,
)
self.assertTrue(any("lease" in r.lower() for r in reasons))
if __name__ == "__main__":
unittest.main()
+9 -26
View File
@@ -20,50 +20,33 @@ def run(script, *args):
branch = arg
break
lock_dir_ctx = None
extra_env = os.environ.copy()
lock_file = Path("/tmp/gitea_issue_lock.json")
created_lock = False
if script == "worktree-start" and branch:
import re
import tempfile
import issue_lock_store
import json
m = re.search(r"issue-(\d+)", branch)
if not m:
m = re.search(r"pr-(\d+)", branch)
issue_num = int(m.group(1)) if m else 999
lock_dir_ctx = tempfile.TemporaryDirectory()
extra_env["GITEA_ISSUE_LOCK_DIR"] = lock_dir_ctx.name
record = {
lock_file.write_text(json.dumps({
"issue_number": issue_num,
"branch_name": branch,
"remote": "prgs",
"org": "Scaled-Tech-Consulting",
"repo": "Gitea-Tools",
"worktree_path": "/tmp/test-worktree",
"work_lease": {
"operation_type": "author_issue_work",
"expires_at": "2999-01-01T00:00:00Z",
},
}
path = issue_lock_store.lock_file_path(
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
issue_number=issue_num,
lock_dir=lock_dir_ctx.name,
)
issue_lock_store.save_lock_file(path, record)
"repo": "Gitea-Tools"
}), encoding="utf-8")
created_lock = True
try:
proc = subprocess.run(
["bash", str(SCRIPTS / script), *args],
capture_output=True, text=True, cwd=str(REPO),
env=extra_env,
)
return proc.returncode, proc.stdout, proc.stderr
finally:
if lock_dir_ctx is not None:
lock_dir_ctx.cleanup()
if created_lock and lock_file.exists():
lock_file.unlink()
class TestWorktreeStart(unittest.TestCase):