Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 67 Pull Requests Actions Packages Projects Releases Wiki Activity

fix: durable author worktree resolution without control fallback (Closes #618)

Browse Source 
Author mutation tools now resolve workspace via explicit worktree_path,
env bindings, or the active author issue lock — never silent fallback to
the control checkout/master. Missing configured bindings fail closed with
operator recovery; create_issue and create_issue_comment agree.

Recovered onto 0568f44 from preserved candidate cbf56ccd (AUTHOR_RECOVERY).
LLM_LOCK_ID=author-618-recovery-508eb3162d01-1784569919

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
jcwalker3
2026-07-20 12:55:40 -05:00
co-authored by Claude Opus 4.8
parent 0568f44cb2
commit 5ed2ab8a38
9 changed files with 1335 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_author_mutation_worktree.py
+60 -13
View File
@@ -79,18 +79,33 @@ class TestPreflightIntegration(unittest.TestCase):
mcp_server._preflight_whoami_called = True
mcp_server._preflight_capability_called = True
mcp_server._preflight_resolved_role = "author"
mcp_server._preflight_resolved_task = None
control_root = "/repo/Gitea-Tools"
with mock.patch.object(mcp_server, "PROJECT_ROOT", control_root):
with mock.patch.object(mcp_server, "_enforce_root_checkout_guard"):
with mock.patch("gitea_auth.get_profile", return_value={"profile_name": "gitea-author"}):
with mock.patch.dict(
"os.environ",
{"GITEA_TEST_PORCELAIN": ""},
clear=False,
with mock.patch(
"gitea_mcp_server._session_author_lock_worktree",
return_value=None,
):
with mock.patch(
"gitea_auth.get_profile",
return_value={"profile_name": "gitea-author"},
):
with self.assertRaises(RuntimeError) as ctx:
mcp_server.verify_preflight_purity()
self.assertIn("Branches-only mutation guard", str(ctx.exception))
with mock.patch.dict(
"os.environ",
{"GITEA_TEST_PORCELAIN": ""},
clear=False,
):
with self.assertRaises(RuntimeError) as ctx:
mcp_server.verify_preflight_purity()
blob = str(ctx.exception)
self.assertTrue(
"Branches-only mutation guard" in blob
or "control checkout" in blob
or "author worktree" in blob.lower()
or "#618" in blob,
msg=blob,
)
def test_verify_preflight_allows_branches_worktree(self):
import mcp_server
@@ -98,14 +113,46 @@ class TestPreflightIntegration(unittest.TestCase):
mcp_server._preflight_whoami_called = True
mcp_server._preflight_capability_called = True
mcp_server._preflight_resolved_role = "author"
mcp_server._preflight_resolved_task = None
worktree = "/repo/Gitea-Tools/branches/issue-274"
healthy_ctx = {
"workspace_path": worktree,
"workspace_binding_source": "worktree_path argument",
"workspace_role_kind": "author",
"ignored_bindings": [],
"process_project_root": "/repo/Gitea-Tools",
"canonical_repo_root": "/repo/Gitea-Tools",
"roots_aligned": True,
"bound_worktree_missing": False,
"author_worktree_block": False,
"author_worktree_reasons": [],
"author_worktree_resolution": {
"proven": True,
"block": False,
"bound_worktree_missing": False,
"workspace_path": worktree,
"workspace_binding_source": "worktree_path argument",
"reasons": [],
},
"path_exists": True,
"in_git_worktree_list": True,
"inspected_git_root": worktree,
}
with mock.patch.object(mcp_server, "_enforce_root_checkout_guard"):
with mock.patch.dict(
"os.environ",
{"GITEA_TEST_PORCELAIN": ""},
clear=False,
with mock.patch.object(
mcp_server, "_session_author_lock_worktree", return_value=None
):
mcp_server.verify_preflight_purity(worktree_path=worktree)
with mock.patch.object(
mcp_server,
"_resolve_namespace_mutation_context",
return_value=healthy_ctx,
):
with mock.patch.dict(
"os.environ",
{"GITEA_TEST_PORCELAIN": ""},
clear=False,
):
mcp_server.verify_preflight_purity(worktree_path=worktree)
if __name__ == "__main__":
tests/test_create_issue_workspace_guard.py
+8
View File
@@ -26,15 +26,23 @@ class TestCreateIssueWorkspaceGuard(unittest.TestCase):
srv._preflight_whoami_called = True
srv._preflight_capability_called = True
srv._preflight_resolved_role = "author"
srv._preflight_resolved_task = "create_issue"
srv._preflight_whoami_violation = False
srv._preflight_capability_violation = False
# Disable early return in verify_preflight_purity for testing
self._orig_in_test = srv._preflight_in_test_mode
srv._preflight_in_test_mode = lambda: False
# #618: isolate from ambient session issue locks
self._lock_patch = patch(
"gitea_mcp_server._session_author_lock_worktree", return_value=None
)
self._lock_patch.start()
def tearDown(self):
srv._preflight_in_test_mode = self._orig_in_test
srv._preflight_resolved_task = None
self._lock_patch.stop()
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
@patch("gitea_mcp_server._profile_permission_block", return_value=None)
tests/test_issue_618_author_worktree_resolution.py
+349
View File
@@ -0,0 +1,349 @@
"""Regression tests for durable author worktree resolution (#618)."""
from __future__ import annotations
import os
import sys
import tempfile
import unittest
from pathlib import Path
from unittest import mock
from unittest.mock import MagicMock, patch
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import author_mutation_worktree as amw # noqa: E402
import gitea_mcp_server as srv # noqa: E402
import namespace_workspace_binding as nwb # noqa: E402
FAKE_AUTH = {"Authorization": "token test-token"}
current_file_path = Path(__file__).resolve()
if "branches" in current_file_path.parts:
CONTROL_CHECKOUT_ROOT = str(current_file_path.parents[3])
else:
CONTROL_CHECKOUT_ROOT = str(current_file_path.parents[1])
class TestDurableAuthorWorktreeResolution(unittest.TestCase):
def test_missing_author_env_fails_closed_no_control_fallback(self):
missing = "/nonexistent/branches/mcp-author-clean-ns"
result = amw.resolve_durable_author_worktree(
process_project_root=CONTROL_CHECKOUT_ROOT,
author_worktree_env=missing,
canonical_repo_root=CONTROL_CHECKOUT_ROOT,
)
self.assertTrue(result["block"])
self.assertTrue(result["bound_worktree_missing"])
self.assertFalse(result["silent_control_fallback"])
self.assertIn(amw.BOUND_WORKTREE_MISSING_MESSAGE, result["reasons"][0])
self.assertNotEqual(
os.path.realpath(result["workspace_path"]),
os.path.realpath(CONTROL_CHECKOUT_ROOT),
)
def test_missing_active_env_fails_closed(self):
missing = "/nonexistent/branches/deleted-active"
result = amw.resolve_durable_author_worktree(
process_project_root=CONTROL_CHECKOUT_ROOT,
active_worktree_env=missing,
canonical_repo_root=CONTROL_CHECKOUT_ROOT,
)
self.assertTrue(result["block"])
self.assertTrue(result["bound_worktree_missing"])
self.assertIn(amw.ACTIVE_WORKTREE_ENV, result["workspace_binding_source"])
def test_derives_from_active_author_issue_lock(self):
lock_wt = os.path.join(CONTROL_CHECKOUT_ROOT, "branches", "issue-618-lock")
result = amw.resolve_durable_author_worktree(
process_project_root=CONTROL_CHECKOUT_ROOT,
session_lock_worktree=lock_wt,
canonical_repo_root=CONTROL_CHECKOUT_ROOT,
validate=False,
)
self.assertEqual(
result["workspace_path"], os.path.realpath(os.path.abspath(lock_wt))
)
self.assertIn("issue lock", result["workspace_binding_source"])
def test_explicit_worktree_path_wins_over_lock(self):
explicit = os.path.join(CONTROL_CHECKOUT_ROOT, "branches", "issue-618-explicit")
lock_wt = os.path.join(CONTROL_CHECKOUT_ROOT, "branches", "issue-618-lock")
result = amw.resolve_durable_author_worktree(
worktree_path=explicit,
process_project_root=CONTROL_CHECKOUT_ROOT,
session_lock_worktree=lock_wt,
canonical_repo_root=CONTROL_CHECKOUT_ROOT,
validate=False,
)
self.assertEqual(
result["workspace_path"], os.path.realpath(os.path.abspath(explicit))
)
self.assertEqual(result["workspace_binding_source"], "worktree_path argument")
def test_no_binding_does_not_silently_use_control_checkout(self):
result = amw.resolve_durable_author_worktree(
process_project_root=CONTROL_CHECKOUT_ROOT,
canonical_repo_root=CONTROL_CHECKOUT_ROOT,
)
self.assertTrue(result["block"])
self.assertFalse(result["silent_control_fallback"])
blob = " ".join(result["reasons"])
self.assertIn("control checkout", blob)
self.assertIn("forbidden", blob)
def test_process_root_under_branches_is_allowed(self):
branches_root = os.path.join(CONTROL_CHECKOUT_ROOT, "branches", "session-wt")
result = amw.resolve_durable_author_worktree(
process_project_root=branches_root,
canonical_repo_root=CONTROL_CHECKOUT_ROOT,
validate=False,
)
self.assertFalse(result["block"])
self.assertEqual(
result["workspace_path"], os.path.realpath(branches_root)
)
self.assertIn("branches/", result["workspace_binding_source"])
def test_lock_ownership_mismatch_fails_closed(self):
with tempfile.TemporaryDirectory() as tmp:
root = tmp
branches = os.path.join(root, "branches")
os.makedirs(os.path.join(branches, "a"))
os.makedirs(os.path.join(branches, "b"))
# Seed a fake .git so membership/list may soft-fail without hard error
os.makedirs(os.path.join(root, ".git"))
result = amw.resolve_durable_author_worktree(
worktree_path=os.path.join(branches, "a"),
process_project_root=root,
session_lock_worktree=os.path.join(branches, "b"),
canonical_repo_root=root,
validate=True,
)
self.assertTrue(result["block"])
self.assertTrue(
any("lock" in r.lower() and "match" in r.lower() for r in result["reasons"])
)
def test_traversal_safety_blocks_escape(self):
assessment = amw.assess_path_traversal_safety(
path="/tmp/other-repo/branches/evil",
canonical_repo_root=CONTROL_CHECKOUT_ROOT,
)
self.assertTrue(assessment["block"])
self.assertTrue(any("escapes" in r for r in assessment["reasons"]))
def test_bound_worktree_existence_reports_null_git_root(self):
assessment = amw.assess_bound_worktree_existence(
configured_path="/nonexistent/branches/gone",
binding_source=f"{amw.AUTHOR_WORKTREE_ENV} environment variable",
canonical_repo_root=CONTROL_CHECKOUT_ROOT,
profile_name="prgs-author",
)
self.assertTrue(assessment["block"])
self.assertIsNone(assessment["inspected_git_root"])
self.assertFalse(assessment["path_exists"])
msg = amw.format_bound_worktree_missing_error(assessment)
self.assertIn(amw.BOUND_WORKTREE_MISSING_MESSAGE, msg)
self.assertIn("prgs-author", msg)
self.assertIn("recreate or repoint", msg.lower())
class TestNamespaceAuthorNoDemotion(unittest.TestCase):
def test_author_missing_env_not_demoted_to_process_root(self):
missing = "/nonexistent/branches/mcp-author-clean-ns"
demotions: list[str] = []
path, source = nwb.resolve_namespace_workspace(
role_kind="author",
process_project_root=CONTROL_CHECKOUT_ROOT,
env={amw.AUTHOR_WORKTREE_ENV: missing},
demotions=demotions,
verify_paths=True,
)
self.assertIn("AUTHOR", source)
self.assertNotEqual(os.path.realpath(path), os.path.realpath(CONTROL_CHECKOUT_ROOT))
self.assertTrue(any("not demoted" in d for d in demotions))
def test_reviewer_still_demotes_missing_env(self):
"""#702 demotion retained for non-author roles."""
demotions: list[str] = []
path, source = nwb.resolve_namespace_workspace(
role_kind="reviewer",
process_project_root=CONTROL_CHECKOUT_ROOT,
env={"GITEA_ACTIVE_WORKTREE": "/nonexistent/branches/review-gone"},
demotions=demotions,
verify_paths=True,
)
self.assertEqual(source, "MCP server process root (default)")
self.assertEqual(path, os.path.realpath(CONTROL_CHECKOUT_ROOT))
self.assertTrue(demotions)
def test_mutation_context_surfaces_missing_binding_health(self):
ctx = nwb.resolve_namespace_mutation_context(
role_kind="author",
worktree_path=None,
process_project_root=CONTROL_CHECKOUT_ROOT,
env={amw.AUTHOR_WORKTREE_ENV: "/nonexistent/branches/mcp-author-clean-ns"},
profile_name="prgs-author",
)
self.assertTrue(ctx.get("bound_worktree_missing"))
self.assertTrue(ctx.get("author_worktree_block"))
self.assertIsNone(ctx.get("inspected_git_root"))
self.assertFalse(ctx.get("path_exists"))
class TestCreateIssueAndCommentAgreeOnMissingWorktree(unittest.TestCase):
"""AC3/AC4: create_issue and create_issue_comment enforce the same rule."""
def setUp(self):
srv._preflight_whoami_called = True
srv._preflight_capability_called = True
srv._preflight_resolved_role = "author"
srv._preflight_resolved_task = None
srv._preflight_whoami_violation = False
srv._preflight_capability_violation = False
self._orig_in_test = srv._preflight_in_test_mode
srv._preflight_in_test_mode = lambda: False
self._lock_patch = patch(
"gitea_mcp_server._session_author_lock_worktree", return_value=None
)
self._lock_patch.start()
self.addCleanup(self._restore)
def _restore(self):
srv._preflight_in_test_mode = self._orig_in_test
srv._preflight_resolved_task = None
self._lock_patch.stop()
os.environ.pop(amw.AUTHOR_WORKTREE_ENV, None)
os.environ.pop(amw.ACTIVE_WORKTREE_ENV, None)
def _assert_blocked_missing(self, result_or_exc):
if isinstance(result_or_exc, BaseException):
blob = str(result_or_exc)
else:
blob = " ".join(
str(x)
for x in (
result_or_exc.get("reasons") or [],
result_or_exc.get("message"),
result_or_exc.get("blocker_kind"),
)
if x
)
if not blob:
blob = str(result_or_exc)
self.assertTrue(
amw.BOUND_WORKTREE_MISSING_MESSAGE in blob
or "does not exist" in blob
or "bound worktree" in blob.lower(),
msg=blob,
)
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
@patch("gitea_mcp_server._profile_permission_block", return_value=None)
@patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
@patch(
"gitea_mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []),
)
@patch("gitea_mcp_server.api_request")
@patch("gitea_mcp_server.api_get_all", return_value=[])
def test_create_issue_blocked_when_author_env_missing(
self, _get_all, mock_api, _role, _ns, _prof, _auth
):
missing = os.path.join(
CONTROL_CHECKOUT_ROOT, "branches", "nonexistent-618-author-env"
)
os.environ[amw.AUTHOR_WORKTREE_ENV] = missing
srv._preflight_resolved_task = "create_issue"
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
try:
res = srv.gitea_create_issue(title="Test issue", body="body text here")
except RuntimeError as exc:
self._assert_blocked_missing(exc)
else:
self.assertFalse(res.get("success", True) and res.get("number"))
self._assert_blocked_missing(res)
mock_api.assert_not_called()
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
@patch("gitea_mcp_server.api_request")
def test_create_issue_comment_blocked_when_author_env_missing(self, mock_api, _auth):
missing = os.path.join(
CONTROL_CHECKOUT_ROOT, "branches", "nonexistent-618-author-env"
)
os.environ[amw.AUTHOR_WORKTREE_ENV] = missing
srv._preflight_resolved_task = "comment_issue"
author_env = {
"GITEA_PROFILE_NAME": "gitea-author",
"GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.issue.comment",
amw.AUTHOR_WORKTREE_ENV: missing,
}
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
with patch.dict(os.environ, author_env, clear=False):
try:
res = srv.gitea_create_issue_comment(
issue_number=618,
body="evidence comment",
remote="prgs",
)
except RuntimeError as exc:
self._assert_blocked_missing(exc)
else:
self.assertFalse(res.get("success", True))
self._assert_blocked_missing(res)
mock_api.assert_not_called()
class TestRuntimeContextUnhealthyMissingWorktree(unittest.TestCase):
def setUp(self):
srv._preflight_whoami_called = True
srv._preflight_capability_called = True
srv._preflight_resolved_role = "author"
srv._preflight_whoami_violation = False
srv._preflight_capability_violation = False
self._lock_patch = patch(
"gitea_mcp_server._session_author_lock_worktree", return_value=None
)
self._lock_patch.start()
def tearDown(self):
self._lock_patch.stop()
os.environ.pop(amw.AUTHOR_WORKTREE_ENV, None)
def test_assess_preflight_reports_null_git_root_and_missing(self):
missing = "/nonexistent/branches/mcp-author-clean-ns"
os.environ[amw.AUTHOR_WORKTREE_ENV] = missing
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
with patch("gitea_mcp_server.get_profile", return_value={
"profile_name": "prgs-author",
"allowed_operations": ["gitea.pr.create"],
"forbidden_operations": [],
}):
status = srv.assess_preflight_status()
self.assertFalse(status["preflight_ready"])
blob = " ".join(status["preflight_block_reasons"])
self.assertIn(amw.BOUND_WORKTREE_MISSING_MESSAGE, blob)
details = status["preflight_workspace"]
self.assertIsNotNone(details)
self.assertTrue(details.get("bound_worktree_missing"))
self.assertIsNone(details.get("inspected_git_root"))
self.assertFalse(details.get("path_exists"))
self.assertFalse(details.get("workspace_healthy"))
class TestThreadLedgerExample(unittest.TestCase):
def test_bound_worktree_missing_ledger_example_exists(self):
import thread_state_ledger_examples as examples
names = [name for name, _h, _l in examples.EXAMPLES]
self.assertIn("bound_worktree_missing_blocker", names)
for name, _handoff, ledger in examples.EXAMPLES:
if name == "bound_worktree_missing_blocker":
self.assertIn(amw.BOUND_WORKTREE_MISSING_MESSAGE, ledger)
self.assertIn("inspected_git_root", ledger)
self.assertIn("operator", ledger.lower())
break
if __name__ == "__main__":
unittest.main()
tests/test_namespace_workspace_binding.py
+14 -3
View File
@@ -224,9 +224,20 @@ class TestNamespaceWorkspaceIntegration(unittest.TestCase):
"gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
return_value={"current_branch": "master"},
):
with self.assertRaises(RuntimeError) as ctx:
srv.verify_preflight_purity("prgs")
self.assertIn("stable control checkout", str(ctx.exception))
with mock.patch(
"gitea_mcp_server._session_author_lock_worktree",
return_value=None,
):
with self.assertRaises(RuntimeError) as ctx:
srv.verify_preflight_purity("prgs")
blob = str(ctx.exception)
self.assertTrue(
"stable control checkout" in blob
or "control checkout" in blob
or "#618" in blob
or "author worktree" in blob.lower(),
msg=blob,
)
@mock.patch("subprocess.run")
@mock.patch("os.path.isdir", return_value=True)