Compare commits

..
Author SHA1 Message Date
sysadminandClaude Opus 4.8 80aa295635 feat: non-destructive branch recovery after lost issue locks (Closes #440)
Implements the #420 recovery umbrella: keyed persistent issue locks with
own-branch adoption, structured branch ownership parsing, create_pr lock
resolution after MCP restart, and workflow docs forbidding destructive
branch deletion as the default recovery path.

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

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-07 16:27:10 -04:00
16 changed files with 1272 additions and 839 deletions
+19 -13
View File
@@ -274,20 +274,26 @@ 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`
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.
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).
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.
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 (#440). Branch ownership
uses structured ``(fix|feat|docs|chore)/issue-<n>-`` parsing — not broad
substring matches like ``issue-420`` inside ``issue-4200``.
Remote branches matching the issue number are also treated as active work unless
the recovery review proves the branch is abandoned or superseded. Never delete
+117 -88
View File
@@ -537,8 +537,10 @@ import role_namespace_gate # noqa: E402
import task_capability_map # noqa: E402
import review_proofs # noqa: E402
import agent_temp_artifacts
import issue_branch_ownership # noqa: E402
import issue_lock_worktree # 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
@@ -549,8 +551,9 @@ import review_merge_state_machine # noqa: E402
import native_mcp_preference # noqa: E402
# Fail-closed exact-issue-lock file (#204): written by gitea_lock_issue,
# consumed by gitea_create_pr and scripts/worktree-start.
# 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.
ISSUE_LOCK_FILE = "/tmp/gitea_issue_lock.json"
WORK_LEASE_TTL_HOURS = 4
AUTHOR_ISSUE_WORK_LEASE = "author_issue_work"
@@ -582,23 +585,59 @@ def _parse_work_lease_timestamp(value: str | None) -> datetime | None:
return None
def _load_existing_issue_lock() -> dict | None:
return issue_lock_store.read_lock_file(ISSUE_LOCK_FILE)
def _load_issue_lock_for_scope(
def _load_existing_issue_lock(
*,
remote: str | None = None,
org: str | None = None,
repo: str | None = None,
issue_number: int | None = None,
) -> dict | None:
if remote and org and repo and issue_number is not None:
return issue_lock_store.load_issue_lock(
remote=remote,
org=org,
repo=repo,
issue_number=issue_number,
)
return issue_lock_store.read_session_issue_lock()
def _resolve_issue_lock_for_pr(
*,
issue_number: int,
remote: str,
org: str,
repo: str,
) -> 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()
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
def _work_lease_claimant(host: str | None) -> dict:
@@ -688,6 +727,19 @@ def _branch_entry_name(branch: dict | str) -> str:
return str(branch.get("name") or branch.get("ref") or "")
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
@@ -1136,13 +1188,8 @@ def gitea_lock_issue(
worktree_path, PROJECT_ROOT
)
h, o, r = _resolve(remote, host, org, repo)
active_lease_block = _active_work_lease_block(
_load_issue_lock_for_scope(
issue_number=issue_number,
remote=remote,
org=o,
repo=r,
),
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),
issue_number=issue_number,
branch_name=branch_name,
worktree_path=resolved_worktree,
@@ -1180,7 +1227,7 @@ def gitea_lock_issue(
pr_title = pr.get("title", "")
pr_body = pr.get("body", "")
if expected_pattern in pr_head:
if issue_branch_ownership.branch_tracks_issue(pr_head, issue_number):
raise ValueError(
f"Issue #{issue_number} is already tied to an open PR (PR #{pr.get('number')}, branch '{pr_head}') (fail closed)"
)
@@ -1200,13 +1247,24 @@ def gitea_lock_issue(
branches = api_get_all(branch_url, auth)
except Exception as e:
raise RuntimeError(f"Could not list branches to verify issue lock: {e}")
for branch in branches:
name = _branch_entry_name(branch)
if expected_pattern in name:
raise ValueError(
f"Issue #{issue_number} already has matching branch '{name}' "
"(fail closed)"
)
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,
@@ -1214,36 +1272,21 @@ def gitea_lock_issue(
worktree_path=resolved_worktree,
host=h,
)
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),
)
except RuntimeError:
raise
except Exception as e:
raise RuntimeError(f"Could not acquire issue lock: {e}") from e
data = {
"issue_number": issue_number,
"branch_name": branch_name,
"remote": remote,
"org": o,
"repo": r,
"worktree_path": resolved_worktree,
"work_lease": work_lease,
}
lock_file_path = _save_issue_lock(data)
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(
acquisition["record"],
freshness=acquisition["freshness"],
competing_live_locks=competing,
released=False,
)
result = {
"success": True,
"message": (
@@ -1254,12 +1297,22 @@ def gitea_lock_issue(
"branch_name": branch_name,
"worktree_path": resolved_worktree,
"work_lease": work_lease,
"lock_path": acquisition["lock_path"],
"session_id": acquisition["session_id"],
"lock_freshness": acquisition["freshness"],
"legacy_pointer": acquisition["legacy_pointer"],
"lock_proof": lock_proof,
"lock_file_path": lock_file_path,
}
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): "
@@ -1318,19 +1371,8 @@ 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 / #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
# ── Issue Lock Validation (Issue #194 / #196 / #443) ──
lock_data = _resolve_issue_lock_for_pr(remote=remote, org=o, repo=r, head=head)
locked_issue = lock_data.get("issue_number")
locked_branch = lock_data.get("branch_name")
@@ -1347,10 +1389,6 @@ def gitea_create_pr(
f"PR head branch '{head}' does not match locked branch '{locked_branch}' (fail closed)"
)
ownership = issue_lock_store.verify_lock_for_mutation(lock_data)
if ownership["block"]:
raise ValueError(ownership["reasons"][0])
# Check for forbidden terms anywhere in title/body
forbidden_terms = ["equivalent", "related", "same as"]
text_to_check = f"{title} {body}".lower()
@@ -2830,7 +2868,7 @@ def _prepare_commit_payload_files(files: list[dict]) -> tuple[list[dict], list[d
processed_files = []
source_proofs = []
lock_data = _load_existing_issue_lock() or {}
lock_data = issue_lock_store.read_session_issue_lock() or {}
locked_worktree = lock_data.get("worktree_path")
if locked_worktree:
@@ -6201,15 +6239,6 @@ def gitea_reconcile_issue_claims(
heartbeat_lease_minutes=heartbeat_lease_minutes,
reclaim_after_minutes=reclaim_after_minutes,
)
live_locks = issue_lock_store.list_live_locks()
inventory["live_issue_locks"] = live_locks
inventory["live_issue_lock_numbers"] = sorted(
{
int(entry["issue_number"])
for entry in live_locks
if entry.get("issue_number") is not None
}
)
inventory["cleanup_plan"] = issue_claim_heartbeat.build_cleanup_plan(inventory)
inventory["success"] = True
inventory["performed"] = False
+31
View File
@@ -0,0 +1,31 @@
"""Structured issue/branch ownership evidence (#440).
Author branches must follow ``(fix|feat|docs|chore)/issue-<n>-<desc>``. Duplicate-work
and recovery gates use this parser instead of broad substring checks like
``issue-420`` inside unrelated names (for example ``issue-4200``).
"""
from __future__ import annotations
import re
IMPLEMENTATION_BRANCH_RE = re.compile(
r"^(?:fix|feat|docs|chore)/issue-(\d+)(?:-.+)?$"
)
def parse_tracked_issue_number(branch_name: str) -> int | None:
"""Return the issue number encoded in a canonical author branch name."""
text = (branch_name or "").strip()
if not text:
return None
match = IMPLEMENTATION_BRANCH_RE.match(text)
if not match:
return None
return int(match.group(1))
def branch_tracks_issue(branch_name: str, issue_number: int) -> bool:
"""True when ``branch_name`` structurally belongs to ``issue_number``."""
parsed = parse_tracked_issue_number(branch_name)
return parsed == issue_number if parsed is not None else False
+117
View File
@@ -0,0 +1,117 @@
"""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
import issue_branch_ownership
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."""
requested = (requested_branch or "").strip()
matches: list[tuple[str, str | None]] = []
for entry in existing_branches or []:
name = _branch_name(entry).strip()
if issue_branch_ownership.branch_tracks_issue(name, issue_number):
matches.append((name, _branch_sha(entry)))
competing = sorted({name for name, _ in matches if name != requested})
exact = [(name, sha) for name, sha in matches if name == requested]
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,
}
+326 -424
View File
@@ -1,70 +1,220 @@
"""Atomic per-issue lock store (#438).
"""Keyed, persistent issue-lock storage (#443).
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.
Replaces the single global ``/tmp/gitea_issue_lock.json`` slot with per-issue
lock files under ``GITEA_ISSUE_LOCK_DIR`` (default
``~/.cache/gitea-tools/issue-locks``). Each MCP session binds its active lock
via a per-process pointer file so concurrent repos/issues never clobber each
other.
"""
from __future__ import annotations
import 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
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
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._+-]+")
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(value: str) -> str:
def _sanitize_segment(value: str) -> str:
text = (value or "").strip()
if not text:
return "unknown"
return "".join(c if c.isalnum() or c in "-_" else "-" for c in text)
return "_"
return _SAFE_SEGMENT_RE.sub("_", text)
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 issue_lock_path(
def lock_key(
*,
remote: str,
org: str,
repo: str,
issue_number: int,
) -> str:
return "-".join(
_sanitize_segment(part)
for part in (remote, org, repo, str(issue_number))
)
def lock_file_path(
*,
remote: str,
org: str,
repo: str,
issue_number: int,
lock_dir: str | None = None,
) -> str:
base = (lock_dir or DEFAULT_LOCK_DIR).strip()
return os.path.join(base, f"{lock_scope_key(remote, org, repo, issue_number)}.json")
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 flock_path(json_path: str) -> str:
return f"{json_path}.lock"
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 _iso(value: datetime) -> str:
return value.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
def _ensure_lock_dir(lock_dir: str | None = None) -> str:
root = (lock_dir or default_lock_dir()).strip()
os.makedirs(root, mode=0o700, exist_ok=True)
return root
def parse_timestamp(value: str | None) -> datetime | None:
def read_lock_file(path: str) -> dict[str, Any] | None:
lock_path = (path or "").strip()
if not lock_path or not os.path.exists(lock_path):
return None
try:
with open(lock_path, encoding="utf-8") as handle:
data = json.load(handle)
except (OSError, json.JSONDecodeError):
return None
return data if isinstance(data, dict) else None
def save_lock_file(path: str, data: dict[str, Any]) -> None:
lock_path = (path or "").strip()
if not lock_path:
raise ValueError("lock path is required (fail closed)")
parent = os.path.dirname(lock_path) or "."
os.makedirs(parent, mode=0o700, exist_ok=True)
payload = json.dumps(data, indent=2, sort_keys=True) + "\n"
fd, temp_path = tempfile.mkstemp(prefix=".lock-", suffix=".json", dir=parent)
try:
with os.fdopen(fd, "w", encoding="utf-8") as handle:
handle.write(payload)
handle.flush()
os.fsync(handle.fileno())
os.replace(temp_path, lock_path)
finally:
if os.path.exists(temp_path):
try:
os.remove(temp_path)
except OSError:
pass
def bind_session_lock(lock_data: dict[str, Any], lock_dir: str | None = None) -> str:
"""Persist a keyed lock and bind it to the current process session."""
remote = str(lock_data.get("remote") or "")
org = str(lock_data.get("org") or "")
repo = str(lock_data.get("repo") or "")
issue_number = int(lock_data.get("issue_number") or 0)
if not remote or not org or not repo or issue_number <= 0:
raise ValueError("lock record must include remote, org, repo, and issue_number")
root = _ensure_lock_dir(lock_dir)
path = lock_file_path(
remote=remote,
org=org,
repo=repo,
issue_number=issue_number,
lock_dir=root,
)
record = dict(lock_data)
record["lock_file_path"] = path
record["session_pid"] = os.getpid()
save_lock_file(path, record)
pointer = {
"pid": os.getpid(),
"lock_file_path": path,
"issue_number": issue_number,
"branch_name": record.get("branch_name"),
"remote": remote,
"org": org,
"repo": repo,
}
save_lock_file(session_pointer_path(root), pointer)
return path
def read_session_issue_lock(lock_dir: str | None = None) -> dict[str, Any] | None:
root = (lock_dir or default_lock_dir()).strip()
pointer = read_lock_file(session_pointer_path(root))
if not pointer:
return None
lock_path = str(pointer.get("lock_file_path") or "").strip()
if not lock_path:
return None
return read_lock_file(lock_path)
def load_issue_lock(
*,
remote: str,
org: str,
repo: str,
issue_number: int,
lock_dir: str | None = None,
) -> dict[str, Any] | None:
return read_lock_file(
lock_file_path(
remote=remote,
org=org,
repo=repo,
issue_number=issue_number,
lock_dir=lock_dir,
)
)
def iter_lock_files(lock_dir: str | None = None) -> list[str]:
root = (lock_dir or default_lock_dir()).strip()
if not os.path.isdir(root):
return []
paths: list[str] = []
for name in os.listdir(root):
if not name.endswith(".json") or name.startswith("session-"):
continue
paths.append(os.path.join(root, name))
return sorted(paths)
def find_lock_for_branch(
*,
remote: str,
org: str,
repo: str,
branch_name: str,
lock_dir: str | None = None,
) -> dict[str, Any] | None:
target = (branch_name or "").strip()
if not target:
return None
for path in iter_lock_files(lock_dir):
lock = read_lock_file(path)
if not lock:
continue
if (
str(lock.get("remote") or "") == remote
and str(lock.get("org") or "") == org
and str(lock.get("repo") or "") == repo
and str(lock.get("branch_name") or "").strip() == target
):
lock = dict(lock)
lock.setdefault("lock_file_path", path)
return lock
return None
def _lease_now(now: datetime | None = None) -> datetime:
return now or datetime.now(timezone.utc)
def _parse_lease_timestamp(value: str | None) -> datetime | None:
text = (value or "").strip()
if not text:
return None
@@ -74,410 +224,162 @@ def parse_timestamp(value: str | None) -> datetime | None:
return None
def is_process_alive(pid: int | None) -> bool:
if not pid or pid <= 0:
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
try:
os.kill(int(pid), 0)
return expires <= _lease_now(now)
def is_lease_live(lock: dict[str, Any] | None, *, now: datetime | None = None) -> bool:
if not lock:
return False
lease = lock.get("work_lease")
if not isinstance(lease, dict):
return True
except OSError as exc:
return exc.errno != errno.ESRCH
except (TypeError, ValueError):
expires = _parse_lease_timestamp(lease.get("expires_at"))
if expires is None:
return True
return expires > _lease_now(now)
def _same_realpath(left: str | None, right: str | None) -> bool:
if not left or not right:
return False
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 os.path.realpath(left) == os.path.realpath(right)
except OSError:
return left == right
def assess_same_issue_lease_conflict(
existing_lock: dict[str, Any] | None,
*,
issue_number: int,
branch_name: str,
worktree_path: str,
operation_type: str = AUTHOR_ISSUE_WORK_LEASE,
now: datetime | None = None,
) -> str | None:
"""Return a fail-closed error when a competing live lease blocks acquisition."""
if not existing_lock:
return None
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)
fd = os.open(lock_path, os.O_CREAT | os.O_RDWR, 0o600)
try:
try:
fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
except BlockingIOError as exc:
raise LockContentionError(
f"could not acquire exclusive lock on '{lock_path}'"
) from exc
yield fd
finally:
try:
fcntl.flock(fd, fcntl.LOCK_UN)
finally:
os.close(fd)
def _lease_expires_at(lock_data: dict[str, Any] | None) -> datetime | None:
if not lock_data:
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
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"))
existing_branch = existing_lock.get("branch_name")
existing_worktree = existing_lock.get("worktree_path")
same_owner = (
existing_branch == branch_name
and _same_realpath(str(existing_worktree or ""), worktree_path)
)
if is_lease_expired(existing_lock, now=now):
return (
f"Issue #{issue_number} has an expired {operation_type} lease on "
f"branch '{existing_branch}' from worktree '{existing_worktree}'. "
"Recovery review is required before takeover (fail closed)"
)
if same_owner:
return None
return (
f"Issue #{issue_number} already has an active {operation_type} lease on "
f"branch '{existing_branch}' from worktree '{existing_worktree}' "
"(fail closed)"
)
def assess_lock_freshness(
lock_data: dict[str, Any] | None,
def assess_foreign_lock_overwrite(
existing_lock: dict[str, Any] | None,
incoming_lock: dict[str, Any],
*,
now: datetime | None = None,
) -> dict[str, Any]:
"""Classify a lock as live, expired, stale, or absent."""
current = now or datetime.now(timezone.utc)
if not lock_data:
return {
"status": "absent",
"live": False,
"stale": False,
"reason": "no lock record",
}
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("pid")
pid_alive = is_process_alive(pid) if pid is not None else False
if expires_at and expires_at <= current:
return {
"status": "expired",
"live": False,
"stale": True,
"reason": f"lease expired at {expires_at.isoformat()}",
"pid_alive": pid_alive,
}
if pid is not None and not pid_alive:
return {
"status": "stale",
"live": False,
"stale": True,
"reason": f"owner pid {pid} is not alive",
"pid_alive": False,
}
return {
"status": "live",
"live": True,
"stale": False,
"reason": "lock heartbeat and lease are fresh",
"pid_alive": pid_alive,
"heartbeat_at": _iso(heartbeat_at) if heartbeat_at else None,
"expires_at": _iso(expires_at) if expires_at else None,
}
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
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 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,
work_lease: dict[str, Any],
claimant: 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)
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)"
)
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,
}
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 resolve_lock_for_issue(
*,
issue_number: int,
remote: str | None = None,
org: str | None = None,
repo: str | None = None,
lock_dir: str | None = None,
) -> dict[str, Any] | None:
"""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_lock_for_branch(
branch_name: str,
*,
lock_dir: str | None = None,
) -> 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
) -> str | None:
"""Block writes that would clobber an unrelated live lease on the same key."""
if not existing_lock:
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
same_issue = existing_lock.get("issue_number") == incoming_lock.get("issue_number")
same_branch = existing_lock.get("branch_name") == incoming_lock.get("branch_name")
same_worktree = _same_realpath(
str(existing_lock.get("worktree_path") or ""),
str(incoming_lock.get("worktree_path") or ""),
)
if same_issue and same_branch and same_worktree:
return None
if not is_lease_live(existing_lock, now=now):
return None
return (
"Refusing to overwrite a live foreign issue lock "
f"(issue #{existing_lock.get('issue_number')}, "
f"branch '{existing_lock.get('branch_name')}', "
f"worktree '{existing_lock.get('worktree_path')}') (fail closed)"
)
def find_live_lock_for_branch(
branch_name: str,
lock_dir: str | None = None,
) -> dict[str, Any] | None:
target = (branch_name or "").strip()
if not target:
return None
for path in iter_lock_files(lock_dir):
lock = read_lock_file(path)
if not lock:
continue
if str(lock.get("branch_name") or "").strip() != target:
continue
if not is_lease_live(lock):
continue
record = dict(lock)
record.setdefault("lock_file_path", path)
return record
return None
def verify_lock_for_mutation(
lock_data: dict[str, Any] | None,
*,
issue_number: int | None = None,
def resolve_locked_branch_for_session(
branch_name: str | None = None,
worktree_path: str | None = None,
) -> dict[str, Any]:
"""Re-check lock ownership immediately before a mutation (#438)."""
reasons: list[str] = []
if not lock_data:
return {"proven": False, "block": True, "reasons": ["issue lock is missing (fail closed)"]}
freshness = assess_lock_freshness(lock_data)
if not freshness["live"]:
reasons.append(f"issue lock is not live: {freshness['reason']} (fail closed)")
if issue_number is not None and lock_data.get("issue_number") != issue_number:
reasons.append(
f"issue lock targets #{lock_data.get('issue_number')}, expected #{issue_number} (fail closed)"
)
if branch_name is not None and lock_data.get("branch_name") != branch_name:
reasons.append(
f"issue lock branch '{lock_data.get('branch_name')}' does not match "
f"'{branch_name}' (fail closed)"
)
if worktree_path is not None:
locked = os.path.realpath(str(lock_data.get("worktree_path") or ""))
declared = os.path.realpath(worktree_path)
if locked != declared:
reasons.append(
f"issue lock worktree '{locked}' does not match declared '{declared}' (fail closed)"
)
return {
"proven": not reasons,
"block": bool(reasons),
"reasons": reasons,
"freshness": freshness,
"lock_proof": format_lock_proof(lock_data, freshness=freshness),
}
def list_live_locks(*, lock_dir: str | None = None, now: datetime | None = None) -> list[dict[str, Any]]:
"""Return live per-issue locks for queue visibility."""
base = (lock_dir or DEFAULT_LOCK_DIR).strip()
if not os.path.isdir(base):
return []
live: list[dict[str, Any]] = []
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 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
def format_lock_proof(
lock_data: dict[str, Any] | None,
*,
freshness: dict[str, Any] | None = None,
competing_live_locks: list[dict[str, Any]] | None = None,
released: bool | None = None,
lock_dir: str | None = None,
) -> str:
"""Canonical issue-lock proof string for final reports."""
if not lock_data:
return "issue lock proof: not acquired"
fresh = freshness or assess_lock_freshness(lock_data)
owner = lock_data.get("claimant") or {}
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('pid')}",
f"session {lock_data.get('session_id')}",
f"freshness {fresh.get('status')}",
]
if competing_live_locks is not None:
parts.append(
"no competing live lock"
if not competing_live_locks
else f"competing live locks {len(competing_live_locks)}"
)
if released is True:
parts.append("lock released")
elif released is False:
parts.append("lock retained")
return "; ".join(parts)
if branch_name:
lock = find_live_lock_for_branch(branch_name, lock_dir)
if lock:
return str(lock.get("branch_name") or "")
lock = read_session_issue_lock(lock_dir)
return str((lock or {}).get("branch_name") or "")
def has_active_issue_lock(
branch: str,
*,
lock_dir: str | None = None,
) -> bool:
target = (branch or "").strip()
if not target:
return False
for path in iter_lock_files(lock_dir):
lock = read_lock_file(path)
if not lock:
continue
if str(lock.get("branch_name") or "").strip() != target:
continue
if is_lease_live(lock):
return True
return False
+9 -18
View File
@@ -16,7 +16,6 @@ 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)
@@ -38,26 +37,18 @@ def resolve_worktree_path(project_root: str, branch: str) -> str:
def read_issue_lock(path: str | None = None) -> dict[str, Any] | None:
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
if path:
return issue_lock_store.read_lock_file(path.strip())
return issue_lock_store.read_session_issue_lock()
def has_active_issue_lock(branch: str, lock_path: str | None = None) -> bool:
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()
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)
def collect_open_pr_heads(open_prs: list[dict[str, Any]]) -> set[str]:
+6 -11
View File
@@ -37,25 +37,20 @@ 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=$(PYTHONPATH="$repo_root" python3 - <<'PY' "$branch")
locked_branch=$(python3 -c "
import sys
sys.path.insert(0, '$repo_root')
import issue_lock_store
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
)
print(issue_lock_store.resolve_locked_branch_for_session('$branch'))
")
if [[ -z "$locked_branch" ]]; then
echo "Error: No issue lock found for branch '$branch'. Lock the issue before branch creation (fail closed)." >&2
echo "Error: No session issue lock is bound. Call gitea_lock_issue before branch creation (fail closed)." >&2
exit 2
fi
if [[ "$branch" != "$locked_branch" ]]; then
@@ -309,6 +309,15 @@ Do not implement unclaimed work.
If the claim/lock gates are broken, produce a recovery handoff.
### Lost lock recovery after push (non-destructive)
If the MCP server restarts after the issue branch was pushed but before PR creation:
* **Do not** delete the remote branch as the normal recovery path.
* Call `gitea_lock_issue` again with the same issue number and exact branch name. Own-branch adoption rebinds the session when open-PR, competing-lock, and worktree safety checks pass.
* Call `gitea_create_pr` afterward. Durable keyed locks resolve from the session pointer or by matching the PR `head` branch without manual lock seeding.
* If adoption is blocked by an open PR, a competing live lease, or a different same-issue branch, stop and produce a recovery handoff.
Create a tooling issue only if this run is explicitly authorized to switch to issue-creation mode and exact `create_issue` capability is proven.
Report:
+5 -3
View File
@@ -74,18 +74,20 @@ ISSUE_WRITE_ENV = {
class TestIssueLockArtifactWarning(unittest.TestCase):
def setUp(self):
self._env_patcher = patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True)
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.start()
def tearDown(self):
self._env_patcher.stop()
self._lock_dir.cleanup()
@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, _lock_file, *_mocks):
def test_lock_success_includes_artifact_warning(self, mock_state, *_mocks):
mock_state.return_value = {
"current_branch": "master",
"porcelain_status": "?? _emit_payload.py\n",
+11 -5
View File
@@ -66,7 +66,10 @@ class TestCommitPayloads(unittest.TestCase):
)
self.locked_worktree_path = os.path.realpath(self.locked_worktree_dir.name)
self.lock_file_path = "/tmp/gitea_issue_lock.json"
import issue_lock_store
self._lock_dir = tempfile.TemporaryDirectory()
os.environ["GITEA_ISSUE_LOCK_DIR"] = self._lock_dir.name
self.lock_data = {
"issue_number": 263,
"branch_name": "feat/issue-263-native-commit-payloads",
@@ -74,9 +77,12 @@ class TestCommitPayloads(unittest.TestCase):
"org": "Example-Org",
"repo": "Example-Repo",
"worktree_path": self.locked_worktree_path,
"work_lease": {
"operation_type": "author_issue_work",
"expires_at": "2999-01-01T00:00:00Z",
},
}
with open(self.lock_file_path, "w", encoding="utf-8") as fh:
fh.write(json.dumps(self.lock_data))
self.lock_file_path = issue_lock_store.bind_session_lock(self.lock_data)
# Reset preflight status to bypass/pass verification in tests
self.orig_whoami_called = mcp_server._preflight_whoami_called
@@ -93,8 +99,7 @@ class TestCommitPayloads(unittest.TestCase):
self._dir.cleanup()
self.locked_worktree_dir.cleanup()
if os.path.exists(self.lock_file_path):
os.remove(self.lock_file_path)
self._lock_dir.cleanup()
def _env(self, profile: str) -> dict:
return {
@@ -103,6 +108,7 @@ 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")
+35
View File
@@ -0,0 +1,35 @@
"""Unit tests for structured issue/branch ownership parsing (#440)."""
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from issue_branch_ownership import ( # noqa: E402
branch_tracks_issue,
parse_tracked_issue_number,
)
class TestIssueBranchOwnership(unittest.TestCase):
def test_parses_canonical_author_branch(self):
self.assertEqual(
parse_tracked_issue_number("feat/issue-420-server-code-parity"),
420,
)
def test_issue_4200_does_not_track_issue_420(self):
self.assertEqual(parse_tracked_issue_number("feat/issue-4200-unrelated"), 4200)
self.assertFalse(branch_tracks_issue("feat/issue-4200-unrelated", 420))
def test_branch_tracks_issue_exact_match(self):
self.assertTrue(
branch_tracks_issue("fix/issue-440-branch-recovery", 440)
)
def test_unrelated_branch_does_not_track(self):
self.assertFalse(branch_tracks_issue("feat/issue-999-other", 440))
if __name__ == "__main__":
unittest.main()
+76
View File
@@ -0,0 +1,76 @@
"""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)
def test_issue_number_substring_collision_is_ignored(self):
result = assess_own_branch_adoption(
issue_number=420,
requested_branch=REQ,
existing_branches=[{"name": "feat/issue-4200-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()
+152
View File
@@ -0,0 +1,152 @@
"""Integration tests for non-destructive lock/PR recovery (#440)."""
import os
import sys
import tempfile
import unittest
from pathlib import Path
from unittest import mock
from unittest.mock import patch
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import issue_lock_adoption # noqa: E402
import issue_lock_store as ils # noqa: E402
import mcp_server # noqa: E402
from mcp_server import gitea_create_pr, gitea_lock_issue # noqa: E402
from tests.test_mcp_server import CREATE_PR_ENV, ISSUE_WRITE_ENV # noqa: E402
FAKE_AUTH = "token fake"
BRANCH = "feat/issue-420-server-code-parity"
def _lock_record(**overrides):
record = {
"issue_number": 420,
"branch_name": BRANCH,
"remote": "prgs",
"org": "Scaled-Tech-Consulting",
"repo": mcp_server.REMOTES["prgs"]["repo"],
"worktree_path": "/tmp/wt-420",
"work_lease": {
"operation_type": "author_issue_work",
"expires_at": "2999-01-01T00:00:00Z",
},
}
record.update(overrides)
return record
def _clean_git_state():
return {
"current_branch": BRANCH,
"porcelain_status": "",
"base_equivalent": True,
"inspected_git_root": os.getcwd(),
"base_branch": "master",
}
class TestIssueLockRecovery(unittest.TestCase):
def setUp(self):
self._tmpdir = tempfile.TemporaryDirectory()
self._lock_dir = self._tmpdir.name
self._env = {
**ISSUE_WRITE_ENV,
"GITEA_ISSUE_LOCK_DIR": self._lock_dir,
}
self._create_pr_env = {
**CREATE_PR_ENV,
"GITEA_ISSUE_LOCK_DIR": self._lock_dir,
}
def tearDown(self):
self._tmpdir.cleanup()
def test_adoption_after_restart_without_session_pointer(self):
with patch.dict(os.environ, self._env, clear=True):
with mock.patch("os.getpid", return_value=9999):
self.assertIsNone(ils.read_session_issue_lock())
with mock.patch(
"mcp_server.issue_lock_worktree.read_worktree_git_state",
return_value=_clean_git_state(),
), mock.patch("mcp_server.api_get_all") as mock_api, mock.patch(
"mcp_server.get_auth_header", return_value=FAKE_AUTH
):
mock_api.side_effect = [
[],
[{"name": BRANCH, "commit": {"id": "934688a"}}],
]
res = gitea_lock_issue(
issue_number=420,
branch_name=BRANCH,
remote="prgs",
worktree_path=os.path.realpath(os.getcwd()),
)
self.assertTrue(res["success"])
self.assertIn("adoption", res)
self.assertEqual(res["adoption"]["branch_head_commit"], "934688a")
@mock.patch(
"mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []),
)
@mock.patch("mcp_server.api_request")
@mock.patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_pr_resolves_keyed_lock_after_restart(self, _auth, mock_api, _role):
mock_api.return_value = {"number": 501, "html_url": "https://example/pr/501"}
worktree = os.path.realpath(os.getcwd())
with patch.dict(os.environ, self._create_pr_env, clear=True):
path = ils.lock_file_path(
remote="prgs",
org=mcp_server.REMOTES["prgs"]["org"],
repo=mcp_server.REMOTES["prgs"]["repo"],
issue_number=420,
lock_dir=self._lock_dir,
)
ils.save_lock_file(path, _lock_record(worktree_path=worktree))
with mock.patch("os.getpid", return_value=4242):
self.assertIsNone(ils.read_session_issue_lock())
res = gitea_create_pr(
title="feat: recovery Closes #420",
head=BRANCH,
base="master",
remote="prgs",
worktree_path=worktree,
)
self.assertEqual(res["number"], 501)
def test_open_pr_blocks_adoption(self):
with patch.dict(os.environ, self._env, clear=True):
with mock.patch(
"mcp_server.issue_lock_worktree.read_worktree_git_state",
return_value=_clean_git_state(),
), mock.patch("mcp_server.api_get_all") as mock_api, mock.patch(
"mcp_server.get_auth_header", return_value=FAKE_AUTH
):
mock_api.side_effect = [
[{"number": 99, "head": {"ref": BRANCH}, "title": "", "body": ""}],
[{"name": BRANCH, "commit": {"id": "934688a"}}],
]
with self.assertRaises(ValueError) as ctx:
gitea_lock_issue(
issue_number=420,
branch_name=BRANCH,
remote="prgs",
worktree_path=os.path.realpath(os.getcwd()),
)
self.assertIn("already tied to an open PR", str(ctx.exception))
def test_structured_ownership_ignores_issue_4200_collision(self):
result = issue_lock_adoption.assess_own_branch_adoption(
issue_number=420,
requested_branch=BRANCH,
existing_branches=[{"name": "feat/issue-4200-unrelated"}],
)
self.assertEqual(result["outcome"], issue_lock_adoption.NO_MATCH)
if __name__ == "__main__":
unittest.main()
+146 -171
View File
@@ -1,205 +1,180 @@
"""Tests for atomic per-issue lock store (#438)."""
from __future__ import annotations
"""Unit tests for keyed issue-lock storage (#443)."""
import json
import os
import sys
import tempfile
import threading
import unittest
from datetime import datetime, timedelta, timezone
from unittest.mock import patch
from pathlib import Path
from unittest import mock
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import issue_lock_store # noqa: E402
import issue_lock_store as ils # noqa: E402
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 _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",
}
lease.update(overrides)
return lease
def _lock_record(**overrides) -> dict:
record = {
"issue_number": 420,
"branch_name": "feat/issue-420-server-code-parity",
"remote": "prgs",
"org": "Scaled-Tech-Consulting",
"repo": "Gitea-Tools",
"worktree_path": "/tmp/wt-420",
"work_lease": _lease("2999-01-01T00:00:00Z"),
}
record.update(overrides)
return record
class TestIssueLockStore(unittest.TestCase):
def setUp(self):
self.tempdir = tempfile.mkdtemp(prefix="issue-lock-store-")
self.legacy = os.path.join(self.tempdir, "legacy.json")
self.addCleanup(self._cleanup)
self._dir = tempfile.TemporaryDirectory()
self.lock_dir = self._dir.name
self._env = mock.patch.dict(os.environ, {"GITEA_ISSUE_LOCK_DIR": self.lock_dir})
self._env.start()
def _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 tearDown(self):
self._env.stop()
self._dir.cleanup()
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,
def test_concurrent_repo_locks_do_not_overwrite(self):
lock_a = _lock_record(
issue_number=108,
branch_name="feat/issue-108-root-menu",
repo="mcp-control-plane",
worktree_path="/tmp/wt-108",
)
lock_b = _lock_record(
issue_number=420,
branch_name="feat/issue-420-server-code-parity",
repo="Gitea-Tools",
worktree_path="/tmp/wt-420",
)
path_a = ils.bind_session_lock(lock_a)
with mock.patch("os.getpid", return_value=9999):
path_b = ils.bind_session_lock(lock_b)
self.assertNotEqual(path_a, path_b)
self.assertTrue(os.path.exists(path_a))
self.assertTrue(os.path.exists(path_b))
stored_a = ils.read_lock_file(path_a)
stored_b = ils.read_lock_file(path_b)
self.assertEqual(stored_a["issue_number"], 108)
self.assertEqual(stored_b["issue_number"], 420)
def test_concurrent_issue_locks_same_repo_do_not_overwrite(self):
lock_a = _lock_record(issue_number=427, branch_name="feat/issue-427-a")
lock_b = _lock_record(issue_number=428, branch_name="feat/issue-428-b")
path_a = ils.bind_session_lock(lock_a)
with mock.patch("os.getpid", return_value=4242):
path_b = ils.bind_session_lock(lock_b)
self.assertNotEqual(path_a, path_b)
self.assertEqual(ils.read_lock_file(path_a)["issue_number"], 427)
self.assertEqual(ils.read_lock_file(path_b)["issue_number"], 428)
def test_foreign_live_lease_blocks_overwrite(self):
existing = _lock_record(
branch_name="feat/issue-420-other",
worktree_path="/tmp/other",
work_lease=_lease("2999-01-01T00:00:00Z"),
)
path = ils.lock_file_path(
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
worktree_path=worktree_path,
work_lease=_lease(issue_number=issue_number, branch=branch),
claimant={"profile": "prgs-author", "username": "jcwalker3"},
lock_dir=self.tempdir,
**kwargs,
issue_number=420,
)
ils.save_lock_file(path, existing)
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"])
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_concurrent_acquire_same_issue_only_one_wins(self):
barrier = threading.Barrier(2)
results: list[dict | Exception] = []
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 worker():
barrier.wait()
try:
results.append(self._acquire())
except Exception as exc: # noqa: BLE001
results.append(exc)
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)
threads = [threading.Thread(target=worker) for _ in range(2)]
for thread in threads:
thread.start()
for thread in threads:
thread.join()
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)
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()
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(
"active lock" in failure_text or "lock contention" in failure_text,
failures[0],
ils.has_active_issue_lock("feat/issue-420-server-code-parity")
)
self.assertFalse(ils.has_active_issue_lock("feat/issue-999-other"))
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",
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,
)
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=438,
branch_name="feat/issue-438-lock-hardening",
worktree_path="/tmp/wt",
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,
)
self.assertTrue(result["block"])
self.assertIn("not live", result["reasons"][0])
ils.save_lock_file(path_b, _lock_record())
def test_list_live_locks_excludes_stale_records(self):
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,
},
)
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,
{
"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 = 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)
self.assertTrue(os.path.exists(path_a))
self.assertTrue(os.path.exists(path_b))
self.assertEqual(ils.read_lock_file(path_a)["issue_number"], 108)
if __name__ == "__main__":
+187 -97
View File
@@ -6,6 +6,7 @@ the MCP protocol) with mocked API responses.
import json
import os
import sys
import tempfile
import unittest
from unittest.mock import patch, MagicMock
@@ -45,6 +46,7 @@ from gitea_auth import get_profile # noqa: E402
import gitea_config # noqa: E402
import mcp_server
import issue_lock_store
FAKE_AUTH = "Basic dGVzdDp0ZXN0"
@@ -97,9 +99,6 @@ CREATE_PR_ENV = {
),
}
ISSUE_LOCK_FILE = "/tmp/gitea_issue_lock.json"
def _sample_issue_lock(issue_number=123, branch_name="feat/x", **overrides):
record = {
"issue_number": issue_number,
@@ -107,11 +106,27 @@ 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": {
"operation_type": "author_issue_work",
"expires_at": "2999-01-01T00:00:00Z",
},
}
record.update(overrides)
return record
def _bind_test_lock(**overrides) -> str:
remote = overrides.get("remote", "dadeschools")
record = _sample_issue_lock(**overrides)
if remote in mcp_server.REMOTES:
profile = mcp_server.REMOTES[remote]
record.setdefault("org", profile["org"])
record.setdefault("repo", profile["repo"])
record["remote"] = remote
return issue_lock_store.bind_session_lock(record)
# ---------------------------------------------------------------------------
# Create Issue
# ---------------------------------------------------------------------------
@@ -170,18 +185,21 @@ class TestCreatePR(unittest.TestCase):
return_value=(True, []))
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@patch("os.path.exists", return_value=True)
@patch("builtins.open")
def test_creates_pr(self, mock_open, mock_exists, _auth, mock_api, _role):
lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x"))
mock_open.return_value.__enter__.return_value.read.return_value = lock_json
def test_creates_pr(self, _auth, mock_api, _role):
worktree = os.path.realpath(os.getcwd())
mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
with tempfile.TemporaryDirectory() as lock_dir:
env = {**CREATE_PR_ENV, "GITEA_ISSUE_LOCK_DIR": lock_dir}
with patch.dict(os.environ, env, clear=True):
_bind_test_lock(issue_number=123, branch_name="feat/x", worktree_path=worktree)
result = gitea_create_pr(
title="feat: X Closes #123",
head="feat/x",
base="main",
worktree_path=worktree,
)
self.assertEqual(result["number"], 3)
self.assertNotIn("url", result)
mock_exists.assert_called_with(ISSUE_LOCK_FILE)
mock_open.assert_called_with(ISSUE_LOCK_FILE, "r", encoding="utf-8")
payload = mock_api.call_args[0][3]
self.assertEqual(payload["head"], "feat/x")
self.assertEqual(payload["base"], "main")
@@ -191,30 +209,42 @@ class TestCreatePR(unittest.TestCase):
return_value=(True, []))
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@patch("os.path.exists", return_value=True)
@patch("builtins.open")
def test_create_pr_reveal_opt_in_includes_url(self, mock_open, mock_exists, _auth, mock_api, _role):
lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x"))
mock_open.return_value.__enter__.return_value.read.return_value = lock_json
def test_create_pr_reveal_opt_in_includes_url(self, _auth, mock_api, _role):
worktree = os.path.realpath(os.getcwd())
mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
env = {**CREATE_PR_ENV, "GITEA_MCP_REVEAL_ENDPOINTS": "1"}
with patch.dict(os.environ, env, clear=True):
result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
with tempfile.TemporaryDirectory() as lock_dir:
env = {**CREATE_PR_ENV, "GITEA_ISSUE_LOCK_DIR": lock_dir, "GITEA_MCP_REVEAL_ENDPOINTS": "1"}
with patch.dict(os.environ, env, clear=True):
_bind_test_lock(issue_number=123, branch_name="feat/x", worktree_path=worktree)
result = gitea_create_pr(
title="feat: X Closes #123",
head="feat/x",
base="main",
worktree_path=worktree,
)
self.assertIn("pulls/3", result["url"])
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@patch("os.path.exists", return_value=True)
@patch("builtins.open")
def test_create_pr_locked_issue_mismatch_fails(self, mock_open, mock_exists, _auth, _role):
lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x"))
mock_open.return_value.__enter__.return_value.read.return_value = lock_json
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
with self.assertRaises(ValueError) as ctx:
gitea_create_pr(title="feat: X Closes #999", head="feat/x", base="main")
def test_create_pr_locked_issue_mismatch_fails(self, _auth, _role):
worktree = os.path.realpath(os.getcwd())
with tempfile.TemporaryDirectory() as lock_dir:
env = {**CREATE_PR_ENV, "GITEA_ISSUE_LOCK_DIR": lock_dir}
with patch.dict(os.environ, env, clear=True):
_bind_test_lock(
issue_number=123,
branch_name="feat/x",
worktree_path=worktree,
)
with self.assertRaises(ValueError) as ctx:
gitea_create_pr(
title="feat: X Closes #999",
head="feat/x",
base="main",
worktree_path=worktree,
)
self.assertIn("Closes #123", str(ctx.exception))
mock_open.assert_called_with(ISSUE_LOCK_FILE, "r", encoding="utf-8")
# ---------------------------------------------------------------------------
@@ -3042,38 +3072,23 @@ class TestIssueLocking(unittest.TestCase):
"""Test issue locking and PR gating constraints."""
def setUp(self):
import shutil
import tempfile
self._lock_dir = tempfile.mkdtemp(prefix="mcp-issue-lock-")
self._session_lock = os.path.join(self._lock_dir, "session.json")
self._lock_dir = tempfile.TemporaryDirectory()
env = {
**ISSUE_WRITE_ENV,
"GITEA_ISSUE_LOCK_DIR": self._lock_dir,
"GITEA_ISSUE_LOCK_FILE": self._session_lock,
"GITEA_ISSUE_LOCK_DIR": self._lock_dir.name,
}
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()
def tearDown(self):
import shutil
for patcher in reversed(getattr(self, "_patchers", [])):
patcher.stop()
self._env_patcher.stop()
shutil.rmtree(getattr(self, "_lock_dir", ""), ignore_errors=True)
self._lock_dir.cleanup()
def _create_pr_env(self) -> dict:
return {
**CREATE_PR_ENV,
"GITEA_ISSUE_LOCK_DIR": self._lock_dir.name,
}
@patch(
"mcp_server.issue_lock_worktree.read_worktree_git_state",
@@ -3092,9 +3107,8 @@ 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.assertTrue(os.path.exists(ISSUE_LOCK_FILE))
with open(ISSUE_LOCK_FILE, encoding="utf-8") as f:
lock = json.load(f)
self.assertIn("lock_file_path", res)
lock = issue_lock_store.read_lock_file(res["lock_file_path"])
self.assertIn("worktree_path", lock)
self.assertIn("work_lease", lock)
@@ -3150,34 +3164,85 @@ class TestIssueLocking(unittest.TestCase):
]
with self.assertRaises(ValueError) as ctx:
gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
self.assertIn("already has matching branch", str(ctx.exception))
self.assertIn("not the requested branch", str(ctx.exception))
def test_lock_issue_blocks_active_same_operation_lease(self):
with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
json.dump({
@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"
mock_api.side_effect = [
[],
[{"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,
),
{
"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))
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({
@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,
),
{
"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))
@@ -3244,9 +3309,7 @@ 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):
if os.path.exists(ISSUE_LOCK_FILE):
os.remove(ISSUE_LOCK_FILE)
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
with patch.dict(os.environ, self._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))
@@ -3255,37 +3318,64 @@ 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):
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):
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 self.assertRaises(ValueError) as ctx:
gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-different", remote="prgs")
gitea_create_pr(
title="feat: X Closes #196",
head="feat/issue-196-different",
remote="prgs",
worktree_path=worktree,
)
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):
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):
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):
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")
gitea_create_pr(
title=f"feat: X {term}",
head="feat/issue-196-mutations",
remote="prgs",
worktree_path=worktree,
)
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):
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):
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 self.assertRaises(ValueError) as ctx:
gitea_create_pr(title="feat: X refs #196", head="feat/issue-196-mutations", remote="prgs")
gitea_create_pr(
title="feat: X refs #196",
head="feat/issue-196-mutations",
remote="prgs",
worktree_path=worktree,
)
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",
@@ -3293,13 +3383,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")
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):
_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 self.assertRaises(ValueError) as ctx:
gitea_create_pr(
title="feat: lock scratch worktree Closes #249",
@@ -3316,13 +3406,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"}
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):
_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):
res = gitea_create_pr(
title="feat: issue-lock scratch worktree Closes #249",
head="feat/issue-249-issue-lock-scratch-worktree",
+26 -9
View File
@@ -20,33 +20,50 @@ def run(script, *args):
branch = arg
break
lock_file = Path("/tmp/gitea_issue_lock.json")
created_lock = False
lock_dir_ctx = None
extra_env = os.environ.copy()
if script == "worktree-start" and branch:
import re
import json
import tempfile
import issue_lock_store
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_file.write_text(json.dumps({
lock_dir_ctx = tempfile.TemporaryDirectory()
extra_env["GITEA_ISSUE_LOCK_DIR"] = lock_dir_ctx.name
record = {
"issue_number": issue_num,
"branch_name": branch,
"remote": "prgs",
"org": "Scaled-Tech-Consulting",
"repo": "Gitea-Tools"
}), encoding="utf-8")
created_lock = True
"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)
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 created_lock and lock_file.exists():
lock_file.unlink()
if lock_dir_ctx is not None:
lock_dir_ctx.cleanup()
class TestWorktreeStart(unittest.TestCase):