Compare commits
5
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
aab54d4825 | ||
|
|
a09c485fc0 | ||
|
|
8c1d22a658 | ||
|
|
6a56260768 | ||
|
|
2066623986 |
+155
-44
@@ -386,6 +386,68 @@ 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,
|
||||||
@@ -397,7 +459,13 @@ 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()
|
||||||
@@ -407,34 +475,50 @@ 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 {
|
return _author_bootstrap_assessment(
|
||||||
"not_applicable": True,
|
not_applicable=True,
|
||||||
"allowed": False,
|
allowed=False,
|
||||||
"block": False,
|
block=False,
|
||||||
"proven": False,
|
reasons=["task is not author_issue_bootstrap"],
|
||||||
"reasons": ["task is not author_issue_bootstrap"],
|
workspace=workspace,
|
||||||
}
|
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 {
|
return _author_bootstrap_assessment(
|
||||||
"not_applicable": False,
|
not_applicable=True,
|
||||||
"allowed": True,
|
allowed=False,
|
||||||
"block": False,
|
block=False,
|
||||||
"proven": True,
|
reasons=["workspace is under branches/; ordinary #274 path applies"],
|
||||||
"bootstrap_path": "existing_branches_worktree",
|
workspace=workspace,
|
||||||
"reasons": [
|
root=root,
|
||||||
"workspace is already a registered worktree under branches/"
|
branch=branch or None,
|
||||||
],
|
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 workspace != root:
|
if not root or 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 branch not in author_mutation_worktree.BASE_BRANCHES:
|
if not branch:
|
||||||
|
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))})"
|
||||||
@@ -444,37 +528,64 @@ 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])}"
|
||||||
)
|
)
|
||||||
|
|
||||||
if remote_master_sha_error:
|
# Fail closed on missing tip proof (same bar as create_issue bootstrap #757).
|
||||||
|
if not local_tip:
|
||||||
reasons.append(
|
reasons.append(
|
||||||
f"could not verify live master tip: {remote_master_sha_error}"
|
"control checkout HEAD SHA is unknown; base equivalence to live "
|
||||||
|
"master cannot be proven (fail closed)"
|
||||||
)
|
)
|
||||||
elif remote_master_sha and head_sha:
|
resolver_error = (remote_master_sha_error or "").strip() or None
|
||||||
h = head_sha.strip().lower()
|
if resolver_error:
|
||||||
rm = remote_master_sha.strip().lower()
|
|
||||||
if h != rm:
|
|
||||||
reasons.append(
|
reasons.append(
|
||||||
f"control checkout HEAD ({h[:12]}) != live master tip ({rm[:12]})"
|
f"live master tip could not be resolved ({resolver_error}); "
|
||||||
|
"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 {
|
return _author_bootstrap_assessment(
|
||||||
"not_applicable": False,
|
not_applicable=False,
|
||||||
"allowed": False,
|
allowed=False,
|
||||||
"block": True,
|
block=True,
|
||||||
"proven": False,
|
reasons=reasons,
|
||||||
"reasons": reasons,
|
workspace=workspace,
|
||||||
}
|
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,
|
||||||
|
)
|
||||||
|
|
||||||
return {
|
# Allowed: empty reasons so bootstrap_permits_control_checkout can pass.
|
||||||
"not_applicable": False,
|
return _author_bootstrap_assessment(
|
||||||
"allowed": True,
|
not_applicable=False,
|
||||||
"block": False,
|
allowed=True,
|
||||||
"proven": True,
|
block=False,
|
||||||
"bootstrap_path": "clean_canonical_control_checkout",
|
reasons=[],
|
||||||
"reasons": [
|
workspace=workspace,
|
||||||
"control checkout is clean on accepted base branch matching live master"
|
root=root,
|
||||||
],
|
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
|
||||||
|
|||||||
@@ -241,9 +241,14 @@ 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` from inspected repository state. It is
|
:func:`assess_create_issue_bootstrap` or
|
||||||
never accepted from an MCP tool argument, so no caller can assert
|
:func:`author_issue_bootstrap.assess_author_issue_bootstrap` from inspected
|
||||||
eligibility it has not proven.
|
repository state. It is never accepted from an MCP tool argument, so no
|
||||||
|
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
|
||||||
@@ -264,9 +269,16 @@ def bootstrap_permits_control_checkout(
|
|||||||
if assessment.get("reasons"):
|
if assessment.get("reasons"):
|
||||||
return False
|
return False
|
||||||
|
|
||||||
# Scope proof: only the create_issue bootstrap, only via the clean
|
# Scope proof: create_issue (#749) or author issue bootstrap (#850/#892),
|
||||||
# canonical control checkout path.
|
# only via the clean canonical control checkout path.
|
||||||
if assessment.get("task_scope") != "create_issue_only":
|
task_scope = assessment.get("task_scope")
|
||||||
|
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
|
||||||
|
|||||||
+32
-1
@@ -1546,6 +1546,7 @@ def verify_preflight_purity(
|
|||||||
task=task,
|
task=task,
|
||||||
target_issue_number=target_issue_number,
|
target_issue_number=target_issue_number,
|
||||||
require_author_lock=require_author_lock,
|
require_author_lock=require_author_lock,
|
||||||
|
bootstrap_assessment=bootstrap_assessment,
|
||||||
)
|
)
|
||||||
# #604: common anti-stomp preflight after legacy + #683 enforcers.
|
# #604: common anti-stomp preflight after legacy + #683 enforcers.
|
||||||
_run_anti_stomp_preflight(
|
_run_anti_stomp_preflight(
|
||||||
@@ -1585,6 +1586,7 @@ def verify_preflight_purity(
|
|||||||
task=task,
|
task=task,
|
||||||
target_issue_number=target_issue_number,
|
target_issue_number=target_issue_number,
|
||||||
require_author_lock=require_author_lock,
|
require_author_lock=require_author_lock,
|
||||||
|
bootstrap_assessment=bootstrap_assessment,
|
||||||
)
|
)
|
||||||
if force_anti_stomp:
|
if force_anti_stomp:
|
||||||
_run_anti_stomp_preflight(
|
_run_anti_stomp_preflight(
|
||||||
@@ -1652,8 +1654,15 @@ def _enforce_issue_scope_guard(
|
|||||||
task: str | None = None,
|
task: str | None = None,
|
||||||
target_issue_number: int | None = None,
|
target_issue_number: int | None = None,
|
||||||
require_author_lock: bool = False,
|
require_author_lock: bool = False,
|
||||||
|
bootstrap_assessment: object = _BOOTSTRAP_UNSET,
|
||||||
) -> None:
|
) -> None:
|
||||||
"""#683: fail closed on missing/out-of-scope issue ownership for mutations."""
|
"""#683: fail closed on missing/out-of-scope issue ownership for mutations.
|
||||||
|
|
||||||
|
#941: the shared bootstrap assessment is threaded in so this guard judges
|
||||||
|
the author issue-worktree bootstrap on the same server-derived evidence as
|
||||||
|
the #274 branches-only and #604 anti-stomp guards. Callers that supply
|
||||||
|
none fall back to computing it here, which preserves behaviour.
|
||||||
|
"""
|
||||||
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)
|
||||||
@@ -1703,11 +1712,32 @@ def _enforce_issue_scope_guard(
|
|||||||
import create_issue_bootstrap as _cib
|
import create_issue_bootstrap as _cib
|
||||||
|
|
||||||
is_create_issue = _cib.is_create_issue_task(task)
|
is_create_issue = _cib.is_create_issue_task(task)
|
||||||
|
# #941: consume the caller-computed bootstrap assessment when one was
|
||||||
|
# threaded in, so this guard and the #274/#604 guards judge identical
|
||||||
|
# evidence. Falling back preserves behaviour for callers that supply none.
|
||||||
|
bootstrap = (
|
||||||
|
_create_issue_bootstrap_assessment(task, worktree_path)
|
||||||
|
if bootstrap_assessment is _BOOTSTRAP_UNSET
|
||||||
|
else bootstrap_assessment
|
||||||
|
)
|
||||||
|
# #941: the author issue-worktree bootstrap is pre-ownership for the same
|
||||||
|
# reason create_issue is — it exists to break the lock<->worktree cycle, so
|
||||||
|
# no owning lock can exist yet. The exemption is granted by the canonical
|
||||||
|
# shared decision over server-derived evidence, never by a task-name list,
|
||||||
|
# and fails closed on missing, malformed, cross-scope, dirty, drifted, or
|
||||||
|
# wrongly bound evidence.
|
||||||
|
bootstrap_waives_ownership = _cib.bootstrap_permits_control_checkout(
|
||||||
|
bootstrap,
|
||||||
|
task=task,
|
||||||
|
workspace_path=workspace,
|
||||||
|
canonical_repo_root=ctx["canonical_repo_root"],
|
||||||
|
)
|
||||||
require_lock = bool(require_author_lock) or (
|
require_lock = bool(require_author_lock) or (
|
||||||
authorish
|
authorish
|
||||||
and workflow_scope_guard.production_guards_forced()
|
and workflow_scope_guard.production_guards_forced()
|
||||||
and role == "author"
|
and role == "author"
|
||||||
and not is_create_issue
|
and not is_create_issue
|
||||||
|
and not bootstrap_waives_ownership
|
||||||
)
|
)
|
||||||
assessment = workflow_scope_guard.assess_production_mutation_guards(
|
assessment = workflow_scope_guard.assess_production_mutation_guards(
|
||||||
workspace_path=workspace,
|
workspace_path=workspace,
|
||||||
@@ -1720,6 +1750,7 @@ def _enforce_issue_scope_guard(
|
|||||||
require_author_lock=require_lock,
|
require_author_lock=require_lock,
|
||||||
in_test_mode=_preflight_in_test_mode(),
|
in_test_mode=_preflight_in_test_mode(),
|
||||||
mutation_task=task,
|
mutation_task=task,
|
||||||
|
bootstrap_assessment=bootstrap,
|
||||||
)
|
)
|
||||||
workflow_scope_guard.raise_if_blocked(assessment)
|
workflow_scope_guard.raise_if_blocked(assessment)
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,215 @@
|
|||||||
|
"""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()
|
||||||
@@ -0,0 +1,346 @@
|
|||||||
|
"""Regression: author bootstrap scope reaches workflow_scope_guard (#941).
|
||||||
|
|
||||||
|
PR #926 (#892) made ``bootstrap_permits_control_checkout`` accept
|
||||||
|
``task_scope=author_issue_bootstrap`` and wired that canonical decision into
|
||||||
|
the #274 branches-only enforcer and the #604 anti-stomp preflight. A third
|
||||||
|
enforcement path was left unwired.
|
||||||
|
|
||||||
|
``workflow_scope_guard.assess_root_source_mutation`` kept its own copy of the
|
||||||
|
clean-root author decision, gated on ``create_issue_bootstrap.is_create_issue_task``
|
||||||
|
— a task-name allowlist that never contained ``bootstrap_author_issue_worktree``.
|
||||||
|
So the real call path
|
||||||
|
|
||||||
|
gitea_bootstrap_author_issue_worktree
|
||||||
|
-> verify_preflight_purity
|
||||||
|
-> _enforce_issue_scope_guard
|
||||||
|
-> workflow_scope_guard.assess_production_mutation_guards
|
||||||
|
|
||||||
|
raised ProductionGuardError(missing_issue_worktree) before
|
||||||
|
``assess_author_issue_bootstrap`` was ever consulted.
|
||||||
|
|
||||||
|
These tests drive the real enforcer, not the authorization helper in
|
||||||
|
isolation. A helper-only test cannot observe this defect: #892's own predicate
|
||||||
|
tests all passed while the live bootstrap stayed blocked.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import os
|
||||||
|
import subprocess
|
||||||
|
import tempfile
|
||||||
|
import unittest
|
||||||
|
from unittest import mock
|
||||||
|
|
||||||
|
import author_issue_bootstrap as aib
|
||||||
|
import create_issue_bootstrap as cib
|
||||||
|
import workflow_scope_guard
|
||||||
|
|
||||||
|
BOOTSTRAP_TASK = "bootstrap_author_issue_worktree"
|
||||||
|
BOOTSTRAP_TOOL = "gitea_bootstrap_author_issue_worktree"
|
||||||
|
|
||||||
|
|
||||||
|
def _make_control_repo(tmp: str) -> tuple[str, str]:
|
||||||
|
"""Create a clean control checkout on master and return (path, head)."""
|
||||||
|
repo = os.path.join(tmp, "repo")
|
||||||
|
os.makedirs(os.path.join(repo, "branches"))
|
||||||
|
subprocess.check_call(
|
||||||
|
["git", "init", "-b", "master", repo],
|
||||||
|
stdout=subprocess.DEVNULL,
|
||||||
|
stderr=subprocess.DEVNULL,
|
||||||
|
)
|
||||||
|
subprocess.check_call(
|
||||||
|
[
|
||||||
|
"git", "-C", repo,
|
||||||
|
"-c", "user.email=t@t", "-c", "user.name=t",
|
||||||
|
"commit", "--allow-empty", "-m", "init",
|
||||||
|
],
|
||||||
|
stdout=subprocess.DEVNULL,
|
||||||
|
stderr=subprocess.DEVNULL,
|
||||||
|
)
|
||||||
|
head = subprocess.check_output(
|
||||||
|
["git", "-C", repo, "rev-parse", "HEAD"], text=True
|
||||||
|
).strip()
|
||||||
|
return repo, head
|
||||||
|
|
||||||
|
|
||||||
|
def _assessment(
|
||||||
|
repo: str,
|
||||||
|
head: str,
|
||||||
|
*,
|
||||||
|
task: str = BOOTSTRAP_TASK,
|
||||||
|
porcelain: str = "",
|
||||||
|
remote: str | None = None,
|
||||||
|
) -> dict:
|
||||||
|
return aib.assess_author_issue_bootstrap(
|
||||||
|
workspace_path=repo,
|
||||||
|
canonical_repo_root=repo,
|
||||||
|
current_branch="master",
|
||||||
|
head_sha=head,
|
||||||
|
porcelain_status=porcelain,
|
||||||
|
remote_master_sha=head if remote is None else remote,
|
||||||
|
task=task,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class _ControlCheckoutHarness(unittest.TestCase):
|
||||||
|
"""Drive the real server guard against a temporary clean control checkout."""
|
||||||
|
|
||||||
|
def setUp(self):
|
||||||
|
self._tmp = tempfile.TemporaryDirectory()
|
||||||
|
self.addCleanup(self._tmp.cleanup)
|
||||||
|
self.repo, self.head = _make_control_repo(self._tmp.name)
|
||||||
|
|
||||||
|
# #683 force-on: production guards must execute under pytest.
|
||||||
|
patcher = mock.patch.dict(
|
||||||
|
os.environ,
|
||||||
|
{workflow_scope_guard.FORCE_PRODUCTION_GUARDS_ENV: "1"},
|
||||||
|
)
|
||||||
|
patcher.start()
|
||||||
|
self.addCleanup(patcher.stop)
|
||||||
|
|
||||||
|
def _enforce(
|
||||||
|
self,
|
||||||
|
task: str,
|
||||||
|
*,
|
||||||
|
porcelain: str = "",
|
||||||
|
assessment: object = "auto",
|
||||||
|
role_kind: str = "author",
|
||||||
|
):
|
||||||
|
"""Call the real _enforce_issue_scope_guard for *task*."""
|
||||||
|
import gitea_mcp_server as srv
|
||||||
|
|
||||||
|
if assessment == "auto":
|
||||||
|
assessment = _assessment(
|
||||||
|
self.repo, self.head, task=task, porcelain=porcelain
|
||||||
|
)
|
||||||
|
|
||||||
|
ctx = {
|
||||||
|
"workspace_path": self.repo,
|
||||||
|
"canonical_repo_root": self.repo,
|
||||||
|
"workspace_role_kind": role_kind,
|
||||||
|
"workspace_binding_source": "process_project_root",
|
||||||
|
}
|
||||||
|
git_state = {
|
||||||
|
"current_branch": "master",
|
||||||
|
"head_sha": self.head,
|
||||||
|
"porcelain_status": porcelain,
|
||||||
|
}
|
||||||
|
|
||||||
|
with mock.patch.object(
|
||||||
|
srv, "_resolve_namespace_mutation_context", return_value=ctx
|
||||||
|
), mock.patch.object(
|
||||||
|
srv.issue_lock_worktree,
|
||||||
|
"read_worktree_git_state",
|
||||||
|
return_value=git_state,
|
||||||
|
), mock.patch.object(
|
||||||
|
srv,
|
||||||
|
"_session_issue_lock_snapshot",
|
||||||
|
return_value={
|
||||||
|
"locked_issue_number": None,
|
||||||
|
"lock_branch_name": None,
|
||||||
|
"worktrees_match": False,
|
||||||
|
},
|
||||||
|
), mock.patch.object(
|
||||||
|
srv, "_actual_profile_role", return_value=role_kind
|
||||||
|
), mock.patch.object(
|
||||||
|
srv, "_effective_workspace_role", return_value=role_kind
|
||||||
|
), mock.patch.object(
|
||||||
|
srv, "_create_issue_bootstrap_assessment", return_value=assessment
|
||||||
|
):
|
||||||
|
srv._enforce_issue_scope_guard(None, task=task)
|
||||||
|
|
||||||
|
|
||||||
|
class TestRealPathBootstrapReachesGuard(_ControlCheckoutHarness):
|
||||||
|
"""The defect and its fix, observed through the real enforcer."""
|
||||||
|
|
||||||
|
def test_bootstrap_task_passes_scope_guard_from_clean_control(self):
|
||||||
|
# Pre-fix this raises ProductionGuardError(missing_issue_worktree)
|
||||||
|
# because the guard consulted a task-name allowlist instead of the
|
||||||
|
# canonical authorization decision.
|
||||||
|
self._enforce(BOOTSTRAP_TASK)
|
||||||
|
|
||||||
|
def test_bootstrap_tool_alias_passes_scope_guard(self):
|
||||||
|
self._enforce(BOOTSTRAP_TOOL)
|
||||||
|
|
||||||
|
def test_guard_consults_canonical_predicate(self):
|
||||||
|
"""The guard must reach bootstrap_permits_control_checkout, not a name list."""
|
||||||
|
real = cib.bootstrap_permits_control_checkout
|
||||||
|
seen: list[str | None] = []
|
||||||
|
|
||||||
|
def _spy(assessment, *, task, workspace_path, canonical_repo_root):
|
||||||
|
seen.append(task)
|
||||||
|
return real(
|
||||||
|
assessment,
|
||||||
|
task=task,
|
||||||
|
workspace_path=workspace_path,
|
||||||
|
canonical_repo_root=canonical_repo_root,
|
||||||
|
)
|
||||||
|
|
||||||
|
with mock.patch.object(
|
||||||
|
cib, "bootstrap_permits_control_checkout", side_effect=_spy
|
||||||
|
):
|
||||||
|
self._enforce(BOOTSTRAP_TASK)
|
||||||
|
|
||||||
|
self.assertIn(
|
||||||
|
BOOTSTRAP_TASK,
|
||||||
|
seen,
|
||||||
|
"workflow_scope_guard did not consult the canonical bootstrap "
|
||||||
|
"authorization decision",
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class TestFailClosedOnBadEvidence(_ControlCheckoutHarness):
|
||||||
|
"""Missing, malformed, or mismatched scope evidence must still block."""
|
||||||
|
|
||||||
|
def _assert_blocked(self, **kwargs):
|
||||||
|
with self.assertRaises(workflow_scope_guard.ProductionGuardError):
|
||||||
|
self._enforce(BOOTSTRAP_TASK, **kwargs)
|
||||||
|
|
||||||
|
def test_missing_assessment_fails_closed(self):
|
||||||
|
self._assert_blocked(assessment=None)
|
||||||
|
|
||||||
|
def test_malformed_assessment_fails_closed(self):
|
||||||
|
self._assert_blocked(assessment={"allowed": True})
|
||||||
|
|
||||||
|
def test_non_dict_assessment_fails_closed(self):
|
||||||
|
self._assert_blocked(assessment="allowed")
|
||||||
|
|
||||||
|
def test_wrong_task_scope_fails_closed(self):
|
||||||
|
bad = dict(_assessment(self.repo, self.head))
|
||||||
|
bad["task_scope"] = "create_issue_only"
|
||||||
|
self._assert_blocked(assessment=bad)
|
||||||
|
|
||||||
|
def test_nonempty_reasons_fail_closed(self):
|
||||||
|
bad = dict(_assessment(self.repo, self.head), reasons=["note"])
|
||||||
|
self._assert_blocked(assessment=bad)
|
||||||
|
|
||||||
|
def test_mismatched_base_tips_fail_closed(self):
|
||||||
|
bad = dict(_assessment(self.repo, self.head))
|
||||||
|
bad["remote_master_sha"] = "b" * 40
|
||||||
|
self._assert_blocked(assessment=bad)
|
||||||
|
|
||||||
|
def test_unverified_base_tips_fail_closed(self):
|
||||||
|
bad = dict(_assessment(self.repo, self.head), base_tips_verified=False)
|
||||||
|
self._assert_blocked(assessment=bad)
|
||||||
|
|
||||||
|
def test_mismatched_workspace_binding_fails_closed(self):
|
||||||
|
bad = dict(_assessment(self.repo, self.head))
|
||||||
|
bad["workspace_path"] = os.path.join(self.repo, "elsewhere")
|
||||||
|
self._assert_blocked(assessment=bad)
|
||||||
|
|
||||||
|
def test_mismatched_repo_root_binding_fails_closed(self):
|
||||||
|
bad = dict(_assessment(self.repo, self.head))
|
||||||
|
bad["canonical_repo_root"] = os.path.join(self.repo, "other-root")
|
||||||
|
self._assert_blocked(assessment=bad)
|
||||||
|
|
||||||
|
def test_blocked_assessment_fails_closed(self):
|
||||||
|
bad = dict(_assessment(self.repo, self.head), block=True, allowed=False)
|
||||||
|
self._assert_blocked(assessment=bad)
|
||||||
|
|
||||||
|
|
||||||
|
class TestOrdinaryControlCheckoutMutationStillForbidden(_ControlCheckoutHarness):
|
||||||
|
"""The waiver must not leak to ordinary author work."""
|
||||||
|
|
||||||
|
def test_ordinary_author_task_still_blocked(self):
|
||||||
|
with self.assertRaises(workflow_scope_guard.ProductionGuardError):
|
||||||
|
self._enforce("commit_files", assessment=None)
|
||||||
|
|
||||||
|
def test_lock_issue_still_blocked_from_control(self):
|
||||||
|
with self.assertRaises(workflow_scope_guard.ProductionGuardError):
|
||||||
|
self._enforce("lock_issue", assessment=None)
|
||||||
|
|
||||||
|
def test_bootstrap_assessment_cannot_license_other_task(self):
|
||||||
|
# Cross-task smuggling: valid bootstrap evidence must not waive a
|
||||||
|
# different author mutation.
|
||||||
|
good = _assessment(self.repo, self.head)
|
||||||
|
with self.assertRaises(workflow_scope_guard.ProductionGuardError):
|
||||||
|
self._enforce("commit_files", assessment=good)
|
||||||
|
|
||||||
|
def test_dirty_control_checkout_still_blocked_for_bootstrap(self):
|
||||||
|
with self.assertRaises(workflow_scope_guard.ProductionGuardError):
|
||||||
|
self._enforce(BOOTSTRAP_TASK, porcelain=" M gitea_mcp_server.py\n")
|
||||||
|
|
||||||
|
|
||||||
|
class TestCreateIssueBehaviorUnchanged(_ControlCheckoutHarness):
|
||||||
|
"""#749 create_issue keeps its own sanctioned path."""
|
||||||
|
|
||||||
|
def test_create_issue_still_allowed_from_clean_control(self):
|
||||||
|
self._enforce("create_issue", assessment=None)
|
||||||
|
|
||||||
|
def test_create_issue_tool_alias_still_allowed(self):
|
||||||
|
self._enforce("gitea_create_issue", assessment=None)
|
||||||
|
|
||||||
|
def test_create_issue_blocked_when_control_dirty(self):
|
||||||
|
with self.assertRaises(workflow_scope_guard.ProductionGuardError):
|
||||||
|
self._enforce(
|
||||||
|
"create_issue",
|
||||||
|
porcelain=" M gitea_mcp_server.py\n",
|
||||||
|
assessment=None,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class TestGuardUnitLevelWiring(unittest.TestCase):
|
||||||
|
"""assess_root_source_mutation itself must accept and honour the evidence."""
|
||||||
|
|
||||||
|
def setUp(self):
|
||||||
|
self._tmp = tempfile.TemporaryDirectory()
|
||||||
|
self.addCleanup(self._tmp.cleanup)
|
||||||
|
self.repo, self.head = _make_control_repo(self._tmp.name)
|
||||||
|
patcher = mock.patch.dict(
|
||||||
|
os.environ,
|
||||||
|
{workflow_scope_guard.FORCE_PRODUCTION_GUARDS_ENV: "1"},
|
||||||
|
)
|
||||||
|
patcher.start()
|
||||||
|
self.addCleanup(patcher.stop)
|
||||||
|
|
||||||
|
def _assess(self, *, task=BOOTSTRAP_TASK, bootstrap_assessment="auto"):
|
||||||
|
if bootstrap_assessment == "auto":
|
||||||
|
bootstrap_assessment = _assessment(self.repo, self.head, task=task)
|
||||||
|
return workflow_scope_guard.assess_root_source_mutation(
|
||||||
|
workspace_path=self.repo,
|
||||||
|
canonical_repo_root=self.repo,
|
||||||
|
porcelain_status="",
|
||||||
|
current_branch="master",
|
||||||
|
role_kind="author",
|
||||||
|
mutation_task=task,
|
||||||
|
bootstrap_assessment=bootstrap_assessment,
|
||||||
|
)
|
||||||
|
|
||||||
|
def test_valid_evidence_unblocks(self):
|
||||||
|
result = self._assess()
|
||||||
|
self.assertFalse(result["block"])
|
||||||
|
self.assertIsNone(result["blocker_kind"])
|
||||||
|
|
||||||
|
def test_absent_evidence_blocks(self):
|
||||||
|
result = self._assess(bootstrap_assessment=None)
|
||||||
|
self.assertTrue(result["block"])
|
||||||
|
self.assertEqual(
|
||||||
|
result["blocker_kind"], workflow_scope_guard.BLOCKER_MISSING_WORKTREE
|
||||||
|
)
|
||||||
|
|
||||||
|
def test_reconciler_exemption_preserved(self):
|
||||||
|
result = workflow_scope_guard.assess_root_source_mutation(
|
||||||
|
workspace_path=self.repo,
|
||||||
|
canonical_repo_root=self.repo,
|
||||||
|
porcelain_status="",
|
||||||
|
current_branch="master",
|
||||||
|
role_kind="reconciler",
|
||||||
|
mutation_task=BOOTSTRAP_TASK,
|
||||||
|
)
|
||||||
|
self.assertFalse(result["block"])
|
||||||
|
|
||||||
|
def test_signature_accepts_evidence_without_it_being_required(self):
|
||||||
|
# Callers that supply no evidence keep the pre-existing behaviour.
|
||||||
|
result = workflow_scope_guard.assess_root_source_mutation(
|
||||||
|
workspace_path=self.repo,
|
||||||
|
canonical_repo_root=self.repo,
|
||||||
|
porcelain_status="",
|
||||||
|
current_branch="master",
|
||||||
|
role_kind="author",
|
||||||
|
mutation_task="create_issue",
|
||||||
|
)
|
||||||
|
self.assertFalse(result["block"])
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
unittest.main()
|
||||||
+30
-1
@@ -279,6 +279,7 @@ def assess_root_source_mutation(
|
|||||||
locked_issue_number: int | None = None,
|
locked_issue_number: int | None = None,
|
||||||
role_kind: str | None = None,
|
role_kind: str | None = None,
|
||||||
mutation_task: str | None = None,
|
mutation_task: str | None = None,
|
||||||
|
bootstrap_assessment: Any | None = None,
|
||||||
) -> dict[str, Any]:
|
) -> dict[str, Any]:
|
||||||
"""Fail closed for diagnostic/source edits on the control/root checkout.
|
"""Fail closed for diagnostic/source edits on the control/root checkout.
|
||||||
|
|
||||||
@@ -286,6 +287,13 @@ def assess_root_source_mutation(
|
|||||||
tracked source/test files on the control checkout always block, including
|
tracked source/test files on the control checkout always block, including
|
||||||
temporary/diagnostic/test-only intent.
|
temporary/diagnostic/test-only intent.
|
||||||
|
|
||||||
|
#941: ``bootstrap_author_issue_worktree`` is judged by the canonical
|
||||||
|
``create_issue_bootstrap.bootstrap_permits_control_checkout`` decision over
|
||||||
|
*bootstrap_assessment* — the same server-derived evidence the #274 and
|
||||||
|
#604 guards consume — instead of a task-name allowlist local to this
|
||||||
|
module. Evidence that is absent, malformed, wrongly scoped, or bound to
|
||||||
|
another workspace leaves the ordinary block in force.
|
||||||
|
|
||||||
#749: ``create_issue`` is a pure remote mutation with no local tree write.
|
#749: ``create_issue`` is a pure remote mutation with no local tree write.
|
||||||
When *mutation_task* is create_issue and the control checkout has no dirty
|
When *mutation_task* is create_issue and the control checkout has no dirty
|
||||||
source/test files, the missing-worktree signal is suppressed so the
|
source/test files, the missing-worktree signal is suppressed so the
|
||||||
@@ -336,6 +344,19 @@ def assess_root_source_mutation(
|
|||||||
if _cib is not None and _cib.is_create_issue_task(mutation_task):
|
if _cib is not None and _cib.is_create_issue_task(mutation_task):
|
||||||
# #749: clean-root create_issue is the sanctioned bootstrap path.
|
# #749: clean-root create_issue is the sanctioned bootstrap path.
|
||||||
create_issue_bootstrap = True
|
create_issue_bootstrap = True
|
||||||
|
elif _cib is not None and _cib.bootstrap_permits_control_checkout(
|
||||||
|
bootstrap_assessment,
|
||||||
|
task=mutation_task,
|
||||||
|
workspace_path=workspace,
|
||||||
|
canonical_repo_root=root,
|
||||||
|
):
|
||||||
|
# #941: the author issue-worktree bootstrap is authorized by the
|
||||||
|
# canonical shared decision over server-derived task-scope
|
||||||
|
# evidence, never by a task-name allowlist kept in this module.
|
||||||
|
# The predicate fails closed on missing, malformed, cross-scope,
|
||||||
|
# dirty, drifted, or wrongly bound evidence, so this arm cannot
|
||||||
|
# widen the waiver beyond the one sanctioned bootstrap task.
|
||||||
|
create_issue_bootstrap = True
|
||||||
else:
|
else:
|
||||||
# Explicit missing-worktree signal for force-on author entrypoints.
|
# Explicit missing-worktree signal for force-on author entrypoints.
|
||||||
reasons.append(
|
reasons.append(
|
||||||
@@ -393,8 +414,15 @@ def assess_production_mutation_guards(
|
|||||||
require_author_lock: bool = False,
|
require_author_lock: bool = False,
|
||||||
in_test_mode: bool = False,
|
in_test_mode: bool = False,
|
||||||
mutation_task: str | None = None,
|
mutation_task: str | None = None,
|
||||||
|
bootstrap_assessment: Any | None = None,
|
||||||
) -> dict[str, Any]:
|
) -> dict[str, Any]:
|
||||||
"""Compose root + scope production guards when they must be active (#683)."""
|
"""Compose root + scope production guards when they must be active (#683).
|
||||||
|
|
||||||
|
#941: *bootstrap_assessment* is the server-derived author-bootstrap
|
||||||
|
evidence, forwarded unchanged to :func:`assess_root_source_mutation` so
|
||||||
|
this guard reaches the same canonical decision as the #274 and #604
|
||||||
|
guards. Omitting it preserves the pre-existing behaviour.
|
||||||
|
"""
|
||||||
if not production_guards_active(in_test_mode=in_test_mode):
|
if not production_guards_active(in_test_mode=in_test_mode):
|
||||||
return {
|
return {
|
||||||
"proven": True,
|
"proven": True,
|
||||||
@@ -414,6 +442,7 @@ def assess_production_mutation_guards(
|
|||||||
locked_issue_number=locked_issue_number,
|
locked_issue_number=locked_issue_number,
|
||||||
role_kind=role_kind,
|
role_kind=role_kind,
|
||||||
mutation_task=mutation_task,
|
mutation_task=mutation_task,
|
||||||
|
bootstrap_assessment=bootstrap_assessment,
|
||||||
)
|
)
|
||||||
if root_assess["block"]:
|
if root_assess["block"]:
|
||||||
return {**root_assess, "skipped": False}
|
return {**root_assess, "skipped": False}
|
||||||
|
|||||||
Reference in New Issue
Block a user