fix(mcp): require proven base equivalence for the create-issue bootstrap
Remediates review #473 (REQUEST_CHANGES) on PR #759 / issue #757 AC3/AC4. The bootstrap compared SHAs only inside: if remote_tip and local_tip and remote_tip != local_tip: so agreement was assumed whenever either tip was unknown. A missing local HEAD, an unresolvable live master, or a resolver exception silently granted the control-checkout exemption instead of blocking it. An unknown tip is missing evidence, not proof of equivalence. Changes: * assess_create_issue_bootstrap now blocks unless BOTH tips are known and equal, with distinct reasons for missing local HEAD, unknown live master, and resolver failure. Adds a remote_master_sha_error parameter so a failed resolution is reported as missing evidence rather than absence of a constraint. * Both normalized SHAs and a derived base_tips_verified flag are recorded on the assessment. normalize_sha() treats only case and surrounding whitespace as equivalent spellings of a commit. * bootstrap_permits_control_checkout re-derives the comparison from the recorded tips instead of trusting base_tips_verified, so a hand-built or truncated assessment cannot assert agreement it never proved. * _create_issue_bootstrap_assessment captures the resolver exception and forwards it, replacing the silent except -> None. One shared assessment still serves both the #274 and #604 guards, and _BOOTSTRAP_UNSET is preserved. No MCP tool signature gains a bootstrap argument (AC6). No issue or PR number is special-cased (AC10). Tests: 16 new assertions across two classes covering missing local, missing remote, both missing, empty/whitespace tips, mismatch, resolver exception at the assessment site, and resolver exception through the real verify_preflight_purity path; plus predicate rejection of stripped tips, a forged base_tips_verified flag, and a missing flag. All 16 fail against the sources atadc61255and pass after this change. The valid non-create_issue lock_issue test is retained unchanged. Validation: focused #757 suite 51 passed (31 subtests); affected guard and preflight suites 217 passed (67 subtests); full suite 3637 passed, 2 failed, 6 skipped, 431 subtests. Both failures (test_issue_702 F1 worktree recovery; test_reconciler_supersession_close org/repo forwarding) are the documented pre-existing baseline failures onbde5c5fband are not caused by this change. Refs #757 Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]> Claude-Session: https://claude.ai/code/session_015KRvvrFtaM5FEvQ6LvDJMH
This commit is contained in:
@@ -52,6 +52,18 @@ def is_create_issue_task(task: str | None) -> bool:
|
|||||||
return (task or "").strip() in CREATE_ISSUE_TASKS
|
return (task or "").strip() in CREATE_ISSUE_TASKS
|
||||||
|
|
||||||
|
|
||||||
|
def normalize_sha(value: str | None) -> str | None:
|
||||||
|
"""Normalize a Git object id for comparison, or ``None`` when unknown.
|
||||||
|
|
||||||
|
Whitespace and case are the only permitted variation between two spellings
|
||||||
|
of the same commit; anything else is a different commit. Empty and
|
||||||
|
whitespace-only values normalize to ``None`` so an unknown tip can never
|
||||||
|
compare equal to another unknown tip.
|
||||||
|
"""
|
||||||
|
normalized = (value or "").strip().lower()
|
||||||
|
return normalized or None
|
||||||
|
|
||||||
|
|
||||||
def assess_create_issue_bootstrap(
|
def assess_create_issue_bootstrap(
|
||||||
*,
|
*,
|
||||||
workspace_path: str,
|
workspace_path: str,
|
||||||
@@ -60,6 +72,7 @@ def assess_create_issue_bootstrap(
|
|||||||
head_sha: str | None = None,
|
head_sha: str | None = None,
|
||||||
porcelain_status: str = "",
|
porcelain_status: str = "",
|
||||||
remote_master_sha: str | None = None,
|
remote_master_sha: 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 create_issue may proceed from the control checkout.
|
"""Assess whether create_issue may proceed from the control checkout.
|
||||||
@@ -142,8 +155,32 @@ def assess_create_issue_bootstrap(
|
|||||||
f"is not an accepted base branch ({'/'.join(sorted(BASE_BRANCHES))})"
|
f"is not an accepted base branch ({'/'.join(sorted(BASE_BRANCHES))})"
|
||||||
)
|
)
|
||||||
|
|
||||||
remote_tip = (remote_master_sha or "").strip() or None
|
# #757 AC3/AC4: base-equivalence must be *proven*, never assumed. An
|
||||||
local_tip = (head_sha or "").strip() or None
|
# unknown tip on either side is not evidence of agreement, so a missing
|
||||||
|
# local HEAD, an unresolvable live master, or a resolver failure all block.
|
||||||
|
remote_tip = normalize_sha(remote_master_sha)
|
||||||
|
local_tip = normalize_sha(head_sha)
|
||||||
|
resolver_error = (remote_master_sha_error or "").strip() or None
|
||||||
|
|
||||||
|
if not local_tip:
|
||||||
|
reasons.append(
|
||||||
|
"create_issue bootstrap blocked: control checkout HEAD SHA is "
|
||||||
|
"unknown; base equivalence to live master cannot be proven "
|
||||||
|
"(fail closed)"
|
||||||
|
)
|
||||||
|
|
||||||
|
if resolver_error:
|
||||||
|
reasons.append(
|
||||||
|
"create_issue bootstrap blocked: live master tip could not be "
|
||||||
|
f"resolved ({resolver_error}); base equivalence cannot be proven "
|
||||||
|
"(fail closed)"
|
||||||
|
)
|
||||||
|
elif not remote_tip:
|
||||||
|
reasons.append(
|
||||||
|
"create_issue bootstrap blocked: live master tip is unknown; base "
|
||||||
|
"equivalence to live master cannot be proven (fail closed)"
|
||||||
|
)
|
||||||
|
|
||||||
if remote_tip and local_tip and remote_tip != local_tip:
|
if remote_tip and local_tip and remote_tip != local_tip:
|
||||||
reasons.append(
|
reasons.append(
|
||||||
"create_issue bootstrap blocked: control checkout HEAD does not match "
|
"create_issue bootstrap blocked: control checkout HEAD does not match "
|
||||||
@@ -162,6 +199,8 @@ def assess_create_issue_bootstrap(
|
|||||||
dirty=dirty,
|
dirty=dirty,
|
||||||
under_branches=False,
|
under_branches=False,
|
||||||
exact_next_action=EXACT_NEXT_ACTION_BOOTSTRAP,
|
exact_next_action=EXACT_NEXT_ACTION_BOOTSTRAP,
|
||||||
|
local_head_sha=local_tip,
|
||||||
|
remote_master_sha=remote_tip,
|
||||||
)
|
)
|
||||||
|
|
||||||
return _result(
|
return _result(
|
||||||
@@ -176,6 +215,8 @@ def assess_create_issue_bootstrap(
|
|||||||
under_branches=False,
|
under_branches=False,
|
||||||
exact_next_action=EXACT_NEXT_ACTION_POST_CREATE,
|
exact_next_action=EXACT_NEXT_ACTION_POST_CREATE,
|
||||||
bootstrap_path="clean_canonical_control_checkout",
|
bootstrap_path="clean_canonical_control_checkout",
|
||||||
|
local_head_sha=local_tip,
|
||||||
|
remote_master_sha=remote_tip,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
@@ -235,6 +276,17 @@ def bootstrap_permits_control_checkout(
|
|||||||
if assessment.get("under_branches") is not False:
|
if assessment.get("under_branches") is not False:
|
||||||
return False
|
return False
|
||||||
|
|
||||||
|
# Base-equivalence proof (#757 AC3/AC4): both tips must be recorded,
|
||||||
|
# nonempty, and equal. Re-derived here rather than trusted from the
|
||||||
|
# assessment's own flag, so a hand-built or truncated assessment cannot
|
||||||
|
# assert agreement it never proved.
|
||||||
|
if assessment.get("base_tips_verified") is not True:
|
||||||
|
return False
|
||||||
|
local_tip = normalize_sha(assessment.get("local_head_sha"))
|
||||||
|
remote_tip = normalize_sha(assessment.get("remote_master_sha"))
|
||||||
|
if not local_tip or not remote_tip or local_tip != remote_tip:
|
||||||
|
return False
|
||||||
|
|
||||||
# Binding proof: the assessment must describe *this* workspace and root,
|
# Binding proof: the assessment must describe *this* workspace and root,
|
||||||
# and that workspace must be exactly the canonical control checkout.
|
# and that workspace must be exactly the canonical control checkout.
|
||||||
root = os.path.realpath(canonical_repo_root or "")
|
root = os.path.realpath(canonical_repo_root or "")
|
||||||
@@ -279,7 +331,14 @@ def _result(
|
|||||||
under_branches: bool,
|
under_branches: bool,
|
||||||
exact_next_action: str | None = None,
|
exact_next_action: str | None = None,
|
||||||
bootstrap_path: str | None = None,
|
bootstrap_path: str | None = None,
|
||||||
|
local_head_sha: str | None = None,
|
||||||
|
remote_master_sha: str | None = None,
|
||||||
) -> dict[str, Any]:
|
) -> dict[str, Any]:
|
||||||
|
# #757 AC3/AC4: equality is recorded only when BOTH tips are known, so a
|
||||||
|
# consumer can never read agreement out of two missing values.
|
||||||
|
base_tips_verified = bool(
|
||||||
|
local_head_sha and remote_master_sha and local_head_sha == remote_master_sha
|
||||||
|
)
|
||||||
return {
|
return {
|
||||||
"not_applicable": not_applicable,
|
"not_applicable": not_applicable,
|
||||||
"allowed": allowed,
|
"allowed": allowed,
|
||||||
@@ -294,4 +353,7 @@ def _result(
|
|||||||
"exact_next_action": exact_next_action,
|
"exact_next_action": exact_next_action,
|
||||||
"bootstrap_path": bootstrap_path,
|
"bootstrap_path": bootstrap_path,
|
||||||
"task_scope": "create_issue_only",
|
"task_scope": "create_issue_only",
|
||||||
|
"local_head_sha": local_head_sha,
|
||||||
|
"remote_master_sha": remote_master_sha,
|
||||||
|
"base_tips_verified": base_tips_verified,
|
||||||
}
|
}
|
||||||
|
|||||||
+7
-1
@@ -845,12 +845,17 @@ def _create_issue_bootstrap_assessment(
|
|||||||
ctx = _resolve_namespace_mutation_context(worktree_path)
|
ctx = _resolve_namespace_mutation_context(worktree_path)
|
||||||
workspace = ctx["workspace_path"]
|
workspace = ctx["workspace_path"]
|
||||||
git_state = issue_lock_worktree.read_worktree_git_state(workspace)
|
git_state = issue_lock_worktree.read_worktree_git_state(workspace)
|
||||||
|
# #757 AC3/AC4: a resolver failure is not "no constraint" — it is missing
|
||||||
|
# evidence, and must reach the assessor as such so the bootstrap fails
|
||||||
|
# closed instead of proceeding without base-equivalence proof.
|
||||||
|
remote_master_sha_error: str | None = None
|
||||||
try:
|
try:
|
||||||
remote_master_sha = root_checkout_guard.resolve_remote_master_sha(
|
remote_master_sha = root_checkout_guard.resolve_remote_master_sha(
|
||||||
ctx["canonical_repo_root"]
|
ctx["canonical_repo_root"]
|
||||||
)
|
)
|
||||||
except Exception:
|
except Exception as exc:
|
||||||
remote_master_sha = None
|
remote_master_sha = None
|
||||||
|
remote_master_sha_error = f"{type(exc).__name__}: {exc}".strip() or "resolver failed"
|
||||||
return _cib.assess_create_issue_bootstrap(
|
return _cib.assess_create_issue_bootstrap(
|
||||||
workspace_path=workspace,
|
workspace_path=workspace,
|
||||||
canonical_repo_root=ctx["canonical_repo_root"],
|
canonical_repo_root=ctx["canonical_repo_root"],
|
||||||
@@ -858,6 +863,7 @@ def _create_issue_bootstrap_assessment(
|
|||||||
head_sha=git_state.get("head_sha"),
|
head_sha=git_state.get("head_sha"),
|
||||||
porcelain_status=git_state.get("porcelain_status") or "",
|
porcelain_status=git_state.get("porcelain_status") or "",
|
||||||
remote_master_sha=remote_master_sha,
|
remote_master_sha=remote_master_sha,
|
||||||
|
remote_master_sha_error=remote_master_sha_error,
|
||||||
task=task,
|
task=task,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|||||||
@@ -609,5 +609,201 @@ class TestNativeCreateIssueEndToEnd(unittest.TestCase):
|
|||||||
self.assertIn("control checkout", str(ctx.exception))
|
self.assertIn("control checkout", str(ctx.exception))
|
||||||
|
|
||||||
|
|
||||||
|
class TestBaseEquivalenceProofRequired(unittest.TestCase):
|
||||||
|
"""#757 AC3/AC4: an unknown tip is not evidence of agreement.
|
||||||
|
|
||||||
|
The original fix compared SHAs only inside
|
||||||
|
``if remote_tip and local_tip and remote_tip != local_tip``, so a missing
|
||||||
|
local HEAD, an unresolvable live master, or a resolver failure all fell
|
||||||
|
through and *granted* the bootstrap exemption. Base equivalence must be
|
||||||
|
proven, and anything less must fail closed.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def _permits(self, assessment):
|
||||||
|
return cib.bootstrap_permits_control_checkout(
|
||||||
|
assessment,
|
||||||
|
task="create_issue",
|
||||||
|
workspace_path=CONTROL_CHECKOUT_ROOT,
|
||||||
|
canonical_repo_root=CONTROL_CHECKOUT_ROOT,
|
||||||
|
)
|
||||||
|
|
||||||
|
def _assert_fails_closed(self, assessment, *, expect_reason):
|
||||||
|
self.assertTrue(assessment["block"], "assessor must block")
|
||||||
|
self.assertFalse(assessment["allowed"])
|
||||||
|
self.assertFalse(assessment["proven"])
|
||||||
|
self.assertFalse(assessment["base_tips_verified"])
|
||||||
|
self.assertFalse(self._permits(assessment), "predicate must refuse")
|
||||||
|
joined = " ".join(assessment["reasons"]).lower()
|
||||||
|
self.assertIn(expect_reason, joined)
|
||||||
|
|
||||||
|
def test_missing_local_head_fails_closed(self):
|
||||||
|
self._assert_fails_closed(
|
||||||
|
proven_bootstrap(head=None),
|
||||||
|
expect_reason="control checkout head sha is unknown",
|
||||||
|
)
|
||||||
|
|
||||||
|
def test_missing_remote_master_fails_closed(self):
|
||||||
|
self._assert_fails_closed(
|
||||||
|
proven_bootstrap(remote_sha=None),
|
||||||
|
expect_reason="live master tip is unknown",
|
||||||
|
)
|
||||||
|
|
||||||
|
def test_both_tips_missing_fails_closed(self):
|
||||||
|
assessment = proven_bootstrap(head=None, remote_sha=None)
|
||||||
|
self._assert_fails_closed(
|
||||||
|
assessment, expect_reason="control checkout head sha is unknown"
|
||||||
|
)
|
||||||
|
self.assertIn("live master tip is unknown", " ".join(assessment["reasons"]))
|
||||||
|
|
||||||
|
def test_empty_and_whitespace_tips_fail_closed(self):
|
||||||
|
for head, remote in (("", MASTER_SHA), (MASTER_SHA, ""), (" ", " ")):
|
||||||
|
with self.subTest(head=repr(head), remote=repr(remote)):
|
||||||
|
assessment = proven_bootstrap(head=head, remote_sha=remote)
|
||||||
|
self.assertTrue(assessment["block"])
|
||||||
|
self.assertFalse(self._permits(assessment))
|
||||||
|
|
||||||
|
def test_mismatched_tips_fail_closed(self):
|
||||||
|
self._assert_fails_closed(
|
||||||
|
proven_bootstrap(head=STALE_SHA),
|
||||||
|
expect_reason="does not match",
|
||||||
|
)
|
||||||
|
|
||||||
|
def test_resolver_failure_fails_closed(self):
|
||||||
|
assessment = cib.assess_create_issue_bootstrap(
|
||||||
|
workspace_path=CONTROL_CHECKOUT_ROOT,
|
||||||
|
canonical_repo_root=CONTROL_CHECKOUT_ROOT,
|
||||||
|
current_branch="master",
|
||||||
|
head_sha=MASTER_SHA,
|
||||||
|
porcelain_status="",
|
||||||
|
remote_master_sha=None,
|
||||||
|
remote_master_sha_error="TimeoutError: remote unreachable",
|
||||||
|
task="create_issue",
|
||||||
|
)
|
||||||
|
self._assert_fails_closed(
|
||||||
|
assessment, expect_reason="could not be resolved"
|
||||||
|
)
|
||||||
|
# The operator must be able to see *why*, not just that it blocked.
|
||||||
|
self.assertIn("remote unreachable", " ".join(assessment["reasons"]))
|
||||||
|
|
||||||
|
def test_proven_bootstrap_records_both_normalized_shas(self):
|
||||||
|
assessment = proven_bootstrap()
|
||||||
|
self.assertEqual(assessment["local_head_sha"], MASTER_SHA)
|
||||||
|
self.assertEqual(assessment["remote_master_sha"], MASTER_SHA)
|
||||||
|
self.assertTrue(assessment["base_tips_verified"])
|
||||||
|
self.assertTrue(self._permits(assessment))
|
||||||
|
|
||||||
|
def test_tips_are_normalized_before_comparison(self):
|
||||||
|
"""Case and surrounding whitespace are not a different commit."""
|
||||||
|
assessment = proven_bootstrap(
|
||||||
|
head=f" {MASTER_SHA.upper()} ", remote_sha=MASTER_SHA
|
||||||
|
)
|
||||||
|
self.assertTrue(assessment["allowed"])
|
||||||
|
self.assertEqual(assessment["local_head_sha"], MASTER_SHA)
|
||||||
|
self.assertTrue(self._permits(assessment))
|
||||||
|
|
||||||
|
def test_predicate_rejects_assessment_with_tips_stripped(self):
|
||||||
|
"""A recorded proof that is later removed cannot still permit."""
|
||||||
|
for field in ("local_head_sha", "remote_master_sha"):
|
||||||
|
with self.subTest(field=field):
|
||||||
|
assessment = dict(proven_bootstrap())
|
||||||
|
assessment[field] = None
|
||||||
|
self.assertFalse(self._permits(assessment))
|
||||||
|
|
||||||
|
def test_predicate_rejects_forged_verified_flag(self):
|
||||||
|
"""base_tips_verified is re-derived, never trusted on its own."""
|
||||||
|
assessment = dict(proven_bootstrap())
|
||||||
|
assessment["local_head_sha"] = MASTER_SHA
|
||||||
|
assessment["remote_master_sha"] = STALE_SHA
|
||||||
|
assessment["base_tips_verified"] = True
|
||||||
|
self.assertFalse(self._permits(assessment))
|
||||||
|
|
||||||
|
def test_predicate_rejects_missing_verified_flag(self):
|
||||||
|
assessment = dict(proven_bootstrap())
|
||||||
|
assessment.pop("base_tips_verified")
|
||||||
|
self.assertFalse(self._permits(assessment))
|
||||||
|
|
||||||
|
|
||||||
|
class TestServerAssessmentFailsClosedOnResolverError(unittest.TestCase):
|
||||||
|
"""AC3/AC4 at the single server-derived computation site."""
|
||||||
|
|
||||||
|
def setUp(self):
|
||||||
|
# verify_preflight_purity short-circuits under pytest; the production
|
||||||
|
# path only runs with test mode disabled (same setup the #757 e2e uses).
|
||||||
|
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._orig_in_test = srv._preflight_in_test_mode
|
||||||
|
srv._preflight_in_test_mode = lambda: False
|
||||||
|
|
||||||
|
def tearDown(self):
|
||||||
|
srv._preflight_in_test_mode = self._orig_in_test
|
||||||
|
srv._preflight_resolved_task = None
|
||||||
|
|
||||||
|
def _git_state(self):
|
||||||
|
return {
|
||||||
|
"current_branch": "master",
|
||||||
|
"head_sha": MASTER_SHA,
|
||||||
|
"porcelain_status": "",
|
||||||
|
}
|
||||||
|
|
||||||
|
def test_resolver_exception_produces_blocking_assessment(self):
|
||||||
|
"""A raising resolver must not become a silent, permissive None."""
|
||||||
|
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT), \
|
||||||
|
patch.object(
|
||||||
|
srv.issue_lock_worktree,
|
||||||
|
"read_worktree_git_state",
|
||||||
|
return_value=self._git_state(),
|
||||||
|
), \
|
||||||
|
patch.object(
|
||||||
|
srv.root_checkout_guard,
|
||||||
|
"resolve_remote_master_sha",
|
||||||
|
side_effect=TimeoutError("remote unreachable"),
|
||||||
|
):
|
||||||
|
assessment = srv._create_issue_bootstrap_assessment("create_issue")
|
||||||
|
|
||||||
|
self.assertIsNotNone(assessment)
|
||||||
|
self.assertTrue(assessment["block"])
|
||||||
|
self.assertFalse(assessment["allowed"])
|
||||||
|
self.assertFalse(assessment["base_tips_verified"])
|
||||||
|
self.assertIsNone(assessment["remote_master_sha"])
|
||||||
|
self.assertIn("could not be resolved", " ".join(assessment["reasons"]))
|
||||||
|
self.assertFalse(
|
||||||
|
cib.bootstrap_permits_control_checkout(
|
||||||
|
assessment,
|
||||||
|
task="create_issue",
|
||||||
|
workspace_path=CONTROL_CHECKOUT_ROOT,
|
||||||
|
canonical_repo_root=CONTROL_CHECKOUT_ROOT,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
|
def test_resolver_exception_blocks_real_preflight(self):
|
||||||
|
"""Production path: verify_preflight_purity must fail closed."""
|
||||||
|
srv._preflight_resolved_task = "create_issue"
|
||||||
|
try:
|
||||||
|
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT), \
|
||||||
|
patch.object(srv, "_actual_profile_role", return_value="author"), \
|
||||||
|
patch.object(
|
||||||
|
srv, "_effective_workspace_role", return_value="author"
|
||||||
|
), \
|
||||||
|
patch.object(
|
||||||
|
srv.issue_lock_worktree,
|
||||||
|
"read_worktree_git_state",
|
||||||
|
return_value=self._git_state(),
|
||||||
|
), \
|
||||||
|
patch.object(srv, "_get_workspace_porcelain", return_value=""), \
|
||||||
|
patch.object(
|
||||||
|
srv.root_checkout_guard,
|
||||||
|
"resolve_remote_master_sha",
|
||||||
|
side_effect=TimeoutError("remote unreachable"),
|
||||||
|
), \
|
||||||
|
patch.object(srv, "_enforce_root_checkout_guard"):
|
||||||
|
with self.assertRaises(RuntimeError):
|
||||||
|
srv.verify_preflight_purity(remote="prgs", task="create_issue")
|
||||||
|
finally:
|
||||||
|
srv._preflight_resolved_task = None
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
unittest.main()
|
unittest.main()
|
||||||
|
|||||||
Reference in New Issue
Block a user