Compare commits

..
3 Commits
Author SHA1 Message Date
sysadminandClaude Opus 4.8 842776f916 fix: correct worktree-start heredoc bash syntax (#438)
Move the command-substitution closing paren after the heredoc terminator
so bash -n passes and worktree lock checks run correctly.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-07 18:01:54 -04:00
sysadmin 2cace4f7cd resolve conflicts for PR #464 2026-07-07 17:25:13 -04:00
sysadminandClaude Opus 4.8 3ad7547a3f feat: harden issue lock store against concurrent session clobbering (Closes #438)
Replace read-modify-write on the global /tmp/gitea_issue_lock.json pointer with
atomic per-issue locks under GITEA_ISSUE_LOCK_DIR using flock serialization,
PID/session metadata, fail-closed stale recovery, and pre-mutation freshness
re-checks in gitea_create_pr. Queue inventory now surfaces live_issue_locks.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-07 17:21:53 -04:00
21 changed files with 730 additions and 2575 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,
+13 -17
View File
@@ -274,24 +274,20 @@ 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).
acquires an atomic per-issue lock under `GITEA_ISSUE_LOCK_DIR` (default
`/tmp/gitea_issue_locks/`) and updates the legacy session pointer at
`GITEA_ISSUE_LOCK_FILE` only when safe. The payload records issue number,
branch, repo scope, worktree path, claimant identity/profile, PID, session id,
created timestamp, expiry timestamp, and last heartbeat timestamp. An active
same-issue/same-operation lease blocks duplicate work. An expired or dead-PID
lease still blocks takeover until a recovery review records why the prior work
is abandoned, completed, or unsafe to continue.
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.
Author/reviewer/reconciler final reports must include **Issue lock proof** with:
lock acquired, lock owner, lock freshness, no competing live lock (when
applicable), and whether the lock was released or intentionally retained.
`gitea_lock_issue` returns a canonical `lock_proof` string for handoffs.
`gitea_list_claim_inventory` exposes `live_issue_locks` for queue visibility.
**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
+116 -467
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)))
@@ -534,12 +540,10 @@ 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 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
@@ -547,9 +551,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"
@@ -581,59 +584,23 @@ 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 _load_existing_issue_lock() -> dict | None:
return issue_lock_store.read_lock_file(ISSUE_LOCK_FILE)
def _resolve_issue_lock_for_pr(
def _load_issue_lock_for_scope(
*,
issue_number: int,
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)
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
) -> dict | None:
return issue_lock_store.resolve_lock_for_issue(
issue_number=issue_number,
remote=remote,
org=org,
repo=repo,
) or _load_existing_issue_lock()
def _work_lease_claimant(host: str | None) -> dict:
@@ -836,19 +803,6 @@ def _enforce_locked_issue_duplicate_recheck(
return None
def _branch_entry_commit_sha(branch: dict | str) -> str | None:
"""Best-effort head SHA for a Gitea branch entry (None when absent)."""
if not isinstance(branch, dict):
return None
commit = branch.get("commit")
if isinstance(commit, dict):
sha = commit.get("id") or commit.get("sha")
if sha:
return str(sha)
sha = branch.get("commit_sha")
return str(sha) if sha else None
def _reveal_endpoints() -> bool:
"""Admin/debug opt-in (#120): include endpoint URLs and token source
names in tool output. Off by default so normal LLM-facing responses
@@ -1297,8 +1251,13 @@ def gitea_lock_issue(
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_issue_lock_for_scope(
issue_number=issue_number,
remote=remote,
org=o,
repo=r,
),
issue_number=issue_number,
branch_name=branch_name,
worktree_path=resolved_worktree,
@@ -1337,68 +1296,47 @@ def gitea_lock_issue(
f"duplicate work gate blocked issue #{issue_number} (fail closed)"
]))
branch_url = f"{repo_api_url(h, o, r)}/branches"
try:
branches = api_get_all(branch_url, auth)
except Exception as e:
raise RuntimeError(f"Could not list branches to verify issue lock: {e}")
existing_branch_entries = [
{
"name": _branch_entry_name(branch),
"commit_sha": _branch_entry_commit_sha(branch),
}
for branch in branches
]
adoption = issue_lock_adoption.assess_own_branch_adoption(
issue_number=issue_number,
requested_branch=branch_name,
existing_branches=existing_branch_entries,
)
if adoption["block"]:
competing = ", ".join(adoption["competing_branches"])
raise ValueError(
f"Issue #{issue_number} already has matching branch '{competing}' "
"that is not the requested branch (fail closed)"
)
work_lease = _build_author_issue_work_lease(
issue_number=issue_number,
branch_name=branch_name,
worktree_path=resolved_worktree,
host=h,
)
data = {
"issue_number": issue_number,
"branch_name": branch_name,
"remote": remote,
"org": o,
"repo": r,
"worktree_path": resolved_worktree,
"work_lease": work_lease,
"lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
tool="gitea_lock_issue",
claimant=work_lease.get("claimant"),
),
}
lock_provenance = issue_lock_provenance.build_sanctioned_lock_provenance(
tool="gitea_lock_issue",
claimant=work_lease.get("claimant"),
)
try:
acquisition = issue_lock_store.acquire_issue_lock(
issue_number=issue_number,
branch_name=branch_name,
remote=remote,
org=o,
repo=r,
worktree_path=resolved_worktree,
work_lease=work_lease,
claimant=_work_lease_claimant(h),
lock_provenance=lock_provenance,
)
except RuntimeError:
raise
except Exception as e:
raise RuntimeError(f"Could not acquire issue lock: {e}") from e
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)
agent_artifacts = agent_temp_artifacts.find_agent_temp_artifacts_from_porcelain(
git_state.get("porcelain_status") or ""
)
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,
acquisition["record"],
freshness=acquisition["freshness"],
competing_live_locks=competing,
released=False,
)
agent_artifacts = agent_temp_artifacts.find_agent_temp_artifacts_from_porcelain(
git_state.get("porcelain_status") or ""
)
result = {
"success": True,
"message": (
@@ -1409,24 +1347,12 @@ def gitea_lock_issue(
"branch_name": branch_name,
"worktree_path": resolved_worktree,
"work_lease": work_lease,
"lock_file_path": lock_file_path,
"lock_freshness": freshness,
"lock_path": acquisition["lock_path"],
"session_id": acquisition["session_id"],
"lock_freshness": acquisition["freshness"],
"legacy_pointer": acquisition["legacy_pointer"],
"lock_proof": lock_proof,
}
if adoption["adopt"]:
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_status="written",
)
result["message"] = (
f"Adopted existing branch '{branch_name}' and locked issue "
f"#{issue_number} for recovery (fail-closed check complete)."
)
if agent_artifacts:
result["warnings"] = [
"Agent temp artifacts at repo root (delete before implementation): "
@@ -1518,8 +1444,19 @@ 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 / #438) ──
lock_data = _load_existing_issue_lock()
if not lock_data:
raise RuntimeError("Issue lock is missing (fail closed). Call gitea_lock_issue first.")
scoped_lock = issue_lock_store.resolve_lock_for_issue(
issue_number=int(lock_data.get("issue_number") or 0),
remote=lock_data.get("remote"),
org=lock_data.get("org"),
repo=lock_data.get("repo"),
)
if scoped_lock:
lock_data = scoped_lock
lock_provenance_check = issue_lock_provenance.assess_lock_file_for_create_pr(
lock_data
@@ -1544,12 +1481,7 @@ 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,
)
ownership = issue_lock_store.verify_lock_for_mutation(lock_data)
if ownership["block"]:
raise ValueError(ownership["reasons"][0])
@@ -2123,7 +2055,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 +2497,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:
@@ -3062,7 +2979,7 @@ 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 = _load_existing_issue_lock() or {}
locked_worktree = lock_data.get("worktree_path")
if locked_worktree:
@@ -3373,19 +3290,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 +4589,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,
-116
View File
@@ -1,116 +0,0 @@
"""Own-branch lock adoption / recovery for ``gitea_lock_issue`` (#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.
This module isolates the pure decision so it can be unit-tested apart from the
MCP server's live Gitea calls.
Adoption is granted only for the issue's *exact* requested branch. Any other
branch that merely contains the same ``issue-<n>`` marker is competing work and
stays fail-closed. Open-PR, competing-live-lock, capability, and worktree
safety checks are enforced by the caller before this decision is consulted;
this module additionally records whether they passed for proof purposes.
"""
from __future__ import annotations
ADOPT = "adopt_existing_branch"
BLOCK_COMPETING = "block_competing_branch"
NO_MATCH = "no_matching_branch"
def _branch_name(entry) -> str:
if isinstance(entry, dict):
return str(entry.get("name") or "")
return str(entry or "")
def _branch_sha(entry) -> str | None:
if isinstance(entry, dict):
sha = entry.get("commit_sha")
if sha:
return str(sha)
return None
def assess_own_branch_adoption(
*,
issue_number: int,
requested_branch: str,
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:
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]
if competing:
return {
"outcome": BLOCK_COMPETING,
"adopt": False,
"block": True,
"reason": (
f"issue #{issue_number} already has matching branch(es) "
f"{competing} that are not the requested branch "
f"'{requested}' (fail closed)"
),
"matched_branch": None,
"matched_head_sha": None,
"competing_branches": competing,
}
if exact:
name, sha = exact[0]
return {
"outcome": ADOPT,
"adopt": True,
"block": False,
"reason": (
f"existing branch '{name}' is the exact requested branch for "
f"issue #{issue_number}; adopting it for lock recovery"
),
"matched_branch": name,
"matched_head_sha": sha,
"competing_branches": [],
}
return {
"outcome": NO_MATCH,
"adopt": False,
"block": False,
"reason": f"no existing branch matches issue #{issue_number}",
"matched_branch": None,
"matched_head_sha": None,
"competing_branches": [],
}
def build_adoption_proof(
*,
issue_number: int,
branch_name: str,
assessment: dict,
open_pr_checked: bool,
competing_lock_checked: bool,
lock_file_path: str,
lock_file_status: str,
) -> dict:
"""Assemble the proof block returned by ``gitea_lock_issue`` on adoption."""
return {
"issue_number": issue_number,
"branch_name": branch_name,
"branch_head_commit": assessment.get("matched_head_sha"),
"adoption_reason": assessment.get("reason"),
"no_existing_pr_proof": bool(open_pr_checked),
"no_competing_live_lock_proof": bool(competing_lock_checked),
"lock_file_path": lock_file_path,
"lock_file_status": lock_file_status,
}
+266 -392
View File
@@ -1,10 +1,8 @@
"""Keyed, persistent issue-lock storage (#443) with flock hardening (#438).
"""Atomic per-issue lock store (#438).
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``.
Distinct issues use separate lock files under ``GITEA_ISSUE_LOCK_DIR`` so
concurrent author sessions do not clobber each other. Acquisition is
serialized per issue with ``fcntl.flock`` and fail-closed stale handling.
"""
from __future__ import annotations
@@ -13,76 +11,69 @@ import errno
import fcntl
import json
import os
import re
import tempfile
import uuid
from contextlib import contextmanager
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._+-]+")
DEFAULT_LOCK_DIR = os.environ.get("GITEA_ISSUE_LOCK_DIR", "/tmp/gitea_issue_locks")
LEGACY_LOCK_FILE = os.environ.get("GITEA_ISSUE_LOCK_FILE", "/tmp/gitea_issue_lock.json")
LOCK_VERSION = 1
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
def _sanitize_segment(value: str) -> str:
def _sanitize(value: str) -> str:
text = (value or "").strip()
if not text:
return "_"
return _SAFE_SEGMENT_RE.sub("_", text)
return "unknown"
return "".join(c if c.isalnum() or c in "-_" else "-" for c in 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_scope_key(remote: str, org: str, repo: str, issue_number: int) -> str:
return "_".join(
(
_sanitize(remote),
_sanitize(org),
_sanitize(repo),
f"issue-{issue_number}",
)
)
def lock_file_path(
*,
def issue_lock_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
base = (lock_dir or DEFAULT_LOCK_DIR).strip()
return os.path.join(base, f"{lock_scope_key(remote, org, repo, issue_number)}.json")
def flock_path(json_path: str) -> str:
return f"{json_path}.lock"
def _iso(value: datetime) -> str:
return value.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
def parse_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 is_process_alive(pid: int | None) -> bool:
if not pid or pid <= 0:
return False
@@ -95,6 +86,32 @@ def is_process_alive(pid: int | None) -> bool:
return False
def read_lock_file(path: str) -> dict[str, Any] | None:
if not os.path.exists(path):
return None
try:
with open(path, encoding="utf-8") as handle:
data = json.load(handle)
return data if isinstance(data, dict) else None
except Exception:
return None
def atomic_write_json(path: str, data: dict[str, Any]) -> None:
directory = os.path.dirname(path) or "."
os.makedirs(directory, exist_ok=True)
fd, tmp_path = tempfile.mkstemp(dir=directory, prefix=".lock-", suffix=".tmp")
try:
with os.fdopen(fd, "w", encoding="utf-8") as handle:
json.dump(data, handle)
handle.flush()
os.fsync(handle.fileno())
os.replace(tmp_path, path)
finally:
if os.path.exists(tmp_path):
os.remove(tmp_path)
@contextmanager
def _exclusive_file_lock(lock_path: str):
os.makedirs(os.path.dirname(lock_path) or ".", exist_ok=True)
@@ -114,201 +131,13 @@ def _exclusive_file_lock(lock_path: str):
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):
def _lease_expires_at(lock_data: dict[str, Any] | None) -> datetime | None:
if not lock_data:
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()
record.setdefault("pid", os.getpid())
pointer = {
"pid": os.getpid(),
"lock_file_path": path,
"issue_number": issue_number,
"branch_name": record.get("branch_name"),
"remote": remote,
"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
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:
return assess_lock_freshness(lock, now=now)["live"]
lease = lock_data.get("work_lease")
if isinstance(lease, dict):
return parse_timestamp(lease.get("expires_at"))
return parse_timestamp(lock_data.get("expires_at"))
def assess_lock_freshness(
@@ -317,7 +146,7 @@ def assess_lock_freshness(
now: datetime | None = None,
) -> dict[str, Any]:
"""Classify a lock as live, expired, stale, or absent."""
current = _lease_now(now)
current = now or datetime.now(timezone.utc)
if not lock_data:
return {
"status": "absent",
@@ -326,15 +155,12 @@ def assess_lock_freshness(
"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"))
expires_at = _lease_expires_at(lock_data)
heartbeat_at = parse_timestamp(lock_data.get("last_heartbeat_at"))
if heartbeat_at is None and isinstance(lock_data.get("work_lease"), dict):
heartbeat_at = parse_timestamp(lock_data["work_lease"].get("last_heartbeat_at"))
pid = lock_data.get("session_pid")
if pid is None:
pid = lock_data.get("pid")
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:
@@ -361,142 +187,192 @@ def assess_lock_freshness(
"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,
"heartbeat_at": _iso(heartbeat_at) if heartbeat_at else None,
"expires_at": _iso(expires_at) if expires_at else None,
}
def _same_realpath(left: str | None, right: str | None) -> bool:
if not left or not right:
def _same_owner(
existing: dict[str, Any],
*,
branch_name: str,
worktree_path: str,
claimant: dict[str, Any] | None,
) -> bool:
same_branch = existing.get("branch_name") == branch_name
same_worktree = os.path.realpath(str(existing.get("worktree_path") or "")) == os.path.realpath(
worktree_path
)
if not (same_branch and same_worktree):
return False
try:
return os.path.realpath(left) == os.path.realpath(right)
except OSError:
return left == right
if claimant and isinstance(existing.get("claimant"), dict):
return (
existing["claimant"].get("profile") == claimant.get("profile")
and existing["claimant"].get("username") == claimant.get("username")
)
return same_branch and same_worktree
def assess_same_issue_lease_conflict(
existing_lock: dict[str, Any] | None,
def update_legacy_session_pointer(record: dict[str, Any]) -> dict[str, Any]:
"""Update the legacy global pointer without clobbering another live session."""
existing = read_lock_file(LEGACY_LOCK_FILE)
if existing:
freshness = assess_lock_freshness(existing)
if freshness["status"] == "live":
different_issue = existing.get("issue_number") != record.get("issue_number")
different_pid = existing.get("pid") != os.getpid()
if different_issue and different_pid and is_process_alive(existing.get("pid")):
return {
"updated": False,
"reason": (
"retained global pointer for live issue "
f"#{existing.get('issue_number')} pid={existing.get('pid')}; "
f"per-issue lock at '{record.get('lock_path')}' is authoritative"
),
}
atomic_write_json(LEGACY_LOCK_FILE, record)
return {"updated": True, "path": LEGACY_LOCK_FILE}
def acquire_issue_lock(
*,
issue_number: int,
branch_name: str,
remote: str,
org: str,
repo: 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
work_lease: dict[str, Any],
claimant: dict[str, Any] | None = None,
lock_provenance: dict[str, Any] | None = None,
lock_dir: str | None = None,
allow_stale_recovery: bool = False,
) -> dict[str, Any]:
"""Acquire an atomic per-issue lock; fail closed on live competing ownership."""
path = issue_lock_path(remote, org, repo, issue_number, lock_dir=lock_dir)
sentinel = flock_path(path)
now = datetime.now(timezone.utc)
session_id = str(uuid.uuid4())
resolved_worktree = os.path.realpath(worktree_path)
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
try:
with _exclusive_file_lock(sentinel):
existing = read_lock_file(path)
freshness = assess_lock_freshness(existing, now=now)
if existing and freshness["live"]:
if not _same_owner(
existing,
branch_name=branch_name,
worktree_path=resolved_worktree,
claimant=claimant,
):
owner = existing.get("claimant") or {}
raise RuntimeError(
f"Issue #{issue_number} already has an active lock owned by "
f"pid={existing.get('pid')} session={existing.get('session_id')} "
f"profile={owner.get('profile')} on branch "
f"'{existing.get('branch_name')}' (fail closed)"
)
elif existing and freshness["stale"]:
if not allow_stale_recovery:
raise RuntimeError(
f"Issue #{issue_number} has a stale lock ({freshness['reason']}). "
"Recovery review is required before takeover (fail closed)"
)
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)"
)
record: dict[str, Any] = {
"lock_version": LOCK_VERSION,
"issue_number": issue_number,
"branch_name": branch_name,
"remote": remote,
"org": org,
"repo": repo,
"worktree_path": resolved_worktree,
"work_lease": work_lease,
"pid": os.getpid(),
"session_id": session_id,
"claimant": claimant or {},
"created_at": _iso(now),
"last_heartbeat_at": _iso(now),
"lock_path": path,
}
if lock_provenance:
record["lock_provenance"] = lock_provenance
atomic_write_json(path, record)
pointer = update_legacy_session_pointer(record)
except LockContentionError as exc:
competing = read_lock_file(path)
if competing:
freshness = assess_lock_freshness(competing, now=now)
owner = competing.get("claimant") or {}
raise RuntimeError(
f"Issue #{issue_number} lock contention: {exc}; competing owner "
f"pid={competing.get('pid')} session={competing.get('session_id')} "
f"profile={owner.get('profile')} status={freshness['status']} (fail closed)"
) from exc
raise RuntimeError(f"Issue #{issue_number} lock contention: {exc} (fail closed)") from exc
freshness = assess_lock_freshness(record, now=now)
return {
"acquired": True,
"lock_path": path,
"session_id": session_id,
"freshness": freshness,
"legacy_pointer": pointer,
"lock_proof": format_lock_proof(record, freshness=freshness),
"record": record,
}
def assess_foreign_lock_overwrite(
existing_lock: dict[str, Any] | None,
incoming_lock: dict[str, Any],
def resolve_lock_for_issue(
*,
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,
issue_number: int,
remote: str | None = None,
org: str | None = None,
repo: str | None = None,
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
"""Load the per-issue lock, falling back to the legacy global pointer."""
if remote and org and repo:
scoped = read_lock_file(
issue_lock_path(remote, org, repo, issue_number, lock_dir=lock_dir)
)
if scoped:
return scoped
legacy = read_lock_file(LEGACY_LOCK_FILE)
if legacy and legacy.get("issue_number") == issue_number:
return legacy
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,
def resolve_lock_for_branch(
branch_name: 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
) -> dict[str, Any] | None:
"""Resolve a lock using the issue number embedded in a branch name."""
import re
match = re.search(r"issue-(\d+)", branch_name or "", re.IGNORECASE)
if not match:
legacy = read_lock_file(LEGACY_LOCK_FILE)
if legacy and legacy.get("branch_name") == branch_name:
return legacy
return None
issue_number = int(match.group(1))
base = (lock_dir or DEFAULT_LOCK_DIR).strip()
if os.path.isdir(base):
suffix = f"_issue-{issue_number}.json"
for name in os.listdir(base):
if name.endswith(suffix):
record = read_lock_file(os.path.join(base, name))
if record and record.get("branch_name") == branch_name:
return record
legacy = read_lock_file(LEGACY_LOCK_FILE)
if legacy and legacy.get("issue_number") == issue_number:
return legacy
return None
def verify_lock_for_mutation(
@@ -543,37 +419,36 @@ def verify_lock_for_mutation(
}
def list_live_locks(
*,
lock_dir: str | None = None,
now: datetime | None = None,
) -> list[dict[str, Any]]:
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."""
base = (lock_dir or DEFAULT_LOCK_DIR).strip()
if not os.path.isdir(base):
return []
live: list[dict[str, Any]] = []
for path in iter_lock_files(lock_dir):
record = read_lock_file(path)
for name in sorted(os.listdir(base)):
if not name.endswith(".json"):
continue
record = read_lock_file(os.path.join(base, name))
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,
}
)
if freshness["live"]:
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("pid"),
"session_id": record.get("session_id"),
"claimant": record.get("claimant"),
"freshness": freshness,
"lock_path": record.get("lock_path") or os.path.join(base, name),
}
)
return live
@@ -589,14 +464,13 @@ def format_lock_proof(
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"pid {lock_data.get('pid')}",
f"session {lock_data.get('session_id')}",
f"freshness {fresh.get('status')}",
]
if competing_live_locks is not None:
+18 -9
View File
@@ -16,6 +16,7 @@ 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 +38,26 @@ 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 = issue_lock_store.resolve_lock_for_branch(branch)
if lock and (lock.get("branch_name") or "").strip() == (branch or "").strip():
freshness = issue_lock_store.assess_lock_freshness(lock)
return freshness["live"]
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
@@ -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,
}
+11 -6
View File
@@ -37,20 +37,25 @@ 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 "
locked_branch=$(PYTHONPATH="$repo_root" python3 - "$branch" <<'PY'
import sys
sys.path.insert(0, '$repo_root')
import issue_lock_store
print(issue_lock_store.resolve_locked_branch_for_session('$branch'))
")
branch = sys.argv[1]
lock = issue_lock_store.resolve_lock_for_branch(branch)
if not lock:
print("", end="")
sys.exit(1)
print(lock.get("branch_name", ""), end="")
PY
)
if [[ -z "$locked_branch" ]]; then
echo "Error: No session issue lock is bound. Call gitea_lock_issue before branch creation (fail closed)." >&2
echo "Error: No issue lock found for branch '$branch'. Lock the issue before branch creation (fail closed)." >&2
exit 2
fi
if [[ "$branch" != "$locked_branch" ]]; then
@@ -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:
+3 -6
View File
@@ -74,24 +74,21 @@ 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",
+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)
+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")
+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)
-68
View File
@@ -1,68 +0,0 @@
"""Unit tests for own-branch lock adoption decision (#442 / #443)."""
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from issue_lock_adoption import ( # noqa: E402
ADOPT,
BLOCK_COMPETING,
NO_MATCH,
assess_own_branch_adoption,
build_adoption_proof,
)
REQ = "feat/issue-420-server-code-parity"
class TestAssessOwnBranchAdoption(unittest.TestCase):
def test_exact_own_branch_is_adopted(self):
result = assess_own_branch_adoption(
issue_number=420,
requested_branch=REQ,
existing_branches=[{"name": REQ, "commit_sha": "934688a"}],
)
self.assertEqual(result["outcome"], ADOPT)
self.assertTrue(result["adopt"])
def test_different_branch_same_issue_blocks(self):
result = assess_own_branch_adoption(
issue_number=420,
requested_branch=REQ,
existing_branches=[{"name": "feat/issue-420-other-work"}],
)
self.assertEqual(result["outcome"], BLOCK_COMPETING)
self.assertTrue(result["block"])
def test_no_matching_branch_is_normal_path(self):
result = assess_own_branch_adoption(
issue_number=420,
requested_branch=REQ,
existing_branches=[{"name": "feat/issue-999-unrelated"}],
)
self.assertEqual(result["outcome"], NO_MATCH)
class TestBuildAdoptionProof(unittest.TestCase):
def test_proof_has_required_fields(self):
assessment = assess_own_branch_adoption(
issue_number=420,
requested_branch=REQ,
existing_branches=[{"name": REQ, "commit_sha": "934688a"}],
)
proof = build_adoption_proof(
issue_number=420,
branch_name=REQ,
assessment=assessment,
open_pr_checked=True,
competing_lock_checked=True,
lock_file_path="/tmp/example-lock.json",
lock_file_status="written",
)
self.assertEqual(proof["branch_head_commit"], "934688a")
self.assertTrue(proof["no_existing_pr_proof"])
if __name__ == "__main__":
unittest.main()
+147 -197
View File
@@ -1,193 +1,87 @@
"""Unit tests for keyed issue-lock storage (#443) and flock hardening (#438)."""
import json
"""Tests for atomic per-issue lock store (#438)."""
from __future__ import annotations
import os
import sys
import tempfile
import threading
import unittest
from datetime import datetime, timedelta, timezone
from pathlib import Path
from unittest import mock
from unittest.mock import patch
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
import issue_lock_store as ils # noqa: E402
import issue_lock_store # 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 _lease(**overrides):
now = datetime.now(timezone.utc)
lease = {
"operation_type": "author_issue_work",
"issue_number": 438,
"branch": "feat/issue-438-lock-hardening",
"worktree_path": "/tmp/wt",
"expires_at": (now + timedelta(hours=4)).isoformat().replace("+00:00", "Z"),
"last_heartbeat_at": now.isoformat().replace("+00:00", "Z"),
}
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
lease.update(overrides)
return lease
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()
self.tempdir = tempfile.mkdtemp(prefix="issue-lock-store-")
self.legacy = os.path.join(self.tempdir, "legacy.json")
self.addCleanup(self._cleanup)
def tearDown(self):
self._env.stop()
self._dir.cleanup()
def _cleanup(self):
for root, _dirs, files in os.walk(self.tempdir, topdown=False):
for name in files:
try:
os.remove(os.path.join(root, name))
except OSError:
pass
try:
os.rmdir(root)
except OSError:
pass
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(
def _acquire(self, issue_number=438, branch="feat/issue-438-lock-hardening", **kwargs):
worktree_path = kwargs.pop("worktree_path", "/tmp/wt")
return issue_lock_store.acquire_issue_lock(
issue_number=issue_number,
branch_name=branch,
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
issue_number=420,
worktree_path=worktree_path,
work_lease=_lease(issue_number=issue_number, branch=branch),
claimant={"profile": "prgs-author", "username": "jcwalker3"},
lock_dir=self.tempdir,
**kwargs,
)
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_acquire_writes_per_issue_lock_and_legacy_pointer(self):
with patch.object(issue_lock_store, "LEGACY_LOCK_FILE", self.legacy):
result = self._acquire()
path = issue_lock_store.issue_lock_path("prgs", "Scaled-Tech-Consulting", "Gitea-Tools", 438, lock_dir=self.tempdir)
self.assertTrue(os.path.exists(path))
self.assertTrue(result["acquired"])
self.assertIn("lock_proof", result)
with open(self.legacy, encoding="utf-8") as handle:
legacy = __import__("json").load(handle)
self.assertEqual(legacy["issue_number"], 438)
self.assertEqual(legacy["session_id"], result["session_id"])
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)
def test_concurrent_bind_same_issue_only_one_wins(self):
def test_concurrent_acquire_same_issue_only_one_wins(self):
barrier = threading.Barrier(2)
results: list[str | Exception] = []
results: list[dict | Exception] = []
def worker():
barrier.wait()
try:
ils.bind_session_lock(
_lock_record(worktree_path=f"/tmp/wt-{threading.get_ident()}")
)
results.append("ok")
results.append(self._acquire())
except Exception as exc: # noqa: BLE001
results.append(exc)
@@ -197,59 +91,115 @@ class TestIssueLockStore(unittest.TestCase):
for thread in threads:
thread.join()
successes = [item for item in results if item == "ok"]
successes = [item for item in results if isinstance(item, dict)]
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,
"active lock" 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"),
def test_distinct_issues_acquire_without_contention(self):
first = self._acquire(issue_number=438, branch="feat/issue-438-lock-hardening")
second = self._acquire(
issue_number=440,
branch="feat/issue-440-recovery",
worktree_path="/tmp/other-wt",
)
record["pid"] = 999999
record["session_pid"] = 999999
result = ils.verify_lock_for_mutation(
self.assertTrue(first["acquired"])
self.assertTrue(second["acquired"])
self.assertNotEqual(first["session_id"], second["session_id"])
def test_stale_lock_requires_explicit_recovery(self):
path = issue_lock_store.issue_lock_path("prgs", "Scaled-Tech-Consulting", "Gitea-Tools", 438, lock_dir=self.tempdir)
stale = {
"lock_version": 1,
"issue_number": 438,
"branch_name": "feat/issue-438-old",
"remote": "prgs",
"org": "Scaled-Tech-Consulting",
"repo": "Gitea-Tools",
"worktree_path": "/tmp/old",
"pid": 999999,
"session_id": "stale-session",
"work_lease": _lease(expires_at="2000-01-01T00:00:00Z"),
"last_heartbeat_at": "2000-01-01T00:00:00Z",
"lock_path": path,
}
issue_lock_store.atomic_write_json(path, stale)
with self.assertRaises(RuntimeError) as ctx:
self._acquire()
self.assertIn("Recovery review is required", str(ctx.exception))
recovered = self._acquire(allow_stale_recovery=True)
self.assertTrue(recovered["acquired"])
def test_verify_lock_for_mutation_blocks_stale_lock(self):
record = {
"issue_number": 438,
"branch_name": "feat/issue-438-lock-hardening",
"worktree_path": "/tmp/wt",
"pid": 999999,
"work_lease": _lease(expires_at="2000-01-01T00:00:00Z"),
"last_heartbeat_at": "2000-01-01T00:00:00Z",
}
result = issue_lock_store.verify_lock_for_mutation(
record,
issue_number=420,
branch_name="feat/issue-420-server-code-parity",
worktree_path="/tmp/wt-420",
issue_number=438,
branch_name="feat/issue-438-lock-hardening",
worktree_path="/tmp/wt",
)
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,
path = issue_lock_store.issue_lock_path("prgs", "Scaled-Tech-Consulting", "Gitea-Tools", 438, lock_dir=self.tempdir)
issue_lock_store.atomic_write_json(
path,
{
"issue_number": 438,
"branch_name": "feat/issue-438-lock-hardening",
"remote": "prgs",
"org": "Scaled-Tech-Consulting",
"repo": "Gitea-Tools",
"worktree_path": "/tmp/wt",
"pid": os.getpid(),
"session_id": "live",
"work_lease": _lease(),
"last_heartbeat_at": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
"lock_path": path,
},
)
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 = issue_lock_store.issue_lock_path("prgs", "Scaled-Tech-Consulting", "Gitea-Tools", 440, lock_dir=self.tempdir)
issue_lock_store.atomic_write_json(
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",
),
{
"issue_number": 440,
"branch_name": "feat/issue-440-recovery",
"remote": "prgs",
"org": "Scaled-Tech-Consulting",
"repo": "Gitea-Tools",
"worktree_path": "/tmp/wt",
"pid": 999999,
"session_id": "stale",
"work_lease": _lease(issue_number=440, branch="feat/issue-440-recovery", expires_at="2000-01-01T00:00:00Z"),
"last_heartbeat_at": "2000-01-01T00:00:00Z",
"lock_path": stale_path,
},
)
live = ils.list_live_locks(lock_dir=self.lock_dir)
self.assertEqual([entry["issue_number"] for entry in live], [420])
live = issue_lock_store.list_live_locks(lock_dir=self.tempdir)
self.assertEqual([entry["issue_number"] for entry in live], [438])
def test_resolve_lock_for_branch_reads_per_issue_file(self):
self._acquire()
resolved = issue_lock_store.resolve_lock_for_branch(
"feat/issue-438-lock-hardening",
lock_dir=self.tempdir,
)
self.assertIsNotNone(resolved)
self.assertEqual(resolved["issue_number"], 438)
if __name__ == "__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"))
+105 -280
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": (
@@ -176,7 +118,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",
@@ -200,7 +141,20 @@ def _bind_test_lock(**overrides) -> str:
record.setdefault("org", profile["org"])
record.setdefault("repo", profile["repo"])
record["remote"] = remote
return issue_lock_store.bind_session_lock(record)
worktree = record.get("worktree_path") or os.path.realpath(os.getcwd())
acquisition = issue_lock_store.acquire_issue_lock(
issue_number=int(record["issue_number"]),
branch_name=str(record["branch_name"]),
remote=remote,
org=str(record["org"]),
repo=str(record["repo"]),
worktree_path=worktree,
work_lease=record["work_lease"],
claimant=record["work_lease"].get("claimant"),
lock_provenance=record.get("lock_provenance"),
lock_dir=os.environ.get("GITEA_ISSUE_LOCK_DIR"),
)
return acquisition["lock_path"]
# ---------------------------------------------------------------------------
@@ -626,19 +580,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 +840,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 +1721,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 +1816,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 +2055,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 +2120,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 +2341,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 +2375,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 +3056,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):
@@ -3213,13 +3110,30 @@ class TestIssueLocking(unittest.TestCase):
"""Test issue locking and PR gating constraints."""
def setUp(self):
self._lock_dir = tempfile.TemporaryDirectory()
import shutil
import tempfile
self._lock_dir = tempfile.mkdtemp(prefix="mcp-issue-lock-")
self._session_lock = os.path.join(self._lock_dir, "session.json")
env = {
**ISSUE_WRITE_ENV,
"GITEA_ISSUE_LOCK_DIR": self._lock_dir.name,
"GITEA_ISSUE_LOCK_DIR": self._lock_dir,
"GITEA_ISSUE_LOCK_FILE": self._session_lock,
}
self._env_patcher = patch.dict(os.environ, env, clear=True)
self._env_patcher.start()
self._patchers = [
patch("mcp_server.ISSUE_LOCK_FILE", self._session_lock),
patch("mcp_server.issue_lock_store.DEFAULT_LOCK_DIR", self._lock_dir),
patch("mcp_server.issue_lock_store.LEGACY_LOCK_FILE", self._session_lock),
patch("tests.test_mcp_server.ISSUE_LOCK_FILE", self._session_lock),
patch(
"mcp_server.issue_lock_worktree.read_worktree_git_state",
return_value=_clean_master_git_state_for_lock(),
),
]
for patcher in self._patchers:
patcher.start()
self._dup_fetcher_patcher = patch(
"mcp_server.issue_duplicate_context_fetcher",
return_value=([], [], {"status": "not_claimed"}),
@@ -3227,23 +3141,20 @@ class TestIssueLocking(unittest.TestCase):
self.mock_dup_fetcher = self._dup_fetcher_patcher.start()
def tearDown(self):
self._dup_fetcher_patcher.stop()
self._env_patcher.stop()
self._lock_dir.cleanup()
import shutil
def _create_pr_env(self) -> dict:
return {
**CREATE_PR_ENV,
"GITEA_ISSUE_LOCK_DIR": self._lock_dir.name,
}
self._dup_fetcher_patcher.stop()
for patcher in reversed(getattr(self, "_patchers", [])):
patcher.stop()
self._env_patcher.stop()
shutil.rmtree(getattr(self, "_lock_dir", ""), ignore_errors=True)
@patch(
"mcp_server.issue_lock_worktree.read_worktree_git_state",
return_value=_clean_master_git_state_for_lock(),
)
@patch("mcp_server.api_get_all", return_value=[])
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_lock_issue_success(self, _auth, _api, _git_state):
def test_lock_issue_success(self, _auth, _git_state):
res = gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
self.assertTrue(res["success"])
self.assertEqual(res["work_lease"]["operation_type"], "author_issue_work")
@@ -3253,8 +3164,9 @@ 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)
@@ -3310,81 +3222,32 @@ class TestIssueLocking(unittest.TestCase):
gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
self.assertIn("remote branch(es) already match issue pattern", str(ctx.exception))
@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)
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"])
self.assertIn("adoption", res)
self.assertEqual(res["adoption"]["branch_head_commit"], "abc123")
@patch(
"mcp_server.issue_lock_worktree.read_worktree_git_state",
return_value=_clean_master_git_state_for_lock(),
)
@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,
),
{
def test_lock_issue_blocks_active_same_operation_lease(self):
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))
@patch(
"mcp_server.issue_lock_worktree.read_worktree_git_state",
return_value=_clean_master_git_state_for_lock(),
)
@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,
),
{
def test_lock_issue_blocks_expired_same_operation_lease_for_recovery(self):
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))
@@ -3451,7 +3314,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))
@@ -3460,64 +3325,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",
@@ -3525,13 +3363,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",
@@ -3545,35 +3383,22 @@ 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())
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,
),
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",
)
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")
@@ -3583,13 +3408,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",
+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()
+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):