Compare commits

..
5 changed files with 86 additions and 392 deletions
+44 -155
View File
@@ -386,68 +386,6 @@ def run_compensating_recovery(
return recovery_info return recovery_info
def _normalize_sha(value: str | None) -> str | None:
"""Normalize a Git object id for comparison, or ``None`` when unknown."""
normalized = (value or "").strip().lower()
return normalized or None
def _author_bootstrap_assessment(
*,
not_applicable: bool,
allowed: bool,
block: bool,
reasons: list[str],
workspace: str,
root: str,
branch: str | None,
dirty: list[str],
under_branches: bool,
bootstrap_path: str | None = None,
local_head_sha: str | None = None,
remote_master_sha: str | None = None,
exact_next_action: str | None = None,
) -> dict[str, Any]:
"""Structured author-bootstrap assessment consumable by bootstrap_permits (#892).
Field shape mirrors :func:`create_issue_bootstrap._result` so the shared
``bootstrap_permits_control_checkout`` predicate can prove control-checkout
eligibility for ``gitea_bootstrap_author_issue_worktree`` the same way it
does for ``create_issue``. Allowed control assessments must use empty
``reasons`` — narrative belongs in other fields, not the refusal list.
"""
local_tip = _normalize_sha(local_head_sha)
remote_tip = _normalize_sha(remote_master_sha)
base_tips_verified = bool(local_tip and remote_tip and local_tip == remote_tip)
return {
"not_applicable": not_applicable,
"allowed": allowed,
"block": block,
"proven": bool(allowed and not block and not not_applicable),
"reasons": list(reasons),
"workspace_path": workspace,
"canonical_repo_root": root,
"current_branch": branch,
"dirty_files": list(dirty),
"under_branches": under_branches,
"exact_next_action": exact_next_action,
"bootstrap_path": bootstrap_path,
"task_scope": "author_issue_bootstrap",
"local_head_sha": local_tip,
"remote_master_sha": remote_tip,
"base_tips_verified": base_tips_verified,
}
EXACT_NEXT_ACTION_AUTHOR_BOOTSTRAP = (
"Restore the canonical control checkout to a clean accepted base branch "
"(master/main/dev) that matches live master, with no tracked local edits. "
"Re-resolve bootstrap_author_issue_worktree, then re-run "
"gitea_bootstrap_author_issue_worktree from that clean control checkout. "
"Do not use shell git worktree add as the primary path once bootstrap is healthy."
)
def assess_author_issue_bootstrap( def assess_author_issue_bootstrap(
*, *,
workspace_path: str, workspace_path: str,
@@ -459,13 +397,7 @@ def assess_author_issue_bootstrap(
remote_master_sha_error: str | None = None, remote_master_sha_error: str | None = None,
task: str | None = None, task: str | None = None,
) -> dict[str, Any]: ) -> dict[str, Any]:
"""Assess whether author issue worktree bootstrap may proceed from control or worktree root. """Assess whether author issue worktree bootstrap may proceed from control or worktree root."""
#892: control-checkout successes emit the full field set required by
``create_issue_bootstrap.bootstrap_permits_control_checkout`` (empty reasons,
task_scope, base tip proof, binding paths) so the #274/#604 guards can
waive control-checkout for this one sanctioned bootstrap task.
"""
root = os.path.realpath(canonical_repo_root or "") root = os.path.realpath(canonical_repo_root or "")
workspace = os.path.realpath(workspace_path or root or ".") workspace = os.path.realpath(workspace_path or root or ".")
branch = (current_branch or "").strip() branch = (current_branch or "").strip()
@@ -475,50 +407,34 @@ def assess_author_issue_bootstrap(
if root if root
else False else False
) )
local_tip = _normalize_sha(head_sha)
remote_tip = _normalize_sha(remote_master_sha)
if not is_author_issue_bootstrap_task(task): if not is_author_issue_bootstrap_task(task):
return _author_bootstrap_assessment( return {
not_applicable=True, "not_applicable": True,
allowed=False, "allowed": False,
block=False, "block": False,
reasons=["task is not author_issue_bootstrap"], "proven": False,
workspace=workspace, "reasons": ["task is not author_issue_bootstrap"],
root=root, }
branch=branch or None,
dirty=dirty,
under_branches=under_branches,
)
# Already under branches/: ordinary #274 path applies; not a control waiver.
if under_branches: if under_branches:
return _author_bootstrap_assessment( return {
not_applicable=True, "not_applicable": False,
allowed=False, "allowed": True,
block=False, "block": False,
reasons=["workspace is under branches/; ordinary #274 path applies"], "proven": True,
workspace=workspace, "bootstrap_path": "existing_branches_worktree",
root=root, "reasons": [
branch=branch or None, "workspace is already a registered worktree under branches/"
dirty=dirty, ],
under_branches=True, }
bootstrap_path="existing_branches_worktree",
local_head_sha=local_tip,
remote_master_sha=remote_tip,
)
reasons: list[str] = [] reasons: list[str] = []
if not root or workspace != root: if workspace != root:
reasons.append( reasons.append(
"bootstrap requires workspace to be canonical control checkout or branches/ worktree" "bootstrap requires workspace to be canonical control checkout or branches/ worktree"
) )
if not branch: if branch not in author_mutation_worktree.BASE_BRANCHES:
reasons.append(
"control checkout is detached HEAD; expected an accepted base branch "
f"({', '.join(sorted(author_mutation_worktree.BASE_BRANCHES))})"
)
elif branch not in author_mutation_worktree.BASE_BRANCHES:
reasons.append( reasons.append(
f"control checkout branch '{branch}' is not an accepted base branch " f"control checkout branch '{branch}' is not an accepted base branch "
f"({', '.join(sorted(author_mutation_worktree.BASE_BRANCHES))})" f"({', '.join(sorted(author_mutation_worktree.BASE_BRANCHES))})"
@@ -528,64 +444,37 @@ def assess_author_issue_bootstrap(
f"control checkout has tracked local edits: {', '.join(dirty[:5])}" f"control checkout has tracked local edits: {', '.join(dirty[:5])}"
) )
# Fail closed on missing tip proof (same bar as create_issue bootstrap #757). if remote_master_sha_error:
if not local_tip:
reasons.append( reasons.append(
"control checkout HEAD SHA is unknown; base equivalence to live " f"could not verify live master tip: {remote_master_sha_error}"
"master cannot be proven (fail closed)"
) )
resolver_error = (remote_master_sha_error or "").strip() or None elif remote_master_sha and head_sha:
if resolver_error: h = head_sha.strip().lower()
rm = remote_master_sha.strip().lower()
if h != rm:
reasons.append( reasons.append(
f"live master tip could not be resolved ({resolver_error}); " f"control checkout HEAD ({h[:12]}) != live master tip ({rm[:12]})"
"base equivalence cannot be proven (fail closed)"
)
elif not remote_tip:
reasons.append(
"live master tip is unknown; base equivalence cannot be proven "
"(fail closed)"
)
elif local_tip and remote_tip and local_tip != remote_tip:
reasons.append(
f"control checkout HEAD ({local_tip[:12]}) != live master tip "
f"({remote_tip[:12]})"
) )
if reasons: if reasons:
return _author_bootstrap_assessment( return {
not_applicable=False, "not_applicable": False,
allowed=False, "allowed": False,
block=True, "block": True,
reasons=reasons, "proven": False,
workspace=workspace, "reasons": reasons,
root=root, }
branch=branch or None,
dirty=dirty,
under_branches=False,
local_head_sha=local_tip,
remote_master_sha=remote_tip,
exact_next_action=EXACT_NEXT_ACTION_AUTHOR_BOOTSTRAP,
)
# Allowed: empty reasons so bootstrap_permits_control_checkout can pass. return {
return _author_bootstrap_assessment( "not_applicable": False,
not_applicable=False, "allowed": True,
allowed=True, "block": False,
block=False, "proven": True,
reasons=[], "bootstrap_path": "clean_canonical_control_checkout",
workspace=workspace, "reasons": [
root=root, "control checkout is clean on accepted base branch matching live master"
branch=branch or None, ],
dirty=dirty, }
under_branches=False,
bootstrap_path="clean_canonical_control_checkout",
local_head_sha=local_tip,
remote_master_sha=remote_tip,
exact_next_action=(
"Call gitea_bootstrap_author_issue_worktree with the allocated "
"issue/lease pins; it will create the branches/ worktree and lock."
),
)
import fcntl import fcntl
+6 -18
View File
@@ -241,14 +241,9 @@ def bootstrap_permits_control_checkout(
caller's ordinary block in force. caller's ordinary block in force.
``assessment`` is server-derived only: it is produced by ``assessment`` is server-derived only: it is produced by
:func:`assess_create_issue_bootstrap` or :func:`assess_create_issue_bootstrap` from inspected repository state. It is
:func:`author_issue_bootstrap.assess_author_issue_bootstrap` from inspected never accepted from an MCP tool argument, so no caller can assert
repository state. It is never accepted from an MCP tool argument, so no eligibility it has not proven.
caller can assert eligibility it has not proven.
#892: author issue worktree bootstrap uses the same predicate with
``task_scope='author_issue_bootstrap'`` so a clean control checkout can
create the first ``branches/`` worktree without the lock↔worktree cycle.
""" """
if not isinstance(assessment, dict): if not isinstance(assessment, dict):
return False return False
@@ -269,16 +264,9 @@ def bootstrap_permits_control_checkout(
if assessment.get("reasons"): if assessment.get("reasons"):
return False return False
# Scope proof: create_issue (#749) or author issue bootstrap (#850/#892), # Scope proof: only the create_issue bootstrap, only via the clean
# only via the clean canonical control checkout path. # canonical control checkout path.
task_scope = assessment.get("task_scope") if assessment.get("task_scope") != "create_issue_only":
if is_create_issue_task(task):
if task_scope != "create_issue_only":
return False
elif author_issue_bootstrap.is_author_issue_bootstrap_task(task):
if task_scope != "author_issue_bootstrap":
return False
else:
return False return False
if assessment.get("bootstrap_path") != "clean_canonical_control_checkout": if assessment.get("bootstrap_path") != "clean_canonical_control_checkout":
return False return False
+2 -2
View File
@@ -577,10 +577,10 @@ def _get_workspace_porcelain(worktree_path: str | None = None) -> str:
def _parse_porcelain_entries(porcelain: str) -> dict[str, str]: def _parse_porcelain_entries(porcelain: str) -> dict[str, str]:
"""Map tracked path -> full porcelain line (untracked ``??`` ignored).""" """Map path -> full porcelain line (including untracked ``??``)."""
entries: dict[str, str] = {} entries: dict[str, str] = {}
for line in (porcelain or "").splitlines(): for line in (porcelain or "").splitlines():
if not line or len(line) < 4 or line.startswith("??"): if not line or len(line) < 4:
continue continue
path = line[3:].strip() path = line[3:].strip()
if " -> " in path: if " -> " in path:
@@ -0,0 +1,32 @@
"""Regression test for Issue #697: Native workspace preflight explicitly includes untracked files.
Verifies that `_parse_porcelain_entries` includes untracked files (lines starting with '??')
so workspace preflight fails closed on untracked or dirty state.
"""
from __future__ import annotations
import unittest
import gitea_mcp_server as mcp_server
class TestIssue697UntrackedPreflight(unittest.TestCase):
def test_parse_porcelain_entries_includes_untracked_files(self):
porcelain = " M gitea_mcp_server.py\n?? untracked_file.txt\n?? new_dir/test.py\n"
entries = mcp_server._parse_porcelain_entries(porcelain)
self.assertIn("untracked_file.txt", entries)
self.assertIn("new_dir/test.py", entries)
self.assertEqual(entries["untracked_file.txt"], "?? untracked_file.txt")
self.assertEqual(entries["new_dir/test.py"], "?? new_dir/test.py")
def test_parse_porcelain_entries_includes_tracked_and_untracked(self):
porcelain = " M tracked.py\n?? untracked.py\n D deleted.py\n"
entries = mcp_server._parse_porcelain_entries(porcelain)
self.assertEqual(len(entries), 3)
self.assertIn("tracked.py", entries)
self.assertIn("untracked.py", entries)
self.assertIn("deleted.py", entries)
if __name__ == "__main__":
unittest.main()
@@ -1,215 +0,0 @@
"""Regression: author worktree bootstrap from clean control checkout (#892).
#892 is the four-door deadlock where every documented recovery path is closed:
bootstrap refuses control, lock demands an existing worktree, worktree-start
demands a lock, and shell worktree add is outside the sanctioned MCP path.
Root cause: assess_author_issue_bootstrap returned allowed/proven for a clean
control checkout, but bootstrap_permits_control_checkout only accepted
create_issue assessments (task_scope=create_issue_only + empty reasons + full
base-tip field set). Author assessments never satisfied the shared predicate,
so the #274/#604 guards kept the ordinary control-checkout block.
"""
from __future__ import annotations
import os
import tempfile
import unittest
from unittest import mock
import author_issue_bootstrap as aib
import create_issue_bootstrap as cib
CONTROL = "/repo/Gitea-Tools"
MASTER = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
OTHER = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
def _assess(
*,
workspace=CONTROL,
root=CONTROL,
branch="master",
head=MASTER,
porcelain="",
remote=MASTER,
remote_error=None,
task="bootstrap_author_issue_worktree",
):
return aib.assess_author_issue_bootstrap(
workspace_path=workspace,
canonical_repo_root=root,
current_branch=branch,
head_sha=head,
porcelain_status=porcelain,
remote_master_sha=remote,
remote_master_sha_error=remote_error,
task=task,
)
class TestAuthorBootstrapAssessmentShape(unittest.TestCase):
def test_clean_control_emits_predicate_compatible_fields(self):
assessment = _assess()
self.assertTrue(assessment["allowed"])
self.assertTrue(assessment["proven"])
self.assertFalse(assessment["block"])
self.assertFalse(assessment["not_applicable"])
self.assertEqual(assessment["reasons"], [])
self.assertEqual(assessment["task_scope"], "author_issue_bootstrap")
self.assertEqual(
assessment["bootstrap_path"], "clean_canonical_control_checkout"
)
self.assertEqual(assessment["dirty_files"], [])
self.assertIs(assessment["under_branches"], False)
self.assertTrue(assessment["base_tips_verified"])
self.assertEqual(assessment["local_head_sha"], MASTER)
self.assertEqual(assessment["remote_master_sha"], MASTER)
self.assertEqual(assessment["workspace_path"], os.path.realpath(CONTROL))
self.assertEqual(
assessment["canonical_repo_root"], os.path.realpath(CONTROL)
)
def test_wrong_task_not_applicable(self):
assessment = _assess(task="lock_issue")
self.assertTrue(assessment["not_applicable"])
self.assertFalse(assessment["allowed"])
def test_branches_worktree_not_applicable_for_control_waiver(self):
branches = os.path.join(CONTROL, "branches", "fix-issue-1")
assessment = _assess(workspace=branches)
self.assertTrue(assessment["not_applicable"])
self.assertFalse(assessment["allowed"])
self.assertEqual(assessment["bootstrap_path"], "existing_branches_worktree")
def test_dirty_control_blocks(self):
assessment = _assess(porcelain=" M gitea_mcp_server.py\n")
self.assertTrue(assessment["block"])
self.assertFalse(assessment["allowed"])
self.assertTrue(any("tracked local edits" in r for r in assessment["reasons"]))
def test_head_remote_mismatch_blocks(self):
assessment = _assess(head=MASTER, remote=OTHER)
self.assertTrue(assessment["block"])
self.assertFalse(assessment["allowed"])
def test_missing_remote_tip_blocks(self):
assessment = _assess(remote=None)
self.assertTrue(assessment["block"])
self.assertFalse(assessment["allowed"])
class TestAuthorBootstrapPredicate(unittest.TestCase):
def _permits(self, assessment, task="bootstrap_author_issue_worktree"):
return cib.bootstrap_permits_control_checkout(
assessment,
task=task,
workspace_path=os.path.realpath(CONTROL),
canonical_repo_root=os.path.realpath(CONTROL),
)
def test_clean_author_bootstrap_permits(self):
self.assertTrue(self._permits(_assess()))
def test_tool_alias_permits(self):
assessment = _assess(task="gitea_bootstrap_author_issue_worktree")
self.assertTrue(
self._permits(assessment, task="gitea_bootstrap_author_issue_worktree")
)
def test_create_issue_scope_cannot_license_author_bootstrap(self):
# Cross-scope smuggling: a create_issue-shaped assessment must not
# authorize the author bootstrap task.
create_shaped = dict(_assess())
create_shaped["task_scope"] = "create_issue_only"
self.assertFalse(self._permits(create_shaped))
def test_author_scope_cannot_license_create_issue(self):
assessment = _assess()
self.assertFalse(
cib.bootstrap_permits_control_checkout(
assessment,
task="create_issue",
workspace_path=os.path.realpath(CONTROL),
canonical_repo_root=os.path.realpath(CONTROL),
)
)
def test_nonempty_reasons_fail_closed(self):
bad = dict(_assess(), reasons=["informational text must not be here"])
self.assertFalse(self._permits(bad))
def test_dirty_fails_closed(self):
self.assertFalse(self._permits(_assess(porcelain=" M x.py\n")))
def test_mismatch_fails_closed(self):
self.assertFalse(self._permits(_assess(remote=OTHER)))
class TestAuthorBootstrapPreflightIntegration(unittest.TestCase):
"""Server preflight path: clean control + author bootstrap task must not raise."""
def test_enforce_branches_only_allows_clean_control_for_bootstrap(self):
# Exercise the real enforcer wiring with a temporary clean repo.
import gitea_mcp_server as srv
with tempfile.TemporaryDirectory() as tmp:
repo = os.path.join(tmp, "repo")
os.makedirs(os.path.join(repo, "branches"))
# Minimal git repo on master at a known tip.
import subprocess
subprocess.check_call(["git", "init", "-b", "master", repo])
subprocess.check_call(
["git", "-C", repo, "commit", "--allow-empty", "-m", "init"]
)
head = subprocess.check_output(
["git", "-C", repo, "rev-parse", "HEAD"], text=True
).strip()
assessment = aib.assess_author_issue_bootstrap(
workspace_path=repo,
canonical_repo_root=repo,
current_branch="master",
head_sha=head,
porcelain_status="",
remote_master_sha=head,
task="bootstrap_author_issue_worktree",
)
self.assertTrue(
cib.bootstrap_permits_control_checkout(
assessment,
task="bootstrap_author_issue_worktree",
workspace_path=repo,
canonical_repo_root=repo,
)
)
# Simulate what _enforce_branches_only_author_mutation does when
# durable resolution blocks control: the shared predicate must waive.
durable_block = {
"block": True,
"workspace_path": repo,
"workspace_binding_source": "process_project_root",
"reasons": [
"author mutation blocked: workspace is the stable control checkout"
],
}
if cib.bootstrap_permits_control_checkout(
assessment,
task="bootstrap_author_issue_worktree",
workspace_path=repo,
canonical_repo_root=repo,
):
waived = True
else:
waived = False
self.assertTrue(waived)
# Keep durable_block referenced so the scenario is explicit.
self.assertTrue(durable_block["block"])
if __name__ == "__main__":
unittest.main()