docs/llm-workflow-runbooks.md

@@ -492,6 +492,25 @@ Root-level matches are listed in `.gitignore` so they never get committed.
 `gitea_get_runtime_context` and `gitea_lock_issue` surface **warnings** (not
 hard blocks) when these artifacts are still present.
 
+## Capability preflight lifetime (#470)
+
+After `gitea_resolve_task_capability(task=…)` proves the mutation is allowed,
+interleaved **read-only** calls preserve that proof until a gated mutation
+consumes it:
+
+- Safe reads: `gitea_whoami`, `gitea_view_pr`, `gitea_view_issue`, `gitea_list_*`,
+  `gitea_get_runtime_context`, `gitea_check_pr_eligibility`, and related
+  read-only inventory/eligibility tools (see `preflight_contract.py`).
+- Each mutation consumes the proof once; call `gitea_resolve_task_capability`
+  again immediately before the next mutation on the same task.
+- Resolving capability for a **different** task replaces the prior task binding.
+- Workspace edits before resolve, profile switches, or a dirty whoami baseline
+  invalidate proof (fail closed).
+
+If a mutation fails with “capability has not been resolved” or “task mismatch”,
+re-run `gitea_resolve_task_capability(task="<mutation>")` immediately before
+retrying — do not guess or skip the resolve step.
+
 Implementation work and review work must use separate branch folders. For
 example, an implementation branch might live under
 `branches/fix-issue-123-example`, while a review branch for the resulting PR

gitea_mcp_server.py

