Compare commits

..
15 changed files with 576 additions and 405 deletions
+44 -155
View File
@@ -386,68 +386,6 @@ def run_compensating_recovery(
return recovery_info return recovery_info
def _normalize_sha(value: str | None) -> str | None:
"""Normalize a Git object id for comparison, or ``None`` when unknown."""
normalized = (value or "").strip().lower()
return normalized or None
def _author_bootstrap_assessment(
*,
not_applicable: bool,
allowed: bool,
block: bool,
reasons: list[str],
workspace: str,
root: str,
branch: str | None,
dirty: list[str],
under_branches: bool,
bootstrap_path: str | None = None,
local_head_sha: str | None = None,
remote_master_sha: str | None = None,
exact_next_action: str | None = None,
) -> dict[str, Any]:
"""Structured author-bootstrap assessment consumable by bootstrap_permits (#892).
Field shape mirrors :func:`create_issue_bootstrap._result` so the shared
``bootstrap_permits_control_checkout`` predicate can prove control-checkout
eligibility for ``gitea_bootstrap_author_issue_worktree`` the same way it
does for ``create_issue``. Allowed control assessments must use empty
``reasons`` — narrative belongs in other fields, not the refusal list.
"""
local_tip = _normalize_sha(local_head_sha)
remote_tip = _normalize_sha(remote_master_sha)
base_tips_verified = bool(local_tip and remote_tip and local_tip == remote_tip)
return {
"not_applicable": not_applicable,
"allowed": allowed,
"block": block,
"proven": bool(allowed and not block and not not_applicable),
"reasons": list(reasons),
"workspace_path": workspace,
"canonical_repo_root": root,
"current_branch": branch,
"dirty_files": list(dirty),
"under_branches": under_branches,
"exact_next_action": exact_next_action,
"bootstrap_path": bootstrap_path,
"task_scope": "author_issue_bootstrap",
"local_head_sha": local_tip,
"remote_master_sha": remote_tip,
"base_tips_verified": base_tips_verified,
}
EXACT_NEXT_ACTION_AUTHOR_BOOTSTRAP = (
"Restore the canonical control checkout to a clean accepted base branch "
"(master/main/dev) that matches live master, with no tracked local edits. "
"Re-resolve bootstrap_author_issue_worktree, then re-run "
"gitea_bootstrap_author_issue_worktree from that clean control checkout. "
"Do not use shell git worktree add as the primary path once bootstrap is healthy."
)
def assess_author_issue_bootstrap( def assess_author_issue_bootstrap(
*, *,
workspace_path: str, workspace_path: str,
@@ -459,13 +397,7 @@ def assess_author_issue_bootstrap(
remote_master_sha_error: str | None = None, remote_master_sha_error: str | None = None,
task: str | None = None, task: str | None = None,
) -> dict[str, Any]: ) -> dict[str, Any]:
"""Assess whether author issue worktree bootstrap may proceed from control or worktree root. """Assess whether author issue worktree bootstrap may proceed from control or worktree root."""
#892: control-checkout successes emit the full field set required by
``create_issue_bootstrap.bootstrap_permits_control_checkout`` (empty reasons,
task_scope, base tip proof, binding paths) so the #274/#604 guards can
waive control-checkout for this one sanctioned bootstrap task.
"""
root = os.path.realpath(canonical_repo_root or "") root = os.path.realpath(canonical_repo_root or "")
workspace = os.path.realpath(workspace_path or root or ".") workspace = os.path.realpath(workspace_path or root or ".")
branch = (current_branch or "").strip() branch = (current_branch or "").strip()
@@ -475,50 +407,34 @@ def assess_author_issue_bootstrap(
if root if root
else False else False
) )
local_tip = _normalize_sha(head_sha)
remote_tip = _normalize_sha(remote_master_sha)
if not is_author_issue_bootstrap_task(task): if not is_author_issue_bootstrap_task(task):
return _author_bootstrap_assessment( return {
not_applicable=True, "not_applicable": True,
allowed=False, "allowed": False,
block=False, "block": False,
reasons=["task is not author_issue_bootstrap"], "proven": False,
workspace=workspace, "reasons": ["task is not author_issue_bootstrap"],
root=root, }
branch=branch or None,
dirty=dirty,
under_branches=under_branches,
)
# Already under branches/: ordinary #274 path applies; not a control waiver.
if under_branches: if under_branches:
return _author_bootstrap_assessment( return {
not_applicable=True, "not_applicable": False,
allowed=False, "allowed": True,
block=False, "block": False,
reasons=["workspace is under branches/; ordinary #274 path applies"], "proven": True,
workspace=workspace, "bootstrap_path": "existing_branches_worktree",
root=root, "reasons": [
branch=branch or None, "workspace is already a registered worktree under branches/"
dirty=dirty, ],
under_branches=True, }
bootstrap_path="existing_branches_worktree",
local_head_sha=local_tip,
remote_master_sha=remote_tip,
)
reasons: list[str] = [] reasons: list[str] = []
if not root or workspace != root: if workspace != root:
reasons.append( reasons.append(
"bootstrap requires workspace to be canonical control checkout or branches/ worktree" "bootstrap requires workspace to be canonical control checkout or branches/ worktree"
) )
if not branch: if branch not in author_mutation_worktree.BASE_BRANCHES:
reasons.append(
"control checkout is detached HEAD; expected an accepted base branch "
f"({', '.join(sorted(author_mutation_worktree.BASE_BRANCHES))})"
)
elif branch not in author_mutation_worktree.BASE_BRANCHES:
reasons.append( reasons.append(
f"control checkout branch '{branch}' is not an accepted base branch " f"control checkout branch '{branch}' is not an accepted base branch "
f"({', '.join(sorted(author_mutation_worktree.BASE_BRANCHES))})" f"({', '.join(sorted(author_mutation_worktree.BASE_BRANCHES))})"
@@ -528,64 +444,37 @@ def assess_author_issue_bootstrap(
f"control checkout has tracked local edits: {', '.join(dirty[:5])}" f"control checkout has tracked local edits: {', '.join(dirty[:5])}"
) )
# Fail closed on missing tip proof (same bar as create_issue bootstrap #757). if remote_master_sha_error:
if not local_tip:
reasons.append( reasons.append(
"control checkout HEAD SHA is unknown; base equivalence to live " f"could not verify live master tip: {remote_master_sha_error}"
"master cannot be proven (fail closed)"
) )
resolver_error = (remote_master_sha_error or "").strip() or None elif remote_master_sha and head_sha:
if resolver_error: h = head_sha.strip().lower()
rm = remote_master_sha.strip().lower()
if h != rm:
reasons.append( reasons.append(
f"live master tip could not be resolved ({resolver_error}); " f"control checkout HEAD ({h[:12]}) != live master tip ({rm[:12]})"
"base equivalence cannot be proven (fail closed)"
)
elif not remote_tip:
reasons.append(
"live master tip is unknown; base equivalence cannot be proven "
"(fail closed)"
)
elif local_tip and remote_tip and local_tip != remote_tip:
reasons.append(
f"control checkout HEAD ({local_tip[:12]}) != live master tip "
f"({remote_tip[:12]})"
) )
if reasons: if reasons:
return _author_bootstrap_assessment( return {
not_applicable=False, "not_applicable": False,
allowed=False, "allowed": False,
block=True, "block": True,
reasons=reasons, "proven": False,
workspace=workspace, "reasons": reasons,
root=root, }
branch=branch or None,
dirty=dirty,
under_branches=False,
local_head_sha=local_tip,
remote_master_sha=remote_tip,
exact_next_action=EXACT_NEXT_ACTION_AUTHOR_BOOTSTRAP,
)
# Allowed: empty reasons so bootstrap_permits_control_checkout can pass. return {
return _author_bootstrap_assessment( "not_applicable": False,
not_applicable=False, "allowed": True,
allowed=True, "block": False,
block=False, "proven": True,
reasons=[], "bootstrap_path": "clean_canonical_control_checkout",
workspace=workspace, "reasons": [
root=root, "control checkout is clean on accepted base branch matching live master"
branch=branch or None, ],
dirty=dirty, }
under_branches=False,
bootstrap_path="clean_canonical_control_checkout",
local_head_sha=local_tip,
remote_master_sha=remote_tip,
exact_next_action=(
"Call gitea_bootstrap_author_issue_worktree with the allocated "
"issue/lease pins; it will create the branches/ worktree and lock."
),
)
import fcntl import fcntl
+6 -18
View File
@@ -241,14 +241,9 @@ def bootstrap_permits_control_checkout(
caller's ordinary block in force. caller's ordinary block in force.
``assessment`` is server-derived only: it is produced by ``assessment`` is server-derived only: it is produced by
:func:`assess_create_issue_bootstrap` or :func:`assess_create_issue_bootstrap` from inspected repository state. It is
:func:`author_issue_bootstrap.assess_author_issue_bootstrap` from inspected never accepted from an MCP tool argument, so no caller can assert
repository state. It is never accepted from an MCP tool argument, so no eligibility it has not proven.
caller can assert eligibility it has not proven.
#892: author issue worktree bootstrap uses the same predicate with
``task_scope='author_issue_bootstrap'`` so a clean control checkout can
create the first ``branches/`` worktree without the lock↔worktree cycle.
""" """
if not isinstance(assessment, dict): if not isinstance(assessment, dict):
return False return False
@@ -269,16 +264,9 @@ def bootstrap_permits_control_checkout(
if assessment.get("reasons"): if assessment.get("reasons"):
return False return False
# Scope proof: create_issue (#749) or author issue bootstrap (#850/#892), # Scope proof: only the create_issue bootstrap, only via the clean
# only via the clean canonical control checkout path. # canonical control checkout path.
task_scope = assessment.get("task_scope") if assessment.get("task_scope") != "create_issue_only":
if is_create_issue_task(task):
if task_scope != "create_issue_only":
return False
elif author_issue_bootstrap.is_author_issue_bootstrap_task(task):
if task_scope != "author_issue_bootstrap":
return False
else:
return False return False
if assessment.get("bootstrap_path") != "clean_canonical_control_checkout": if assessment.get("bootstrap_path") != "clean_canonical_control_checkout":
return False return False
+12
View File
@@ -153,7 +153,19 @@ 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).
+23
View File
@@ -86,3 +86,26 @@ 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.
## Connected vs Attached Tool Surface (#708)
MCP servers can report **Connected** at the CLI / host inventory layer while the **active LLM session exposes none of their tool namespaces**.
### Core principle
* **Connected status at host layer ≠ attached tools in active session.**
* Required preflight proof is **live tool visibility + `gitea_whoami` call** through the target namespace, not host `Connected` status alone.
* When servers report Connected but namespaces are absent from attached tools, classify as `mcp_connected_namespaces_missing`.
### Forbidden unsafe fallbacks
When `mcp_connected_namespaces_missing` is detected, workflows must **fail closed** and must **never** encourage or perform:
* direct imports of MCP server Python modules
* CLI or raw Gitea API mutations as a substitute for native tools
* profile hopping to another MCP profile/namespace to bypass the empty session
* session-state overrides or hand-edited session/ledger files
* process kills (`pkill`), config mtime touches, or `.env` edits
Only sanctioned recovery: **client reconnect path**, followed by full preflight (`whoami` → capability resolve → task).
+50 -1
View File
@@ -1169,10 +1169,57 @@ 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 only command/args and the two GITEA_MCP_* env vars — never a token Contains command/args and the GITEA_MCP_* / GITEA_CLIENT_MANAGED 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()
@@ -1183,11 +1230,13 @@ 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*.
+117 -4
View File
@@ -14585,6 +14585,56 @@ 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).
@@ -14601,6 +14651,10 @@ 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(
@@ -14634,6 +14688,10 @@ 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:
@@ -18081,6 +18139,9 @@ 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,
@@ -18097,6 +18158,9 @@ 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"),
@@ -18110,6 +18174,13 @@ def gitea_get_runtime_context(
PROJECT_ROOT), PROJECT_ROOT),
} }
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:
@@ -20567,7 +20638,9 @@ 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
running_profiles = {} all_profile_procs: dict[str, list[dict]] = {}
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:
@@ -20599,16 +20672,55 @@ 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
if profile not in running_profiles or start_time > running_profiles[profile]["start_time"]: proc_info = {
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.
@@ -20643,6 +20755,7 @@ 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,
+101
View File
@@ -51,6 +51,14 @@ EOF_PATTERNS = (
"eof", "eof",
) )
ERROR_CONNECTED_NAMESPACES_MISSING = "mcp_connected_namespaces_missing"
UNSAFE_FALLBACK_WARNING = (
"Workflow Safety Hard Stop (#708): Connected-but-namespaces-missing recovery must "
"NEVER use direct imports, Gitea API mutations, profile hopping, session-state "
"overrides, PID kills, or config mtime touches. Use client reconnect only."
)
SAFE_ENV_KEYS = ( SAFE_ENV_KEYS = (
"GITEA_MCP_PROFILE", "GITEA_MCP_PROFILE",
"GITEA_PROFILE_NAME", "GITEA_PROFILE_NAME",
@@ -60,6 +68,83 @@ SAFE_ENV_KEYS = (
) )
def assess_connected_namespace_attachment(
*,
connected_servers: list[str] | tuple[str, ...] | set[str] | None = None,
attached_session_namespaces: list[str] | tuple[str, ...] | set[str] | None = None,
required_namespaces: list[str] | tuple[str, ...] | set[str] | None = None,
) -> dict[str, Any]:
"""Assess whether host-connected MCP servers have attached tool namespaces in the active session (#708).
Addresses the Connected-but-namespaces-missing defect: CLI/host status may report Connected
while the active LLM session tool surface exposes 0 attached tool namespaces.
Returns structured telemetry and detection details.
"""
connected = [str(s).strip() for s in (connected_servers or []) if str(s).strip()]
attached = set(str(ns).strip() for ns in (attached_session_namespaces or []) if str(ns).strip())
req = [str(r).strip() for r in (required_namespaces or DEFAULT_NAMESPACES) if str(r).strip()]
proof: dict[str, dict[str, bool]] = {}
missing: list[str] = []
for s in connected:
is_attached = s in attached
proof[s] = {"connected": True, "attached": is_attached}
if not is_attached and s in req:
missing.append(s)
for r in req:
if r not in proof:
proof[r] = {"connected": r in connected, "attached": r in attached}
if r in connected and r not in attached and r not in missing:
missing.append(r)
attachment_healthy = len(missing) == 0 and len(connected) > 0
discovery_status = (
"namespaces_attached" if attachment_healthy
else ("connected_but_namespaces_missing" if len(connected) > 0 else "disconnected")
)
reasons: list[str] = []
remediation: list[str] = []
if missing:
reasons.append(
f"MCP server(s) {missing} report Connected at host/CLI layer but tool namespaces "
f"are missing from active session attached tools (Connected ≠ attached tools, #708)."
)
remediation.append(
"Reconnect the IDE/client MCP session to attach namespaces to the active session. "
"Do not use direct imports, CLI API mutations, profile hopping, or session file overrides."
)
elif not connected:
reasons.append("No MCP servers reported Connected.")
remediation.append("Start or reconnect Gitea MCP servers in client config.")
else:
reasons.append("All connected MCP server namespaces are attached to the active session.")
return {
"success": attachment_healthy,
"attachment_healthy": attachment_healthy,
"discovery_status": discovery_status,
"connected_servers": connected,
"attached_session_namespaces": list(attached),
"missing_namespaces": missing,
"proof_of_connected_vs_attached": proof,
"error_type": None if attachment_healthy else ERROR_CONNECTED_NAMESPACES_MISSING,
"reasons": reasons,
"remediation": remediation,
"exact_next_action": (
"Reconnect the IDE/client MCP session so tool namespaces attach to the active session. "
"Do not use direct imports, CLI API mutations, profile hopping, or session-state overrides."
if not attachment_healthy
else "None; session tool namespaces attached."
),
"unsafe_fallback_policy": UNSAFE_FALLBACK_WARNING,
}
def _as_list(value: Any) -> list[str] | None: def _as_list(value: Any) -> list[str] | None:
if value is None: if value is None:
return None return None
@@ -225,6 +310,16 @@ 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,
@@ -240,6 +335,9 @@ 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,
@@ -248,6 +346,9 @@ 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,
} }
+2
View File
@@ -44,6 +44,8 @@ def _reset_mutation_authority(monkeypatch):
]: ]:
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
+2 -2
View File
@@ -35,7 +35,7 @@ CONFIG = {
], ],
"forbidden_operations": [], "forbidden_operations": [],
"execution_profile": "full-author", "execution_profile": "full-author",
"allowed_repositories": ["Example-Org/Example-Repo"], "allowed_repositories": ["Scaled-Tech-Consulting/Gitea-Tools", "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": ["Example-Org/Example-Repo"], "allowed_repositories": ["Scaled-Tech-Consulting/Gitea-Tools", "Example-Org/Example-Repo"],
}, },
}, },
"rules": {"allow_runtime_switching": False}, "rules": {"allow_runtime_switching": False},
+1 -1
View File
@@ -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"}) self.assertEqual(set(entry["env"]), {"GITEA_MCP_CONFIG", "GITEA_MCP_PROFILE", "GITEA_CLIENT_MANAGED"})
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"):
@@ -0,0 +1,139 @@
"""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,69 @@
"""Unit regression tests for Issue #708: Connected-but-namespaces-missing detection and attachment safety."""
import mcp_namespace_health
def test_assess_connected_namespace_attachment_success():
connected = ["gitea-author", "gitea-reviewer", "gitea-merger"]
attached = ["gitea-author", "gitea-reviewer", "gitea-merger"]
res = mcp_namespace_health.assess_connected_namespace_attachment(
connected_servers=connected,
attached_session_namespaces=attached,
)
assert res["success"] is True
assert res["attachment_healthy"] is True
assert res["discovery_status"] == "namespaces_attached"
assert res["error_type"] is None
assert res["missing_namespaces"] == []
assert res["exact_next_action"] == "None; session tool namespaces attached."
def test_assess_connected_namespace_attachment_missing():
connected = ["gitea-author", "gitea-reviewer", "gitea-merger"]
attached = ["gitea-author"]
res = mcp_namespace_health.assess_connected_namespace_attachment(
connected_servers=connected,
attached_session_namespaces=attached,
)
assert res["success"] is False
assert res["attachment_healthy"] is False
assert res["discovery_status"] == "connected_but_namespaces_missing"
assert res["error_type"] == "mcp_connected_namespaces_missing"
assert "gitea-reviewer" in res["missing_namespaces"]
assert "gitea-merger" in res["missing_namespaces"]
assert "Reconnect the IDE/client MCP session" in res["exact_next_action"]
def test_assess_connected_namespace_attachment_disconnected():
res = mcp_namespace_health.assess_connected_namespace_attachment(
connected_servers=[],
attached_session_namespaces=[],
)
assert res["success"] is False
assert res["attachment_healthy"] is False
assert res["discovery_status"] == "disconnected"
def test_proof_of_connected_vs_attached_mapping():
connected = ["gitea-author", "gitea-reviewer"]
attached = ["gitea-author"]
res = mcp_namespace_health.assess_connected_namespace_attachment(
connected_servers=connected,
attached_session_namespaces=attached,
)
proof = res["proof_of_connected_vs_attached"]
assert proof["gitea-author"] == {"connected": True, "attached": True}
assert proof["gitea-reviewer"] == {"connected": True, "attached": False}
def test_unsafe_fallback_policy_enforcement():
res = mcp_namespace_health.assess_connected_namespace_attachment(
connected_servers=["gitea-author"],
attached_session_namespaces=[],
)
policy = res["unsafe_fallback_policy"]
assert "Workflow Safety Hard Stop (#708)" in policy
assert "direct imports" in policy
assert "API mutations" in policy
assert "profile hopping" in policy
assert "session-state overrides" in policy
@@ -1,215 +0,0 @@
"""Regression: author worktree bootstrap from clean control checkout (#892).
#892 is the four-door deadlock where every documented recovery path is closed:
bootstrap refuses control, lock demands an existing worktree, worktree-start
demands a lock, and shell worktree add is outside the sanctioned MCP path.
Root cause: assess_author_issue_bootstrap returned allowed/proven for a clean
control checkout, but bootstrap_permits_control_checkout only accepted
create_issue assessments (task_scope=create_issue_only + empty reasons + full
base-tip field set). Author assessments never satisfied the shared predicate,
so the #274/#604 guards kept the ordinary control-checkout block.
"""
from __future__ import annotations
import os
import tempfile
import unittest
from unittest import mock
import author_issue_bootstrap as aib
import create_issue_bootstrap as cib
CONTROL = "/repo/Gitea-Tools"
MASTER = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
OTHER = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
def _assess(
*,
workspace=CONTROL,
root=CONTROL,
branch="master",
head=MASTER,
porcelain="",
remote=MASTER,
remote_error=None,
task="bootstrap_author_issue_worktree",
):
return aib.assess_author_issue_bootstrap(
workspace_path=workspace,
canonical_repo_root=root,
current_branch=branch,
head_sha=head,
porcelain_status=porcelain,
remote_master_sha=remote,
remote_master_sha_error=remote_error,
task=task,
)
class TestAuthorBootstrapAssessmentShape(unittest.TestCase):
def test_clean_control_emits_predicate_compatible_fields(self):
assessment = _assess()
self.assertTrue(assessment["allowed"])
self.assertTrue(assessment["proven"])
self.assertFalse(assessment["block"])
self.assertFalse(assessment["not_applicable"])
self.assertEqual(assessment["reasons"], [])
self.assertEqual(assessment["task_scope"], "author_issue_bootstrap")
self.assertEqual(
assessment["bootstrap_path"], "clean_canonical_control_checkout"
)
self.assertEqual(assessment["dirty_files"], [])
self.assertIs(assessment["under_branches"], False)
self.assertTrue(assessment["base_tips_verified"])
self.assertEqual(assessment["local_head_sha"], MASTER)
self.assertEqual(assessment["remote_master_sha"], MASTER)
self.assertEqual(assessment["workspace_path"], os.path.realpath(CONTROL))
self.assertEqual(
assessment["canonical_repo_root"], os.path.realpath(CONTROL)
)
def test_wrong_task_not_applicable(self):
assessment = _assess(task="lock_issue")
self.assertTrue(assessment["not_applicable"])
self.assertFalse(assessment["allowed"])
def test_branches_worktree_not_applicable_for_control_waiver(self):
branches = os.path.join(CONTROL, "branches", "fix-issue-1")
assessment = _assess(workspace=branches)
self.assertTrue(assessment["not_applicable"])
self.assertFalse(assessment["allowed"])
self.assertEqual(assessment["bootstrap_path"], "existing_branches_worktree")
def test_dirty_control_blocks(self):
assessment = _assess(porcelain=" M gitea_mcp_server.py\n")
self.assertTrue(assessment["block"])
self.assertFalse(assessment["allowed"])
self.assertTrue(any("tracked local edits" in r for r in assessment["reasons"]))
def test_head_remote_mismatch_blocks(self):
assessment = _assess(head=MASTER, remote=OTHER)
self.assertTrue(assessment["block"])
self.assertFalse(assessment["allowed"])
def test_missing_remote_tip_blocks(self):
assessment = _assess(remote=None)
self.assertTrue(assessment["block"])
self.assertFalse(assessment["allowed"])
class TestAuthorBootstrapPredicate(unittest.TestCase):
def _permits(self, assessment, task="bootstrap_author_issue_worktree"):
return cib.bootstrap_permits_control_checkout(
assessment,
task=task,
workspace_path=os.path.realpath(CONTROL),
canonical_repo_root=os.path.realpath(CONTROL),
)
def test_clean_author_bootstrap_permits(self):
self.assertTrue(self._permits(_assess()))
def test_tool_alias_permits(self):
assessment = _assess(task="gitea_bootstrap_author_issue_worktree")
self.assertTrue(
self._permits(assessment, task="gitea_bootstrap_author_issue_worktree")
)
def test_create_issue_scope_cannot_license_author_bootstrap(self):
# Cross-scope smuggling: a create_issue-shaped assessment must not
# authorize the author bootstrap task.
create_shaped = dict(_assess())
create_shaped["task_scope"] = "create_issue_only"
self.assertFalse(self._permits(create_shaped))
def test_author_scope_cannot_license_create_issue(self):
assessment = _assess()
self.assertFalse(
cib.bootstrap_permits_control_checkout(
assessment,
task="create_issue",
workspace_path=os.path.realpath(CONTROL),
canonical_repo_root=os.path.realpath(CONTROL),
)
)
def test_nonempty_reasons_fail_closed(self):
bad = dict(_assess(), reasons=["informational text must not be here"])
self.assertFalse(self._permits(bad))
def test_dirty_fails_closed(self):
self.assertFalse(self._permits(_assess(porcelain=" M x.py\n")))
def test_mismatch_fails_closed(self):
self.assertFalse(self._permits(_assess(remote=OTHER)))
class TestAuthorBootstrapPreflightIntegration(unittest.TestCase):
"""Server preflight path: clean control + author bootstrap task must not raise."""
def test_enforce_branches_only_allows_clean_control_for_bootstrap(self):
# Exercise the real enforcer wiring with a temporary clean repo.
import gitea_mcp_server as srv
with tempfile.TemporaryDirectory() as tmp:
repo = os.path.join(tmp, "repo")
os.makedirs(os.path.join(repo, "branches"))
# Minimal git repo on master at a known tip.
import subprocess
subprocess.check_call(["git", "init", "-b", "master", repo])
subprocess.check_call(
["git", "-C", repo, "commit", "--allow-empty", "-m", "init"]
)
head = subprocess.check_output(
["git", "-C", repo, "rev-parse", "HEAD"], text=True
).strip()
assessment = aib.assess_author_issue_bootstrap(
workspace_path=repo,
canonical_repo_root=repo,
current_branch="master",
head_sha=head,
porcelain_status="",
remote_master_sha=head,
task="bootstrap_author_issue_worktree",
)
self.assertTrue(
cib.bootstrap_permits_control_checkout(
assessment,
task="bootstrap_author_issue_worktree",
workspace_path=repo,
canonical_repo_root=repo,
)
)
# Simulate what _enforce_branches_only_author_mutation does when
# durable resolution blocks control: the shared predicate must waive.
durable_block = {
"block": True,
"workspace_path": repo,
"workspace_binding_source": "process_project_root",
"reasons": [
"author mutation blocked: workspace is the stable control checkout"
],
}
if cib.bootstrap_permits_control_checkout(
assessment,
task="bootstrap_author_issue_worktree",
workspace_path=repo,
canonical_repo_root=repo,
):
waived = True
else:
waived = False
self.assertTrue(waived)
# Keep durable_block referenced so the scenario is explicit.
self.assertTrue(durable_block["block"])
if __name__ == "__main__":
unittest.main()
+3 -3
View File
@@ -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" mock_run_env12345.stdout = "GITEA_MCP_PROFILE=prgs-reconciler GITEA_CLIENT_MANAGED=1"
mock_run_env54321 = MagicMock() mock_run_env54321 = MagicMock()
mock_run_env54321.stdout = "GITEA_MCP_PROFILE=prgs-author" mock_run_env54321.stdout = "GITEA_MCP_PROFILE=prgs-author GITEA_CLIENT_MANAGED=1"
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" mock_run_env.stdout = "GITEA_MCP_PROFILE=prgs-author GITEA_CLIENT_MANAGED=1"
mock_run_git = MagicMock() mock_run_git = MagicMock()
mock_run_git.stdout = "FAKE2" # different SHA mock_run_git.stdout = "FAKE2" # different SHA
+2 -1
View File
@@ -243,9 +243,10 @@ 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): def test_activate_profile_succeeds_when_enabled(self, mock_auth, mock_api, mock_trusted):
self._write_config(CONFIG_SWITCHING_ENABLED) self._write_config(CONFIG_SWITCHING_ENABLED)
# Setup mock responses for whoami checks # Setup mock responses for whoami checks