Compare commits

..
Author SHA1 Message Date
sysadmin d67b2f54eb resolve conflicts for PR #467 2026-07-07 17:25:13 -04:00
sysadminandClaude Opus 4.8 a2cabb64b1 feat: non-destructive lock recovery for pushed branches (Closes #440)
Adds structured issue-branch ownership parsing, wires it into lock adoption
and gitea_lock_issue validation, documents the restart recovery workflow, and
adds regression tests for adoption, durable create_pr resolution, and open-PR
blocking without remote branch deletion.

Built on keyed persistent lock store and own-branch adoption (#443 / #442).

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-07 17:22:24 -04:00
sysadminandClaude Opus 4.8 69e9e25fcf feat: replace global issue lock with keyed persistent store (Closes #443)
Store per remote/org/repo/issue locks under GITEA_ISSUE_LOCK_DIR with
atomic writes and per-session binding. Integrate own-branch adoption for
lock recovery, update worktree-start and cleanup reconcile, and add tests
documenting the ban on manual global lock seeding.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-07 17:21:53 -04:00
20 changed files with 408 additions and 1819 deletions
-127
View File
@@ -7,11 +7,8 @@ project's ``branches/`` directory, never from the stable control checkout.
from __future__ import annotations
import os
import subprocess
BASE_BRANCHES = frozenset({"master", "main", "dev"})
ACTIVE_WORKTREE_ENV = "GITEA_ACTIVE_WORKTREE"
AUTHOR_WORKTREE_ENV = "GITEA_AUTHOR_WORKTREE"
def _normalize_path(path: str) -> str:
@@ -51,130 +48,6 @@ def resolve_mutation_workspace(
return os.path.realpath(project_root)
def _realpath_git_common_dir(workspace_path: str, common_dir: str) -> str:
"""Resolve ``git rev-parse --git-common-dir`` relative to *workspace_path*."""
raw = (common_dir or "").strip()
if not raw:
return raw
if os.path.isabs(raw):
return os.path.realpath(raw)
return os.path.realpath(os.path.join(workspace_path, raw))
def resolve_canonical_repo_root(workspace_path: str, fallback_project_root: str) -> str:
"""Return the stable repository root for *workspace_path* via git metadata (#460)."""
path = (workspace_path or "").strip()
fallback = os.path.realpath(fallback_project_root)
if not path:
return fallback
try:
res = subprocess.run(
["git", "-C", path, "rev-parse", "--git-common-dir"],
capture_output=True,
text=True,
check=True,
)
common = _realpath_git_common_dir(path, res.stdout)
except Exception:
return fallback
if common.endswith(f"{os.sep}.git"):
return os.path.dirname(common)
if os.path.basename(common) == ".git":
return os.path.dirname(common)
return fallback
def resolve_author_mutation_context(
worktree_path: str | None,
process_project_root: str,
*,
active_worktree_env: str | None = None,
author_worktree_env: str | None = None,
) -> dict:
"""Shared workspace resolution for runtime_context and mutation guards (#460)."""
workspace = resolve_mutation_workspace(
worktree_path,
process_project_root,
active_worktree_env=active_worktree_env,
author_worktree_env=author_worktree_env,
)
process_root = os.path.realpath(process_project_root)
# Canonical repository identity comes from the MCP process checkout (#460),
# not from the declared task workspace being validated.
canonical_root = resolve_canonical_repo_root(process_root, process_root)
return {
"workspace_path": workspace,
"process_project_root": process_root,
"canonical_repo_root": canonical_root,
"roots_aligned": canonical_root == process_root,
}
def assess_workspace_repo_membership(
*,
workspace_path: str,
canonical_repo_root: str,
) -> dict:
"""Fail closed when *workspace_path* is not a git worktree of *canonical_repo_root*."""
workspace = os.path.realpath(workspace_path)
root = os.path.realpath(canonical_repo_root)
reasons: list[str] = []
if not os.path.exists(workspace):
reasons.append(f"worktree path '{workspace}' does not exist")
return _membership_assessment(False, reasons, workspace, root, None)
if not os.path.isdir(workspace):
reasons.append(f"worktree path '{workspace}' is not a directory")
return _membership_assessment(False, reasons, workspace, root, None)
try:
res = subprocess.run(
["git", "-C", workspace, "rev-parse", "--git-common-dir"],
capture_output=True,
text=True,
check=True,
)
common_dir = _realpath_git_common_dir(workspace, res.stdout)
except Exception:
reasons.append(f"worktree '{workspace}' is not a valid git repository")
return _membership_assessment(False, reasons, workspace, root, None)
expected_dir = os.path.realpath(os.path.join(root, ".git"))
if common_dir != expected_dir:
reasons.append(
f"worktree '{workspace}' does not belong to the target repository '{root}'"
)
return _membership_assessment(not reasons, reasons, workspace, root, common_dir)
def _membership_assessment(
proven: bool,
reasons: list[str],
workspace: str,
root: str,
common_dir: str | None,
) -> dict:
return {
"proven": proven,
"block": not proven,
"reasons": reasons,
"workspace_path": workspace,
"canonical_repo_root": root,
"git_common_dir": common_dir,
}
def format_workspace_repo_membership_error(assessment: dict) -> str:
workspace = assessment.get("workspace_path") or "(unknown)"
root = assessment.get("canonical_repo_root") or "(unknown)"
reasons = "; ".join(assessment.get("reasons") or ["unknown repository membership violation"])
return (
f"Branches-only mutation guard (#274): {reasons} (fail closed). "
f"canonical repository root: {root}; workspace: {workspace}."
)
def assess_author_mutation_worktree(
*,
workspace_path: str,
+22
View File
@@ -308,6 +308,28 @@ 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
+57 -367
View File
@@ -190,22 +190,14 @@ def _ensure_process_start_porcelain() -> str:
def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str:
"""Resolve the workspace root inspected by pre-flight guards."""
return author_mutation_worktree.resolve_mutation_workspace(
worktree_path,
PROJECT_ROOT,
active_worktree_env=os.environ.get(ACTIVE_WORKTREE_ENV),
author_worktree_env=os.environ.get(AUTHOR_WORKTREE_ENV),
)
def _resolve_author_mutation_context(worktree_path: str | None = None) -> dict:
"""Canonical workspace + repository root for runtime_context and guards (#460)."""
return author_mutation_worktree.resolve_author_mutation_context(
worktree_path,
PROJECT_ROOT,
active_worktree_env=os.environ.get(ACTIVE_WORKTREE_ENV),
author_worktree_env=os.environ.get(AUTHOR_WORKTREE_ENV),
)
path = (worktree_path or "").strip()
if not path:
path = (os.environ.get(ACTIVE_WORKTREE_ENV) or "").strip()
if not path:
path = (os.environ.get(AUTHOR_WORKTREE_ENV) or "").strip()
if not path:
path = PROJECT_ROOT
return os.path.realpath(os.path.abspath(path))
def _get_git_root(path: str) -> str | None:
@@ -275,31 +267,21 @@ def _format_preflight_files(files: list[str]) -> str:
def _preflight_workspace_details(worktree_path: str | None, dirty_files: list[str]) -> dict:
ctx = _resolve_author_mutation_context(worktree_path)
workspace = ctx["workspace_path"]
workspace = _resolve_preflight_workspace_path(worktree_path)
inspected_root = _get_git_root(workspace)
process_root = ctx["process_project_root"]
canonical_root = ctx["canonical_repo_root"]
control_root = os.path.realpath(PROJECT_ROOT)
active_root = os.path.realpath(inspected_root or workspace)
if active_root == canonical_root:
if active_root == control_root:
dirty_scope = "control checkout"
else:
dirty_scope = "active task workspace"
details = {
"mcp_server_process_root": process_root,
"canonical_repository_root": canonical_root,
return {
"mcp_server_process_root": control_root,
"active_task_workspace_root": active_root,
"inspected_git_root": inspected_root,
"dirty_files": list(dirty_files),
"dirty_scope": dirty_scope,
"workspace_roots_aligned": ctx["roots_aligned"],
}
if not ctx["roots_aligned"]:
details["workspace_root_mismatch"] = (
"runtime_context and mutation guard use canonical repository root "
f"'{canonical_root}' instead of MCP process root '{process_root}'"
)
return details
def _format_preflight_workspace_details(details: dict) -> str:
@@ -420,12 +402,16 @@ def _enforce_branches_only_author_mutation(worktree_path: str | None = None) ->
"""#274: author mutations must run from a branches/ session worktree."""
if _preflight_resolved_role == "reviewer":
return
ctx = _resolve_author_mutation_context(worktree_path)
workspace = ctx["workspace_path"]
workspace = author_mutation_worktree.resolve_mutation_workspace(
worktree_path,
PROJECT_ROOT,
active_worktree_env=os.environ.get(ACTIVE_WORKTREE_ENV),
author_worktree_env=os.environ.get(AUTHOR_WORKTREE_ENV),
)
git_state = issue_lock_worktree.read_worktree_git_state(workspace)
assessment = author_mutation_worktree.assess_author_mutation_worktree(
workspace_path=workspace,
project_root=ctx["canonical_repo_root"],
project_root=PROJECT_ROOT,
current_branch=git_state.get("current_branch"),
)
if assessment["block"]:
@@ -454,23 +440,43 @@ def verify_preflight_purity(remote: str | None = None, worktree_path: str | None
"Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)"
)
ctx = _resolve_author_mutation_context(worktree_path)
workspace = ctx["workspace_path"]
canonical_root = ctx["canonical_repo_root"]
process_root = ctx["process_project_root"]
workspace = author_mutation_worktree.resolve_mutation_workspace(
worktree_path,
PROJECT_ROOT,
active_worktree_env=os.environ.get(ACTIVE_WORKTREE_ENV),
author_worktree_env=os.environ.get(AUTHOR_WORKTREE_ENV),
)
real_workspace = os.path.realpath(workspace)
real_root = os.path.realpath(PROJECT_ROOT)
if real_workspace != process_root:
if real_workspace != real_root:
if not _preflight_in_test_mode():
membership = author_mutation_worktree.assess_workspace_repo_membership(
workspace_path=workspace,
canonical_repo_root=canonical_root,
)
if membership["block"]:
if not os.path.exists(real_workspace):
raise RuntimeError(
author_mutation_worktree.format_workspace_repo_membership_error(
membership
f"Branches-only mutation guard (#274): worktree path '{workspace}' does not exist (fail closed)"
)
if not os.path.isdir(real_workspace):
raise RuntimeError(
f"Branches-only mutation guard (#274): worktree path '{workspace}' is not a directory (fail closed)"
)
try:
res = subprocess.run(
["git", "-C", real_workspace, "rev-parse", "--git-common-dir"],
capture_output=True,
text=True,
check=True,
)
common_dir = os.path.realpath(res.stdout.strip())
expected_dir = os.path.realpath(os.path.join(real_root, ".git"))
if common_dir != expected_dir:
raise RuntimeError(
f"Branches-only mutation guard (#274): worktree '{workspace}' does not belong to the target repository '{PROJECT_ROOT}' (fail closed)"
)
except Exception as e:
if isinstance(e, RuntimeError):
raise e
raise RuntimeError(
f"Branches-only mutation guard (#274): worktree '{workspace}' is not a valid git repository (fail closed)"
)
dirty_files = sorted(_parse_porcelain_entries(_get_workspace_porcelain(workspace)))
@@ -535,11 +541,11 @@ 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 reviewer_pr_lease # noqa: E402
import merged_cleanup_reconcile # noqa: E402
import reconciler_profile # noqa: E402
import reconciliation_workflow # noqa: E402
@@ -1281,11 +1287,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 branch name includes issue number
expected_pattern = f"issue-{issue_number}"
if expected_pattern not in branch_name:
# 1. Enforce canonical issue branch ownership (#440)
if not issue_branch_ownership.branch_belongs_to_issue(branch_name, issue_number):
raise ValueError(
f"Branch name '{branch_name}' must contain locked issue pattern '{expected_pattern}' (fail closed)"
f"Branch name '{branch_name}' must match "
f"(fix|feat|docs|chore)/issue-{issue_number}-<desc> (fail closed)"
)
blocked = _profile_permission_block(
@@ -1382,19 +1388,6 @@ def gitea_lock_issue(
}
lock_file_path = _save_issue_lock(data)
lock_record = issue_lock_store.read_lock_file(lock_file_path) or data
freshness = issue_lock_store.assess_lock_freshness(lock_record)
competing = [
entry
for entry in issue_lock_store.list_live_locks()
if entry.get("issue_number") != issue_number
]
lock_proof = issue_lock_store.format_lock_proof(
lock_record,
freshness=freshness,
competing_live_locks=competing,
released=False,
)
agent_artifacts = agent_temp_artifacts.find_agent_temp_artifacts_from_porcelain(
git_state.get("porcelain_status") or ""
@@ -1410,8 +1403,6 @@ def gitea_lock_issue(
"worktree_path": resolved_worktree,
"work_lease": work_lease,
"lock_file_path": lock_file_path,
"lock_freshness": freshness,
"lock_proof": lock_proof,
}
if adoption["adopt"]:
result["adoption"] = issue_lock_adoption.build_adoption_proof(
@@ -1544,15 +1535,6 @@ def gitea_create_pr(
f"PR head branch '{head}' does not match locked branch '{locked_branch}' (fail closed)"
)
ownership = issue_lock_store.verify_lock_for_mutation(
lock_data,
issue_number=locked_issue,
branch_name=head,
worktree_path=worktree_path,
)
if ownership["block"]:
raise ValueError(ownership["reasons"][0])
# Check for forbidden terms anywhere in title/body
forbidden_terms = ["equivalent", "related", "same as"]
text_to_check = f"{title} {body}".lower()
@@ -2123,7 +2105,6 @@ 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,
@@ -2566,20 +2547,6 @@ 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:
@@ -3373,19 +3340,6 @@ 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:
@@ -4685,261 +4639,6 @@ 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,
@@ -6715,15 +6414,6 @@ def gitea_reconcile_issue_claims(
heartbeat_lease_minutes=heartbeat_lease_minutes,
reclaim_after_minutes=reclaim_after_minutes,
)
live_locks = issue_lock_store.list_live_locks()
inventory["live_issue_locks"] = live_locks
inventory["live_issue_lock_numbers"] = sorted(
{
int(entry["issue_number"])
for entry in live_locks
if entry.get("issue_number") is not None
}
)
inventory["cleanup_plan"] = issue_claim_heartbeat.build_cleanup_plan(inventory)
inventory["success"] = True
inventory["performed"] = False
+28
View File
@@ -0,0 +1,28 @@
"""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
+4 -3
View File
@@ -1,4 +1,4 @@
"""Own-branch lock adoption / recovery for ``gitea_lock_issue`` (#442 / #443).
"""Own-branch lock adoption / recovery for ``gitea_lock_issue`` (#440 / #442 / #443).
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,6 +14,8 @@ this module additionally records whether they passed for proof purposes.
from __future__ import annotations
import issue_branch_ownership
ADOPT = "adopt_existing_branch"
BLOCK_COMPETING = "block_competing_branch"
NO_MATCH = "no_matching_branch"
@@ -40,13 +42,12 @@ def assess_own_branch_adoption(
existing_branches,
) -> dict:
"""Decide whether an existing matching branch is adoptable."""
marker = f"issue-{issue_number}"
requested = (requested_branch or "").strip()
matches: list[tuple[str, str | None]] = []
for entry in existing_branches or []:
name = _branch_name(entry).strip()
if marker in name:
if issue_branch_ownership.branch_belongs_to_issue(name, issue_number):
matches.append((name, _branch_sha(entry)))
competing = sorted({name for name, _ in matches if name != requested})
+14 -241
View File
@@ -1,21 +1,18 @@
"""Keyed, persistent issue-lock storage (#443) with flock hardening (#438).
"""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. Acquisition is serialized per issue with ``fcntl.flock``.
other.
"""
from __future__ import annotations
import errno
import fcntl
import json
import os
import re
import tempfile
from contextlib import contextmanager
from datetime import datetime, timedelta, timezone
from typing import Any
@@ -27,10 +24,6 @@ AUTHOR_ISSUE_WORK_LEASE = "author_issue_work"
_SAFE_SEGMENT_RE = re.compile(r"[^A-Za-z0-9._+-]+")
class LockContentionError(RuntimeError):
"""Raised when an exclusive per-issue lock cannot be acquired."""
def default_lock_dir() -> str:
raw = (os.environ.get(LOCK_DIR_ENV) or DEFAULT_LOCK_DIR).strip()
return raw or DEFAULT_LOCK_DIR
@@ -79,41 +72,6 @@ def _ensure_lock_dir(lock_dir: str | None = None) -> str:
return root
def flock_path(json_path: str) -> str:
return f"{json_path}.lock"
def is_process_alive(pid: int | None) -> bool:
if not pid or pid <= 0:
return False
try:
os.kill(int(pid), 0)
return True
except OSError as exc:
return exc.errno != errno.ESRCH
except (TypeError, ValueError):
return False
@contextmanager
def _exclusive_file_lock(lock_path: str):
os.makedirs(os.path.dirname(lock_path) or ".", exist_ok=True)
fd = os.open(lock_path, os.O_CREAT | os.O_RDWR, 0o600)
try:
try:
fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
except BlockingIOError as exc:
raise LockContentionError(
f"could not acquire exclusive lock on '{lock_path}'"
) from exc
yield fd
finally:
try:
fcntl.flock(fd, fcntl.LOCK_UN)
finally:
os.close(fd)
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):
@@ -168,7 +126,7 @@ def bind_session_lock(lock_data: dict[str, Any], lock_dir: str | None = None) ->
record = dict(lock_data)
record["lock_file_path"] = path
record["session_pid"] = os.getpid()
record.setdefault("pid", os.getpid())
save_lock_file(path, record)
pointer = {
"pid": os.getpid(),
@@ -179,32 +137,7 @@ def bind_session_lock(lock_data: dict[str, Any], lock_dir: str | None = None) ->
"org": org,
"repo": repo,
}
sentinel = flock_path(path)
try:
with _exclusive_file_lock(sentinel):
existing = read_lock_file(path)
overwrite_block = assess_foreign_lock_overwrite(existing, record)
if overwrite_block:
raise RuntimeError(overwrite_block)
lease_block = assess_same_issue_lease_conflict(
existing,
issue_number=issue_number,
branch_name=str(record.get("branch_name") or ""),
worktree_path=str(record.get("worktree_path") or ""),
)
if lease_block:
raise RuntimeError(lease_block)
save_lock_file(path, record)
save_lock_file(session_pointer_path(root), pointer)
except LockContentionError as exc:
competing = read_lock_file(path)
if competing:
owner_pid = competing.get("session_pid") or competing.get("pid")
raise RuntimeError(
f"Issue #{issue_number} lock contention: {exc}; competing owner "
f"pid={owner_pid} (fail closed)"
) from exc
raise RuntimeError(f"Issue #{issue_number} lock contention: {exc} (fail closed)") from exc
save_lock_file(session_pointer_path(root), pointer)
return path
@@ -308,62 +241,15 @@ def is_lease_expired(lock: dict[str, Any] | None, *, now: datetime | None = None
def is_lease_live(lock: dict[str, Any] | None, *, now: datetime | None = None) -> bool:
return assess_lock_freshness(lock, now=now)["live"]
def assess_lock_freshness(
lock_data: dict[str, Any] | None,
*,
now: datetime | None = None,
) -> dict[str, Any]:
"""Classify a lock as live, expired, stale, or absent."""
current = _lease_now(now)
if not lock_data:
return {
"status": "absent",
"live": False,
"stale": False,
"reason": "no lock record",
}
expires_at = lease_expires_at(lock_data)
lease = lock_data.get("work_lease")
heartbeat_at = _parse_lease_timestamp(lock_data.get("last_heartbeat_at"))
if heartbeat_at is None and isinstance(lease, dict):
heartbeat_at = _parse_lease_timestamp(lease.get("last_heartbeat_at"))
pid = lock_data.get("session_pid")
if pid is None:
pid = lock_data.get("pid")
pid_alive = is_process_alive(pid) if pid is not None else False
if expires_at and expires_at <= current:
return {
"status": "expired",
"live": False,
"stale": True,
"reason": f"lease expired at {expires_at.isoformat()}",
"pid_alive": pid_alive,
}
if pid is not None and not pid_alive:
return {
"status": "stale",
"live": False,
"stale": True,
"reason": f"owner pid {pid} is not alive",
"pid_alive": False,
}
return {
"status": "live",
"live": True,
"stale": False,
"reason": "lock heartbeat and lease are fresh",
"pid_alive": pid_alive,
"heartbeat_at": heartbeat_at.isoformat() if heartbeat_at else None,
"expires_at": expires_at.isoformat() if expires_at else None,
}
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:
@@ -496,117 +382,4 @@ def has_active_issue_lock(
continue
if is_lease_live(lock):
return True
return False
def verify_lock_for_mutation(
lock_data: dict[str, Any] | None,
*,
issue_number: int | None = None,
branch_name: str | None = None,
worktree_path: str | None = None,
) -> dict[str, Any]:
"""Re-check lock ownership immediately before a mutation (#438)."""
reasons: list[str] = []
if not lock_data:
return {"proven": False, "block": True, "reasons": ["issue lock is missing (fail closed)"]}
freshness = assess_lock_freshness(lock_data)
if not freshness["live"]:
reasons.append(f"issue lock is not live: {freshness['reason']} (fail closed)")
if issue_number is not None and lock_data.get("issue_number") != issue_number:
reasons.append(
f"issue lock targets #{lock_data.get('issue_number')}, expected #{issue_number} (fail closed)"
)
if branch_name is not None and lock_data.get("branch_name") != branch_name:
reasons.append(
f"issue lock branch '{lock_data.get('branch_name')}' does not match "
f"'{branch_name}' (fail closed)"
)
if worktree_path is not None:
locked = os.path.realpath(str(lock_data.get("worktree_path") or ""))
declared = os.path.realpath(worktree_path)
if locked != declared:
reasons.append(
f"issue lock worktree '{locked}' does not match declared '{declared}' (fail closed)"
)
return {
"proven": not reasons,
"block": bool(reasons),
"reasons": reasons,
"freshness": freshness,
"lock_proof": format_lock_proof(lock_data, freshness=freshness),
}
def list_live_locks(
*,
lock_dir: str | None = None,
now: datetime | None = None,
) -> list[dict[str, Any]]:
"""Return live per-issue locks for queue visibility."""
live: list[dict[str, Any]] = []
for path in iter_lock_files(lock_dir):
record = read_lock_file(path)
if not record:
continue
freshness = assess_lock_freshness(record, now=now)
if not freshness["live"]:
continue
live.append(
{
"issue_number": record.get("issue_number"),
"branch_name": record.get("branch_name"),
"remote": record.get("remote"),
"org": record.get("org"),
"repo": record.get("repo"),
"worktree_path": record.get("worktree_path"),
"pid": record.get("session_pid") or record.get("pid"),
"claimant": (
record.get("claimant")
or (record.get("work_lease") or {}).get("claimant")
),
"freshness": freshness,
"lock_path": record.get("lock_file_path") or path,
}
)
return live
def format_lock_proof(
lock_data: dict[str, Any] | None,
*,
freshness: dict[str, Any] | None = None,
competing_live_locks: list[dict[str, Any]] | None = None,
released: bool | None = None,
) -> str:
"""Canonical issue-lock proof string for final reports."""
if not lock_data:
return "issue lock proof: not acquired"
fresh = freshness or assess_lock_freshness(lock_data)
owner = lock_data.get("claimant") or {}
if not owner and isinstance(lock_data.get("work_lease"), dict):
owner = lock_data["work_lease"].get("claimant") or {}
parts = [
"issue lock proof:",
f"acquired issue #{lock_data.get('issue_number')}",
f"branch {lock_data.get('branch_name')}",
f"owner {owner.get('profile') or 'unknown'}",
f"pid {lock_data.get('session_pid') or lock_data.get('pid')}",
f"freshness {fresh.get('status')}",
]
if competing_live_locks is not None:
parts.append(
"no competing live lock"
if not competing_live_locks
else f"competing live locks {len(competing_live_locks)}"
)
if released is True:
parts.append("lock released")
elif released is False:
parts.append("lock retained")
return "; ".join(parts)
return False
-382
View File
@@ -1,382 +0,0 @@
"""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,
}
@@ -732,26 +732,6 @@ 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:
-1
View File
@@ -87,7 +87,6 @@ class TestIssueLockArtifactWarning(unittest.TestCase):
"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("issue_lock_worktree.read_worktree_git_state")
+3 -29
View File
@@ -286,21 +286,6 @@ 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}
@@ -359,22 +344,11 @@ 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")
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()
r = gitea_submit_pr_review(pr_number=8, action="approve",
body="LGTM", remote="prgs",
final_review_decision_ready=True)
self.assertTrue(r["performed"])
recs = self._records()
self.assertEqual(len(recs), 1)
+10 -22
View File
@@ -106,32 +106,20 @@ class TestCreateIssueWorkspaceGuard(unittest.TestCase):
@patch("gitea_mcp_server.issue_lock_worktree.read_worktree_git_state", return_value={"current_branch": "feat/issue-1"})
@patch("os.path.exists", return_value=True)
@patch("os.path.isdir", return_value=True)
@patch("author_mutation_worktree.subprocess.run")
@patch("subprocess.run")
def test_create_issue_wrong_repo_fails_closed(self, mock_run, mock_amw_run, mock_isdir, mock_exists, _git, _get_all, mock_api, _role, _ns, _prof, _auth):
def test_create_issue_wrong_repo_fails_closed(self, mock_run, mock_isdir, mock_exists, _git, _get_all, mock_api, _role, _ns, _prof, _auth):
# Mock subprocess.run for git --git-common-dir to return a different path
mock_res = MagicMock()
mock_res.stdout = "/Users/jasonwalker/Development/some-other-repo/.git\n"
mock_run.return_value = mock_res
wrong_repo_path = os.path.join(CONTROL_CHECKOUT_ROOT, "branches", "feat-issue-1")
def _subprocess_side_effect(cmd, *args, **kwargs):
mock_res = MagicMock(returncode=0)
if "--git-common-dir" in cmd:
cwd = cmd[cmd.index("-C") + 1] if "-C" in cmd else ""
if cwd == wrong_repo_path:
mock_res.stdout = "/Users/jasonwalker/Development/some-other-repo/.git\n"
else:
mock_res.stdout = f"{CONTROL_CHECKOUT_ROOT}/.git\n"
else:
mock_res.stdout = ""
return mock_res
mock_run.side_effect = _subprocess_side_effect
mock_amw_run.side_effect = _subprocess_side_effect
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
with patch("gitea_mcp_server._get_workspace_porcelain", return_value=""):
with self.assertRaises(RuntimeError) as ctx:
srv.gitea_create_issue(
title="Test issue", body="body", worktree_path=wrong_repo_path
)
with self.assertRaises(RuntimeError) as ctx:
srv.gitea_create_issue(
title="Test issue", body="body", worktree_path=wrong_repo_path
)
self.assertIn("does not belong to the target repository", str(ctx.exception))
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
+36
View File
@@ -0,0 +1,36 @@
"""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()
+9
View File
@@ -43,6 +43,15 @@ class TestAssessOwnBranchAdoption(unittest.TestCase):
)
self.assertEqual(result["outcome"], NO_MATCH)
def test_substring_only_branch_name_is_ignored(self):
result = assess_own_branch_adoption(
issue_number=420,
requested_branch=REQ,
existing_branches=[{"name": "feat/my-issue-420-backport"}],
)
self.assertEqual(result["outcome"], NO_MATCH)
self.assertFalse(result["block"])
class TestBuildAdoptionProof(unittest.TestCase):
def test_proof_has_required_fields(self):
+170
View File
@@ -0,0 +1,170 @@
"""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()
+1 -76
View File
@@ -1,9 +1,8 @@
"""Unit tests for keyed issue-lock storage (#443) and flock hardening (#438)."""
"""Unit tests for keyed issue-lock storage (#443)."""
import json
import os
import sys
import tempfile
import threading
import unittest
from datetime import datetime, timedelta, timezone
from pathlib import Path
@@ -177,80 +176,6 @@ class TestIssueLockStore(unittest.TestCase):
self.assertTrue(os.path.exists(path_b))
self.assertEqual(ils.read_lock_file(path_a)["issue_number"], 108)
def test_concurrent_bind_same_issue_only_one_wins(self):
barrier = threading.Barrier(2)
results: list[str | Exception] = []
def worker():
barrier.wait()
try:
ils.bind_session_lock(
_lock_record(worktree_path=f"/tmp/wt-{threading.get_ident()}")
)
results.append("ok")
except Exception as exc: # noqa: BLE001
results.append(exc)
threads = [threading.Thread(target=worker) for _ in range(2)]
for thread in threads:
thread.start()
for thread in threads:
thread.join()
successes = [item for item in results if item == "ok"]
failures = [item for item in results if isinstance(item, Exception)]
self.assertEqual(len(successes), 1)
self.assertEqual(len(failures), 1)
failure_text = str(failures[0]).lower()
self.assertTrue(
"active" in failure_text or "lock contention" in failure_text,
failures[0],
)
def test_verify_lock_for_mutation_blocks_stale_lock(self):
record = _lock_record(
work_lease=_lease("2000-01-01T00:00:00Z"),
)
record["pid"] = 999999
record["session_pid"] = 999999
result = ils.verify_lock_for_mutation(
record,
issue_number=420,
branch_name="feat/issue-420-server-code-parity",
worktree_path="/tmp/wt-420",
)
self.assertTrue(result["block"])
self.assertIn("not live", result["reasons"][0])
def test_list_live_locks_excludes_stale_records(self):
live_path = ils.lock_file_path(
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
issue_number=420,
)
ils.save_lock_file(
live_path,
_lock_record(worktree_path="/tmp/wt-420"),
)
stale_path = ils.lock_file_path(
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
issue_number=440,
)
ils.save_lock_file(
stale_path,
_lock_record(
issue_number=440,
branch_name="feat/issue-440-recovery",
work_lease=_lease("2000-01-01T00:00:00Z"),
worktree_path="/tmp/wt-440",
),
)
live = ils.list_live_locks(lock_dir=self.lock_dir)
self.assertEqual([entry["issue_number"] for entry in live], [420])
if __name__ == "__main__":
unittest.main()
+21 -42
View File
@@ -1,4 +1,5 @@
"""Tests for early duplicate-work detection (#400)."""
import json
import os
import sys
import tempfile
@@ -8,8 +9,6 @@ from unittest.mock import patch
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import issue_lock_provenance
import issue_lock_store
import issue_work_duplicate_gate as dup_gate
import mcp_server
from issue_work_duplicate_gate import (
@@ -123,9 +122,8 @@ 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, _api):
def test_lock_issue_uses_injected_fetcher(self, _auth):
seen = {}
def fetcher(h, o, r, auth, issue_number):
@@ -142,29 +140,26 @@ class TestInjectableDuplicateFetcher(unittest.TestCase):
"porcelain_status": "",
"base_equivalent": True,
},
):
with tempfile.TemporaryDirectory() as lock_dir:
with patch.dict(os.environ, {
"GITEA_ALLOWED_OPERATIONS": "gitea.issue.comment",
"GITEA_ISSUE_LOCK_DIR": lock_dir,
}, clear=True):
mcp_server.gitea_lock_issue(
issue_number=400,
branch_name="feat/issue-400-duplicate-work-preflight",
remote="prgs",
)
), patch.dict(os.environ, {
"GITEA_ALLOWED_OPERATIONS": "gitea.issue.comment",
}, clear=True):
with patch.object(mcp_server, "ISSUE_LOCK_FILE", tempfile.mktemp()):
mcp_server.gitea_lock_issue(
issue_number=400,
branch_name="feat/issue-400-duplicate-work-preflight",
remote="prgs",
)
self.assertEqual(seen["issue_number"], 400)
class TestMcpDuplicateRecheck(unittest.TestCase):
def setUp(self):
self._dir = tempfile.TemporaryDirectory()
self._env_patch = patch.dict(
os.environ,
{"GITEA_ISSUE_LOCK_DIR": self._dir.name},
clear=False,
self.lock_path = os.path.join(self._dir.name, "gitea_issue_lock.json")
self._lock_patch = patch.object(
mcp_server, "ISSUE_LOCK_FILE", self.lock_path
)
self._env_patch.start()
self._lock_patch.start()
self._remotes = patch.dict(mcp_server.REMOTES, {
"prgs": {"host": "gitea.example.com", "org": "Example-Org",
"repo": "Example-Repo"},
@@ -177,27 +172,12 @@ class TestMcpDuplicateRecheck(unittest.TestCase):
self._dir.cleanup()
def _write_lock(self, issue_number=400, branch="feat/issue-400-x"):
worktree_path = os.path.realpath(os.getcwd())
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",
}
issue_lock_store.bind_session_lock({
"issue_number": issue_number,
"branch_name": branch,
"remote": "prgs",
"org": "Example-Org",
"repo": "Example-Repo",
"worktree_path": worktree_path,
"work_lease": work_lease,
"lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
tool="gitea_lock_issue",
claimant=work_lease.get("claimant"),
),
})
with open(self.lock_path, "w", encoding="utf-8") as fh:
json.dump({
"issue_number": issue_number,
"branch_name": branch,
"remote": "prgs",
}, fh)
@patch("mcp_server._assess_issue_duplicate_gate")
@patch("mcp_server.get_profile", return_value={
@@ -261,7 +241,6 @@ class TestMcpDuplicateRecheck(unittest.TestCase):
base="master",
body="Closes #400",
remote="prgs",
worktree_path=os.path.realpath(os.getcwd()),
)
self.assertFalse(result["success"])
self.assertIsNone(result.get("number"))
+30 -147
View File
@@ -81,64 +81,6 @@ 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": (
@@ -626,19 +568,6 @@ 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},
@@ -899,11 +828,9 @@ class TestMergePR(unittest.TestCase):
pr_number=8, confirmation=self._confirm(8),
expected_head_sha="deadbeef", remote="prgs")
self.assertFalse(r["performed"])
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.assertIn(
"expected head SHA does not match current PR head (fail closed)",
r["reasons"])
self._assert_no_merge_call(mock_api)
@patch("mcp_server.api_request")
@@ -1782,20 +1709,7 @@ 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, {
@@ -1890,19 +1804,8 @@ 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 {
@@ -2140,11 +2043,9 @@ class TestSubmitPrReview(unittest.TestCase):
final_review_decision_ready=True,
)
self.assertFalse(r["performed"])
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.assertIn(
"expected head SHA does not match current PR head (fail closed)",
r["reasons"])
self._assert_no_mutation(mock_api)
def test_head_sha_match_allows(self):
@@ -2207,9 +2108,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(8, "approve", remote="prgs")
gitea_mark_final_review_decision(5, "approve", remote="prgs")
r = gitea_submit_pr_review(
pr_number=8, action="approve", remote="prgs",
pr_number=5, action="approve", remote="prgs",
final_review_decision_ready=True,
)
self.assertFalse(r["performed"])
@@ -2428,13 +2329,6 @@ 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"}
@@ -2469,13 +2363,6 @@ 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"}
@@ -3157,12 +3044,10 @@ 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("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")
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):
@@ -3261,7 +3146,7 @@ class TestIssueLocking(unittest.TestCase):
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 contain locked issue pattern", str(ctx.exception))
self.assertIn("must match", str(ctx.exception))
@patch(
"mcp_server.issue_lock_worktree.read_worktree_git_state",
@@ -3318,7 +3203,6 @@ class TestIssueLocking(unittest.TestCase):
@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"
self.mock_dup_fetcher.return_value = ([], [branch], {"status": "not_claimed"})
mock_api.return_value = [{"name": branch, "commit": {"id": "abc123"}}]
res = gitea_lock_issue(issue_number=196, branch_name=branch, remote="prgs")
self.assertTrue(res["success"])
@@ -3546,17 +3430,16 @@ class TestIssueLocking(unittest.TestCase):
@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())
with tempfile.TemporaryDirectory() as lock_dir:
env = {**self._create_pr_env(), "GITEA_ISSUE_LOCK_DIR": lock_dir}
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=lock_dir,
),
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,
),
_sample_issue_lock(
issue_number=447,
branch_name="feat/issue-447-lock-provenance",
@@ -3566,14 +3449,14 @@ class TestIssueLocking(unittest.TestCase):
worktree_path=worktree,
lock_provenance=None,
),
)
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,
)
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())
@patch("mcp_server.api_request")
+3 -16
View File
@@ -156,22 +156,9 @@ class TestPRQueueInventory(unittest.TestCase):
]
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
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,
)
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)
self.assertTrue(result["success"])
self.assertIn("=== PR Queue Inventory ===", result["message"])
self.assertIn("Repository:", result["message"])
-193
View File
@@ -1,193 +0,0 @@
"""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()
-153
View File
@@ -1,153 +0,0 @@
"""Tests for runtime_context / mutation-guard workspace alignment (#460)."""
from __future__ import annotations
import os
import sys
import unittest
from pathlib import Path
from unittest import mock
from unittest.mock import MagicMock
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import author_mutation_worktree as amw # noqa: E402
import gitea_mcp_server as srv # noqa: E402
CONTROL_ROOT = str(Path(__file__).resolve().parents[3])
BRANCHES_WORKTREE = str(Path(__file__).resolve().parents[1])
MCP_PROCESS_ROOT = BRANCHES_WORKTREE
class TestCanonicalRepoRoot(unittest.TestCase):
@mock.patch("subprocess.run")
def test_resolves_main_repo_from_branches_worktree(self, mock_run):
mock_run.return_value = MagicMock(
returncode=0,
stdout=f"{CONTROL_ROOT}/.git\n",
)
root = amw.resolve_canonical_repo_root(BRANCHES_WORKTREE, MCP_PROCESS_ROOT)
self.assertEqual(root, CONTROL_ROOT)
def test_falls_back_when_git_unavailable(self):
root = amw.resolve_canonical_repo_root("/missing/path", MCP_PROCESS_ROOT)
self.assertEqual(root, os.path.realpath(MCP_PROCESS_ROOT))
class TestWorkspaceRepoMembership(unittest.TestCase):
@mock.patch("os.path.isdir", return_value=True)
@mock.patch("os.path.exists", return_value=True)
@mock.patch("subprocess.run")
def test_valid_branches_worktree_accepted(self, mock_run, *_exists):
mock_run.return_value = MagicMock(
returncode=0,
stdout=f"{CONTROL_ROOT}/.git\n",
)
result = amw.assess_workspace_repo_membership(
workspace_path=BRANCHES_WORKTREE,
canonical_repo_root=CONTROL_ROOT,
)
self.assertTrue(result["proven"])
self.assertFalse(result["block"])
@mock.patch("os.path.isdir", return_value=True)
@mock.patch("os.path.exists", return_value=True)
@mock.patch("subprocess.run")
def test_wrong_repo_rejected(self, mock_run, *_exists):
mock_run.return_value = MagicMock(
returncode=0,
stdout="/other/repo/.git\n",
)
result = amw.assess_workspace_repo_membership(
workspace_path=BRANCHES_WORKTREE,
canonical_repo_root=CONTROL_ROOT,
)
self.assertTrue(result["block"])
self.assertIn("does not belong", result["reasons"][0])
@mock.patch("os.path.exists", return_value=False)
def test_missing_worktree_rejected(self, *_exists):
result = amw.assess_workspace_repo_membership(
workspace_path=f"{CONTROL_ROOT}/branches/missing-worktree",
canonical_repo_root=CONTROL_ROOT,
)
self.assertTrue(result["block"])
self.assertIn("does not exist", result["reasons"][0])
class TestRuntimeContextGuardAlignment(unittest.TestCase):
def setUp(self):
srv._preflight_whoami_called = True
srv._preflight_capability_called = True
srv._preflight_resolved_role = "author"
self._orig_in_test = srv._preflight_in_test_mode
srv._preflight_in_test_mode = lambda: False
self._env_patch = mock.patch.dict(
os.environ,
{},
clear=False,
)
self._env_patch.start()
os.environ.pop("GITEA_ACTIVE_WORKTREE", None)
os.environ.pop("GITEA_AUTHOR_WORKTREE", None)
def tearDown(self):
srv._preflight_in_test_mode = self._orig_in_test
self._env_patch.stop()
def test_runtime_context_and_guard_share_resolved_workspace(self):
with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT):
ctx = srv._resolve_author_mutation_context(BRANCHES_WORKTREE)
status = srv.assess_preflight_status(worktree_path=BRANCHES_WORKTREE)
self.assertEqual(ctx["workspace_path"], os.path.realpath(BRANCHES_WORKTREE))
self.assertEqual(ctx["canonical_repo_root"], CONTROL_ROOT)
self.assertFalse(ctx["roots_aligned"])
self.assertEqual(
status["preflight_workspace"]["active_task_workspace_root"],
os.path.realpath(BRANCHES_WORKTREE),
)
self.assertEqual(
status["preflight_workspace"]["canonical_repository_root"],
CONTROL_ROOT,
)
self.assertIn("workspace_root_mismatch", status["preflight_workspace"])
@mock.patch("subprocess.run")
@mock.patch("os.path.isdir", return_value=True)
@mock.patch("os.path.exists", return_value=True)
def test_declared_branches_worktree_passes_when_mcp_root_differs(
self, _exists, _isdir, mock_run
):
mock_run.return_value = MagicMock(
returncode=0,
stdout=f"{CONTROL_ROOT}/.git\n",
)
with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT):
with mock.patch.dict("os.environ", {"GITEA_TEST_PORCELAIN": ""}, clear=False):
srv.verify_preflight_purity(worktree_path=BRANCHES_WORKTREE)
def test_stable_checkout_still_rejected(self):
with mock.patch.object(srv, "PROJECT_ROOT", CONTROL_ROOT):
with mock.patch.dict("os.environ", {"GITEA_TEST_PORCELAIN": ""}, clear=False):
with self.assertRaises(RuntimeError) as ctx:
srv.verify_preflight_purity()
self.assertIn("stable control checkout", str(ctx.exception))
@mock.patch("os.path.isdir", return_value=True)
@mock.patch("os.path.exists", return_value=True)
@mock.patch("subprocess.run")
def test_non_branches_worktree_rejected(self, mock_run, *_exists):
outside = "/tmp/outside-repo-checkout"
mock_run.return_value = MagicMock(
returncode=0,
stdout=f"{CONTROL_ROOT}/.git\n",
)
with mock.patch.object(srv, "PROJECT_ROOT", CONTROL_ROOT):
with mock.patch.dict("os.environ", {"GITEA_TEST_PORCELAIN": ""}, clear=False):
with self.assertRaises(RuntimeError) as ctx:
srv.verify_preflight_purity(worktree_path=outside)
self.assertIn("not under", str(ctx.exception))
if __name__ == "__main__":
unittest.main()