docs/llm-workflow-runbooks.md

@@ -254,7 +254,20 @@ Never create a parallel branch or PR for the same issue unless the old branch
 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. Full portable wording:
+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,
+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.
+
+Remote branches matching the issue number are also treated as active work unless
+the recovery review proves the branch is abandoned or superseded. Never delete
+or clean up a branch when it has an active lease, dirty worktree, open PR, or is
+the only copy of unmerged work.
+
+Full portable wording:
 [`skills/llm-project-workflow/SKILL.md`](../skills/llm-project-workflow/SKILL.md).
 
 ## Global LLM Worktree Rule

docs/wiki/Workflow.md

@@ -17,6 +17,10 @@
 0. Work Selection Rule — verify a work lease before any mutations (open PRs,
    issue-linked PRs, branches, worktrees, dirty worktrees, active leases/
    handoffs, merged-PR completion). Stop if another session owns the lease.
+   `gitea_lock_issue` records an operation-scoped `author_issue_work` lease
+   with issue, branch, worktree, claimant, created, expiry, and heartbeat
+   fields; active or expired same-operation leases require recovery review
+   before takeover.
 0b. Global LLM Worktree Rule — main checkout on `master`/`main`/`dev` only;
    mutate only from a `branches/` worktree after proving root, cwd, branch,
    stable main-checkout branch, and session worktree path (no exceptions).
@@ -31,4 +35,4 @@
 Wiki publication, credential provisioning, and cross-repo mirror operations are
 operator-confirmed actions — see [Runbooks](Runbooks.md).
 
-Full Gitea-specific runbooks: `docs/llm-workflow-runbooks.md` in the repository.
+Full Gitea-specific runbooks: `docs/llm-workflow-runbooks.md` in the repository.

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"] = [

tests/test_mcp_server.py

@@ -3048,10 +3048,18 @@ class TestIssueLocking(unittest.TestCase):
         mock_api.return_value = [] # no open PRs
         res = gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
         self.assertTrue(res["success"])
+        self.assertEqual(res["work_lease"]["operation_type"], "author_issue_work")
+        self.assertEqual(res["work_lease"]["issue_number"], 196)
+        self.assertEqual(res["work_lease"]["branch"], "feat/issue-196-mutations")
+        self.assertIn("created_at", res["work_lease"])
+        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("worktree_path", lock)
+        self.assertIn("work_lease", lock)
 
     def test_lock_issue_mismatch_branch_fails(self):
         with self.assertRaises(ValueError) as ctx:
@@ -3092,6 +3100,51 @@ class TestIssueLocking(unittest.TestCase):
             gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
         self.assertIn("already tied to an open PR", str(ctx.exception))
 
+    @patch(
+        "mcp_server.issue_lock_worktree.read_worktree_git_state",
+        return_value=_clean_master_git_state_for_lock(),
+    )
+    @patch("mcp_server.api_get_all")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_lock_issue_reused_by_remote_branch(self, _auth, mock_api, _git_state):
+        mock_api.side_effect = [
+            [],
+            [{"name": "feat/issue-196-existing-work"}],
+        ]
+        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))
+
+    def test_lock_issue_blocks_active_same_operation_lease(self):
+        with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
+            json.dump({
+                "issue_number": 196,
+                "branch_name": "feat/issue-196-other-work",
+                "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({
+                "issue_number": 196,
+                "branch_name": "feat/issue-196-other-work",
+                "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))
+
     @patch("mcp_server.api_get_all", return_value=[])
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_lock_from_clean_scratch_worktree(self, _auth, _api):