Compare commits

..
Author SHA1 Message Date
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
sysadmin 6d0015cabc Merge pull request 'fix(worktree-audit): make cleanup audit merged-PR aware for issue worktrees (Closes #858)' (#859) from fix/issue-858-audit-merged-pr-aware into master 2026-07-23 22:02:22 -05:00
sysadminandClaude Opus 4.8 fe259e6d38 fix(worktree-audit): make cleanup audit merged-PR aware for issue worktrees (Closes #858)
gitea_audit_worktree_cleanup had no PR linkage. Issue worktrees therefore
reported pr_number=null and classified as active_issue_work with
removable=false permanently, even once their PR was merged and the head was
already contained in master. Observed on live master 9301739910: 42
issue_work worktrees, 0 removable, 0 with pr_number populated, while
gitea_reconcile_merged_cleanups reported the same worktree safe to remove.

Two independent gaps caused it:

* build_worktree_metadata was never given a pr_number, and only open PRs were
  fetched, so no owning-PR evidence existed at all.
* clean_stale_removable was unreachable for issue_work: it required
  ttl_expired, derived from a last_used_at that nothing populates, and
  is_ttl_expired fail-safes to False when the timestamp is unknown.

This adds deterministic merged-PR linkage and gates removal on the complete
cleanup policy:

* build_pr_index / resolve_owning_pr link a worktree branch to exactly one
  owning PR. Competing PRs on one branch, a still-open owner, a head-branch
  mismatch, or missing PR state all fail closed while still reporting the
  resolved pr_number.
* assess_merged_pr_worktree_cleanup requires all of: conclusive merged
  ownership, branch agreement, containment of the head in authoritative
  master, no open/competing PR, no active lease, no issue lock, no live
  session, a clean tree, and a non-protected checkout. Unknown state blocks.
* Containment reuses merged_cleanup_reconcile.is_head_ancestor_of_ref so the
  audit and the PR-scoped reconciler agree on what "already landed" means.

Lease evidence is now supplied. audit_branches_directory already accepted
leased_branches but the MCP tool never passed it, so has_active_lease was
false for every worktree in a live run. That was inert only while issue
worktrees could never become removable; it is wired to authoritative
control-plane leases here, scoped so a lease on issue N protects that issue's
work worktree and not a baseline or review tree merely named after it.

Issue work no longer becomes removable on TTL age alone, since age is not
proof that a branch landed and would otherwise reclaim a worktree holding
unmerged commits. conflict_fix keeps its existing TTL behaviour, and review,
baseline, merge-simulation, detached, dirty, open-PR, and protected
classifications are unchanged.

The assessor still performs no deletion and gains no cleanup mutation. This
is assessor-side only and does not implement the PR-scoped executor or the
expired-lease reclaim policy tracked separately by #855.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-23 21:19:32 -04:00
19 changed files with 2407 additions and 1834 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("/")
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:
"""True when *path* resolves inside a canonical ``branches/`` directory."""
if not path or not str(path).strip():
"""True when *path* resolves inside ``<project_root>/branches/``."""
normalized = _normalize_path(path)
if not normalized:
return False
try:
real_path = os.path.realpath(os.path.abspath(str(path).strip()))
except Exception:
if "/branches/" in f"{normalized}/":
return True
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
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(
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))
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(
worktree_path: str | None,
process_project_root: str,
+1 -2
View File
@@ -247,8 +247,7 @@ def bootstrap_permits_control_checkout(
"""
if not isinstance(assessment, dict):
return False
import author_issue_bootstrap
if not is_create_issue_task(task) and not author_issue_bootstrap.is_author_issue_bootstrap_task(task):
if not is_create_issue_task(task):
return False
# 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_authorize_reconciliation_cleanup_phase`
- `gitea_authorize_review_correction`
- `gitea_bootstrap_author_issue_worktree`
- `gitea_capability_stop_terminal_report`
- `gitea_capture_branches_worktree_snapshot`
- `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 |
| `merge_pr` | controller | privileged | `gitea.pr.merge` | 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
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
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
`authorize(action_id, principal, for_execution=False)` returns a decision
@@ -199,8 +208,8 @@ breaking the request it describes.
| Class | Applies to | Default |
|-------|-----------|---------|
| `standard` | Routine gated writes | 90 days |
| `privileged` | `review_pr`, `close_pr`, and any unclassifiable action | 365 days |
| `break_glass` | `merge_pr`, `delete_branch` | 730 days |
| `privileged` | `review_pr`, `close_pr`, `system.reload_namespace`, and any unclassifiable action | 365 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
retention is auditable per record rather than inferred from file age. An
+146 -201
View File
@@ -958,32 +958,6 @@ def _create_issue_bootstrap_assessment(
"""
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):
return None
@@ -1476,7 +1450,7 @@ def verify_preflight_purity(
dirty_files = sorted(
_parse_porcelain_entries(_get_workspace_porcelain(workspace))
)
if dirty_files and task != "commit_files":
if dirty_files:
raise RuntimeError(
nwb.format_namespace_workspace_binding_error(
role_kind=role,
@@ -9036,7 +9010,6 @@ def gitea_commit_files(
host: str | None = None,
org: str | None = None,
repo: str | None = None,
worktree_path: str | None = None,
) -> dict:
"""Commit changes to multiple files in a Gitea repository in a single atomic commit.
@@ -9049,46 +9022,10 @@ def gitea_commit_files(
host: Override the Gitea host.
org: Override the owner/organization.
repo: Override the repository name.
worktree_path: Optional worktree path for author mutation context.
Returns:
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(
"commit_files"
)
@@ -9101,7 +9038,7 @@ def gitea_commit_files(
"reasons": block_reasons,
}
blocked = _namespace_mutation_block(
"commit_files", commit="", branch="", remote=remote, worktree_path=worktree_path
"commit_files", commit="", branch="", remote=remote
)
if blocked:
return blocked
@@ -9129,7 +9066,7 @@ def gitea_commit_files(
)
# #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)
h, o, r = _resolve(remote, host, org, repo)
@@ -9406,96 +9343,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 = ("merge", "squash", "rebase")
@@ -11395,25 +11242,22 @@ def gitea_reconcile_merged_cleanups(
if dry_run:
report["dry_run"] = True
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}
verify_preflight_purity(
remote, task="reconcile_merged_cleanups", org=org, repo=repo
)
actions: list[dict] = []
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 {}
project_root = _canonical_local_git_root()
if remote_assessment.get("safe_to_delete_remote"):
import urllib.parse
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
def _ownership_records_for_branch(
head_branch: str, pr_num_int: int | None
) -> list[dict]:
ownership_bundle = _collect_branch_ownership_records(
remote=remote,
host=h,
@@ -11421,7 +11265,7 @@ def gitea_reconcile_merged_cleanups(
repo=r,
branch=head_branch,
pr_number=pr_num_int,
project_root=_canonical_local_git_root(),
project_root=project_root,
auth=auth,
base_api=base,
)
@@ -11442,6 +11286,18 @@ def gitea_reconcile_merged_cleanups(
"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(
remote=remote,
org=o,
@@ -11451,8 +11307,7 @@ def gitea_reconcile_merged_cleanups(
records=ownership_records,
)
if ownership.get("block"):
actions.append(
{
return {
"action": "delete_remote_branch",
"branch": head_branch,
"success": False,
@@ -11461,13 +11316,10 @@ def gitea_reconcile_merged_cleanups(
"verified_absent": False,
"blocker_kind": "active_branch_ownership",
"reasons": ownership.get("reasons") or [],
"blocking_categories": ownership.get(
"blocking_categories"
)
or [],
"blocking_categories": ownership.get("blocking_categories") or [],
"after_worktree_removal": after_worktree_removal,
"ownership_reassessed": after_worktree_removal,
}
)
continue
encoded = urllib.parse.quote(head_branch, safe="")
url = f"{base}/branches/{encoded}"
@@ -11482,6 +11334,7 @@ def gitea_reconcile_merged_cleanups(
"branch": head_branch,
"source": "reconcile_merged_cleanups",
"ownership_checked": True,
"after_worktree_removal": after_worktree_removal,
},
):
api_request("DELETE", url, auth)
@@ -11490,8 +11343,7 @@ def gitea_reconcile_merged_cleanups(
readback
)
verified = bool(readback_assessment.get("verified_absent"))
actions.append(
{
return {
"action": "delete_remote_branch",
"branch": head_branch,
"success": bool(readback_assessment.get("ok")),
@@ -11500,22 +11352,53 @@ def gitea_reconcile_merged_cleanups(
"verified_absent": verified,
"readback": readback_assessment.get("readback"),
"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"):
result = merged_cleanup_reconcile.remove_local_worktree(
_canonical_local_git_root(),
project_root,
head_branch,
worktree_path=local_assessment.get("worktree_path"),
)
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 []:
if not scratch.get("safe_to_remove_worktree"):
continue
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})
@@ -11751,6 +11634,7 @@ def gitea_audit_worktree_cleanup(
org: str | None = None,
repo: str | None = None,
ttl_hours: float = worktree_cleanup_audit.DEFAULT_TTL_HOURS,
merged_pr_limit: int = 200,
) -> dict:
"""Read-only: classify every session-owned worktree under ``branches/`` (#401).
@@ -11761,17 +11645,26 @@ def gitea_audit_worktree_cleanup(
the active issue-lock branch is read from the local lock file and treated
as active work. Deletes nothing and mutates no Gitea state.
Fails closed if the live open-PR list cannot be fetched: without it,
removability cannot be proven, so no candidates are returned.
Merged PRs are fetched as well, so an issue worktree can be linked to the
PR that owns its branch (#858). Such a worktree only becomes removable
when that owning PR is unambiguous and merged, the worktree head is
already contained in authoritative master, and nothing else protects it
no open or competing PR, lease, issue lock, live session, dirty file, or
protected/control checkout. Anything unproven keeps it classified as
active issue work.
Fails closed if the live open-PR list, the merged-PR list, or the
control-plane lease state cannot be read: without them removability
cannot be proven, so no candidates are returned.
Args:
remote: Known instance 'dadeschools' or 'prgs'.
host: Override the Gitea host.
org: Override the owner/organization.
repo: Override the repository name.
ttl_hours: Age (hours) after which a clean issue/conflict-fix
worktree becomes stale-removable (default from
GITEA_WORKTREE_TTL_HOURS).
ttl_hours: Age (hours) after which a clean conflict-fix worktree
becomes stale-removable (default from GITEA_WORKTREE_TTL_HOURS).
merged_pr_limit: Max closed PRs scanned for merged-PR ownership.
Returns:
dict with per-worktree classifications, counts, removable
@@ -11807,22 +11700,84 @@ def gitea_audit_worktree_cleanup(
if (pr.get("head") or {}).get("ref")
}
# #858: merged PRs are the ownership evidence that lets a landed issue
# worktree stop being reported as active work. Without them the audit can
# never agree with the PR-scoped reconciler, so treat a fetch failure the
# same way an open-PR fetch failure is treated: fail closed.
try:
closed_prs = api_get_all(
f"{repo_api_url(h, o, r)}/pulls?state=closed", auth, limit=merged_pr_limit
)
except Exception as exc:
return {
"success": False,
"performed": False,
"open_pr_state_verified": True,
"merged_pr_state_verified": False,
"reasons": [
"could not fetch merged PRs; worktree ownership unverified "
f"(fail closed): {_redact(str(exc))}"
],
}
merged_prs = [pr for pr in closed_prs if (pr.get("merged") or pr.get("merged_at"))]
pr_index = worktree_cleanup_audit.build_pr_index(list(open_prs) + merged_prs)
# #858: the auditor already accepted lease evidence but nothing ever
# supplied it, so every worktree looked unleased. Removability is now
# reachable for issue worktrees, so authoritative control-plane leases
# must be readable or the audit fails closed.
db, lease_errs = _control_plane_db_or_error()
if db is None:
return {
"success": False,
"performed": False,
"open_pr_state_verified": True,
"merged_pr_state_verified": True,
"lease_state_verified": False,
"reasons": [
"could not read control-plane leases; worktree protection "
"unverified (fail closed)",
*lease_errs,
],
}
lease_result = lease_lifecycle.list_active_leases(
db, remote=remote, org=o, repo=r, include_non_active=False, limit=500
)
leased_issue_numbers: set[int] = set()
live_session_paths: set[str] = set()
for lease in lease_result.get("leases") or []:
if lease.get("work_kind") == "issue" and lease.get("work_number") is not None:
try:
leased_issue_numbers.add(int(lease["work_number"]))
except (TypeError, ValueError):
pass
if lease.get("worktree_path"):
live_session_paths.add(str(lease["worktree_path"]))
active_issue_branches: set[str] = set()
lock = merged_cleanup_reconcile.read_issue_lock(ISSUE_LOCK_FILE)
if lock and lock.get("branch_name"):
active_issue_branches.add(str(lock["branch_name"]).strip())
master_ref = f"{remote}/master" if remote in REMOTES else "origin/master"
report = worktree_cleanup_audit.audit_branches_directory(
_canonical_local_git_root(),
open_pr_branches=open_pr_branches,
active_issue_branches=active_issue_branches,
now=datetime.now(timezone.utc),
ttl_hours=ttl_hours,
pr_index=pr_index,
leased_issue_numbers=leased_issue_numbers,
live_session_paths=live_session_paths,
master_ref=master_ref,
)
return {
"success": True,
"performed": False,
"open_pr_state_verified": True,
"merged_pr_state_verified": True,
"lease_state_verified": True,
"master_ref": master_ref,
"task_mode": "work-issue",
**report,
}
@@ -19356,12 +19311,6 @@ def gitea_resolve_task_capability(
task: str,
remote: str = "dadeschools",
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:
"""Read-only / side-effect free: resolve capability, profile, and namespace for a task.
@@ -19378,14 +19327,13 @@ def gitea_resolve_task_capability(
remote: Known remote instance name.
host: Optional override for the Gitea host.
"""
task_key = task_capability_map._canonical_preflight_task(task)
TASK_MAP = task_capability_map.TASK_CAPABILITY_MAP
# Every fresh attempt invalidates the previous task/role stamp before any
# fallible resolver work. Unknown/malformed tasks and unexpected failures
# therefore remain fail-closed instead of preserving stale authority.
_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).
profile = get_profile()
h = host or (REMOTES.get(remote, {}).get("host") if remote in REMOTES else None)
@@ -19431,8 +19379,8 @@ def gitea_resolve_task_capability(
result["cleared_stale_denial"] = True
return result
required_permission = task_capability_map.required_permission(task_key)
required_role = task_capability_map.required_role(task_key)
required_permission = task_capability_map.required_permission(task)
required_role = task_capability_map.required_role(task)
role_exclusive_tasks = task_capability_map.ROLE_EXCLUSIVE_TASKS
infra_assessment = role_session_router.assess_infra_stop(PROJECT_ROOT)
@@ -19618,10 +19566,7 @@ def gitea_resolve_task_capability(
available_in_session = allowed_in_current_session
runtime_stale_blocker = False
if (
"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":
if "PYTEST_CURRENT_TEST" not in os.environ or "GITEA_FORCE_MCP_RUNTIME_CHECK" in os.environ:
runtime_reasons = _check_mcp_runtimes_diagnostics(task, matching_profiles)
if runtime_reasons:
restart_required = True
-2
View File
@@ -73,8 +73,6 @@ AUTHOR_TASKS = frozenset({
"claim_issue",
"create_branch",
"push_branch",
"bootstrap_author_issue_worktree",
"gitea_bootstrap_author_issue_worktree",
"create_pr",
"comment_pr",
"address_pr_change_requests",
+15 -13
View File
@@ -58,14 +58,6 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
"permission": "gitea.branch.create",
"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": {
"permission": "gitea.branch.push",
"role": "author",
@@ -343,6 +335,21 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
"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
# ownership in the control-plane DB (not a separate Gitea write permission).
"list_workflow_leases": {
@@ -485,8 +492,6 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
# merger lease (#763).
_PREFLIGHT_TASK_TRANSITIONS = frozenset({
("review_pr", "acquire_reviewer_pr_lease"),
("work_issue", "bootstrap_author_issue_worktree"),
("bootstrap_author_issue_worktree", "lock_issue"),
})
@@ -533,8 +538,6 @@ ROLE_EXCLUSIVE_TASKS: frozenset[str] = frozenset(
"gitea_release_merger_pr_lease",
"create_branch",
"push_branch",
"bootstrap_author_issue_worktree",
"gitea_bootstrap_author_issue_worktree",
"publish_unpublished_branch",
"create_pr",
"commit_files",
@@ -560,7 +563,6 @@ ISSUE_MUTATION_TOOL_TASKS: dict[str, str] = {
"gitea_set_issue_labels": "set_issue_labels",
"gitea_cleanup_terminal_pr_labels": "cleanup_terminal_pr_labels",
"gitea_create_label": "create_label",
"gitea_bootstrap_author_issue_worktree": "bootstrap_author_issue_worktree",
"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)
)
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):
ROOT = "/repo/Gitea-Tools"
@@ -0,0 +1,551 @@
"""Merged-PR awareness for the worktree cleanup audit (#858).
Before #858 an ``issue_work`` worktree could never leave ``active_issue_work``:
the audit had no PR linkage at all (``pr_number`` was structurally ``None``)
and its only route to ``clean_stale_removable`` was a TTL derived from a
``last_used_at`` that nothing ever populated. A merged, clean, unprotected
worktree was therefore reported as active work forever, disagreeing with the
PR-scoped reconciler.
These tests use fabricated temporary repositories and synthetic PR records
only. Nothing here removes a worktree or deletes a branch.
"""
import os
import subprocess
import sys
import tempfile
import unittest
from unittest.mock import patch
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
import merged_cleanup_reconcile as mcr # noqa: E402
import worktree_cleanup_audit as wca # noqa: E402
MERGED_BRANCH = "feat/issue-777-timeline"
MERGED_PATH = "/repo/branches/issue-777-timeline"
HEAD_SHA = "a" * 40
def _pr(number, branch, *, merged=True, sha=HEAD_SHA, state=None):
"""Synthetic Gitea PR payload."""
return {
"number": number,
"head": {"ref": branch, "sha": sha},
"merged_at": "2026-07-24T01:00:00Z" if merged else None,
"state": state or ("closed" if merged else "open"),
}
def _porcelain(*entries):
out = []
for path, branch, sha in entries:
out.append(f"worktree {path}")
out.append(f"HEAD {sha}")
if branch is None:
out.append("detached")
else:
out.append(f"branch refs/heads/{branch}")
out.append("")
return "\n".join(out)
class _AuditHarness(unittest.TestCase):
"""Runs audit_branches_directory over a fabricated worktree listing."""
PORCELAIN = _porcelain(
("/repo", "master", "f" * 40),
(MERGED_PATH, MERGED_BRANCH, HEAD_SHA),
)
def run_audit(self, *, dirty_paths=(), contained=True, **kwargs):
def fake_dirty(path):
if path in dirty_paths:
return {"exists": True, "dirty": True, "dirty_files": [" M x.py"]}
return {"exists": True, "dirty": False, "dirty_files": []}
with patch.object(
wca, "list_worktrees",
return_value=wca.parse_worktree_porcelain(self.PORCELAIN),
), patch.object(
wca, "read_worktree_dirty", side_effect=fake_dirty
), patch.object(
wca, "git_worktree_list", return_value="(mocked)"
), patch.object(
wca, "is_head_ancestor_of_ref", return_value=contained
):
report = wca.audit_branches_directory("/repo", **kwargs)
return {wt["path"]: wt for wt in report["worktrees"]}, report
def merged_audit(self, **kwargs):
kwargs.setdefault("pr_index", wca.build_pr_index([_pr(849, MERGED_BRANCH)]))
kwargs.setdefault("master_ref", "prgs/master")
return self.run_audit(**kwargs)
class TestMergedWorktreeBecomesRemovable(_AuditHarness):
def test_clean_merged_issue_worktree_is_linked_and_removable(self):
by_path, report = self.merged_audit()
entry = by_path[MERGED_PATH]
self.assertEqual(entry["classification"], wca.CLASS_CLEAN_STALE_REMOVABLE)
self.assertTrue(entry["removable"])
self.assertEqual(entry["merged_pr_linkage"]["status"], wca.LINKAGE_MERGED)
self.assertEqual(entry["merged_pr_cleanup"]["block_reasons"], [])
self.assertIn(MERGED_PATH, [c["path"] for c in report["removable_candidates"]])
def test_pr_number_populated_from_authoritative_linkage(self):
by_path, _ = self.merged_audit()
self.assertEqual(by_path[MERGED_PATH]["pr_number"], 849)
def test_regression_without_pr_evidence_stays_active_issue_work(self):
"""The pre-#858 behaviour, still correct when no PR state is supplied."""
by_path, _ = self.run_audit()
entry = by_path[MERGED_PATH]
self.assertEqual(entry["classification"], wca.CLASS_ACTIVE_ISSUE_WORK)
self.assertFalse(entry["removable"])
self.assertIsNone(entry["pr_number"])
class TestProtectiveSignalsSurvive(_AuditHarness):
def test_open_pr_worktree_is_not_removable(self):
index = wca.build_pr_index([_pr(900, MERGED_BRANCH, merged=False)])
by_path, _ = self.run_audit(
pr_index=index,
master_ref="prgs/master",
open_pr_branches={MERGED_BRANCH},
)
entry = by_path[MERGED_PATH]
self.assertEqual(entry["classification"], wca.CLASS_ACTIVE_OPEN_PR)
self.assertFalse(entry["removable"])
# linkage still reports the owning PR, it just is not merge proof
self.assertEqual(entry["merged_pr_linkage"]["status"], wca.LINKAGE_OPEN)
self.assertEqual(entry["pr_number"], 900)
def test_dirty_tracked_worktree_is_not_removable(self):
by_path, _ = self.merged_audit(dirty_paths=(MERGED_PATH,))
entry = by_path[MERGED_PATH]
self.assertEqual(entry["classification"], wca.CLASS_DIRTY_LOCAL)
self.assertFalse(entry["removable"])
self.assertIn(
"worktree has uncommitted changes",
entry["merged_pr_cleanup"]["block_reasons"],
)
def test_untracked_only_worktree_is_not_removable(self):
"""``git status --porcelain`` reports untracked files as dirty too."""
def untracked(path):
if path == MERGED_PATH:
return {"exists": True, "dirty": True, "dirty_files": ["?? scratch.txt"]}
return {"exists": True, "dirty": False, "dirty_files": []}
with patch.object(
wca, "list_worktrees",
return_value=wca.parse_worktree_porcelain(self.PORCELAIN),
), patch.object(
wca, "read_worktree_dirty", side_effect=untracked
), patch.object(
wca, "git_worktree_list", return_value="(mocked)"
), patch.object(
wca, "is_head_ancestor_of_ref", return_value=True
):
report = wca.audit_branches_directory(
"/repo",
pr_index=wca.build_pr_index([_pr(849, MERGED_BRANCH)]),
master_ref="prgs/master",
)
entry = {wt["path"]: wt for wt in report["worktrees"]}[MERGED_PATH]
self.assertEqual(entry["classification"], wca.CLASS_DIRTY_LOCAL)
self.assertFalse(entry["removable"])
def test_active_lease_by_issue_number_is_protective(self):
by_path, _ = self.merged_audit(leased_issue_numbers={777})
entry = by_path[MERGED_PATH]
self.assertTrue(entry["has_active_lease"])
self.assertEqual(entry["classification"], wca.CLASS_ACTIVE_ISSUE_WORK)
self.assertFalse(entry["removable"])
def test_active_lease_by_branch_is_protective(self):
by_path, _ = self.merged_audit(leased_branches={MERGED_BRANCH})
entry = by_path[MERGED_PATH]
self.assertTrue(entry["has_active_lease"])
self.assertFalse(entry["removable"])
def test_active_issue_lock_is_protective(self):
by_path, _ = self.merged_audit(active_issue_branches={MERGED_BRANCH})
entry = by_path[MERGED_PATH]
self.assertTrue(entry["has_active_issue_lock"])
self.assertEqual(entry["classification"], wca.CLASS_ACTIVE_ISSUE_WORK)
self.assertFalse(entry["removable"])
def test_live_session_worktree_is_protective(self):
by_path, _ = self.merged_audit(live_session_paths={MERGED_PATH})
entry = by_path[MERGED_PATH]
self.assertTrue(entry["has_live_session"])
self.assertEqual(entry["classification"], wca.CLASS_ACTIVE_ISSUE_WORK)
self.assertFalse(entry["removable"])
def test_head_not_contained_in_master_is_not_removable(self):
by_path, _ = self.merged_audit(contained=False)
entry = by_path[MERGED_PATH]
self.assertEqual(entry["classification"], wca.CLASS_ACTIVE_ISSUE_WORK)
self.assertFalse(entry["removable"])
self.assertIn(
"worktree head is not contained in authoritative master "
"(unmerged commits remain)",
entry["merged_pr_cleanup"]["block_reasons"],
)
def test_unknown_containment_fails_closed(self):
by_path, _ = self.merged_audit(contained=None)
entry = by_path[MERGED_PATH]
self.assertFalse(entry["removable"])
self.assertIn(
"containment of the worktree head in master is unknown",
entry["merged_pr_cleanup"]["block_reasons"],
)
def test_missing_master_ref_fails_closed(self):
by_path, _ = self.run_audit(
pr_index=wca.build_pr_index([_pr(849, MERGED_BRANCH)])
)
self.assertFalse(by_path[MERGED_PATH]["removable"])
def test_unmerged_owning_pr_is_not_removable(self):
index = wca.build_pr_index([_pr(901, MERGED_BRANCH, merged=False)])
by_path, _ = self.run_audit(pr_index=index, master_ref="prgs/master")
entry = by_path[MERGED_PATH]
self.assertFalse(entry["removable"])
self.assertIn(
"owning PR #901 is not merged",
entry["merged_pr_cleanup"]["block_reasons"],
)
def test_control_checkout_is_never_removable(self):
by_path, _ = self.merged_audit()
control = by_path["/repo"]
self.assertTrue(control["is_protected"])
self.assertEqual(control["classification"], wca.CLASS_UNSAFE_UNKNOWN)
self.assertFalse(control["removable"])
def test_control_checkout_not_removable_even_if_linked_and_merged(self):
"""A merged PR on the control checkout must not unlock removal."""
porcelain = _porcelain(("/repo", MERGED_BRANCH, HEAD_SHA))
with patch.object(
wca, "list_worktrees", return_value=wca.parse_worktree_porcelain(porcelain)
), patch.object(
wca, "read_worktree_dirty",
return_value={"exists": True, "dirty": False, "dirty_files": []},
), patch.object(
wca, "git_worktree_list", return_value="(mocked)"
), patch.object(
wca, "is_head_ancestor_of_ref", return_value=True
):
report = wca.audit_branches_directory(
"/repo",
pr_index=wca.build_pr_index([_pr(849, MERGED_BRANCH)]),
master_ref="prgs/master",
)
entry = report["worktrees"][0]
self.assertEqual(entry["classification"], wca.CLASS_UNSAFE_UNKNOWN)
self.assertFalse(entry["removable"])
class TestAmbiguousLinkageFailsClosed(_AuditHarness):
def test_competing_prs_on_one_branch_fail_closed(self):
index = wca.build_pr_index(
[_pr(849, MERGED_BRANCH), _pr(860, MERGED_BRANCH)]
)
by_path, _ = self.run_audit(pr_index=index, master_ref="prgs/master")
entry = by_path[MERGED_PATH]
self.assertEqual(entry["merged_pr_linkage"]["status"], wca.LINKAGE_AMBIGUOUS)
self.assertIsNone(entry["pr_number"])
self.assertEqual(entry["classification"], wca.CLASS_ACTIVE_ISSUE_WORK)
self.assertFalse(entry["removable"])
def test_merged_plus_open_pr_on_one_branch_fails_closed(self):
index = wca.build_pr_index(
[_pr(849, MERGED_BRANCH), _pr(861, MERGED_BRANCH, merged=False)]
)
by_path, _ = self.run_audit(pr_index=index, master_ref="prgs/master")
entry = by_path[MERGED_PATH]
self.assertEqual(entry["merged_pr_linkage"]["status"], wca.LINKAGE_AMBIGUOUS)
self.assertFalse(entry["removable"])
def test_no_owning_pr_fails_closed(self):
by_path, _ = self.run_audit(
pr_index=wca.build_pr_index([_pr(849, "feat/other-branch")]),
master_ref="prgs/master",
)
entry = by_path[MERGED_PATH]
self.assertEqual(entry["merged_pr_linkage"]["status"], wca.LINKAGE_NONE)
self.assertFalse(entry["removable"])
def test_malformed_pr_records_are_dropped_not_guessed(self):
index = wca.build_pr_index(
[
{"number": None, "head": {"ref": MERGED_BRANCH}},
{"number": 5, "head": {}},
{"number": "not-an-int", "head": {"ref": MERGED_BRANCH}},
]
)
self.assertEqual(index, {})
self.assertEqual(
wca.resolve_owning_pr(branch=MERGED_BRANCH, pr_index=index)["status"],
wca.LINKAGE_NONE,
)
def test_detached_worktree_has_no_branch_linkage(self):
self.assertEqual(
wca.resolve_owning_pr(branch=None, pr_index={})["status"],
wca.LINKAGE_UNKNOWN,
)
class TestUnrelatedClassificationsUnchanged(unittest.TestCase):
"""Non-issue_work worktrees keep their pre-#858 classifications."""
PORCELAIN = _porcelain(
("/repo", "master", "f" * 40),
("/repo/branches/review-pr42", "review-pr42", "2" * 40),
("/repo/branches/baseline-master-x", "baseline-master-x", "3" * 40),
("/repo/branches/conflict-fix-pr50", "conflict-fix-pr50", "4" * 40),
("/repo/branches/review-pr99", None, "5" * 40),
)
def _audit(self, **kwargs):
with patch.object(
wca, "list_worktrees",
return_value=wca.parse_worktree_porcelain(self.PORCELAIN),
), patch.object(
wca, "read_worktree_dirty",
return_value={"exists": True, "dirty": False, "dirty_files": []},
), patch.object(
wca, "git_worktree_list", return_value="(mocked)"
), patch.object(
wca, "is_head_ancestor_of_ref", return_value=True
):
report = wca.audit_branches_directory("/repo", **kwargs)
return {wt["path"]: wt for wt in report["worktrees"]}
def test_classifications_identical_with_and_without_pr_evidence(self):
without = self._audit()
with_evidence = self._audit(
pr_index=wca.build_pr_index([_pr(849, MERGED_BRANCH)]),
master_ref="prgs/master",
)
self.assertEqual(
{p: e["classification"] for p, e in without.items()},
{p: e["classification"] for p, e in with_evidence.items()},
)
def test_lease_on_issue_does_not_capture_similarly_named_scratch_trees(self):
"""A lease on issue 777 protects issue work, not baseline/review trees."""
porcelain = _porcelain(
("/repo/branches/baseline-master-issue-777", "baseline-issue-777", "7" * 40),
("/repo/branches/issue-777-timeline", MERGED_BRANCH, HEAD_SHA),
)
with patch.object(
wca, "list_worktrees", return_value=wca.parse_worktree_porcelain(porcelain)
), patch.object(
wca, "read_worktree_dirty",
return_value={"exists": True, "dirty": False, "dirty_files": []},
), patch.object(
wca, "git_worktree_list", return_value="(mocked)"
), patch.object(
wca, "is_head_ancestor_of_ref", return_value=True
):
report = wca.audit_branches_directory(
"/repo",
pr_index=wca.build_pr_index([_pr(849, MERGED_BRANCH)]),
master_ref="prgs/master",
leased_issue_numbers={777},
)
by_path = {wt["path"]: wt for wt in report["worktrees"]}
baseline = by_path["/repo/branches/baseline-master-issue-777"]
self.assertFalse(baseline["has_active_lease"])
self.assertEqual(baseline["classification"], wca.CLASS_CLEAN_STALE_REMOVABLE)
issue_work = by_path["/repo/branches/issue-777-timeline"]
self.assertTrue(issue_work["has_active_lease"])
self.assertFalse(issue_work["removable"])
def test_review_and_baseline_still_removable(self):
by_path = self._audit(
pr_index=wca.build_pr_index([]), master_ref="prgs/master"
)
self.assertEqual(
by_path["/repo/branches/review-pr42"]["classification"],
wca.CLASS_CLEAN_STALE_REMOVABLE,
)
self.assertEqual(
by_path["/repo/branches/baseline-master-x"]["classification"],
wca.CLASS_CLEAN_STALE_REMOVABLE,
)
self.assertEqual(
by_path["/repo/branches/review-pr99"]["classification"],
wca.CLASS_DETACHED_REVIEW_LEFTOVER,
)
def test_conflict_fix_ttl_behaviour_unchanged(self):
"""conflict_fix still needs only TTL expiry; #858 did not touch it."""
self.assertEqual(
wca.classify_worktree(
workflow_type=wca.WORKFLOW_CONFLICT_FIX,
is_dirty=False,
ttl_expired=True,
),
wca.CLASS_CLEAN_STALE_REMOVABLE,
)
self.assertEqual(
wca.classify_worktree(
workflow_type=wca.WORKFLOW_CONFLICT_FIX,
is_dirty=False,
ttl_expired=False,
),
wca.CLASS_ACTIVE_ISSUE_WORK,
)
def test_issue_work_ttl_alone_no_longer_grants_removal(self):
"""Age is not landing proof: TTL alone must not reclaim issue work."""
self.assertEqual(
wca.classify_worktree(
workflow_type=wca.WORKFLOW_ISSUE_WORK,
is_dirty=False,
ttl_expired=True,
),
wca.CLASS_ACTIVE_ISSUE_WORK,
)
class TestAssessorPerformsNoDeletion(_AuditHarness):
def test_audit_never_removes_a_worktree(self):
with patch.object(wca, "remove_worktree") as removal:
self.merged_audit()
removal.assert_not_called()
def test_audit_shells_out_to_no_destructive_git_command(self):
seen = []
real_run = subprocess.run
def recording_run(cmd, *args, **kwargs):
seen.append(cmd)
return real_run(["true"], *args, **kwargs)
with patch.object(subprocess, "run", side_effect=recording_run):
wca.audit_branches_directory("/nonexistent-repo-for-audit")
joined = [" ".join(c) if isinstance(c, list) else str(c) for c in seen]
for cmd in joined:
self.assertNotIn("worktree remove", cmd)
self.assertNotIn("branch -D", cmd)
self.assertNotIn("push", cmd)
class TestAgreementWithPrScopedReconciler(unittest.TestCase):
"""The audit and merged_cleanup_reconcile must agree on identical input.
Uses a real throwaway git repository so containment is computed by git
rather than asserted. Nothing outside the temporary directory is touched.
"""
def _git(self, *args):
subprocess.run(
["git", "-C", self.root, *args],
check=True,
capture_output=True,
text=True,
)
def setUp(self):
self._tmp = tempfile.TemporaryDirectory()
self.root = os.path.realpath(self._tmp.name)
self._git("init", "-b", "master", ".")
self._git("config", "user.email", "[email protected]")
self._git("config", "user.name", "Test")
with open(os.path.join(self.root, "seed.txt"), "w") as fh:
fh.write("seed\n")
self._git("add", "seed.txt")
self._git("commit", "-m", "seed")
self.branch = "feat/issue-777-timeline"
self._git("checkout", "-b", self.branch)
with open(os.path.join(self.root, "feature.txt"), "w") as fh:
fh.write("feature\n")
self._git("add", "feature.txt")
self._git("commit", "-m", "feature")
self.head_sha = subprocess.run(
["git", "-C", self.root, "rev-parse", "HEAD"],
capture_output=True, text=True, check=True,
).stdout.strip()
self._git("checkout", "master")
self._git("merge", "--no-ff", "-m", "merge feature", self.branch)
self.worktree = os.path.join(self.root, "branches", "issue-777-timeline")
self._git("worktree", "add", self.worktree, self.branch)
def tearDown(self):
self._tmp.cleanup()
def _pr_index(self):
return wca.build_pr_index(
[
{
"number": 849,
"head": {"ref": self.branch, "sha": self.head_sha},
"merged_at": "2026-07-24T01:00:00Z",
}
]
)
def _audit_entry(self):
report = wca.audit_branches_directory(
self.root, pr_index=self._pr_index(), master_ref="master"
)
return next(wt for wt in report["worktrees"] if wt["path"] == self.worktree)
def _reconciler_entry(self):
return mcr.assess_local_worktree_cleanup(
pr_number=849,
head_branch=self.branch,
merged=True,
worktree_state=mcr.resolve_cleanup_worktree_state(
project_root=self.root,
head_branch=self.branch,
issue_number=777,
pr_head_sha=self.head_sha,
target_ref="master",
),
active_lock=False,
)
def test_both_assessors_agree_the_worktree_is_safe(self):
audit_entry = self._audit_entry()
reconciler = self._reconciler_entry()
self.assertTrue(reconciler["safe_to_remove_worktree"], reconciler)
self.assertTrue(audit_entry["removable"], audit_entry)
self.assertEqual(audit_entry["pr_number"], reconciler["pr_number"])
self.assertEqual(audit_entry["merged_pr_cleanup"]["block_reasons"], [])
self.assertEqual(reconciler["block_reasons"], [])
def test_both_assessors_agree_a_dirty_worktree_is_unsafe(self):
with open(os.path.join(self.worktree, "feature.txt"), "a") as fh:
fh.write("local edit\n")
audit_entry = self._audit_entry()
reconciler = self._reconciler_entry()
self.assertFalse(audit_entry["removable"])
self.assertFalse(reconciler["safe_to_remove_worktree"])
def test_worktree_still_present_after_audit(self):
self._audit_entry()
self.assertTrue(os.path.isdir(self.worktree))
if __name__ == "__main__":
unittest.main()
@@ -139,8 +139,6 @@ EXPECTED_ROLE_EXCLUSIVE_TASKS = frozenset(
"gitea_release_merger_pr_lease",
"create_branch",
"push_branch",
"bootstrap_author_issue_worktree",
"gitea_bootstrap_author_issue_worktree",
# #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.
"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()
+24 -2
View File
@@ -134,13 +134,35 @@ class TestClassification(unittest.TestCase):
self.assertEqual(cls, wca.CLASS_ACTIVE_OPEN_PR)
self.assertFalse(wca.is_removable(cls))
def test_stale_clean_issue_worktree_removable(self):
# Scenario 5: clean issue worktree, TTL expired, no lock -> removable.
def test_stale_clean_issue_worktree_needs_merged_pr_proof(self):
# Scenario 5 (#858): age is not proof that the branch landed, so a
# TTL-expired issue worktree stays active work. Only authoritative
# merged-PR evidence makes it removable, which is what keeps a
# worktree holding unmerged commits from being reclaimed by age.
cls = wca.classify_worktree(
workflow_type=wca.WORKFLOW_ISSUE_WORK,
is_dirty=False,
ttl_expired=True,
)
self.assertEqual(cls, wca.CLASS_ACTIVE_ISSUE_WORK)
self.assertFalse(wca.is_removable(cls))
cls = wca.classify_worktree(
workflow_type=wca.WORKFLOW_ISSUE_WORK,
is_dirty=False,
ttl_expired=True,
merged_pr_cleanup={"proven": True},
)
self.assertEqual(cls, wca.CLASS_CLEAN_STALE_REMOVABLE)
self.assertTrue(wca.is_removable(cls))
def test_stale_clean_conflict_fix_worktree_removable(self):
# conflict_fix keeps the original TTL rule; #858 changed issue work only.
cls = wca.classify_worktree(
workflow_type=wca.WORKFLOW_CONFLICT_FIX,
is_dirty=False,
ttl_expired=True,
)
self.assertEqual(cls, wca.CLASS_CLEAN_STALE_REMOVABLE)
self.assertTrue(wca.is_removable(cls))
+28
View File
@@ -236,6 +236,34 @@ _ACTION_SPECS: tuple[ConsoleAction, ...] = (
phase=3,
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}
+13
View File
@@ -110,6 +110,8 @@ def _format_target(action_id: str, params: dict[str, Any]) -> str:
)
if action_id == "create_issue":
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"
@@ -165,6 +167,17 @@ def build_action_registry() -> ActionRegistry:
"gitea_create_issue_comment", "Post a PR review thread comment."),
("close_pr", "Close PR", "close_pr", "gitea_edit_pr",
"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(
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,
}
+271 -4
View File
@@ -34,7 +34,11 @@ import subprocess
from datetime import datetime, timezone
from typing import Any
from merged_cleanup_reconcile import branch_worktree_folder, read_local_worktree_state
from merged_cleanup_reconcile import (
branch_worktree_folder,
is_head_ancestor_of_ref,
read_local_worktree_state,
)
from reviewer_worktree import parse_dirty_tracked_files, REVIEW_WORKTREE_RE
PROTECTED_BRANCHES = frozenset({"master", "main", "dev"})
@@ -67,6 +71,14 @@ REMOVABLE_CLASSES = frozenset(
{CLASS_CLEAN_STALE_REMOVABLE, CLASS_DETACHED_REVIEW_LEFTOVER}
)
# Merged-PR linkage outcomes for issue worktrees (#858). Only ``LINKAGE_MERGED``
# is ownership proof; every other outcome leaves the worktree protected.
LINKAGE_MERGED = "merged_pr"
LINKAGE_OPEN = "open_pr"
LINKAGE_NONE = "no_owning_pr"
LINKAGE_AMBIGUOUS = "ambiguous"
LINKAGE_UNKNOWN = "unknown"
_ISSUE_REF_RE = re.compile(r"issue-(\d+)", re.IGNORECASE)
_ISSUE_BRANCH_PREFIXES = ("feat/", "fix/", "docs/", "chore/")
@@ -169,6 +181,186 @@ def is_ttl_expired(
return (now_dt - last).total_seconds() > ttl_hours * 3600.0
def build_pr_index(prs: list[dict[str, Any]] | None) -> dict[str, list[dict[str, Any]]]:
"""Index PR records by head branch for deterministic worktree linkage (#858).
Accepts Gitea PR payloads (``head`` as a dict) and pre-flattened records
(``head_branch``/``head_sha``). Records without a usable head branch or
number are dropped rather than guessed at, so a branch is only ever linked
to a PR the caller actually proved.
"""
index: dict[str, list[dict[str, Any]]] = {}
for pr in prs or []:
head = pr.get("head")
if isinstance(head, dict):
head_branch = head.get("ref")
head_sha = head.get("sha")
else:
head_branch = pr.get("head_branch") or (head if isinstance(head, str) else None)
head_sha = pr.get("head_sha")
number = pr.get("number")
if not head_branch or number is None:
continue
try:
pr_number = int(number)
except (TypeError, ValueError):
continue
index.setdefault(str(head_branch).strip(), []).append(
{
"pr_number": pr_number,
"head_branch": str(head_branch).strip(),
"head_sha": head_sha,
"merged": bool(pr.get("merged") or pr.get("merged_at")),
"state": pr.get("state"),
}
)
return index
def resolve_owning_pr(
*,
branch: str | None,
pr_index: dict[str, list[dict[str, Any]]] | None,
) -> dict[str, Any]:
"""Resolve the single PR that owns ``branch``, failing closed when unclear.
Ownership is only ``LINKAGE_MERGED`` when exactly one PR claims the branch
and that PR is merged. Several distinct PRs on one branch is a competing
claim (``LINKAGE_AMBIGUOUS``), and a still-open owner is reported as
``LINKAGE_OPEN`` — both keep the worktree protected while still exposing
the PR number the audit resolved.
"""
if pr_index is None:
return {
"status": LINKAGE_UNKNOWN,
"pr_number": None,
"candidate_pr_numbers": [],
"reasons": ["live PR state was not supplied; ownership unproven"],
}
branch_name = (branch or "").strip()
if not branch_name:
return {
"status": LINKAGE_UNKNOWN,
"pr_number": None,
"candidate_pr_numbers": [],
"reasons": ["worktree has no attached branch; ownership unproven"],
}
candidates = list(pr_index.get(branch_name) or [])
numbers = sorted({c["pr_number"] for c in candidates})
if not candidates:
return {
"status": LINKAGE_NONE,
"pr_number": None,
"candidate_pr_numbers": [],
"reasons": [f"no PR claims branch '{branch_name}'"],
}
if len(numbers) > 1:
return {
"status": LINKAGE_AMBIGUOUS,
"pr_number": None,
"candidate_pr_numbers": numbers,
"reasons": [
f"branch '{branch_name}' is claimed by competing PRs {numbers}; "
"ownership is ambiguous"
],
}
owner = candidates[0]
pr_number = owner["pr_number"]
if owner.get("head_branch") != branch_name:
return {
"status": LINKAGE_UNKNOWN,
"pr_number": pr_number,
"candidate_pr_numbers": numbers,
"reasons": [
f"PR #{pr_number} head branch '{owner.get('head_branch')}' does not "
f"match worktree branch '{branch_name}'"
],
}
if not owner.get("merged"):
return {
"status": LINKAGE_OPEN,
"pr_number": pr_number,
"candidate_pr_numbers": numbers,
"pr_head_sha": owner.get("head_sha"),
"reasons": [f"owning PR #{pr_number} is not merged"],
}
return {
"status": LINKAGE_MERGED,
"pr_number": pr_number,
"candidate_pr_numbers": numbers,
"pr_head_sha": owner.get("head_sha"),
"reasons": [],
}
def assess_merged_pr_worktree_cleanup(
*,
linkage: dict[str, Any] | None,
head_sha: str | None,
head_in_master: bool | None,
is_dirty: bool,
has_open_pr: bool,
has_active_lease: bool,
has_active_issue_lock: bool,
is_protected: bool,
has_live_session: bool = False,
) -> dict[str, Any]:
"""Decide whether a merged issue worktree satisfies the full cleanup policy.
Every condition must be independently proven: conclusive merged-PR
ownership, agreement between the worktree branch and the PR head branch,
containment of the worktree head in authoritative master (which is what
proves no unmerged commits remain), absence of any open/competing PR,
lease, issue lock, or live session, a clean tree, and a worktree that is
not the protected control checkout. Anything unknown blocks.
"""
link = linkage or {
"status": LINKAGE_UNKNOWN,
"pr_number": None,
"reasons": ["no linkage assessment supplied"],
}
status = link.get("status")
reasons: list[str] = []
if status != LINKAGE_MERGED:
reasons.extend(
link.get("reasons") or ["owning PR could not be conclusively identified"]
)
if is_protected:
reasons.append("worktree is protected or the stable control checkout")
if is_dirty:
reasons.append("worktree has uncommitted changes")
if has_open_pr:
reasons.append("worktree branch has an open PR")
if has_active_lease:
reasons.append("worktree has an active lease")
if has_active_issue_lock:
reasons.append("an active issue lock references this branch")
if has_live_session:
reasons.append("a live process or session is using this worktree")
if not head_sha:
reasons.append("worktree head sha is unknown")
if head_in_master is None:
reasons.append("containment of the worktree head in master is unknown")
elif not head_in_master:
reasons.append(
"worktree head is not contained in authoritative master "
"(unmerged commits remain)"
)
proven = not reasons
return {
"linkage_status": status,
"pr_number": link.get("pr_number"),
"pr_head_sha": link.get("pr_head_sha"),
"head_in_master": head_in_master,
"proven": proven,
"block_reasons": reasons,
}
def classify_worktree(
*,
workflow_type: str,
@@ -181,6 +373,8 @@ def classify_worktree(
ttl_expired: bool = False,
is_protected: bool = False,
metadata_known: bool = True,
merged_pr_cleanup: dict[str, Any] | None = None,
has_live_session: bool = False,
) -> str:
"""Classify a worktree, safety-first: any preservation signal wins.
@@ -199,6 +393,8 @@ def classify_worktree(
return CLASS_ACTIVE_ISSUE_WORK # never auto-deleted (criterion 8)
if has_active_issue_lock:
return CLASS_ACTIVE_ISSUE_WORK
if has_live_session:
return CLASS_ACTIVE_ISSUE_WORK # a live session still owns this tree
if not metadata_known or workflow_type == WORKFLOW_UNKNOWN:
return CLASS_UNSAFE_UNKNOWN # never auto-deleted without proof
@@ -207,7 +403,15 @@ def classify_worktree(
if is_detached or branch_gone:
return CLASS_DETACHED_REVIEW_LEFTOVER
return CLASS_CLEAN_STALE_REMOVABLE
# issue_work / conflict_fix: only removable once the TTL has expired.
if workflow_type == WORKFLOW_ISSUE_WORK:
# #858: an issue worktree becomes removable only on authoritative
# merged-PR evidence satisfying the whole cleanup policy. Age alone
# never proves the branch landed, so TTL cannot qualify one by itself
# — otherwise a worktree holding unmerged commits would be reclaimed.
if (merged_pr_cleanup or {}).get("proven"):
return CLASS_CLEAN_STALE_REMOVABLE
return CLASS_ACTIVE_ISSUE_WORK
# conflict_fix: only removable once the TTL has expired.
if ttl_expired:
return CLASS_CLEAN_STALE_REMOVABLE
return CLASS_ACTIVE_ISSUE_WORK
@@ -400,6 +604,20 @@ def remove_worktree(project_root: str, path: str) -> dict[str, Any]:
}
def head_contained_in_ref(
project_root: str, head_sha: str | None, ref: str | None
) -> bool | None:
"""Return True when ``head_sha`` is already contained in ``ref``.
Shares :mod:`merged_cleanup_reconcile`'s ancestry check so the audit and
the PR-scoped reconciler agree on what "already landed" means (#858).
Returns None when containment cannot be determined, which fails closed.
"""
if not head_sha or not ref:
return None
return is_head_ancestor_of_ref(project_root, head_sha, ref)
def _is_under_branches(project_root: str, path: str) -> bool:
branches_root = os.path.join(os.path.abspath(project_root), "branches")
return os.path.abspath(path or "").startswith(branches_root + os.sep)
@@ -413,16 +631,30 @@ def audit_branches_directory(
active_issue_branches: set[str] | None = None,
now: datetime | str | None = None,
ttl_hours: float = DEFAULT_TTL_HOURS,
pr_index: dict[str, list[dict[str, Any]]] | None = None,
leased_issue_numbers: set[int] | None = None,
live_session_paths: set[str] | None = None,
master_ref: str | None = None,
) -> dict[str, Any]:
"""Classify every session-owned worktree under ``branches/``.
Read-only: shells out to git for discovery and dirty state, then applies
the pure classifier. Returns per-worktree classifications, counts, the
list of removable candidates, and the ``git worktree list`` proof.
``pr_index`` (see :func:`build_pr_index`) supplies the authoritative PR
ownership used to link issue worktrees to their merged PR (#858).
``master_ref`` is the ref a worktree head must be contained in before it
can be considered landed. Both are optional and their absence only ever
fails closed: without them no issue worktree becomes removable.
"""
open_pr_branches = open_pr_branches or set()
leased_branches = leased_branches or set()
active_issue_branches = active_issue_branches or set()
leased_issue_numbers = leased_issue_numbers or set()
live_session_paths = {
os.path.abspath(p) for p in (live_session_paths or set()) if p
}
worktrees: list[dict[str, Any]] = []
for entry in list_worktrees(project_root):
@@ -433,12 +665,42 @@ def audit_branches_directory(
)
dirty_state = read_worktree_dirty(path)
is_dirty = bool(dirty_state.get("dirty"))
head_sha = entry.get("head")
linkage = resolve_owning_pr(branch=branch, pr_index=pr_index)
metadata = build_worktree_metadata(
path=path, branch=branch, head_sha=entry.get("head")
path=path,
branch=branch,
head_sha=head_sha,
pr_number=linkage.get("pr_number"),
)
has_open_pr = bool(branch) and branch in open_pr_branches
has_active_lease = bool(branch) and branch in leased_branches
# A lease on issue N protects that issue's own work worktree. It must
# not incidentally protect a baseline/review scratch tree that merely
# carries the same issue marker in its name, which would change the
# classification of worktrees this policy does not own.
has_active_lease = (bool(branch) and branch in leased_branches) or (
metadata["workflow_type"] == WORKFLOW_ISSUE_WORK
and metadata.get("issue_number") is not None
and metadata["issue_number"] in leased_issue_numbers
)
has_active_lock = bool(branch) and branch in active_issue_branches
has_live_session = bool(path) and os.path.abspath(path) in live_session_paths
head_in_master = (
head_contained_in_ref(project_root, head_sha, master_ref)
if master_ref
else None
)
merged_pr_cleanup = assess_merged_pr_worktree_cleanup(
linkage=linkage,
head_sha=head_sha,
head_in_master=head_in_master,
is_dirty=is_dirty,
has_open_pr=has_open_pr,
has_active_lease=has_active_lease,
has_active_issue_lock=has_active_lock,
is_protected=is_protected,
has_live_session=has_live_session,
)
ttl_expired = is_ttl_expired(
last_used_at=metadata.get("last_used_at"), now=now, ttl_hours=ttl_hours
)
@@ -452,6 +714,8 @@ def audit_branches_directory(
branch_gone=branch is None and not entry.get("detached"),
ttl_expired=ttl_expired,
is_protected=is_protected,
merged_pr_cleanup=merged_pr_cleanup,
has_live_session=has_live_session,
)
metadata["cleanup_eligibility"] = classification
worktrees.append(
@@ -463,7 +727,10 @@ def audit_branches_directory(
"has_open_pr": has_open_pr,
"has_active_lease": has_active_lease,
"has_active_issue_lock": has_active_lock,
"has_live_session": has_live_session,
"is_protected": is_protected,
"merged_pr_linkage": linkage,
"merged_pr_cleanup": merged_pr_cleanup,
"classification": classification,
"removable": is_removable(classification),
}