feat: enforce author work leases (Closes #267)

gitea_mcp_server.py

@@ -21,6 +21,7 @@ import json
 import functools
 import contextlib
 import subprocess
+from datetime import datetime, timedelta, timezone
 
 
 # Mutation-authority record (#199, refs #194). Deliberately in-process, NOT a
@@ -482,6 +483,132 @@ import merged_cleanup_reconcile  # noqa: E402
 # Fail-closed exact-issue-lock file (#204): written by gitea_lock_issue,
 # consumed by gitea_create_pr and scripts/worktree-start.
 ISSUE_LOCK_FILE = "/tmp/gitea_issue_lock.json"
+WORK_LEASE_TTL_HOURS = 4
+AUTHOR_ISSUE_WORK_LEASE = "author_issue_work"
+VALID_WORK_LEASE_OPERATIONS = frozenset({
+    AUTHOR_ISSUE_WORK_LEASE,
+    "review_pr_work",
+    "fix_pr_changes",
+    "cleanup_branch_work",
+    "issue_filing_work",
+    "recovery_work",
+})
+
+
+def _work_lease_now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _work_lease_timestamp(value: datetime) -> str:
+    return value.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
+
+
+def _parse_work_lease_timestamp(value: str | None) -> datetime | None:
+    text = (value or "").strip()
+    if not text:
+        return None
+    try:
+        return datetime.fromisoformat(text.replace("Z", "+00:00")).astimezone(timezone.utc)
+    except ValueError:
+        return None
+
+
+def _load_existing_issue_lock() -> dict | None:
+    if not os.path.exists(ISSUE_LOCK_FILE):
+        return None
+    try:
+        with open(ISSUE_LOCK_FILE, encoding="utf-8") as f:
+            data = json.load(f)
+        return data if isinstance(data, dict) else None
+    except Exception:
+        return None
+
+
+def _work_lease_claimant(host: str | None) -> dict:
+    profile = get_profile()
+    username = _IDENTITY_CACHE.get(host) if host else None
+    if username is None and host and not _preflight_in_test_mode():
+        username = _authenticated_username(host)
+    return {
+        "username": username,
+        "profile": profile.get("profile_name"),
+    }
+
+
+def _build_author_issue_work_lease(
+    *,
+    issue_number: int,
+    branch_name: str,
+    worktree_path: str,
+    host: str | None,
+) -> dict:
+    created = _work_lease_now()
+    expires = created + timedelta(hours=WORK_LEASE_TTL_HOURS)
+    return {
+        "operation_type": AUTHOR_ISSUE_WORK_LEASE,
+        "issue_number": issue_number,
+        "pr_number": None,
+        "branch": branch_name,
+        "worktree_path": worktree_path,
+        "claimant": _work_lease_claimant(host),
+        "created_at": _work_lease_timestamp(created),
+        "expires_at": _work_lease_timestamp(expires),
+        "last_heartbeat_at": _work_lease_timestamp(created),
+    }
+
+
+def _active_work_lease_block(
+    existing_lock: dict | None,
+    *,
+    issue_number: int,
+    branch_name: str,
+    worktree_path: str,
+    operation_type: str,
+) -> str | None:
+    if not existing_lock:
+        return None
+    existing_issue = existing_lock.get("issue_number")
+    existing_branch = existing_lock.get("branch_name")
+    existing_worktree = existing_lock.get("worktree_path")
+    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
+
+    same_owner = (
+        existing_branch == branch_name
+        and os.path.realpath(str(existing_worktree or "")) == os.path.realpath(worktree_path)
+    )
+    expires_at = (
+        _parse_work_lease_timestamp(lease.get("expires_at"))
+        if isinstance(lease, dict)
+        else None
+    )
+    if expires_at and expires_at <= _work_lease_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 _branch_entry_name(branch: dict | str) -> str:
+    if isinstance(branch, str):
+        return branch
+    if not isinstance(branch, dict):
+        return ""
+    return str(branch.get("name") or branch.get("ref") or "")
 
 
 def _reveal_endpoints() -> bool:
@@ -929,6 +1056,16 @@ def gitea_lock_issue(
     resolved_worktree = issue_lock_worktree.resolve_author_worktree_path(
         worktree_path, PROJECT_ROOT
     )
+    active_lease_block = _active_work_lease_block(
+        _load_existing_issue_lock(),
+        issue_number=issue_number,
+        branch_name=branch_name,
+        worktree_path=resolved_worktree,
+        operation_type=AUTHOR_ISSUE_WORK_LEASE,
+    )
+    if active_lease_block:
+        raise RuntimeError(active_lease_block)
+
     git_state = issue_lock_worktree.read_worktree_git_state(resolved_worktree)
     verify_preflight_purity(remote, worktree_path=resolved_worktree)
     lock_assessment = issue_lock_worktree.assess_issue_lock_worktree(
@@ -974,6 +1111,25 @@ def gitea_lock_issue(
                 f"Issue #{issue_number} is already tied to an open PR (PR #{pr.get('number')}) via Closes/Fixes reference (fail closed)"
             )
 
+    branch_url = f"{repo_api_url(h, o, r)}/branches"
+    try:
+        branches = api_get_all(branch_url, auth)
+    except Exception as e:
+        raise RuntimeError(f"Could not list branches to verify issue lock: {e}")
+    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)"
+            )
+
+    work_lease = _build_author_issue_work_lease(
+        issue_number=issue_number,
+        branch_name=branch_name,
+        worktree_path=resolved_worktree,
+        host=h,
+    )
     data = {
         "issue_number": issue_number,
         "branch_name": branch_name,
@@ -981,6 +1137,7 @@ def gitea_lock_issue(
         "org": o,
         "repo": r,
         "worktree_path": resolved_worktree,
+        "work_lease": work_lease,
     }
 
     try:
@@ -1001,6 +1158,7 @@ def gitea_lock_issue(
         "issue_number": issue_number,
         "branch_name": branch_name,
         "worktree_path": resolved_worktree,
+        "work_lease": work_lease,
     }
     if agent_artifacts:
         result["warnings"] = [