diff --git a/docs/llm-workflow-runbooks.md b/docs/llm-workflow-runbooks.md index d67769d..69caeb7 100644 --- a/docs/llm-workflow-runbooks.md +++ b/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="")` 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 diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 411a80b..8ab9ea3 100644 --- a/gitea_mcp_server.py +++ b/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, diff --git a/preflight_contract.py b/preflight_contract.py new file mode 100644 index 0000000..60859c7 --- /dev/null +++ b/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." + ) \ No newline at end of file diff --git a/tests/test_preflight_read_survival.py b/tests/test_preflight_read_survival.py index 6da8af6..c422fc6 100644 --- a/tests/test_preflight_read_survival.py +++ b/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() \ No newline at end of file