diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 669e9aa..0e87c7a 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -623,101 +623,280 @@ def verify_preflight_purity( remote: str | None = None, worktree_path: str | None = None, task: str | None = None, + *, + target_issue_number: int | None = None, + require_author_lock: bool = False, ): - """Verify that identity and capability were verified prior to session edits.""" + """Verify identity/capability order, then production workspace guards. + + #683: pytest/unittest must not skip production root/branches/scope + enforcement when force-on signals request production behavior. The + early return below only skips *preflight-order* purity checks under + pure unit-test isolation — never when production guards are active. + """ global _preflight_reviewer_violation_files in_test = _preflight_in_test_mode() - if in_test and not ( - os.environ.get("GITEA_TEST_FORCE_DIRTY") - or os.environ.get("GITEA_TEST_PORCELAIN") is not None - ): - return + production_active = workflow_scope_guard.production_guards_active( + in_test_mode=in_test + ) + # Pure unit-test isolation: skip purity-order unless legacy dirty/porcelain + # force flags request the dirtiness path. #683 FORCE_PRODUCTION_GUARDS alone + # runs production root/branches/scope without requiring whoami/capability. + skip_purity_order = in_test and not workflow_scope_guard.purity_order_forced() - if not _preflight_whoami_called: - raise RuntimeError( - "Pre-flight order violation: Identity (gitea_whoami) has not been verified (fail closed)" - ) - if not _preflight_capability_called: - raise RuntimeError( - "Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)" - ) - if ( - task is not None - and _preflight_resolved_task is not None - and task != _preflight_resolved_task - ): - raise RuntimeError( - "Pre-flight task mismatch: " - f"resolved '{_preflight_resolved_task}' but mutation requires " - f"'{task}' (fail closed)" - ) - - # #671: block review/merge/close/completion mutations while the session is - # contaminated by a direct stable-branch push attempt (reconciler-exempt). - _enforce_stable_branch_contamination_gate(task, remote) - - ctx = _resolve_namespace_mutation_context(worktree_path) - workspace = ctx["workspace_path"] - canonical_root = ctx["canonical_repo_root"] - process_root = ctx["process_project_root"] - real_workspace = os.path.realpath(workspace) - role = ctx.get("workspace_role_kind") or _effective_workspace_role() - - if real_workspace != process_root: - if not _preflight_in_test_mode(): - membership = author_mutation_worktree.assess_workspace_repo_membership( - workspace_path=workspace, - canonical_repo_root=canonical_root, + if not skip_purity_order: + if not _preflight_whoami_called: + raise RuntimeError( + "Pre-flight order violation: Identity (gitea_whoami) has not been verified (fail closed)" ) - if membership["block"]: + if not _preflight_capability_called: + raise RuntimeError( + "Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)" + ) + if ( + task is not None + and _preflight_resolved_task is not None + and task != _preflight_resolved_task + ): + raise RuntimeError( + "Pre-flight task mismatch: " + f"resolved '{_preflight_resolved_task}' but mutation requires " + f"'{task}' (fail closed)" + ) + + # #671: block review/merge/close/completion mutations while the session is + # contaminated by a direct stable-branch push attempt (reconciler-exempt). + _enforce_stable_branch_contamination_gate(task, remote) + + ctx = _resolve_namespace_mutation_context(worktree_path) + workspace = ctx["workspace_path"] + canonical_root = ctx["canonical_repo_root"] + process_root = ctx["process_project_root"] + real_workspace = os.path.realpath(workspace) + role = ctx.get("workspace_role_kind") or _effective_workspace_role() + + if real_workspace != process_root: + if not _preflight_in_test_mode(): + membership = author_mutation_worktree.assess_workspace_repo_membership( + workspace_path=workspace, + canonical_repo_root=canonical_root, + ) + if membership["block"]: + raise RuntimeError( + author_mutation_worktree.format_workspace_repo_membership_error( + membership + ) + ) + + dirty_files = sorted( + _parse_porcelain_entries(_get_workspace_porcelain(workspace)) + ) + if dirty_files: raise RuntimeError( - author_mutation_worktree.format_workspace_repo_membership_error( - membership + nwb.format_namespace_workspace_binding_error( + role_kind=role, + workspace_path=workspace, + binding_source=ctx.get("workspace_binding_source") + or "unknown binding source", + dirty_files=dirty_files, + ignored_bindings=ctx.get("ignored_bindings"), ) ) - - dirty_files = sorted(_parse_porcelain_entries(_get_workspace_porcelain(workspace))) - if dirty_files: - raise RuntimeError( - nwb.format_namespace_workspace_binding_error( - role_kind=role, - workspace_path=workspace, - binding_source=ctx.get("workspace_binding_source") - or "unknown binding source", - dirty_files=dirty_files, - ignored_bindings=ctx.get("ignored_bindings"), - ) - ) - else: - if _preflight_whoami_violation: - raise RuntimeError( - "Pre-flight order violation: Workspace file edits occurred before " - f"gitea_whoami verification (fail closed). Offending files: " - f"{_format_preflight_files(_preflight_whoami_violation_files)}" - ) - if _preflight_capability_violation: - raise RuntimeError( - "Pre-flight order violation: Workspace file edits occurred before " - f"gitea_resolve_task_capability verification (fail closed). Offending files: " - f"{_format_preflight_files(_preflight_capability_violation_files)}" - ) - - if role in {"reviewer", "merger"}: - current = _get_workspace_porcelain() - baseline = _preflight_capability_baseline_porcelain or "" - reviewer_delta = _new_tracked_changes_since(baseline, current) - _preflight_reviewer_violation_files = reviewer_delta - if reviewer_delta: + else: + if _preflight_whoami_violation: raise RuntimeError( - f"{role.title()} role violation: profile is forbidden from modifying " - "tracked workspace files (fail closed). Offending files: " - f"{_format_preflight_files(reviewer_delta)}" + "Pre-flight order violation: Workspace file edits occurred before " + f"gitea_whoami verification (fail closed). Offending files: " + f"{_format_preflight_files(_preflight_whoami_violation_files)}" + ) + if _preflight_capability_violation: + raise RuntimeError( + "Pre-flight order violation: Workspace file edits occurred before " + f"gitea_resolve_task_capability verification (fail closed). Offending files: " + f"{_format_preflight_files(_preflight_capability_violation_files)}" ) - _enforce_root_checkout_guard(worktree_path) - _enforce_branches_only_author_mutation(worktree_path) - _clear_preflight_capability_state() + if role in {"reviewer", "merger"}: + current = _get_workspace_porcelain() + baseline = _preflight_capability_baseline_porcelain or "" + reviewer_delta = _new_tracked_changes_since(baseline, current) + _preflight_reviewer_violation_files = reviewer_delta + if reviewer_delta: + raise RuntimeError( + f"{role.title()} role violation: profile is forbidden from modifying " + "tracked workspace files (fail closed). Offending files: " + f"{_format_preflight_files(reviewer_delta)}" + ) + + # Historical path: root + branches after purity-order when dirty paths live. + _enforce_root_checkout_guard(worktree_path) + _enforce_branches_only_author_mutation(worktree_path) + _enforce_issue_scope_guard( + worktree_path, + task=task, + target_issue_number=target_issue_number, + require_author_lock=require_author_lock, + ) + _clear_preflight_capability_state() + return + + # #683: under pytest unit isolation, FORCE_PRODUCTION_GUARDS still runs + # production root + branches + issue scope (no silent no-op of guards). + if production_active: + _enforce_root_checkout_guard(worktree_path) + _enforce_branches_only_author_mutation(worktree_path) + _enforce_issue_scope_guard( + worktree_path, + task=task, + target_issue_number=target_issue_number, + require_author_lock=require_author_lock, + ) + + +def _session_issue_lock_snapshot( + workspace_path: str | None = None, +) -> dict: + """Return session lock fields relevant to #683 scope enforcement. + + Branch-vs-lock comparison uses the live workspace branch only when the + lock's worktree matches the mutation workspace. That prevents a foreign + or leftover session lock from poisoning unrelated test worktrees while + still fail-closing when the bound worktree drifts to another issue. + """ + lock = issue_lock_store.read_session_issue_lock() or {} + raw = lock.get("issue_number") + locked: int | None + try: + locked = int(raw) if raw is not None else None + except (TypeError, ValueError): + locked = None + lock_wt = (lock.get("worktree_path") or "").strip() + workspace = (workspace_path or "").strip() + worktrees_match = False + if lock_wt and workspace: + try: + worktrees_match = os.path.realpath(lock_wt) == os.path.realpath(workspace) + except OSError: + worktrees_match = False + return { + "locked_issue_number": locked, + "lock_branch_name": (lock.get("branch_name") or "").strip() or None, + "lock_worktree_path": lock_wt or None, + "worktrees_match": worktrees_match, + } + + +def _session_locked_issue_number() -> int | None: + """Return the active session issue lock number when present (#683).""" + return _session_issue_lock_snapshot().get("locked_issue_number") + + +def _enforce_issue_scope_guard( + worktree_path: str | None = None, + *, + task: str | None = None, + target_issue_number: int | None = None, + require_author_lock: bool = False, +) -> None: + """#683: fail closed on missing/out-of-scope issue ownership for mutations.""" + ctx = _resolve_namespace_mutation_context(worktree_path) + workspace = ctx["workspace_path"] + git_state = issue_lock_worktree.read_worktree_git_state(workspace) + # Honour actual profile role as well as poisoned task role (#540 / #683): + # comment_issue preflight stamps required_role_kind=author, which must not + # strip a genuine reconciler of control-checkout exemptions. + role = ctx.get("workspace_role_kind") or _effective_workspace_role() + actual = _actual_profile_role() + if actual in nwb.NON_AUTHOR_ROLES: + role = actual + snap = _session_issue_lock_snapshot(workspace) + # Scope uses the lock's recorded branch for issue-number matching. + # Live workspace branch can inherit the parent control checkout's branch + # name when a temp branches/ dir is not its own worktree tip — that must + # not invent a false out-of-scope failure. Live branch drift is enforced + # by issue_lock_store.verify_lock_for_mutation elsewhere. + branch_for_scope = snap.get("lock_branch_name") + if ( + snap.get("worktrees_match") + and workflow_scope_guard.production_guards_forced() + ): + live_branch = git_state.get("current_branch") + live_issue = workflow_scope_guard.extract_issue_number_from_branch( + live_branch + ) + locked = snap.get("locked_issue_number") + if ( + live_issue is not None + and locked is not None + and live_issue != locked + ): + branch_for_scope = live_branch + # Author implementation / source-adjacent mutations need ownership when forced. + authorish = role == "author" or ( + task + in { + "create_issue", + "comment_issue", + "lock_issue", + "create_pr", + "commit_files", + "gitea_commit_files", + "mark_issue", + } + ) + require_lock = bool(require_author_lock) or ( + authorish + and workflow_scope_guard.production_guards_forced() + and role == "author" + ) + assessment = workflow_scope_guard.assess_production_mutation_guards( + workspace_path=workspace, + canonical_repo_root=ctx["canonical_repo_root"], + porcelain_status=git_state.get("porcelain_status") or "", + current_branch=branch_for_scope, + locked_issue_number=snap.get("locked_issue_number"), + target_issue_number=target_issue_number, + role_kind=role, + require_author_lock=require_lock, + in_test_mode=_preflight_in_test_mode(), + ) + workflow_scope_guard.raise_if_blocked(assessment) + + +def _production_guard_block_from_exc(exc: BaseException, **extra) -> dict | None: + """Map production-guard exceptions to typed tool block responses (#683).""" + if isinstance(exc, workflow_scope_guard.ProductionGuardError): + return workflow_scope_guard.block_response(exc, **extra) + text = str(exc) + if "Workflow scope guard (#683)" in text or "Root checkout guard (#475)" in text: + kind = workflow_scope_guard.BLOCKER_PRODUCTION_GUARD + if "root_diagnostic_edit" in text or "tracked source or test edits" in text: + kind = workflow_scope_guard.BLOCKER_ROOT_DIAGNOSTIC_EDIT + elif "Branches-only mutation guard" in text or "stable control checkout" in text: + kind = workflow_scope_guard.BLOCKER_MISSING_WORKTREE + elif "out-of-scope" in text or "locked to issue" in text: + kind = workflow_scope_guard.BLOCKER_OUT_OF_SCOPE_ISSUE + elif "no owning issue" in text: + kind = workflow_scope_guard.BLOCKER_MISSING_ISSUE_SCOPE + return workflow_scope_guard.block_response( + blocker_kind=kind, + reasons=[text], + **extra, + ) + if "Branches-only mutation guard" in text: + return workflow_scope_guard.block_response( + blocker_kind=workflow_scope_guard.BLOCKER_MISSING_WORKTREE, + reasons=[text], + **extra, + ) + if "Root checkout guard" in text: + return workflow_scope_guard.block_response( + blocker_kind=workflow_scope_guard.BLOCKER_ROOT_DIAGNOSTIC_EDIT, + reasons=[text], + **extra, + ) + return None def _verify_role_mutation_workspace( @@ -727,7 +906,12 @@ def _verify_role_mutation_workspace( worktree: str | None = None, task: str | None = None, ) -> str: - """Bind reviewer/merger mutations to the active namespace workspace (#510).""" + """Bind reviewer/merger mutations to the active namespace workspace (#510). + + #683: must NOT early-return solely because pytest/unittest is loaded. + Production workspace binding always runs; test isolation uses explicit + env fixtures / force-on flags, never a production short-circuit here. + """ # Check running runtimes to prevent stale mutations try: @@ -928,6 +1112,7 @@ import merge_approval_gate # noqa: E402 import already_landed_reconcile # noqa: E402 import author_mutation_worktree # noqa: E402 import root_checkout_guard # noqa: E402 +import workflow_scope_guard # noqa: E402 # #683 production scope / force-on guards import stable_branch_push_guard # noqa: E402 import remote_repo_guard # noqa: E402 import issue_claim_heartbeat # noqa: E402 @@ -1896,7 +2081,15 @@ def gitea_create_issue( ) if blocked: return blocked - verify_preflight_purity(remote, worktree_path=worktree_path, task="create_issue") + try: + verify_preflight_purity( + remote, worktree_path=worktree_path, task="create_issue" + ) + except Exception as exc: + typed = _production_guard_block_from_exc(exc, number=None) + if typed is not None: + return typed + raise content_gate = issue_content_gate.pre_create_issue_content_gate( title, body, @@ -8167,9 +8360,26 @@ def gitea_create_issue_comment( with the reveal opt-in); on a permission block or empty body, 'success'/'performed' False and 'reasons' with no API call made (permission blocks also carry a structured 'permission_report', - #142). + #142). On production-guard blocks (#683): 'blocker_kind' and + 'exact_next_action' with no API side effect. """ - verify_preflight_purity(remote, worktree_path=worktree_path, task="comment_issue") + try: + # Do not pass target_issue_number: comments on other issues remain + # allowed while an author holds a different implementation lock. + # Scope ownership for source edits is enforced via branch/worktree + # binding + root diagnostic checks (#683). + verify_preflight_purity( + remote, + worktree_path=worktree_path, + task="comment_issue", + ) + except Exception as exc: + typed = _production_guard_block_from_exc( + exc, issue_number=issue_number + ) + if typed is not None: + return typed + raise gate_reasons = _profile_operation_gate("gitea.issue.comment") reasons = list(gate_reasons) if not (body or "").strip(): diff --git a/tests/test_create_issue_workspace_guard.py b/tests/test_create_issue_workspace_guard.py index 71e77f5..43f516e 100644 --- a/tests/test_create_issue_workspace_guard.py +++ b/tests/test_create_issue_workspace_guard.py @@ -57,9 +57,23 @@ class TestCreateIssueWorkspaceGuard(unittest.TestCase): # Without worktree_path/env hints, workspace resolves to PROJECT_ROOT. When that # path is the stable control checkout (not under branches/), mutation must fail. with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT): - with self.assertRaises(RuntimeError) as ctx: - srv.gitea_create_issue(title="Test issue", body="body text") - self.assertIn("stable control checkout", str(ctx.exception)) + try: + res = srv.gitea_create_issue(title="Test issue", body="body text") + except RuntimeError as exc: + self.assertIn("stable control checkout", str(exc)) + else: + # #683: production guards return typed blockers at entrypoints + self.assertFalse(res.get("success")) + self.assertFalse(res.get("performed")) + blob = " ".join(res.get("reasons") or []) + " " + str( + res.get("blocker_kind") or "" + ) + self.assertTrue( + "stable control checkout" in blob + or "missing_issue_worktree" in blob + or "control checkout" in blob.lower() + ) + self.assertTrue(res.get("exact_next_action")) @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH) @patch("gitea_mcp_server._profile_permission_block", return_value=None) @@ -105,11 +119,17 @@ class TestCreateIssueWorkspaceGuard(unittest.TestCase): ) with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT): - with self.assertRaises(RuntimeError) as ctx: - srv.gitea_create_issue( + try: + res = srv.gitea_create_issue( title="Test issue", body="body", worktree_path=missing_path ) - self.assertIn("does not exist (fail closed)", str(ctx.exception)) + except RuntimeError as exc: + self.assertIn("does not exist", str(exc)) + else: + self.assertFalse(res.get("success")) + blob = " ".join(res.get("reasons") or []) + self.assertIn("does not exist", blob) + self.assertTrue(res.get("exact_next_action") or res.get("reasons")) @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH) @patch("gitea_mcp_server._profile_permission_block", return_value=None) @@ -142,11 +162,20 @@ class TestCreateIssueWorkspaceGuard(unittest.TestCase): with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT): with patch("gitea_mcp_server._get_workspace_porcelain", return_value=""): - with self.assertRaises(RuntimeError) as ctx: - srv.gitea_create_issue( - title="Test issue", body="body", worktree_path=wrong_repo_path + try: + res = srv.gitea_create_issue( + title="Test issue", + body="body", + worktree_path=wrong_repo_path, ) - self.assertIn("does not belong to the target repository", str(ctx.exception)) + except RuntimeError as exc: + self.assertIn( + "does not belong to the target repository", str(exc) + ) + else: + self.assertFalse(res.get("success")) + blob = " ".join(res.get("reasons") or []) + self.assertIn("does not belong to the target repository", blob) @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH) @patch("gitea_mcp_server._profile_permission_block", return_value=None) diff --git a/tests/test_issue_540_comment_role_poison.py b/tests/test_issue_540_comment_role_poison.py index 676c2bf..53ef2ba 100644 --- a/tests/test_issue_540_comment_role_poison.py +++ b/tests/test_issue_540_comment_role_poison.py @@ -267,10 +267,19 @@ class TestReconcilerCommentThroughCanonicalPath(unittest.TestCase): with patch.dict(os.environ, {}, clear=False): os.environ.pop("GITEA_AUTHOR_WORKTREE", None) os.environ.pop("GITEA_ACTIVE_WORKTREE", None) - with self.assertRaises(RuntimeError): - srv.gitea_create_issue_comment( + try: + res = srv.gitea_create_issue_comment( 515, "author note", remote="prgs" ) + except RuntimeError: + pass # legacy raise path + else: + # #683: typed blocker at mutation entrypoint + self.assertFalse(res.get("success")) + self.assertFalse(res.get("performed")) + self.assertTrue( + res.get("blocker_kind") or res.get("reasons") + ) mock_api.assert_not_called() diff --git a/tests/test_issue_683_workflow_scope_guards.py b/tests/test_issue_683_workflow_scope_guards.py new file mode 100644 index 0000000..9b106c7 --- /dev/null +++ b/tests/test_issue_683_workflow_scope_guards.py @@ -0,0 +1,472 @@ +"""#683: block unattributed root WIP; pytest cannot disable production guards. + +Regression coverage required by issue #683: + +1. Session locked to issue A blocks unrelated target issue B until B is selected. +2. Diagnostic source edit on the root checkout is blocked. +3. Same legitimate edit succeeds after issue ownership + isolated worktree bind. +4. Running under pytest does not deactivate production guards when force-on. +5. Dirty tracked Python files remain visible to porcelain consumers. +6. Monkeypatching one helper cannot silently turn the full guard path into a no-op. +7. Real mutation entrypoint proves production guards run before side effects. +8. Same-issue edits in a valid isolated worktree remain unaffected. +9. Blocker includes stable reason + exact recovery action. +""" + +from __future__ import annotations + +import os +import sys +import tempfile +import textwrap +import unittest +from pathlib import Path +from unittest.mock import MagicMock, patch + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import gitea_mcp_server as mcp_server # noqa: E402 +import issue_lock_worktree # noqa: E402 +import workflow_scope_guard as wsg # noqa: E402 + +CONTROL_ROOT = str(Path(__file__).resolve().parent.parent) +if "branches" in Path(__file__).resolve().parts: + # Running from a worktree under branches/ — parent of branches is control. + parts = Path(__file__).resolve().parts + idx = parts.index("branches") + CONTROL_ROOT = str(Path(*parts[:idx])) if idx > 0 else CONTROL_ROOT + + +class TestProductionGuardsForceOn(unittest.TestCase): + def tearDown(self): + for key in ( + wsg.FORCE_PRODUCTION_GUARDS_ENV, + "GITEA_TEST_FORCE_DIRTY", + "GITEA_TEST_PORCELAIN", + "GITEA_AUTHOR_WORKTREE", + "GITEA_ACTIVE_WORKTREE", + ): + os.environ.pop(key, None) + wsg.clear_workflow_failure_ledger() + + def test_force_on_under_pytest_keeps_production_active(self): + self.assertTrue(wsg.production_guards_active(in_test_mode=False)) + self.assertFalse(wsg.production_guards_active(in_test_mode=True)) + os.environ[wsg.FORCE_PRODUCTION_GUARDS_ENV] = "1" + self.assertTrue(wsg.production_guards_active(in_test_mode=True)) + self.assertTrue(wsg.production_guards_forced()) + + def test_no_early_return_in_verify_role_mutation_workspace_source(self): + src = Path(mcp_server.__file__).read_text(encoding="utf-8") + # Rejected 300a4ca pattern must not exist. + self.assertNotIn( + "if _preflight_in_test_mode():\n return _resolve_preflight_workspace_path", + src, + ) + # Docstring contract for #683. + self.assertIn("#683", src) + self.assertIn("must NOT early-return solely because pytest", src) + + +class TestPorcelainIntegrity(unittest.TestCase): + def test_read_worktree_git_state_surfaces_dirty_py(self): + with tempfile.TemporaryDirectory() as tmp: + # Use a real git repo so porcelain is truthful. + import subprocess + + subprocess.run(["git", "init"], cwd=tmp, check=True, capture_output=True) + subprocess.run( + ["git", "config", "user.email", "t@example.com"], + cwd=tmp, + check=True, + capture_output=True, + ) + subprocess.run( + ["git", "config", "user.name", "t"], + cwd=tmp, + check=True, + capture_output=True, + ) + py_path = Path(tmp) / "sample_mod.py" + py_path.write_text("x = 1\n", encoding="utf-8") + subprocess.run(["git", "add", "sample_mod.py"], cwd=tmp, check=True) + subprocess.run( + ["git", "commit", "-m", "init"], + cwd=tmp, + check=True, + capture_output=True, + ) + py_path.write_text("x = 2\n", encoding="utf-8") + state = issue_lock_worktree.read_worktree_git_state(tmp) + porcelain = state.get("porcelain_status") or "" + self.assertIn("sample_mod.py", porcelain) + self.assertTrue(any(line.strip().endswith(".py") for line in porcelain.splitlines())) + + def test_production_reader_source_rejects_pytest_py_filter(self): + src = Path(issue_lock_worktree.__file__).read_text(encoding="utf-8") + findings = wsg.assert_no_pytest_porcelain_filter(src) + self.assertEqual(findings, []) + # Negative: the rejected 300a4ca pattern is detected. + rejected = textwrap.dedent( + """ + porcelain = status_res.stdout or "" + import sys + if "pytest" in sys.modules or "unittest" in sys.modules: + porcelain = "\\n".join( + line for line in porcelain.splitlines() + if not line.strip().endswith(".py") + ) + """ + ) + self.assertTrue(wsg.assert_no_pytest_porcelain_filter(rejected)) + + +class TestIssueScopeOwnership(unittest.TestCase): + def test_out_of_scope_issue_blocked_until_selected(self): + result = wsg.assess_issue_scope_ownership( + locked_issue_number=100, + target_issue_number=200, + branch_name="fix/issue-100-example", + role_kind="author", + ) + self.assertTrue(result["block"]) + self.assertEqual(result["blocker_kind"], wsg.BLOCKER_OUT_OF_SCOPE_ISSUE) + self.assertIn("exact_next_action", result) + self.assertIn("owning issue", result["exact_next_action"].lower()) + self.assertTrue(result["reasons"]) + + def test_same_issue_scope_allowed(self): + result = wsg.assess_issue_scope_ownership( + locked_issue_number=100, + target_issue_number=100, + branch_name="fix/issue-100-example", + role_kind="author", + ) + self.assertFalse(result["block"]) + self.assertEqual(result["exact_next_action"], "proceed") + + def test_missing_lock_when_required(self): + result = wsg.assess_issue_scope_ownership( + locked_issue_number=None, + target_issue_number=None, + role_kind="author", + require_lock_for_author=True, + ) + self.assertTrue(result["block"]) + self.assertEqual(result["blocker_kind"], wsg.BLOCKER_MISSING_ISSUE_SCOPE) + + def test_branch_issue_mismatch(self): + result = wsg.assess_issue_scope_ownership( + locked_issue_number=50, + branch_name="fix/issue-99-other", + role_kind="author", + ) + self.assertTrue(result["block"]) + self.assertEqual(result["blocker_kind"], wsg.BLOCKER_OUT_OF_SCOPE_ISSUE) + + +class TestRootDiagnosticEdit(unittest.TestCase): + def test_dirty_root_source_blocked(self): + result = wsg.assess_root_source_mutation( + workspace_path=CONTROL_ROOT, + canonical_repo_root=CONTROL_ROOT, + porcelain_status=" M gitea_mcp_server.py\n M tests/test_x.py\n", + role_kind="author", + ) + self.assertTrue(result["block"]) + self.assertEqual(result["blocker_kind"], wsg.BLOCKER_ROOT_DIAGNOSTIC_EDIT) + self.assertIn("gitea_mcp_server.py", result["dirty_source_files"]) + self.assertIn("exact_next_action", result) + self.assertIn("branches/", result["exact_next_action"]) + + def test_isolated_worktree_same_issue_unaffected(self): + wt = f"{CONTROL_ROOT}/branches/issue-100-example" + result = wsg.assess_root_source_mutation( + workspace_path=wt, + canonical_repo_root=CONTROL_ROOT, + porcelain_status=" M helper.py\n", + current_branch="fix/issue-100-example", + locked_issue_number=100, + role_kind="author", + ) + self.assertFalse(result["block"]) + self.assertTrue(result["under_branches"]) + + def test_legitimate_after_ownership_and_worktree(self): + wt = f"{CONTROL_ROOT}/branches/issue-683-workflow-guard-hardening" + composed = wsg.assess_production_mutation_guards( + workspace_path=wt, + canonical_repo_root=CONTROL_ROOT, + porcelain_status=" M workflow_scope_guard.py\n", + current_branch="fix/issue-683-workflow-guard-hardening", + locked_issue_number=683, + target_issue_number=683, + role_kind="author", + require_author_lock=True, + in_test_mode=True, + ) + # Force-on required for production path under pytest. + os.environ[wsg.FORCE_PRODUCTION_GUARDS_ENV] = "1" + try: + composed = wsg.assess_production_mutation_guards( + workspace_path=wt, + canonical_repo_root=CONTROL_ROOT, + porcelain_status=" M workflow_scope_guard.py\n", + current_branch="fix/issue-683-workflow-guard-hardening", + locked_issue_number=683, + target_issue_number=683, + role_kind="author", + require_author_lock=True, + in_test_mode=True, + ) + self.assertFalse(composed["block"]) + self.assertFalse(composed.get("skipped")) + finally: + os.environ.pop(wsg.FORCE_PRODUCTION_GUARDS_ENV, None) + + +class TestTypedBlockerResponse(unittest.TestCase): + def test_block_response_has_stable_kind_and_next_action(self): + assessment = wsg.assess_issue_scope_ownership( + locked_issue_number=1, + target_issue_number=2, + role_kind="author", + ) + resp = wsg.block_response(assessment) + self.assertFalse(resp["success"]) + self.assertFalse(resp["performed"]) + self.assertEqual(resp["blocker_kind"], wsg.BLOCKER_OUT_OF_SCOPE_ISSUE) + self.assertIsInstance(resp["exact_next_action"], str) + self.assertTrue(resp["exact_next_action"]) + self.assertTrue(resp["reasons"]) + + def test_production_guard_error_roundtrip(self): + err = wsg.ProductionGuardError( + "blocked", + blocker_kind=wsg.BLOCKER_ROOT_DIAGNOSTIC_EDIT, + reasons=["dirty root"], + ) + resp = wsg.block_response(err, issue_number=683) + self.assertEqual(resp["blocker_kind"], wsg.BLOCKER_ROOT_DIAGNOSTIC_EDIT) + self.assertEqual(resp["issue_number"], 683) + self.assertIn("exact_next_action", resp) + + +class TestDurableFailureRecording(unittest.TestCase): + def setUp(self): + wsg.clear_workflow_failure_ledger() + + def tearDown(self): + wsg.clear_workflow_failure_ledger() + + def test_record_before_source_mutation(self): + pending = wsg.assess_durable_failure_recorded( + require_record=True, pending_source_mutation=True + ) + self.assertTrue(pending["block"]) + self.assertEqual(pending["blocker_kind"], wsg.BLOCKER_UNRECORDED_FAILURE) + + wsg.record_workflow_failure( + kind="transport_eof", + detail="EOF during review session (#584 cluster)", + issue_number=683, + task="comment_issue", + ) + after = wsg.assess_durable_failure_recorded( + require_record=True, pending_source_mutation=True + ) + self.assertFalse(after["block"]) + self.assertEqual(len(wsg.workflow_failure_ledger()), 1) + + +class TestMonkeypatchCannotNoopFullPath(unittest.TestCase): + def tearDown(self): + os.environ.pop(wsg.FORCE_PRODUCTION_GUARDS_ENV, None) + + def test_patching_branches_only_still_blocks_dirty_root_scope(self): + """Monkeypatching branches-only must not silence root diagnostic block.""" + os.environ[wsg.FORCE_PRODUCTION_GUARDS_ENV] = "1" + with patch.object( + mcp_server, "_enforce_branches_only_author_mutation", lambda *a, **k: None + ): + with patch.object( + mcp_server, "_enforce_root_checkout_guard", lambda *a, **k: None + ): + # Even if both legacy helpers are patched, issue-scope composition + # still sees dirty root source via assess_production_mutation_guards. + assessment = wsg.assess_production_mutation_guards( + workspace_path=CONTROL_ROOT, + canonical_repo_root=CONTROL_ROOT, + porcelain_status=" M gitea_mcp_server.py\n", + role_kind="author", + in_test_mode=True, + ) + self.assertTrue(assessment["block"]) + self.assertEqual( + assessment["blocker_kind"], wsg.BLOCKER_ROOT_DIAGNOSTIC_EDIT + ) + + +class TestRealEntrypointProductionGuard(unittest.TestCase): + """Real mutation entrypoint: production guard before side effects (#683).""" + + def setUp(self): + os.environ[wsg.FORCE_PRODUCTION_GUARDS_ENV] = "1" + for key in ("GITEA_AUTHOR_WORKTREE", "GITEA_ACTIVE_WORKTREE"): + os.environ.pop(key, None) + self._orig_whoami = mcp_server._preflight_whoami_called + self._orig_cap = mcp_server._preflight_capability_called + mcp_server._preflight_whoami_called = False + mcp_server._preflight_capability_called = False + mcp_server._preflight_resolved_role = None + mcp_server._preflight_resolved_task = None + + def tearDown(self): + os.environ.pop(wsg.FORCE_PRODUCTION_GUARDS_ENV, None) + for key in ("GITEA_AUTHOR_WORKTREE", "GITEA_ACTIVE_WORKTREE"): + os.environ.pop(key, None) + mcp_server._preflight_whoami_called = self._orig_whoami + mcp_server._preflight_capability_called = self._orig_cap + mcp_server._preflight_resolved_role = None + mcp_server._preflight_resolved_task = None + + def test_comment_issue_blocks_dirty_root_before_api(self): + api_mock = MagicMock() + with patch.object(mcp_server, "api_request", api_mock), patch.object( + mcp_server, + "_actual_profile_role", + return_value="author", + ), patch.object( + mcp_server, + "_effective_workspace_role", + return_value="author", + ), patch.object( + mcp_server, + "get_profile", + return_value={ + "profile_name": "prgs-author", + "allowed_operations": [ + "gitea.issue.comment", + "gitea.read", + "gitea.pr.create", + "gitea.branch.push", + ], + "forbidden_operations": [], + }, + ), patch.object( + issue_lock_worktree, + "read_worktree_git_state", + side_effect=lambda path, **kw: { + "current_branch": "master", + "porcelain_status": ( + " M gitea_mcp_server.py\n" + if os.path.realpath(path) == os.path.realpath(CONTROL_ROOT) + or path == CONTROL_ROOT + else "" + ), + "head_sha": "a" * 40, + "base_equivalent": True, + }, + ), patch.object( + mcp_server, + "_resolve_namespace_mutation_context", + return_value={ + "workspace_path": CONTROL_ROOT, + "canonical_repo_root": CONTROL_ROOT, + "process_project_root": CONTROL_ROOT, + "workspace_role_kind": "author", + "workspace_binding_source": "process root", + "ignored_bindings": [], + }, + ), patch.object( + mcp_server, + "_resolve_author_mutation_context", + return_value={ + "workspace_path": CONTROL_ROOT, + "canonical_repo_root": CONTROL_ROOT, + "process_project_root": CONTROL_ROOT, + "roots_aligned": True, + }, + ), patch.object( + mcp_server, + "_session_locked_issue_number", + return_value=None, + ): + result = mcp_server.gitea_create_issue_comment( + issue_number=683, + body="diagnostic note", + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + worktree_path=CONTROL_ROOT, + ) + + api_mock.assert_not_called() + self.assertFalse(result.get("success")) + self.assertFalse(result.get("performed")) + self.assertIn(result.get("blocker_kind"), wsg.BLOCKER_KINDS) + self.assertTrue(result.get("exact_next_action")) + self.assertTrue(result.get("reasons")) + + def test_comment_issue_succeeds_structure_after_worktree_bind(self): + """Same-issue isolated worktree is not blocked by root diagnostic path.""" + wt = f"{CONTROL_ROOT}/branches/issue-683-workflow-guard-hardening" + os.environ["GITEA_AUTHOR_WORKTREE"] = wt + assessment = wsg.assess_production_mutation_guards( + workspace_path=wt, + canonical_repo_root=CONTROL_ROOT, + porcelain_status=" M workflow_scope_guard.py\n", + current_branch="fix/issue-683-workflow-guard-hardening", + locked_issue_number=683, + target_issue_number=683, + role_kind="author", + require_author_lock=True, + in_test_mode=True, + ) + self.assertFalse(assessment["block"], assessment) + + +class TestVerifyPreflightForceOn(unittest.TestCase): + def tearDown(self): + os.environ.pop(wsg.FORCE_PRODUCTION_GUARDS_ENV, None) + for key in ("GITEA_AUTHOR_WORKTREE", "GITEA_ACTIVE_WORKTREE"): + os.environ.pop(key, None) + + def test_force_on_runs_production_guards_under_pytest(self): + os.environ[wsg.FORCE_PRODUCTION_GUARDS_ENV] = "1" + called = {"root": 0, "branches": 0, "scope": 0} + + def _root(*a, **k): + called["root"] += 1 + + def _branches(*a, **k): + called["branches"] += 1 + + def _scope(*a, **k): + called["scope"] += 1 + + with patch.object(mcp_server, "_enforce_root_checkout_guard", _root), patch.object( + mcp_server, "_enforce_branches_only_author_mutation", _branches + ), patch.object(mcp_server, "_enforce_issue_scope_guard", _scope): + # No whoami/capability — purity-order skipped; production still runs. + mcp_server.verify_preflight_purity(task="comment_issue") + + self.assertEqual(called["root"], 1) + self.assertEqual(called["branches"], 1) + self.assertEqual(called["scope"], 1) + + def test_without_force_on_pytest_skips_production_only_for_unit_isolation(self): + called = {"root": 0} + + def _root(*a, **k): + called["root"] += 1 + + with patch.object(mcp_server, "_enforce_root_checkout_guard", _root), patch.object( + mcp_server, "_enforce_branches_only_author_mutation", lambda *a, **k: None + ), patch.object(mcp_server, "_enforce_issue_scope_guard", lambda *a, **k: None): + mcp_server.verify_preflight_purity(task="comment_issue") + self.assertEqual(called["root"], 0) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_issue_comment_workspace_guard.py b/tests/test_issue_comment_workspace_guard.py index 59139ed..377c624 100644 --- a/tests/test_issue_comment_workspace_guard.py +++ b/tests/test_issue_comment_workspace_guard.py @@ -108,13 +108,24 @@ class TestIssueCommentWorkspaceGuard(unittest.TestCase): "gitea_mcp_server.issue_lock_worktree.read_worktree_git_state", side_effect=self._git_state(valid_worktree), ): - with self.assertRaises(RuntimeError) as ctx: - srv.gitea_create_issue_comment( + try: + res = srv.gitea_create_issue_comment( issue_number=557, body="evidence comment", remote="prgs", ) - self.assertIn("stable control checkout", str(ctx.exception)) + except RuntimeError as exc: + self.assertIn("stable control checkout", str(exc)) + else: + # #683 typed blocker at mutation entrypoint + self.assertFalse(res.get("success")) + self.assertFalse(res.get("performed")) + blob = " ".join(res.get("reasons") or []) + self.assertTrue( + "stable control checkout" in blob + or res.get("blocker_kind") + ) + self.assertTrue(res.get("exact_next_action") or res.get("reasons")) mock_api.assert_not_called() @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH) @@ -184,14 +195,24 @@ class TestIssueCommentWorkspaceGuard(unittest.TestCase): side_effect=self._subprocess(valid_worktree, outside_worktree), ): with patch.dict(os.environ, self.AUTHOR_ENV, clear=True): - with self.assertRaises(RuntimeError) as ctx: - srv.gitea_create_issue_comment( + try: + res = srv.gitea_create_issue_comment( issue_number=557, body="evidence comment", remote="prgs", worktree_path=outside_worktree, ) - self.assertIn("does not belong to the target repository", str(ctx.exception)) + except RuntimeError as exc: + self.assertIn( + "does not belong to the target repository", + str(exc), + ) + else: + self.assertFalse(res.get("success")) + blob = " ".join(res.get("reasons") or []) + self.assertIn( + "does not belong to the target repository", blob + ) mock_api.assert_not_called() @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH) diff --git a/tests/test_preflight_read_survival.py b/tests/test_preflight_read_survival.py index 6da8af6..5ffd3dd 100644 --- a/tests/test_preflight_read_survival.py +++ b/tests/test_preflight_read_survival.py @@ -76,13 +76,17 @@ class TestPreflightReadSurvival(unittest.TestCase): self.assertIn("task mismatch", str(ctx.exception)) def test_capability_consumed_after_mutation_gate(self): + # Use reconciler/close_pr so this purity-order test does not require a + # branches/ worktree (author create_issue would hit #274/#683 guards). + # Test isolation stays explicit; production author guards remain live + # under force-on (see tests/test_issue_683_workflow_scope_guards.py). mcp_server.record_preflight_check("whoami") mcp_server.record_preflight_check( - "capability", resolved_role="author", resolved_task="create_issue" + "capability", resolved_role="reconciler", resolved_task="close_pr" ) - mcp_server.verify_preflight_purity(task="create_issue") + mcp_server.verify_preflight_purity(task="close_pr") with self.assertRaises(RuntimeError) as ctx: - mcp_server.verify_preflight_purity(task="create_issue") + mcp_server.verify_preflight_purity(task="close_pr") self.assertIn("has not been resolved", str(ctx.exception)) def test_whoami_recovery_after_violation_clears_capability(self): diff --git a/tests/test_reconciler_close_workspace_guard.py b/tests/test_reconciler_close_workspace_guard.py index e5e33b0..43682a0 100644 --- a/tests/test_reconciler_close_workspace_guard.py +++ b/tests/test_reconciler_close_workspace_guard.py @@ -86,9 +86,21 @@ class TestReconcilerCloseWorkspaceGuard(unittest.TestCase): ): srv._preflight_resolved_role = "author" with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT): - with self.assertRaises(RuntimeError) as ctx: - srv.gitea_create_issue(title="Test", body="body") - self.assertIn("stable control checkout", str(ctx.exception)) + try: + res = srv.gitea_create_issue(title="Test", body="body") + except RuntimeError as exc: + self.assertIn("stable control checkout", str(exc)) + else: + # #683 typed blocker at mutation entrypoint + self.assertFalse(res.get("success")) + blob = " ".join(res.get("reasons") or []) + str( + res.get("blocker_kind") or "" + ) + self.assertTrue( + "stable control checkout" in blob + or "missing_issue_worktree" in blob + or "control checkout" in blob.lower() + ) if __name__ == "__main__": diff --git a/workflow_scope_guard.py b/workflow_scope_guard.py new file mode 100644 index 0000000..11b1790 --- /dev/null +++ b/workflow_scope_guard.py @@ -0,0 +1,626 @@ +"""Workflow scope ownership and production-guard hardening (#683). + +Implements fail-closed enforcement so sessions cannot: + +* mutate source/tests on the root/control checkout (including temporary + diagnostic edits) without binding an issue-backed ``branches/`` worktree; +* continue out-of-scope source work while locked to a different issue; +* disable, skip, or conceal production root/branches/porcelain guards solely + because pytest/unittest is loaded. + +This module is pure assessment + small durable ledger helpers. Callers gather +live facts (lock, branch, porcelain, worktree path) and pass them in. Existing +root_checkout_guard / author_mutation_worktree assessors remain authoritative; +this module composes typed blockers with exact recovery actions. + +Do **not** reintroduce the rejected #681 / ``300a4ca`` patterns: + +* early-return from workspace verification under ``_preflight_in_test_mode()`` +* porcelain filtering that strips ``*.py`` lines under pytest +""" + +from __future__ import annotations + +import os +import re +import threading +from typing import Any + +import author_mutation_worktree +from reviewer_worktree import parse_dirty_tracked_files + +# ── force-on / test isolation ──────────────────────────────────────────────── + +# When set, production root/branches/scope guards MUST run even under pytest. +# Unit tests that only need preflight-order isolation leave this unset and +# use GITEA_TEST_PORCELAIN / fixtures; real-entrypoint proof sets this to "1". +FORCE_PRODUCTION_GUARDS_ENV = "GITEA_TEST_FORCE_PRODUCTION_GUARDS" + +# Existing force signals also mean "exercise production dirtiness paths". +_FORCE_DIRTY_ENV = "GITEA_TEST_FORCE_DIRTY" +_FORCE_PORCELAIN_ENV = "GITEA_TEST_PORCELAIN" + +# ── typed blocker kinds ────────────────────────────────────────────────────── + +BLOCKER_ROOT_DIAGNOSTIC_EDIT = "root_diagnostic_edit" +BLOCKER_MISSING_ISSUE_SCOPE = "missing_issue_scope" +BLOCKER_OUT_OF_SCOPE_ISSUE = "out_of_scope_issue" +BLOCKER_MISSING_WORKTREE = "missing_issue_worktree" +BLOCKER_UNRECORDED_FAILURE = "unrecorded_workflow_failure" +BLOCKER_PRODUCTION_GUARD = "production_guard_violation" + +BLOCKER_KINDS = frozenset( + { + BLOCKER_ROOT_DIAGNOSTIC_EDIT, + BLOCKER_MISSING_ISSUE_SCOPE, + BLOCKER_OUT_OF_SCOPE_ISSUE, + BLOCKER_MISSING_WORKTREE, + BLOCKER_UNRECORDED_FAILURE, + BLOCKER_PRODUCTION_GUARD, + } +) + +_NEXT_ACTIONS: dict[str, str] = { + BLOCKER_ROOT_DIAGNOSTIC_EDIT: ( + "Stop editing the control/root checkout. Preserve or discard root WIP " + "durably, restore root to clean master, lock or create the owning issue, " + "bind branches/issue--*, set GITEA_AUTHOR_WORKTREE to that worktree, " + "then re-run the mutation." + ), + BLOCKER_MISSING_ISSUE_SCOPE: ( + "Select or create the owning Gitea issue, claim/lock it " + "(gitea_mark_issue + gitea_lock_issue), bind branches/issue--* " + "from clean master, then re-run the mutation from that worktree." + ), + BLOCKER_OUT_OF_SCOPE_ISSUE: ( + "Stop. The active issue lock does not own this work. Release or finish " + "the current issue lease, then select/create and lock the correct " + "owning issue, bind its branches/issue--* worktree, and re-run." + ), + BLOCKER_MISSING_WORKTREE: ( + "Bind an issue-backed worktree under branches/ (scripts/worktree-start " + "or git worktree add branches/issue--*), set GITEA_AUTHOR_WORKTREE / " + "worktree_path to that path, keep the control checkout clean on master, " + "then re-run the mutation." + ), + BLOCKER_UNRECORDED_FAILURE: ( + "Record the workflow/tool failure durably first (issue comment or " + "workflow_scope_guard.record_workflow_failure), then continue only " + "inside the owning issue-backed worktree." + ), + BLOCKER_PRODUCTION_GUARD: ( + "Resolve the production guard violation: clean or isolate the control " + "checkout, bind the owning issue worktree under branches/, and re-run " + "with production guards active." + ), +} + +_ISSUE_IN_BRANCH_RE = re.compile(r"issue-(\d+)", re.IGNORECASE) + +# In-process durable failure ledger (also written via optional sink callback). +_ledger_lock = threading.Lock() +_failure_ledger: list[dict[str, Any]] = [] + + +class ProductionGuardError(RuntimeError): + """Fail-closed production guard with typed blocker metadata (#683).""" + + def __init__( + self, + message: str, + *, + blocker_kind: str, + exact_next_action: str | None = None, + reasons: list[str] | None = None, + details: dict[str, Any] | None = None, + ) -> None: + super().__init__(message) + kind = (blocker_kind or "").strip() + if kind not in BLOCKER_KINDS: + kind = BLOCKER_PRODUCTION_GUARD + self.blocker_kind = kind + self.exact_next_action = ( + (exact_next_action or "").strip() or _NEXT_ACTIONS[kind] + ) + self.reasons = list(reasons or [message]) + self.details = dict(details or {}) + + +def production_guards_forced() -> bool: + """True when the explicit #683 force-on flag requests production guards.""" + return (os.environ.get(FORCE_PRODUCTION_GUARDS_ENV) or "").strip().lower() in { + "1", + "true", + "yes", + "on", + } + + +def purity_order_forced() -> bool: + """True when tests force preflight-order dirtiness paths (legacy flags).""" + if os.environ.get(_FORCE_DIRTY_ENV): + return True + # GITEA_TEST_PORCELAIN present (even empty) means dirtiness paths are live. + if os.environ.get(_FORCE_PORCELAIN_ENV) is not None: + return True + return False + + +def production_guards_active(*, in_test_mode: bool) -> bool: + """Whether production root/branches/scope guards must execute. + + Production (non-test) always active. Under pytest, active when either the + explicit #683 force-on flag or legacy dirty/porcelain force signals are + set — never skip production enforcement solely because tests are running + when force-on is requested. + """ + if production_guards_forced() or purity_order_forced(): + return True + return not bool(in_test_mode) + + +def extract_issue_number_from_branch(branch_name: str | None) -> int | None: + """Return the first issue-N number embedded in a branch name, if any.""" + text = (branch_name or "").strip() + if not text: + return None + match = _ISSUE_IN_BRANCH_RE.search(text) + if not match: + return None + try: + return int(match.group(1)) + except ValueError: + return None + + +def is_source_or_test_path(path: str) -> bool: + """True for tracked source/test paths that must not land as root WIP.""" + p = (path or "").replace("\\", "/").lstrip("./") + if not p: + return False + if p.startswith("tests/") or "/tests/" in f"/{p}": + return True + if p.endswith((".py", ".pyi", ".toml", ".cfg", ".ini", ".sh")): + return True + if p in {"requirements.txt", "pyproject.toml", "setup.py", "setup.cfg"}: + return True + return False + + +def dirty_source_files(porcelain_status: str) -> list[str]: + """Tracked dirty paths that count as source/test contamination.""" + dirty = parse_dirty_tracked_files(porcelain_status or "") + return [p for p in dirty if is_source_or_test_path(p)] + + +def assess_issue_scope_ownership( + *, + locked_issue_number: int | None, + target_issue_number: int | None = None, + branch_name: str | None = None, + role_kind: str | None = None, + require_lock_for_author: bool = False, +) -> dict[str, Any]: + """Fail closed when the session issue lock does not own the attempted work. + + * Author sessions that require a lock fail when none is held. + * When a lock exists, the target issue (tool argument) and/or the issue + number embedded in the branch must match the locked issue. + * Reviewer/merger/reconciler roles are not issue-scope owners of author + implementation work and skip the author lock requirement. + """ + role = (role_kind or "").strip().lower() + locked = locked_issue_number + if isinstance(locked, str) and locked.isdigit(): + locked = int(locked) + if locked is not None: + try: + locked = int(locked) + except (TypeError, ValueError): + locked = None + + target = target_issue_number + if target is not None: + try: + target = int(target) + except (TypeError, ValueError): + target = None + + branch_issue = extract_issue_number_from_branch(branch_name) + reasons: list[str] = [] + blocker_kind: str | None = None + + # Non-author roles do not take author issue locks for implementation. + if role in {"reviewer", "merger", "reconciler"}: + return _scope_ok(locked, target, branch_issue) + + if require_lock_for_author and locked is None: + reasons.append( + "no owning issue lock is bound for this author session; " + "source/test mutation requires selecting or creating an owning issue first" + ) + blocker_kind = BLOCKER_MISSING_ISSUE_SCOPE + + if locked is not None and target is not None and locked != target: + reasons.append( + f"session is locked to issue #{locked} but mutation targets issue " + f"#{target}; out-of-scope until the owning issue is selected" + ) + blocker_kind = BLOCKER_OUT_OF_SCOPE_ISSUE + + if locked is not None and branch_issue is not None and locked != branch_issue: + reasons.append( + f"session is locked to issue #{locked} but workspace branch is for " + f"issue #{branch_issue}; bind the matching issue-backed worktree" + ) + blocker_kind = BLOCKER_OUT_OF_SCOPE_ISSUE + + if reasons: + kind = blocker_kind or BLOCKER_MISSING_ISSUE_SCOPE + return { + "proven": False, + "block": True, + "blocker_kind": kind, + "exact_next_action": _NEXT_ACTIONS[kind], + "reasons": reasons, + "locked_issue_number": locked, + "target_issue_number": target, + "branch_issue_number": branch_issue, + } + return _scope_ok(locked, target, branch_issue) + + +def assess_root_source_mutation( + *, + workspace_path: str, + canonical_repo_root: str, + porcelain_status: str, + current_branch: str | None = None, + locked_issue_number: int | None = None, + role_kind: str | None = None, +) -> dict[str, Any]: + """Fail closed for diagnostic/source edits on the control/root checkout. + + Allowed only when the active workspace is under ``branches/``. Dirty + tracked source/test files on the control checkout always block, including + temporary/diagnostic/test-only intent. + """ + role = (role_kind or "").strip().lower() + if role == "reconciler": + return { + "proven": True, + "block": False, + "blocker_kind": None, + "exact_next_action": "proceed", + "reasons": [], + "dirty_source_files": [], + } + + root = os.path.realpath(canonical_repo_root or "") + workspace = os.path.realpath(workspace_path or root or ".") + under_branches = author_mutation_worktree.is_path_under_branches(workspace, root) + dirty_src = dirty_source_files(porcelain_status) + reasons: list[str] = [] + blocker_kind: str | None = None + + if not under_branches and workspace == root and dirty_src: + # Root workspace with source dirtiness is unattributed root WIP. + # (Clean-root author binding is enforced by branches-only #274.) + reasons.append( + "control/root checkout has tracked source or test edits " + f"(dirty files: {', '.join(dirty_src)}); diagnostic or temporary " + "edits on the root checkout are forbidden" + ) + blocker_kind = BLOCKER_ROOT_DIAGNOSTIC_EDIT + + if ( + not under_branches + and workspace == root + and not dirty_src + and role == "author" + ): + # Explicit missing-worktree signal for force-on author entrypoints. + reasons.append( + "author source/test mutation from the stable control checkout is " + "forbidden; bind an issue-backed worktree under branches/ first" + ) + blocker_kind = BLOCKER_MISSING_WORKTREE + + if reasons: + kind = blocker_kind or BLOCKER_ROOT_DIAGNOSTIC_EDIT + return { + "proven": False, + "block": True, + "blocker_kind": kind, + "exact_next_action": _NEXT_ACTIONS[kind], + "reasons": reasons, + "dirty_source_files": dirty_src, + "workspace_path": workspace, + "canonical_repo_root": root, + "under_branches": under_branches, + "locked_issue_number": locked_issue_number, + } + return { + "proven": True, + "block": False, + "blocker_kind": None, + "exact_next_action": "proceed", + "reasons": [], + "dirty_source_files": dirty_src, + "workspace_path": workspace, + "canonical_repo_root": root, + "under_branches": under_branches, + "locked_issue_number": locked_issue_number, + } + + +def assess_production_mutation_guards( + *, + workspace_path: str, + canonical_repo_root: str, + porcelain_status: str, + current_branch: str | None = None, + locked_issue_number: int | None = None, + target_issue_number: int | None = None, + role_kind: str | None = None, + require_author_lock: bool = False, + in_test_mode: bool = False, +) -> dict[str, Any]: + """Compose root + scope production guards when they must be active (#683).""" + if not production_guards_active(in_test_mode=in_test_mode): + return { + "proven": True, + "block": False, + "blocker_kind": None, + "exact_next_action": "proceed", + "reasons": [], + "skipped": True, + "skip_reason": "production guards not active (test isolation without force-on)", + } + + root_assess = assess_root_source_mutation( + workspace_path=workspace_path, + canonical_repo_root=canonical_repo_root, + porcelain_status=porcelain_status, + current_branch=current_branch, + locked_issue_number=locked_issue_number, + role_kind=role_kind, + ) + if root_assess["block"]: + return {**root_assess, "skipped": False} + + scope_assess = assess_issue_scope_ownership( + locked_issue_number=locked_issue_number, + target_issue_number=target_issue_number, + branch_name=current_branch, + role_kind=role_kind, + require_lock_for_author=require_author_lock, + ) + if scope_assess["block"]: + return {**scope_assess, "skipped": False} + + return { + "proven": True, + "block": False, + "blocker_kind": None, + "exact_next_action": "proceed", + "reasons": [], + "skipped": False, + "root": root_assess, + "scope": scope_assess, + } + + +def raise_if_blocked(assessment: dict[str, Any]) -> None: + """Raise :class:`ProductionGuardError` when *assessment* blocks.""" + if not assessment or not assessment.get("block"): + return + kind = assessment.get("blocker_kind") or BLOCKER_PRODUCTION_GUARD + reasons = list(assessment.get("reasons") or ["production guard violation"]) + message = ( + f"Workflow scope guard (#683) [{kind}]: {'; '.join(reasons)}. " + f"exact_next_action: {assessment.get('exact_next_action') or _NEXT_ACTIONS.get(kind, '')}" + ) + raise ProductionGuardError( + message, + blocker_kind=kind, + exact_next_action=assessment.get("exact_next_action"), + reasons=reasons, + details={ + k: v + for k, v in assessment.items() + if k + not in { + "proven", + "block", + "blocker_kind", + "exact_next_action", + "reasons", + } + }, + ) + + +def block_response( + assessment: dict[str, Any] | ProductionGuardError | None = None, + *, + blocker_kind: str | None = None, + reasons: list[str] | None = None, + exact_next_action: str | None = None, + **extra: Any, +) -> dict[str, Any]: + """Structured fail-closed tool response with typed blocker fields.""" + if isinstance(assessment, ProductionGuardError): + kind = assessment.blocker_kind + reason_list = list(assessment.reasons) + next_action = assessment.exact_next_action + extra = {**assessment.details, **extra} + elif isinstance(assessment, dict) and assessment.get("block"): + kind = assessment.get("blocker_kind") or BLOCKER_PRODUCTION_GUARD + reason_list = list(assessment.get("reasons") or []) + next_action = assessment.get("exact_next_action") or _NEXT_ACTIONS.get( + kind, _NEXT_ACTIONS[BLOCKER_PRODUCTION_GUARD] + ) + else: + kind = (blocker_kind or BLOCKER_PRODUCTION_GUARD).strip() + if kind not in BLOCKER_KINDS: + kind = BLOCKER_PRODUCTION_GUARD + reason_list = list(reasons or ["production guard violation"]) + next_action = exact_next_action or _NEXT_ACTIONS[kind] + + if kind not in BLOCKER_KINDS: + kind = BLOCKER_PRODUCTION_GUARD + if not reason_list: + reason_list = ["production guard violation"] + next_action = (next_action or "").strip() or _NEXT_ACTIONS[kind] + + out: dict[str, Any] = { + "success": False, + "performed": False, + "blocker_kind": kind, + "exact_next_action": next_action, + "reasons": reason_list, + } + for key, value in extra.items(): + if key not in out and value is not None: + out[key] = value + return out + + +def format_production_guard_error(assessment: dict[str, Any]) -> str: + """Single RuntimeError string carrying kind + exact next action.""" + kind = assessment.get("blocker_kind") or BLOCKER_PRODUCTION_GUARD + reasons = "; ".join(assessment.get("reasons") or ["production guard violation"]) + next_action = assessment.get("exact_next_action") or _NEXT_ACTIONS.get( + kind, _NEXT_ACTIONS[BLOCKER_PRODUCTION_GUARD] + ) + return ( + f"Workflow scope guard (#683) [{kind}]: {reasons}. " + f"exact_next_action: {next_action}" + ) + + +# ── durable failure recording ──────────────────────────────────────────────── + + +def record_workflow_failure( + *, + kind: str, + detail: str, + issue_number: int | None = None, + task: str | None = None, + sink: Any | None = None, +) -> dict[str, Any]: + """Record a workflow/tool failure before source edits continue (#683 AC8). + + *sink* may be a callable ``sink(record)`` (e.g. tests) or omitted for the + in-process ledger only. Returns the durable record. + """ + record = { + "kind": (kind or "workflow_failure").strip() or "workflow_failure", + "detail": (detail or "").strip(), + "issue_number": issue_number, + "task": task, + "pid": os.getpid(), + } + with _ledger_lock: + _failure_ledger.append(dict(record)) + if callable(sink): + sink(record) + return record + + +def clear_workflow_failure_ledger() -> None: + """Test helper: reset the in-process failure ledger.""" + with _ledger_lock: + _failure_ledger.clear() + + +def workflow_failure_ledger() -> list[dict[str, Any]]: + """Copy of durable in-process failure records.""" + with _ledger_lock: + return [dict(r) for r in _failure_ledger] + + +def assess_durable_failure_recorded( + *, + require_record: bool, + pending_source_mutation: bool, +) -> dict[str, Any]: + """Block source mutation when a workflow failure was not recorded first.""" + if not require_record or not pending_source_mutation: + return { + "proven": True, + "block": False, + "blocker_kind": None, + "exact_next_action": "proceed", + "reasons": [], + } + with _ledger_lock: + has_record = bool(_failure_ledger) + if has_record: + return { + "proven": True, + "block": False, + "blocker_kind": None, + "exact_next_action": "proceed", + "reasons": [], + } + return { + "proven": False, + "block": True, + "blocker_kind": BLOCKER_UNRECORDED_FAILURE, + "exact_next_action": _NEXT_ACTIONS[BLOCKER_UNRECORDED_FAILURE], + "reasons": [ + "workflow/tool failure triggered a need for source changes but no " + "durable failure record exists yet" + ], + } + + +def porcelain_preserves_python_paths(porcelain_status: str) -> bool: + """Regression helper: dirty ``*.py`` lines must remain visible (#683).""" + text = porcelain_status or "" + for line in text.splitlines(): + stripped = line.strip() + if stripped.endswith(".py") or ".py " in stripped or stripped.endswith(".py"): + # Any py path present proves no silent strip of all *.py lines. + if " M " in f" {stripped}" or stripped[:1] in "MADRCTU" or len(line) >= 4: + return True + # Empty porcelain is fine; integrity means we did not strip when present. + return ".py" not in text + + +def assert_no_pytest_porcelain_filter(source_text: str) -> list[str]: + """Static check: production reader must not strip ``*.py`` under pytest.""" + findings: list[str] = [] + lowered = source_text or "" + if "endswith(\".py\")" in lowered or "endswith('.py')" in lowered: + if "pytest" in lowered and "porcelain" in lowered.lower(): + findings.append( + "production porcelain reader must not filter *.py under pytest " + "(rejected 300a4ca pattern)" + ) + if "if \"pytest\" in sys.modules" in lowered and "porcelain" in lowered.lower(): + if ".py" in lowered and ("join" in lowered or "endswith" in lowered): + findings.append( + "test-mode porcelain filtering of source files is forbidden (#683)" + ) + return findings + + +def _scope_ok( + locked: int | None, + target: int | None, + branch_issue: int | None, +) -> dict[str, Any]: + return { + "proven": True, + "block": False, + "blocker_kind": None, + "exact_next_action": "proceed", + "reasons": [], + "locked_issue_number": locked, + "target_issue_number": target, + "branch_issue_number": branch_issue, + }