Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
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
|
||||||
|
|||||||
@@ -589,47 +589,6 @@ When dynamic profile switching is enabled and a profile is activated via `gitea_
|
|||||||
2. Call `gitea_whoami` with the target remote to prove and verify the fresh Gitea authenticated identity.
|
2. Call `gitea_whoami` with the target remote to prove and verify the fresh Gitea authenticated identity.
|
||||||
This guarantees the active profile operations align with the actual Gitea authenticated user credential.
|
This guarantees the active profile operations align with the actual Gitea authenticated user credential.
|
||||||
|
|
||||||
### 4. Review-State Invalidation on Profile Switch (#690)
|
|
||||||
|
|
||||||
A cross-profile activation (e.g. reviewer → author → reviewer) is a session
|
|
||||||
boundary for formal review state. On any switch where the activated profile
|
|
||||||
differs from the previous one, `gitea_activate_profile` invalidates, in
|
|
||||||
memory **and** in durable session state (for both the old and new profile
|
|
||||||
identities):
|
|
||||||
|
|
||||||
- preflight identity/capability stamps (`gitea_whoami` / `gitea_resolve_task_capability` proof),
|
|
||||||
- review workflow-load proof (`gitea_load_review_workflow`),
|
|
||||||
- the review decision lock (including `final_review_decision_ready` markers),
|
|
||||||
- the reviewer PR session lease binding,
|
|
||||||
- live namespace-health assessments.
|
|
||||||
|
|
||||||
Before any formal verdict (`gitea_mark_final_review_decision` /
|
|
||||||
`gitea_submit_pr_review`) the full reviewer preflight must be re-established
|
|
||||||
under the new profile: `gitea_whoami`, `gitea_load_review_workflow`,
|
|
||||||
`gitea_resolve_task_capability(review_pr)`, live head re-pin, and lease
|
|
||||||
re-acquire/adopt. Switching back to the earlier profile does **not**
|
|
||||||
resurrect the prior run — durable state keyed by either profile identity is
|
|
||||||
cleared at switch time.
|
|
||||||
|
|
||||||
The supported pattern remains **separate session/namespace per role**
|
|
||||||
(dual-namespace, §2): file author-side follow-ups from an author session,
|
|
||||||
not by hopping profiles inside a formal review run. Runtime profile
|
|
||||||
switching is the operator-approved exception and always carries the
|
|
||||||
re-preflight cost above.
|
|
||||||
|
|
||||||
### 5. Namespace Provenance (#690)
|
|
||||||
|
|
||||||
The server cannot derive its own client-managed MCP namespace name, so a
|
|
||||||
launcher may declare it via the `GITEA_MCP_NAMESPACE` environment variable
|
|
||||||
(e.g. `gitea-reviewer`). `gitea_whoami`, `gitea_get_runtime_context`, and
|
|
||||||
`gitea_resolve_task_capability` report `namespace_provenance` — the
|
|
||||||
configured client namespace, the active execution profile, and, for tasks
|
|
||||||
with a required namespace (`review_pr` → `gitea-reviewer`, `merge_pr` →
|
|
||||||
`gitea-merger`), a mismatch verdict. A declared namespace that disagrees
|
|
||||||
with the requested task's required namespace **fails closed**. An
|
|
||||||
undeclared namespace is reported as `unknown` and is never treated as
|
|
||||||
proof either way.
|
|
||||||
|
|
||||||
## Gitea MCP Runtime Isolation and Worktree Safety
|
## Gitea MCP Runtime Isolation and Worktree Safety
|
||||||
|
|
||||||
To ensure high availability and prevent broken feature worktrees from disabling essential security/identity controls, the Gitea MCP server implements runtime isolation:
|
To ensure high availability and prevent broken feature worktrees from disabling essential security/identity controls, the Gitea MCP server implements runtime isolation:
|
||||||
|
|||||||
@@ -86,24 +86,3 @@ When a namespace returns EOF, follow
|
|||||||
|
|
||||||
When blocked, repair the IDE namespace and re-record a healthy
|
When blocked, repair the IDE namespace and re-record a healthy
|
||||||
`client_namespace` assessment before retrying the mutation.
|
`client_namespace` assessment before retrying the mutation.
|
||||||
|
|
||||||
## Namespace provenance (#690)
|
|
||||||
|
|
||||||
A server process cannot derive the name of the client-managed namespace it is
|
|
||||||
registered under, so the launcher may declare it with the
|
|
||||||
`GITEA_MCP_NAMESPACE` environment variable (e.g. `GITEA_MCP_NAMESPACE=gitea-reviewer`).
|
|
||||||
|
|
||||||
- `gitea_whoami`, `gitea_get_runtime_context`, and
|
|
||||||
`gitea_resolve_task_capability` report `namespace_provenance`: the declared
|
|
||||||
client namespace, the active execution profile, and — for tasks with a
|
|
||||||
required namespace (`review_pr`/`submit_review` → `gitea-reviewer`,
|
|
||||||
`merge_pr` → `gitea-merger`) — a `mismatch` verdict.
|
|
||||||
- A declared namespace that disagrees with the requested task's required
|
|
||||||
namespace **fails closed** (`allowed_in_current_session=false` with a STOP
|
|
||||||
guidance entry).
|
|
||||||
- An undeclared namespace is reported as `namespace_source="unknown"` and is
|
|
||||||
never treated as proof either way.
|
|
||||||
- A profile switch via `gitea_activate_profile` clears all recorded live
|
|
||||||
namespace-health assessments; re-probe through the client before further
|
|
||||||
review/merge mutations.
|
|
||||||
|
|
||||||
|
|||||||
+1
-119
@@ -839,75 +839,6 @@ def _invalidate_preflight_identity_state() -> None:
|
|||||||
_clear_preflight_capability_state()
|
_clear_preflight_capability_state()
|
||||||
|
|
||||||
|
|
||||||
# #690: session-boundary invalidation record for the most recent cross-profile
|
|
||||||
# activation. Surfaced in runtime diagnostics so a formal review run can prove
|
|
||||||
# its state was reset by a profile switch and must be fully re-established.
|
|
||||||
_PROFILE_SWITCH_INVALIDATION: dict | None = None
|
|
||||||
|
|
||||||
|
|
||||||
def _invalidate_review_state_on_profile_switch(
|
|
||||||
before_profile: str,
|
|
||||||
after_profile: str,
|
|
||||||
) -> dict:
|
|
||||||
"""Invalidate review-critical session state on a profile switch (#690).
|
|
||||||
|
|
||||||
Workflow-load proof, reviewer lease binding, the review decision lock,
|
|
||||||
live namespace health, and preflight identity/capability stamps recorded
|
|
||||||
under the prior profile are contaminated for the new role. Durable state
|
|
||||||
keyed by *either* profile identity is cleared so a reviewer → author →
|
|
||||||
reviewer hop cannot resurrect a stale review run: the full reviewer
|
|
||||||
preflight (gitea_whoami, gitea_load_review_workflow,
|
|
||||||
gitea_resolve_task_capability(review_pr), live head re-pin, lease
|
|
||||||
re-acquire/adopt) must be re-established before any formal verdict.
|
|
||||||
"""
|
|
||||||
global _PROFILE_SWITCH_INVALIDATION
|
|
||||||
invalidated: list[str] = []
|
|
||||||
|
|
||||||
_invalidate_preflight_identity_state()
|
|
||||||
invalidated.append("preflight_identity_capability")
|
|
||||||
|
|
||||||
review_workflow_load.clear_review_workflow_load()
|
|
||||||
invalidated.append("review_workflow_load")
|
|
||||||
|
|
||||||
_save_review_decision_lock(None)
|
|
||||||
invalidated.append("review_decision_lock")
|
|
||||||
|
|
||||||
reviewer_pr_lease.clear_session_lease()
|
|
||||||
invalidated.append("reviewer_session_lease")
|
|
||||||
|
|
||||||
if _LIVE_NAMESPACE_HEALTH:
|
|
||||||
_LIVE_NAMESPACE_HEALTH.clear()
|
|
||||||
invalidated.append("live_namespace_health")
|
|
||||||
|
|
||||||
# Durable records keyed by either profile identity must not survive the
|
|
||||||
# switch, or activating author → reviewer → author could revive a stale
|
|
||||||
# review run without re-preflight.
|
|
||||||
for identity in {before_profile, after_profile}:
|
|
||||||
if not identity:
|
|
||||||
continue
|
|
||||||
try:
|
|
||||||
mcp_session_state.clear_state(
|
|
||||||
kind=mcp_session_state.KIND_DECISION_LOCK,
|
|
||||||
profile_identity=identity,
|
|
||||||
)
|
|
||||||
mcp_session_state.clear_state(
|
|
||||||
kind=mcp_session_state.KIND_WORKFLOW_LOAD,
|
|
||||||
profile_identity=identity,
|
|
||||||
)
|
|
||||||
except Exception:
|
|
||||||
pass # best-effort durable cleanup; in-memory state already reset
|
|
||||||
invalidated.append("durable_profile_state")
|
|
||||||
|
|
||||||
_PROFILE_SWITCH_INVALIDATION = {
|
|
||||||
"from_profile": before_profile,
|
|
||||||
"to_profile": after_profile,
|
|
||||||
"invalidated": invalidated,
|
|
||||||
"invalidated_at": datetime.now(timezone.utc).isoformat(),
|
|
||||||
"re_preflight_required": True,
|
|
||||||
}
|
|
||||||
return dict(_PROFILE_SWITCH_INVALIDATION)
|
|
||||||
|
|
||||||
|
|
||||||
def record_preflight_check(
|
def record_preflight_check(
|
||||||
type_name: str,
|
type_name: str,
|
||||||
resolved_role: str | None = None,
|
resolved_role: str | None = None,
|
||||||
@@ -17279,11 +17210,6 @@ def gitea_whoami(
|
|||||||
"session_context_audit": session_ctx.mutation_context_audit_fields(),
|
"session_context_audit": session_ctx.mutation_context_audit_fields(),
|
||||||
"identity_match": not id_match.get("block"),
|
"identity_match": not id_match.get("block"),
|
||||||
"identity_match_reasons": id_match.get("reasons") or [],
|
"identity_match_reasons": id_match.get("reasons") or [],
|
||||||
# #690 AC4: report launcher-declared client namespace alongside the
|
|
||||||
# active execution profile so drift is visible in diagnostics.
|
|
||||||
"namespace_provenance": mcp_namespace_health.namespace_provenance(
|
|
||||||
active_profile=profile["profile_name"]
|
|
||||||
),
|
|
||||||
}
|
}
|
||||||
if id_match.get("block"):
|
if id_match.get("block"):
|
||||||
_invalidate_preflight_identity_state()
|
_invalidate_preflight_identity_state()
|
||||||
@@ -18182,11 +18108,6 @@ def gitea_get_runtime_context(
|
|||||||
"shell_health": native_mcp_preference.shell_health_status(),
|
"shell_health": native_mcp_preference.shell_health_status(),
|
||||||
"workflow_load_proof": review_workflow_load.workflow_load_status(
|
"workflow_load_proof": review_workflow_load.workflow_load_status(
|
||||||
PROJECT_ROOT),
|
PROJECT_ROOT),
|
||||||
# #690: namespace provenance + profile-switch invalidation evidence.
|
|
||||||
"namespace_provenance": mcp_namespace_health.namespace_provenance(
|
|
||||||
active_profile=profile["profile_name"]
|
|
||||||
),
|
|
||||||
"profile_switch_invalidation": _PROFILE_SWITCH_INVALIDATION,
|
|
||||||
}
|
}
|
||||||
|
|
||||||
# #702: read-only visibility into the inherited GITEA_ACTIVE_WORKTREE
|
# #702: read-only visibility into the inherited GITEA_ACTIVE_WORKTREE
|
||||||
@@ -18690,17 +18611,6 @@ def gitea_activate_profile(
|
|||||||
source="gitea_activate_profile",
|
source="gitea_activate_profile",
|
||||||
)
|
)
|
||||||
|
|
||||||
# 4.7 #690: a profile switch is a session-boundary event for review state.
|
|
||||||
# Any workflow-load proof, reviewer lease, decision lock, namespace
|
|
||||||
# health, or preflight stamp recorded under the prior profile is
|
|
||||||
# contaminated for the new role and must be re-established under the new
|
|
||||||
# profile before any formal review verdict.
|
|
||||||
switch_invalidation = None
|
|
||||||
if before_profile != after_profile:
|
|
||||||
switch_invalidation = _invalidate_review_state_on_profile_switch(
|
|
||||||
before_profile, after_profile
|
|
||||||
)
|
|
||||||
|
|
||||||
# 5. Audit the switch if auditing is on
|
# 5. Audit the switch if auditing is on
|
||||||
_audit(
|
_audit(
|
||||||
"activate_profile",
|
"activate_profile",
|
||||||
@@ -18711,12 +18621,11 @@ def gitea_activate_profile(
|
|||||||
"before": before_profile,
|
"before": before_profile,
|
||||||
"after": after_profile,
|
"after": after_profile,
|
||||||
"session_context": session_ctx.mutation_context_audit_fields(),
|
"session_context": session_ctx.mutation_context_audit_fields(),
|
||||||
"review_state_invalidated": bool(switch_invalidation),
|
|
||||||
},
|
},
|
||||||
username=after_identity,
|
username=after_identity,
|
||||||
)
|
)
|
||||||
|
|
||||||
result = {
|
return {
|
||||||
"success": True,
|
"success": True,
|
||||||
"message": f"Successfully activated profile '{profile_name}' (fresh identity verification complete).",
|
"message": f"Successfully activated profile '{profile_name}' (fresh identity verification complete).",
|
||||||
"before_profile": before_profile,
|
"before_profile": before_profile,
|
||||||
@@ -18726,18 +18635,6 @@ def gitea_activate_profile(
|
|||||||
"session_context_audit": session_ctx.mutation_context_audit_fields(),
|
"session_context_audit": session_ctx.mutation_context_audit_fields(),
|
||||||
"auto_profile_substitution": False,
|
"auto_profile_substitution": False,
|
||||||
}
|
}
|
||||||
if switch_invalidation is not None:
|
|
||||||
result["review_state_invalidation"] = switch_invalidation
|
|
||||||
result["re_preflight_required"] = True
|
|
||||||
result["exact_next_action"] = (
|
|
||||||
"Profile switch invalidated workflow-load proof, reviewer lease, "
|
|
||||||
"decision lock, and preflight stamps (#690). Before any formal "
|
|
||||||
"review verdict, re-run the full reviewer preflight: "
|
|
||||||
"gitea_whoami, gitea_load_review_workflow, "
|
|
||||||
"gitea_resolve_task_capability(review_pr), live head re-pin, and "
|
|
||||||
"lease re-acquire/adopt."
|
|
||||||
)
|
|
||||||
return result
|
|
||||||
|
|
||||||
|
|
||||||
@mcp.tool()
|
@mcp.tool()
|
||||||
@@ -20982,22 +20879,12 @@ def gitea_resolve_task_capability(
|
|||||||
f"{required_role} task '{task}' even if nearby permissions are "
|
f"{required_role} task '{task}' even if nearby permissions are "
|
||||||
"present (fail closed)."
|
"present (fail closed)."
|
||||||
)
|
)
|
||||||
# #690 AC4: when the launcher declares a client namespace, a task with a
|
|
||||||
# required namespace must fail closed on mismatch (e.g. review_pr served
|
|
||||||
# from an author namespace).
|
|
||||||
ns_provenance = mcp_namespace_health.namespace_provenance(
|
|
||||||
task=task_key, active_profile=profile.get("profile_name")
|
|
||||||
)
|
|
||||||
ns_mismatch_reason = None
|
|
||||||
if ns_provenance.get("mismatch"):
|
|
||||||
ns_mismatch_reason = "; ".join(ns_provenance.get("reasons") or [])
|
|
||||||
cross_host_block = bool(remote_assess.get("block"))
|
cross_host_block = bool(remote_assess.get("block"))
|
||||||
identity_block = bool(id_assess.get("block"))
|
identity_block = bool(id_assess.get("block"))
|
||||||
drift_block = bool(ctx_assess.get("block"))
|
drift_block = bool(ctx_assess.get("block"))
|
||||||
allowed_in_current_session = (
|
allowed_in_current_session = (
|
||||||
permission_allowed_in_current_session
|
permission_allowed_in_current_session
|
||||||
and role_matches_current_session
|
and role_matches_current_session
|
||||||
and not ns_provenance.get("mismatch")
|
|
||||||
and not cross_host_block
|
and not cross_host_block
|
||||||
and not identity_block
|
and not identity_block
|
||||||
and not drift_block
|
and not drift_block
|
||||||
@@ -21048,8 +20935,6 @@ def gitea_resolve_task_capability(
|
|||||||
)
|
)
|
||||||
if role_mismatch_reason:
|
if role_mismatch_reason:
|
||||||
deny_parts.append(role_mismatch_reason)
|
deny_parts.append(role_mismatch_reason)
|
||||||
if ns_mismatch_reason:
|
|
||||||
deny_parts.append(ns_mismatch_reason)
|
|
||||||
if deny_parts:
|
if deny_parts:
|
||||||
reason_msg = "; ".join(deny_parts)
|
reason_msg = "; ".join(deny_parts)
|
||||||
elif configured and switching:
|
elif configured and switching:
|
||||||
@@ -21127,8 +21012,6 @@ def gitea_resolve_task_capability(
|
|||||||
task_role_guidance = []
|
task_role_guidance = []
|
||||||
if role_mismatch_reason:
|
if role_mismatch_reason:
|
||||||
task_role_guidance.append(f"STOP: {role_mismatch_reason}")
|
task_role_guidance.append(f"STOP: {role_mismatch_reason}")
|
||||||
if ns_mismatch_reason:
|
|
||||||
task_role_guidance.append(f"STOP: {ns_mismatch_reason}")
|
|
||||||
if required_role == "reviewer":
|
if required_role == "reviewer":
|
||||||
if allowed_in_current_session:
|
if allowed_in_current_session:
|
||||||
task_role_guidance.append(
|
task_role_guidance.append(
|
||||||
@@ -21195,7 +21078,6 @@ def gitea_resolve_task_capability(
|
|||||||
"session_context_audit": session_ctx.mutation_context_audit_fields(),
|
"session_context_audit": session_ctx.mutation_context_audit_fields(),
|
||||||
"profile_remote_compatible": not cross_host_block,
|
"profile_remote_compatible": not cross_host_block,
|
||||||
"identity_match": not identity_block,
|
"identity_match": not identity_block,
|
||||||
"namespace_provenance": ns_provenance,
|
|
||||||
"auto_profile_substitution": False,
|
"auto_profile_substitution": False,
|
||||||
}
|
}
|
||||||
# #685: report typed reconnect blocker without mutating config or exiting.
|
# #685: report typed reconnect blocker without mutating config or exiting.
|
||||||
|
|||||||
@@ -16,7 +16,6 @@ Probe sources
|
|||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import os
|
|
||||||
from typing import Any
|
from typing import Any
|
||||||
|
|
||||||
|
|
||||||
@@ -58,56 +57,8 @@ SAFE_ENV_KEYS = (
|
|||||||
"GITEA_SERVICE",
|
"GITEA_SERVICE",
|
||||||
"GITEA_EXECUTION_ROLE",
|
"GITEA_EXECUTION_ROLE",
|
||||||
"GITEA_MCP_CONFIG",
|
"GITEA_MCP_CONFIG",
|
||||||
"GITEA_MCP_NAMESPACE",
|
|
||||||
)
|
)
|
||||||
|
|
||||||
# Optional launcher-provided env declaring the client-managed MCP namespace
|
|
||||||
# this process is registered under (e.g. ``gitea-reviewer``). The server
|
|
||||||
# cannot derive its own IDE namespace name, so the launcher declares it; when
|
|
||||||
# declared, reviewers/mergers can fail closed on a namespace/task mismatch
|
|
||||||
# (#690 AC4). Absence means "unknown" — reported, never guessed.
|
|
||||||
NAMESPACE_ENV = "GITEA_MCP_NAMESPACE"
|
|
||||||
|
|
||||||
|
|
||||||
def configured_client_namespace(env: dict[str, str] | None = None) -> str | None:
|
|
||||||
"""Return the launcher-declared client namespace, or None when unknown."""
|
|
||||||
source = os.environ if env is None else env
|
|
||||||
value = (source.get(NAMESPACE_ENV) or "").strip()
|
|
||||||
return value or None
|
|
||||||
|
|
||||||
|
|
||||||
def namespace_provenance(
|
|
||||||
task: str | None = None,
|
|
||||||
*,
|
|
||||||
active_profile: str | None = None,
|
|
||||||
env: dict[str, str] | None = None,
|
|
||||||
) -> dict[str, Any]:
|
|
||||||
"""Report configured client namespace vs active execution profile (#690).
|
|
||||||
|
|
||||||
When *task* carries a required namespace (``TASK_REQUIRED_NAMESPACES``)
|
|
||||||
and the launcher declared a different one, ``mismatch`` is True and the
|
|
||||||
caller must fail closed for that task. An undeclared namespace is
|
|
||||||
reported as unknown — never treated as proof either way.
|
|
||||||
"""
|
|
||||||
configured = configured_client_namespace(env)
|
|
||||||
required = TASK_REQUIRED_NAMESPACES.get(task or "")
|
|
||||||
mismatch = bool(configured and required and configured != required)
|
|
||||||
reasons: list[str] = []
|
|
||||||
if mismatch:
|
|
||||||
reasons.append(
|
|
||||||
f"configured client namespace '{configured}' does not match "
|
|
||||||
f"required namespace '{required}' for task '{task}' (fail closed)"
|
|
||||||
)
|
|
||||||
return {
|
|
||||||
"configured_namespace": configured,
|
|
||||||
"namespace_source": NAMESPACE_ENV if configured else "unknown",
|
|
||||||
"active_profile": active_profile,
|
|
||||||
"requested_task": task,
|
|
||||||
"required_namespace": required,
|
|
||||||
"mismatch": mismatch,
|
|
||||||
"reasons": reasons,
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
def _as_list(value: Any) -> list[str] | None:
|
def _as_list(value: Any) -> list[str] | None:
|
||||||
if value is None:
|
if value is None:
|
||||||
|
|||||||
@@ -41,7 +41,6 @@ def _reset_mutation_authority(monkeypatch):
|
|||||||
"GITEA_REVIEWER_WORKTREE",
|
"GITEA_REVIEWER_WORKTREE",
|
||||||
"GITEA_MERGER_WORKTREE",
|
"GITEA_MERGER_WORKTREE",
|
||||||
"GITEA_RECONCILER_WORKTREE",
|
"GITEA_RECONCILER_WORKTREE",
|
||||||
"GITEA_MCP_NAMESPACE",
|
|
||||||
]:
|
]:
|
||||||
monkeypatch.delenv(env_key, raising=False)
|
monkeypatch.delenv(env_key, raising=False)
|
||||||
|
|
||||||
@@ -116,7 +115,6 @@ def _reset_mutation_authority(monkeypatch):
|
|||||||
monkeypatch.setattr(mcp_server, "_ACTOR_IDENTITY_CACHE", {})
|
monkeypatch.setattr(mcp_server, "_ACTOR_IDENTITY_CACHE", {})
|
||||||
monkeypatch.setattr(mcp_server, "_REVIEW_DECISION_LOCK", None)
|
monkeypatch.setattr(mcp_server, "_REVIEW_DECISION_LOCK", None)
|
||||||
monkeypatch.setattr(mcp_server, "_LIVE_NAMESPACE_HEALTH", {})
|
monkeypatch.setattr(mcp_server, "_LIVE_NAMESPACE_HEALTH", {})
|
||||||
monkeypatch.setattr(mcp_server, "_PROFILE_SWITCH_INVALIDATION", None)
|
|
||||||
monkeypatch.setattr(mcp_server, "_preflight_whoami_called", False)
|
monkeypatch.setattr(mcp_server, "_preflight_whoami_called", False)
|
||||||
monkeypatch.setattr(mcp_server, "_preflight_capability_called", False)
|
monkeypatch.setattr(mcp_server, "_preflight_capability_called", False)
|
||||||
monkeypatch.setattr(mcp_server, "_preflight_resolved_role", None)
|
monkeypatch.setattr(mcp_server, "_preflight_resolved_role", None)
|
||||||
|
|||||||
@@ -1,274 +0,0 @@
|
|||||||
"""Regression coverage for #690: cross-role profile activation invalidation.
|
|
||||||
|
|
||||||
A mid-run profile switch (e.g. reviewer → author → reviewer) must invalidate
|
|
||||||
workflow-load proof, reviewer lease binding, review decision lock, live
|
|
||||||
namespace health, and preflight identity/capability stamps, and must require
|
|
||||||
a full reviewer preflight before any formal verdict. Namespace provenance
|
|
||||||
must be reported and fail closed on task/namespace mismatch.
|
|
||||||
"""
|
|
||||||
import json
|
|
||||||
import os
|
|
||||||
import sys
|
|
||||||
import tempfile
|
|
||||||
import unittest
|
|
||||||
from unittest.mock import patch
|
|
||||||
|
|
||||||
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
|
||||||
|
|
||||||
import gitea_config
|
|
||||||
import mcp_namespace_health
|
|
||||||
import mcp_server
|
|
||||||
import mcp_session_state
|
|
||||||
import review_workflow_load
|
|
||||||
import reviewer_pr_lease
|
|
||||||
|
|
||||||
from tests.test_runtime_clarity import CONFIG_SWITCHING_ENABLED
|
|
||||||
|
|
||||||
|
|
||||||
class TestProfileSwitchReviewGuard(unittest.TestCase):
|
|
||||||
def setUp(self):
|
|
||||||
self._remotes_patch = patch.dict(mcp_server.REMOTES, {
|
|
||||||
"dadeschools": {"host": "gitea.example.com", "org": "Example-Org", "repo": "Example-Repo"},
|
|
||||||
"prgs": {"host": "gitea.example.com", "org": "Example-Org", "repo": "Example-Repo"},
|
|
||||||
})
|
|
||||||
self._remotes_patch.start()
|
|
||||||
mcp_server._IDENTITY_CACHE.clear()
|
|
||||||
gitea_config._active_profile_override = None
|
|
||||||
self._dir = tempfile.TemporaryDirectory()
|
|
||||||
self.config_path = os.path.join(self._dir.name, "profiles.json")
|
|
||||||
with open(self.config_path, "w", encoding="utf-8") as fh:
|
|
||||||
fh.write(json.dumps(CONFIG_SWITCHING_ENABLED))
|
|
||||||
|
|
||||||
def tearDown(self):
|
|
||||||
self._remotes_patch.stop()
|
|
||||||
mcp_server._IDENTITY_CACHE.clear()
|
|
||||||
gitea_config._active_profile_override = None
|
|
||||||
self._dir.cleanup()
|
|
||||||
|
|
||||||
def _env(self, profile="reviewer-profile"):
|
|
||||||
return {
|
|
||||||
"GITEA_MCP_CONFIG": self.config_path,
|
|
||||||
"GITEA_MCP_PROFILE": profile,
|
|
||||||
"GITEA_TOKEN_AUTHOR": "author-pass",
|
|
||||||
"GITEA_TOKEN_REVIEWER": "reviewer-pass",
|
|
||||||
"GITEA_TOKEN_MERGER": "merger-pass",
|
|
||||||
}
|
|
||||||
|
|
||||||
def _seed_contaminated_review_state(self):
|
|
||||||
"""Simulate an in-flight reviewer run under reviewer-profile."""
|
|
||||||
mcp_server._preflight_whoami_called = True
|
|
||||||
mcp_server._preflight_capability_called = True
|
|
||||||
mcp_server._preflight_resolved_role = "reviewer"
|
|
||||||
mcp_server._preflight_resolved_task = "review_pr"
|
|
||||||
review_workflow_load._REVIEW_WORKFLOW_LOAD = {"loaded": True}
|
|
||||||
mcp_server._REVIEW_DECISION_LOCK = {
|
|
||||||
"session_profile": "reviewer-profile",
|
|
||||||
"final_review_decision_ready": True,
|
|
||||||
"ready_pr_number": 688,
|
|
||||||
}
|
|
||||||
reviewer_pr_lease.record_session_lease(
|
|
||||||
{"session_id": "lease-session-1", "pr_number": 688}
|
|
||||||
)
|
|
||||||
mcp_server._LIVE_NAMESPACE_HEALTH["gitea-reviewer"] = {
|
|
||||||
"namespace": "gitea-reviewer",
|
|
||||||
"healthy": True,
|
|
||||||
"ide_namespace_proven": True,
|
|
||||||
}
|
|
||||||
# Durable records keyed by the reviewer identity must also be cleared.
|
|
||||||
mcp_session_state.save_state(
|
|
||||||
kind=mcp_session_state.KIND_WORKFLOW_LOAD,
|
|
||||||
payload={"loaded": True},
|
|
||||||
profile_identity="reviewer-profile",
|
|
||||||
)
|
|
||||||
mcp_session_state.save_state(
|
|
||||||
kind=mcp_session_state.KIND_DECISION_LOCK,
|
|
||||||
payload={"final_review_decision_ready": True, "ready_pr_number": 688},
|
|
||||||
profile_identity="reviewer-profile",
|
|
||||||
)
|
|
||||||
|
|
||||||
def _activate(self, target, logins):
|
|
||||||
with patch.object(
|
|
||||||
mcp_server, "get_auth_header", side_effect=[f"token p" for _ in logins]
|
|
||||||
), patch.object(
|
|
||||||
mcp_server, "api_request", side_effect=[{"login": l} for l in logins]
|
|
||||||
), patch.object(
|
|
||||||
mcp_server,
|
|
||||||
"_workspace_repository_slug",
|
|
||||||
return_value="Example-Org/Example-Repo",
|
|
||||||
), patch.object(
|
|
||||||
mcp_server, "_canonical_repository_slug", return_value=(None, [])
|
|
||||||
):
|
|
||||||
return mcp_server.gitea_activate_profile(profile_name=target)
|
|
||||||
|
|
||||||
# -----------------------------------------------------------------
|
|
||||||
# AC1/AC2/AC3: switch invalidates review state; re-preflight required
|
|
||||||
# -----------------------------------------------------------------
|
|
||||||
def test_switch_invalidates_review_state_and_blocks_verdict(self):
|
|
||||||
with patch.dict(os.environ, self._env("reviewer-profile"), clear=True):
|
|
||||||
self._seed_contaminated_review_state()
|
|
||||||
res = self._activate("author-profile", ["reviewer-user", "author-user"])
|
|
||||||
|
|
||||||
self.assertTrue(res["success"])
|
|
||||||
self.assertTrue(res["re_preflight_required"])
|
|
||||||
inv = res["review_state_invalidation"]
|
|
||||||
self.assertEqual(inv["from_profile"], "reviewer-profile")
|
|
||||||
self.assertEqual(inv["to_profile"], "author-profile")
|
|
||||||
for item in (
|
|
||||||
"preflight_identity_capability",
|
|
||||||
"review_workflow_load",
|
|
||||||
"review_decision_lock",
|
|
||||||
"reviewer_session_lease",
|
|
||||||
"live_namespace_health",
|
|
||||||
):
|
|
||||||
self.assertIn(item, inv["invalidated"])
|
|
||||||
|
|
||||||
# In-memory state cleared.
|
|
||||||
self.assertFalse(mcp_server._preflight_whoami_called)
|
|
||||||
self.assertFalse(mcp_server._preflight_capability_called)
|
|
||||||
self.assertIsNone(mcp_server._preflight_resolved_task)
|
|
||||||
self.assertIsNone(review_workflow_load._REVIEW_WORKFLOW_LOAD)
|
|
||||||
self.assertIsNone(mcp_server._REVIEW_DECISION_LOCK)
|
|
||||||
self.assertIsNone(reviewer_pr_lease.get_session_lease())
|
|
||||||
self.assertEqual(mcp_server._LIVE_NAMESPACE_HEALTH, {})
|
|
||||||
self.assertIsNotNone(mcp_server._PROFILE_SWITCH_INVALIDATION)
|
|
||||||
|
|
||||||
# Durable records keyed by the reviewer identity are gone.
|
|
||||||
self.assertIsNone(
|
|
||||||
mcp_session_state.load_state(
|
|
||||||
kind=mcp_session_state.KIND_WORKFLOW_LOAD,
|
|
||||||
profile_identity="reviewer-profile",
|
|
||||||
)
|
|
||||||
)
|
|
||||||
self.assertIsNone(
|
|
||||||
mcp_session_state.load_state(
|
|
||||||
kind=mcp_session_state.KIND_DECISION_LOCK,
|
|
||||||
profile_identity="reviewer-profile",
|
|
||||||
)
|
|
||||||
)
|
|
||||||
|
|
||||||
# A formal verdict without re-preflight fails closed.
|
|
||||||
reasons = mcp_server.check_review_decision_gate(
|
|
||||||
688, "APPROVE", final_review_decision_ready=True
|
|
||||||
)
|
|
||||||
self.assertTrue(reasons)
|
|
||||||
|
|
||||||
def test_switch_back_cannot_resurrect_stale_review_run(self):
|
|
||||||
with patch.dict(os.environ, self._env("reviewer-profile"), clear=True):
|
|
||||||
self._seed_contaminated_review_state()
|
|
||||||
self._activate("author-profile", ["reviewer-user", "author-user"])
|
|
||||||
res = self._activate("reviewer-profile", ["author-user", "reviewer-user"])
|
|
||||||
|
|
||||||
self.assertTrue(res["success"])
|
|
||||||
# The pre-switch review run must not reappear.
|
|
||||||
self.assertIsNone(mcp_server._REVIEW_DECISION_LOCK)
|
|
||||||
self.assertIsNone(review_workflow_load._REVIEW_WORKFLOW_LOAD)
|
|
||||||
self.assertIsNone(reviewer_pr_lease.get_session_lease())
|
|
||||||
status = review_workflow_load.workflow_load_status()
|
|
||||||
self.assertFalse(status["workflow_load_valid"])
|
|
||||||
reasons = mcp_server.check_review_decision_gate(
|
|
||||||
688, "APPROVE", final_review_decision_ready=True
|
|
||||||
)
|
|
||||||
self.assertTrue(reasons)
|
|
||||||
|
|
||||||
def test_same_profile_reactivation_keeps_state(self):
|
|
||||||
with patch.dict(os.environ, self._env("reviewer-profile"), clear=True):
|
|
||||||
self._seed_contaminated_review_state()
|
|
||||||
res = self._activate("reviewer-profile", ["reviewer-user", "reviewer-user"])
|
|
||||||
self.assertTrue(res["success"], res)
|
|
||||||
self.assertNotIn("review_state_invalidation", res)
|
|
||||||
self.assertIsNotNone(mcp_server._REVIEW_DECISION_LOCK)
|
|
||||||
self.assertTrue(mcp_server._preflight_whoami_called)
|
|
||||||
|
|
||||||
def test_clean_repreflight_after_switch_allows_gate(self):
|
|
||||||
with patch.dict(os.environ, self._env("reviewer-profile"), clear=True):
|
|
||||||
self._seed_contaminated_review_state()
|
|
||||||
self._activate("author-profile", ["reviewer-user", "author-user"])
|
|
||||||
self._activate("reviewer-profile", ["author-user", "reviewer-user"])
|
|
||||||
|
|
||||||
# Re-establish the full reviewer preflight under the new profile.
|
|
||||||
mcp_server.record_preflight_check("whoami")
|
|
||||||
mcp_server.record_preflight_check(
|
|
||||||
"capability", resolved_role="reviewer", resolved_task="review_pr"
|
|
||||||
)
|
|
||||||
mcp_server.init_review_decision_lock("dadeschools", "review_pr")
|
|
||||||
lock = mcp_server._load_review_decision_lock()
|
|
||||||
self.assertIsNotNone(lock)
|
|
||||||
lock.update(
|
|
||||||
{
|
|
||||||
"final_review_decision_ready": True,
|
|
||||||
"ready_pr_number": 688,
|
|
||||||
"ready_action": "APPROVE",
|
|
||||||
"ready_remote": "dadeschools",
|
|
||||||
"ready_org": "Example-Org",
|
|
||||||
"ready_repo": "Example-Repo",
|
|
||||||
}
|
|
||||||
)
|
|
||||||
mcp_server._save_review_decision_lock(lock)
|
|
||||||
|
|
||||||
with patch.object(
|
|
||||||
mcp_server, "_review_workflow_load_gate_reasons", return_value=[]
|
|
||||||
):
|
|
||||||
reasons = mcp_server.check_review_decision_gate(
|
|
||||||
688,
|
|
||||||
"APPROVE",
|
|
||||||
final_review_decision_ready=True,
|
|
||||||
remote="dadeschools",
|
|
||||||
)
|
|
||||||
self.assertEqual(reasons, [])
|
|
||||||
|
|
||||||
# -----------------------------------------------------------------
|
|
||||||
# AC4: namespace provenance reporting + fail-closed mismatch
|
|
||||||
# -----------------------------------------------------------------
|
|
||||||
def test_namespace_provenance_mismatch_detection(self):
|
|
||||||
prov = mcp_namespace_health.namespace_provenance(
|
|
||||||
task="review_pr",
|
|
||||||
active_profile="reviewer-profile",
|
|
||||||
env={"GITEA_MCP_NAMESPACE": "gitea-author"},
|
|
||||||
)
|
|
||||||
self.assertTrue(prov["mismatch"])
|
|
||||||
self.assertEqual(prov["required_namespace"], "gitea-reviewer")
|
|
||||||
|
|
||||||
prov_ok = mcp_namespace_health.namespace_provenance(
|
|
||||||
task="review_pr",
|
|
||||||
active_profile="reviewer-profile",
|
|
||||||
env={"GITEA_MCP_NAMESPACE": "gitea-reviewer"},
|
|
||||||
)
|
|
||||||
self.assertFalse(prov_ok["mismatch"])
|
|
||||||
|
|
||||||
prov_unknown = mcp_namespace_health.namespace_provenance(
|
|
||||||
task="review_pr", active_profile="reviewer-profile", env={}
|
|
||||||
)
|
|
||||||
self.assertIsNone(prov_unknown["configured_namespace"])
|
|
||||||
self.assertFalse(prov_unknown["mismatch"])
|
|
||||||
self.assertEqual(prov_unknown["namespace_source"], "unknown")
|
|
||||||
|
|
||||||
@patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
|
|
||||||
@patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
|
|
||||||
def test_whoami_reports_namespace_provenance(self, _auth, _api):
|
|
||||||
env = self._env("reviewer-profile")
|
|
||||||
env["GITEA_MCP_NAMESPACE"] = "gitea-reviewer"
|
|
||||||
with patch.dict(os.environ, env, clear=True):
|
|
||||||
res = mcp_server.gitea_whoami(remote="dadeschools")
|
|
||||||
prov = res["namespace_provenance"]
|
|
||||||
self.assertEqual(prov["configured_namespace"], "gitea-reviewer")
|
|
||||||
self.assertEqual(prov["active_profile"], "reviewer-profile")
|
|
||||||
self.assertFalse(prov["mismatch"])
|
|
||||||
|
|
||||||
@patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
|
|
||||||
@patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
|
|
||||||
def test_resolve_fails_closed_on_namespace_mismatch(self, _auth, _api):
|
|
||||||
env = self._env("reviewer-profile")
|
|
||||||
env["GITEA_MCP_NAMESPACE"] = "gitea-author"
|
|
||||||
with patch.dict(os.environ, env, clear=True):
|
|
||||||
res = mcp_server.gitea_resolve_task_capability(
|
|
||||||
task="review_pr", kwargs="{}", remote="dadeschools"
|
|
||||||
)
|
|
||||||
self.assertFalse(res["allowed_in_current_session"])
|
|
||||||
self.assertTrue(res["namespace_provenance"]["mismatch"])
|
|
||||||
self.assertTrue(
|
|
||||||
any("namespace" in g for g in res["task_role_guidance"])
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
|
||||||
unittest.main()
|
|
||||||
@@ -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()
|
||||||
Reference in New Issue
Block a user