Compare commits

..
Author SHA1 Message Date
jcwalker3 4ff4d2acc9 Merge branch 'master' into feat/issue-642-sanctioned-restart-controls 2026-07-24 02:07:44 -05:00
jcwalker3 fc8fe329d2 Merge branch 'master' into feat/issue-642-sanctioned-restart-controls 2026-07-23 22:22:36 -05:00
sysadminandClaude Opus 4.8 e593444eea feat(webui): sanctioned restart and graceful reload controls (Closes #642)
Adds a gated `system.restart_namespace` action so operators and workers can
restart or gracefully reload MCP namespaces through an authorized, audited
path instead of manual host process killing (#630).

- webui/sanctioned_restart.py: restart/reload operation model, dry-run
  intent preview, confirmation-string enforcement, audit emission, and
  post-restart health verification. Fails closed on unknown auth, missing
  capability, or ambiguous target namespace.
- webui/console_authz.py: RBAC entries for the restart capability with
  secret redaction preserved.
- webui/gated_actions.py: registers the restart action in the gated action
  framework so it cannot be invoked without capability + confirmation.
- task_capability_map.py: capability mapping for the restart operation.
- docs/sanctioned-restart-controls.md: operator documentation for the
  sanctioned path and the explicit prohibition on pkill recovery.
- docs/webui-authz-audit.md: audit model updated for restart events.
- tests/test_webui_sanctioned_restart.py: authorized preview, unauthorized
  deny, confirmation enforcement, contamination classification, and audit
  emission coverage.

No unrestricted kill path is exposed; manual pkill remains classified as
contamination and continues to block clean claims.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-23 23:19:21 -04:00
16 changed files with 1484 additions and 1823 deletions
-976
View File
@@ -1,976 +0,0 @@
"""Sanctioned author issue worktree bootstrap for allocated issues (#850).
Bootstraps an allocated author branch, canonical worktree under ``branches/``,
worktree registration, issue lock, and lease/assignment binding without
caller-side Git, Bash, or helper scripts.
Features:
1. Durable phase journal with read-after-write evidence for every phase.
2. Idempotent replay handling via idempotency keys.
3. Authoritative expected-base / concurrency-pin validation.
4. Typed stale-pin refusal without silent rebasing or repointing.
5. Canonical branches-root enforcement (path MUST be inside branches/).
6. Preexisting work preservation (dirty tracked/untracked check, foreign ownership refusal).
7. Compensating recovery limited strictly to artifacts created by this transition.
8. Satisfiable exact_next_action for MCP scheduled workers.
"""
from __future__ import annotations
import json
import os
import shutil
import subprocess
from typing import Any, Mapping
import author_mutation_worktree
import issue_lock_store
import issue_lock_worktree
import lease_lifecycle
from reviewer_worktree import parse_dirty_tracked_files
import task_capability_map
BOOTSTRAP_TASKS = frozenset(
{
"bootstrap_author_issue_worktree",
"gitea_bootstrap_author_issue_worktree",
}
)
PHASE_1_REQUEST_ACCEPTED = "1_request_accepted"
PHASE_2_BRANCH_CONFIRMED = "2_branch_confirmed"
PHASE_3_PATH_RESERVED = "3_path_reserved"
PHASE_4_WORKTREE_CONFIRMED = "4_worktree_confirmed"
PHASE_5_REGISTRATION_VERIFIED = "5_registration_verified"
PHASE_6_STATE_ESTABLISHED = "6_state_established"
PHASE_7_TRANSITION_COMPLETED = "7_transition_completed"
PHASE_COMPENSATING_RECOVERY = "compensating_recovery"
JOURNAL_DIR_NAME = "bootstrap-journals"
def is_author_issue_bootstrap_task(task: str | None) -> bool:
"""True when *task* is the author issue worktree bootstrap task."""
return (task or "").strip() in BOOTSTRAP_TASKS
def get_journal_dir(override: str | None = None) -> str:
"""Return the root directory for durable bootstrap phase journals."""
if override:
path = override
elif os.environ.get("GITEA_BOOTSTRAP_JOURNAL_DIR"):
path = os.environ["GITEA_BOOTSTRAP_JOURNAL_DIR"]
else:
cache_dir = os.path.expanduser("~/.cache/gitea-tools")
path = os.path.join(cache_dir, JOURNAL_DIR_NAME)
os.makedirs(path, exist_ok=True)
return path
def _journal_file_path(idempotency_key: str, journal_dir: str | None = None) -> str:
safe_key = "".join(
c if c.isalnum() or c in ("-", "_", ".") else "_"
for c in idempotency_key
)
return os.path.join(get_journal_dir(journal_dir), f"{safe_key}.json")
def load_phase_journal(
idempotency_key: str, journal_dir: str | None = None
) -> dict[str, Any] | None:
"""Load a durable phase journal if it exists."""
path = _journal_file_path(idempotency_key, journal_dir=journal_dir)
if not os.path.isfile(path):
return None
try:
with open(path, "r", encoding="utf-8") as f:
return json.load(f)
except Exception:
return None
def save_phase_journal(
journal: dict[str, Any], journal_dir: str | None = None
) -> None:
"""Persist a durable phase journal with write-through file sync."""
key = journal["idempotency_key"]
path = _journal_file_path(key, journal_dir=journal_dir)
tmp_path = f"{path}.tmp.{os.getpid()}"
with open(tmp_path, "w", encoding="utf-8") as f:
json.dump(journal, f, indent=2, sort_keys=True)
os.replace(tmp_path, path)
def derive_default_idempotency_key(
remote: str,
org: str | None,
repo: str | None,
issue_number: int,
assignment_id: str | None = None,
lease_id: str | None = None,
) -> str:
parts = [
"bootstrap",
(remote or "prgs").strip(),
(org or "Scaled-Tech-Consulting").strip(),
(repo or "Gitea-Tools").strip(),
f"issue-{issue_number}",
]
if assignment_id:
parts.append(assignment_id.strip())
if lease_id:
parts.append(lease_id.strip())
return ":".join(parts)
def run_compensating_recovery(
journal: dict[str, Any],
canonical_repo_root: str,
journal_dir: str | None = None,
) -> dict[str, Any]:
"""Execute compensating recovery for artifacts created by this transition only."""
artifacts = journal.get("artifacts_created") or {}
pending = journal.get("pending_creations") or {}
rolled_back: list[str] = []
worktree_path = journal.get("worktree_path")
branch_name = journal.get("branch_name")
# Roll back issue lock if created
if artifacts.get("lock_created") or pending.get("lock"):
issue_num = journal.get("issue_number")
session_id = journal.get("owner_session")
if issue_num and session_id:
try:
issue_lock_store.release_session_lock(
issue_number=issue_num,
session=session_id,
lock_dir=journal_dir,
)
rolled_back.append(f"lock:issue-{issue_num}")
except Exception:
pass
artifacts["lock_created"] = False
worktree_created = (
artifacts.get("worktree_registered")
or artifacts.get("worktree_dir_created")
or (pending.get("worktree_path") == worktree_path and worktree_path)
)
if worktree_created and worktree_path:
if os.path.exists(worktree_path):
# Re-verify cleanliness before destructive removal (F-4)
porc_res = subprocess.run(
["git", "-C", worktree_path, "status", "--porcelain"],
capture_output=True,
text=True,
check=False,
)
is_dirty = porc_res.returncode == 0 and bool(porc_res.stdout.strip())
if is_dirty:
rolled_back.append(f"worktree_path_preserved_dirty:{worktree_path}")
else:
try:
subprocess.run(
[
"git",
"-C",
canonical_repo_root,
"worktree",
"remove",
"--force",
worktree_path,
],
capture_output=True,
text=True,
check=False,
)
except Exception:
pass
if os.path.exists(worktree_path):
shutil.rmtree(worktree_path, ignore_errors=True)
try:
subprocess.run(
["git", "-C", canonical_repo_root, "worktree", "prune"],
capture_output=True,
text=True,
check=False,
)
except Exception:
pass
rolled_back.append(f"worktree_path:{worktree_path}")
else:
rolled_back.append(f"worktree_path:{worktree_path}")
branch_created = (
artifacts.get("branch_created")
or (pending.get("branch_name") == branch_name and branch_name)
)
if branch_created and branch_name:
try:
res = subprocess.run(
[
"git",
"-C",
canonical_repo_root,
"rev-parse",
"--verify",
branch_name,
],
capture_output=True,
text=True,
check=False,
)
if res.returncode == 0:
# Check for author commits on branch before branch deletion (F-4)
resolved_base = journal.get("resolved_base_sha") or "master"
rev_list_res = subprocess.run(
[
"git",
"-C",
canonical_repo_root,
"rev-list",
f"{resolved_base}..{branch_name}",
],
capture_output=True,
text=True,
check=False,
)
has_commits = rev_list_res.returncode == 0 and bool(rev_list_res.stdout.strip())
if has_commits:
rolled_back.append(f"branch_preserved_commits:{branch_name}")
else:
subprocess.run(
[
"git",
"-C",
canonical_repo_root,
"branch",
"-D",
branch_name,
],
capture_output=True,
text=True,
check=False,
)
rolled_back.append(f"branch:{branch_name}")
except Exception:
pass
recovery_info = {
"executed": True,
"rolled_back": rolled_back,
"reason": journal.get("failure_reason"),
}
journal["compensating_recovery"] = recovery_info
journal["current_phase"] = PHASE_COMPENSATING_RECOVERY
save_phase_journal(journal, journal_dir=journal_dir)
return recovery_info
def assess_author_issue_bootstrap(
*,
workspace_path: str,
canonical_repo_root: str,
current_branch: str | None = None,
head_sha: str | None = None,
porcelain_status: str = "",
remote_master_sha: str | None = None,
remote_master_sha_error: str | None = None,
task: str | None = None,
) -> dict[str, Any]:
"""Assess whether author issue worktree bootstrap may proceed from control or worktree root."""
root = os.path.realpath(canonical_repo_root or "")
workspace = os.path.realpath(workspace_path or root or ".")
branch = (current_branch or "").strip()
dirty = parse_dirty_tracked_files(porcelain_status or "")
under_branches = (
author_mutation_worktree.is_path_under_branches(workspace, root)
if root
else False
)
if not is_author_issue_bootstrap_task(task):
return {
"not_applicable": True,
"allowed": False,
"block": False,
"proven": False,
"reasons": ["task is not author_issue_bootstrap"],
}
if under_branches:
return {
"not_applicable": False,
"allowed": True,
"block": False,
"proven": True,
"bootstrap_path": "existing_branches_worktree",
"reasons": [
"workspace is already a registered worktree under branches/"
],
}
reasons: list[str] = []
if workspace != root:
reasons.append(
"bootstrap requires workspace to be canonical control checkout or branches/ worktree"
)
if branch not in author_mutation_worktree.BASE_BRANCHES:
reasons.append(
f"control checkout branch '{branch}' is not an accepted base branch "
f"({', '.join(sorted(author_mutation_worktree.BASE_BRANCHES))})"
)
if dirty:
reasons.append(
f"control checkout has tracked local edits: {', '.join(dirty[:5])}"
)
if remote_master_sha_error:
reasons.append(
f"could not verify live master tip: {remote_master_sha_error}"
)
elif remote_master_sha and head_sha:
h = head_sha.strip().lower()
rm = remote_master_sha.strip().lower()
if h != rm:
reasons.append(
f"control checkout HEAD ({h[:12]}) != live master tip ({rm[:12]})"
)
if reasons:
return {
"not_applicable": False,
"allowed": False,
"block": True,
"proven": False,
"reasons": reasons,
}
return {
"not_applicable": False,
"allowed": True,
"block": False,
"proven": True,
"bootstrap_path": "clean_canonical_control_checkout",
"reasons": [
"control checkout is clean on accepted base branch matching live master"
],
}
import fcntl
import stat
class BootstrapTransitionLock:
"""Inter-process file lock scoped to the transition identity / idempotency key."""
def __init__(self, idempotency_key: str, journal_dir: str | None = None):
safe_key = "".join(
c if c.isalnum() or c in ("-", "_", ".") else "_"
for c in idempotency_key
)
lock_dir = os.path.realpath(get_journal_dir(journal_dir))
if not os.path.isdir(lock_dir):
raise RuntimeError(f"Lock directory '{lock_dir}' does not exist or is not a directory")
raw_lock_path = os.path.abspath(os.path.join(lock_dir, f"{safe_key}.lock"))
try:
common = os.path.commonpath([lock_dir, os.path.dirname(raw_lock_path)])
except Exception:
common = None
if common != lock_dir:
raise RuntimeError(f"Lock path '{raw_lock_path}' escapes canonical lock directory '{lock_dir}'")
self.lock_path = raw_lock_path
self.fd = None
def __enter__(self):
if os.path.islink(self.lock_path):
raise RuntimeError(f"Refusing lock acquisition: lock path '{self.lock_path}' is a symlink")
flags = os.O_RDWR | os.O_CREAT
if hasattr(os, "O_NOFOLLOW"):
flags |= os.O_NOFOLLOW
if hasattr(os, "O_CLOEXEC"):
flags |= os.O_CLOEXEC
try:
fd = os.open(self.lock_path, flags, 0o600)
except OSError as exc:
raise RuntimeError(f"Failed to open lock file safely '{self.lock_path}': {exc}") from exc
st = os.fstat(fd)
if not stat.S_ISREG(st.st_mode):
os.close(fd)
raise RuntimeError(f"Lock target '{self.lock_path}' is not a regular file")
self.fd = fd
fcntl.flock(self.fd, fcntl.LOCK_EX)
return self
def __exit__(self, exc_type, exc_val, exc_tb):
if self.fd is not None:
try:
fcntl.flock(self.fd, fcntl.LOCK_UN)
except Exception:
pass
try:
os.close(self.fd)
except Exception:
pass
self.fd = None
def bootstrap_author_issue_worktree(
*,
issue_number: int,
canonical_repo_root: str,
assignment_id: str | None = None,
lease_id: str | None = None,
expected_base_sha: str | None = None,
branch_name: str | None = None,
worktree_path: str | None = None,
idempotency_key: str | None = None,
remote: str = "prgs",
host: str | None = None,
org: str | None = "Scaled-Tech-Consulting",
repo: str | None = "Gitea-Tools",
active_identity: str | None = "jcwalker3",
active_profile: str | None = "prgs-author",
owner_session: str | None = None,
lock_dir: str | None = None,
dry_run: bool = False,
) -> dict[str, Any]:
"""Execute the sanctioned author issue worktree bootstrap transition."""
root = os.path.realpath(canonical_repo_root)
session = (owner_session or "").strip()
if not session:
return {
"success": False,
"reason_code": "missing_owner_session",
"message": "Missing required owner_session parameter (fail closed). Session identifier cannot be fabricated or defaulted.",
"exact_next_action": (
"Pass explicit owner_session resolved from gitea_whoami or session context."
),
}
if active_identity is None or not str(active_identity).strip():
return {
"success": False,
"reason_code": "missing_active_identity",
"message": "Missing required active_identity parameter (fail closed). Identity cannot be fabricated or defaulted.",
"exact_next_action": "Pass explicit active_identity resolved from gitea_whoami.",
}
identity = str(active_identity).strip()
if active_profile is None or not str(active_profile).strip():
return {
"success": False,
"reason_code": "missing_active_profile",
"message": "Missing required active_profile parameter (fail closed). Profile cannot be fabricated or defaulted.",
"exact_next_action": "Pass explicit active_profile resolved from gitea_whoami.",
}
profile = str(active_profile).strip()
# Derive standard inputs
expected_pattern = f"issue-{issue_number}"
target_branch = (branch_name or "").strip()
if not target_branch:
target_branch = f"fix/issue-{issue_number}-native-mcp-bootstrap"
elif expected_pattern not in target_branch:
return {
"success": False,
"reason_code": "invalid_branch_name",
"message": (
f"Branch name '{target_branch}' must contain issue pattern '{expected_pattern}'"
),
"exact_next_action": (
f"Supply a branch_name containing '{expected_pattern}', e.g., 'fix/issue-{issue_number}-...'"
),
}
worktree_name = target_branch.replace("/", "-")
target_worktree = (worktree_path or "").strip()
if not target_worktree:
target_worktree = os.path.join(root, "branches", worktree_name)
target_worktree = os.path.realpath(os.path.abspath(target_worktree))
key = (idempotency_key or "").strip()
if not key:
key = derive_default_idempotency_key(
remote=remote,
org=org,
repo=repo,
issue_number=issue_number,
assignment_id=assignment_id,
lease_id=lease_id,
)
# Acquire cross-process file lock scoped to the idempotency key / transition identity
with BootstrapTransitionLock(key, journal_dir=lock_dir):
# Idempotency check
existing = load_phase_journal(key, journal_dir=lock_dir)
if existing and existing.get("completed"):
if (
existing.get("issue_number") == issue_number
and existing.get("branch_name") == target_branch
and os.path.realpath(existing.get("worktree_path", ""))
== target_worktree
):
return {
"success": True,
"replayed": True,
"message": (
f"Idempotent replay: worktree for issue #{issue_number} already bootstrapped at {target_worktree}"
),
"issue_number": issue_number,
"branch_name": target_branch,
"worktree_path": target_worktree,
"base_sha": existing.get("resolved_base_sha"),
"lease_id": existing.get("lease_id"),
"assignment_id": existing.get("assignment_id"),
"idempotency_key": key,
"phase_journal": existing,
"exact_next_action": (
"Call gitea_whoami, then gitea_resolve_task_capability(task='work_issue') "
"and proceed with author implementation in the bootstrapped worktree."
),
}
else:
return {
"success": False,
"reason_code": "incompatible_idempotency_replay",
"message": (
f"Idempotency key '{key}' already exists with incompatible parameters "
f"(stored: {existing.get('branch_name')}, {existing.get('worktree_path')}; "
f"requested: {target_branch}, {target_worktree})"
),
"exact_next_action": (
"Supply a unique idempotency_key or pass compatible parameters."
),
}
# Initialize or resume Phase Journal
if existing:
journal = existing
artifacts = journal.setdefault("artifacts_created", {})
artifacts.setdefault("branch_created", False)
artifacts.setdefault("worktree_dir_created", False)
artifacts.setdefault("worktree_registered", False)
artifacts.setdefault("lock_created", False)
else:
journal = {
"idempotency_key": key,
"issue_number": issue_number,
"assignment_id": assignment_id,
"lease_id": lease_id,
"expected_base_sha": expected_base_sha,
"resolved_base_sha": None,
"branch_name": target_branch,
"worktree_path": target_worktree,
"active_identity": identity,
"active_profile": profile,
"owner_session": session,
"remote": remote,
"org": org,
"repo": repo,
"phases": {},
"artifacts_created": {
"branch_created": False,
"worktree_dir_created": False,
"worktree_registered": False,
"lock_created": False,
},
"current_phase": PHASE_1_REQUEST_ACCEPTED,
"completed": False,
}
# Fetch current live master SHA
try:
rev_res = subprocess.run(
["git", "-C", root, "rev-parse", "HEAD"],
capture_output=True,
text=True,
check=True,
)
live_master_sha = rev_res.stdout.strip()
except Exception as exc:
return {
"success": False,
"reason_code": "git_rev_parse_failed",
"message": f"Could not determine repository HEAD: {exc}",
"exact_next_action": "Verify repository git state and retry.",
}
# Phase 1: REQUEST_ACCEPTED & Concurrency Pin Check
if expected_base_sha:
exp_norm = expected_base_sha.strip().lower()
live_norm = live_master_sha.lower()
if exp_norm != live_norm:
journal["failure_reason"] = (
f"stale concurrency pin: expected {exp_norm[:12]} != live {live_norm[:12]}"
)
save_phase_journal(journal, journal_dir=lock_dir)
return {
"success": False,
"reason_code": "stale_concurrency_pin",
"message": (
f"Expected base SHA {exp_norm[:12]} does not match live master SHA {live_norm[:12]} (fail closed)."
),
"expected_base_sha": expected_base_sha,
"live_master_sha": live_master_sha,
"exact_next_action": (
"Re-evaluate assignment against current live master SHA and retry with updated expected_base_sha."
),
}
journal["resolved_base_sha"] = live_master_sha
journal["phases"][PHASE_1_REQUEST_ACCEPTED] = {
"status": "completed",
"live_master_sha": live_master_sha,
"expected_base_sha": expected_base_sha,
}
journal["current_phase"] = PHASE_2_BRANCH_CONFIRMED
save_phase_journal(journal, journal_dir=lock_dir)
if dry_run:
return {
"success": True,
"dry_run": True,
"message": f"Dry-run: validated bootstrap intent for issue #{issue_number}",
"issue_number": issue_number,
"branch_name": target_branch,
"worktree_path": target_worktree,
"base_sha": live_master_sha,
"phase_journal": journal,
"exact_next_action": "Run without dry_run=True to execute bootstrap.",
}
# Phase 2: BRANCH_CONFIRMED
was_branch_created_previously = journal["artifacts_created"].get("branch_created", False)
pending_branch = (journal.get("pending_creations") or {}).get("branch_name")
branch_check = subprocess.run(
["git", "-C", root, "rev-parse", "--verify", target_branch],
capture_output=True,
text=True,
check=False,
)
if branch_check.returncode == 0:
branch_head = branch_check.stdout.strip()
# Verify branch head descends from base
anc_check = subprocess.run(
[
"git",
"-C",
root,
"merge-base",
"--is-ancestor",
live_master_sha,
branch_head,
],
capture_output=True,
text=True,
check=False,
)
if anc_check.returncode != 0 and branch_head.lower() != live_master_sha.lower():
# F-8: Check if branch shares a common ancestor with live master
mb_check = subprocess.run(
["git", "-C", root, "merge-base", live_master_sha, branch_head],
capture_output=True,
text=True,
check=False,
)
if mb_check.returncode != 0 or not mb_check.stdout.strip():
journal["failure_reason"] = (
f"existing branch '{target_branch}' HEAD ({branch_head[:12]}) is incompatible with live master ({live_master_sha[:12]})"
)
save_phase_journal(journal, journal_dir=lock_dir)
return {
"success": False,
"reason_code": "incompatible_existing_branch",
"message": (
f"Existing branch '{target_branch}' HEAD ({branch_head[:12]}) is incompatible with live master ({live_master_sha[:12]})."
),
"exact_next_action": (
"Inspect or sync the existing branch with master before bootstrapping."
),
}
# Preserve creation provenance monotonically across interruption and replay
if was_branch_created_previously or pending_branch == target_branch:
journal["artifacts_created"]["branch_created"] = True
else:
journal["artifacts_created"]["branch_created"] = False
else:
# Persist creation intent/provenance to disk BEFORE executing external mutation
journal.setdefault("pending_creations", {})["branch_name"] = target_branch
journal["artifacts_created"]["branch_created"] = True
save_phase_journal(journal, journal_dir=lock_dir)
# Create branch
create_res = subprocess.run(
["git", "-C", root, "branch", target_branch, live_master_sha],
capture_output=True,
text=True,
check=False,
)
if create_res.returncode != 0:
journal["artifacts_created"]["branch_created"] = False
journal.get("pending_creations", {}).pop("branch_name", None)
journal["failure_reason"] = (
f"failed to create git branch '{target_branch}': {create_res.stderr.strip()}"
)
save_phase_journal(journal, journal_dir=lock_dir)
return {
"success": False,
"reason_code": "branch_creation_failed",
"message": f"Failed to create git branch '{target_branch}': {create_res.stderr.strip()}",
"exact_next_action": "Verify branch availability and retry.",
}
journal["phases"][PHASE_2_BRANCH_CONFIRMED] = {
"status": "completed",
"branch_name": target_branch,
"created": journal["artifacts_created"]["branch_created"],
}
journal["current_phase"] = PHASE_3_PATH_RESERVED
save_phase_journal(journal, journal_dir=lock_dir)
# Phase 3: PATH_RESERVED & Phase 4: WORKTREE_CONFIRMED
if not author_mutation_worktree.is_path_under_branches(
target_worktree, root
):
journal["failure_reason"] = (
f"target_worktree '{target_worktree}' is outside canonical branches/ root"
)
run_compensating_recovery(journal, root, journal_dir=lock_dir)
return {
"success": False,
"reason_code": "path_outside_canonical_branches_root",
"message": (
f"Worktree path '{target_worktree}' is outside canonical branches/ root (fail closed)."
),
"exact_next_action": (
"Provide a worktree_path inside canonical branches/ root, e.g., 'branches/issue-...'"
),
}
was_dir_created_previously = journal["artifacts_created"].get("worktree_dir_created", False)
was_registered_previously = journal["artifacts_created"].get("worktree_registered", False)
pending_wt = (journal.get("pending_creations") or {}).get("worktree_path")
dir_exists = os.path.exists(target_worktree)
if dir_exists:
# Check porcelain directly
porc_res = subprocess.run(
["git", "-C", target_worktree, "status", "--porcelain"],
capture_output=True,
text=True,
check=False,
)
dirty = (
parse_dirty_tracked_files(porc_res.stdout)
if porc_res.returncode == 0
else []
)
if dirty or (porc_res.returncode == 0 and porc_res.stdout.strip()):
journal["failure_reason"] = (
f"target worktree '{target_worktree}' contains dirty tracked/untracked files"
)
run_compensating_recovery(journal, root, journal_dir=lock_dir)
return {
"success": False,
"reason_code": "preexisting_dirty_worktree",
"message": (
f"Preexisting worktree '{target_worktree}' has dirty tracked/untracked files (fail closed)."
),
"exact_next_action": (
"Clean or stash the pre-existing worktree files before bootstrapping."
),
}
# Check registered branch
wt_state = issue_lock_worktree.read_worktree_git_state(target_worktree)
wt_branch = (wt_state.get("current_branch") or "").strip()
if wt_branch and wt_branch != target_branch:
journal["failure_reason"] = (
f"existing worktree '{target_worktree}' is on branch '{wt_branch}' != expected '{target_branch}'"
)
run_compensating_recovery(journal, root, journal_dir=lock_dir)
return {
"success": False,
"reason_code": "incompatible_existing_directory",
"message": (
f"Existing worktree '{target_worktree}' is registered to branch '{wt_branch}' instead of '{target_branch}'."
),
"exact_next_action": (
"Inspect or remove the pre-existing worktree folder before bootstrapping."
),
}
if was_dir_created_previously or pending_wt == target_worktree:
journal["artifacts_created"]["worktree_dir_created"] = True
journal["artifacts_created"]["worktree_registered"] = True
else:
journal["artifacts_created"]["worktree_dir_created"] = False
journal["artifacts_created"]["worktree_registered"] = False
else:
# Persist creation intent/provenance to disk BEFORE executing external worktree add mutation
journal.setdefault("pending_creations", {})["worktree_path"] = target_worktree
journal["artifacts_created"]["worktree_dir_created"] = True
journal["artifacts_created"]["worktree_registered"] = True
save_phase_journal(journal, journal_dir=lock_dir)
wt_add_res = subprocess.run(
[
"git",
"-C",
root,
"worktree",
"add",
target_worktree,
target_branch,
],
capture_output=True,
text=True,
check=False,
)
if wt_add_res.returncode != 0:
journal["artifacts_created"]["worktree_dir_created"] = False
journal["artifacts_created"]["worktree_registered"] = False
journal.get("pending_creations", {}).pop("worktree_path", None)
journal["failure_reason"] = (
f"git worktree add failed: {wt_add_res.stderr.strip()}"
)
run_compensating_recovery(journal, root, journal_dir=lock_dir)
return {
"success": False,
"reason_code": "worktree_add_failed",
"message": f"Failed to execute git worktree add: {wt_add_res.stderr.strip()}",
"exact_next_action": "Verify git worktree capabilities and retry.",
}
journal["phases"][PHASE_3_PATH_RESERVED] = {
"status": "completed",
"worktree_path": target_worktree,
"preexisting_dir": dir_exists,
}
journal["current_phase"] = PHASE_4_WORKTREE_CONFIRMED
save_phase_journal(journal, journal_dir=lock_dir)
# Phase 5: REGISTRATION_VERIFIED
wt_list_res = subprocess.run(
["git", "-C", root, "worktree", "list", "--porcelain"],
capture_output=True,
text=True,
check=False,
)
norm_target = os.path.realpath(target_worktree)
found_registration = False
if wt_list_res.returncode == 0:
for block in wt_list_res.stdout.split("\n\n"):
lines = block.strip().splitlines()
worktree_line = next(
(l[9:].strip() for l in lines if l.startswith("worktree ")),
None,
)
if worktree_line and os.path.realpath(worktree_line) == norm_target:
found_registration = True
break
if not found_registration:
journal["failure_reason"] = (
f"worktree registration for '{target_worktree}' not found in git worktree list"
)
run_compensating_recovery(journal, root, journal_dir=lock_dir)
return {
"success": False,
"reason_code": "worktree_registration_verification_failed",
"message": f"Worktree '{target_worktree}' registration verification failed.",
"exact_next_action": "Check git worktree list integrity and retry.",
}
journal["phases"][PHASE_4_WORKTREE_CONFIRMED] = {
"status": "completed",
"worktree_path": target_worktree,
}
journal["phases"][PHASE_5_REGISTRATION_VERIFIED] = {
"status": "completed",
"registered": True,
}
journal["current_phase"] = PHASE_6_STATE_ESTABLISHED
save_phase_journal(journal, journal_dir=lock_dir)
# Phase 6: STATE_ESTABLISHED — Issue Lock Acquisition
from datetime import datetime, timezone
try:
lock_data = {
"remote": remote,
"org": org or "Scaled-Tech-Consulting",
"repo": repo or "Gitea-Tools",
"issue_number": issue_number,
"branch": target_branch,
"branch_name": target_branch,
"worktree_path": target_worktree,
"owner_session": session,
"claimant": {
"username": identity,
"profile": profile,
},
"assignment_id": assignment_id,
"lease_id": lease_id,
"expected_base_sha": live_master_sha,
"created_at": datetime.now(timezone.utc).isoformat(),
}
journal.setdefault("pending_creations", {})["lock"] = True
journal["artifacts_created"]["lock_created"] = True
save_phase_journal(journal, journal_dir=lock_dir)
lock_res = issue_lock_store.bind_session_lock(lock_data, lock_dir=lock_dir)
except Exception as exc:
journal["artifacts_created"]["lock_created"] = False
journal.get("pending_creations", {}).pop("lock", None)
journal["failure_reason"] = f"issue lock binding failed: {exc}"
run_compensating_recovery(journal, root, journal_dir=lock_dir)
return {
"success": False,
"reason_code": "issue_lock_acquisition_failed",
"message": f"Could not bind canonical issue lock for issue #{issue_number}: {exc}",
"exact_next_action": "Verify lease/assignment state and retry.",
}
journal["phases"][PHASE_6_STATE_ESTABLISHED] = {
"status": "completed",
"lock": lock_res,
}
journal["phases"][PHASE_7_TRANSITION_COMPLETED] = {
"status": "completed",
}
journal["current_phase"] = PHASE_7_TRANSITION_COMPLETED
journal["completed"] = True
save_phase_journal(journal, journal_dir=lock_dir)
return {
"success": True,
"replayed": False,
"message": (
f"Successfully bootstrapped author issue worktree for issue #{issue_number} "
f"at branch '{target_branch}' and worktree '{target_worktree}'."
),
"issue_number": issue_number,
"branch_name": target_branch,
"worktree_path": target_worktree,
"base_sha": live_master_sha,
"lease_id": lease_id,
"assignment_id": assignment_id,
"idempotency_key": key,
"lock_state": lock_res,
"phase_journal": journal,
"exact_next_action": (
"Call gitea_whoami, then gitea_resolve_task_capability(task='work_issue') "
"and proceed with author implementation in the bootstrapped worktree."
),
}
+36 -57
View File
@@ -40,67 +40,23 @@ def _normalize_path(path: str) -> str:
return (path or "").replace("\\", "/").rstrip("/") return (path or "").replace("\\", "/").rstrip("/")
def get_canonical_branches_root(project_root: str | None = None) -> str:
"""Return the absolute path of the canonical branches directory for *project_root*."""
root = os.path.realpath(project_root) if project_root else os.path.realpath(os.getcwd())
canonical_repo_root = resolve_canonical_repo_root(root, root)
return os.path.realpath(os.path.join(canonical_repo_root, "branches"))
def is_path_under_branches(path: str, project_root: str | None = None) -> bool: def is_path_under_branches(path: str, project_root: str | None = None) -> bool:
"""True when *path* resolves inside a canonical ``branches/`` directory.""" """True when *path* resolves inside ``<project_root>/branches/``."""
if not path or not str(path).strip(): normalized = _normalize_path(path)
if not normalized:
return False return False
try: if "/branches/" in f"{normalized}/":
real_path = os.path.realpath(os.path.abspath(str(path).strip())) return True
except Exception: if normalized.endswith("/branches"):
return True
if project_root:
root = _normalize_path(os.path.realpath(project_root))
real = _normalize_path(os.path.realpath(path))
if real.startswith(f"{root}/"):
rel = real[len(root) + 1 :]
return rel == "branches" or rel.startswith("branches/")
return False return False
branches_root = get_canonical_branches_root(project_root or real_path)
try:
common = os.path.commonpath([branches_root, real_path])
except Exception:
return False
if common != branches_root:
return False
rel = os.path.relpath(real_path, branches_root)
return rel != "." and not rel.startswith("..")
def resolve_canonical_repo_root(workspace_path: str, fallback_project_root: str) -> str:
"""Return the stable repository root for *workspace_path* via git metadata (#460)."""
p = (workspace_path or "").strip()
if p:
try:
res = subprocess.run(
["git", "-C", p, "rev-parse", "--git-common-dir"],
capture_output=True,
text=True,
check=True,
)
common = _realpath_git_common_dir(p, res.stdout)
if common.endswith(f"{os.sep}.git") or os.path.basename(common) == ".git":
candidate_root = os.path.dirname(common)
real_p = os.path.realpath(p)
try:
if os.path.commonpath([candidate_root, real_p]) == candidate_root:
return candidate_root
except Exception:
pass
except Exception:
pass
fallback = os.path.realpath(fallback_project_root or workspace_path or ".")
norm = fallback.replace("\\", "/")
if "/branches/" in norm:
return os.path.realpath(norm.split("/branches/")[0])
elif norm.endswith("/branches"):
return os.path.realpath(os.path.dirname(fallback))
return fallback
def resolve_mutation_workspace( def resolve_mutation_workspace(
worktree_path: str | None, worktree_path: str | None,
@@ -131,6 +87,29 @@ def _realpath_git_common_dir(workspace_path: str, common_dir: str) -> str:
return os.path.realpath(os.path.join(workspace_path, raw)) return os.path.realpath(os.path.join(workspace_path, raw))
def resolve_canonical_repo_root(workspace_path: str, fallback_project_root: str) -> str:
"""Return the stable repository root for *workspace_path* via git metadata (#460)."""
path = (workspace_path or "").strip()
fallback = os.path.realpath(fallback_project_root)
if not path:
return fallback
try:
res = subprocess.run(
["git", "-C", path, "rev-parse", "--git-common-dir"],
capture_output=True,
text=True,
check=True,
)
common = _realpath_git_common_dir(path, res.stdout)
except Exception:
return fallback
if common.endswith(f"{os.sep}.git"):
return os.path.dirname(common)
if os.path.basename(common) == ".git":
return os.path.dirname(common)
return fallback
def resolve_author_mutation_context( def resolve_author_mutation_context(
worktree_path: str | None, worktree_path: str | None,
process_project_root: str, process_project_root: str,
+1 -2
View File
@@ -247,8 +247,7 @@ def bootstrap_permits_control_checkout(
""" """
if not isinstance(assessment, dict): if not isinstance(assessment, dict):
return False return False
import author_issue_bootstrap if not is_create_issue_task(task):
if not is_create_issue_task(task) and not author_issue_bootstrap.is_author_issue_bootstrap_task(task):
return False return False
# Positive proof: the assessment must affirmatively allow, with no # Positive proof: the assessment must affirmatively allow, with no
-1
View File
@@ -69,7 +69,6 @@ that gates each call, not which tools exist.
- `gitea_audit_worktree_cleanup` - `gitea_audit_worktree_cleanup`
- `gitea_authorize_reconciliation_cleanup_phase` - `gitea_authorize_reconciliation_cleanup_phase`
- `gitea_authorize_review_correction` - `gitea_authorize_review_correction`
- `gitea_bootstrap_author_issue_worktree`
- `gitea_capability_stop_terminal_report` - `gitea_capability_stop_terminal_report`
- `gitea_capture_branches_worktree_snapshot` - `gitea_capture_branches_worktree_snapshot`
- `gitea_check_pr_eligibility` - `gitea_check_pr_eligibility`
+122
View File
@@ -0,0 +1,122 @@
# Sanctioned restart and graceful reload controls (#642)
Sessions used to recover MCP connectivity by killing the host daemon
(`pkill -f mcp_server.py`, #630). That is forbidden and stays forbidden: it
kills every namespace on the host, contaminates whichever session survives, and
leaves no audit trail. This document describes the sanctioned replacement,
implemented in `webui/sanctioned_restart.py`.
## What the console will and will not do
The console **never** restarts anything. It authorizes an intent, records it,
and hands off to a host supervisor. There is no code path in which the console
sends a signal, spawns a process, or renders a kill command — a regression test
asserts the module contains no `subprocess`, `signal`, `os.kill`, `os.system`,
or `popen` reference, and that no returned payload contains a kill command.
## Operations
| Mode | Action | Minimum role | Behaviour |
|------|--------|--------------|-----------|
| `reload` | `system.reload_namespace` | controller | Host supervisor reloads the namespace in place, draining in-flight requests. |
| `restart` | `system.restart_namespace` | admin | Host supervisor restarts the namespace. In-flight requests are lost. |
Scope is always exactly one namespace. A fleet-wide restart is an explicit
non-goal: `all`, `*`, `fleet`, and an empty scope are refused with
`fleet_scope_not_permitted`, because that is precisely the blast radius the
forbidden kill already had. An unrecognised namespace is refused rather than
passed through to the host.
## The gate sequence
`assess_restart_request()` applies every gate in order and reports the first
failure with a stable reason code:
| Order | Gate | Reason code on failure |
|-------|------|------------------------|
| 1 | Mode is `restart` or `reload` | `unknown_mode` |
| 2 | Scope is a single known namespace | `fleet_scope_not_permitted`, `unknown_namespace` |
| 3 | Principal holds the required console role | `unauthorized` |
| 4 | Confirmation phrase supplied | `confirmation_required` |
| 5 | Confirmation names this namespace and mode | `confirmation_mismatch` |
| 6 | Out-of-band operator authorization present | `operator_authorization_missing` |
| 7 | Runtime is not contaminated | `contaminated_runtime` |
| 8 | Host restart hook configured | `restart_hook_not_configured` |
Passing every gate yields `host_action_required`, never "restarted".
### Confirmation binds the namespace
The required phrase is `"<mode> <namespace>"` — for example
`restart gitea-author`. Binding the namespace into the phrase is the point: a
confirmation typed for one namespace cannot be replayed against another.
### Operator authorization is not self-assertable
Host daemon maintenance is authorized out of band through
`GITEA_OPERATOR_DAEMON_MAINTENANCE_AUTHORIZATION`, read from the process
environment and nowhere else (#630; #710 finding F1). A worker session cannot
set an environment variable for an already-running daemon, so this cannot be
faked the way a tool argument could.
### The host hook
`GITEA_SANCTIONED_RESTART_HOOK` holds an opaque reference the *host* resolves —
a supervisor label such as a launchd job name, never a command line. With no
hook configured the request is refused; the console does not fall back to a
process kill. The value is read server-side and never rendered to a client.
## Manual kill remains contamination
`classify_restart_command()` classifies an operator-proposed recovery command.
A manual `pkill`/`kill`/`killall` of the MCP daemon is contamination, not a
restart: it returns `clean_claim_allowed: false` and builds a durable
contamination marker (redacted command only, never secrets) naming
`system.restart_namespace` as the sanctioned alternative.
A live, uncleared contamination marker also blocks a restart. This is stricter
than #630's task-scoped gate, which deliberately lets a contaminated worker keep
commenting and handing off: restarting a contaminated runtime would launder the
contamination rather than resolve it. Clear the marker through the reconciler
path first.
## Post-restart health verification
After the host supervisor acts, `verify_post_restart_health()` decides whether
the session may claim to be clean:
| Status | Meaning | Clean claim |
|--------|---------|-------------|
| `clean` | Required tool callable, proven through the live client namespace | Allowed |
| `unproven` | Reported healthy without live client-namespace evidence | Refused |
| `unhealthy` | Probe failed | Refused |
Only `probe_source=client_namespace` evidence clears a session. Static tool
registration is not proof, and neither is an offline subprocess probe — an IDE
client can hold a registered tool list while live calls fail with
`client is closing: EOF` (see
[`mcp-namespace-health.md`](mcp-namespace-health.md)).
## Audit
Every attempt — allowed or denied — is recorded through
`webui.console_audit` with actor, target namespace, mode, result, and reason
code, and is redacted before it is persisted. `system.restart_namespace` is
break-glass, so its records are retained for 730 days. Records carry
`process_kill_executed: false`, which is a fact about the code path rather than
a claim: no such path exists.
## Environment variables
| Variable | Purpose |
|----------|---------|
| `GITEA_SANCTIONED_RESTART_HOOK` | Host supervisor reference; absent means restart is refused. |
| `GITEA_OPERATOR_DAEMON_MAINTENANCE_AUTHORIZATION` | Out-of-band operator authorization reference. |
| `WEBUI_AUDIT_LOG` | Console audit sink; absent means records are built but not persisted. |
## Non-goals
* No unrestricted `kill` from the UI, in any role, in any phase.
* No fleet-wide restart.
* No silent auto-restart loop: every attempt is confirmed and audited.
* This does not implement the Phase 1 health API (#634).
+11 -2
View File
@@ -91,6 +91,8 @@ already define, and a regression test asserts each mapping matches.
| `close_pr` | controller | privileged | `gitea.pr.close` | Yes | No | No | 3 | | `close_pr` | controller | privileged | `gitea.pr.close` | Yes | No | No | 3 |
| `merge_pr` | controller | privileged | `gitea.pr.merge` | Yes | **Yes** | **Yes** | 3 | | `merge_pr` | controller | privileged | `gitea.pr.merge` | Yes | **Yes** | **Yes** | 3 |
| `delete_branch` | admin | destructive | `gitea.branch.delete` | Yes | **Yes** | **Yes** | 3 | | `delete_branch` | admin | destructive | `gitea.branch.delete` | Yes | **Yes** | **Yes** | 3 |
| `system.reload_namespace` | controller | privileged | `runtime.reload_namespace` | Yes | No | No | 2 |
| `system.restart_namespace` | admin | destructive | `runtime.restart_namespace` | Yes | **Yes** | **Yes** | 2 |
**Dual control** means the acting principal may not be the sole authority: a **Dual control** means the acting principal may not be the sole authority: a
second distinct principal must confirm. **Break-glass** means the action is second distinct principal must confirm. **Break-glass** means the action is
@@ -102,6 +104,13 @@ honouring it.
`delete_branch` is admin-only rather than controller because it is the one `delete_branch` is admin-only rather than controller because it is the one
irreversible action in the set. irreversible action in the set.
`system.restart_namespace` is admin-only for the same reason: restarting a
namespace drops every in-flight request on it. `system.reload_namespace` drains
first, so it is privileged but not destructive. Neither action is ever executed
by the console — both hand off to a host supervisor, and neither exposes a raw
process kill. See
[`sanctioned-restart-controls.md`](sanctioned-restart-controls.md) (#642).
### Authorization decision ### Authorization decision
`authorize(action_id, principal, for_execution=False)` returns a decision `authorize(action_id, principal, for_execution=False)` returns a decision
@@ -199,8 +208,8 @@ breaking the request it describes.
| Class | Applies to | Default | | Class | Applies to | Default |
|-------|-----------|---------| |-------|-----------|---------|
| `standard` | Routine gated writes | 90 days | | `standard` | Routine gated writes | 90 days |
| `privileged` | `review_pr`, `close_pr`, and any unclassifiable action | 365 days | | `privileged` | `review_pr`, `close_pr`, `system.reload_namespace`, and any unclassifiable action | 365 days |
| `break_glass` | `merge_pr`, `delete_branch` | 730 days | | `break_glass` | `merge_pr`, `delete_branch`, `system.restart_namespace` | 730 days |
Each record carries its own class, day count, and computed `expires_at`, so Each record carries its own class, day count, and computed `expires_at`, so
retention is auditable per record rather than inferred from file age. An retention is auditable per record rather than inferred from file age. An
+69 -196
View File
@@ -977,32 +977,6 @@ def _create_issue_bootstrap_assessment(
""" """
import create_issue_bootstrap as _cib import create_issue_bootstrap as _cib
import author_issue_bootstrap as _aib
if _aib.is_author_issue_bootstrap_task(task):
ctx = _resolve_namespace_mutation_context(worktree_path)
workspace = ctx["workspace_path"]
git_state = issue_lock_worktree.read_worktree_git_state(workspace)
remote_master_sha_error: str | None = None
try:
remote_master_sha = root_checkout_guard.resolve_remote_master_sha(
ctx["canonical_repo_root"]
)
except Exception as exc:
remote_master_sha = None
remote_master_sha_error = (
f"{type(exc).__name__}: {exc}".strip() or "resolver failed"
)
return _aib.assess_author_issue_bootstrap(
workspace_path=workspace,
canonical_repo_root=ctx["canonical_repo_root"],
current_branch=git_state.get("current_branch"),
head_sha=git_state.get("head_sha"),
porcelain_status=git_state.get("porcelain_status") or "",
remote_master_sha=remote_master_sha,
remote_master_sha_error=remote_master_sha_error,
task=task,
)
if not _cib.is_create_issue_task(task): if not _cib.is_create_issue_task(task):
return None return None
@@ -1495,7 +1469,7 @@ def verify_preflight_purity(
dirty_files = sorted( dirty_files = sorted(
_parse_porcelain_entries(_get_workspace_porcelain(workspace)) _parse_porcelain_entries(_get_workspace_porcelain(workspace))
) )
if dirty_files and task != "commit_files": if dirty_files:
raise RuntimeError( raise RuntimeError(
nwb.format_namespace_workspace_binding_error( nwb.format_namespace_workspace_binding_error(
role_kind=role, role_kind=role,
@@ -9313,7 +9287,6 @@ def gitea_commit_files(
host: str | None = None, host: str | None = None,
org: str | None = None, org: str | None = None,
repo: str | None = None, repo: str | None = None,
worktree_path: str | None = None,
) -> dict: ) -> dict:
"""Commit changes to multiple files in a Gitea repository in a single atomic commit. """Commit changes to multiple files in a Gitea repository in a single atomic commit.
@@ -9326,46 +9299,10 @@ def gitea_commit_files(
host: Override the Gitea host. host: Override the Gitea host.
org: Override the owner/organization. org: Override the owner/organization.
repo: Override the repository name. repo: Override the repository name.
worktree_path: Optional worktree path for author mutation context.
Returns: Returns:
dict with success status and commit/branch information. dict with success status and commit/branch information.
""" """
if worktree_path is None:
lock_data = issue_lock_store.read_session_issue_lock() or {}
worktree_path = lock_data.get("worktree_path")
if not worktree_path:
try:
prof = get_profile()
uname = prof.get("username") or prof.get("profile_name")
for path in issue_lock_store.iter_lock_files():
lk = issue_lock_store.read_lock_file(path) or {}
claimant = lk.get("claimant") or {}
if lk.get("remote") == remote and (claimant.get("username") == uname or lk.get("profile") == prof.get("profile_name")):
issue_lock_store.bind_session_lock(lk, renewal_sanctioned=True)
worktree_path = lk.get("worktree_path")
break
except Exception:
pass
if worktree_path is None and files:
for f in files:
p = f.get("workspace_path") or f.get("local_path") or ""
if p and os.path.isabs(p):
real_p = os.path.realpath(p)
real_root = os.path.realpath(PROJECT_ROOT)
branches_dir = os.path.join(real_root, "branches")
if real_p.startswith(branches_dir + os.sep):
rel_sub = os.path.relpath(real_p, branches_dir)
wt_folder = rel_sub.split(os.sep)[0]
if wt_folder and wt_folder != "..":
worktree_path = os.path.join(branches_dir, wt_folder)
break
if worktree_path:
os.environ["GITEA_AUTHOR_WORKTREE"] = worktree_path
os.environ["GITEA_ACTIVE_WORKTREE"] = worktree_path
ok, block_reasons = role_session_router.check_author_mutation_after_reviewer_stop( ok, block_reasons = role_session_router.check_author_mutation_after_reviewer_stop(
"commit_files" "commit_files"
) )
@@ -9378,7 +9315,7 @@ def gitea_commit_files(
"reasons": block_reasons, "reasons": block_reasons,
} }
blocked = _namespace_mutation_block( blocked = _namespace_mutation_block(
"commit_files", commit="", branch="", remote=remote, worktree_path=worktree_path "commit_files", commit="", branch="", remote=remote
) )
if blocked: if blocked:
return blocked return blocked
@@ -9406,7 +9343,7 @@ def gitea_commit_files(
) )
# #735: forward explicit org/repo into shared anti-stomp preflight. # #735: forward explicit org/repo into shared anti-stomp preflight.
verify_preflight_purity(remote=remote, worktree_path=worktree_path, task="commit_files", org=org, repo=repo) verify_preflight_purity(remote, task="commit_files", org=org, repo=repo)
processed_files, source_proofs = _prepare_commit_payload_files(files) processed_files, source_proofs = _prepare_commit_payload_files(files)
h, o, r = _resolve(remote, host, org, repo) h, o, r = _resolve(remote, host, org, repo)
@@ -9683,96 +9620,6 @@ def gitea_publish_unpublished_issue_branch(
} }
@mcp.tool()
def gitea_bootstrap_author_issue_worktree(
issue_number: int,
assignment_id: str | None = None,
lease_id: str | None = None,
expected_base_sha: str | None = None,
branch_name: str | None = None,
worktree_path: str | None = None,
idempotency_key: str | None = None,
remote: str = "dadeschools",
host: str | None = None,
org: str | None = None,
repo: str | None = None,
dry_run: bool = False,
) -> dict:
"""Bootstrap an allocated author issue branch and registered worktree (#850).
Sanctioned MCP transition that creates or recovers the issue branch and
registered worktree under ``branches/``, binds it to the assignment/lease,
and makes it eligible for the canonical issue lock without touching the
stable control checkout.
Args:
issue_number: Allocated issue number to bootstrap.
assignment_id: Optional allocation assignment ID.
lease_id: Optional workflow lease ID.
expected_base_sha: Authoritative expected base SHA / concurrency pin.
branch_name: Optional custom branch name (must match issue-<N> pattern).
worktree_path: Optional custom worktree path under branches/.
idempotency_key: Optional key for idempotent replay/resume.
remote: Known instance 'dadeschools' or 'prgs'.
host: Override Gitea host.
org: Override Org.
repo: Override Repo.
dry_run: Report planned transition without mutating repository.
"""
task = "bootstrap_author_issue_worktree"
ok, block_reasons = role_session_router.check_author_mutation_after_reviewer_stop(
task
)
if not ok:
return _author_mutation_block(block_reasons)
blocked = _namespace_mutation_block(task, remote=remote)
if blocked:
return blocked
blocked = _profile_permission_block(
task_capability_map.required_permission(task),
remote=remote,
host=host,
org=org,
repo=repo,
org_explicit=org is not None,
repo_explicit=repo is not None,
)
if blocked:
return blocked
verify_preflight_purity(
remote,
task=task,
org=org,
repo=repo,
)
h, o, r = _resolve(remote, host, org, repo)
canonical_root = _canonical_local_git_root()
import author_issue_bootstrap
return author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=issue_number,
canonical_repo_root=canonical_root,
assignment_id=assignment_id,
lease_id=lease_id,
expected_base_sha=expected_base_sha,
branch_name=branch_name,
worktree_path=worktree_path,
idempotency_key=idempotency_key,
remote=remote,
host=h,
org=o,
repo=r,
active_identity=_active_username(),
active_profile=_active_profile_name(),
owner_session=_current_session_id(),
dry_run=dry_run,
)
# Merge methods supported by the Gitea merge API. # Merge methods supported by the Gitea merge API.
_MERGE_METHODS = ("merge", "squash", "rebase") _MERGE_METHODS = ("merge", "squash", "rebase")
@@ -11672,25 +11519,22 @@ def gitea_reconcile_merged_cleanups(
if dry_run: if dry_run:
report["dry_run"] = True report["dry_run"] = True
report["executed"] = False report["executed"] = False
# #851: surface planned lifecycle order so dry-run matches execute.
report["planned_execution_orders"] = {
str(entry.get("pr_number")): entry.get("planned_execution_order") or []
for entry in (report.get("entries") or [])
}
return {"success": True, "performed": False, **report} return {"success": True, "performed": False, **report}
verify_preflight_purity( verify_preflight_purity(
remote, task="reconcile_merged_cleanups", org=org, repo=repo remote, task="reconcile_merged_cleanups", org=org, repo=repo
) )
actions: list[dict] = [] actions: list[dict] = []
for entry in report.get("entries") or []: project_root = _canonical_local_git_root()
head_branch = entry.get("head_branch") or ""
remote_assessment = entry.get("remote_branch") or {}
local_assessment = entry.get("local_worktree") or {}
if remote_assessment.get("safe_to_delete_remote"): def _ownership_records_for_branch(
import urllib.parse head_branch: str, pr_num_int: int | None
) -> list[dict]:
pr_num = entry.get("pr_number")
try:
pr_num_int = int(pr_num) if pr_num is not None else None
except (TypeError, ValueError):
pr_num_int = None
ownership_bundle = _collect_branch_ownership_records( ownership_bundle = _collect_branch_ownership_records(
remote=remote, remote=remote,
host=h, host=h,
@@ -11698,7 +11542,7 @@ def gitea_reconcile_merged_cleanups(
repo=r, repo=r,
branch=head_branch, branch=head_branch,
pr_number=pr_num_int, pr_number=pr_num_int,
project_root=_canonical_local_git_root(), project_root=project_root,
auth=auth, auth=auth,
base_api=base, base_api=base,
) )
@@ -11719,6 +11563,18 @@ def gitea_reconcile_merged_cleanups(
"role": "inventory", "role": "inventory",
} }
) )
return ownership_records
def _attempt_owned_remote_delete(
*,
head_branch: str,
pr_num_int: int | None,
after_worktree_removal: bool = False,
) -> dict:
"""Fail-closed remote delete with live ownership reassessment (#851)."""
import urllib.parse
ownership_records = _ownership_records_for_branch(head_branch, pr_num_int)
ownership = branch_cleanup_guard.assess_active_branch_ownership( ownership = branch_cleanup_guard.assess_active_branch_ownership(
remote=remote, remote=remote,
org=o, org=o,
@@ -11728,8 +11584,7 @@ def gitea_reconcile_merged_cleanups(
records=ownership_records, records=ownership_records,
) )
if ownership.get("block"): if ownership.get("block"):
actions.append( return {
{
"action": "delete_remote_branch", "action": "delete_remote_branch",
"branch": head_branch, "branch": head_branch,
"success": False, "success": False,
@@ -11738,13 +11593,10 @@ def gitea_reconcile_merged_cleanups(
"verified_absent": False, "verified_absent": False,
"blocker_kind": "active_branch_ownership", "blocker_kind": "active_branch_ownership",
"reasons": ownership.get("reasons") or [], "reasons": ownership.get("reasons") or [],
"blocking_categories": ownership.get( "blocking_categories": ownership.get("blocking_categories") or [],
"blocking_categories" "after_worktree_removal": after_worktree_removal,
) "ownership_reassessed": after_worktree_removal,
or [],
} }
)
continue
encoded = urllib.parse.quote(head_branch, safe="") encoded = urllib.parse.quote(head_branch, safe="")
url = f"{base}/branches/{encoded}" url = f"{base}/branches/{encoded}"
@@ -11759,6 +11611,7 @@ def gitea_reconcile_merged_cleanups(
"branch": head_branch, "branch": head_branch,
"source": "reconcile_merged_cleanups", "source": "reconcile_merged_cleanups",
"ownership_checked": True, "ownership_checked": True,
"after_worktree_removal": after_worktree_removal,
}, },
): ):
api_request("DELETE", url, auth) api_request("DELETE", url, auth)
@@ -11767,8 +11620,7 @@ def gitea_reconcile_merged_cleanups(
readback readback
) )
verified = bool(readback_assessment.get("verified_absent")) verified = bool(readback_assessment.get("verified_absent"))
actions.append( return {
{
"action": "delete_remote_branch", "action": "delete_remote_branch",
"branch": head_branch, "branch": head_branch,
"success": bool(readback_assessment.get("ok")), "success": bool(readback_assessment.get("ok")),
@@ -11777,22 +11629,53 @@ def gitea_reconcile_merged_cleanups(
"verified_absent": verified, "verified_absent": verified,
"readback": readback_assessment.get("readback"), "readback": readback_assessment.get("readback"),
"reasons": readback_assessment.get("reasons") or [], "reasons": readback_assessment.get("reasons") or [],
"after_worktree_removal": after_worktree_removal,
"ownership_reassessed": after_worktree_removal,
} }
)
for entry in report.get("entries") or []:
head_branch = entry.get("head_branch") or ""
remote_assessment = entry.get("remote_branch") or {}
local_assessment = entry.get("local_worktree") or {}
pr_num = entry.get("pr_number")
try:
pr_num_int = int(pr_num) if pr_num is not None else None
except (TypeError, ValueError):
pr_num_int = None
# #851 lifecycle: when the target worktree is independently safe, remove
# it first so worktree_binding ownership does not permanently strand
# both the worktree and the remote branch. Never skip worktree removal
# merely because remote delete would be blocked by that binding.
# Ownership protection for remote delete remains fail-closed below.
worktree_removed = False
if local_assessment.get("safe_to_remove_worktree"): if local_assessment.get("safe_to_remove_worktree"):
result = merged_cleanup_reconcile.remove_local_worktree( result = merged_cleanup_reconcile.remove_local_worktree(
_canonical_local_git_root(), project_root,
head_branch, head_branch,
worktree_path=local_assessment.get("worktree_path"), worktree_path=local_assessment.get("worktree_path"),
) )
actions.append({"action": "remove_local_worktree", **result}) actions.append({"action": "remove_local_worktree", **result})
# Idempotent resume: absent worktree is already gone.
msg = (result.get("message") or "").lower()
worktree_removed = bool(result.get("success")) or (
"not found" in msg
)
if remote_assessment.get("safe_to_delete_remote"):
actions.append(
_attempt_owned_remote_delete(
head_branch=head_branch,
pr_num_int=pr_num_int,
after_worktree_removal=worktree_removed,
)
)
for scratch in report.get("reviewer_scratch_entries") or []: for scratch in report.get("reviewer_scratch_entries") or []:
if not scratch.get("safe_to_remove_worktree"): if not scratch.get("safe_to_remove_worktree"):
continue continue
result = merged_cleanup_reconcile.remove_reviewer_scratch_worktree( result = merged_cleanup_reconcile.remove_reviewer_scratch_worktree(
_canonical_local_git_root(), scratch.get("worktree_path") or "" project_root, scratch.get("worktree_path") or ""
) )
actions.append({"action": "remove_reviewer_scratch_worktree", **result}) actions.append({"action": "remove_reviewer_scratch_worktree", **result})
@@ -19705,12 +19588,6 @@ def gitea_resolve_task_capability(
task: str, task: str,
remote: str = "dadeschools", remote: str = "dadeschools",
host: str | None = None, host: str | None = None,
org: str | None = None,
repo: str | None = None,
issue_number: int | None = None,
worktree_path: str | None = None,
pr_number: int | None = None,
**kwargs: Any,
) -> dict: ) -> dict:
"""Read-only / side-effect free: resolve capability, profile, and namespace for a task. """Read-only / side-effect free: resolve capability, profile, and namespace for a task.
@@ -19727,14 +19604,13 @@ def gitea_resolve_task_capability(
remote: Known remote instance name. remote: Known remote instance name.
host: Optional override for the Gitea host. host: Optional override for the Gitea host.
""" """
task_key = task_capability_map._canonical_preflight_task(task)
TASK_MAP = task_capability_map.TASK_CAPABILITY_MAP TASK_MAP = task_capability_map.TASK_CAPABILITY_MAP
# Every fresh attempt invalidates the previous task/role stamp before any # Every fresh attempt invalidates the previous task/role stamp before any
# fallible resolver work. Unknown/malformed tasks and unexpected failures # fallible resolver work. Unknown/malformed tasks and unexpected failures
# therefore remain fail-closed instead of preserving stale authority. # therefore remain fail-closed instead of preserving stale authority.
_clear_resolved_capability_stamp() _clear_resolved_capability_stamp()
if task_key not in TASK_MAP: if task not in TASK_MAP:
# #723: structured fail-closed unknown_task (never raise into internal_error). # #723: structured fail-closed unknown_task (never raise into internal_error).
profile = get_profile() profile = get_profile()
h = host or (REMOTES.get(remote, {}).get("host") if remote in REMOTES else None) h = host or (REMOTES.get(remote, {}).get("host") if remote in REMOTES else None)
@@ -19780,8 +19656,8 @@ def gitea_resolve_task_capability(
result["cleared_stale_denial"] = True result["cleared_stale_denial"] = True
return result return result
required_permission = task_capability_map.required_permission(task_key) required_permission = task_capability_map.required_permission(task)
required_role = task_capability_map.required_role(task_key) required_role = task_capability_map.required_role(task)
role_exclusive_tasks = task_capability_map.ROLE_EXCLUSIVE_TASKS role_exclusive_tasks = task_capability_map.ROLE_EXCLUSIVE_TASKS
infra_assessment = role_session_router.assess_infra_stop(PROJECT_ROOT) infra_assessment = role_session_router.assess_infra_stop(PROJECT_ROOT)
@@ -19967,10 +19843,7 @@ def gitea_resolve_task_capability(
available_in_session = allowed_in_current_session available_in_session = allowed_in_current_session
runtime_stale_blocker = False runtime_stale_blocker = False
if ( if "PYTEST_CURRENT_TEST" not in os.environ or "GITEA_FORCE_MCP_RUNTIME_CHECK" in os.environ:
"PYTEST_CURRENT_TEST" not in os.environ
or "GITEA_FORCE_MCP_RUNTIME_CHECK" in os.environ
) and os.environ.get("GITEA_ALLOW_STALE_RUNTIME") != "1":
runtime_reasons = _check_mcp_runtimes_diagnostics(task, matching_profiles) runtime_reasons = _check_mcp_runtimes_diagnostics(task, matching_profiles)
if runtime_reasons: if runtime_reasons:
restart_required = True restart_required = True
-2
View File
@@ -73,8 +73,6 @@ AUTHOR_TASKS = frozenset({
"claim_issue", "claim_issue",
"create_branch", "create_branch",
"push_branch", "push_branch",
"bootstrap_author_issue_worktree",
"gitea_bootstrap_author_issue_worktree",
"create_pr", "create_pr",
"comment_pr", "comment_pr",
"address_pr_change_requests", "address_pr_change_requests",
+15 -13
View File
@@ -69,14 +69,6 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
"permission": "gitea.branch.create", "permission": "gitea.branch.create",
"role": "author", "role": "author",
}, },
"bootstrap_author_issue_worktree": {
"permission": "gitea.branch.create",
"role": "author",
},
"gitea_bootstrap_author_issue_worktree": {
"permission": "gitea.branch.create",
"role": "author",
},
"push_branch": { "push_branch": {
"permission": "gitea.branch.push", "permission": "gitea.branch.push",
"role": "author", "role": "author",
@@ -354,6 +346,21 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
"role": "controller", "role": "controller",
}, },
# #642: sanctioned host-daemon lifecycle controls. Deliberately *not* a
# ``gitea.*`` operation — restarting an MCP namespace is a host action, not
# a Gitea API call, and no configured Gitea profile should be able to
# satisfy it by accident. Authority comes from the console RBAC model plus
# out-of-band operator authorization (#630); these entries exist so the
# console cannot invent an authority the capability layer never declared.
"restart_namespace": {
"permission": "runtime.restart_namespace",
"role": "controller",
},
"reload_namespace": {
"permission": "runtime.reload_namespace",
"role": "controller",
},
# #601 first-class lease lifecycle — inspect/list need read; mutations gate on # #601 first-class lease lifecycle — inspect/list need read; mutations gate on
# ownership in the control-plane DB (not a separate Gitea write permission). # ownership in the control-plane DB (not a separate Gitea write permission).
"list_workflow_leases": { "list_workflow_leases": {
@@ -496,8 +503,6 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
# merger lease (#763). # merger lease (#763).
_PREFLIGHT_TASK_TRANSITIONS = frozenset({ _PREFLIGHT_TASK_TRANSITIONS = frozenset({
("review_pr", "acquire_reviewer_pr_lease"), ("review_pr", "acquire_reviewer_pr_lease"),
("work_issue", "bootstrap_author_issue_worktree"),
("bootstrap_author_issue_worktree", "lock_issue"),
}) })
@@ -544,8 +549,6 @@ ROLE_EXCLUSIVE_TASKS: frozenset[str] = frozenset(
"gitea_release_merger_pr_lease", "gitea_release_merger_pr_lease",
"create_branch", "create_branch",
"push_branch", "push_branch",
"bootstrap_author_issue_worktree",
"gitea_bootstrap_author_issue_worktree",
"publish_unpublished_branch", "publish_unpublished_branch",
"create_pr", "create_pr",
"commit_files", "commit_files",
@@ -571,7 +574,6 @@ ISSUE_MUTATION_TOOL_TASKS: dict[str, str] = {
"gitea_set_issue_labels": "set_issue_labels", "gitea_set_issue_labels": "set_issue_labels",
"gitea_cleanup_terminal_pr_labels": "cleanup_terminal_pr_labels", "gitea_cleanup_terminal_pr_labels": "cleanup_terminal_pr_labels",
"gitea_create_label": "create_label", "gitea_create_label": "create_label",
"gitea_bootstrap_author_issue_worktree": "bootstrap_author_issue_worktree",
"gitea_commit_files": "commit_files", "gitea_commit_files": "commit_files",
} }
-483
View File
@@ -1,483 +0,0 @@
"""Regression test suite for native author issue worktree bootstrap (#850)."""
from __future__ import annotations
import json
import os
import shutil
import subprocess
import tempfile
import unittest
from unittest import mock
import author_issue_bootstrap
import task_capability_map
def _concurrent_bootstrap_worker(args: tuple[str, int, str, str, str, str]) -> dict:
repo_dir, issue_num, key, lock_dir, journal_dir, master_sha = args
os.environ["GITEA_BOOTSTRAP_JOURNAL_DIR"] = journal_dir
return author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=issue_num,
canonical_repo_root=repo_dir,
expected_base_sha=master_sha,
idempotency_key=key,
lock_dir=lock_dir,
owner_session="session-concurrent-test",
active_identity="jcwalker3",
active_profile="prgs-author",
)
class TestAuthorIssueBootstrap(unittest.TestCase):
"""Test suite covering AC1-AC10 and comment #14959 specification."""
def setUp(self):
self.tmp_dir = tempfile.mkdtemp(prefix="test_bootstrap_")
self.repo_dir = os.path.join(self.tmp_dir, "repo")
os.makedirs(self.repo_dir)
# Initialize synthetic git repo
subprocess.run(["git", "init", "-b", "master"], cwd=self.repo_dir, check=True, capture_output=True)
subprocess.run(["git", "config", "user.name", "Test User"], cwd=self.repo_dir, check=True)
subprocess.run(["git", "config", "user.email", "[email protected]"], cwd=self.repo_dir, check=True)
readme = os.path.join(self.repo_dir, "README.md")
with open(readme, "w", encoding="utf-8") as f:
f.write("# Test Repo\n")
subprocess.run(["git", "add", "README.md"], cwd=self.repo_dir, check=True, capture_output=True)
subprocess.run(["git", "commit", "-m", "initial commit"], cwd=self.repo_dir, check=True, capture_output=True)
rev_res = subprocess.run(["git", "rev-parse", "HEAD"], cwd=self.repo_dir, capture_output=True, text=True, check=True)
self.master_sha = rev_res.stdout.strip()
self.branches_dir = os.path.join(self.repo_dir, "branches")
os.makedirs(self.branches_dir, exist_ok=True)
self.lock_dir = os.path.join(self.tmp_dir, "locks")
os.makedirs(self.lock_dir, exist_ok=True)
self.journal_dir = os.path.join(self.tmp_dir, "journals")
os.makedirs(self.journal_dir, exist_ok=True)
os.environ["GITEA_BOOTSTRAP_JOURNAL_DIR"] = self.journal_dir
def tearDown(self):
os.environ.pop("GITEA_BOOTSTRAP_JOURNAL_DIR", None)
shutil.rmtree(self.tmp_dir, ignore_errors=True)
def test_bootstrap_success_path(self):
"""AC1/AC3/AC8: Successful bootstrap creates branch, worktree, registration, and lock proof."""
key = "test_key_success_1"
res = author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=850,
canonical_repo_root=self.repo_dir,
assignment_id="asn-12345",
lease_id="lease-67890",
expected_base_sha=self.master_sha,
idempotency_key=key,
remote="prgs",
lock_dir=self.lock_dir,
owner_session="session-test-1234",
)
self.assertTrue(res.get("success"), f"Bootstrap failed: {res}")
self.assertFalse(res.get("replayed"))
self.assertEqual(res.get("issue_number"), 850)
self.assertEqual(res.get("base_sha"), self.master_sha)
self.assertIn("branches/fix-issue-850-native-mcp-bootstrap", res.get("worktree_path"))
# Verify worktree directory exists and is registered
worktree_path = res["worktree_path"]
self.assertTrue(os.path.isdir(worktree_path))
wt_list = subprocess.run(["git", "-C", self.repo_dir, "worktree", "list"], capture_output=True, text=True, check=True)
self.assertIn(worktree_path, wt_list.stdout)
# Verify phase journal written
journal = author_issue_bootstrap.load_phase_journal(key, journal_dir=self.lock_dir)
self.assertIsNotNone(journal)
self.assertTrue(journal.get("completed"))
self.assertEqual(journal.get("current_phase"), author_issue_bootstrap.PHASE_7_TRANSITION_COMPLETED)
def test_idempotent_replay(self):
"""Item 2: Replaying with identical key returns cached transition without duplicate creation."""
key = "test_key_idempotent_1"
res1 = author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=850,
canonical_repo_root=self.repo_dir,
idempotency_key=key,
lock_dir=self.lock_dir,
owner_session="session-test-1234",
)
self.assertTrue(res1["success"], f"res1 failed: {res1}")
self.assertFalse(res1.get("replayed"))
# Second call
res2 = author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=850,
canonical_repo_root=self.repo_dir,
idempotency_key=key,
lock_dir=self.lock_dir,
owner_session="session-test-1234",
)
self.assertTrue(res2["success"], f"res2 failed: {res2}")
self.assertTrue(res2.get("replayed"))
self.assertEqual(res1["worktree_path"], res2["worktree_path"])
def test_stale_concurrency_pin_refusal(self):
"""Item 3: Mismatched expected base SHA fails closed without silent rebasing."""
stale_sha = "0000000000000000000000000000000000000000"
res = author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=850,
canonical_repo_root=self.repo_dir,
expected_base_sha=stale_sha,
lock_dir=self.lock_dir,
owner_session="session-test-1234",
)
self.assertFalse(res["success"])
self.assertEqual(res.get("reason_code"), "stale_concurrency_pin")
self.assertIn("exact_next_action", res)
def test_path_outside_branches_root_refusal(self):
"""Item 6: Worktree path outside branches/ root is refused."""
outside_path = os.path.join(self.tmp_dir, "outside_worktree")
res = author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=850,
canonical_repo_root=self.repo_dir,
worktree_path=outside_path,
lock_dir=self.lock_dir,
owner_session="session-test-1234",
)
self.assertFalse(res["success"])
self.assertEqual(res.get("reason_code"), "path_outside_canonical_branches_root")
def test_preexisting_dirty_worktree_preservation(self):
"""Item 6: Preexisting dirty worktree fails closed and is NOT modified or cleaned."""
branch = "fix/issue-850-dirty-test"
wt_path = os.path.join(self.branches_dir, "fix-issue-850-dirty-test")
subprocess.run(["git", "-C", self.repo_dir, "worktree", "add", "-b", branch, wt_path], check=True, capture_output=True)
# Create dirty untracked file
dirty_file = os.path.join(wt_path, "dirty.txt")
with open(dirty_file, "w") as f:
f.write("dirty edits\n")
res = author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=850,
canonical_repo_root=self.repo_dir,
branch_name=branch,
worktree_path=wt_path,
lock_dir=self.lock_dir,
owner_session="session-test-1234",
)
self.assertFalse(res["success"])
self.assertEqual(res.get("reason_code"), "preexisting_dirty_worktree")
# Prove dirty file is preserved byte-for-byte
self.assertTrue(os.path.exists(dirty_file))
with open(dirty_file, "r") as f:
self.assertEqual(f.read(), "dirty edits\n")
def test_compensating_recovery_on_failed_phase(self):
"""AC4/Item 4: Failure during transition rolls back ONLY newly created artifacts."""
key = "test_key_recovery_1"
# Simulate partial progress in journal
journal = {
"idempotency_key": key,
"issue_number": 850,
"branch_name": "fix/issue-850-recovery-test",
"worktree_path": os.path.join(self.branches_dir, "fix-issue-850-recovery-test"),
"artifacts_created": {
"branch_created": True,
"worktree_dir_created": True,
"worktree_registered": True,
"lock_created": False,
},
"failure_reason": "simulated lock failure",
"current_phase": author_issue_bootstrap.PHASE_5_REGISTRATION_VERIFIED,
"completed": False,
}
# Create the branch and worktree manually to simulate partial state
subprocess.run(["git", "-C", self.repo_dir, "branch", journal["branch_name"]], check=True, capture_output=True)
subprocess.run(["git", "-C", self.repo_dir, "worktree", "add", journal["worktree_path"], journal["branch_name"]], check=True, capture_output=True)
# Run compensating recovery
rec = author_issue_bootstrap.run_compensating_recovery(journal, self.repo_dir)
self.assertTrue(rec["executed"])
self.assertIn(f"worktree_path:{journal['worktree_path']}", rec["rolled_back"])
self.assertIn(f"branch:{journal['branch_name']}", rec["rolled_back"])
# Prove worktree directory and branch were rolled back
self.assertFalse(os.path.exists(journal["worktree_path"]))
branch_check = subprocess.run(["git", "-C", self.repo_dir, "rev-parse", "--verify", journal["branch_name"]], capture_output=True, text=True, check=False)
self.assertNotEqual(branch_check.returncode, 0)
def test_cross_process_concurrency(self):
"""Review #525 Finding 1: Genuine cross-process concurrency locking prevents corruption."""
import concurrent.futures
key = "test_concurrent_key_850"
args = (self.repo_dir, 850, key, self.lock_dir, self.journal_dir, self.master_sha)
with concurrent.futures.ProcessPoolExecutor(max_workers=2) as executor:
fut1 = executor.submit(_concurrent_bootstrap_worker, args)
fut2 = executor.submit(_concurrent_bootstrap_worker, args)
res1 = fut1.result(timeout=10)
res2 = fut2.result(timeout=10)
self.assertTrue(res1["success"], f"res1 failed: {res1}")
self.assertTrue(res2["success"], f"res2 failed: {res2}")
# One process performs creation, the other process receives idempotent replay
replayed_count = sum(1 for r in (res1, res2) if r.get("replayed"))
created_count = sum(1 for r in (res1, res2) if not r.get("replayed"))
self.assertEqual(replayed_count, 1)
self.assertEqual(created_count, 1)
self.assertEqual(res1["worktree_path"], res2["worktree_path"])
def test_interrupted_replay_preserves_artifacts_created_provenance(self):
"""Review #525 Finding 2: Replaying incomplete journal preserves creation provenance monotonically."""
key = "test_key_interrupted_replay_1"
branch = "fix/issue-850-interrupted-replay"
wt_path = os.path.join(self.branches_dir, "fix-issue-850-interrupted-replay")
# Simulate Phase 2/3 completion where branch and worktree directory were created by this transition
journal = {
"idempotency_key": key,
"issue_number": 850,
"branch_name": branch,
"worktree_path": wt_path,
"active_identity": "jcwalker3",
"active_profile": "prgs-author",
"remote": "prgs",
"org": "Scaled-Tech-Consulting",
"repo": "Gitea-Tools",
"phases": {
author_issue_bootstrap.PHASE_1_REQUEST_ACCEPTED: {"status": "completed"},
author_issue_bootstrap.PHASE_2_BRANCH_CONFIRMED: {"status": "completed", "created": True},
},
"artifacts_created": {
"branch_created": True,
"worktree_dir_created": True,
"worktree_registered": True,
"lock_created": False,
},
"current_phase": author_issue_bootstrap.PHASE_3_PATH_RESERVED,
"completed": False,
}
# Pre-create the branch and worktree on disk to simulate partial state after crash
subprocess.run(["git", "-C", self.repo_dir, "branch", branch, self.master_sha], check=True, capture_output=True)
subprocess.run(["git", "-C", self.repo_dir, "worktree", "add", wt_path, branch], check=True, capture_output=True)
author_issue_bootstrap.save_phase_journal(journal, journal_dir=self.lock_dir)
# Now resume/replay the transition but simulate lock binding failure during Phase 6
with mock.patch("issue_lock_store.bind_session_lock", side_effect=RuntimeError("Lock failure test")):
res = author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=850,
canonical_repo_root=self.repo_dir,
branch_name=branch,
worktree_path=wt_path,
idempotency_key=key,
lock_dir=self.lock_dir,
owner_session="session-test-1234",
)
self.assertFalse(res["success"])
self.assertEqual(res.get("reason_code"), "issue_lock_acquisition_failed")
# Verify that compensating recovery correctly deleted transition-created branch & worktree
# because creation provenance was preserved across replay (NOT downgraded to False!)
self.assertFalse(os.path.exists(wt_path))
branch_check = subprocess.run(["git", "-C", self.repo_dir, "rev-parse", "--verify", branch], capture_output=True, text=True, check=False)
self.assertNotEqual(branch_check.returncode, 0)
def test_transition_created_only_compensation(self):
"""Review #525 Finding 4: Preexisting branch is NOT deleted by compensation when only worktree was transition-created."""
key = "test_key_preexisting_branch_compensation"
preexisting_branch = "fix/issue-850-preexisting"
wt_path = os.path.join(self.branches_dir, "fix-issue-850-preexisting")
# Create branch BEFORE bootstrap (preexisting branch)
subprocess.run(["git", "-C", self.repo_dir, "branch", preexisting_branch, self.master_sha], check=True, capture_output=True)
# Call bootstrap with simulated failure during Phase 6 (lock binding)
with mock.patch("issue_lock_store.bind_session_lock", side_effect=RuntimeError("Simulated lock failure")):
res = author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=850,
canonical_repo_root=self.repo_dir,
branch_name=preexisting_branch,
worktree_path=wt_path,
idempotency_key=key,
lock_dir=self.lock_dir,
owner_session="session-test-1234",
)
self.assertFalse(res["success"])
# Worktree dir was created by transition -> removed by compensation
self.assertFalse(os.path.exists(wt_path))
# Preexisting branch was NOT created by transition -> MUST BE PRESERVED!
branch_check = subprocess.run(["git", "-C", self.repo_dir, "rev-parse", "--verify", preexisting_branch], capture_output=True, text=True, check=False)
self.assertEqual(branch_check.returncode, 0, "Preexisting branch was deleted by mistake!")
def test_incompatible_idempotency_replay_refusal(self):
"""Review #525 Finding 4: Replaying key with incompatible parameters returns refusal."""
key = "test_key_incompatible_replay"
res1 = author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=850,
canonical_repo_root=self.repo_dir,
branch_name="fix/issue-850-param-a",
idempotency_key=key,
lock_dir=self.lock_dir,
owner_session="session-test-1234",
)
self.assertTrue(res1["success"])
# Second call with different branch_name
res2 = author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=850,
canonical_repo_root=self.repo_dir,
branch_name="fix/issue-850-param-b",
idempotency_key=key,
lock_dir=self.lock_dir,
owner_session="session-test-1234",
)
self.assertFalse(res2["success"])
self.assertEqual(res2.get("reason_code"), "incompatible_idempotency_replay")
def test_exact_next_action_satisfiable_via_mcp(self):
"""Review #525 Finding 4: exact_next_action provides satisfiable MCP actions, not shell commands."""
key = "test_key_next_action_mcp"
stale_sha = "0000000000000000000000000000000000000000"
res = author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=850,
canonical_repo_root=self.repo_dir,
expected_base_sha=stale_sha,
lock_dir=self.lock_dir,
owner_session="session-test-1234",
)
next_action = res.get("exact_next_action", "")
self.assertNotIn("scripts/worktree-start", next_action)
self.assertNotIn("git worktree add", next_action)
self.assertNotIn("bash", next_action.lower())
def test_missing_owner_session_refusal(self):
"""Finding D: Missing owner_session context fails closed with typed refusal and zero mutation."""
res = author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=850,
canonical_repo_root=self.repo_dir,
owner_session=None,
lock_dir=self.lock_dir,
)
self.assertFalse(res["success"])
self.assertEqual(res.get("reason_code"), "missing_owner_session")
self.assertIn("exact_next_action", res)
def test_symlink_lock_file_refusal(self):
"""Finding C: BootstrapTransitionLock refuses to follow symlinks."""
key = "test_symlink_lock_key"
safe_key = "".join(c if c.isalnum() or c in ("-", "_", ".") else "_" for c in key)
lock_path = os.path.join(self.lock_dir, f"{safe_key}.lock")
target_file = os.path.join(self.tmp_dir, "fake_target")
with open(target_file, "w") as f:
f.write("target")
os.symlink(target_file, lock_path)
with self.assertRaises(RuntimeError) as ctx:
with author_issue_bootstrap.BootstrapTransitionLock(key, journal_dir=self.lock_dir):
pass
self.assertIn("symlink", str(ctx.exception).lower())
def test_lock_directory_escape_refusal(self):
"""Finding C: BootstrapTransitionLock refuses keys that escape lock directory."""
with mock.patch("os.path.abspath", return_value="/tmp/outside/evil_key.lock"):
with self.assertRaises(RuntimeError) as ctx:
author_issue_bootstrap.BootstrapTransitionLock("key", journal_dir=self.lock_dir)
self.assertIn("escapes", str(ctx.exception).lower())
def test_missing_active_identity_refusal(self):
"""F-5: Missing active_identity parameter fails closed."""
res = author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=850,
canonical_repo_root=self.repo_dir,
owner_session="session-test-1234",
active_identity=None,
active_profile="prgs-author",
lock_dir=self.lock_dir,
)
self.assertFalse(res["success"])
self.assertEqual(res.get("reason_code"), "missing_active_identity")
def test_missing_active_profile_refusal(self):
"""F-5: Missing active_profile parameter fails closed."""
res = author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=850,
canonical_repo_root=self.repo_dir,
owner_session="session-test-1234",
active_identity="jcwalker3",
active_profile=None,
lock_dir=self.lock_dir,
)
self.assertFalse(res["success"])
self.assertEqual(res.get("reason_code"), "missing_active_profile")
def test_dirty_worktree_preserved_during_recovery(self):
"""F-4: Compensating recovery does not delete dirty worktree."""
branch = "fix/issue-850-rec-dirty"
wt_path = os.path.join(self.branches_dir, "fix-issue-850-rec-dirty")
subprocess.run(["git", "-C", self.repo_dir, "worktree", "add", "-b", branch, wt_path], check=True, capture_output=True)
dirty_file = os.path.join(wt_path, "dirty.txt")
with open(dirty_file, "w") as f:
f.write("uncommitted work")
journal = {
"idempotency_key": "test_dirty_rec",
"issue_number": 850,
"branch_name": branch,
"worktree_path": wt_path,
"artifacts_created": {
"worktree_dir_created": True,
"worktree_registered": True,
},
"failure_reason": "test dirty recovery",
}
rec = author_issue_bootstrap.run_compensating_recovery(journal, self.repo_dir, journal_dir=self.lock_dir)
self.assertTrue(os.path.exists(wt_path))
self.assertIn(f"worktree_path_preserved_dirty:{wt_path}", rec["rolled_back"])
def test_branch_with_commits_preserved_during_recovery(self):
"""F-4: Compensating recovery does not delete branch with author commits."""
branch = "fix/issue-850-rec-commits"
subprocess.run(["git", "-C", self.repo_dir, "branch", branch, self.master_sha], check=True, capture_output=True)
# Add a commit on the branch
wt_path = os.path.join(self.branches_dir, "fix-issue-850-rec-commits")
subprocess.run(["git", "-C", self.repo_dir, "worktree", "add", wt_path, branch], check=True, capture_output=True)
cfile = os.path.join(wt_path, "commit.txt")
with open(cfile, "w") as f:
f.write("author commit")
subprocess.run(["git", "-C", wt_path, "add", "commit.txt"], check=True, capture_output=True)
subprocess.run(["git", "-C", wt_path, "commit", "-m", "author commit"], check=True, capture_output=True)
subprocess.run(["git", "-C", self.repo_dir, "worktree", "remove", "--force", wt_path], check=True, capture_output=True)
journal = {
"idempotency_key": "test_commits_rec",
"issue_number": 850,
"branch_name": branch,
"resolved_base_sha": self.master_sha,
"artifacts_created": {
"branch_created": True,
},
"failure_reason": "test commit branch recovery",
}
rec = author_issue_bootstrap.run_compensating_recovery(journal, self.repo_dir, journal_dir=self.lock_dir)
branch_check = subprocess.run(["git", "-C", self.repo_dir, "rev-parse", "--verify", branch], capture_output=True, text=True, check=False)
self.assertEqual(branch_check.returncode, 0, "Branch with commits was deleted!")
self.assertIn(f"branch_preserved_commits:{branch}", rec["rolled_back"])
def test_task_capability_map_integration(self):
"""Verify task_capability_map has bootstrap_author_issue_worktree configured correctly."""
self.assertEqual(task_capability_map.required_role("bootstrap_author_issue_worktree"), "author")
self.assertEqual(task_capability_map.required_permission("bootstrap_author_issue_worktree"), "gitea.branch.create")
self.assertTrue(task_capability_map.preflight_task_matches("work_issue", "bootstrap_author_issue_worktree"))
self.assertTrue(task_capability_map.preflight_task_matches("bootstrap_author_issue_worktree", "lock_issue"))
if __name__ == "__main__":
unittest.main()
-15
View File
@@ -28,21 +28,6 @@ class TestPathUnderBranches(unittest.TestCase):
amw.is_path_under_branches("/repo/other-checkout", self.ROOT) amw.is_path_under_branches("/repo/other-checkout", self.ROOT)
) )
def test_unrelated_branches_dir_fails(self):
self.assertFalse(
amw.is_path_under_branches("/tmp/branches/evil", self.ROOT)
)
def test_prefix_confusion_fails(self):
self.assertFalse(
amw.is_path_under_branches(f"{self.ROOT}/branches-other/foo", self.ROOT)
)
def test_traversal_fails(self):
self.assertFalse(
amw.is_path_under_branches(f"{self.ROOT}/branches/../evil", self.ROOT)
)
class TestAssessAuthorMutationWorktree(unittest.TestCase): class TestAssessAuthorMutationWorktree(unittest.TestCase):
ROOT = "/repo/Gitea-Tools" ROOT = "/repo/Gitea-Tools"
@@ -139,8 +139,6 @@ EXPECTED_ROLE_EXCLUSIVE_TASKS = frozenset(
"gitea_release_merger_pr_lease", "gitea_release_merger_pr_lease",
"create_branch", "create_branch",
"push_branch", "push_branch",
"bootstrap_author_issue_worktree",
"gitea_bootstrap_author_issue_worktree",
# #812 AC20: publishing an unpublished local head is author-only for the # #812 AC20: publishing an unpublished local head is author-only for the
# same reason every other push is — it writes a branch to the remote. # same reason every other push is — it writes a branch to the remote.
"publish_unpublished_branch", "publish_unpublished_branch",
+502
View File
@@ -0,0 +1,502 @@
"""Sanctioned restart / graceful reload control tests (#642).
Acceptance criteria under test:
1. The sanctioned restart path is implemented behind gates (capability,
confirmation, operator authorization, host hook).
2. Manual ``pkill`` stays forbidden and is classified as contamination.
3. Post-restart mutations require clean health/session proof.
4. Authorized restart preview, unauthorized deny, contamination classification.
5. No entry point exposes a raw kill.
"""
import json
import os
import tempfile
import unittest
import mcp_namespace_health
import runtime_recovery_guard
from task_capability_map import TASK_CAPABILITY_MAP
from webui import console_audit, console_authz, gated_actions, sanctioned_restart
NAMESPACE = "gitea-author"
# An operator-authorized, hook-configured host. Passed explicitly so no test
# depends on (or mutates) the real process environment.
READY_ENV = {
sanctioned_restart.RESTART_HOOK_ENV: "launchd:cc.prgs.gitea-author",
runtime_recovery_guard.OPERATOR_AUTHORIZATION_ENV: "ops-ticket-4821",
}
def admin(subject: str = "[email protected]") -> console_authz.Principal:
return console_authz.Principal(
subject=subject,
role=console_authz.ADMIN,
identity_source=console_authz.IDENTITY_ACCESS_PROXY,
authenticated=True,
)
def viewer() -> console_authz.Principal:
return console_authz.Principal(
subject="[email protected]",
role=console_authz.VIEWER,
identity_source=console_authz.IDENTITY_ACCESS_PROXY,
authenticated=True,
)
class TestCapabilityWiring(unittest.TestCase):
"""AC1: authority is declared, not invented by the console."""
def test_actions_resolve_through_the_capability_map(self):
for action_id in (
sanctioned_restart.ACTION_RESTART_NAMESPACE,
sanctioned_restart.ACTION_RELOAD_NAMESPACE,
):
with self.subTest(action=action_id):
action = console_authz.get_action(action_id)
self.assertIsNotNone(action)
self.assertIn(action.task_key, TASK_CAPABILITY_MAP)
self.assertEqual(
action.mcp_permission,
TASK_CAPABILITY_MAP[action.task_key]["permission"],
)
def test_restart_permission_is_not_a_gitea_operation(self):
"""No configured Gitea profile should satisfy a host restart."""
permission = TASK_CAPABILITY_MAP["restart_namespace"]["permission"]
self.assertFalse(permission.startswith("gitea."))
def test_restart_is_destructive_dual_control_break_glass(self):
action = console_authz.get_action(
sanctioned_restart.ACTION_RESTART_NAMESPACE
)
self.assertEqual(action.action_class, console_authz.CLASS_DESTRUCTIVE)
self.assertEqual(action.minimum_role, console_authz.ADMIN)
self.assertTrue(action.dual_control)
self.assertTrue(action.break_glass)
self.assertTrue(action.requires_confirmation)
def test_reload_is_privileged_but_not_destructive(self):
action = console_authz.get_action(
sanctioned_restart.ACTION_RELOAD_NAMESPACE
)
self.assertEqual(action.action_class, console_authz.CLASS_PRIVILEGED)
self.assertTrue(action.requires_confirmation)
class TestPreview(unittest.TestCase):
"""AC4: an authorized preview renders the plan without executing it."""
def test_preview_lists_the_mutation_ledger(self):
preview = sanctioned_restart.build_restart_preview(
NAMESPACE, principal=admin(), env=READY_ENV
)
steps = [entry["step"] for entry in preview["mutation_ledger"]]
self.assertEqual(
steps, ["quiesce", "host_restart_hook", "health_recheck", "audit"]
)
self.assertTrue(preview["scope_valid"])
self.assertTrue(preview["post_restart_verification_required"])
def test_reload_preview_drains_instead_of_restarting(self):
preview = sanctioned_restart.build_restart_preview(
NAMESPACE, sanctioned_restart.MODE_RELOAD,
principal=admin(), env=READY_ENV,
)
steps = [entry["step"] for entry in preview["mutation_ledger"]]
self.assertIn("host_graceful_reload", steps)
self.assertNotIn("host_restart_hook", steps)
def test_preview_never_enables_execution(self):
preview = sanctioned_restart.build_restart_preview(
NAMESPACE, principal=admin(), env=READY_ENV
)
self.assertFalse(preview["execution_enabled"])
self.assertFalse(preview["authorization"]["execution_enabled"])
def test_confirmation_phrase_binds_the_namespace(self):
self.assertTrue(
sanctioned_restart.confirmation_matches(
NAMESPACE, sanctioned_restart.MODE_RESTART,
"restart gitea-author",
)
)
# A phrase typed for one namespace must not authorize another.
self.assertFalse(
sanctioned_restart.confirmation_matches(
"gitea-merger", sanctioned_restart.MODE_RESTART,
"restart gitea-author",
)
)
class TestGates(unittest.TestCase):
"""AC1/AC4: every gate denies with a stable reason code."""
def _assess(self, **kwargs):
params = {
"principal": admin(),
"confirmation": f"restart {NAMESPACE}",
"env": READY_ENV,
}
params.update(kwargs)
namespace = params.pop("namespace", NAMESPACE)
mode = params.pop("mode", sanctioned_restart.MODE_RESTART)
return sanctioned_restart.assess_restart_request(
namespace, mode, **params
)
def test_authorized_confirmed_request_passes_every_gate(self):
result = self._assess()
self.assertTrue(result["allowed"])
self.assertEqual(
result["reason_code"], sanctioned_restart.ALLOW_HOST_ACTION_REQUIRED
)
def test_passing_every_gate_is_not_an_execution_grant(self):
"""An allowed request still never lets the console touch the process."""
result = self._assess()
self.assertTrue(result["allowed"])
self.assertFalse(result["execution_enabled"])
self.assertFalse(result["console_executes"])
def test_unauthorized_principal_is_denied(self):
result = self._assess(principal=viewer())
self.assertFalse(result["allowed"])
self.assertEqual(
result["reason_code"], sanctioned_restart.DENY_UNAUTHORIZED
)
def test_anonymous_principal_is_denied(self):
result = self._assess(principal=None)
self.assertFalse(result["allowed"])
self.assertEqual(
result["reason_code"], sanctioned_restart.DENY_UNAUTHORIZED
)
def test_missing_confirmation_is_denied(self):
result = self._assess(confirmation=None)
self.assertFalse(result["allowed"])
self.assertEqual(
result["reason_code"], sanctioned_restart.DENY_CONFIRMATION_MISSING
)
def test_confirmation_for_another_namespace_is_denied(self):
result = self._assess(confirmation="restart gitea-merger")
self.assertFalse(result["allowed"])
self.assertEqual(
result["reason_code"], sanctioned_restart.DENY_CONFIRMATION_MISMATCH
)
def test_missing_operator_authorization_is_denied(self):
env = {sanctioned_restart.RESTART_HOOK_ENV: "launchd:cc.prgs.author"}
result = self._assess(env=env)
self.assertFalse(result["allowed"])
self.assertEqual(
result["reason_code"],
sanctioned_restart.DENY_OPERATOR_AUTHORIZATION,
)
def test_missing_host_hook_is_denied_without_kill_fallback(self):
env = {
runtime_recovery_guard.OPERATOR_AUTHORIZATION_ENV: "ops-ticket-1",
}
result = self._assess(env=env)
self.assertFalse(result["allowed"])
self.assertEqual(
result["reason_code"], sanctioned_restart.DENY_HOOK_NOT_CONFIGURED
)
def test_fleet_scope_is_refused(self):
for scope in ("all", "*", "fleet"):
with self.subTest(scope=scope):
result = self._assess(
namespace=scope, confirmation=f"restart {scope}"
)
self.assertFalse(result["allowed"])
self.assertEqual(
result["reason_code"], sanctioned_restart.DENY_FLEET_SCOPE
)
def test_unknown_namespace_is_refused(self):
result = self._assess(
namespace="gitea-nope", confirmation="restart gitea-nope"
)
self.assertFalse(result["allowed"])
self.assertEqual(
result["reason_code"], sanctioned_restart.DENY_UNKNOWN_NAMESPACE
)
def test_unknown_mode_is_refused(self):
result = self._assess(mode="obliterate")
self.assertFalse(result["allowed"])
self.assertEqual(
result["reason_code"], sanctioned_restart.DENY_UNKNOWN_MODE
)
def test_live_contamination_marker_blocks_restart(self):
marker = runtime_recovery_guard.build_contamination_record(
reason_class=runtime_recovery_guard.REASON_MANUAL_DAEMON_KILL,
command_redacted="pkill -f mcp_server.py",
)
result = self._assess(contamination_marker=marker)
self.assertFalse(result["allowed"])
self.assertEqual(
result["reason_code"], sanctioned_restart.DENY_CONTAMINATED_RUNTIME
)
def test_reconciler_cleared_marker_no_longer_blocks(self):
marker = runtime_recovery_guard.build_contamination_record(
reason_class=runtime_recovery_guard.REASON_MANUAL_DAEMON_KILL,
command_redacted="pkill -f mcp_server.py",
)
marker = dict(marker, cleared_by_reconciler=True)
result = self._assess(contamination_marker=marker)
self.assertTrue(result["allowed"])
class TestExecutionNeverKills(unittest.TestCase):
"""AC5: no path exposes or runs a raw process kill."""
def test_authorized_execution_defers_to_the_host_supervisor(self):
result = sanctioned_restart.execute_restart(
NAMESPACE,
principal=admin(),
confirmation=f"restart {NAMESPACE}",
env=READY_ENV,
)
self.assertTrue(result["allowed"])
self.assertFalse(result["success"])
self.assertFalse(result["process_kill_executed"])
self.assertEqual(
result["outcome"], sanctioned_restart.ALLOW_HOST_ACTION_REQUIRED
)
def test_denied_execution_reports_the_refusing_gate(self):
result = sanctioned_restart.execute_restart(
NAMESPACE, principal=viewer(), confirmation=f"restart {NAMESPACE}",
env=READY_ENV,
)
self.assertFalse(result["allowed"])
self.assertEqual(
result["outcome"], sanctioned_restart.DENY_UNAUTHORIZED
)
self.assertFalse(result["process_kill_executed"])
def test_module_never_spawns_a_process(self):
path = os.path.join(
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
"webui", "sanctioned_restart.py",
)
with open(path, encoding="utf-8") as handle:
source = handle.read()
for forbidden in (
"import subprocess", "import signal", "os.kill", "os.system",
"popen",
):
with self.subTest(forbidden=forbidden):
self.assertNotIn(forbidden, source.lower())
def test_no_surface_returns_a_kill_command(self):
payloads = [
sanctioned_restart.build_restart_preview(
NAMESPACE, principal=admin(), env=READY_ENV
),
sanctioned_restart.restart_policy(),
sanctioned_restart.execute_restart(
NAMESPACE, principal=admin(),
confirmation=f"restart {NAMESPACE}", env=READY_ENV,
),
]
for payload in payloads:
rendered = json.dumps(payload, default=str).lower()
self.assertNotIn("kill -9", rendered)
self.assertNotIn("pkill -f", rendered)
def test_policy_declares_no_raw_kill_and_no_silent_restart(self):
policy = sanctioned_restart.restart_policy()
self.assertFalse(policy["raw_kill_exposed"])
self.assertFalse(policy["console_executes_process_kill"])
self.assertFalse(policy["fleet_scope_permitted"])
self.assertFalse(policy["silent_auto_restart_permitted"])
self.assertTrue(policy["audit_required"])
class TestContaminationClassification(unittest.TestCase):
"""AC2: manual pkill is contamination, and it blocks clean claims."""
def test_manual_daemon_pkill_is_contamination(self):
result = sanctioned_restart.classify_restart_command(
"pkill -f mcp_server.py"
)
self.assertTrue(result["contamination"])
self.assertFalse(result["clean_claim_allowed"])
self.assertIsNotNone(result["contamination_marker"])
self.assertEqual(
result["sanctioned_alternative"],
sanctioned_restart.ACTION_RESTART_NAMESPACE,
)
def test_broad_process_kill_is_contamination(self):
result = sanctioned_restart.classify_restart_command("killall -9 Python")
self.assertTrue(result["contamination"])
self.assertFalse(result["clean_claim_allowed"])
def test_marker_names_the_sanctioned_alternative(self):
result = sanctioned_restart.classify_restart_command(
"pkill -f mcp_server.py"
)
marker = result["contamination_marker"]
self.assertIn(
sanctioned_restart.ACTION_RESTART_NAMESPACE, marker["detail"]
)
def test_benign_command_is_not_contamination(self):
result = sanctioned_restart.classify_restart_command("git status")
self.assertFalse(result["contamination"])
self.assertTrue(result["clean_claim_allowed"])
def test_no_command_is_not_contamination(self):
result = sanctioned_restart.classify_restart_command(None)
self.assertFalse(result["contamination"])
self.assertTrue(result["clean_claim_allowed"])
class TestPostRestartHealth(unittest.TestCase):
"""AC3: a clean post-restart claim needs live client-namespace proof."""
def test_live_client_probe_clears_the_session(self):
result = sanctioned_restart.verify_post_restart_health(
NAMESPACE,
probe_result={"success": True},
probe_source=mcp_namespace_health.PROBE_SOURCE_CLIENT,
registered_tools=["gitea_whoami"],
required_tool="gitea_whoami",
)
self.assertEqual(result["status"], sanctioned_restart.HEALTH_CLEAN)
self.assertTrue(result["clean_claim_allowed"])
self.assertTrue(result["mutations_allowed"])
def test_offline_probe_does_not_clear_the_session(self):
result = sanctioned_restart.verify_post_restart_health(
NAMESPACE,
probe_result={"success": True},
probe_source=mcp_namespace_health.PROBE_SOURCE_OFFLINE,
registered_tools=["gitea_whoami"],
required_tool="gitea_whoami",
)
self.assertFalse(result["clean_claim_allowed"])
self.assertFalse(result["mutations_allowed"])
def test_failed_probe_is_unhealthy(self):
result = sanctioned_restart.verify_post_restart_health(
NAMESPACE,
probe_result={"success": False, "error": "client is closing: EOF"},
probe_source=mcp_namespace_health.PROBE_SOURCE_CLIENT,
registered_tools=["gitea_whoami"],
required_tool="gitea_whoami",
)
self.assertEqual(result["status"], sanctioned_restart.HEALTH_UNHEALTHY)
self.assertFalse(result["clean_claim_allowed"])
def test_static_registration_alone_never_clears_the_session(self):
result = sanctioned_restart.verify_post_restart_health(
NAMESPACE,
registered_tools=["gitea_whoami"],
required_tool="gitea_whoami",
)
self.assertFalse(result["clean_claim_allowed"])
class TestAuditEmission(unittest.TestCase):
"""Every restart attempt is audited with actor, target, and result."""
def _run(self, principal, sink):
prior = os.environ.get(console_audit.AUDIT_LOG_ENV)
os.environ[console_audit.AUDIT_LOG_ENV] = sink
try:
return sanctioned_restart.execute_restart(
NAMESPACE,
principal=principal,
confirmation=f"restart {NAMESPACE}",
env=READY_ENV,
request_id="req-642",
)
finally:
if prior is None:
os.environ.pop(console_audit.AUDIT_LOG_ENV, None)
else:
os.environ[console_audit.AUDIT_LOG_ENV] = prior
def test_allowed_attempt_is_written_with_actor_and_target(self):
with tempfile.TemporaryDirectory() as tmp:
sink = os.path.join(tmp, "audit.jsonl")
result = self._run(admin(), sink)
self.assertTrue(result["audit"]["written"])
with open(sink, encoding="utf-8") as handle:
record = json.loads(handle.read().strip())
self.assertEqual(
record["action"], sanctioned_restart.ACTION_RESTART_NAMESPACE
)
self.assertEqual(record["target"]["namespace"], NAMESPACE)
self.assertEqual(record["target"]["mode"], "restart")
self.assertEqual(record["result"], console_audit.RESULT_ALLOWED)
self.assertEqual(record["actor"]["subject"], "[email protected]")
self.assertFalse(record["metadata"]["process_kill_executed"])
def test_denied_attempt_is_audited_too(self):
with tempfile.TemporaryDirectory() as tmp:
sink = os.path.join(tmp, "audit.jsonl")
self._run(viewer(), sink)
with open(sink, encoding="utf-8") as handle:
record = json.loads(handle.read().strip())
self.assertEqual(record["result"], console_audit.RESULT_DENIED)
self.assertEqual(
record["reason_code"], sanctioned_restart.DENY_UNAUTHORIZED
)
def test_restart_audit_uses_break_glass_retention(self):
action = console_authz.get_action(
sanctioned_restart.ACTION_RESTART_NAMESPACE
)
self.assertEqual(
console_audit.retention_class_for(action),
console_audit.RETENTION_BREAK_GLASS,
)
class TestRegistrySurface(unittest.TestCase):
"""AC5: the console surfaces the control, still disabled, with no kill."""
def test_registry_exposes_both_actions_disabled(self):
registry = gated_actions.load_action_registry()
for action_id in (
sanctioned_restart.ACTION_RESTART_NAMESPACE,
sanctioned_restart.ACTION_RELOAD_NAMESPACE,
):
with self.subTest(action=action_id):
action = registry.get(action_id)
self.assertIsNotNone(action)
self.assertFalse(action.enabled)
def test_registry_preview_names_the_namespace_target(self):
preview = gated_actions.preview_action(
sanctioned_restart.ACTION_RESTART_NAMESPACE, namespace=NAMESPACE
)
target = preview["mutation_ledger"][0]["target"]
self.assertIn(NAMESPACE, target)
self.assertFalse(preview["enabled"])
def test_registry_attempt_fails_closed(self):
result = gated_actions.attempt_action(
sanctioned_restart.ACTION_RESTART_NAMESPACE, namespace=NAMESPACE
)
self.assertFalse(result["success"])
if __name__ == "__main__":
unittest.main()
+28
View File
@@ -236,6 +236,34 @@ _ACTION_SPECS: tuple[ConsoleAction, ...] = (
phase=3, phase=3,
summary="Remove a remote feature branch.", summary="Remove a remote feature branch.",
), ),
# #642: sanctioned daemon lifecycle. These exist so operators have an
# audited path off `pkill -f mcp_server.py` (#630). Restart drops every
# in-flight request on a namespace, so it carries the same dual-control and
# break-glass weight as a merge; reload drains first and is privileged but
# not destructive. Neither ever exposes a raw kill: execution is handed to
# a host supervisor by ``webui.sanctioned_restart``.
ConsoleAction(
action_id="system.reload_namespace",
task_key="reload_namespace",
action_class=CLASS_PRIVILEGED,
minimum_role=CONTROLLER,
requires_confirmation=True,
dual_control=False,
break_glass=False,
phase=2,
summary="Gracefully reload one MCP namespace via the host supervisor.",
),
ConsoleAction(
action_id="system.restart_namespace",
task_key="restart_namespace",
action_class=CLASS_DESTRUCTIVE,
minimum_role=ADMIN,
requires_confirmation=True,
dual_control=True,
break_glass=True,
phase=2,
summary="Restart one MCP namespace via the host supervisor.",
),
) )
ACTIONS: dict[str, ConsoleAction] = {a.action_id: a for a in _ACTION_SPECS} ACTIONS: dict[str, ConsoleAction] = {a.action_id: a for a in _ACTION_SPECS}
+13
View File
@@ -110,6 +110,8 @@ def _format_target(action_id: str, params: dict[str, Any]) -> str:
) )
if action_id == "create_issue": if action_id == "create_issue":
return f"issue {params.get('title', '?')!r}" return f"issue {params.get('title', '?')!r}"
if action_id in {"system.restart_namespace", "system.reload_namespace"}:
return f"MCP namespace {params.get('namespace', '?')!r}"
return "unspecified" return "unspecified"
@@ -165,6 +167,17 @@ def build_action_registry() -> ActionRegistry:
"gitea_create_issue_comment", "Post a PR review thread comment."), "gitea_create_issue_comment", "Post a PR review thread comment."),
("close_pr", "Close PR", "close_pr", "gitea_edit_pr", ("close_pr", "Close PR", "close_pr", "gitea_edit_pr",
"Close a pull request without merge."), "Close a pull request without merge."),
# #642: the sanctioned replacement for the forbidden manual daemon-kill
# recovery path (#630). The "tool" is a host supervisor hook, not an MCP
# call — the console never signals a process. Preview and gating live in
# ``webui.sanctioned_restart``; these stay disabled like every other
# registry entry.
("system.reload_namespace", "Reload MCP namespace", "reload_namespace",
"host.supervisor_reload",
"Gracefully reload one MCP namespace via the host supervisor."),
("system.restart_namespace", "Restart MCP namespace",
"restart_namespace", "host.supervisor_restart",
"Restart one MCP namespace via the host supervisor."),
) )
actions = tuple( actions = tuple(
GatedAction( GatedAction(
+613
View File
@@ -0,0 +1,613 @@
"""Sanctioned MCP restart and graceful reload controls (#642, Phase 2).
Sessions have historically recovered MCP connectivity by killing the host
daemon (``pkill -f mcp_server.py``, #630). That path stays forbidden: it kills
every namespace on the host, contaminates the surviving session, and leaves no
audit trail. This module is the sanctioned replacement.
A restart is modelled as a *gated action*, never as a command:
1. **Capability** — the console action resolves through ``console_authz``
against ``task_capability_map``, so the console cannot invent an authority
the MCP layer does not already define.
2. **Preview** — :func:`build_restart_preview` renders a mutation ledger and
the exact confirmation phrase. It never returns a shell command.
3. **Confirmation** — the operator echoes a phrase naming the exact namespace
and mode. A phrase for one namespace never authorizes another.
4. **Operator authorization** — host daemon maintenance is authorized out of
band through the environment (#630, and #710 finding F1: a worker session
cannot set an env var for an already-running daemon, so this cannot be
self-asserted the way a tool argument could).
5. **Execution** — :func:`execute_restart` never spawns a process. Once every
gate passes it hands the request to the configured host-managed restart
hook; with no hook configured it fails closed.
6. **Health recheck** — :func:`verify_post_restart_health` requires live
client-namespace probe evidence before any post-restart clean claim.
Manual ``pkill`` remains forbidden and is classified as contamination by
:func:`classify_restart_command`, which blocks clean claims (#630 AC3).
This module performs no I/O beyond reading its own environment configuration,
imports no MCP client, and holds no credential.
"""
from __future__ import annotations
import os
from dataclasses import asdict, dataclass
from typing import Any
import mcp_namespace_health
import runtime_recovery_guard
from webui import console_audit, console_authz
# --- Operations -------------------------------------------------------------
MODE_RESTART = "restart"
MODE_RELOAD = "reload"
MODES: tuple[str, ...] = (MODE_RESTART, MODE_RELOAD)
ACTION_RESTART_NAMESPACE = "system.restart_namespace"
ACTION_RELOAD_NAMESPACE = "system.reload_namespace"
ACTION_FOR_MODE: dict[str, str] = {
MODE_RESTART: ACTION_RESTART_NAMESPACE,
MODE_RELOAD: ACTION_RELOAD_NAMESPACE,
}
# Namespaces the console may target. An unlisted name fails closed rather than
# being passed through to a host hook.
KNOWN_NAMESPACES: tuple[str, ...] = tuple(
sorted(
set(mcp_namespace_health.DEFAULT_NAMESPACES)
| {"gitea-author", "gitea-reviewer", "gitea-merger",
"gitea-reconciler", "gitea-controller"}
)
)
# Scope tokens that would mean "everything at once". Explicit non-goal: the
# console never offers a fleet-wide restart, because that is the blast radius
# `pkill -f mcp_server.py` already had.
_FLEET_TOKENS = frozenset({"*", "all", "fleet", "any", ""})
# --- Environment configuration ----------------------------------------------
# Read server-side only; the value is an opaque host hook reference (e.g. a
# launchd label), never a command line, and is never rendered to a client.
RESTART_HOOK_ENV = "GITEA_SANCTIONED_RESTART_HOOK"
# --- Reason codes -----------------------------------------------------------
DENY_UNKNOWN_MODE = "unknown_mode"
DENY_UNKNOWN_NAMESPACE = "unknown_namespace"
DENY_FLEET_SCOPE = "fleet_scope_not_permitted"
DENY_UNAUTHORIZED = "unauthorized"
DENY_CONFIRMATION_MISSING = "confirmation_required"
DENY_CONFIRMATION_MISMATCH = "confirmation_mismatch"
DENY_OPERATOR_AUTHORIZATION = "operator_authorization_missing"
DENY_HOOK_NOT_CONFIGURED = "restart_hook_not_configured"
DENY_CONTAMINATED_RUNTIME = "contaminated_runtime"
ALLOW_HOST_ACTION_REQUIRED = "host_action_required"
# Post-restart verification outcomes.
HEALTH_CLEAN = "clean"
HEALTH_UNPROVEN = "unproven"
HEALTH_UNHEALTHY = "unhealthy"
def _clean(value: Any) -> str:
return str(value or "").strip()
# --- Mutation ledger --------------------------------------------------------
@dataclass(frozen=True)
class RestartLedgerEntry:
"""One planned step, shown before anything is asked of the host."""
sequence: int
step: str
summary: str
executes_process_kill: bool = False
def _mutation_ledger(namespace: str, mode: str) -> tuple[RestartLedgerEntry, ...]:
if mode == MODE_RELOAD:
middle = RestartLedgerEntry(
sequence=2,
step="host_graceful_reload",
summary=(
f"Ask the configured host supervisor to reload {namespace} "
"in place, draining in-flight requests. The console does not "
"signal the process itself."
),
)
else:
middle = RestartLedgerEntry(
sequence=2,
step="host_restart_hook",
summary=(
f"Ask the configured host supervisor to restart {namespace}. "
"The console never sends a signal and never runs a kill."
),
)
return (
RestartLedgerEntry(
sequence=1,
step="quiesce",
summary=(
f"Stop admitting new gated mutations for {namespace} and "
"record the intent before anything restarts."
),
),
middle,
RestartLedgerEntry(
sequence=3,
step="health_recheck",
summary=(
f"Re-probe {namespace} through the live client namespace and "
"prove the required tool is callable again."
),
),
RestartLedgerEntry(
sequence=4,
step="audit",
summary=(
"Append actor, target namespace, mode, and result to the "
"console audit log."
),
),
)
# --- Confirmation -----------------------------------------------------------
def confirmation_phrase(namespace: str, mode: str) -> str:
"""Exact phrase an operator must echo, naming the namespace and mode.
Binding the namespace into the phrase is the point: a confirmation typed
for ``gitea-author`` cannot be replayed against ``gitea-merger``.
"""
return f"{_clean(mode)} {_clean(namespace)}"
def confirmation_matches(
namespace: str, mode: str, confirmation: str | None
) -> bool:
"""Compare *confirmation* to the required phrase (exact, whitespace-trimmed)."""
return _clean(confirmation) == confirmation_phrase(namespace, mode)
# --- Scope validation -------------------------------------------------------
def _validate_scope(namespace: str, mode: str) -> tuple[str, str] | None:
"""Return ``(reason_code, detail)`` when the scope is refused."""
ns = _clean(namespace)
md = _clean(mode)
if md not in MODES:
return (
DENY_UNKNOWN_MODE,
f"Mode {md!r} is not one of {', '.join(MODES)}.",
)
if ns.lower() in _FLEET_TOKENS:
return (
DENY_FLEET_SCOPE,
(
"Fleet-wide restart is an explicit non-goal: it reproduces the "
"blast radius of `pkill -f mcp_server.py` (#630). Restart one "
"namespace at a time."
),
)
if ns not in KNOWN_NAMESPACES:
return (
DENY_UNKNOWN_NAMESPACE,
f"Namespace {ns!r} is not a known MCP namespace.",
)
return None
# --- Host hook --------------------------------------------------------------
def restart_hook(env: dict[str, str] | None = None) -> dict[str, Any]:
"""Report the configured host-managed restart hook.
The hook is a reference the *host* resolves (a supervisor label), not a
command this process runs. ``configured=False`` fails restart closed.
"""
source = env if env is not None else os.environ
reference = _clean(source.get(RESTART_HOOK_ENV))
return {
"configured": bool(reference),
"reference": reference or None,
"source": RESTART_HOOK_ENV if reference else None,
"self_assertable": False,
"console_executes_process": False,
}
# --- Preview ----------------------------------------------------------------
def build_restart_preview(
namespace: str,
mode: str = MODE_RESTART,
*,
principal: console_authz.Principal | None = None,
env: dict[str, str] | None = None,
) -> dict[str, Any]:
"""Render the dry-run preview for a restart/reload request.
Read-only: no authorization is granted, no host is contacted, and the
result never contains a shell command.
"""
ns = _clean(namespace)
md = _clean(mode)
action_id = ACTION_FOR_MODE.get(md, ACTION_RESTART_NAMESPACE)
action = console_authz.get_action(action_id)
decision = console_authz.authorize(action_id, principal)
scope_error = _validate_scope(ns, md)
hook = restart_hook(env)
operator = runtime_recovery_guard.operator_authorization(env)
return {
"action_id": action_id,
"namespace": ns,
"mode": md,
"scope_valid": scope_error is None,
"scope_reason_code": scope_error[0] if scope_error else None,
"scope_detail": scope_error[1] if scope_error else None,
"required_role": action.minimum_role if action else None,
"required_permission": action.mcp_permission if action else None,
"action_class": action.action_class if action else None,
"dual_control": action.dual_control if action else True,
"break_glass": action.break_glass if action else True,
"requires_confirmation": True,
"confirmation_phrase": confirmation_phrase(ns, md),
"mutation_ledger": [asdict(entry) for entry in _mutation_ledger(ns, md)],
"authorization": decision.to_dict(),
"operator_authorization": operator,
"restart_hook": hook,
"execution_enabled": False,
"raw_process_kill_exposed": False,
"known_namespaces": list(KNOWN_NAMESPACES),
"post_restart_verification_required": True,
}
# --- Gate -------------------------------------------------------------------
def assess_restart_request(
namespace: str,
mode: str = MODE_RESTART,
*,
principal: console_authz.Principal | None = None,
confirmation: str | None = None,
contamination_marker: dict[str, Any] | None = None,
env: dict[str, str] | None = None,
) -> dict[str, Any]:
"""Decide whether a restart request may proceed to the host hook.
Every gate must pass. The first failure wins and is reported with a stable
reason code; a pass never means "restarted", only "may be handed to the
configured host hook".
"""
ns = _clean(namespace)
md = _clean(mode)
action_id = ACTION_FOR_MODE.get(md, ACTION_RESTART_NAMESPACE)
preview = build_restart_preview(ns, md, principal=principal, env=env)
def refuse(reason_code: str, detail: str) -> dict[str, Any]:
return {
"allowed": False,
"gates_passed": False,
"reason_code": reason_code,
"detail": detail,
"action_id": action_id,
"namespace": ns,
"mode": md,
"preview": preview,
"execution_enabled": False,
}
scope_error = _validate_scope(ns, md)
if scope_error is not None:
return refuse(*scope_error)
# Authority is checked as an authorization decision, not an execution
# grant. ``for_execution=True`` asks "may the console perform this write?",
# and the answer here is permanently no: step 2 of the ledger is a request
# to the host supervisor, so the console's Phase 2 execution gate is not
# the relevant gate. Every branch below keeps ``execution_enabled`` False
# and :func:`execute_restart` never touches a process.
decision = console_authz.authorize(action_id, principal)
if not decision.allowed:
return refuse(DENY_UNAUTHORIZED, decision.detail)
if not _clean(confirmation):
return refuse(
DENY_CONFIRMATION_MISSING,
(
"Type the confirmation phrase "
f"{preview['confirmation_phrase']!r} to proceed."
),
)
if not confirmation_matches(ns, md, confirmation):
return refuse(
DENY_CONFIRMATION_MISMATCH,
(
"Confirmation does not name this namespace and mode; expected "
f"{preview['confirmation_phrase']!r}."
),
)
operator = preview["operator_authorization"]
if not operator["authorized"]:
return refuse(
DENY_OPERATOR_AUTHORIZATION,
(
"Host daemon maintenance requires out-of-band operator "
"authorization via "
f"{runtime_recovery_guard.OPERATOR_AUTHORIZATION_ENV}."
),
)
# #630's task-scoped gate deliberately lets a contaminated worker keep
# commenting and handing off. Restart is stricter and unconditional: a
# runtime already contaminated by a manual kill must be reconciled before
# it is restarted again, or the restart just launders the contamination.
if contamination_marker and not contamination_marker.get(
"cleared_by_reconciler"
):
return refuse(
DENY_CONTAMINATED_RUNTIME,
(
"A live contamination marker is present; clear it through the "
"reconciler path before restarting."
),
)
hook = preview["restart_hook"]
if not hook["configured"]:
return refuse(
DENY_HOOK_NOT_CONFIGURED,
(
"No host-managed restart hook is configured "
f"({RESTART_HOOK_ENV}). The console will not fall back to a "
"process kill."
),
)
return {
"allowed": True,
"gates_passed": True,
"reason_code": ALLOW_HOST_ACTION_REQUIRED,
"detail": (
"Every gate passed. The restart must be performed by the "
"configured host supervisor; the console does not signal the "
"process."
),
"action_id": action_id,
"namespace": ns,
"mode": md,
"preview": preview,
"execution_enabled": False,
"console_executes": False,
"console_active_phase": console_authz.ACTIVE_PHASE,
}
# --- Execution --------------------------------------------------------------
def execute_restart(
namespace: str,
mode: str = MODE_RESTART,
*,
principal: console_authz.Principal | None = None,
confirmation: str | None = None,
contamination_marker: dict[str, Any] | None = None,
env: dict[str, str] | None = None,
request_id: str | None = None,
session_id: str | None = None,
) -> dict[str, Any]:
"""Run every gate, audit the outcome, and hand off to the host.
This function never spawns a process, never sends a signal, and never
builds a command line. ``success`` is False in both directions: a refused
request is refused, and an authorized request still requires the host
supervisor to act.
"""
assessment = assess_restart_request(
namespace,
mode,
principal=principal,
confirmation=confirmation,
contamination_marker=contamination_marker,
env=env,
)
action_id = assessment["action_id"]
allowed = assessment["allowed"]
audit = console_audit.record_event(
action_id=action_id,
result=(
console_audit.RESULT_ALLOWED if allowed
else console_audit.RESULT_DENIED
),
principal=principal,
target={"namespace": assessment["namespace"], "mode": assessment["mode"]},
reason_code=assessment["reason_code"],
detail=assessment["detail"],
request_id=request_id,
session_id=session_id,
metadata={
"gates_passed": assessment["gates_passed"],
"process_kill_executed": False,
"post_restart_verification_required": True,
},
)
return {
"success": False,
"outcome": (
ALLOW_HOST_ACTION_REQUIRED if allowed else assessment["reason_code"]
),
"allowed": allowed,
"detail": assessment["detail"],
"namespace": assessment["namespace"],
"mode": assessment["mode"],
"action_id": action_id,
"process_kill_executed": False,
"host_hook": assessment["preview"]["restart_hook"],
"next_action": (
"Have the host supervisor perform the restart, then call "
"verify_post_restart_health with live client-namespace evidence "
"before claiming a clean session."
if allowed
else assessment["detail"]
),
"assessment": assessment,
"audit": audit,
}
# --- Contamination classification -------------------------------------------
def classify_restart_command(
command: str | None,
*,
mcp_pids: list[Any] | tuple[Any, ...] | None = None,
session_id: str | None = None,
remote: str | None = None,
role: str | None = None,
) -> dict[str, Any]:
"""Classify an operator-proposed recovery command (#630 AC2).
A manual ``pkill``/``kill`` of the MCP daemon is contamination, not a
restart. When contaminating, a durable marker is returned so downstream
gated mutations and clean claims fail closed.
"""
classification = runtime_recovery_guard.classify_recovery_command(
command, mcp_pids=mcp_pids
)
contaminating = bool(classification.get("contamination"))
marker = None
if contaminating:
marker = runtime_recovery_guard.build_contamination_record(
reason_class=(
classification.get("reason_class")
or runtime_recovery_guard.REASON_MANUAL_DAEMON_KILL
),
command_redacted=classification.get("redacted_command"),
session_id=session_id,
remote=remote,
role=role,
detail=(
"Manual daemon kill is forbidden; use the sanctioned "
f"{ACTION_RESTART_NAMESPACE} gated action instead."
),
)
return {
"contamination": contaminating,
"sanctioned": not contaminating and not classification.get("process_kill"),
"clean_claim_allowed": not contaminating,
"reason_class": classification.get("reason_class"),
"redacted_command": classification.get("redacted_command"),
"classification": classification,
"contamination_marker": marker,
"sanctioned_alternative": ACTION_RESTART_NAMESPACE,
}
# --- Post-restart health verification ---------------------------------------
def verify_post_restart_health(
namespace: str,
*,
probe_result: dict[str, Any] | None = None,
probe_source: str | None = None,
registered_tools: list[str] | tuple[str, ...] | None = None,
required_tool: str | None = None,
profile: str | None = None,
) -> dict[str, Any]:
"""Require live proof a namespace is callable before any clean claim (AC3).
Static registration is not proof and neither is an offline subprocess
probe: only ``probe_source=client_namespace`` evidence can clear a
post-restart session for mutations.
"""
ns = _clean(namespace)
health = mcp_namespace_health.classify_namespace_probe(
ns,
required_tool=required_tool,
registered_tools=registered_tools,
probe_result=probe_result,
profile=profile,
probe_source=probe_source,
)
healthy = bool(health.get("healthy"))
proven = bool(health.get("ide_namespace_proven"))
if healthy and proven:
status = HEALTH_CLEAN
elif healthy:
status = HEALTH_UNPROVEN
else:
status = HEALTH_UNHEALTHY
reasons = list(health.get("reasons") or [])
if status == HEALTH_UNPROVEN:
reasons.append(
"Namespace reported healthy without live client-namespace "
"evidence; a post-restart clean claim requires "
f"probe_source={mcp_namespace_health.PROBE_SOURCE_CLIENT!r}."
)
return {
"namespace": ns,
"status": status,
"healthy": healthy,
"ide_namespace_proven": proven,
"clean_claim_allowed": status == HEALTH_CLEAN,
"mutations_allowed": status == HEALTH_CLEAN,
"reasons": reasons,
"health": health,
}
# --- Policy surface ---------------------------------------------------------
def restart_policy() -> dict[str, Any]:
"""Machine-readable description of the sanctioned restart contract."""
return {
"policy_version": 1,
"modes": list(MODES),
"actions": [ACTION_RESTART_NAMESPACE, ACTION_RELOAD_NAMESPACE],
"known_namespaces": list(KNOWN_NAMESPACES),
"fleet_scope_permitted": False,
"console_executes_process_kill": False,
"raw_kill_exposed": False,
"requires_confirmation": True,
"confirmation_binds_namespace": True,
"operator_authorization_env": (
runtime_recovery_guard.OPERATOR_AUTHORIZATION_ENV
),
"restart_hook_env": RESTART_HOOK_ENV,
"manual_kill_classified_as": runtime_recovery_guard.CONTAMINATION_KIND,
"post_restart_clean_claim_requires": (
mcp_namespace_health.PROBE_SOURCE_CLIENT
),
"audit_required": True,
"silent_auto_restart_permitted": False,
}