diff --git a/docs/llm-workflow-runbooks.md b/docs/llm-workflow-runbooks.md index 09b4d5c..439e653 100644 --- a/docs/llm-workflow-runbooks.md +++ b/docs/llm-workflow-runbooks.md @@ -274,12 +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 the issue-lock payload with issue -number, optional PR number, branch, worktree path, claimant identity/profile, +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 lease still -blocks takeover until a recovery review records why the prior work is abandoned, -completed, or unsafe to continue. +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. + +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 diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 17938e3..40eb6f9 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -539,6 +539,7 @@ import review_proofs # noqa: E402 import agent_temp_artifacts import issue_lock_worktree # noqa: E402 import issue_lock_provenance # noqa: E402 +import issue_lock_store # noqa: E402 import already_landed_reconcile # noqa: E402 import author_mutation_worktree # noqa: E402 import issue_claim_heartbeat # noqa: E402 @@ -584,14 +585,22 @@ def _parse_work_lease_timestamp(value: str | None) -> datetime | None: def _load_existing_issue_lock() -> dict | None: - if not os.path.exists(ISSUE_LOCK_FILE): - return None - try: - with open(ISSUE_LOCK_FILE, "r", encoding="utf-8") as f: - data = json.load(f) - return data if isinstance(data, dict) else None - except Exception: - return None + return issue_lock_store.read_lock_file(ISSUE_LOCK_FILE) + + +def _load_issue_lock_for_scope( + *, + 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() def _work_lease_claimant(host: str | None) -> dict: @@ -1241,8 +1250,14 @@ def gitea_lock_issue( resolved_worktree = issue_lock_worktree.resolve_author_worktree_path( worktree_path, PROJECT_ROOT ) + h, o, r = _resolve(remote, host, org, repo) active_lease_block = _active_work_lease_block( - _load_existing_issue_lock(), + _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, @@ -1266,7 +1281,6 @@ def gitea_lock_issue( issue_lock_worktree.format_issue_lock_worktree_error(lock_assessment) ) - h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) duplicate_gate = _assess_issue_duplicate_gate( issue_number, @@ -1288,29 +1302,41 @@ def gitea_lock_issue( 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: - with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f: - json.dump(data, f) + 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 write issue lock file: {e}") + raise RuntimeError(f"Could not acquire issue lock: {e}") from e 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": ( @@ -1321,6 +1347,11 @@ 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, } if agent_artifacts: result["warnings"] = [ @@ -1413,15 +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) ── - if not os.path.exists(ISSUE_LOCK_FILE): + # ── 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.") - try: - with open(ISSUE_LOCK_FILE, "r", encoding="utf-8") as f: - lock_data = json.load(f) - except Exception as e: - raise RuntimeError(f"Could not read issue lock file: {e} (fail closed)") + 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 @@ -1446,6 +1481,10 @@ 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() @@ -2940,13 +2979,7 @@ def _prepare_commit_payload_files(files: list[dict]) -> tuple[list[dict], list[d processed_files = [] source_proofs = [] - lock_data = {} - if os.path.exists(ISSUE_LOCK_FILE): - try: - with open(ISSUE_LOCK_FILE, "r", encoding="utf-8") as f: - lock_data = json.load(f) - except Exception: - pass + lock_data = _load_existing_issue_lock() or {} locked_worktree = lock_data.get("worktree_path") if locked_worktree: @@ -6331,6 +6364,15 @@ 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 diff --git a/issue_lock_store.py b/issue_lock_store.py new file mode 100644 index 0000000..e115c4f --- /dev/null +++ b/issue_lock_store.py @@ -0,0 +1,486 @@ +"""Atomic per-issue lock store (#438). + +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 + +import errno +import fcntl +import json +import os +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 + + +class LockContentionError(RuntimeError): + """Raised when an exclusive per-issue lock cannot be acquired.""" + + +def _sanitize(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) + + +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( + 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") + + +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 + try: + os.kill(int(pid), 0) + return True + except OSError as exc: + return exc.errno != errno.ESRCH + except (TypeError, ValueError): + 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) + 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: + 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")) + + +def assess_lock_freshness( + lock_data: dict[str, Any] | None, + *, + now: datetime | None = None, +) -> dict[str, Any]: + """Classify a lock as live, expired, stale, or absent.""" + current = 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_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) + + 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, + } + 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 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 + 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( + lock_data: dict[str, Any] | None, + *, + issue_number: int | None = None, + branch_name: str | None = None, + worktree_path: str | None = None, +) -> dict[str, Any]: + """Re-check lock ownership immediately before a mutation (#438).""" + reasons: list[str] = [] + if not lock_data: + return {"proven": False, "block": True, "reasons": ["issue lock is missing (fail closed)"]} + + freshness = assess_lock_freshness(lock_data) + if not freshness["live"]: + reasons.append(f"issue lock is not live: {freshness['reason']} (fail closed)") + + if issue_number is not None and lock_data.get("issue_number") != issue_number: + reasons.append( + f"issue lock targets #{lock_data.get('issue_number')}, expected #{issue_number} (fail closed)" + ) + + if branch_name is not None and lock_data.get("branch_name") != branch_name: + reasons.append( + f"issue lock branch '{lock_data.get('branch_name')}' does not match " + f"'{branch_name}' (fail closed)" + ) + + if worktree_path is not None: + locked = os.path.realpath(str(lock_data.get("worktree_path") or "")) + declared = os.path.realpath(worktree_path) + if locked != declared: + reasons.append( + f"issue lock worktree '{locked}' does not match declared '{declared}' (fail closed)" + ) + + return { + "proven": not reasons, + "block": bool(reasons), + "reasons": reasons, + "freshness": freshness, + "lock_proof": format_lock_proof(lock_data, freshness=freshness), + } + + +def list_live_locks(*, lock_dir: str | None = None, now: datetime | None = None) -> list[dict[str, Any]]: + """Return live per-issue locks for queue visibility.""" + 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, +) -> 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) \ No newline at end of file diff --git a/merged_cleanup_reconcile.py b/merged_cleanup_reconcile.py index 60204f6..0fab716 100644 --- a/merged_cleanup_reconcile.py +++ b/merged_cleanup_reconcile.py @@ -13,6 +13,7 @@ import subprocess from typing import Any from reviewer_worktree import parse_dirty_tracked_files +import issue_lock_store PROTECTED_BRANCHES = frozenset({"master", "main", "dev"}) ISSUE_LOCK_FILE = os.environ.get("GITEA_ISSUE_LOCK_FILE", "/tmp/gitea_issue_lock.json") @@ -49,6 +50,10 @@ def read_issue_lock(path: str | None = None) -> dict[str, Any] | None: 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 diff --git a/scripts/worktree-start b/scripts/worktree-start index 09c3bee..771ac7a 100755 --- a/scripts/worktree-start +++ b/scripts/worktree-start @@ -37,14 +37,27 @@ 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 - if [[ ! -f "/tmp/gitea_issue_lock.json" ]]; then - echo "Error: Issue lock file '/tmp/gitea_issue_lock.json' is missing. You must lock exactly one issue before branch creation (fail closed)." >&2 + locked_branch=$(PYTHONPATH="$repo_root" python3 - <<'PY' "$branch") +import sys +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 + ) + if [[ -z "$locked_branch" ]]; then + echo "Error: No issue lock found for branch '$branch'. Lock the issue before branch creation (fail closed)." >&2 exit 2 fi - locked_branch=$(python3 -c "import json; print(json.load(open('/tmp/gitea_issue_lock.json')).get('branch_name', ''))") if [[ "$branch" != "$locked_branch" ]]; then echo "Error: Requested branch '$branch' does not match locked branch '$locked_branch' (fail closed)." >&2 exit 2 @@ -68,8 +81,6 @@ EOF fi fi -script_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" -repo_root="$(cd "$script_dir/.." && pwd)" worktree_name="${branch//\//-}" worktree_path="$repo_root/branches/$worktree_name" diff --git a/tests/test_issue_lock_store.py b/tests/test_issue_lock_store.py new file mode 100644 index 0000000..2a5468c --- /dev/null +++ b/tests/test_issue_lock_store.py @@ -0,0 +1,206 @@ +"""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 unittest.mock import patch + +sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) + +import issue_lock_store # 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"), + } + lease.update(overrides) + return lease + + +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) + + 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 _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", + worktree_path=worktree_path, + work_lease=_lease(issue_number=issue_number, branch=branch), + claimant={"profile": "prgs-author", "username": "jcwalker3"}, + lock_dir=self.tempdir, + **kwargs, + ) + + 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_concurrent_acquire_same_issue_only_one_wins(self): + barrier = threading.Barrier(2) + results: list[dict | Exception] = [] + + def worker(): + barrier.wait() + try: + results.append(self._acquire()) + except Exception as exc: # noqa: BLE001 + results.append(exc) + + threads = [threading.Thread(target=worker) for _ in range(2)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + successes = [item for item in results if 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 lock" in failure_text or "lock contention" in failure_text, + failures[0], + ) + + 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", + ) + 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", + ) + self.assertTrue(result["block"]) + self.assertIn("not live", result["reasons"][0]) + + 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) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index aabb7e7..e672030 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -3069,8 +3069,30 @@ class TestIssueLocking(unittest.TestCase): """Test issue locking and PR gating constraints.""" def setUp(self): - self._env_patcher = patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True) + 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, + "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"}), @@ -3078,10 +3100,13 @@ class TestIssueLocking(unittest.TestCase): self.mock_dup_fetcher = self._dup_fetcher_patcher.start() def tearDown(self): + import shutil + self._dup_fetcher_patcher.stop() + for patcher in reversed(getattr(self, "_patchers", [])): + patcher.stop() self._env_patcher.stop() - if os.path.exists(ISSUE_LOCK_FILE): - os.remove(ISSUE_LOCK_FILE) + shutil.rmtree(getattr(self, "_lock_dir", ""), ignore_errors=True) @patch( "mcp_server.issue_lock_worktree.read_worktree_git_state",