@@ -507,7 +507,17 @@ def verify_preflight_purity(
         )
     if not _preflight_capability_called:
         raise RuntimeError(
-            "Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)"
+            preflight_contract.format_missing_capability_error(task)
+        )
+    if (
+        task is not None
+        and _preflight_resolved_task is not None
+        and task != _preflight_resolved_task
+    ):
+        raise RuntimeError(
+            preflight_contract.format_task_mismatch_error(
+                _preflight_resolved_task, task
+            )
         )
     if (
         task is not None
@@ -606,6 +616,7 @@ import stacked_pr_support  # noqa: E402
 import merge_approval_gate  # noqa: E402
 import already_landed_reconcile  # noqa: E402
 import author_mutation_worktree  # noqa: E402
+import preflight_contract  # noqa: E402
 import issue_claim_heartbeat  # noqa: E402
 import issue_work_duplicate_gate  # noqa: E402
 import reviewer_pr_lease  # noqa: E402
@@ -6982,7 +6993,9 @@ def gitea_acquire_conflict_fix_lease(
         task_capability_map.required_permission("comment_issue"))
     if blocked:
         return blocked
-    verify_preflight_purity(remote, worktree_path=worktree_path)
+    verify_preflight_purity(
+        remote, worktree_path=worktree_path, task="comment_issue"
+    )
     comments = _list_pr_lease_comments(
         pr_number,
         remote=remote,

preflight_contract.py

@@ -0,0 +1,66 @@
+"""Capability preflight lifetime contract (#470).
+
+Defines which read-only MCP tools preserve an existing capability proof and the
+canonical sequencing operators must follow before mutations.
+"""
+
+from __future__ import annotations
+
+# Read-only tools that must not invalidate capability proof (#470).
+# gitea_whoami is read-only but re-pins identity; capability is preserved when
+# identity was already verified clean (#469).
+READ_ONLY_PREFLIGHT_TOOLS = frozenset({
+    "gitea_whoami",
+    "gitea_get_authenticated_user",
+    "gitea_get_current_user",
+    "gitea_view_pr",
+    "gitea_view_issue",
+    "gitea_list_prs",
+    "gitea_list_issues",
+    "gitea_list_issue_comments",
+    "gitea_get_runtime_context",
+    "gitea_resolve_task_capability",
+    "gitea_check_pr_eligibility",
+    "gitea_get_pr_review_feedback",
+    "gitea_assess_work_issue_duplicate",
+    "gitea_route_task_session",
+    "gitea_audit_config",
+    "gitea_list_labels",
+    "gitea_get_profile",
+    "gitea_list_profiles",
+})
+
+PREFLIGHT_CONTRACT_SUMMARY = (
+    "Capability preflight is session-scoped per MCP process, profile, and task. "
+    "After gitea_resolve_task_capability(task=...), interleaved read-only calls "
+    "(whoami, view_*, list_*, get_runtime_context, eligibility checks) preserve "
+    "the proof until a gated mutation consumes it. Each mutation consumes the proof "
+    "once; re-resolve immediately before the next mutation. A new resolve for a "
+    "different task replaces the prior task binding. Profile/session changes or "
+    "workspace edits before resolve invalidate proof."
+)
+
+
+def format_missing_capability_error(task: str | None = None) -> str:
+    base = (
+        "Pre-flight order violation: Task capability "
+        "(gitea_resolve_task_capability) has not been resolved (fail closed)"
+    )
+    if task:
+        return (
+            f"{base}. Re-run gitea_resolve_task_capability(task=\"{task}\") "
+            "immediately before this mutation."
+        )
+    return (
+        f"{base}. Re-run gitea_resolve_task_capability for the mutation task "
+        "immediately before acting."
+    )
+
+
+def format_task_mismatch_error(resolved: str, required: str) -> str:
+    return (
+        "Pre-flight task mismatch: "
+        f"resolved '{resolved}' but mutation requires '{required}' (fail closed). "
+        f"Re-run gitea_resolve_task_capability(task=\"{required}\") "
+        "immediately before this mutation."
+    )

tests/test_preflight_read_survival.py

@@ -1,4 +1,4 @@
-"""#469: capability preflight survives interleaved read-only whoami calls."""
+"""#469/#470: capability preflight lifetime across safe read-only calls."""
 
 import os
 import unittest
@@ -101,6 +101,59 @@ class TestPreflightReadSurvival(unittest.TestCase):
         )
         mcp_server.verify_preflight_purity(task="review_pr")
 
+    def test_close_pr_sequence_with_interleaved_reads(self):
+        """resolve(close_pr) → whoami/view reads → close_pr gate (#470)."""
+        mcp_server.record_preflight_check("whoami")
+        mcp_server.record_preflight_check(
+            "capability", resolved_role="reconciler", resolved_task="close_pr"
+        )
+        # Simulate live-state revalidation between resolve and mutation.
+        mcp_server.record_preflight_check("whoami")
+        self.assertTrue(mcp_server._preflight_capability_called)
+        self.assertEqual(mcp_server._preflight_resolved_task, "close_pr")
+        mcp_server.verify_preflight_purity(task="close_pr")
+        self.assertFalse(mcp_server._preflight_capability_called)
+
+    def test_fresh_capability_resolve_replaces_prior_task(self):
+        mcp_server.record_preflight_check("whoami")
+        mcp_server.record_preflight_check(
+            "capability", resolved_role="author", resolved_task="create_issue"
+        )
+        mcp_server.record_preflight_check(
+            "capability", resolved_role="reconciler", resolved_task="close_pr"
+        )
+        self.assertEqual(mcp_server._preflight_resolved_task, "close_pr")
+        mcp_server.verify_preflight_purity(task="close_pr")
+
+    def test_missing_capability_error_names_re_resolve(self):
+        mcp_server.record_preflight_check("whoami")
+        with self.assertRaises(RuntimeError) as ctx:
+            mcp_server.verify_preflight_purity(task="close_pr")
+        msg = str(ctx.exception)
+        self.assertIn("gitea_resolve_task_capability", msg)
+        self.assertIn('task="close_pr"', msg)
+
+    def test_task_mismatch_error_names_re_resolve(self):
+        mcp_server.record_preflight_check("whoami")
+        mcp_server.record_preflight_check(
+            "capability", resolved_role="author", resolved_task="create_issue"
+        )
+        with self.assertRaises(RuntimeError) as ctx:
+            mcp_server.verify_preflight_purity(task="close_pr")
+        msg = str(ctx.exception)
+        self.assertIn("task mismatch", msg)
+        self.assertIn('task="close_pr"', msg)
+
+    def test_consumed_capability_error_names_re_resolve(self):
+        mcp_server.record_preflight_check("whoami")
+        mcp_server.record_preflight_check(
+            "capability", resolved_role="reconciler", resolved_task="close_pr"
+        )
+        mcp_server.verify_preflight_purity(task="close_pr")
+        with self.assertRaises(RuntimeError) as ctx:
+            mcp_server.verify_preflight_purity(task="close_pr")
+        self.assertIn('task="close_pr"', str(ctx.exception))
+
 
 if __name__ == "__main__":
     unittest.main()