Compare commits

..
Author SHA1 Message Date
sysadmin 2b3f5baaeb fix(mcp): invalidate review session state on cross-profile activation (Closes #690)
A mid-run profile switch (reviewer -> author -> reviewer) left workflow-load
proof, reviewer lease binding, the review decision lock, live namespace
health, and preflight identity/capability stamps intact, so a formal verdict
could be recorded under contaminated session state.

- gitea_activate_profile now invalidates all review-critical session state
  on a cross-profile switch, in memory and in durable state keyed by either
  profile identity, and reports the invalidation + re-preflight requirement.
- Full reviewer preflight (whoami, load_review_workflow,
  resolve_task_capability(review_pr), head re-pin, lease re-acquire) is
  required before any formal verdict after a switch; switching back cannot
  resurrect the stale run.
- Namespace provenance: optional launcher-declared GITEA_MCP_NAMESPACE is
  reported by whoami/runtime context/capability resolution, and a declared
  namespace that disagrees with a task's required namespace fails closed.
- Docs: supported pattern is separate session/namespace per role, not
  in-process profile hopping mid-review.
2026-07-25 19:17:19 -04:00
7 changed files with 508 additions and 35 deletions
+41
View File
@@ -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. 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:
+21
View File
@@ -86,3 +86,24 @@ 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.
+121 -3
View File
@@ -577,10 +577,10 @@ def _get_workspace_porcelain(worktree_path: str | None = None) -> str:
def _parse_porcelain_entries(porcelain: str) -> dict[str, str]: def _parse_porcelain_entries(porcelain: str) -> dict[str, str]:
"""Map path -> full porcelain line (including untracked ``??``).""" """Map tracked path -> full porcelain line (untracked ``??`` ignored)."""
entries: dict[str, str] = {} entries: dict[str, str] = {}
for line in (porcelain or "").splitlines(): for line in (porcelain or "").splitlines():
if not line or len(line) < 4: if not line or len(line) < 4 or line.startswith("??"):
continue continue
path = line[3:].strip() path = line[3:].strip()
if " -> " in path: if " -> " in path:
@@ -839,6 +839,75 @@ 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,
@@ -17210,6 +17279,11 @@ 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()
@@ -18108,6 +18182,11 @@ 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
@@ -18611,6 +18690,17 @@ 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",
@@ -18621,11 +18711,12 @@ 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,
) )
return { result = {
"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,
@@ -18635,6 +18726,18 @@ 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()
@@ -20879,12 +20982,22 @@ 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
@@ -20935,6 +21048,8 @@ 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:
@@ -21012,6 +21127,8 @@ 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(
@@ -21078,6 +21195,7 @@ 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.
+49
View File
@@ -16,6 +16,7 @@ Probe sources
from __future__ import annotations from __future__ import annotations
import os
from typing import Any from typing import Any
@@ -57,8 +58,56 @@ 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:
+2
View File
@@ -41,6 +41,7 @@ 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)
@@ -115,6 +116,7 @@ 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)
@@ -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()
@@ -1,32 +0,0 @@
"""Regression test for Issue #697: Native workspace preflight explicitly includes untracked files.
Verifies that `_parse_porcelain_entries` includes untracked files (lines starting with '??')
so workspace preflight fails closed on untracked or dirty state.
"""
from __future__ import annotations
import unittest
import gitea_mcp_server as mcp_server
class TestIssue697UntrackedPreflight(unittest.TestCase):
def test_parse_porcelain_entries_includes_untracked_files(self):
porcelain = " M gitea_mcp_server.py\n?? untracked_file.txt\n?? new_dir/test.py\n"
entries = mcp_server._parse_porcelain_entries(porcelain)
self.assertIn("untracked_file.txt", entries)
self.assertIn("new_dir/test.py", entries)
self.assertEqual(entries["untracked_file.txt"], "?? untracked_file.txt")
self.assertEqual(entries["new_dir/test.py"], "?? new_dir/test.py")
def test_parse_porcelain_entries_includes_tracked_and_untracked(self):
porcelain = " M tracked.py\n?? untracked.py\n D deleted.py\n"
entries = mcp_server._parse_porcelain_entries(porcelain)
self.assertEqual(len(entries), 3)
self.assertIn("tracked.py", entries)
self.assertIn("untracked.py", entries)
self.assertIn("deleted.py", entries)
if __name__ == "__main__":
unittest.main()