Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
2b3f5baaeb |
@@ -589,6 +589,47 @@ 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.
|
||||
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
|
||||
|
||||
To ensure high availability and prevent broken feature worktrees from disabling essential security/identity controls, the Gitea MCP server implements runtime isolation:
|
||||
|
||||
@@ -86,3 +86,24 @@ When a namespace returns EOF, follow
|
||||
|
||||
When blocked, repair the IDE namespace and re-record a healthy
|
||||
`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.
|
||||
|
||||
|
||||
+119
-1
@@ -839,6 +839,75 @@ def _invalidate_preflight_identity_state() -> None:
|
||||
_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(
|
||||
type_name: str,
|
||||
resolved_role: str | None = None,
|
||||
@@ -17210,6 +17279,11 @@ def gitea_whoami(
|
||||
"session_context_audit": session_ctx.mutation_context_audit_fields(),
|
||||
"identity_match": not id_match.get("block"),
|
||||
"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"):
|
||||
_invalidate_preflight_identity_state()
|
||||
@@ -18108,6 +18182,11 @@ def gitea_get_runtime_context(
|
||||
"shell_health": native_mcp_preference.shell_health_status(),
|
||||
"workflow_load_proof": review_workflow_load.workflow_load_status(
|
||||
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
|
||||
@@ -18611,6 +18690,17 @@ def 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
|
||||
_audit(
|
||||
"activate_profile",
|
||||
@@ -18621,11 +18711,12 @@ def gitea_activate_profile(
|
||||
"before": before_profile,
|
||||
"after": after_profile,
|
||||
"session_context": session_ctx.mutation_context_audit_fields(),
|
||||
"review_state_invalidated": bool(switch_invalidation),
|
||||
},
|
||||
username=after_identity,
|
||||
)
|
||||
|
||||
return {
|
||||
result = {
|
||||
"success": True,
|
||||
"message": f"Successfully activated profile '{profile_name}' (fresh identity verification complete).",
|
||||
"before_profile": before_profile,
|
||||
@@ -18635,6 +18726,18 @@ def gitea_activate_profile(
|
||||
"session_context_audit": session_ctx.mutation_context_audit_fields(),
|
||||
"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()
|
||||
@@ -20879,12 +20982,22 @@ def gitea_resolve_task_capability(
|
||||
f"{required_role} task '{task}' even if nearby permissions are "
|
||||
"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"))
|
||||
identity_block = bool(id_assess.get("block"))
|
||||
drift_block = bool(ctx_assess.get("block"))
|
||||
allowed_in_current_session = (
|
||||
permission_allowed_in_current_session
|
||||
and role_matches_current_session
|
||||
and not ns_provenance.get("mismatch")
|
||||
and not cross_host_block
|
||||
and not identity_block
|
||||
and not drift_block
|
||||
@@ -20935,6 +21048,8 @@ def gitea_resolve_task_capability(
|
||||
)
|
||||
if role_mismatch_reason:
|
||||
deny_parts.append(role_mismatch_reason)
|
||||
if ns_mismatch_reason:
|
||||
deny_parts.append(ns_mismatch_reason)
|
||||
if deny_parts:
|
||||
reason_msg = "; ".join(deny_parts)
|
||||
elif configured and switching:
|
||||
@@ -21012,6 +21127,8 @@ def gitea_resolve_task_capability(
|
||||
task_role_guidance = []
|
||||
if 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 allowed_in_current_session:
|
||||
task_role_guidance.append(
|
||||
@@ -21078,6 +21195,7 @@ def gitea_resolve_task_capability(
|
||||
"session_context_audit": session_ctx.mutation_context_audit_fields(),
|
||||
"profile_remote_compatible": not cross_host_block,
|
||||
"identity_match": not identity_block,
|
||||
"namespace_provenance": ns_provenance,
|
||||
"auto_profile_substitution": False,
|
||||
}
|
||||
# #685: report typed reconnect blocker without mutating config or exiting.
|
||||
|
||||
@@ -16,6 +16,7 @@ Probe sources
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from typing import Any
|
||||
|
||||
|
||||
@@ -57,8 +58,56 @@ SAFE_ENV_KEYS = (
|
||||
"GITEA_SERVICE",
|
||||
"GITEA_EXECUTION_ROLE",
|
||||
"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:
|
||||
if value is None:
|
||||
|
||||
@@ -41,6 +41,7 @@ def _reset_mutation_authority(monkeypatch):
|
||||
"GITEA_REVIEWER_WORKTREE",
|
||||
"GITEA_MERGER_WORKTREE",
|
||||
"GITEA_RECONCILER_WORKTREE",
|
||||
"GITEA_MCP_NAMESPACE",
|
||||
]:
|
||||
monkeypatch.delenv(env_key, raising=False)
|
||||
|
||||
@@ -115,6 +116,7 @@ def _reset_mutation_authority(monkeypatch):
|
||||
monkeypatch.setattr(mcp_server, "_ACTOR_IDENTITY_CACHE", {})
|
||||
monkeypatch.setattr(mcp_server, "_REVIEW_DECISION_LOCK", None)
|
||||
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_capability_called", False)
|
||||
monkeypatch.setattr(mcp_server, "_preflight_resolved_role", None)
|
||||
|
||||
@@ -0,0 +1,274 @@
|
||||
"""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()
|
||||
Reference in New Issue
Block a user