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.
|
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:
|
||||||
|
|||||||
@@ -153,19 +153,7 @@ not a tool argument: a session must never be able to authorize itself.
|
|||||||
## Related
|
## Related
|
||||||
|
|
||||||
- #630 — manual daemon killing as contaminated recovery (this contrast, enforced).
|
- #630 — manual daemon killing as contaminated recovery (this contrast, enforced).
|
||||||
- #657 — restart-path inventory and daemon classification.
|
|
||||||
- #686 — manual server launch detection & fail-closed provenance gate.
|
|
||||||
- #531 / #544 — stale-runtime detection (`ps`-based); sibling failure mode.
|
- #531 / #544 — stale-runtime detection (`ps`-based); sibling failure mode.
|
||||||
- #558 / `docs/mcp-daemon-import-guard.md` — why shell imports are not a repair.
|
- #558 / `docs/mcp-daemon-import-guard.md` — why shell imports are not a repair.
|
||||||
- `docs/mcp-client-registration.md` — per-server registration contract.
|
- `docs/mcp-client-registration.md` — per-server registration contract.
|
||||||
- `docs/mcp-namespace-health.md` — probe sources and mutation enforcement.
|
- `docs/mcp-namespace-health.md` — probe sources and mutation enforcement.
|
||||||
|
|
||||||
## Sanctioned reconnect vs forbidden manual launch (#686)
|
|
||||||
|
|
||||||
In addition to manual process killing (#630), manually launching a duplicate role server from an ad hoc shell (`python3 mcp_server.py`) is forbidden and fail-closed:
|
|
||||||
|
|
||||||
- **Why manual launches are unsupported:** A terminal-launched `mcp_server.py` holds its own stdio transport; it can never bind to the IDE client's stdio pipes. It cannot restore a dropped IDE namespace, and a manual duplicate process masks stale client-managed runtimes for that profile, defeating stale-runtime gates.
|
|
||||||
- **Sanctioned path:** Supported recovery is IDE/client-managed reconnect only (`/mcp reconnect`, IDE restart, or sanctioned reconnect exposure).
|
|
||||||
- **Fail-closed enforcement (#686):** Mutating tools on a server lacking client-managed launch provenance (`GITEA_CLIENT_MANAGED=1`) refuse execution fail-closed with typed blocker `unsupported_manual_launch` and an exact next action. Unsupported `GITEA_*` env overrides (e.g. `GITEA_DUMMY`) are surfaced in diagnostics rather than silently ignored.
|
|
||||||
- **Inventory & staleness:** Staleness diagnostics ignore non-client-managed duplicates when evaluating runtime freshness and inventory duplicate processes per profile (#657, #686).
|
|
||||||
|
|
||||||
|
|||||||
@@ -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.
|
||||||
|
|
||||||
|
|||||||
+1
-50
@@ -1169,57 +1169,10 @@ def server_command():
|
|||||||
return python, [os.path.join(root, "mcp_server.py")]
|
return python, [os.path.join(root, "mcp_server.py")]
|
||||||
|
|
||||||
|
|
||||||
RECOGNIZED_GITEA_ENV_KEYS = frozenset({
|
|
||||||
"GITEA_MCP_CONFIG",
|
|
||||||
"GITEA_MCP_PROFILE",
|
|
||||||
"GITEA_PROFILE_NAME",
|
|
||||||
"GITEA_SERVICE",
|
|
||||||
"GITEA_EXECUTION_ROLE",
|
|
||||||
"GITEA_CLIENT_MANAGED",
|
|
||||||
"GITEA_MCP_CLIENT_MANAGED",
|
|
||||||
"GITEA_SERVER_PROVENANCE",
|
|
||||||
"GITEA_AUTHOR_WORKTREE",
|
|
||||||
"GITEA_ACTIVE_WORKTREE",
|
|
||||||
"GITEA_DISABLE_KEYCHAIN",
|
|
||||||
"GITEA_CONTROL_PLANE_DB",
|
|
||||||
"GITEA_DB_PATH",
|
|
||||||
"GITEA_LOG_LEVEL",
|
|
||||||
"GITEA_DEBUG",
|
|
||||||
"GITEA_HMAC_SECRET",
|
|
||||||
"GITEA_IRRECOVERABLE_HMAC_SECRET",
|
|
||||||
"GITEA_FORCE_MCP_RUNTIME_CHECK",
|
|
||||||
"GITEA_FORCE_CLIENT_MANAGED",
|
|
||||||
})
|
|
||||||
|
|
||||||
RECOGNIZED_GITEA_ENV_PREFIXES = (
|
|
||||||
"GITEA_TOKEN_",
|
|
||||||
"GITEA_PASS_",
|
|
||||||
"GITEA_USER_",
|
|
||||||
"GITEA_URL_",
|
|
||||||
"GITEA_HOST_",
|
|
||||||
"GITEA_REMOTE_",
|
|
||||||
"GITEA_HTTP_HEADER_",
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
def get_unconsumed_gitea_env_overrides(env=None) -> dict[str, str]:
|
|
||||||
"""Find unsupported GITEA_* env vars present in *env* (defaults to os.environ)."""
|
|
||||||
target = os.environ if env is None else env
|
|
||||||
unconsumed = {}
|
|
||||||
for key, value in target.items():
|
|
||||||
if key.startswith("GITEA_"):
|
|
||||||
if key in RECOGNIZED_GITEA_ENV_KEYS:
|
|
||||||
continue
|
|
||||||
if any(key.startswith(p) for p in RECOGNIZED_GITEA_ENV_PREFIXES):
|
|
||||||
continue
|
|
||||||
unconsumed[key] = str(value)
|
|
||||||
return unconsumed
|
|
||||||
|
|
||||||
|
|
||||||
def launcher_entry(profile_name, config_path=None):
|
def launcher_entry(profile_name, config_path=None):
|
||||||
"""Return a thin MCP launcher entry for *profile_name*.
|
"""Return a thin MCP launcher entry for *profile_name*.
|
||||||
|
|
||||||
Contains command/args and the GITEA_MCP_* / GITEA_CLIENT_MANAGED env vars — never a token
|
Contains only command/args and the two GITEA_MCP_* env vars — never a token
|
||||||
or password. Suitable for Claude / Gemini / Codex ``mcpServers`` blocks.
|
or password. Suitable for Claude / Gemini / Codex ``mcpServers`` blocks.
|
||||||
"""
|
"""
|
||||||
command, args = server_command()
|
command, args = server_command()
|
||||||
@@ -1230,13 +1183,11 @@ def launcher_entry(profile_name, config_path=None):
|
|||||||
"env": {
|
"env": {
|
||||||
"GITEA_MCP_CONFIG": config_path or DEFAULT_CONFIG_PATH,
|
"GITEA_MCP_CONFIG": config_path or DEFAULT_CONFIG_PATH,
|
||||||
"GITEA_MCP_PROFILE": profile_name,
|
"GITEA_MCP_PROFILE": profile_name,
|
||||||
"GITEA_CLIENT_MANAGED": "1",
|
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
def keychain_set(item_id, token, account=None, runner=subprocess.run):
|
def keychain_set(item_id, token, account=None, runner=subprocess.run):
|
||||||
"""Store *token* in the macOS keychain under service *item_id*.
|
"""Store *token* in the macOS keychain under service *item_id*.
|
||||||
|
|
||||||
|
|||||||
+123
-118
@@ -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,
|
||||||
@@ -14585,56 +14654,6 @@ def _session_context_mutation_block(
|
|||||||
return blocked
|
return blocked
|
||||||
|
|
||||||
|
|
||||||
def _is_client_managed_process() -> bool:
|
|
||||||
"""Check whether the current MCP server process has client-managed launch provenance (#686)."""
|
|
||||||
val = (
|
|
||||||
os.environ.get("GITEA_CLIENT_MANAGED")
|
|
||||||
or os.environ.get("GITEA_MCP_CLIENT_MANAGED")
|
|
||||||
or os.environ.get("GITEA_SERVER_PROVENANCE")
|
|
||||||
or os.environ.get("GITEA_FORCE_CLIENT_MANAGED")
|
|
||||||
or ""
|
|
||||||
).strip().lower()
|
|
||||||
|
|
||||||
if val in ("0", "false", "no", "manual", "manual_launch"):
|
|
||||||
return False
|
|
||||||
|
|
||||||
if val in ("1", "true", "yes", "client_managed"):
|
|
||||||
return True
|
|
||||||
|
|
||||||
# A terminal launch has an active TTY on stdin
|
|
||||||
try:
|
|
||||||
if sys.stdin and sys.stdin.isatty():
|
|
||||||
return False
|
|
||||||
except Exception:
|
|
||||||
pass
|
|
||||||
|
|
||||||
# Standard client launch or test runner with stdio pipe and profile env
|
|
||||||
if "GITEA_MCP_CONFIG" in os.environ or "GITEA_MCP_PROFILE" in os.environ or "GITEA_PROFILE_NAME" in os.environ:
|
|
||||||
return True
|
|
||||||
|
|
||||||
return False
|
|
||||||
|
|
||||||
|
|
||||||
def _provenance_mutation_block(**extra_fields) -> dict | None:
|
|
||||||
"""Refuse mutating tool calls on processes lacking client-managed launch provenance (#686)."""
|
|
||||||
if _is_client_managed_process():
|
|
||||||
return None
|
|
||||||
unconsumed = gitea_config.get_unconsumed_gitea_env_overrides()
|
|
||||||
blocked = {
|
|
||||||
"success": False,
|
|
||||||
"performed": False,
|
|
||||||
"blocker_kind": "unsupported_manual_launch",
|
|
||||||
"reasons": [
|
|
||||||
"mutation denied: server process was launched manually from a terminal without client-managed provenance (fail closed). Manually launched mcp_server.py processes cannot receive IDE stdio or serve workflow mutations."
|
|
||||||
],
|
|
||||||
"exact_next_action": "BLOCKED + RECONNECT: Reconnect the IDE/client-managed MCP server namespace instead of an ad hoc terminal launch. Hand-launched processes and mcp_config.json hand-edits are classified as workflow contamination.",
|
|
||||||
"provenance": "manual_launch",
|
|
||||||
"unconsumed_gitea_env": unconsumed,
|
|
||||||
}
|
|
||||||
blocked.update(extra_fields)
|
|
||||||
return blocked
|
|
||||||
|
|
||||||
|
|
||||||
def _profile_permission_block(required_operation: str, **extra_fields) -> dict | None:
|
def _profile_permission_block(required_operation: str, **extra_fields) -> dict | None:
|
||||||
"""Structured operation-gate denial for gated tools (#69, #142, #897).
|
"""Structured operation-gate denial for gated tools (#69, #142, #897).
|
||||||
|
|
||||||
@@ -14651,10 +14670,6 @@ def _profile_permission_block(required_operation: str, **extra_fields) -> dict |
|
|||||||
# #714: evaluate active profile only — never auto-switch.
|
# #714: evaluate active profile only — never auto-switch.
|
||||||
_ensure_matching_profile(required_operation, req_role, extra_fields.get("remote"))
|
_ensure_matching_profile(required_operation, req_role, extra_fields.get("remote"))
|
||||||
|
|
||||||
prov_block = _provenance_mutation_block(**extra_fields)
|
|
||||||
if prov_block is not None:
|
|
||||||
return prov_block
|
|
||||||
|
|
||||||
reasons = _profile_operation_gate(required_operation)
|
reasons = _profile_operation_gate(required_operation)
|
||||||
if reasons:
|
if reasons:
|
||||||
return _build_operation_gate_refusal(
|
return _build_operation_gate_refusal(
|
||||||
@@ -14688,10 +14703,6 @@ def _namespace_mutation_block(mutation_task: str, **extra_fields) -> dict | None
|
|||||||
# #714: evaluate active profile only — never auto-switch.
|
# #714: evaluate active profile only — never auto-switch.
|
||||||
_ensure_matching_profile(required_permission, required_role, extra_fields.get("remote"))
|
_ensure_matching_profile(required_permission, required_role, extra_fields.get("remote"))
|
||||||
|
|
||||||
prov_block = _provenance_mutation_block(**extra_fields)
|
|
||||||
if prov_block is not None:
|
|
||||||
return prov_block
|
|
||||||
|
|
||||||
try:
|
try:
|
||||||
profile = get_profile()
|
profile = get_profile()
|
||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
@@ -17268,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()
|
||||||
@@ -18139,9 +18155,6 @@ def gitea_get_runtime_context(
|
|||||||
source="gitea_get_runtime_context",
|
source="gitea_get_runtime_context",
|
||||||
)
|
)
|
||||||
|
|
||||||
is_client_managed = _is_client_managed_process()
|
|
||||||
unconsumed_env = gitea_config.get_unconsumed_gitea_env_overrides()
|
|
||||||
|
|
||||||
result = {
|
result = {
|
||||||
"active_profile": profile["profile_name"],
|
"active_profile": profile["profile_name"],
|
||||||
"authenticated_username": username,
|
"authenticated_username": username,
|
||||||
@@ -18158,9 +18171,6 @@ def gitea_get_runtime_context(
|
|||||||
"review_merge_blocked_reasons": blocked_reasons,
|
"review_merge_blocked_reasons": blocked_reasons,
|
||||||
"suggested_fix": suggested_fix,
|
"suggested_fix": suggested_fix,
|
||||||
"safe_next_action": safe_next_action,
|
"safe_next_action": safe_next_action,
|
||||||
"server_provenance": "client_managed" if is_client_managed else "manual_launch",
|
|
||||||
"is_client_managed": is_client_managed,
|
|
||||||
"unconsumed_gitea_env": unconsumed_env,
|
|
||||||
"preflight_ready": preflight["preflight_ready"],
|
"preflight_ready": preflight["preflight_ready"],
|
||||||
"preflight_block_reasons": preflight["preflight_block_reasons"],
|
"preflight_block_reasons": preflight["preflight_block_reasons"],
|
||||||
"preflight_workspace": preflight.get("preflight_workspace"),
|
"preflight_workspace": preflight.get("preflight_workspace"),
|
||||||
@@ -18172,15 +18182,13 @@ 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,
|
||||||
}
|
}
|
||||||
|
|
||||||
if not is_client_managed:
|
|
||||||
result["safe_next_action"] = (
|
|
||||||
"BLOCKED + RECONNECT: Serving process lacks client-managed launch provenance (manual launch). "
|
|
||||||
"Reconnect the IDE/client-managed MCP server namespace instead of an ad hoc terminal launch."
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
# #702: read-only visibility into the inherited GITEA_ACTIVE_WORKTREE
|
# #702: read-only visibility into the inherited GITEA_ACTIVE_WORKTREE
|
||||||
# binding; recovery itself runs during capability resolution.
|
# binding; recovery itself runs during capability resolution.
|
||||||
try:
|
try:
|
||||||
@@ -18682,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",
|
||||||
@@ -18692,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,
|
||||||
@@ -18706,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()
|
||||||
@@ -20638,9 +20670,7 @@ def _check_mcp_runtimes_diagnostics(task: str, matching_profiles: list[str]) ->
|
|||||||
self_pid = os.getpid()
|
self_pid = os.getpid()
|
||||||
self_stale = False
|
self_stale = False
|
||||||
|
|
||||||
all_profile_procs: dict[str, list[dict]] = {}
|
running_profiles = {}
|
||||||
unsupported_env_found = set()
|
|
||||||
|
|
||||||
for line in proc.stdout.splitlines()[1:]:
|
for line in proc.stdout.splitlines()[1:]:
|
||||||
line = line.strip()
|
line = line.strip()
|
||||||
if not line or "mcp_server.py" not in line:
|
if not line or "mcp_server.py" not in line:
|
||||||
@@ -20672,55 +20702,16 @@ def _check_mcp_runtimes_diagnostics(task: str, matching_profiles: list[str]) ->
|
|||||||
if match:
|
if match:
|
||||||
profile = match.group(1)
|
profile = match.group(1)
|
||||||
|
|
||||||
is_client_managed = bool(
|
|
||||||
re.search(r'\bGITEA_CLIENT_MANAGED=(1|true|yes|client_managed)\b', env_out, re.IGNORECASE)
|
|
||||||
or re.search(r'\bGITEA_MCP_CLIENT_MANAGED=(1|true|yes|client_managed)\b', env_out, re.IGNORECASE)
|
|
||||||
or re.search(r'\bGITEA_SERVER_PROVENANCE=client_managed\b', env_out, re.IGNORECASE)
|
|
||||||
)
|
|
||||||
|
|
||||||
for env_match in re.finditer(r'\b(GITEA_[A-Z0-9_]+)=([^\s]+)', env_out):
|
|
||||||
k, v = env_match.group(1), env_match.group(2)
|
|
||||||
if k not in gitea_config.RECOGNIZED_GITEA_ENV_KEYS and not any(k.startswith(p) for p in gitea_config.RECOGNIZED_GITEA_ENV_PREFIXES):
|
|
||||||
unsupported_env_found.add(f"{k}={v}")
|
|
||||||
|
|
||||||
is_stale = (start_time < code_mtime) or git_stale
|
is_stale = (start_time < code_mtime) or git_stale
|
||||||
if pid == self_pid and is_stale:
|
if pid == self_pid and is_stale:
|
||||||
self_stale = True
|
self_stale = True
|
||||||
|
|
||||||
proc_info = {
|
if profile not in running_profiles or start_time > running_profiles[profile]["start_time"]:
|
||||||
|
running_profiles[profile] = {
|
||||||
"pid": pid,
|
"pid": pid,
|
||||||
"start_time": start_time,
|
"start_time": start_time,
|
||||||
"is_stale": is_stale,
|
"is_stale": is_stale
|
||||||
"is_client_managed": is_client_managed,
|
|
||||||
}
|
}
|
||||||
if profile not in all_profile_procs:
|
|
||||||
all_profile_procs[profile] = []
|
|
||||||
all_profile_procs[profile].append(proc_info)
|
|
||||||
|
|
||||||
running_profiles = {}
|
|
||||||
for profile, procs in all_profile_procs.items():
|
|
||||||
if len(procs) > 1:
|
|
||||||
pids_str = ", ".join(str(p["pid"]) for p in procs)
|
|
||||||
reasons.append(
|
|
||||||
f"stale-runtime: Duplicate MCP server process(es) detected for profile '{profile}' (PIDs: {pids_str}). "
|
|
||||||
"Manual or duplicate launches defeat staleness detection and cannot receive client stdio."
|
|
||||||
)
|
|
||||||
client_procs = [p for p in procs if p["is_client_managed"]]
|
|
||||||
if client_procs:
|
|
||||||
client_procs.sort(key=lambda p: p["start_time"], reverse=True)
|
|
||||||
running_profiles[profile] = client_procs[0]
|
|
||||||
else:
|
|
||||||
pids_str = ", ".join(str(p["pid"]) for p in procs)
|
|
||||||
reasons.append(
|
|
||||||
f"stale-runtime: Manually launched MCP process(es) detected without client-managed provenance for profile '{profile}' (PIDs: {pids_str}). "
|
|
||||||
"Manual launches cannot serve client stdio and are ignored for runtime freshness."
|
|
||||||
)
|
|
||||||
|
|
||||||
if unsupported_env_found:
|
|
||||||
reasons.append(
|
|
||||||
f"unsupported-env: Unsupported GITEA_* environment variable override(s) detected: {', '.join(sorted(unsupported_env_found))}. "
|
|
||||||
"Unknown env overrides are unsupported."
|
|
||||||
)
|
|
||||||
|
|
||||||
if self_stale:
|
if self_stale:
|
||||||
# #685: report-only — no config utime, no thread, no os._exit.
|
# #685: report-only — no config utime, no thread, no os._exit.
|
||||||
@@ -20755,7 +20746,6 @@ def _check_mcp_runtimes_diagnostics(task: str, matching_profiles: list[str]) ->
|
|||||||
return reasons
|
return reasons
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
@mcp.tool()
|
@mcp.tool()
|
||||||
def gitea_resolve_task_capability(
|
def gitea_resolve_task_capability(
|
||||||
task: str,
|
task: str,
|
||||||
@@ -20992,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
|
||||||
@@ -21048,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:
|
||||||
@@ -21125,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(
|
||||||
@@ -21191,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
-16
@@ -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:
|
||||||
@@ -225,16 +274,6 @@ def classify_namespace_probe(
|
|||||||
# on bad data without treating success as IDE proof).
|
# on bad data without treating success as IDE proof).
|
||||||
blocks = namespace_health_blocks_task("merge_pr", healthy)
|
blocks = namespace_health_blocks_task("merge_pr", healthy)
|
||||||
|
|
||||||
import gitea_config
|
|
||||||
raw_env = process.get("env") if isinstance(process, dict) else None
|
|
||||||
unconsumed_env = gitea_config.get_unconsumed_gitea_env_overrides(raw_env)
|
|
||||||
is_client_managed = bool(
|
|
||||||
env_summary.get("GITEA_CLIENT_MANAGED") in ("1", "true", "yes", "client_managed")
|
|
||||||
or env_summary.get("GITEA_MCP_CLIENT_MANAGED") in ("1", "true", "yes", "client_managed")
|
|
||||||
or env_summary.get("GITEA_SERVER_PROVENANCE") == "client_managed"
|
|
||||||
)
|
|
||||||
provenance = "client_managed" if is_client_managed else "manual_launch"
|
|
||||||
|
|
||||||
return {
|
return {
|
||||||
"success": healthy,
|
"success": healthy,
|
||||||
"healthy": healthy,
|
"healthy": healthy,
|
||||||
@@ -250,9 +289,6 @@ def classify_namespace_probe(
|
|||||||
"error_message": error_message or None,
|
"error_message": error_message or None,
|
||||||
"reasons": reasons,
|
"reasons": reasons,
|
||||||
"remediation": remediation,
|
"remediation": remediation,
|
||||||
"provenance": provenance,
|
|
||||||
"is_client_managed": is_client_managed,
|
|
||||||
"unconsumed_gitea_env": unconsumed_env,
|
|
||||||
"diagnostics": {
|
"diagnostics": {
|
||||||
"namespace": ns,
|
"namespace": ns,
|
||||||
"required_tool": tool,
|
"required_tool": tool,
|
||||||
@@ -261,9 +297,6 @@ def classify_namespace_probe(
|
|||||||
"env": env_summary,
|
"env": env_summary,
|
||||||
"config_path": config_path,
|
"config_path": config_path,
|
||||||
"probe_source": source,
|
"probe_source": source,
|
||||||
"provenance": provenance,
|
|
||||||
"is_client_managed": is_client_managed,
|
|
||||||
"unconsumed_gitea_env": unconsumed_env,
|
|
||||||
},
|
},
|
||||||
"blocks_merge_workflow": blocks,
|
"blocks_merge_workflow": blocks,
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,3 +0,0 @@
|
|||||||
[pytest]
|
|
||||||
testpaths = tests
|
|
||||||
norecursedirs = branches .git venv __pycache__ graphify-out
|
|
||||||
+2
-2
@@ -41,11 +41,10 @@ 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)
|
||||||
|
|
||||||
monkeypatch.setenv("GITEA_CLIENT_MANAGED", "1")
|
|
||||||
|
|
||||||
# Isolate durable session-state files so tests never share host cache (#559).
|
# Isolate durable session-state files so tests never share host cache (#559).
|
||||||
import tempfile
|
import tempfile
|
||||||
|
|
||||||
@@ -117,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)
|
||||||
|
|||||||
@@ -35,7 +35,7 @@ CONFIG = {
|
|||||||
],
|
],
|
||||||
"forbidden_operations": [],
|
"forbidden_operations": [],
|
||||||
"execution_profile": "full-author",
|
"execution_profile": "full-author",
|
||||||
"allowed_repositories": ["Scaled-Tech-Consulting/Gitea-Tools", "Example-Org/Example-Repo"],
|
"allowed_repositories": ["Example-Org/Example-Repo"],
|
||||||
},
|
},
|
||||||
"reviewer-no-commit": {
|
"reviewer-no-commit": {
|
||||||
"enabled": True,
|
"enabled": True,
|
||||||
@@ -50,7 +50,7 @@ CONFIG = {
|
|||||||
"gitea.repo.commit", "gitea.pr.create", "gitea.branch.push"
|
"gitea.repo.commit", "gitea.pr.create", "gitea.branch.push"
|
||||||
],
|
],
|
||||||
"execution_profile": "reviewer-no-commit",
|
"execution_profile": "reviewer-no-commit",
|
||||||
"allowed_repositories": ["Scaled-Tech-Consulting/Gitea-Tools", "Example-Org/Example-Repo"],
|
"allowed_repositories": ["Example-Org/Example-Repo"],
|
||||||
},
|
},
|
||||||
},
|
},
|
||||||
"rules": {"allow_runtime_switching": False},
|
"rules": {"allow_runtime_switching": False},
|
||||||
|
|||||||
@@ -175,7 +175,7 @@ class TestLauncherSnippets(unittest.TestCase):
|
|||||||
def test_only_safe_keys_no_secrets(self):
|
def test_only_safe_keys_no_secrets(self):
|
||||||
entry = gitea_config.launcher_entry("prgs", "/cfg/profiles.json")["gitea-tools"]
|
entry = gitea_config.launcher_entry("prgs", "/cfg/profiles.json")["gitea-tools"]
|
||||||
self.assertEqual(set(entry), {"command", "args", "env"})
|
self.assertEqual(set(entry), {"command", "args", "env"})
|
||||||
self.assertEqual(set(entry["env"]), {"GITEA_MCP_CONFIG", "GITEA_MCP_PROFILE", "GITEA_CLIENT_MANAGED"})
|
self.assertEqual(set(entry["env"]), {"GITEA_MCP_CONFIG", "GITEA_MCP_PROFILE"})
|
||||||
self.assertEqual(entry["env"]["GITEA_MCP_PROFILE"], "prgs")
|
self.assertEqual(entry["env"]["GITEA_MCP_PROFILE"], "prgs")
|
||||||
blob = json.dumps(entry).lower()
|
blob = json.dumps(entry).lower()
|
||||||
for word in ("token", "password", "secret"):
|
for word in ("token", "password", "secret"):
|
||||||
|
|||||||
@@ -1,139 +0,0 @@
|
|||||||
"""Tests for Issue #686: Detect and reject manually launched duplicate MCP role servers."""
|
|
||||||
import os
|
|
||||||
import unittest
|
|
||||||
from unittest.mock import patch, MagicMock
|
|
||||||
from datetime import datetime
|
|
||||||
|
|
||||||
import gitea_config
|
|
||||||
import gitea_mcp_server
|
|
||||||
import mcp_namespace_health
|
|
||||||
|
|
||||||
|
|
||||||
class TestIssue686ManualMcpProvenance(unittest.TestCase):
|
|
||||||
|
|
||||||
def test_client_managed_process_detection(self):
|
|
||||||
"""Test _is_client_managed_process correctly detects provenance markers."""
|
|
||||||
with patch.dict(os.environ, {"GITEA_CLIENT_MANAGED": "1"}, clear=True):
|
|
||||||
self.assertTrue(gitea_mcp_server._is_client_managed_process())
|
|
||||||
|
|
||||||
with patch.dict(os.environ, {"GITEA_MCP_CLIENT_MANAGED": "true"}, clear=True):
|
|
||||||
self.assertTrue(gitea_mcp_server._is_client_managed_process())
|
|
||||||
|
|
||||||
with patch.dict(os.environ, {"GITEA_SERVER_PROVENANCE": "client_managed"}, clear=True):
|
|
||||||
self.assertTrue(gitea_mcp_server._is_client_managed_process())
|
|
||||||
|
|
||||||
with patch.dict(os.environ, {"GITEA_CLIENT_MANAGED": "0"}, clear=True):
|
|
||||||
self.assertFalse(gitea_mcp_server._is_client_managed_process())
|
|
||||||
|
|
||||||
def test_unconsumed_gitea_env_overrides(self):
|
|
||||||
"""Test surfacing of unsupported GITEA_* env overrides (e.g. GITEA_DUMMY)."""
|
|
||||||
env = {
|
|
||||||
"GITEA_MCP_PROFILE": "prgs-author",
|
|
||||||
"GITEA_CLIENT_MANAGED": "1",
|
|
||||||
"GITEA_DUMMY": "2",
|
|
||||||
"GITEA_UNKNOWN_FLAG": "abc",
|
|
||||||
}
|
|
||||||
unconsumed = gitea_config.get_unconsumed_gitea_env_overrides(env)
|
|
||||||
self.assertIn("GITEA_DUMMY", unconsumed)
|
|
||||||
self.assertEqual(unconsumed["GITEA_DUMMY"], "2")
|
|
||||||
self.assertIn("GITEA_UNKNOWN_FLAG", unconsumed)
|
|
||||||
self.assertNotIn("GITEA_MCP_PROFILE", unconsumed)
|
|
||||||
self.assertNotIn("GITEA_CLIENT_MANAGED", unconsumed)
|
|
||||||
|
|
||||||
def test_manual_server_mutation_fail_closed(self):
|
|
||||||
"""AC 2: Mutating tools on a server without client-managed provenance fail closed with a typed blocker."""
|
|
||||||
with patch.dict(os.environ, {"GITEA_CLIENT_MANAGED": "0"}, clear=True):
|
|
||||||
block = gitea_mcp_server._provenance_mutation_block(task="create_issue")
|
|
||||||
self.assertIsNotNone(block)
|
|
||||||
self.assertFalse(block["success"])
|
|
||||||
self.assertFalse(block["performed"])
|
|
||||||
self.assertEqual(block["blocker_kind"], "unsupported_manual_launch")
|
|
||||||
self.assertEqual(block["provenance"], "manual_launch")
|
|
||||||
self.assertTrue(any("mutation denied: server process was launched manually" in r for r in block["reasons"]))
|
|
||||||
self.assertIn("BLOCKED + RECONNECT", block["exact_next_action"])
|
|
||||||
|
|
||||||
def test_client_managed_server_mutation_passes_provenance_gate(self):
|
|
||||||
"""AC 3: Clean client-managed baseline passes the provenance gate."""
|
|
||||||
with patch.dict(os.environ, {"GITEA_CLIENT_MANAGED": "1"}, clear=True):
|
|
||||||
block = gitea_mcp_server._provenance_mutation_block(task="create_issue")
|
|
||||||
self.assertIsNone(block)
|
|
||||||
|
|
||||||
@patch("subprocess.run")
|
|
||||||
@patch("os.path.getmtime")
|
|
||||||
@patch("os.path.exists")
|
|
||||||
@patch("os.getpid")
|
|
||||||
def test_manual_duplicate_does_not_mask_stale_runtime(
|
|
||||||
self, mock_getpid, mock_exists, mock_getmtime, mock_run
|
|
||||||
):
|
|
||||||
"""AC 1 & AC 3: Staleness detection ignores manual duplicates and reports stale supported runtimes."""
|
|
||||||
mock_getpid.return_value = 12345
|
|
||||||
mock_exists.return_value = True
|
|
||||||
|
|
||||||
code_time = datetime(2026, 7, 8, 14, 0, 0)
|
|
||||||
mock_getmtime.return_value = code_time.timestamp()
|
|
||||||
|
|
||||||
# PID 12345: stale client-managed process (started at 13:00)
|
|
||||||
# PID 99999: fresh manual duplicate process (started at 15:00, no GITEA_CLIENT_MANAGED)
|
|
||||||
ps_output = (
|
|
||||||
" PID LSTART COMMAND\n"
|
|
||||||
"12345 Wed Jul 8 13:00:00 2026 /path/to/python mcp_server.py\n"
|
|
||||||
"99999 Wed Jul 8 15:00:00 2026 /path/to/python mcp_server.py\n"
|
|
||||||
)
|
|
||||||
|
|
||||||
mock_run_ps = MagicMock()
|
|
||||||
mock_run_ps.stdout = ps_output
|
|
||||||
|
|
||||||
mock_env_12345 = MagicMock()
|
|
||||||
mock_env_12345.stdout = "GITEA_MCP_PROFILE=prgs-author GITEA_CLIENT_MANAGED=1"
|
|
||||||
|
|
||||||
mock_env_99999 = MagicMock()
|
|
||||||
mock_env_99999.stdout = "GITEA_MCP_PROFILE=prgs-author GITEA_DUMMY=2"
|
|
||||||
|
|
||||||
def side_effect(args, **kwargs):
|
|
||||||
if args[0] == "ps" and "eww" in args:
|
|
||||||
pid = args[2]
|
|
||||||
if pid == "12345":
|
|
||||||
return mock_env_12345
|
|
||||||
elif pid == "99999":
|
|
||||||
return mock_env_99999
|
|
||||||
elif args[0] == "ps":
|
|
||||||
return mock_run_ps
|
|
||||||
raise ValueError(f"Unexpected args: {args}")
|
|
||||||
|
|
||||||
mock_run.side_effect = side_effect
|
|
||||||
|
|
||||||
reasons = gitea_mcp_server._check_mcp_runtimes_diagnostics("create_issue", ["prgs-author"])
|
|
||||||
|
|
||||||
# Manual duplicate process must be flagged
|
|
||||||
self.assertTrue(any("Duplicate MCP server process(es) detected" in r for r in reasons))
|
|
||||||
# Unsupported env override (GITEA_DUMMY=2) must be flagged
|
|
||||||
self.assertTrue(any("unsupported-env: Unsupported GITEA_* environment variable override(s) detected: GITEA_DUMMY=2" in r for r in reasons))
|
|
||||||
# Stale runtime must NOT be masked by fresh manual process 99999!
|
|
||||||
self.assertTrue(any("All matching profiles for task 'create_issue' (['prgs-author']) are running but stale" in r for r in reasons))
|
|
||||||
|
|
||||||
def test_namespace_health_classification_includes_provenance(self):
|
|
||||||
"""AC 1 & 4: mcp_namespace_health diagnostics include provenance and unconsumed_gitea_env."""
|
|
||||||
process = {
|
|
||||||
"pid": 5555,
|
|
||||||
"profile": "prgs-author",
|
|
||||||
"env": {
|
|
||||||
"GITEA_MCP_PROFILE": "prgs-author",
|
|
||||||
"GITEA_DUMMY": "99",
|
|
||||||
},
|
|
||||||
}
|
|
||||||
res = mcp_namespace_health.classify_namespace_probe(
|
|
||||||
"gitea-author",
|
|
||||||
configured=True,
|
|
||||||
registered_tools=["gitea_whoami"],
|
|
||||||
probe_result={"success": True},
|
|
||||||
process=process,
|
|
||||||
probe_source="client_namespace",
|
|
||||||
)
|
|
||||||
self.assertEqual(res["provenance"], "manual_launch")
|
|
||||||
self.assertFalse(res["is_client_managed"])
|
|
||||||
self.assertEqual(res["unconsumed_gitea_env"], {"GITEA_DUMMY": "99"})
|
|
||||||
self.assertEqual(res["diagnostics"]["provenance"], "manual_launch")
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
|
||||||
unittest.main()
|
|
||||||
@@ -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()
|
||||||
@@ -35,10 +35,10 @@ class TestMcpStaleRuntime(unittest.TestCase):
|
|||||||
|
|
||||||
# Mock env output for ps eww
|
# Mock env output for ps eww
|
||||||
mock_run_env12345 = MagicMock()
|
mock_run_env12345 = MagicMock()
|
||||||
mock_run_env12345.stdout = "GITEA_MCP_PROFILE=prgs-reconciler GITEA_CLIENT_MANAGED=1"
|
mock_run_env12345.stdout = "GITEA_MCP_PROFILE=prgs-reconciler"
|
||||||
|
|
||||||
mock_run_env54321 = MagicMock()
|
mock_run_env54321 = MagicMock()
|
||||||
mock_run_env54321.stdout = "GITEA_MCP_PROFILE=prgs-author GITEA_CLIENT_MANAGED=1"
|
mock_run_env54321.stdout = "GITEA_MCP_PROFILE=prgs-author"
|
||||||
|
|
||||||
def side_effect(args, **kwargs):
|
def side_effect(args, **kwargs):
|
||||||
if args[0] == "ps" and "eww" in args:
|
if args[0] == "ps" and "eww" in args:
|
||||||
@@ -91,7 +91,7 @@ class TestMcpStaleRuntime(unittest.TestCase):
|
|||||||
mock_run_ps.stdout = ps_output
|
mock_run_ps.stdout = ps_output
|
||||||
|
|
||||||
mock_run_env = MagicMock()
|
mock_run_env = MagicMock()
|
||||||
mock_run_env.stdout = "GITEA_MCP_PROFILE=prgs-author GITEA_CLIENT_MANAGED=1"
|
mock_run_env.stdout = "GITEA_MCP_PROFILE=prgs-author"
|
||||||
|
|
||||||
mock_run_git = MagicMock()
|
mock_run_git = MagicMock()
|
||||||
mock_run_git.stdout = "FAKE2" # different SHA
|
mock_run_git.stdout = "FAKE2" # different SHA
|
||||||
|
|||||||
@@ -243,10 +243,9 @@ class TestRuntimeClarity(unittest.TestCase):
|
|||||||
self.assertIn("switching is disabled", res["message"].lower())
|
self.assertIn("switching is disabled", res["message"].lower())
|
||||||
self.assertIsNone(gitea_config._active_profile_override)
|
self.assertIsNone(gitea_config._active_profile_override)
|
||||||
|
|
||||||
@patch("mcp_server._trusted_session_repository", return_value={"repository": "Example-Org/Example-Repo", "org": "Example-Org", "repo": "Example-Repo", "reasons": []})
|
|
||||||
@patch("mcp_server.api_request")
|
@patch("mcp_server.api_request")
|
||||||
@patch("mcp_server.get_auth_header")
|
@patch("mcp_server.get_auth_header")
|
||||||
def test_activate_profile_succeeds_when_enabled(self, mock_auth, mock_api, mock_trusted):
|
def test_activate_profile_succeeds_when_enabled(self, mock_auth, mock_api):
|
||||||
self._write_config(CONFIG_SWITCHING_ENABLED)
|
self._write_config(CONFIG_SWITCHING_ENABLED)
|
||||||
|
|
||||||
# Setup mock responses for whoami checks
|
# Setup mock responses for whoami checks
|
||||||
|
|||||||
Reference in New Issue
Block a user