Compare commits

..
Author SHA1 Message Date
sysadmin 3fd02a3c63 fix: anti-stomp capability auth, inventory wiring, side-effect proof (#604)
Address REQUEST_CHANGES on PR #680:

Blocker A — role vs capability agreement
- authorization_compatible allows reviewer/merger when they hold the task's
  required permission (e.g. gitea.issue.comment on comment_issue/mark_issue/
  set_issue_labels/lock_issue) without broad role-bypass.
- Unauthorized escalation without the permission remains fail closed.
- Regression coverage for allowed and denied reviewer/merger cases.

Blocker B — MUTATION_TASKS matches runtime wiring
- Remove unenforced entries; document dedicated-gate exclusions
  (mark_final_review_decision, save/resume_review_draft, etc.).
- Wire non-closing gitea_edit_pr as edit_pr; acquire lease uses
  acquire_reviewer_pr_lease task through verify_preflight_purity.
- Inventory↔wiring consistency tests replace set-membership-only coverage.

Blocker C — entrypoint side-effect ordering
- Behavioral tests for merge_pr, submit_pr_review, and comment_issue prove
  the assessor block aborts before Gitea API mutation and local durable writes.

Validation: 43 focused anti-stomp + related suites green; full suite
2639 passed / 6 skipped.

Closes #604
2026-07-12 09:00:45 -04:00
sysadmin ea8e186549 feat: common anti-stomp preflight before mutation tools (Closes #604)
Add a shared fail-closed anti-stomp preflight that mutation tools invoke
before create/comment/lease/review/approve/request-changes/merge/cleanup/
label mutations. Composes repo/role/root/worktree/lease/terminal-lock/
head-SHA/stale-runtime/workflow-hash/contamination checks into a typed
blocker with an exact next action. No agent bypass flags.
2026-07-12 05:10:30 -04:00
19 changed files with 2319 additions and 5083 deletions
+807
View File
@@ -0,0 +1,807 @@
"""Common anti-stomp preflight for every MCP mutation tool (#604).
Even with leases, mutation tools need a **shared** preflight that prevents
stale sessions, wrong worktrees, wrong repos, old prompts, terminal locks,
foreign leases, contaminated approvals, and root-checkout mutations from
slipping through.
This module is the pure assessment core. Callers (MCP mutation entrypoints)
gather live facts and pass them in — nothing here performs git, network, or
durable-state I/O. Existing happy-path guards remain authoritative; this
module composes them into one typed, fail-closed result.
Required checks (issue #604):
* repo/org verification
* profile/role verification
* root checkout clean and not used for mutation (when role requires branches/)
* worktree under branches when required
* active lease ownership (when lease is required for the mutation)
* terminal lock status (when applicable)
* expected head SHA (when pinned)
* stale runtime status
* workflow hash (when a workflow load is required)
* source contamination status
* no manual state/mtime/source-file bypass
Failure returns a typed blocker and an exact next action. There is **no**
agent-facing bypass flag.
"""
from __future__ import annotations
from typing import Any
import author_mutation_worktree
import master_parity_gate
import remote_repo_guard
import root_checkout_guard
import stable_branch_push_guard
# ── public constants ──────────────────────────────────────────────────────────
# Typed blocker kinds returned in the structured result.
BLOCKER_WRONG_REPO = "wrong_repo"
BLOCKER_WRONG_ROLE = "wrong_role"
BLOCKER_ROOT_CHECKOUT = "root_checkout_mutation"
BLOCKER_WRONG_WORKTREE = "wrong_worktree"
BLOCKER_FOREIGN_LEASE = "foreign_lease"
BLOCKER_TERMINAL_LOCK = "terminal_lock"
BLOCKER_HEAD_SHA = "head_sha_mismatch"
BLOCKER_STALE_RUNTIME = "stale_runtime"
BLOCKER_WORKFLOW_HASH = "workflow_hash"
BLOCKER_SOURCE_CONTAMINATION = "source_contamination"
BLOCKER_MANUAL_BYPASS = "manual_bypass"
BLOCKER_KINDS = frozenset({
BLOCKER_WRONG_REPO,
BLOCKER_WRONG_ROLE,
BLOCKER_ROOT_CHECKOUT,
BLOCKER_WRONG_WORKTREE,
BLOCKER_FOREIGN_LEASE,
BLOCKER_TERMINAL_LOCK,
BLOCKER_HEAD_SHA,
BLOCKER_STALE_RUNTIME,
BLOCKER_WORKFLOW_HASH,
BLOCKER_SOURCE_CONTAMINATION,
BLOCKER_MANUAL_BYPASS,
})
# Mutation tasks that must invoke the shared anti-stomp preflight before
# acting (issue #604 AC1). Every member must appear as a live task= kwarg
# to verify_preflight_purity / _run_anti_stomp_preflight (or the review
# anti_task map) in gitea_mcp_server.py. Inventory↔wiring tests fail if
# a declared task is not wired.
MUTATION_TASKS = frozenset({
"create_issue",
"comment_issue",
"close_issue",
"mark_issue",
"lock_issue",
"set_issue_labels",
"create_label",
"create_pr",
"close_pr",
"edit_pr",
"commit_files",
"gitea_commit_files",
"delete_branch",
"cleanup_merged_pr_branch",
"cleanup_stale_claims",
"reconcile_merged_cleanups",
"reconcile_already_landed_pr",
"reconcile_close_superseded_pr",
"post_heartbeat",
"acquire_reviewer_pr_lease",
"gitea_acquire_reviewer_pr_lease",
"adopt_merger_pr_lease",
"review_pr",
"submit_pr_review",
"approve_pr",
"request_changes_pr",
"merge_pr",
})
# Intentionally excluded from MUTATION_TASKS. Each has a dedicated
# fail-closed gate that preserves decision-lock ownership, workflow-hash,
# head-SHA, and repository consistency. Do not re-add without either
# wiring shared preflight or updating this rationale (#604 Blocker B).
DEDICATED_GATE_MUTATIONS: dict[str, str] = {
"mark_final_review_decision": (
"Local review-decision lock only. Enforced by session lock ownership, "
"workflow-hash, expected head SHA, PR work lease, eligibility, and "
"terminal-lock gates. No Gitea review POST; shared anti-stomp would "
"duplicate without side-effect-ordering value."
),
"save_review_draft": (
"Local session-state draft save with live PR head/base consistency "
"checks. Not a Gitea review/merge mutation; dedicated head-SHA and "
"worktree resolution gates apply."
),
"resume_review_draft": (
"Live-state resume with terminal lock, lease ownership, head/base SHA, "
"and parity checks. When submit=True, delegates to gitea_submit_pr_review "
"which runs shared anti-stomp before the Gitea review POST."
),
"cleanup_stale_review_decision_lock": (
"Moot-lock cleanup with identity match, live PR merged/closed proof, "
"and reviewer capability gate (#594). Distinct from shared anti-stomp."
),
"gitea_cleanup_stale_review_decision_lock": (
"Alias of cleanup_stale_review_decision_lock; dedicated #594 path."
),
"heartbeat_reviewer_pr_lease": (
"Lease heartbeat posts via verify_preflight_purity(task='review_pr') "
"plus in-session lease ownership checks; not a separate mutation class."
),
"release_reviewer_pr_lease": (
"Lease release uses workspace binding + ownership checks; posts only "
"when the session owns the active lease."
),
}
# Roles that must mutate from a branches/ worktree (not the control checkout).
_BRANCHES_REQUIRED_ROLES = frozenset({"author"})
# Reviewer/merger mutations that require an owned lease + head pinning when
# the caller supplies those facts.
_LEASE_AWARE_TASKS = frozenset({
"review_pr",
"submit_pr_review",
"approve_pr",
"request_changes_pr",
"merge_pr",
"acquire_reviewer_pr_lease",
"gitea_acquire_reviewer_pr_lease",
"adopt_merger_pr_lease",
})
_NEXT_ACTIONS: dict[str, str] = {
BLOCKER_WRONG_REPO: (
"Pass explicit org= and repo= matching the local git remote "
"(e.g. org=Scaled-Tech-Consulting repo=Gitea-Tools) and retry."
),
BLOCKER_WRONG_ROLE: (
"Call gitea_resolve_task_capability, switch to the required "
"namespace/profile, re-verify with gitea_whoami, then retry."
),
BLOCKER_ROOT_CHECKOUT: (
"Leave the root control checkout on clean master; create or switch "
"to a session-owned worktree under branches/ and retry the mutation "
"with worktree_path set."
),
BLOCKER_WRONG_WORKTREE: (
"Create or bind a session-owned worktree under branches/, set "
"worktree_path (or GITEA_ACTIVE_WORKTREE), and retry."
),
BLOCKER_FOREIGN_LEASE: (
"Stop. Do not stomp a foreign lease. Wait for expiry, request "
"takeover through the sanctioned path, or hand off to the owner."
),
BLOCKER_TERMINAL_LOCK: (
"Stop. A terminal review-decision lock is active for this head. "
"Do not re-approve/merge; follow the #332/#620 recovery path."
),
BLOCKER_HEAD_SHA: (
"Re-fetch the live PR head with gitea_view_pr, re-pin "
"expected_head_sha to the current head, re-validate, then retry."
),
BLOCKER_STALE_RUNTIME: (
"Restart the Gitea MCP server so it reloads master's capability "
"gates, re-run gitea_whoami + gitea_resolve_task_capability, then retry."
),
BLOCKER_WORKFLOW_HASH: (
"Reload the canonical workflow via gitea_load_review_workflow, then "
"rerun the full review-merge workflow from inventory (no approve/merge replay)."
),
BLOCKER_SOURCE_CONTAMINATION: (
"Stop. Session is source-contaminated. Hand off to a reconciler for "
"audit/clear; do not clear markers by deleting session-state files."
),
BLOCKER_MANUAL_BYPASS: (
"Stop. Manual state/mtime/source-file bypass is forbidden. Use "
"sanctioned MCP tools only; never delete or rewrite session-state "
"or lock files by hand."
),
}
def is_mutation_task(task: str | None) -> bool:
"""True when *task* is in the #604 mutation set (normalized)."""
name = (task or "").strip().lower().removeprefix("gitea_")
if not name:
return False
if name in MUTATION_TASKS:
return True
# Accept gitea_ prefixed membership for aliases already in the set.
return f"gitea_{name}" in MUTATION_TASKS
def roles_compatible(active_role: str | None, required_role: str | None) -> bool:
"""Whether *active_role* may perform a task that maps to *required_role*.
Exact match always passes. Reconcilers may run author-class mutations
(comment/close/cleanup) that the capability map stamps as ``author`` —
matching the long-standing reconciler exemption for control-checkout
work. No other cross-role substitutions are allowed.
Capability-authorized multi-role tasks (e.g. reviewer holding
``gitea.issue.comment`` for ``comment_issue``) are handled by
:func:`authorization_compatible`, not by expanding this role matrix.
"""
active = (active_role or "").strip().lower()
required = (required_role or "").strip().lower()
if not active or not required:
return True
if active == required:
return True
if active == "reconciler" and required in {"author", "reconciler"}:
return True
return False
def authorization_compatible(
active_role: str | None,
required_role: str | None,
*,
required_permission: str | None = None,
allowed_operations: Any = None,
) -> bool:
"""Whether the active session may run a task given role *and* capability.
Order:
1. :func:`roles_compatible` (exact role or narrow reconciler→author).
2. Capability possession: when *required_permission* is non-empty and
present in *allowed_operations*, allow even if the nominal task role
differs (e.g. reviewer/merger with ``gitea.issue.comment`` on
``comment_issue`` / ``mark_issue`` / ``set_issue_labels`` /
``lock_issue``).
Fail closed otherwise. This is **not** a broad reviewer→author or
merger→author role rewrite: without the specific permission the check
still denies. Missing *allowed_operations* when a permission is required
also fails closed (callers must supply the live profile op list).
"""
if roles_compatible(active_role, required_role):
return True
perm = (required_permission or "").strip()
if not perm:
return False
if allowed_operations is None:
return False
try:
ops = {str(o).strip() for o in allowed_operations if o is not None}
except TypeError:
return False
return perm in ops
def _blocker(
kind: str,
reasons: list[str],
*,
exact_next_action: str | None = None,
detail: dict | None = None,
) -> dict[str, Any]:
if kind not in BLOCKER_KINDS:
raise ValueError(f"unknown anti-stomp blocker kind: {kind!r}")
return {
"kind": kind,
"reasons": list(reasons),
"exact_next_action": exact_next_action or _NEXT_ACTIONS[kind],
"detail": dict(detail or {}),
}
def _allowed_result(checks: dict[str, Any]) -> dict[str, Any]:
return {
"allowed": True,
"block": False,
"blockers": [],
"reasons": [],
"exact_next_action": "proceed",
"blocker_kind": None,
"checks": checks,
}
def _blocked_result(
blockers: list[dict[str, Any]],
checks: dict[str, Any],
) -> dict[str, Any]:
primary = blockers[0]
all_reasons: list[str] = []
for b in blockers:
for r in b.get("reasons") or []:
if r not in all_reasons:
all_reasons.append(r)
return {
"allowed": False,
"block": True,
"blockers": blockers,
"reasons": all_reasons,
"exact_next_action": primary["exact_next_action"],
"blocker_kind": primary["kind"],
"checks": checks,
}
def assess_anti_stomp_preflight(
*,
task: str | None = None,
# repo/org
remote: str | None = None,
resolved_org: str | None = None,
resolved_repo: str | None = None,
local_remote_url: str | None = None,
org_explicit: bool = False,
repo_explicit: bool = False,
check_repo: bool = True,
# profile/role + capability
profile_name: str | None = None,
profile_role: str | None = None,
required_role: str | None = None,
required_permission: str | None = None,
allowed_operations: Any = None,
check_role: bool = True,
# root checkout + worktree
workspace_path: str | None = None,
project_root: str | None = None,
current_branch: str | None = None,
root_head_sha: str | None = None,
root_porcelain: str | None = None,
remote_master_sha: str | None = None,
check_root_checkout: bool = True,
check_worktree: bool = True,
# stale runtime (master parity)
startup_head: str | None = None,
current_code_head: str | None = None,
check_stale_runtime: bool = True,
# lease ownership
lease_required: bool = False,
foreign_lease: bool | None = None,
lease_owner_session: str | None = None,
active_session_id: str | None = None,
lease_owner_identity: str | None = None,
active_identity: str | None = None,
lease_reasons: list[str] | None = None,
# terminal lock
terminal_lock_blocks: bool | None = None,
terminal_lock_reasons: list[str] | None = None,
# expected head SHA
require_head_sha: bool = False,
expected_head_sha: str | None = None,
live_head_sha: str | None = None,
# workflow hash
workflow_hash_valid: bool | None = None,
workflow_hash_reasons: list[str] | None = None,
# source contamination (stable-branch push / approval contamination)
source_contaminated: bool | None = None,
contamination_reasons: list[str] | None = None,
# manual bypass
manual_bypass_attempted: bool = False,
manual_bypass_reasons: list[str] | None = None,
) -> dict[str, Any]:
"""Fail-closed common anti-stomp assessment (pure).
Only checks for which facts are supplied (or which are required by the
mutation category) are evaluated. Unsupplied optional facts are recorded
as ``skipped`` so callers can see coverage without inventing defaults.
Returns a structured dict with ``allowed``/``block``, typed ``blockers``,
aggregated ``reasons``, ``exact_next_action``, and per-check ``checks``.
"""
task_name = (task or "").strip()
role = (profile_role or "").strip().lower() or None
req_role = (required_role or "").strip().lower() or None
req_perm = (required_permission or "").strip() or None
checks: dict[str, Any] = {
"task": task_name or None,
"profile_name": profile_name,
"profile_role": role,
"required_role": req_role,
"required_permission": req_perm,
}
blockers: list[dict[str, Any]] = []
# ── manual bypass (always first; never skippable when attempted) ─────────
if manual_bypass_attempted:
reasons = list(manual_bypass_reasons or [
"manual state/mtime/source-file bypass attempt detected (fail closed)"
])
blockers.append(_blocker(BLOCKER_MANUAL_BYPASS, reasons))
checks["manual_bypass"] = {"block": True, "reasons": reasons}
else:
checks["manual_bypass"] = {"block": False, "skipped": False}
# ── repo/org ─────────────────────────────────────────────────────────────
if check_repo and resolved_org is not None and resolved_repo is not None:
repo_assessment = remote_repo_guard.assess_remote_repo_match(
remote=remote or "",
resolved_org=resolved_org,
resolved_repo=resolved_repo,
local_remote_url=local_remote_url,
org_explicit=org_explicit,
repo_explicit=repo_explicit,
)
checks["repo"] = {
"block": bool(repo_assessment.get("block")),
"reasons": list(repo_assessment.get("reasons") or []),
"resolved_org": resolved_org,
"resolved_repo": resolved_repo,
}
if repo_assessment.get("block"):
blockers.append(
_blocker(
BLOCKER_WRONG_REPO,
list(repo_assessment.get("reasons") or ["repo/org mismatch"]),
detail={
"resolved_org": resolved_org,
"resolved_repo": resolved_repo,
"local_remote_url": local_remote_url,
},
)
)
else:
checks["repo"] = {"block": False, "skipped": True}
# ── profile/role + capability ────────────────────────────────────────────
# Role match OR possession of the task's required permission (Blocker A).
# Reviewer/merger may run issue-comment-class tasks when they hold
# gitea.issue.comment; unauthorized escalation without the permission
# still fails closed.
if check_role and req_role and role:
authorized = authorization_compatible(
role,
req_role,
required_permission=req_perm,
allowed_operations=allowed_operations,
)
if not authorized:
perm_clause = (
f", required_permission='{req_perm}'" if req_perm else ""
)
reasons = [
f"active profile role '{role}' is not authorized for task "
f"'{task_name or '(unknown)'}' (required_role='{req_role}'"
f"{perm_clause}; profile={profile_name or '(unknown)'})"
]
checks["role"] = {
"block": True,
"reasons": reasons,
"required_permission": req_perm,
"capability_authorized": False,
}
blockers.append(
_blocker(
BLOCKER_WRONG_ROLE,
reasons,
detail={
"profile_name": profile_name,
"profile_role": role,
"required_role": req_role,
"required_permission": req_perm,
},
)
)
else:
checks["role"] = {
"block": False,
"reasons": [],
"required_permission": req_perm,
"capability_authorized": bool(
req_perm
and allowed_operations is not None
and req_perm in {
str(o).strip() for o in (allowed_operations or [])
if o is not None
}
and not roles_compatible(role, req_role)
),
}
else:
checks["role"] = {"block": False, "skipped": True}
# ── stale runtime (master parity) ────────────────────────────────────────
if check_stale_runtime and (
startup_head is not None or current_code_head is not None
):
parity = master_parity_gate.assess_master_parity(
{"startup_head": startup_head},
current_code_head,
)
stale_reasons = master_parity_gate.parity_block_reasons(parity)
checks["stale_runtime"] = {
"block": bool(stale_reasons),
"reasons": list(stale_reasons),
"startup_head": startup_head,
"current_code_head": current_code_head,
"stale": bool(parity.get("stale")),
}
if stale_reasons:
blockers.append(
_blocker(
BLOCKER_STALE_RUNTIME,
list(stale_reasons),
detail={
"startup_head": startup_head,
"current_code_head": current_code_head,
},
)
)
else:
checks["stale_runtime"] = {"block": False, "skipped": True}
# ── root checkout ────────────────────────────────────────────────────────
if (
check_root_checkout
and workspace_path is not None
and project_root is not None
and root_porcelain is not None
):
root_assessment = root_checkout_guard.assess_root_checkout_guard(
workspace_path=workspace_path,
canonical_repo_root=project_root,
current_branch=current_branch,
head_sha=root_head_sha,
porcelain_status=root_porcelain,
remote_master_sha=remote_master_sha,
resolved_role=req_role or role,
actual_role=role,
)
checks["root_checkout"] = {
"block": bool(root_assessment.get("block")),
"reasons": list(root_assessment.get("reasons") or []),
}
if root_assessment.get("block"):
blockers.append(
_blocker(
BLOCKER_ROOT_CHECKOUT,
list(root_assessment.get("reasons") or [
"control checkout is not clean master"
]),
detail={
"workspace_path": workspace_path,
"project_root": project_root,
"current_branch": current_branch,
},
)
)
else:
checks["root_checkout"] = {"block": False, "skipped": True}
# ── worktree under branches (author) ─────────────────────────────────────
worktree_role = role or req_role
if (
check_worktree
and workspace_path is not None
and project_root is not None
and worktree_role in _BRANCHES_REQUIRED_ROLES
):
wt = author_mutation_worktree.assess_author_mutation_worktree(
workspace_path=workspace_path,
project_root=project_root,
current_branch=current_branch,
)
checks["worktree"] = {
"block": bool(wt.get("block")),
"reasons": list(wt.get("reasons") or []),
"under_branches": wt.get("under_branches"),
}
if wt.get("block"):
blockers.append(
_blocker(
BLOCKER_WRONG_WORKTREE,
list(wt.get("reasons") or [
"mutation worktree is not under branches/"
]),
detail={
"workspace_path": workspace_path,
"project_root": project_root,
},
)
)
else:
checks["worktree"] = {
"block": False,
"skipped": worktree_role not in _BRANCHES_REQUIRED_ROLES
or workspace_path is None,
}
# ── foreign lease ────────────────────────────────────────────────────────
lease_needed = lease_required or (
task_name.removeprefix("gitea_") in {
t.removeprefix("gitea_") for t in _LEASE_AWARE_TASKS
}
and foreign_lease is not None
)
if lease_needed or foreign_lease is True:
is_foreign = bool(foreign_lease)
if foreign_lease is None and lease_required:
# Ownership must be proven when lease_required; missing ownership
# facts fail closed.
owner_ok = (
(not lease_owner_session and not active_session_id)
or (
lease_owner_session
and active_session_id
and lease_owner_session == active_session_id
)
)
identity_ok = (
(not lease_owner_identity and not active_identity)
or (
lease_owner_identity
and active_identity
and lease_owner_identity == active_identity
)
)
if not owner_ok or not identity_ok:
is_foreign = True
reasons = list(lease_reasons or [])
if is_foreign and not reasons:
reasons = [
"active lease is owned by a foreign session or identity "
"(anti-stomp fail closed)"
]
checks["lease"] = {
"block": is_foreign,
"reasons": reasons if is_foreign else [],
"lease_owner_session": lease_owner_session,
"active_session_id": active_session_id,
}
if is_foreign:
blockers.append(
_blocker(BLOCKER_FOREIGN_LEASE, reasons)
)
else:
checks["lease"] = {"block": False, "skipped": True}
# ── terminal lock ────────────────────────────────────────────────────────
if terminal_lock_blocks is not None:
reasons = list(terminal_lock_reasons or [])
if terminal_lock_blocks and not reasons:
reasons = [
"terminal review-decision lock is active for this head "
"(anti-stomp fail closed)"
]
checks["terminal_lock"] = {
"block": bool(terminal_lock_blocks),
"reasons": reasons if terminal_lock_blocks else [],
}
if terminal_lock_blocks:
blockers.append(_blocker(BLOCKER_TERMINAL_LOCK, reasons))
else:
checks["terminal_lock"] = {"block": False, "skipped": True}
# ── expected head SHA ────────────────────────────────────────────────────
if require_head_sha or (
expected_head_sha is not None and live_head_sha is not None
):
exp = (expected_head_sha or "").strip().lower()
live = (live_head_sha or "").strip().lower()
mismatch = False
reasons: list[str] = []
if require_head_sha and not exp:
mismatch = True
reasons.append(
"expected_head_sha is required before this mutation "
"(anti-stomp fail closed)"
)
elif exp and live and exp != live:
mismatch = True
reasons.append(
f"expected_head_sha '{exp}' does not match live PR head "
f"'{live}' (anti-stomp fail closed; stale prompt data)"
)
elif require_head_sha and exp and not live:
mismatch = True
reasons.append(
"live PR head SHA could not be determined to corroborate "
"expected_head_sha (anti-stomp fail closed)"
)
checks["head_sha"] = {
"block": mismatch,
"reasons": reasons,
"expected_head_sha": expected_head_sha,
"live_head_sha": live_head_sha,
}
if mismatch:
blockers.append(_blocker(BLOCKER_HEAD_SHA, reasons))
else:
checks["head_sha"] = {"block": False, "skipped": True}
# ── workflow hash ────────────────────────────────────────────────────────
if workflow_hash_valid is not None:
reasons = list(workflow_hash_reasons or [])
if not workflow_hash_valid and not reasons:
reasons = [
"workflow load proof missing or hash stale "
"(anti-stomp fail closed)"
]
checks["workflow_hash"] = {
"block": not bool(workflow_hash_valid),
"reasons": reasons if not workflow_hash_valid else [],
}
if not workflow_hash_valid:
blockers.append(_blocker(BLOCKER_WORKFLOW_HASH, reasons))
else:
checks["workflow_hash"] = {"block": False, "skipped": True}
# ── source contamination ─────────────────────────────────────────────────
if source_contaminated is not None:
reasons = list(contamination_reasons or [])
if source_contaminated and not reasons:
reasons = [
"session is source-contaminated (stable-branch push or "
"contaminated approval path); anti-stomp fail closed"
]
checks["source_contamination"] = {
"block": bool(source_contaminated),
"reasons": reasons if source_contaminated else [],
}
if source_contaminated:
blockers.append(
_blocker(BLOCKER_SOURCE_CONTAMINATION, reasons)
)
else:
checks["source_contamination"] = {"block": False, "skipped": True}
if blockers:
return _blocked_result(blockers, checks)
return _allowed_result(checks)
def format_anti_stomp_error(assessment: dict[str, Any]) -> str:
"""Single RuntimeError message for MCP mutation gates."""
kind = assessment.get("blocker_kind") or "unknown"
reasons = "; ".join(
assessment.get("reasons")
or ["anti-stomp preflight blocked the mutation"]
)
next_action = assessment.get("exact_next_action") or "stop and diagnose"
return (
f"Anti-stomp preflight (#604) blocked mutation "
f"[{kind}]: {reasons}. "
f"exact_next_action: {next_action}"
)
def block_response(
assessment: dict[str, Any],
**extra_fields: Any,
) -> dict[str, Any]:
"""Structured tool-return payload for a blocked mutation."""
payload = {
"success": False,
"performed": False,
"blocked": True,
"anti_stomp": True,
"blocker_kind": assessment.get("blocker_kind"),
"blockers": list(assessment.get("blockers") or []),
"reasons": list(assessment.get("reasons") or []),
"exact_next_action": assessment.get("exact_next_action"),
"checks": assessment.get("checks") or {},
}
payload.update(extra_fields)
return payload
def contamination_from_stable_marker(
marker: dict | None,
*,
task: str | None,
actual_role: str | None,
) -> tuple[bool, list[str]]:
"""Derive source-contamination facts from a #671 durable marker."""
if not marker:
return False, []
gate = stable_branch_push_guard.assess_contamination_gate(
marker,
task=task,
actual_role=actual_role,
)
if gate.get("block"):
return True, list(gate.get("reasons") or [])
return False, []
+372 -768
View File
File diff suppressed because it is too large Load Diff
-1
View File
@@ -13,7 +13,6 @@ from typing import Any
AUTHORIZED_CLEANUP_TOOLS = frozenset({
"gitea_cleanup_post_merge_moot_lease",
"gitea_cleanup_obsolete_reviewer_comment_lease",
"gitea_reconcile_merged_cleanups",
"gitea_delete_branch",
"gitea_cleanup_merged_pr_branch",
+10 -96
View File
@@ -36,25 +36,6 @@ KIND_REVIEW_DRAFT = "review_draft"
# other session proofs; a contaminated session fails closed on gated mutations
# until a reconciler audits and clears it.
KIND_STABLE_BRANCH_CONTAMINATION = "stable_branch_contamination"
# Durable shadow of the in-memory reviewer session lease (#702). Written on
# every sanctioned record/heartbeat and removed on sanctioned clear, so a
# daemon that dies without teardown leaves provable orphan evidence (owner
# pid + session id) for the guarded lease-cleanup path. Never a substitute
# for the in-session lease: mutation gates ignore it.
KIND_REVIEWER_SESSION_LEASE = "reviewer_session_lease"
# Durable audit record written whenever the daemon's sanctioned stale-binding
# recovery clears an inherited GITEA_ACTIVE_WORKTREE binding (#702).
KIND_STALE_BINDING_RECOVERY = "stale_binding_recovery"
# Kinds that carry recovery-critical crash-orphan evidence (#702). They are
# exempt from TTL expiry: a crashed daemon can never refresh its own record,
# so a wall-clock TTL would silently destroy the only owner-exit proof the
# guarded cleanup path accepts. These records live until a sanctioned clear
# terminally reconciles them (durable lifecycle), never until a timer fires.
RECOVERY_CRITICAL_KINDS = frozenset({
KIND_REVIEWER_SESSION_LEASE,
KIND_STALE_BINDING_RECOVERY,
})
@@ -105,7 +86,6 @@ def state_key(
org: str | None = None,
repo: str | None = None,
profile_identity: str | None = None,
instance_id: str | None = None,
) -> str:
"""Build durable filename key.
@@ -113,20 +93,17 @@ def state_key(
so the primary key is kind + profile identity. Remote/org/repo are stored
inside the payload and validated on load (#559), which lets a later daemon
process recover state without already knowing the remote argument.
*instance_id* extends the key for kinds where one-per-profile collides —
concurrent same-profile sessions each own their reviewer-lease shadow
(#702 F5), keyed by their lease session id so a later heartbeat can never
overwrite or misattribute another session's crash evidence.
"""
# Keep remote/org/repo parameters for API stability / future kinds; they are
# intentionally not part of the filename for session-scoped proofs.
_ = (remote, org, repo)
parts = [kind, profile_identity or "unknown-profile"]
instance = (instance_id or "").strip()
if instance:
parts.append(instance)
return "-".join(_sanitize_segment(part) for part in parts)
return "-".join(
_sanitize_segment(part)
for part in (
kind,
profile_identity or "unknown-profile",
)
)
def state_file_path(
@@ -137,7 +114,6 @@ def state_file_path(
repo: str | None = None,
profile_identity: str | None = None,
state_dir: str | None = None,
instance_id: str | None = None,
) -> str:
root = (state_dir or default_state_dir()).strip()
name = state_key(
@@ -146,7 +122,6 @@ def state_file_path(
org=org,
repo=repo,
profile_identity=profile_identity,
instance_id=instance_id,
)
return os.path.join(root, f"{name}.json")
@@ -267,12 +242,10 @@ def identity_match_reasons(
reasons.append("session state missing recorded_at timestamp (fail closed)")
else:
age = _now_utc() - recorded_at
record_kind = (str(record.get("kind") or "")).strip()
if age > timedelta(hours=ttl_hours()):
if record_kind not in RECOVERY_CRITICAL_KINDS:
reasons.append(
f"session state expired after {ttl_hours():g}h (fail closed)"
)
reasons.append(
f"session state expired after {ttl_hours():g}h (fail closed)"
)
if age < timedelta(0):
reasons.append("session state recorded_at is in the future (fail closed)")
return reasons
@@ -286,7 +259,6 @@ def load_state(
repo: str | None = None,
profile_identity: str | None = None,
state_dir: str | None = None,
instance_id: str | None = None,
) -> dict[str, Any] | None:
"""Load durable state payload when identity checks pass."""
profile = current_profile_identity(profile_identity=profile_identity)
@@ -297,7 +269,6 @@ def load_state(
repo=repo,
profile_identity=profile,
state_dir=state_dir,
instance_id=instance_id,
)
lock_path = f"{path}.lock"
with _exclusive_file_lock(lock_path):
@@ -343,7 +314,6 @@ def save_state(
repo: str | None = None,
profile_identity: str | None = None,
state_dir: str | None = None,
instance_id: str | None = None,
) -> dict[str, Any] | None:
"""Persist or clear durable state for the given session identity key."""
profile = current_profile_identity(
@@ -365,7 +335,6 @@ def save_state(
repo=key_repo,
profile_identity=profile,
state_dir=root,
instance_id=instance_id,
)
lock_path = f"{path}.lock"
with _exclusive_file_lock(lock_path):
@@ -387,8 +356,6 @@ def save_state(
body["session_profile_lock"] = profile
body["recorded_at"] = body.get("recorded_at") or now
body["updated_at"] = now
if (instance_id or "").strip():
body["instance_id"] = instance_id.strip()
if key_remote is not None:
body["remote"] = key_remote
if key_org is not None:
@@ -420,7 +387,6 @@ def clear_state(
repo: str | None = None,
profile_identity: str | None = None,
state_dir: str | None = None,
instance_id: str | None = None,
) -> None:
save_state(
kind=kind,
@@ -430,56 +396,4 @@ def clear_state(
repo=repo,
profile_identity=profile_identity,
state_dir=state_dir,
instance_id=instance_id,
)
def list_states(
*,
kind: str,
profile_identity: str | None = None,
state_dir: str | None = None,
) -> list[dict[str, Any]]:
"""Enumerate durable records of *kind* across all instance keys.
Crash recovery cannot know which session id left a shadow behind, so it
must be able to enumerate every record of a kind (#702 F5). Identity
checks (profile / TTL policy) apply per record exactly as in
:func:`load_state`; records that fail closed are omitted.
"""
root = (state_dir or default_state_dir()).strip()
if not root or not os.path.isdir(root):
return []
profile = current_profile_identity(profile_identity=profile_identity)
results: list[dict[str, Any]] = []
try:
names = sorted(os.listdir(root))
except OSError:
return []
for name in names:
if not name.endswith(".json") or name.startswith("."):
continue
envelope = _read_json(os.path.join(root, name))
if not envelope or envelope.get("kind") != kind:
continue
payload = envelope.get("payload")
if not isinstance(payload, dict):
continue
merged = dict(payload)
for key in (
"kind",
"remote",
"org",
"repo",
"profile_identity",
"session_profile_lock",
"recorded_at",
"updated_at",
"writer_pid",
):
if key in envelope and key not in merged:
merged[key] = envelope[key]
if identity_match_reasons(merged, profile_identity=profile):
continue
results.append(merged)
return results
+10 -36
View File
@@ -56,46 +56,23 @@ def resolve_namespace_workspace(
env: dict[str, str] | os._Environ | None = None,
session_lease_worktree: str | None = None,
profile_name: str | None = None,
demotions: list[str] | None = None,
verify_paths: bool = False,
) -> tuple[str, str]:
"""Return ``(resolved_path, binding_source)`` for *role_kind*.
With *verify_paths*, env-sourced candidates whose path no longer exists
are demoted (#702): a binding to a deleted worktree can never name a
valid task workspace, so resolution falls through to the next candidate.
Explicit arguments are never demoted — a caller-declared path must fail
loudly downstream rather than silently rebind. Demotion notes are
appended to *demotions* when provided. Runtime-context and mutation
guards resolve through :func:`resolve_namespace_mutation_context`, which
always verifies.
"""
"""Return ``(resolved_path, binding_source)`` for *role_kind*."""
env_map = env if env is not None else os.environ
role = normalize_role_kind(role_kind, profile_name=profile_name)
role_env_key = ROLE_WORKTREE_ENVS[role]
for candidate, source, env_sourced in (
(worktree_path, "worktree_path argument", False),
(worktree, "worktree argument", False),
(_env_value(env_map, ACTIVE_WORKTREE_ENV),
f"{ACTIVE_WORKTREE_ENV} environment variable", True),
(_env_value(env_map, role_env_key),
f"{role_env_key} environment variable", True),
for candidate, source in (
(worktree_path, "worktree_path argument"),
(worktree, "worktree argument"),
(_env_value(env_map, ACTIVE_WORKTREE_ENV), f"{ACTIVE_WORKTREE_ENV} environment variable"),
(_env_value(env_map, role_env_key), f"{role_env_key} environment variable"),
(session_lease_worktree if role in {"reviewer", "merger"} else None,
"reviewer PR lease worktree", False),
"reviewer PR lease worktree"),
):
text = (candidate or "").strip()
if not text:
continue
real = os.path.realpath(os.path.abspath(text))
if verify_paths and env_sourced and not os.path.isdir(real):
if demotions is not None:
demotions.append(
f"{source} '{real}' demoted: path no longer exists "
"(stale binding, #702)"
)
continue
return real, source
if text:
return os.path.realpath(os.path.abspath(text)), source
return os.path.realpath(process_project_root), "MCP server process root (default)"
@@ -111,7 +88,6 @@ def resolve_namespace_mutation_context(
profile_name: str | None = None,
) -> dict:
"""Shared workspace resolution for runtime_context and mutation guards."""
demotions: list[str] = []
workspace, binding_source = resolve_namespace_workspace(
role_kind=role_kind,
worktree_path=worktree_path,
@@ -120,8 +96,6 @@ def resolve_namespace_mutation_context(
env=env,
session_lease_worktree=session_lease_worktree,
profile_name=profile_name,
demotions=demotions,
verify_paths=True,
)
process_root = os.path.realpath(process_project_root)
role = normalize_role_kind(role_kind, profile_name=profile_name)
@@ -137,7 +111,7 @@ def resolve_namespace_mutation_context(
"workspace_path": workspace,
"workspace_binding_source": binding_source,
"workspace_role_kind": role,
"ignored_bindings": demotions + (pollution.get("ignored_bindings") or []),
"ignored_bindings": pollution.get("ignored_bindings") or [],
"process_project_root": process_root,
"canonical_repo_root": canonical_root,
"roots_aligned": canonical_root == process_root,
+43 -892
View File
File diff suppressed because it is too large Load Diff
-257
View File
@@ -1,257 +0,0 @@
"""Sanctioned recovery for stale env-bound task workspace bindings (#702).
When the MCP daemon dies from an unhandled exception it never runs teardown,
and the auto-reconnected daemon inherits ``GITEA_ACTIVE_WORKTREE`` from its
parent environment. That inherited value can permanently bind the fresh
session to a prior task's worktree (the PR #701 / ``review-pr-654`` incident).
This module classifies the active env binding against live session evidence
and produces a fail-closed recovery plan. Only two situations permit the
daemon to clear the binding on its own:
- the bound path no longer exists on disk (``provably_stale_missing_path``)
- the binding was inherited at daemon boot and a *sanctioned* in-session
reviewer/merger lease later bound a different worktree
(``superseded_by_session_lease``)
Everything else is surfaced (``unverified_inherited``) and left for the
managed reconnect path — never cleared silently, never rebound to a guessed
worktree.
"""
from __future__ import annotations
import os
from typing import Any
ACTIVE_WORKTREE_ENV = "GITEA_ACTIVE_WORKTREE"
CLASSIFICATION_UNBOUND = "unbound"
CLASSIFICATION_CORROBORATED = "corroborated"
CLASSIFICATION_UNVERIFIED_INHERITED = "unverified_inherited"
CLASSIFICATION_PROVABLY_STALE_MISSING_PATH = "provably_stale_missing_path"
CLASSIFICATION_SUPERSEDED_BY_SESSION_LEASE = "superseded_by_session_lease"
CLASSIFICATIONS = frozenset({
CLASSIFICATION_UNBOUND,
CLASSIFICATION_CORROBORATED,
CLASSIFICATION_UNVERIFIED_INHERITED,
CLASSIFICATION_PROVABLY_STALE_MISSING_PATH,
CLASSIFICATION_SUPERSEDED_BY_SESSION_LEASE,
})
# Roles whose task workspace is owned by a lease lifecycle, so an inherited
# generic binding without a corroborating lease is suspect.
LEASE_BOUND_ROLES = frozenset({"reviewer", "merger"})
RECOVERY_ACTION_NONE = "none"
RECOVERY_ACTION_CLEAR_ENV = "clear_env"
def _norm(path: str | None) -> str:
text = (path or "").strip()
if not text:
return ""
return os.path.realpath(os.path.abspath(text))
def snapshot_boot_bindings(
env: dict[str, str] | os._Environ | None = None,
) -> dict[str, Any]:
"""Capture worktree env bindings present when the daemon booted.
A value present in this snapshot was inherited from the parent
environment, not established by any sanctioned tool in this daemon's
lifetime.
"""
env_map = env if env is not None else os.environ
return {
"active_worktree": (env_map.get(ACTIVE_WORKTREE_ENV) or "").strip() or None,
}
def classify_active_worktree_binding(
*,
active_value: str | None,
role_env_value: str | None = None,
session_lease_worktree: str | None = None,
boot_inherited: bool,
path_exists: bool | None,
role_kind: str | None = None,
) -> dict[str, Any]:
"""Classify the GITEA_ACTIVE_WORKTREE binding against live evidence.
Fail closed: unknown evidence never yields a clear-eligible class.
"""
reasons: list[str] = []
active = _norm(active_value)
if not active:
return {
"classification": CLASSIFICATION_UNBOUND,
"clear_eligible": False,
"active_worktree": None,
"reasons": ["no GITEA_ACTIVE_WORKTREE binding present"],
}
role = (role_kind or "").strip().lower()
role_env = _norm(role_env_value)
lease_wt = _norm(session_lease_worktree)
result: dict[str, Any] = {
"active_worktree": active,
"boot_inherited": bool(boot_inherited),
"role_kind": role or None,
"session_lease_worktree": lease_wt or None,
"role_env_worktree": role_env or None,
}
if path_exists is False:
reasons.append(
f"bound worktree '{active}' no longer exists on disk; the binding "
"cannot name a valid task workspace"
)
result.update({
"classification": CLASSIFICATION_PROVABLY_STALE_MISSING_PATH,
"clear_eligible": True,
"reasons": reasons,
})
return result
if lease_wt and lease_wt == active:
reasons.append("binding matches the sanctioned in-session lease worktree")
result.update({
"classification": CLASSIFICATION_CORROBORATED,
"clear_eligible": False,
"reasons": reasons,
})
return result
if lease_wt and lease_wt != active and boot_inherited:
reasons.append(
"boot-inherited binding disagrees with the sanctioned in-session "
f"lease worktree '{lease_wt}'; the lease is live authority"
)
result.update({
"classification": CLASSIFICATION_SUPERSEDED_BY_SESSION_LEASE,
"clear_eligible": True,
"reasons": reasons,
})
return result
if lease_wt and lease_wt != active and not boot_inherited:
# Set after boot by tooling; disagreement is surfaced, never auto-cleared.
reasons.append(
"post-boot binding disagrees with the in-session lease worktree; "
"surfacing only (fail closed, no automatic clear)"
)
result.update({
"classification": CLASSIFICATION_UNVERIFIED_INHERITED,
"clear_eligible": False,
"reasons": reasons,
})
return result
if role_env and role_env == active:
reasons.append("binding matches the role-specific worktree env binding")
result.update({
"classification": CLASSIFICATION_CORROBORATED,
"clear_eligible": False,
"reasons": reasons,
})
return result
if boot_inherited and role in LEASE_BOUND_ROLES and not lease_wt:
reasons.append(
"binding was inherited at daemon boot, no sanctioned in-session "
f"lease corroborates it, and the {role} role binds workspaces "
"through the lease lifecycle; treat as unverified until a fresh "
"lease or managed reconnect re-establishes the workspace"
)
result.update({
"classification": CLASSIFICATION_UNVERIFIED_INHERITED,
"clear_eligible": False,
"reasons": reasons,
})
return result
reasons.append("no contradicting evidence; binding treated as corroborated")
result.update({
"classification": CLASSIFICATION_CORROBORATED,
"clear_eligible": False,
"reasons": reasons,
})
return result
def plan_recovery(classification_result: dict[str, Any]) -> dict[str, Any]:
"""Turn a classification into an explicit, fail-closed recovery plan."""
classification = classification_result.get("classification")
clear_eligible = bool(classification_result.get("clear_eligible"))
reasons = list(classification_result.get("reasons") or [])
if classification not in CLASSIFICATIONS:
return {
"action": RECOVERY_ACTION_NONE,
"clear_allowed": False,
"classification": classification,
"reasons": reasons + ["unknown classification (fail closed)"],
}
if clear_eligible and classification in {
CLASSIFICATION_PROVABLY_STALE_MISSING_PATH,
CLASSIFICATION_SUPERSEDED_BY_SESSION_LEASE,
}:
return {
"action": RECOVERY_ACTION_CLEAR_ENV,
"clear_allowed": True,
"classification": classification,
"active_worktree": classification_result.get("active_worktree"),
"reasons": reasons,
}
exact_next_action = None
if classification == CLASSIFICATION_UNVERIFIED_INHERITED:
exact_next_action = (
"managed recovery required: reconnect the MCP namespace through "
"the managed client configuration (or start a fresh session) so "
"the daemon boots without the inherited GITEA_ACTIVE_WORKTREE, or "
"corroborate the binding by acquiring a lease for that worktree; "
"do not clear or rewrite the environment manually"
)
return {
"action": RECOVERY_ACTION_NONE,
"clear_allowed": False,
"classification": classification,
"active_worktree": classification_result.get("active_worktree"),
"reasons": reasons,
**({"exact_next_action": exact_next_action} if exact_next_action else {}),
}
def apply_recovery(
plan: dict[str, Any],
env: dict[str, str] | os._Environ | None = None,
) -> dict[str, Any]:
"""Apply a clear-eligible plan by removing the stale env binding.
This is the sanctioned in-daemon mechanism (#702 AC2); callers must
persist the returned record for audit. Refuses anything but an explicit
``clear_env`` plan.
"""
env_map = env if env is not None else os.environ
if not plan.get("clear_allowed") or plan.get("action") != RECOVERY_ACTION_CLEAR_ENV:
return {
"performed": False,
"action": RECOVERY_ACTION_NONE,
"reasons": ["recovery plan does not authorize clearing (fail closed)"],
}
cleared_value = (env_map.get(ACTIVE_WORKTREE_ENV) or "").strip() or None
env_map.pop(ACTIVE_WORKTREE_ENV, None)
return {
"performed": True,
"action": RECOVERY_ACTION_CLEAR_ENV,
"cleared_env": ACTIVE_WORKTREE_ENV,
"cleared_value": cleared_value,
"classification": plan.get("classification"),
"reasons": list(plan.get("reasons") or []),
}
+21 -10
View File
@@ -60,6 +60,15 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
"permission": "gitea.pr.close",
"role": "author",
},
# Non-closing PR metadata edits (title/body/base). Closing uses close_pr.
"edit_pr": {
"permission": "gitea.pr.create",
"role": "author",
},
"gitea_edit_pr": {
"permission": "gitea.pr.create",
"role": "author",
},
"address_pr_change_requests": {
"permission": "gitea.branch.push",
"role": "author",
@@ -68,24 +77,26 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
"permission": "gitea.pr.review",
"role": "reviewer",
},
"submit_pr_review": {
"permission": "gitea.pr.review",
"role": "reviewer",
},
"merge_pr": {
"permission": "gitea.pr.merge",
"role": "merger",
},
"acquire_reviewer_pr_lease": {
"permission": "gitea.pr.comment",
"role": "reviewer",
},
"gitea_acquire_reviewer_pr_lease": {
"permission": "gitea.pr.comment",
"role": "reviewer",
},
"adopt_merger_pr_lease": {
"permission": "gitea.pr.comment",
"role": "reviewer",
},
# #691: guarded non-owner cleanup of obsolete comment-backed reviewer leases.
# Apply path posts lease release + audit comments (gitea.pr.comment).
"cleanup_obsolete_reviewer_comment_lease": {
"permission": "gitea.pr.comment",
"role": "reviewer",
},
"gitea_cleanup_obsolete_reviewer_comment_lease": {
"permission": "gitea.pr.comment",
"role": "reviewer",
},
"blind_pr_queue_review": {
"permission": "gitea.pr.review",
"role": "reviewer",
File diff suppressed because it is too large Load Diff
+10 -39
View File
@@ -57,23 +57,9 @@ class TestCreateIssueWorkspaceGuard(unittest.TestCase):
# Without worktree_path/env hints, workspace resolves to PROJECT_ROOT. When that
# path is the stable control checkout (not under branches/), mutation must fail.
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
try:
res = srv.gitea_create_issue(title="Test issue", body="body text")
except RuntimeError as exc:
self.assertIn("stable control checkout", str(exc))
else:
# #683: production guards return typed blockers at entrypoints
self.assertFalse(res.get("success"))
self.assertFalse(res.get("performed"))
blob = " ".join(res.get("reasons") or []) + " " + str(
res.get("blocker_kind") or ""
)
self.assertTrue(
"stable control checkout" in blob
or "missing_issue_worktree" in blob
or "control checkout" in blob.lower()
)
self.assertTrue(res.get("exact_next_action"))
with self.assertRaises(RuntimeError) as ctx:
srv.gitea_create_issue(title="Test issue", body="body text")
self.assertIn("stable control checkout", str(ctx.exception))
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
@patch("gitea_mcp_server._profile_permission_block", return_value=None)
@@ -119,17 +105,11 @@ class TestCreateIssueWorkspaceGuard(unittest.TestCase):
)
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
try:
res = srv.gitea_create_issue(
with self.assertRaises(RuntimeError) as ctx:
srv.gitea_create_issue(
title="Test issue", body="body", worktree_path=missing_path
)
except RuntimeError as exc:
self.assertIn("does not exist", str(exc))
else:
self.assertFalse(res.get("success"))
blob = " ".join(res.get("reasons") or [])
self.assertIn("does not exist", blob)
self.assertTrue(res.get("exact_next_action") or res.get("reasons"))
self.assertIn("does not exist (fail closed)", str(ctx.exception))
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
@patch("gitea_mcp_server._profile_permission_block", return_value=None)
@@ -162,20 +142,11 @@ class TestCreateIssueWorkspaceGuard(unittest.TestCase):
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
with patch("gitea_mcp_server._get_workspace_porcelain", return_value=""):
try:
res = srv.gitea_create_issue(
title="Test issue",
body="body",
worktree_path=wrong_repo_path,
with self.assertRaises(RuntimeError) as ctx:
srv.gitea_create_issue(
title="Test issue", body="body", worktree_path=wrong_repo_path
)
except RuntimeError as exc:
self.assertIn(
"does not belong to the target repository", str(exc)
)
else:
self.assertFalse(res.get("success"))
blob = " ".join(res.get("reasons") or [])
self.assertIn("does not belong to the target repository", blob)
self.assertIn("does not belong to the target repository", str(ctx.exception))
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
@patch("gitea_mcp_server._profile_permission_block", return_value=None)
+2 -11
View File
@@ -267,19 +267,10 @@ class TestReconcilerCommentThroughCanonicalPath(unittest.TestCase):
with patch.dict(os.environ, {}, clear=False):
os.environ.pop("GITEA_AUTHOR_WORKTREE", None)
os.environ.pop("GITEA_ACTIVE_WORKTREE", None)
try:
res = srv.gitea_create_issue_comment(
with self.assertRaises(RuntimeError):
srv.gitea_create_issue_comment(
515, "author note", remote="prgs"
)
except RuntimeError:
pass # legacy raise path
else:
# #683: typed blocker at mutation entrypoint
self.assertFalse(res.get("success"))
self.assertFalse(res.get("performed"))
self.assertTrue(
res.get("blocker_kind") or res.get("reasons")
)
mock_api.assert_not_called()
@@ -1,472 +0,0 @@
"""#683: block unattributed root WIP; pytest cannot disable production guards.
Regression coverage required by issue #683:
1. Session locked to issue A blocks unrelated target issue B until B is selected.
2. Diagnostic source edit on the root checkout is blocked.
3. Same legitimate edit succeeds after issue ownership + isolated worktree bind.
4. Running under pytest does not deactivate production guards when force-on.
5. Dirty tracked Python files remain visible to porcelain consumers.
6. Monkeypatching one helper cannot silently turn the full guard path into a no-op.
7. Real mutation entrypoint proves production guards run before side effects.
8. Same-issue edits in a valid isolated worktree remain unaffected.
9. Blocker includes stable reason + exact recovery action.
"""
from __future__ import annotations
import os
import sys
import tempfile
import textwrap
import unittest
from pathlib import Path
from unittest.mock import MagicMock, patch
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import gitea_mcp_server as mcp_server # noqa: E402
import issue_lock_worktree # noqa: E402
import workflow_scope_guard as wsg # noqa: E402
CONTROL_ROOT = str(Path(__file__).resolve().parent.parent)
if "branches" in Path(__file__).resolve().parts:
# Running from a worktree under branches/ — parent of branches is control.
parts = Path(__file__).resolve().parts
idx = parts.index("branches")
CONTROL_ROOT = str(Path(*parts[:idx])) if idx > 0 else CONTROL_ROOT
class TestProductionGuardsForceOn(unittest.TestCase):
def tearDown(self):
for key in (
wsg.FORCE_PRODUCTION_GUARDS_ENV,
"GITEA_TEST_FORCE_DIRTY",
"GITEA_TEST_PORCELAIN",
"GITEA_AUTHOR_WORKTREE",
"GITEA_ACTIVE_WORKTREE",
):
os.environ.pop(key, None)
wsg.clear_workflow_failure_ledger()
def test_force_on_under_pytest_keeps_production_active(self):
self.assertTrue(wsg.production_guards_active(in_test_mode=False))
self.assertFalse(wsg.production_guards_active(in_test_mode=True))
os.environ[wsg.FORCE_PRODUCTION_GUARDS_ENV] = "1"
self.assertTrue(wsg.production_guards_active(in_test_mode=True))
self.assertTrue(wsg.production_guards_forced())
def test_no_early_return_in_verify_role_mutation_workspace_source(self):
src = Path(mcp_server.__file__).read_text(encoding="utf-8")
# Rejected 300a4ca pattern must not exist.
self.assertNotIn(
"if _preflight_in_test_mode():\n return _resolve_preflight_workspace_path",
src,
)
# Docstring contract for #683.
self.assertIn("#683", src)
self.assertIn("must NOT early-return solely because pytest", src)
class TestPorcelainIntegrity(unittest.TestCase):
def test_read_worktree_git_state_surfaces_dirty_py(self):
with tempfile.TemporaryDirectory() as tmp:
# Use a real git repo so porcelain is truthful.
import subprocess
subprocess.run(["git", "init"], cwd=tmp, check=True, capture_output=True)
subprocess.run(
["git", "config", "user.email", "[email protected]"],
cwd=tmp,
check=True,
capture_output=True,
)
subprocess.run(
["git", "config", "user.name", "t"],
cwd=tmp,
check=True,
capture_output=True,
)
py_path = Path(tmp) / "sample_mod.py"
py_path.write_text("x = 1\n", encoding="utf-8")
subprocess.run(["git", "add", "sample_mod.py"], cwd=tmp, check=True)
subprocess.run(
["git", "commit", "-m", "init"],
cwd=tmp,
check=True,
capture_output=True,
)
py_path.write_text("x = 2\n", encoding="utf-8")
state = issue_lock_worktree.read_worktree_git_state(tmp)
porcelain = state.get("porcelain_status") or ""
self.assertIn("sample_mod.py", porcelain)
self.assertTrue(any(line.strip().endswith(".py") for line in porcelain.splitlines()))
def test_production_reader_source_rejects_pytest_py_filter(self):
src = Path(issue_lock_worktree.__file__).read_text(encoding="utf-8")
findings = wsg.assert_no_pytest_porcelain_filter(src)
self.assertEqual(findings, [])
# Negative: the rejected 300a4ca pattern is detected.
rejected = textwrap.dedent(
"""
porcelain = status_res.stdout or ""
import sys
if "pytest" in sys.modules or "unittest" in sys.modules:
porcelain = "\\n".join(
line for line in porcelain.splitlines()
if not line.strip().endswith(".py")
)
"""
)
self.assertTrue(wsg.assert_no_pytest_porcelain_filter(rejected))
class TestIssueScopeOwnership(unittest.TestCase):
def test_out_of_scope_issue_blocked_until_selected(self):
result = wsg.assess_issue_scope_ownership(
locked_issue_number=100,
target_issue_number=200,
branch_name="fix/issue-100-example",
role_kind="author",
)
self.assertTrue(result["block"])
self.assertEqual(result["blocker_kind"], wsg.BLOCKER_OUT_OF_SCOPE_ISSUE)
self.assertIn("exact_next_action", result)
self.assertIn("owning issue", result["exact_next_action"].lower())
self.assertTrue(result["reasons"])
def test_same_issue_scope_allowed(self):
result = wsg.assess_issue_scope_ownership(
locked_issue_number=100,
target_issue_number=100,
branch_name="fix/issue-100-example",
role_kind="author",
)
self.assertFalse(result["block"])
self.assertEqual(result["exact_next_action"], "proceed")
def test_missing_lock_when_required(self):
result = wsg.assess_issue_scope_ownership(
locked_issue_number=None,
target_issue_number=None,
role_kind="author",
require_lock_for_author=True,
)
self.assertTrue(result["block"])
self.assertEqual(result["blocker_kind"], wsg.BLOCKER_MISSING_ISSUE_SCOPE)
def test_branch_issue_mismatch(self):
result = wsg.assess_issue_scope_ownership(
locked_issue_number=50,
branch_name="fix/issue-99-other",
role_kind="author",
)
self.assertTrue(result["block"])
self.assertEqual(result["blocker_kind"], wsg.BLOCKER_OUT_OF_SCOPE_ISSUE)
class TestRootDiagnosticEdit(unittest.TestCase):
def test_dirty_root_source_blocked(self):
result = wsg.assess_root_source_mutation(
workspace_path=CONTROL_ROOT,
canonical_repo_root=CONTROL_ROOT,
porcelain_status=" M gitea_mcp_server.py\n M tests/test_x.py\n",
role_kind="author",
)
self.assertTrue(result["block"])
self.assertEqual(result["blocker_kind"], wsg.BLOCKER_ROOT_DIAGNOSTIC_EDIT)
self.assertIn("gitea_mcp_server.py", result["dirty_source_files"])
self.assertIn("exact_next_action", result)
self.assertIn("branches/", result["exact_next_action"])
def test_isolated_worktree_same_issue_unaffected(self):
wt = f"{CONTROL_ROOT}/branches/issue-100-example"
result = wsg.assess_root_source_mutation(
workspace_path=wt,
canonical_repo_root=CONTROL_ROOT,
porcelain_status=" M helper.py\n",
current_branch="fix/issue-100-example",
locked_issue_number=100,
role_kind="author",
)
self.assertFalse(result["block"])
self.assertTrue(result["under_branches"])
def test_legitimate_after_ownership_and_worktree(self):
wt = f"{CONTROL_ROOT}/branches/issue-683-workflow-guard-hardening"
composed = wsg.assess_production_mutation_guards(
workspace_path=wt,
canonical_repo_root=CONTROL_ROOT,
porcelain_status=" M workflow_scope_guard.py\n",
current_branch="fix/issue-683-workflow-guard-hardening",
locked_issue_number=683,
target_issue_number=683,
role_kind="author",
require_author_lock=True,
in_test_mode=True,
)
# Force-on required for production path under pytest.
os.environ[wsg.FORCE_PRODUCTION_GUARDS_ENV] = "1"
try:
composed = wsg.assess_production_mutation_guards(
workspace_path=wt,
canonical_repo_root=CONTROL_ROOT,
porcelain_status=" M workflow_scope_guard.py\n",
current_branch="fix/issue-683-workflow-guard-hardening",
locked_issue_number=683,
target_issue_number=683,
role_kind="author",
require_author_lock=True,
in_test_mode=True,
)
self.assertFalse(composed["block"])
self.assertFalse(composed.get("skipped"))
finally:
os.environ.pop(wsg.FORCE_PRODUCTION_GUARDS_ENV, None)
class TestTypedBlockerResponse(unittest.TestCase):
def test_block_response_has_stable_kind_and_next_action(self):
assessment = wsg.assess_issue_scope_ownership(
locked_issue_number=1,
target_issue_number=2,
role_kind="author",
)
resp = wsg.block_response(assessment)
self.assertFalse(resp["success"])
self.assertFalse(resp["performed"])
self.assertEqual(resp["blocker_kind"], wsg.BLOCKER_OUT_OF_SCOPE_ISSUE)
self.assertIsInstance(resp["exact_next_action"], str)
self.assertTrue(resp["exact_next_action"])
self.assertTrue(resp["reasons"])
def test_production_guard_error_roundtrip(self):
err = wsg.ProductionGuardError(
"blocked",
blocker_kind=wsg.BLOCKER_ROOT_DIAGNOSTIC_EDIT,
reasons=["dirty root"],
)
resp = wsg.block_response(err, issue_number=683)
self.assertEqual(resp["blocker_kind"], wsg.BLOCKER_ROOT_DIAGNOSTIC_EDIT)
self.assertEqual(resp["issue_number"], 683)
self.assertIn("exact_next_action", resp)
class TestDurableFailureRecording(unittest.TestCase):
def setUp(self):
wsg.clear_workflow_failure_ledger()
def tearDown(self):
wsg.clear_workflow_failure_ledger()
def test_record_before_source_mutation(self):
pending = wsg.assess_durable_failure_recorded(
require_record=True, pending_source_mutation=True
)
self.assertTrue(pending["block"])
self.assertEqual(pending["blocker_kind"], wsg.BLOCKER_UNRECORDED_FAILURE)
wsg.record_workflow_failure(
kind="transport_eof",
detail="EOF during review session (#584 cluster)",
issue_number=683,
task="comment_issue",
)
after = wsg.assess_durable_failure_recorded(
require_record=True, pending_source_mutation=True
)
self.assertFalse(after["block"])
self.assertEqual(len(wsg.workflow_failure_ledger()), 1)
class TestMonkeypatchCannotNoopFullPath(unittest.TestCase):
def tearDown(self):
os.environ.pop(wsg.FORCE_PRODUCTION_GUARDS_ENV, None)
def test_patching_branches_only_still_blocks_dirty_root_scope(self):
"""Monkeypatching branches-only must not silence root diagnostic block."""
os.environ[wsg.FORCE_PRODUCTION_GUARDS_ENV] = "1"
with patch.object(
mcp_server, "_enforce_branches_only_author_mutation", lambda *a, **k: None
):
with patch.object(
mcp_server, "_enforce_root_checkout_guard", lambda *a, **k: None
):
# Even if both legacy helpers are patched, issue-scope composition
# still sees dirty root source via assess_production_mutation_guards.
assessment = wsg.assess_production_mutation_guards(
workspace_path=CONTROL_ROOT,
canonical_repo_root=CONTROL_ROOT,
porcelain_status=" M gitea_mcp_server.py\n",
role_kind="author",
in_test_mode=True,
)
self.assertTrue(assessment["block"])
self.assertEqual(
assessment["blocker_kind"], wsg.BLOCKER_ROOT_DIAGNOSTIC_EDIT
)
class TestRealEntrypointProductionGuard(unittest.TestCase):
"""Real mutation entrypoint: production guard before side effects (#683)."""
def setUp(self):
os.environ[wsg.FORCE_PRODUCTION_GUARDS_ENV] = "1"
for key in ("GITEA_AUTHOR_WORKTREE", "GITEA_ACTIVE_WORKTREE"):
os.environ.pop(key, None)
self._orig_whoami = mcp_server._preflight_whoami_called
self._orig_cap = mcp_server._preflight_capability_called
mcp_server._preflight_whoami_called = False
mcp_server._preflight_capability_called = False
mcp_server._preflight_resolved_role = None
mcp_server._preflight_resolved_task = None
def tearDown(self):
os.environ.pop(wsg.FORCE_PRODUCTION_GUARDS_ENV, None)
for key in ("GITEA_AUTHOR_WORKTREE", "GITEA_ACTIVE_WORKTREE"):
os.environ.pop(key, None)
mcp_server._preflight_whoami_called = self._orig_whoami
mcp_server._preflight_capability_called = self._orig_cap
mcp_server._preflight_resolved_role = None
mcp_server._preflight_resolved_task = None
def test_comment_issue_blocks_dirty_root_before_api(self):
api_mock = MagicMock()
with patch.object(mcp_server, "api_request", api_mock), patch.object(
mcp_server,
"_actual_profile_role",
return_value="author",
), patch.object(
mcp_server,
"_effective_workspace_role",
return_value="author",
), patch.object(
mcp_server,
"get_profile",
return_value={
"profile_name": "prgs-author",
"allowed_operations": [
"gitea.issue.comment",
"gitea.read",
"gitea.pr.create",
"gitea.branch.push",
],
"forbidden_operations": [],
},
), patch.object(
issue_lock_worktree,
"read_worktree_git_state",
side_effect=lambda path, **kw: {
"current_branch": "master",
"porcelain_status": (
" M gitea_mcp_server.py\n"
if os.path.realpath(path) == os.path.realpath(CONTROL_ROOT)
or path == CONTROL_ROOT
else ""
),
"head_sha": "a" * 40,
"base_equivalent": True,
},
), patch.object(
mcp_server,
"_resolve_namespace_mutation_context",
return_value={
"workspace_path": CONTROL_ROOT,
"canonical_repo_root": CONTROL_ROOT,
"process_project_root": CONTROL_ROOT,
"workspace_role_kind": "author",
"workspace_binding_source": "process root",
"ignored_bindings": [],
},
), patch.object(
mcp_server,
"_resolve_author_mutation_context",
return_value={
"workspace_path": CONTROL_ROOT,
"canonical_repo_root": CONTROL_ROOT,
"process_project_root": CONTROL_ROOT,
"roots_aligned": True,
},
), patch.object(
mcp_server,
"_session_locked_issue_number",
return_value=None,
):
result = mcp_server.gitea_create_issue_comment(
issue_number=683,
body="diagnostic note",
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
worktree_path=CONTROL_ROOT,
)
api_mock.assert_not_called()
self.assertFalse(result.get("success"))
self.assertFalse(result.get("performed"))
self.assertIn(result.get("blocker_kind"), wsg.BLOCKER_KINDS)
self.assertTrue(result.get("exact_next_action"))
self.assertTrue(result.get("reasons"))
def test_comment_issue_succeeds_structure_after_worktree_bind(self):
"""Same-issue isolated worktree is not blocked by root diagnostic path."""
wt = f"{CONTROL_ROOT}/branches/issue-683-workflow-guard-hardening"
os.environ["GITEA_AUTHOR_WORKTREE"] = wt
assessment = wsg.assess_production_mutation_guards(
workspace_path=wt,
canonical_repo_root=CONTROL_ROOT,
porcelain_status=" M workflow_scope_guard.py\n",
current_branch="fix/issue-683-workflow-guard-hardening",
locked_issue_number=683,
target_issue_number=683,
role_kind="author",
require_author_lock=True,
in_test_mode=True,
)
self.assertFalse(assessment["block"], assessment)
class TestVerifyPreflightForceOn(unittest.TestCase):
def tearDown(self):
os.environ.pop(wsg.FORCE_PRODUCTION_GUARDS_ENV, None)
for key in ("GITEA_AUTHOR_WORKTREE", "GITEA_ACTIVE_WORKTREE"):
os.environ.pop(key, None)
def test_force_on_runs_production_guards_under_pytest(self):
os.environ[wsg.FORCE_PRODUCTION_GUARDS_ENV] = "1"
called = {"root": 0, "branches": 0, "scope": 0}
def _root(*a, **k):
called["root"] += 1
def _branches(*a, **k):
called["branches"] += 1
def _scope(*a, **k):
called["scope"] += 1
with patch.object(mcp_server, "_enforce_root_checkout_guard", _root), patch.object(
mcp_server, "_enforce_branches_only_author_mutation", _branches
), patch.object(mcp_server, "_enforce_issue_scope_guard", _scope):
# No whoami/capability — purity-order skipped; production still runs.
mcp_server.verify_preflight_purity(task="comment_issue")
self.assertEqual(called["root"], 1)
self.assertEqual(called["branches"], 1)
self.assertEqual(called["scope"], 1)
def test_without_force_on_pytest_skips_production_only_for_unit_isolation(self):
called = {"root": 0}
def _root(*a, **k):
called["root"] += 1
with patch.object(mcp_server, "_enforce_root_checkout_guard", _root), patch.object(
mcp_server, "_enforce_branches_only_author_mutation", lambda *a, **k: None
), patch.object(mcp_server, "_enforce_issue_scope_guard", lambda *a, **k: None):
mcp_server.verify_preflight_purity(task="comment_issue")
self.assertEqual(called["root"], 0)
if __name__ == "__main__":
unittest.main()
@@ -1,616 +0,0 @@
"""Guarded cleanup for obsolete comment-backed reviewer leases (#691).
Reproduces PR #688 lease comment 10749 class of defect: completed
REQUEST_CHANGES on head A, author pushes head B, foreign lease remains
pinned to A (and after expiry still has no non-owner cleanup path that
diagnosis can prescribe beyond indefinite wait).
"""
from __future__ import annotations
import sys
import unittest
from datetime import datetime, timedelta, timezone
from pathlib import Path
from unittest.mock import patch
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import merger_lease_adoption as mla # noqa: E402
import reviewer_pr_lease as leases # noqa: E402
# PR #688 / comment 10749 reproduction fixtures
HEAD_A = "c7a444eb4b41cf916fdbd20a4999ffd78af496d0"
HEAD_B = "4a6357800364718a27a36ebc73578d4b929ff4aa"
SESSION_FOREIGN = "25883-7d4c6e6ebd53"
COMMENT_10749 = 10749
REPO = "Scaled-Tech-Consulting/Gitea-Tools"
PR = 688
ISSUE = 687
EXPIRES_10749 = "2026-07-13T04:23:00Z"
def _utc(dt: datetime) -> datetime:
return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc)
def _lease_comment(
*,
pr_number: int = PR,
session_id: str = SESSION_FOREIGN,
phase: str = "claimed",
candidate_head: str = HEAD_A,
comment_id: int = COMMENT_10749,
last_activity: datetime | None = None,
expires_at: datetime | None = None,
worktree: str = "branches/review-pr688-feat-issue-687-reconciler-branch-delete",
repo: str = REPO,
) -> dict:
now = last_activity or datetime.now(timezone.utc)
body = leases.format_lease_body(
repo=repo,
pr_number=pr_number,
issue_number=ISSUE,
reviewer_identity="sysadmin",
profile="prgs-reviewer",
session_id=session_id,
worktree=worktree,
phase=phase,
candidate_head=candidate_head,
target_branch="master",
target_branch_sha="b" * 40,
last_activity=now,
expires_at=expires_at,
)
return {
"id": comment_id,
"body": body,
"user": {"login": "sysadmin"},
"author": "sysadmin",
"created_at": now.isoformat(),
"updated_at": now.isoformat(),
}
def _reviews_for_head(head: str, verdict: str = "REQUEST_CHANGES") -> list[dict]:
return [
{
"verdict": verdict,
"reviewed_head_sha": head,
"dismissed": False,
"reviewer": "sysadmin",
}
]
def _assess(
comments,
*,
current_head=HEAD_B,
formal_reviews=None,
controller=True,
worktree_exists=True,
worktree_clean=True,
owner_process_alive=False,
requesting_session="fresh-controller",
apply=False,
confirmation=None,
**kwargs,
):
return leases.assess_obsolete_reviewer_comment_lease_cleanup(
comments,
pr_number=PR,
current_head_sha=current_head,
formal_reviews=formal_reviews if formal_reviews is not None else _reviews_for_head(HEAD_A),
repo=REPO,
expected_repo=REPO,
requesting_session_id=requesting_session,
controller_recovery_authorized=controller,
worktree_exists=worktree_exists,
worktree_clean=worktree_clean,
worktree_has_unpreserved_work=not worktree_clean if worktree_exists else False,
owner_process_alive=owner_process_alive,
owner_pid_observed=kwargs.get("owner_pid_observed"),
requesting_pid=kwargs.get("requesting_pid", 99999),
current_head_review_in_progress=kwargs.get(
"current_head_review_in_progress", False
),
expected_lease_comment_id=kwargs.get("expected_lease_comment_id"),
expected_session_id=kwargs.get("expected_session_id"),
expected_leased_head=kwargs.get("expected_leased_head"),
confirmation=confirmation
if confirmation is not None
else (leases.cleanup_confirmation_for_pr(PR) if apply else ""),
apply=apply,
now=kwargs.get("now"),
)
class TestIssue691SupersededHeadCleanup(unittest.TestCase):
def setUp(self):
leases.clear_session_lease()
def test_1_request_changes_then_push_expired_cleanup_succeeds(self):
"""Completed REQUEST_CHANGES on A, push B, lease A expires → cleanup OK."""
now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
claim = _lease_comment(
last_activity=expires - timedelta(hours=2),
expires_at=expires,
)
result = _assess(
[claim],
formal_reviews=_reviews_for_head(HEAD_A, "REQUEST_CHANGES"),
now=now,
apply=True,
)
self.assertTrue(result["cleanup_allowed"], result["reasons"])
self.assertEqual(result["classification"], "foreign_expired_superseded_head")
self.assertEqual(
result["exact_next_action"], leases.NEXT_ACTION_CLEANUP_OBSOLETE_LEASE
)
self.assertIn("phase: released", result["release_body"])
self.assertIn("obsolete-superseded-or-expired-lease", result["release_body"])
self.assertIsNotNone(result["audit_comment_body"])
self.assertEqual(result["cleanup_tool"], leases.CLEANUP_OBSOLETE_LEASE_TOOL)
def test_2_approve_then_push_cleanup_policy(self):
"""Completed APPROVE on A, push B → cleanup eligible (completed superseded)."""
now = datetime(2026, 7, 13, 1, 0, 0, tzinfo=timezone.utc)
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc) # not yet expired
claim = _lease_comment(
last_activity=now - timedelta(minutes=10),
expires_at=expires,
)
result = _assess(
[claim],
formal_reviews=_reviews_for_head(HEAD_A, "APPROVED"),
now=now,
)
self.assertTrue(result["cleanup_allowed"], result["reasons"])
self.assertEqual(result["classification"], "foreign_completed_superseded_head")
def test_3_active_current_head_cleanup_denied(self):
now = datetime.now(timezone.utc)
claim = _lease_comment(
candidate_head=HEAD_B,
last_activity=now - timedelta(minutes=5),
expires_at=now + timedelta(hours=2),
)
result = _assess(
[claim],
current_head=HEAD_B,
formal_reviews=_reviews_for_head(HEAD_B, "REQUEST_CHANGES"),
now=now,
)
self.assertFalse(result["cleanup_allowed"])
self.assertEqual(result["classification"], "foreign_active_current_head")
def test_4_foreign_owner_recent_progress_denied(self):
now = datetime.now(timezone.utc)
claim = _lease_comment(
candidate_head=HEAD_B,
last_activity=now - timedelta(minutes=2),
expires_at=now + timedelta(hours=2),
)
result = _assess(
[claim],
current_head=HEAD_B,
formal_reviews=[],
owner_process_alive=True,
now=now,
)
self.assertFalse(result["cleanup_allowed"])
self.assertTrue(
any("recent authenticated progress" in r for r in result["reasons"])
or any("current PR head" in r for r in result["reasons"])
)
def test_5_owner_absent_worktree_dirty_denied(self):
now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
claim = _lease_comment(
candidate_head=HEAD_B,
last_activity=expires - timedelta(hours=1),
expires_at=expires,
)
result = _assess(
[claim],
current_head=HEAD_B,
formal_reviews=[],
worktree_clean=False,
owner_process_alive=False,
now=now,
)
self.assertFalse(result["cleanup_allowed"])
self.assertTrue(any("dirty" in r for r in result["reasons"]))
def test_6_owner_absent_worktree_clean_orphan_ok(self):
now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
claim = _lease_comment(
candidate_head=HEAD_B,
last_activity=expires - timedelta(hours=1),
expires_at=expires,
)
result = _assess(
[claim],
current_head=HEAD_B,
formal_reviews=[],
worktree_clean=True,
owner_process_alive=False,
now=now,
)
self.assertTrue(result["cleanup_allowed"], result["reasons"])
self.assertEqual(result["classification"], "orphaned_owner_missing")
def test_7_pid_reuse_not_ownership(self):
now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
claim = _lease_comment(expires_at=expires, last_activity=expires - timedelta(hours=1))
result = _assess(
[claim],
now=now,
owner_pid_observed=4242,
requesting_pid=4242,
)
self.assertTrue(
any("PID equality" in r or "NOT ownership" in r for r in result["reasons"])
or result["owner_session_evidence"]["pid_is_not_ownership_proof"]
)
# Still eligible via superseded+terminal+expired despite matching PID.
self.assertTrue(result["cleanup_allowed"], result["reasons"])
def test_8_comment_backed_no_db_lease_id(self):
"""Cleanup uses comment ledger only — no control-plane lease_id required."""
now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
claim = _lease_comment(expires_at=expires, last_activity=expires - timedelta(hours=1))
# Explicitly no lease_id field anywhere.
self.assertNotIn("lease_id", claim)
result = _assess([claim], now=now)
self.assertTrue(result["cleanup_allowed"], result["reasons"])
self.assertEqual(result["active_lease"]["comment_id"], COMMENT_10749)
self.assertIsNone(result["active_lease"].get("lease_id"))
def test_9_cleanup_posts_durable_audit_bodies(self):
now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
claim = _lease_comment(expires_at=expires, last_activity=expires - timedelta(hours=1))
result = _assess([claim], now=now, apply=True)
self.assertTrue(result["cleanup_allowed"])
self.assertIn("#691", result["audit_comment_body"])
self.assertIn(str(COMMENT_10749), result["audit_comment_body"])
self.assertIn(HEAD_A, result["audit_comment_body"])
self.assertIn(HEAD_B, result["audit_comment_body"])
def test_10_fresh_acquire_after_cleanup_marker(self):
now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
claim = _lease_comment(
expires_at=expires,
last_activity=expires - timedelta(hours=1),
comment_id=COMMENT_10749,
)
assessed = _assess([claim], now=now, apply=True)
self.assertTrue(assessed["cleanup_allowed"])
release = {
"id": 99999,
"body": assessed["release_body"],
"user": {"login": "sysadmin"},
}
# Newest terminal marker ends active lease.
active = leases.find_active_reviewer_lease(
[claim, release], pr_number=PR, now=now
)
self.assertIsNone(active)
acq = leases.assess_acquire_lease(
[claim, release],
pr_number=PR,
reviewer_identity="sysadmin",
profile="prgs-reviewer",
session_id="fresh-reviewer-1",
repo=REPO,
issue_number=ISSUE,
worktree="branches/review-pr-688-fresh",
candidate_head=HEAD_B,
target_branch="master",
target_branch_sha="b" * 40,
now=now,
)
self.assertTrue(acq["acquire_allowed"], acq["reasons"])
def test_11_no_validation_state_transfer(self):
now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
claim = _lease_comment(expires_at=expires, last_activity=expires - timedelta(hours=1))
result = _assess([claim], now=now, apply=True)
self.assertTrue(result["cleanup_allowed"])
# Release body keeps old candidate_head (no repoint).
self.assertIn(f"candidate_head: {HEAD_A}", result["release_body"])
self.assertNotIn(HEAD_B, result["release_body"].split("candidate_head:")[1].split("\n")[0])
self.assertIn("did not transfer validation", result["audit_comment_body"])
self.assertIn("did not repoint", result["audit_comment_body"])
# Session lease must remain unset by assessment (tool never seeds).
self.assertIsNone(leases.get_session_lease())
def test_12_repeated_cleanup_idempotent(self):
now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
claim = _lease_comment(expires_at=expires, last_activity=expires - timedelta(hours=1))
first = _assess([claim], now=now, apply=True)
self.assertTrue(first["cleanup_allowed"])
release = {"id": 100001, "body": first["release_body"], "user": {"login": "x"}}
second = _assess([claim, release], now=now, apply=True)
self.assertFalse(second["cleanup_allowed"])
self.assertEqual(second["classification"], "no_lease")
def test_13_malformed_expiry_fails_closed(self):
claim = _lease_comment()
# Corrupt expires_at in the body.
claim["body"] = claim["body"].replace(
claim["body"].split("expires_at:")[1].split("\n")[0].strip(),
"not-a-timestamp",
)
result = _assess([claim], formal_reviews=_reviews_for_head(HEAD_A))
self.assertFalse(result["cleanup_allowed"])
self.assertFalse(result["expires_parseable"])
self.assertTrue(any("expires_at" in r for r in result["reasons"]))
def test_14_wrong_identity_evidence_fails_closed(self):
now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
claim = _lease_comment(expires_at=expires, last_activity=expires - timedelta(hours=1))
bad_repo = _assess(
[claim],
now=now,
# force repo mismatch via expected_repo
)
# Patch via direct call with wrong expected_repo
bad = leases.assess_obsolete_reviewer_comment_lease_cleanup(
[claim],
pr_number=PR,
current_head_sha=HEAD_B,
formal_reviews=_reviews_for_head(HEAD_A),
expected_repo="Other-Org/Other-Repo",
requesting_session_id="fresh",
controller_recovery_authorized=True,
worktree_exists=True,
worktree_clean=True,
owner_process_alive=False,
now=now,
)
self.assertFalse(bad["cleanup_allowed"])
self.assertTrue(any("repository identity" in r for r in bad["reasons"]))
bad_pr = leases.assess_obsolete_reviewer_comment_lease_cleanup(
[claim],
pr_number=999,
current_head_sha=HEAD_B,
formal_reviews=_reviews_for_head(HEAD_A),
expected_repo=REPO,
requesting_session_id="fresh",
controller_recovery_authorized=True,
worktree_exists=True,
worktree_clean=True,
owner_process_alive=False,
expected_lease_comment_id=COMMENT_10749,
now=now,
)
self.assertFalse(bad_pr["cleanup_allowed"])
bad_head = _assess(
[claim],
now=now,
expected_leased_head="0" * 40,
)
self.assertFalse(bad_head["cleanup_allowed"])
bad_session = _assess(
[claim],
now=now,
expected_session_id="wrong-session",
)
self.assertFalse(bad_session["cleanup_allowed"])
def test_15_current_head_active_cannot_be_displaced(self):
now = datetime.now(timezone.utc)
claim = _lease_comment(
candidate_head=HEAD_B,
last_activity=now - timedelta(minutes=1),
expires_at=now + timedelta(hours=2),
)
result = _assess(
[claim],
current_head=HEAD_B,
formal_reviews=_reviews_for_head(HEAD_B),
now=now,
current_head_review_in_progress=True,
)
self.assertFalse(result["cleanup_allowed"])
# Acquire also blocked by foreign active lease.
acq = leases.assess_acquire_lease(
[claim],
pr_number=PR,
reviewer_identity="other",
profile="prgs-reviewer",
session_id="thief",
repo=REPO,
issue_number=ISSUE,
worktree="branches/steal",
candidate_head=HEAD_B,
target_branch="master",
target_branch_sha="b" * 40,
now=now,
)
self.assertFalse(acq["acquire_allowed"])
def test_regression_pr688_comment_10749_after_expiry(self):
"""Exact 10749 reproduction after recorded expires_at 2026-07-13T04:23:00Z."""
now = datetime(2026, 7, 13, 4, 30, 0, tzinfo=timezone.utc)
expires = datetime.fromisoformat(EXPIRES_10749.replace("Z", "+00:00"))
claim = _lease_comment(
session_id=SESSION_FOREIGN,
candidate_head=HEAD_A,
comment_id=COMMENT_10749,
last_activity=expires - timedelta(hours=2),
expires_at=expires,
)
# Diagnosis must not prescribe indefinite wait-only for this case.
diag = leases.diagnose_reviewer_pr_lease_handoff(
[claim],
pr_number=PR,
current_session_id="fresh-reviewer",
current_reviewer_identity="sysadmin",
current_head_sha=HEAD_B,
formal_reviews=_reviews_for_head(HEAD_A, "REQUEST_CHANGES"),
worktree_exists=True,
worktree_clean=True,
owner_process_alive=False,
instructed_session_id=SESSION_FOREIGN,
instructed_comment_id=COMMENT_10749,
now=now,
)
self.assertEqual(diag["classification"], "foreign_expired_superseded_head")
self.assertEqual(
diag["next_action"], leases.NEXT_ACTION_CLEANUP_OBSOLETE_LEASE
)
self.assertEqual(diag["cleanup_tool"], leases.CLEANUP_OBSOLETE_LEASE_TOOL)
self.assertIn("CLEANUP OBSOLETE REVIEWER LEASE 688", diag["required_confirmation"])
self.assertEqual(diag["mutation_eligibility"], "cleanup_only")
self.assertFalse(diag["mutation_allowed"])
# Assessment still returns the lease as active/stale class of block for acquire
# when unexpired path... here it's expired so find_active is None; acquire free
# only after cleanup if somehow still active. With expired, find_active is None:
active = leases.find_active_reviewer_lease([claim], pr_number=PR, now=now)
self.assertIsNone(active)
# But superseded unexpired still blocks acquire and diagnosis points to cleanup:
unexpired_now = datetime(2026, 7, 13, 1, 0, 0, tzinfo=timezone.utc)
claim2 = _lease_comment(
session_id=SESSION_FOREIGN,
candidate_head=HEAD_A,
comment_id=COMMENT_10749,
last_activity=unexpired_now - timedelta(minutes=45),
expires_at=expires,
)
active2 = leases.find_active_reviewer_lease(
[claim2], pr_number=PR, now=unexpired_now
)
self.assertIsNotNone(active2)
self.assertEqual(active2["freshness"], "stale_warning")
diag2 = leases.diagnose_reviewer_pr_lease_handoff(
[claim2],
pr_number=PR,
current_session_id="fresh-reviewer",
current_reviewer_identity="sysadmin",
current_head_sha=HEAD_B,
formal_reviews=_reviews_for_head(HEAD_A, "REQUEST_CHANGES"),
worktree_exists=True,
worktree_clean=True,
now=unexpired_now,
)
self.assertEqual(diag2["classification"], "foreign_completed_superseded_head")
self.assertEqual(
diag2["next_action"], leases.NEXT_ACTION_CLEANUP_OBSOLETE_LEASE
)
acq = leases.assess_acquire_lease(
[claim2],
pr_number=PR,
reviewer_identity="sysadmin",
profile="prgs-reviewer",
session_id="fresh-reviewer",
repo=REPO,
issue_number=ISSUE,
worktree="branches/review-pr-688-new",
candidate_head=HEAD_B,
target_branch="master",
target_branch_sha="b" * 40,
now=unexpired_now,
)
self.assertFalse(acq["acquire_allowed"])
def test_missing_controller_auth_denied(self):
now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
claim = _lease_comment(expires_at=expires, last_activity=expires - timedelta(hours=1))
result = _assess([claim], controller=False, now=now)
self.assertFalse(result["cleanup_allowed"])
def test_wrong_confirmation_denied_on_apply(self):
now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
claim = _lease_comment(expires_at=expires, last_activity=expires - timedelta(hours=1))
result = _assess(
[claim], now=now, apply=True, confirmation="MERGE PR 688"
)
self.assertFalse(result["cleanup_allowed"])
self.assertTrue(any("confirmation" in r for r in result["reasons"]))
def test_owner_session_must_use_owner_release(self):
now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
claim = _lease_comment(expires_at=expires, last_activity=expires - timedelta(hours=1))
result = _assess(
[claim],
now=now,
requesting_session=SESSION_FOREIGN,
)
self.assertFalse(result["cleanup_allowed"])
self.assertTrue(any("owns the lease" in r for r in result["reasons"]))
def test_superseded_without_terminal_review_denied(self):
# Pre-expiry the missing terminal review is ambiguous (fail closed);
# post-expiry with unknown owner evidence it is a crash-orphan (#702)
# that still fails closed until owner-process exit is proven.
now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
claim = _lease_comment(expires_at=expires, last_activity=expires - timedelta(hours=1))
result = _assess(
[claim], formal_reviews=[], now=now, owner_process_alive=None
)
self.assertFalse(result["cleanup_allowed"])
self.assertEqual(
result["classification"], "orphaned_expired_superseded_head"
)
fresh_expires = now + timedelta(minutes=60)
fresh = _lease_comment(
expires_at=fresh_expires, last_activity=now - timedelta(minutes=5)
)
pre_expiry = _assess([fresh], formal_reviews=[], now=now)
self.assertFalse(pre_expiry["cleanup_allowed"])
self.assertEqual(
pre_expiry["classification"], "ambiguous_conflicting_evidence"
)
class TestIssue691DiagnoseClassifications(unittest.TestCase):
def setUp(self):
leases.clear_session_lease()
def test_foreign_active_current_head_still_wait(self):
now = datetime.now(timezone.utc)
claim = _lease_comment(
candidate_head=HEAD_B,
last_activity=now - timedelta(minutes=5),
expires_at=now + timedelta(hours=1),
)
claim["id"] = 1
result = leases.diagnose_reviewer_pr_lease_handoff(
[claim],
pr_number=PR,
current_session_id="other",
current_reviewer_identity="sysadmin",
current_head_sha=HEAD_B,
formal_reviews=[],
now=now,
)
self.assertEqual(result["classification"], "foreign_active_current_head")
self.assertEqual(result["next_action"], leases.NEXT_ACTION_WAIT)
self.assertFalse(result["mutation_allowed"])
if __name__ == "__main__":
unittest.main()
@@ -1,693 +0,0 @@
"""Regression tests for the PR #703 REQUEST_CHANGES findings F1F6 (#702).
Formal review at head 889931d independently reproduced six defects:
F1 — stale-binding recovery ran only AFTER the terminal launcher probe in
``gitea_resolve_task_capability``; a worktree removed mid-session wedged
every mutation-task resolution on 'missing cwd' before recovery.
F2 — ``_resolve_preflight_workspace_path`` skipped the existence checks the
canonical mutation context applies; a dead binding could produce empty
porcelain and read as a clean workspace.
F3 — ``orphaned_expired_superseded_head`` cleanup accepted unknown
``worktree_exists``/``worktree_clean`` evidence.
F4 — the 4-hour session-state TTL silently discarded recovery-critical
crash-orphan shadow evidence.
F5 — the single same-profile shadow slot let concurrent sessions overwrite
each other's crash evidence.
F6 — the environment was cleared BEFORE the audit record was persisted, and
audit-write failure was swallowed.
"""
from __future__ import annotations
import json
import os
import shutil
import sys
import tempfile
import unittest
from datetime import datetime, timedelta, timezone
from pathlib import Path
from unittest.mock import patch
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import mcp_session_state as mss # noqa: E402
import namespace_workspace_binding as nwb # noqa: E402
import reviewer_pr_lease as leases # noqa: E402
import stale_binding_recovery as sbr # noqa: E402
import mcp_server # noqa: E402
REPO = "Scaled-Tech-Consulting/Gitea-Tools"
OLD_HEAD = "b" * 40
NEW_HEAD = "6" * 40
CONFIG = {
"version": 2,
"contexts": {
"ctx": {
"enabled": True,
"gitea": {"enabled": True, "base_url": "https://gitea.example.com"},
}
},
"profiles": {
"prgs-author": {
"enabled": True,
"context": "ctx",
"role": "author",
"username": "jcwalker3",
"auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
"allowed_operations": [
"gitea.read", "gitea.issue.create", "gitea.issue.comment",
"gitea.pr.create", "gitea.branch.push",
],
"forbidden_operations": [
"gitea.pr.approve", "gitea.pr.merge", "gitea.pr.review",
],
"execution_profile": "prgs-author",
},
},
"rules": {"allow_runtime_switching": False},
}
def _now() -> datetime:
return datetime.now(timezone.utc)
class _CapabilityHarness(unittest.TestCase):
"""Shared config/env plumbing for resolve-capability integration tests."""
def setUp(self):
self._remotes = patch.dict(mcp_server.REMOTES, {
"prgs": {
"host": "gitea.example.com",
"org": "Scaled-Tech-Consulting",
"repo": "Gitea-Tools",
},
})
self._remotes.start()
mcp_server._IDENTITY_CACHE.clear()
self._dir = tempfile.TemporaryDirectory()
self.config_path = os.path.join(self._dir.name, "profiles.json")
with open(self.config_path, "w", encoding="utf-8") as fh:
fh.write(json.dumps(CONFIG))
def tearDown(self):
self._remotes.stop()
mcp_server._IDENTITY_CACHE.clear()
self._dir.cleanup()
def _env(self, **extra):
env = {
"GITEA_MCP_CONFIG": self.config_path,
"GITEA_MCP_PROFILE": "prgs-author",
"GITEA_TOKEN_AUTHOR": "author-pass",
}
env.update(extra)
return env
def _deleted_worktree(self) -> str:
path = tempfile.mkdtemp(prefix="gitea-removed-worktree-")
shutil.rmtree(path)
return path
class TestF1RecoveryBeforeTerminalProbe(_CapabilityHarness):
"""F1: recovery must run before the terminal cwd probe can wedge."""
@patch("mcp_server.api_request", return_value={"login": "jcwalker3"})
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_removed_worktree_recovers_before_probe(self, _auth, _api):
missing = self._deleted_worktree()
env = self._env(
GITEA_ACTIVE_WORKTREE=missing,
GITEA_FORCE_TERMINAL_PROBE="1",
GITEA_FORCE_STALE_BINDING_RECOVERY="1",
)
with patch.dict(os.environ, env):
res = mcp_server.gitea_resolve_task_capability(
task="create_pr", remote="prgs"
)
# The provably-stale binding was cleared before the probe ran.
self.assertNotIn("GITEA_ACTIVE_WORKTREE", os.environ)
self.assertFalse(res.get("terminal_launcher_unhealthy"))
report = res.get("stale_binding_recovery") or {}
self.assertEqual(
report.get("classification"),
sbr.CLASSIFICATION_PROVABLY_STALE_MISSING_PATH,
)
self.assertTrue(report.get("recovery_performed"))
self.assertTrue(res.get("allowed_in_current_session"))
@patch("mcp_server.api_request", return_value={"login": "jcwalker3"})
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_probe_block_still_carries_recovery_report(self, _auth, _api):
# Recovery disabled (test mode without the force flag) preserves the
# fail-closed probe block, but the block must now carry the
# classification evidence instead of hiding it.
missing = self._deleted_worktree()
env = self._env(
GITEA_ACTIVE_WORKTREE=missing,
GITEA_FORCE_TERMINAL_PROBE="1",
)
with patch.dict(os.environ, env):
res = mcp_server.gitea_resolve_task_capability(
task="create_pr", remote="prgs"
)
self.assertIn("GITEA_ACTIVE_WORKTREE", os.environ)
self.assertTrue(res.get("terminal_launcher_unhealthy"))
report = res.get("stale_binding_recovery") or {}
self.assertEqual(
report.get("classification"),
sbr.CLASSIFICATION_PROVABLY_STALE_MISSING_PATH,
)
self.assertFalse(report.get("recovery_performed"))
@patch("mcp_server.api_request", return_value={"login": "jcwalker3"})
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_unverified_binding_is_never_cleared_by_reordering(self, _auth, _api):
# F1 must not weaken authorization: an existing, uncorroborated
# binding survives resolution untouched even with recovery forced.
existing = tempfile.mkdtemp(prefix="gitea-existing-worktree-")
self.addCleanup(shutil.rmtree, existing, ignore_errors=True)
env = self._env(
GITEA_ACTIVE_WORKTREE=existing,
GITEA_FORCE_TERMINAL_PROBE="1",
GITEA_FORCE_STALE_BINDING_RECOVERY="1",
)
with patch.dict(os.environ, env):
res = mcp_server.gitea_resolve_task_capability(
task="create_pr", remote="prgs"
)
self.assertEqual(os.environ.get("GITEA_ACTIVE_WORKTREE"), existing)
report = res.get("stale_binding_recovery") or {}
self.assertFalse(report.get("recovery_performed"))
class TestF2PreflightPathVerification(_CapabilityHarness):
"""F2: preflight and mutation contexts verify resolved paths alike."""
def test_preflight_and_mutation_context_agree_on_dead_binding(self):
missing = self._deleted_worktree()
env = self._env(GITEA_ACTIVE_WORKTREE=missing)
with patch.dict(os.environ, env):
preflight = mcp_server._resolve_preflight_workspace_path(None)
mutation = mcp_server._resolve_namespace_mutation_context(None)
self.assertEqual(preflight, mutation["workspace_path"])
self.assertNotEqual(preflight, os.path.realpath(missing))
def test_missing_explicit_worktree_never_reads_clean(self):
missing = self._deleted_worktree()
with patch.dict(os.environ, self._env()):
porcelain = mcp_server._get_workspace_porcelain(worktree_path=missing)
self.assertEqual(
porcelain, mcp_server.PREFLIGHT_WORKSPACE_UNAVAILABLE_PORCELAIN
)
# The sentinel parses as a tracked dirty entry — never clean.
self.assertTrue(mcp_server._parse_porcelain_entries(porcelain))
def test_workspace_deleted_during_evaluation_never_reads_clean(self):
# TOCTOU: resolution succeeded, then the path vanished before git ran.
missing = self._deleted_worktree()
with patch.dict(os.environ, self._env()):
with patch.object(
mcp_server,
"_resolve_preflight_workspace_path",
return_value=missing,
):
porcelain = mcp_server._get_workspace_porcelain()
self.assertEqual(
porcelain, mcp_server.PREFLIGHT_WORKSPACE_UNAVAILABLE_PORCELAIN
)
def test_git_failure_never_reads_clean(self):
class _Failed:
returncode = 128
stdout = ""
stderr = "fatal: not a git repository"
with patch.dict(os.environ, self._env()):
with patch.object(
mcp_server.subprocess, "run", return_value=_Failed()
):
porcelain = mcp_server._get_workspace_porcelain()
self.assertEqual(
porcelain, mcp_server.PREFLIGHT_WORKSPACE_UNAVAILABLE_PORCELAIN
)
def test_symlinked_binding_to_deleted_target_is_demoted(self):
target = tempfile.mkdtemp(prefix="gitea-symlink-target-")
holder = tempfile.mkdtemp(prefix="gitea-symlink-holder-")
self.addCleanup(shutil.rmtree, holder, ignore_errors=True)
link = os.path.join(holder, "worktree-link")
os.symlink(target, link)
shutil.rmtree(target)
demotions: list[str] = []
_path, source = nwb.resolve_namespace_workspace(
role_kind="reviewer",
process_project_root=os.getcwd(),
env={"GITEA_ACTIVE_WORKTREE": link},
demotions=demotions,
verify_paths=True,
)
self.assertEqual(source, "MCP server process root (default)")
self.assertEqual(len(demotions), 1)
def test_dead_binding_falls_through_to_live_role_binding(self):
# Path changes during evaluation: the dead candidate demotes at
# verification time and never resurrects over the live fallback.
alive = tempfile.mkdtemp(prefix="gitea-alive-worktree-")
self.addCleanup(shutil.rmtree, alive, ignore_errors=True)
missing = self._deleted_worktree()
path, source = nwb.resolve_namespace_workspace(
role_kind="reviewer",
process_project_root=os.getcwd(),
env={
"GITEA_ACTIVE_WORKTREE": missing,
"GITEA_REVIEWER_WORKTREE": alive,
},
verify_paths=True,
)
self.assertEqual(path, os.path.realpath(alive))
self.assertEqual(
source, "GITEA_REVIEWER_WORKTREE environment variable"
)
class TestF3AffirmativeWorktreeEvidence(unittest.TestCase):
"""F3: unknown worktree evidence must fail closed for crash orphans."""
def _comments(self) -> list[dict]:
stale = _now() - timedelta(hours=3)
body = leases.format_lease_body(
repo=REPO,
pr_number=701,
issue_number=699,
reviewer_identity="sysadmin",
profile="prgs-reviewer",
session_id="65664-4f248d213d60",
worktree="branches/review-pr-654",
phase="validation-start",
candidate_head=OLD_HEAD,
target_branch="master",
target_branch_sha="2" * 40,
last_activity=stale,
expires_at=stale + timedelta(minutes=120),
)
return [{
"id": 11131,
"body": body,
"user": {"login": "sysadmin"},
"author": "sysadmin",
"created_at": stale.isoformat(),
"updated_at": stale.isoformat(),
}]
def _assess(self, **kwargs):
defaults = dict(
pr_number=701,
current_head_sha=NEW_HEAD,
formal_reviews=[],
repo=REPO,
expected_repo=REPO,
requesting_session_id="fresh-controller",
controller_recovery_authorized=True,
owner_process_alive=False,
)
defaults.update(kwargs)
return leases.assess_obsolete_reviewer_comment_lease_cleanup(
self._comments(), **defaults
)
def test_unknown_worktree_evidence_fails_closed(self):
result = self._assess(worktree_exists=None, worktree_clean=None)
self.assertEqual(
result["classification"], "orphaned_expired_superseded_head"
)
self.assertFalse(result["cleanup_allowed"])
self.assertIn(
"affirmative safe evidence",
" ".join(result["fail_closed_reasons"]),
)
def test_unknown_cleanliness_with_existing_worktree_fails_closed(self):
result = self._assess(worktree_exists=True, worktree_clean=None)
self.assertFalse(result["cleanup_allowed"])
def test_affirmative_clean_worktree_is_eligible(self):
result = self._assess(worktree_exists=True, worktree_clean=True)
self.assertEqual(
result["classification"], "orphaned_expired_superseded_head"
)
self.assertTrue(result["cleanup_allowed"])
def test_affirmative_absent_worktree_is_eligible(self):
result = self._assess(worktree_exists=False, worktree_clean=None)
self.assertTrue(result["cleanup_allowed"])
def test_matrix_matches_diagnosis_gate(self):
# The cleanup matrix and the diagnosis path must agree: unknown
# evidence yields acquire-only, affirmative evidence yields cleanup.
diagnosis_unknown = leases.diagnose_reviewer_pr_lease_handoff(
self._comments(),
pr_number=701,
current_session_id=None,
current_reviewer_identity="fresh-reviewer",
current_head_sha=NEW_HEAD,
formal_reviews=[],
owner_process_alive=False,
worktree_exists=None,
worktree_clean=None,
)
assess_unknown = self._assess(worktree_exists=None, worktree_clean=None)
self.assertNotEqual(
diagnosis_unknown["next_action"],
"cleanup_obsolete_reviewer_comment_lease",
)
self.assertFalse(assess_unknown["cleanup_allowed"])
class TestF4CrashShadowDurability(unittest.TestCase):
"""F4: crash-orphan evidence survives the former 4h TTL boundary."""
SID = "82984-abcabcabcabc"
def setUp(self):
self._dir = tempfile.TemporaryDirectory()
self._env = patch.dict(
os.environ, {"GITEA_MCP_SESSION_STATE_DIR": self._dir.name}
)
self._env.start()
leases._SESSION_LEASE = None
def tearDown(self):
self._env.stop()
self._dir.cleanup()
leases._SESSION_LEASE = None
def _age_record(self, kind: str, hours: float, instance_id: str | None = None):
path = mss.state_file_path(kind=kind, instance_id=instance_id)
with open(path, encoding="utf-8") as fh:
envelope = json.load(fh)
stamp = (
(_now() - timedelta(hours=hours)).isoformat().replace("+00:00", "Z")
)
envelope["recorded_at"] = stamp
envelope["updated_at"] = stamp
envelope["payload"]["recorded_at"] = stamp
envelope["payload"]["updated_at"] = stamp
with open(path, "w", encoding="utf-8") as fh:
json.dump(envelope, fh)
def _save_shadow(self) -> None:
mss.save_state(
kind=mss.KIND_REVIEWER_SESSION_LEASE,
instance_id=self.SID,
payload={"session_id": self.SID, "owner_pid": os.getpid()},
)
def _load_shadow(self):
return mss.load_state(
kind=mss.KIND_REVIEWER_SESSION_LEASE, instance_id=self.SID
)
def test_shadow_loads_before_former_ttl_boundary(self):
self._save_shadow()
self._age_record(
mss.KIND_REVIEWER_SESSION_LEASE, hours=3.9, instance_id=self.SID
)
self.assertIsNotNone(self._load_shadow())
def test_shadow_loads_at_former_ttl_boundary(self):
self._save_shadow()
self._age_record(
mss.KIND_REVIEWER_SESSION_LEASE, hours=4.0, instance_id=self.SID
)
self.assertIsNotNone(self._load_shadow())
def test_shadow_loads_long_after_former_ttl_boundary(self):
self._save_shadow()
self._age_record(
mss.KIND_REVIEWER_SESSION_LEASE, hours=27.0, instance_id=self.SID
)
record = self._load_shadow()
self.assertIsNotNone(record)
self.assertEqual(record.get("session_id"), self.SID)
def test_stale_binding_audit_record_is_also_durable(self):
mss.save_state(
kind=mss.KIND_STALE_BINDING_RECOVERY,
payload={"cleared_env": "GITEA_ACTIVE_WORKTREE"},
)
self._age_record(mss.KIND_STALE_BINDING_RECOVERY, hours=27.0)
self.assertIsNotNone(
mss.load_state(kind=mss.KIND_STALE_BINDING_RECOVERY)
)
def test_non_recovery_kinds_keep_ttl(self):
mss.save_state(
kind=mss.KIND_WORKFLOW_LOAD, payload={"workflow_hash": "abc"}
)
self._age_record(mss.KIND_WORKFLOW_LOAD, hours=5.0)
self.assertIsNone(mss.load_state(kind=mss.KIND_WORKFLOW_LOAD))
def test_sanctioned_clear_is_the_terminal_reconciliation(self):
self._save_shadow()
self._age_record(
mss.KIND_REVIEWER_SESSION_LEASE, hours=27.0, instance_id=self.SID
)
mss.clear_state(
kind=mss.KIND_REVIEWER_SESSION_LEASE, instance_id=self.SID
)
self.assertIsNone(self._load_shadow())
self.assertEqual(
mss.list_states(kind=mss.KIND_REVIEWER_SESSION_LEASE), []
)
def test_crash_orphan_recovery_works_after_former_ttl(self):
# End to end: a crashed session's shadow still proves owner exit a
# day later.
leases.record_session_lease(
{
"pr_number": 701,
"session_id": self.SID,
"worktree": "branches/review-pr-654",
"candidate_head": OLD_HEAD,
"repo": REPO,
"comment_id": 11131,
"profile": "prgs-reviewer",
"reviewer_identity": "sysadmin",
"phase": "claimed",
},
lease_provenance={"source": "acquire", "comment_id": 11131},
)
leases._SESSION_LEASE = None # simulated crash: no sanctioned clear
self._age_record(
mss.KIND_REVIEWER_SESSION_LEASE, hours=27.0, instance_id=self.SID
)
shadow = leases.load_session_lease_shadow(session_id=self.SID)
self.assertIsNotNone(shadow)
verdict = leases.assess_crashed_session_lease_orphan(
shadow,
lease={"session_id": self.SID},
current_pid=os.getpid() + 1,
pid_alive=False,
)
self.assertTrue(verdict["orphan_evidence"])
self.assertIs(verdict["owner_process_alive"], False)
class TestF5ConcurrentSessionShadows(unittest.TestCase):
"""F5: concurrent same-profile sessions keep distinct shadow records."""
SID_A = "11111-aaaaaaaaaaaa"
SID_B = "22222-bbbbbbbbbbbb"
def setUp(self):
self._dir = tempfile.TemporaryDirectory()
self._env = patch.dict(
os.environ, {"GITEA_MCP_SESSION_STATE_DIR": self._dir.name}
)
self._env.start()
leases._SESSION_LEASE = None
def tearDown(self):
self._env.stop()
self._dir.cleanup()
leases._SESSION_LEASE = None
def _lease(self, session_id: str, pr_number: int, head: str) -> dict:
return {
"pr_number": pr_number,
"session_id": session_id,
"worktree": f"branches/review-pr-{pr_number}-{session_id[:5]}",
"candidate_head": head,
"repo": REPO,
"comment_id": 11221,
"profile": "prgs-reviewer",
"reviewer_identity": "sysadmin",
"phase": "claimed",
}
def _record(self, lease: dict) -> None:
leases.record_session_lease(
lease,
lease_provenance={"source": "acquire", "comment_id": 11221},
)
def test_interleaved_sessions_do_not_overwrite_each_other(self):
self._record(self._lease(self.SID_A, 703, OLD_HEAD))
self._record(self._lease(self.SID_B, 701, NEW_HEAD))
# Session A heartbeats again after B wrote.
updated_a = self._lease(self.SID_A, 703, OLD_HEAD)
updated_a["phase"] = "validation-start"
self._record(updated_a)
shadow_a = leases.load_session_lease_shadow(session_id=self.SID_A)
shadow_b = leases.load_session_lease_shadow(session_id=self.SID_B)
self.assertEqual(shadow_a.get("session_id"), self.SID_A)
self.assertEqual(shadow_a.get("pr_number"), 703)
self.assertEqual(shadow_a.get("phase"), "validation-start")
self.assertEqual(shadow_b.get("session_id"), self.SID_B)
self.assertEqual(shadow_b.get("pr_number"), 701)
self.assertEqual(shadow_b.get("candidate_head"), NEW_HEAD)
def test_shadow_lookup_never_misattributes(self):
self._record(self._lease(self.SID_A, 703, OLD_HEAD))
self.assertIsNone(
leases.load_session_lease_shadow(session_id=self.SID_B)
)
verdict = leases.assess_crashed_session_lease_orphan(
leases.load_session_lease_shadow(session_id=self.SID_A),
lease={"session_id": self.SID_B},
pid_alive=False,
)
self.assertFalse(verdict["orphan_evidence"])
def test_clearing_one_session_preserves_the_other(self):
self._record(self._lease(self.SID_A, 703, OLD_HEAD))
self._record(self._lease(self.SID_B, 701, NEW_HEAD))
# Sanctioned clear of B (the currently recorded in-memory lease).
leases.clear_session_lease()
self.assertIsNone(
leases.load_session_lease_shadow(session_id=self.SID_B)
)
shadow_a = leases.load_session_lease_shadow(session_id=self.SID_A)
self.assertIsNotNone(shadow_a)
self.assertEqual(shadow_a.get("session_id"), self.SID_A)
def test_reconnected_process_recovers_by_session_id(self):
self._record(self._lease(self.SID_A, 703, OLD_HEAD))
self._record(self._lease(self.SID_B, 701, NEW_HEAD))
# Simulated crash + reconnect: in-memory state is gone, durable
# records remain enumerable and individually attributable.
leases._SESSION_LEASE = None
records = mss.list_states(kind=mss.KIND_REVIEWER_SESSION_LEASE)
self.assertEqual(
sorted(r.get("session_id") for r in records),
sorted([self.SID_A, self.SID_B]),
)
shadow = leases.load_session_lease_shadow(session_id=self.SID_B)
self.assertEqual(shadow.get("pr_number"), 701)
def test_legacy_single_slot_record_still_resolves_by_session_id(self):
# Upgrade path: a record written by pre-F5 code (no instance key)
# remains usable when its payload names the requested session.
mss.save_state(
kind=mss.KIND_REVIEWER_SESSION_LEASE,
payload={"session_id": self.SID_A, "owner_pid": os.getpid()},
)
shadow = leases.load_session_lease_shadow(session_id=self.SID_A)
self.assertIsNotNone(shadow)
self.assertIsNone(
leases.load_session_lease_shadow(session_id=self.SID_B)
)
class TestF6AuditBeforeClear(_CapabilityHarness):
"""F6: the durable audit record is persisted before the env clears."""
def _recovery_env(self, missing: str) -> dict:
return self._env(
GITEA_ACTIVE_WORKTREE=missing,
GITEA_FORCE_STALE_BINDING_RECOVERY="1",
)
def test_audit_write_failure_blocks_the_clear(self):
missing = self._deleted_worktree()
with patch.dict(os.environ, self._recovery_env(missing)):
with patch.object(
mss, "save_state", side_effect=OSError("disk full")
):
report = mcp_server._assess_stale_active_binding(
auto_recover=True
)
self.assertEqual(os.environ.get("GITEA_ACTIVE_WORKTREE"), missing)
self.assertFalse(report["recovery_performed"])
self.assertIs(report.get("audit_persisted"), False)
self.assertTrue(report.get("audit_write_failed"))
self.assertIn("NOT cleared", " ".join(report.get("reasons") or []))
def test_retry_after_audit_failure_recovers(self):
missing = self._deleted_worktree()
with patch.dict(os.environ, self._recovery_env(missing)):
with patch.object(
mss, "save_state", side_effect=OSError("disk full")
):
first = mcp_server._assess_stale_active_binding(
auto_recover=True
)
self.assertFalse(first["recovery_performed"])
# Persistence recovered; the retry clears with a durable audit.
second = mcp_server._assess_stale_active_binding(auto_recover=True)
self.assertTrue(second["recovery_performed"])
self.assertIs(second.get("audit_persisted"), True)
self.assertNotIn("GITEA_ACTIVE_WORKTREE", os.environ)
audit = mss.load_state(kind=mss.KIND_STALE_BINDING_RECOVERY)
self.assertIsNotNone(audit)
self.assertEqual(audit.get("recovery_status"), "performed")
self.assertEqual(audit.get("cleared_value"), missing)
def test_audit_record_exists_before_env_clear(self):
missing = self._deleted_worktree()
observed: list[tuple[str | None, str | None]] = []
real_save = mss.save_state
def _spy(**kwargs):
observed.append(
(
(kwargs.get("payload") or {}).get("recovery_status"),
os.environ.get("GITEA_ACTIVE_WORKTREE"),
)
)
return real_save(**kwargs)
with patch.dict(os.environ, self._recovery_env(missing)):
with patch.object(mss, "save_state", side_effect=_spy):
report = mcp_server._assess_stale_active_binding(
auto_recover=True
)
self.assertTrue(report["recovery_performed"])
self.assertGreaterEqual(len(observed), 2)
pending_status, env_at_pending = observed[0]
performed_status, env_at_performed = observed[-1]
# Ordering proof: the binding was still present while the pending
# audit persisted, and gone by the time the performed status wrote.
self.assertEqual(pending_status, "pending")
self.assertEqual(env_at_pending, missing)
self.assertEqual(performed_status, "performed")
self.assertIsNone(env_at_performed)
def test_recovery_is_idempotent_after_success(self):
missing = self._deleted_worktree()
with patch.dict(os.environ, self._recovery_env(missing)):
first = mcp_server._assess_stale_active_binding(auto_recover=True)
second = mcp_server._assess_stale_active_binding(auto_recover=True)
self.assertTrue(first["recovery_performed"])
self.assertEqual(second["classification"], sbr.CLASSIFICATION_UNBOUND)
self.assertFalse(second["recovery_performed"])
if __name__ == "__main__":
unittest.main()
@@ -1,517 +0,0 @@
"""Regression tests for #702: stale runtime binding + orphaned durable lease.
Reproduces the PR #701 incident (lease 65664-4f248d213d60, comments
11129/11131): the MCP daemon crashed without teardown, destroying the
in-memory session lease while the durable comment lease survived, and the
auto-reconnected daemon inherited a stale ``GITEA_ACTIVE_WORKTREE`` binding
(``branches/review-pr-654``) from its parent environment.
Covers the five mandated scenarios:
1. lost in-session leases — durable shadow survives a crash and yields
provable owner-process evidence
2. old-head leases — expired superseded-head lease with no terminal
review becomes recoverable only with orphan evidence + controller authority
3. runtime/worktree mismatch — env bindings to deleted worktrees are demoted,
never silently bound
4. managed reconnect — boot-inherited uncorroborated bindings are
surfaced for managed recovery; provably stale ones are clear-eligible
5. safe expiry — pre-expiry stays fail-closed wait (no steal);
canonical expiry unlocks acquisition and guarded cleanup
"""
from __future__ import annotations
import os
import sys
import unittest
from datetime import datetime, timedelta, timezone
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import namespace_workspace_binding as nwb # noqa: E402
import reviewer_pr_lease as leases # noqa: E402
import stale_binding_recovery as sbr # noqa: E402
# PR #701 incident fixtures.
OLD_HEAD = "b4d0cb22e452a07c8e816745c704ddb0f43edf0b"
NEW_HEAD = "6b675f5c834b41f9d74e8a54294ff44dddf28ae4"
CRASHED_SESSION = "65664-4f248d213d60"
LEASE_COMMENT = 11131
REPO = "Scaled-Tech-Consulting/Gitea-Tools"
PR = 701
ISSUE = 699
LEASE_WORKTREE = "branches/review-fix-issue-699-structured-auth-mcp-errors"
def _now() -> datetime:
return datetime.now(timezone.utc)
def _lease_comment(
*,
phase: str = "validation-start",
candidate_head: str = OLD_HEAD,
session_id: str = CRASHED_SESSION,
comment_id: int = LEASE_COMMENT,
last_activity: datetime | None = None,
expires_at: datetime | None = None,
worktree: str = LEASE_WORKTREE,
) -> dict:
stamp = last_activity or _now()
body = leases.format_lease_body(
repo=REPO,
pr_number=PR,
issue_number=ISSUE,
reviewer_identity="sysadmin",
profile="prgs-reviewer",
session_id=session_id,
worktree=worktree,
phase=phase,
candidate_head=candidate_head,
target_branch="master",
target_branch_sha="2" * 40,
last_activity=stamp,
expires_at=expires_at,
)
return {
"id": comment_id,
"body": body,
"user": {"login": "sysadmin"},
"author": "sysadmin",
"created_at": stamp.isoformat(),
"updated_at": stamp.isoformat(),
}
def _expired_superseded_comments() -> list[dict]:
stale = _now() - timedelta(hours=3)
return [_lease_comment(
last_activity=stale,
expires_at=stale + timedelta(minutes=120),
)]
def _fresh_superseded_comments() -> list[dict]:
recent = _now() - timedelta(minutes=10)
return [_lease_comment(
last_activity=recent,
expires_at=recent + timedelta(minutes=120),
)]
class TestLostInSessionLeaseShadow(unittest.TestCase):
"""Scenario 1: the durable shadow survives a daemon crash (#702 AC3)."""
def setUp(self):
leases._SESSION_LEASE = None
def tearDown(self):
leases._SESSION_LEASE = None
def _record(self) -> dict:
return leases.record_session_lease(
{
"pr_number": PR,
"issue_number": ISSUE,
"session_id": CRASHED_SESSION,
"reviewer_identity": "sysadmin",
"profile": "prgs-reviewer",
"worktree": LEASE_WORKTREE,
"phase": "claimed",
"candidate_head": OLD_HEAD,
"repo": REPO,
"comment_id": LEASE_COMMENT,
},
lease_provenance={"source": "acquire", "comment_id": LEASE_COMMENT},
)
def test_shadow_written_on_record_and_survives_simulated_crash(self):
self._record()
# Simulated unhandled crash: the in-memory dict dies with the process
# but the sanctioned clear path never runs.
leases._SESSION_LEASE = None
shadow = leases.load_session_lease_shadow()
self.assertIsNotNone(shadow)
self.assertEqual(shadow.get("session_id"), CRASHED_SESSION)
self.assertEqual(shadow.get("pr_number"), PR)
self.assertEqual(int(shadow.get("owner_pid")), os.getpid())
def test_sanctioned_clear_removes_shadow(self):
self._record()
leases.clear_session_lease()
self.assertIsNone(leases.load_session_lease_shadow())
def test_orphan_assessment_dead_owner_pid_proves_exit(self):
self._record()
leases._SESSION_LEASE = None
shadow = leases.load_session_lease_shadow()
lease = {"session_id": CRASHED_SESSION}
verdict = leases.assess_crashed_session_lease_orphan(
shadow, lease=lease, current_pid=os.getpid() + 1, pid_alive=False
)
self.assertTrue(verdict["orphan_evidence"])
self.assertIs(verdict["owner_process_alive"], False)
def test_orphan_assessment_alive_pid_is_not_ownership_proof(self):
self._record()
shadow = leases.load_session_lease_shadow()
verdict = leases.assess_crashed_session_lease_orphan(
shadow,
lease={"session_id": CRASHED_SESSION},
current_pid=os.getpid() + 1,
pid_alive=True,
)
self.assertFalse(verdict["orphan_evidence"])
self.assertIsNone(verdict["owner_process_alive"])
def test_orphan_assessment_session_mismatch_proves_nothing(self):
self._record()
shadow = leases.load_session_lease_shadow()
verdict = leases.assess_crashed_session_lease_orphan(
shadow, lease={"session_id": "other-999"}, pid_alive=False
)
self.assertFalse(verdict["orphan_evidence"])
self.assertIsNone(verdict["owner_process_alive"])
def test_orphan_assessment_without_shadow_is_unknown(self):
verdict = leases.assess_crashed_session_lease_orphan(
None, lease={"session_id": CRASHED_SESSION}
)
self.assertFalse(verdict["orphan_evidence"])
self.assertIsNone(verdict["owner_process_alive"])
class TestOldHeadLeaseCleanup(unittest.TestCase):
"""Scenario 2: expired superseded-head lease with no terminal review."""
def _assess(self, comments, **kwargs):
defaults = dict(
pr_number=PR,
current_head_sha=NEW_HEAD,
formal_reviews=[],
repo=REPO,
expected_repo=REPO,
requesting_session_id="fresh-controller",
controller_recovery_authorized=True,
worktree_exists=True,
worktree_clean=True,
owner_process_alive=False,
)
defaults.update(kwargs)
return leases.assess_obsolete_reviewer_comment_lease_cleanup(
comments, **defaults
)
def test_expired_orphan_with_full_evidence_is_cleanup_eligible(self):
result = self._assess(_expired_superseded_comments())
self.assertEqual(result["classification"], "orphaned_expired_superseded_head")
self.assertTrue(result["cleanup_allowed"])
self.assertIn("phase: released", result["release_body"] or "")
self.assertEqual(
result["exact_next_action"], "cleanup_obsolete_reviewer_comment_lease"
)
def test_expired_orphan_without_owner_evidence_fails_closed(self):
result = self._assess(
_expired_superseded_comments(), owner_process_alive=None
)
self.assertFalse(result["cleanup_allowed"])
self.assertIn(
"provable owner-process-exit evidence",
" ".join(result["fail_closed_reasons"]),
)
def test_expired_orphan_without_controller_authority_fails_closed(self):
result = self._assess(
_expired_superseded_comments(), controller_recovery_authorized=False
)
self.assertFalse(result["cleanup_allowed"])
def test_expired_orphan_with_dirty_worktree_fails_closed(self):
result = self._assess(
_expired_superseded_comments(),
worktree_clean=False,
worktree_has_unpreserved_work=True,
)
self.assertFalse(result["cleanup_allowed"])
def test_unexpired_superseded_no_terminal_stays_ambiguous_wait(self):
# Regression guard: pre-expiry there is still no steal path.
result = self._assess(_fresh_superseded_comments())
self.assertEqual(result["classification"], "ambiguous_conflicting_evidence")
self.assertFalse(result["cleanup_allowed"])
self.assertEqual(result["exact_next_action"], "wait")
class TestRuntimeWorktreeMismatch(unittest.TestCase):
"""Scenario 3: env bindings to deleted worktrees are demoted (#702 AC2)."""
def test_missing_active_env_binding_is_demoted(self):
demotions: list[str] = []
missing = "/nonexistent/branches/review-pr-654"
path, source = nwb.resolve_namespace_workspace(
role_kind="reviewer",
process_project_root=os.getcwd(),
env={"GITEA_ACTIVE_WORKTREE": missing},
demotions=demotions,
verify_paths=True,
)
self.assertEqual(source, "MCP server process root (default)")
self.assertEqual(len(demotions), 1)
self.assertIn("no longer exists", demotions[0])
def test_missing_active_falls_through_to_existing_role_env(self):
existing = os.getcwd()
path, source = nwb.resolve_namespace_workspace(
role_kind="reviewer",
process_project_root="/tmp",
env={
"GITEA_ACTIVE_WORKTREE": "/nonexistent/branches/review-pr-654",
"GITEA_REVIEWER_WORKTREE": existing,
},
verify_paths=True,
)
self.assertEqual(path, os.path.realpath(existing))
self.assertEqual(source, "GITEA_REVIEWER_WORKTREE environment variable")
def test_existing_active_env_binding_still_wins(self):
existing = os.getcwd()
path, source = nwb.resolve_namespace_workspace(
role_kind="reviewer",
process_project_root="/tmp",
env={"GITEA_ACTIVE_WORKTREE": existing},
verify_paths=True,
)
self.assertEqual(path, os.path.realpath(existing))
self.assertEqual(source, "GITEA_ACTIVE_WORKTREE environment variable")
def test_explicit_worktree_path_argument_is_never_demoted(self):
missing = "/nonexistent/branches/explicit"
path, source = nwb.resolve_namespace_workspace(
role_kind="reviewer",
worktree_path=missing,
process_project_root="/tmp",
env={},
verify_paths=True,
)
self.assertEqual(source, "worktree_path argument")
def test_mutation_context_reports_demotion_in_ignored_bindings(self):
ctx = nwb.resolve_namespace_mutation_context(
role_kind="reviewer",
worktree_path=None,
process_project_root=os.getcwd(),
env={"GITEA_ACTIVE_WORKTREE": "/nonexistent/branches/review-pr-654"},
)
joined = " ".join(ctx["ignored_bindings"])
self.assertIn("stale binding, #702", joined)
class TestManagedReconnectRecovery(unittest.TestCase):
"""Scenario 4: boot-inherited bindings and the sanctioned recovery plan."""
def test_uncorroborated_inherited_reviewer_binding_is_unverified(self):
classification = sbr.classify_active_worktree_binding(
active_value=os.getcwd(),
role_env_value=None,
session_lease_worktree=None,
boot_inherited=True,
path_exists=True,
role_kind="reviewer",
)
self.assertEqual(
classification["classification"], sbr.CLASSIFICATION_UNVERIFIED_INHERITED
)
self.assertFalse(classification["clear_eligible"])
plan = sbr.plan_recovery(classification)
self.assertFalse(plan["clear_allowed"])
self.assertIn("managed", plan["exact_next_action"])
self.assertIn("do not clear or rewrite", plan["exact_next_action"])
def test_missing_path_binding_is_provably_stale_and_clear_eligible(self):
classification = sbr.classify_active_worktree_binding(
active_value="/nonexistent/branches/review-pr-654",
boot_inherited=True,
path_exists=False,
role_kind="reviewer",
)
self.assertEqual(
classification["classification"],
sbr.CLASSIFICATION_PROVABLY_STALE_MISSING_PATH,
)
plan = sbr.plan_recovery(classification)
self.assertTrue(plan["clear_allowed"])
def test_inherited_binding_superseded_by_session_lease_is_clear_eligible(self):
classification = sbr.classify_active_worktree_binding(
active_value="/tmp",
session_lease_worktree=os.getcwd(),
boot_inherited=True,
path_exists=True,
role_kind="reviewer",
)
self.assertEqual(
classification["classification"],
sbr.CLASSIFICATION_SUPERSEDED_BY_SESSION_LEASE,
)
self.assertTrue(sbr.plan_recovery(classification)["clear_allowed"])
def test_post_boot_binding_conflict_is_surfaced_not_cleared(self):
classification = sbr.classify_active_worktree_binding(
active_value="/tmp",
session_lease_worktree=os.getcwd(),
boot_inherited=False,
path_exists=True,
role_kind="reviewer",
)
self.assertEqual(
classification["classification"], sbr.CLASSIFICATION_UNVERIFIED_INHERITED
)
self.assertFalse(sbr.plan_recovery(classification)["clear_allowed"])
def test_corroborated_bindings_are_untouched(self):
for kwargs in (
dict(role_env_value=os.getcwd()),
dict(session_lease_worktree=os.getcwd()),
):
classification = sbr.classify_active_worktree_binding(
active_value=os.getcwd(),
boot_inherited=True,
path_exists=True,
role_kind="reviewer",
**kwargs,
)
self.assertEqual(
classification["classification"], sbr.CLASSIFICATION_CORROBORATED
)
self.assertFalse(sbr.plan_recovery(classification)["clear_allowed"])
def test_author_inherited_binding_stays_corroborated(self):
classification = sbr.classify_active_worktree_binding(
active_value=os.getcwd(),
boot_inherited=True,
path_exists=True,
role_kind="author",
)
self.assertEqual(
classification["classification"], sbr.CLASSIFICATION_CORROBORATED
)
def test_apply_recovery_clears_env_only_for_authorized_plan(self):
env = {"GITEA_ACTIVE_WORKTREE": "/nonexistent/branches/review-pr-654"}
plan = sbr.plan_recovery(
sbr.classify_active_worktree_binding(
active_value=env["GITEA_ACTIVE_WORKTREE"],
boot_inherited=True,
path_exists=False,
role_kind="reviewer",
)
)
applied = sbr.apply_recovery(plan, env=env)
self.assertTrue(applied["performed"])
self.assertNotIn("GITEA_ACTIVE_WORKTREE", env)
self.assertIn("review-pr-654", applied["cleared_value"])
def test_apply_recovery_refuses_unauthorized_plan(self):
env = {"GITEA_ACTIVE_WORKTREE": os.getcwd()}
plan = sbr.plan_recovery(
sbr.classify_active_worktree_binding(
active_value=env["GITEA_ACTIVE_WORKTREE"],
boot_inherited=True,
path_exists=True,
role_kind="reviewer",
)
)
applied = sbr.apply_recovery(plan, env=env)
self.assertFalse(applied["performed"])
self.assertIn("GITEA_ACTIVE_WORKTREE", env)
class TestSafeExpiryDiagnosis(unittest.TestCase):
"""Scenario 5: canonical expiry flips wait into acquire/guarded cleanup."""
def _diagnose(self, comments, **kwargs):
defaults = dict(
pr_number=PR,
current_session_id=None,
current_reviewer_identity="fresh-reviewer",
current_head_sha=NEW_HEAD,
formal_reviews=[],
)
defaults.update(kwargs)
return leases.diagnose_reviewer_pr_lease_handoff(comments, **defaults)
def setUp(self):
leases._SESSION_LEASE = None
def test_pre_expiry_superseded_no_terminal_stays_wait(self):
diagnosis = self._diagnose(_fresh_superseded_comments())
self.assertEqual(
diagnosis["classification"], "ambiguous_conflicting_evidence"
)
self.assertEqual(diagnosis["next_action"], "wait")
def test_post_expiry_without_orphan_evidence_unlocks_acquire(self):
diagnosis = self._diagnose(_expired_superseded_comments())
self.assertEqual(
diagnosis["classification"], "orphaned_expired_superseded_head"
)
self.assertEqual(diagnosis["next_action"], "acquire")
def test_post_expiry_with_orphan_evidence_offers_guarded_cleanup(self):
diagnosis = self._diagnose(
_expired_superseded_comments(),
owner_process_alive=False,
worktree_clean=True,
worktree_exists=True,
)
self.assertEqual(
diagnosis["classification"], "orphaned_expired_superseded_head"
)
self.assertEqual(
diagnosis["next_action"], "cleanup_obsolete_reviewer_comment_lease"
)
def test_expired_lease_no_longer_blocks_fresh_acquisition(self):
assessment = leases.assess_acquire_lease(
_expired_superseded_comments(),
pr_number=PR,
reviewer_identity="fresh-reviewer",
profile="prgs-reviewer",
session_id="99999-fresh",
repo=REPO,
issue_number=ISSUE,
worktree="branches/review-pr-701-fresh",
candidate_head=NEW_HEAD,
target_branch="master",
target_branch_sha="2" * 40,
)
self.assertTrue(assessment.get("acquire_allowed"))
def test_unparseable_expiry_fails_closed_in_cleanup(self):
comment = _lease_comment(
last_activity=_now() - timedelta(hours=3), expires_at=None
)
comment["body"] = comment["body"].replace(
"expires_at: ", "expires_at: not-a-timestamp"
)
result = leases.assess_obsolete_reviewer_comment_lease_cleanup(
[comment],
pr_number=PR,
current_head_sha=NEW_HEAD,
formal_reviews=[],
repo=REPO,
expected_repo=REPO,
requesting_session_id="fresh-controller",
controller_recovery_authorized=True,
worktree_exists=True,
worktree_clean=True,
owner_process_alive=False,
)
self.assertFalse(result["cleanup_allowed"])
if __name__ == "__main__":
unittest.main()
+6 -27
View File
@@ -108,24 +108,13 @@ class TestIssueCommentWorkspaceGuard(unittest.TestCase):
"gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
side_effect=self._git_state(valid_worktree),
):
try:
res = srv.gitea_create_issue_comment(
with self.assertRaises(RuntimeError) as ctx:
srv.gitea_create_issue_comment(
issue_number=557,
body="evidence comment",
remote="prgs",
)
except RuntimeError as exc:
self.assertIn("stable control checkout", str(exc))
else:
# #683 typed blocker at mutation entrypoint
self.assertFalse(res.get("success"))
self.assertFalse(res.get("performed"))
blob = " ".join(res.get("reasons") or [])
self.assertTrue(
"stable control checkout" in blob
or res.get("blocker_kind")
)
self.assertTrue(res.get("exact_next_action") or res.get("reasons"))
self.assertIn("stable control checkout", str(ctx.exception))
mock_api.assert_not_called()
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
@@ -195,24 +184,14 @@ class TestIssueCommentWorkspaceGuard(unittest.TestCase):
side_effect=self._subprocess(valid_worktree, outside_worktree),
):
with patch.dict(os.environ, self.AUTHOR_ENV, clear=True):
try:
res = srv.gitea_create_issue_comment(
with self.assertRaises(RuntimeError) as ctx:
srv.gitea_create_issue_comment(
issue_number=557,
body="evidence comment",
remote="prgs",
worktree_path=outside_worktree,
)
except RuntimeError as exc:
self.assertIn(
"does not belong to the target repository",
str(exc),
)
else:
self.assertFalse(res.get("success"))
blob = " ".join(res.get("reasons") or [])
self.assertIn(
"does not belong to the target repository", blob
)
self.assertIn("does not belong to the target repository", str(ctx.exception))
mock_api.assert_not_called()
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
+3 -7
View File
@@ -76,17 +76,13 @@ class TestPreflightReadSurvival(unittest.TestCase):
self.assertIn("task mismatch", str(ctx.exception))
def test_capability_consumed_after_mutation_gate(self):
# Use reconciler/close_pr so this purity-order test does not require a
# branches/ worktree (author create_issue would hit #274/#683 guards).
# Test isolation stays explicit; production author guards remain live
# under force-on (see tests/test_issue_683_workflow_scope_guards.py).
mcp_server.record_preflight_check("whoami")
mcp_server.record_preflight_check(
"capability", resolved_role="reconciler", resolved_task="close_pr"
"capability", resolved_role="author", resolved_task="create_issue"
)
mcp_server.verify_preflight_purity(task="close_pr")
mcp_server.verify_preflight_purity(task="create_issue")
with self.assertRaises(RuntimeError) as ctx:
mcp_server.verify_preflight_purity(task="close_pr")
mcp_server.verify_preflight_purity(task="create_issue")
self.assertIn("has not been resolved", str(ctx.exception))
def test_whoami_recovery_after_violation_clears_capability(self):
+3 -15
View File
@@ -86,21 +86,9 @@ class TestReconcilerCloseWorkspaceGuard(unittest.TestCase):
):
srv._preflight_resolved_role = "author"
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
try:
res = srv.gitea_create_issue(title="Test", body="body")
except RuntimeError as exc:
self.assertIn("stable control checkout", str(exc))
else:
# #683 typed blocker at mutation entrypoint
self.assertFalse(res.get("success"))
blob = " ".join(res.get("reasons") or []) + str(
res.get("blocker_kind") or ""
)
self.assertTrue(
"stable control checkout" in blob
or "missing_issue_worktree" in blob
or "control checkout" in blob.lower()
)
with self.assertRaises(RuntimeError) as ctx:
srv.gitea_create_issue(title="Test", body="body")
self.assertIn("stable control checkout", str(ctx.exception))
if __name__ == "__main__":
-626
View File
@@ -1,626 +0,0 @@
"""Workflow scope ownership and production-guard hardening (#683).
Implements fail-closed enforcement so sessions cannot:
* mutate source/tests on the root/control checkout (including temporary
diagnostic edits) without binding an issue-backed ``branches/`` worktree;
* continue out-of-scope source work while locked to a different issue;
* disable, skip, or conceal production root/branches/porcelain guards solely
because pytest/unittest is loaded.
This module is pure assessment + small durable ledger helpers. Callers gather
live facts (lock, branch, porcelain, worktree path) and pass them in. Existing
root_checkout_guard / author_mutation_worktree assessors remain authoritative;
this module composes typed blockers with exact recovery actions.
Do **not** reintroduce the rejected #681 / ``300a4ca`` patterns:
* early-return from workspace verification under ``_preflight_in_test_mode()``
* porcelain filtering that strips ``*.py`` lines under pytest
"""
from __future__ import annotations
import os
import re
import threading
from typing import Any
import author_mutation_worktree
from reviewer_worktree import parse_dirty_tracked_files
# ── force-on / test isolation ────────────────────────────────────────────────
# When set, production root/branches/scope guards MUST run even under pytest.
# Unit tests that only need preflight-order isolation leave this unset and
# use GITEA_TEST_PORCELAIN / fixtures; real-entrypoint proof sets this to "1".
FORCE_PRODUCTION_GUARDS_ENV = "GITEA_TEST_FORCE_PRODUCTION_GUARDS"
# Existing force signals also mean "exercise production dirtiness paths".
_FORCE_DIRTY_ENV = "GITEA_TEST_FORCE_DIRTY"
_FORCE_PORCELAIN_ENV = "GITEA_TEST_PORCELAIN"
# ── typed blocker kinds ──────────────────────────────────────────────────────
BLOCKER_ROOT_DIAGNOSTIC_EDIT = "root_diagnostic_edit"
BLOCKER_MISSING_ISSUE_SCOPE = "missing_issue_scope"
BLOCKER_OUT_OF_SCOPE_ISSUE = "out_of_scope_issue"
BLOCKER_MISSING_WORKTREE = "missing_issue_worktree"
BLOCKER_UNRECORDED_FAILURE = "unrecorded_workflow_failure"
BLOCKER_PRODUCTION_GUARD = "production_guard_violation"
BLOCKER_KINDS = frozenset(
{
BLOCKER_ROOT_DIAGNOSTIC_EDIT,
BLOCKER_MISSING_ISSUE_SCOPE,
BLOCKER_OUT_OF_SCOPE_ISSUE,
BLOCKER_MISSING_WORKTREE,
BLOCKER_UNRECORDED_FAILURE,
BLOCKER_PRODUCTION_GUARD,
}
)
_NEXT_ACTIONS: dict[str, str] = {
BLOCKER_ROOT_DIAGNOSTIC_EDIT: (
"Stop editing the control/root checkout. Preserve or discard root WIP "
"durably, restore root to clean master, lock or create the owning issue, "
"bind branches/issue-<N>-*, set GITEA_AUTHOR_WORKTREE to that worktree, "
"then re-run the mutation."
),
BLOCKER_MISSING_ISSUE_SCOPE: (
"Select or create the owning Gitea issue, claim/lock it "
"(gitea_mark_issue + gitea_lock_issue), bind branches/issue-<N>-* "
"from clean master, then re-run the mutation from that worktree."
),
BLOCKER_OUT_OF_SCOPE_ISSUE: (
"Stop. The active issue lock does not own this work. Release or finish "
"the current issue lease, then select/create and lock the correct "
"owning issue, bind its branches/issue-<N>-* worktree, and re-run."
),
BLOCKER_MISSING_WORKTREE: (
"Bind an issue-backed worktree under branches/ (scripts/worktree-start "
"or git worktree add branches/issue-<N>-*), set GITEA_AUTHOR_WORKTREE / "
"worktree_path to that path, keep the control checkout clean on master, "
"then re-run the mutation."
),
BLOCKER_UNRECORDED_FAILURE: (
"Record the workflow/tool failure durably first (issue comment or "
"workflow_scope_guard.record_workflow_failure), then continue only "
"inside the owning issue-backed worktree."
),
BLOCKER_PRODUCTION_GUARD: (
"Resolve the production guard violation: clean or isolate the control "
"checkout, bind the owning issue worktree under branches/, and re-run "
"with production guards active."
),
}
_ISSUE_IN_BRANCH_RE = re.compile(r"issue-(\d+)", re.IGNORECASE)
# In-process durable failure ledger (also written via optional sink callback).
_ledger_lock = threading.Lock()
_failure_ledger: list[dict[str, Any]] = []
class ProductionGuardError(RuntimeError):
"""Fail-closed production guard with typed blocker metadata (#683)."""
def __init__(
self,
message: str,
*,
blocker_kind: str,
exact_next_action: str | None = None,
reasons: list[str] | None = None,
details: dict[str, Any] | None = None,
) -> None:
super().__init__(message)
kind = (blocker_kind or "").strip()
if kind not in BLOCKER_KINDS:
kind = BLOCKER_PRODUCTION_GUARD
self.blocker_kind = kind
self.exact_next_action = (
(exact_next_action or "").strip() or _NEXT_ACTIONS[kind]
)
self.reasons = list(reasons or [message])
self.details = dict(details or {})
def production_guards_forced() -> bool:
"""True when the explicit #683 force-on flag requests production guards."""
return (os.environ.get(FORCE_PRODUCTION_GUARDS_ENV) or "").strip().lower() in {
"1",
"true",
"yes",
"on",
}
def purity_order_forced() -> bool:
"""True when tests force preflight-order dirtiness paths (legacy flags)."""
if os.environ.get(_FORCE_DIRTY_ENV):
return True
# GITEA_TEST_PORCELAIN present (even empty) means dirtiness paths are live.
if os.environ.get(_FORCE_PORCELAIN_ENV) is not None:
return True
return False
def production_guards_active(*, in_test_mode: bool) -> bool:
"""Whether production root/branches/scope guards must execute.
Production (non-test) always active. Under pytest, active when either the
explicit #683 force-on flag or legacy dirty/porcelain force signals are
set — never skip production enforcement solely because tests are running
when force-on is requested.
"""
if production_guards_forced() or purity_order_forced():
return True
return not bool(in_test_mode)
def extract_issue_number_from_branch(branch_name: str | None) -> int | None:
"""Return the first issue-N number embedded in a branch name, if any."""
text = (branch_name or "").strip()
if not text:
return None
match = _ISSUE_IN_BRANCH_RE.search(text)
if not match:
return None
try:
return int(match.group(1))
except ValueError:
return None
def is_source_or_test_path(path: str) -> bool:
"""True for tracked source/test paths that must not land as root WIP."""
p = (path or "").replace("\\", "/").lstrip("./")
if not p:
return False
if p.startswith("tests/") or "/tests/" in f"/{p}":
return True
if p.endswith((".py", ".pyi", ".toml", ".cfg", ".ini", ".sh")):
return True
if p in {"requirements.txt", "pyproject.toml", "setup.py", "setup.cfg"}:
return True
return False
def dirty_source_files(porcelain_status: str) -> list[str]:
"""Tracked dirty paths that count as source/test contamination."""
dirty = parse_dirty_tracked_files(porcelain_status or "")
return [p for p in dirty if is_source_or_test_path(p)]
def assess_issue_scope_ownership(
*,
locked_issue_number: int | None,
target_issue_number: int | None = None,
branch_name: str | None = None,
role_kind: str | None = None,
require_lock_for_author: bool = False,
) -> dict[str, Any]:
"""Fail closed when the session issue lock does not own the attempted work.
* Author sessions that require a lock fail when none is held.
* When a lock exists, the target issue (tool argument) and/or the issue
number embedded in the branch must match the locked issue.
* Reviewer/merger/reconciler roles are not issue-scope owners of author
implementation work and skip the author lock requirement.
"""
role = (role_kind or "").strip().lower()
locked = locked_issue_number
if isinstance(locked, str) and locked.isdigit():
locked = int(locked)
if locked is not None:
try:
locked = int(locked)
except (TypeError, ValueError):
locked = None
target = target_issue_number
if target is not None:
try:
target = int(target)
except (TypeError, ValueError):
target = None
branch_issue = extract_issue_number_from_branch(branch_name)
reasons: list[str] = []
blocker_kind: str | None = None
# Non-author roles do not take author issue locks for implementation.
if role in {"reviewer", "merger", "reconciler"}:
return _scope_ok(locked, target, branch_issue)
if require_lock_for_author and locked is None:
reasons.append(
"no owning issue lock is bound for this author session; "
"source/test mutation requires selecting or creating an owning issue first"
)
blocker_kind = BLOCKER_MISSING_ISSUE_SCOPE
if locked is not None and target is not None and locked != target:
reasons.append(
f"session is locked to issue #{locked} but mutation targets issue "
f"#{target}; out-of-scope until the owning issue is selected"
)
blocker_kind = BLOCKER_OUT_OF_SCOPE_ISSUE
if locked is not None and branch_issue is not None and locked != branch_issue:
reasons.append(
f"session is locked to issue #{locked} but workspace branch is for "
f"issue #{branch_issue}; bind the matching issue-backed worktree"
)
blocker_kind = BLOCKER_OUT_OF_SCOPE_ISSUE
if reasons:
kind = blocker_kind or BLOCKER_MISSING_ISSUE_SCOPE
return {
"proven": False,
"block": True,
"blocker_kind": kind,
"exact_next_action": _NEXT_ACTIONS[kind],
"reasons": reasons,
"locked_issue_number": locked,
"target_issue_number": target,
"branch_issue_number": branch_issue,
}
return _scope_ok(locked, target, branch_issue)
def assess_root_source_mutation(
*,
workspace_path: str,
canonical_repo_root: str,
porcelain_status: str,
current_branch: str | None = None,
locked_issue_number: int | None = None,
role_kind: str | None = None,
) -> dict[str, Any]:
"""Fail closed for diagnostic/source edits on the control/root checkout.
Allowed only when the active workspace is under ``branches/``. Dirty
tracked source/test files on the control checkout always block, including
temporary/diagnostic/test-only intent.
"""
role = (role_kind or "").strip().lower()
if role == "reconciler":
return {
"proven": True,
"block": False,
"blocker_kind": None,
"exact_next_action": "proceed",
"reasons": [],
"dirty_source_files": [],
}
root = os.path.realpath(canonical_repo_root or "")
workspace = os.path.realpath(workspace_path or root or ".")
under_branches = author_mutation_worktree.is_path_under_branches(workspace, root)
dirty_src = dirty_source_files(porcelain_status)
reasons: list[str] = []
blocker_kind: str | None = None
if not under_branches and workspace == root and dirty_src:
# Root workspace with source dirtiness is unattributed root WIP.
# (Clean-root author binding is enforced by branches-only #274.)
reasons.append(
"control/root checkout has tracked source or test edits "
f"(dirty files: {', '.join(dirty_src)}); diagnostic or temporary "
"edits on the root checkout are forbidden"
)
blocker_kind = BLOCKER_ROOT_DIAGNOSTIC_EDIT
if (
not under_branches
and workspace == root
and not dirty_src
and role == "author"
):
# Explicit missing-worktree signal for force-on author entrypoints.
reasons.append(
"author source/test mutation from the stable control checkout is "
"forbidden; bind an issue-backed worktree under branches/ first"
)
blocker_kind = BLOCKER_MISSING_WORKTREE
if reasons:
kind = blocker_kind or BLOCKER_ROOT_DIAGNOSTIC_EDIT
return {
"proven": False,
"block": True,
"blocker_kind": kind,
"exact_next_action": _NEXT_ACTIONS[kind],
"reasons": reasons,
"dirty_source_files": dirty_src,
"workspace_path": workspace,
"canonical_repo_root": root,
"under_branches": under_branches,
"locked_issue_number": locked_issue_number,
}
return {
"proven": True,
"block": False,
"blocker_kind": None,
"exact_next_action": "proceed",
"reasons": [],
"dirty_source_files": dirty_src,
"workspace_path": workspace,
"canonical_repo_root": root,
"under_branches": under_branches,
"locked_issue_number": locked_issue_number,
}
def assess_production_mutation_guards(
*,
workspace_path: str,
canonical_repo_root: str,
porcelain_status: str,
current_branch: str | None = None,
locked_issue_number: int | None = None,
target_issue_number: int | None = None,
role_kind: str | None = None,
require_author_lock: bool = False,
in_test_mode: bool = False,
) -> dict[str, Any]:
"""Compose root + scope production guards when they must be active (#683)."""
if not production_guards_active(in_test_mode=in_test_mode):
return {
"proven": True,
"block": False,
"blocker_kind": None,
"exact_next_action": "proceed",
"reasons": [],
"skipped": True,
"skip_reason": "production guards not active (test isolation without force-on)",
}
root_assess = assess_root_source_mutation(
workspace_path=workspace_path,
canonical_repo_root=canonical_repo_root,
porcelain_status=porcelain_status,
current_branch=current_branch,
locked_issue_number=locked_issue_number,
role_kind=role_kind,
)
if root_assess["block"]:
return {**root_assess, "skipped": False}
scope_assess = assess_issue_scope_ownership(
locked_issue_number=locked_issue_number,
target_issue_number=target_issue_number,
branch_name=current_branch,
role_kind=role_kind,
require_lock_for_author=require_author_lock,
)
if scope_assess["block"]:
return {**scope_assess, "skipped": False}
return {
"proven": True,
"block": False,
"blocker_kind": None,
"exact_next_action": "proceed",
"reasons": [],
"skipped": False,
"root": root_assess,
"scope": scope_assess,
}
def raise_if_blocked(assessment: dict[str, Any]) -> None:
"""Raise :class:`ProductionGuardError` when *assessment* blocks."""
if not assessment or not assessment.get("block"):
return
kind = assessment.get("blocker_kind") or BLOCKER_PRODUCTION_GUARD
reasons = list(assessment.get("reasons") or ["production guard violation"])
message = (
f"Workflow scope guard (#683) [{kind}]: {'; '.join(reasons)}. "
f"exact_next_action: {assessment.get('exact_next_action') or _NEXT_ACTIONS.get(kind, '')}"
)
raise ProductionGuardError(
message,
blocker_kind=kind,
exact_next_action=assessment.get("exact_next_action"),
reasons=reasons,
details={
k: v
for k, v in assessment.items()
if k
not in {
"proven",
"block",
"blocker_kind",
"exact_next_action",
"reasons",
}
},
)
def block_response(
assessment: dict[str, Any] | ProductionGuardError | None = None,
*,
blocker_kind: str | None = None,
reasons: list[str] | None = None,
exact_next_action: str | None = None,
**extra: Any,
) -> dict[str, Any]:
"""Structured fail-closed tool response with typed blocker fields."""
if isinstance(assessment, ProductionGuardError):
kind = assessment.blocker_kind
reason_list = list(assessment.reasons)
next_action = assessment.exact_next_action
extra = {**assessment.details, **extra}
elif isinstance(assessment, dict) and assessment.get("block"):
kind = assessment.get("blocker_kind") or BLOCKER_PRODUCTION_GUARD
reason_list = list(assessment.get("reasons") or [])
next_action = assessment.get("exact_next_action") or _NEXT_ACTIONS.get(
kind, _NEXT_ACTIONS[BLOCKER_PRODUCTION_GUARD]
)
else:
kind = (blocker_kind or BLOCKER_PRODUCTION_GUARD).strip()
if kind not in BLOCKER_KINDS:
kind = BLOCKER_PRODUCTION_GUARD
reason_list = list(reasons or ["production guard violation"])
next_action = exact_next_action or _NEXT_ACTIONS[kind]
if kind not in BLOCKER_KINDS:
kind = BLOCKER_PRODUCTION_GUARD
if not reason_list:
reason_list = ["production guard violation"]
next_action = (next_action or "").strip() or _NEXT_ACTIONS[kind]
out: dict[str, Any] = {
"success": False,
"performed": False,
"blocker_kind": kind,
"exact_next_action": next_action,
"reasons": reason_list,
}
for key, value in extra.items():
if key not in out and value is not None:
out[key] = value
return out
def format_production_guard_error(assessment: dict[str, Any]) -> str:
"""Single RuntimeError string carrying kind + exact next action."""
kind = assessment.get("blocker_kind") or BLOCKER_PRODUCTION_GUARD
reasons = "; ".join(assessment.get("reasons") or ["production guard violation"])
next_action = assessment.get("exact_next_action") or _NEXT_ACTIONS.get(
kind, _NEXT_ACTIONS[BLOCKER_PRODUCTION_GUARD]
)
return (
f"Workflow scope guard (#683) [{kind}]: {reasons}. "
f"exact_next_action: {next_action}"
)
# ── durable failure recording ────────────────────────────────────────────────
def record_workflow_failure(
*,
kind: str,
detail: str,
issue_number: int | None = None,
task: str | None = None,
sink: Any | None = None,
) -> dict[str, Any]:
"""Record a workflow/tool failure before source edits continue (#683 AC8).
*sink* may be a callable ``sink(record)`` (e.g. tests) or omitted for the
in-process ledger only. Returns the durable record.
"""
record = {
"kind": (kind or "workflow_failure").strip() or "workflow_failure",
"detail": (detail or "").strip(),
"issue_number": issue_number,
"task": task,
"pid": os.getpid(),
}
with _ledger_lock:
_failure_ledger.append(dict(record))
if callable(sink):
sink(record)
return record
def clear_workflow_failure_ledger() -> None:
"""Test helper: reset the in-process failure ledger."""
with _ledger_lock:
_failure_ledger.clear()
def workflow_failure_ledger() -> list[dict[str, Any]]:
"""Copy of durable in-process failure records."""
with _ledger_lock:
return [dict(r) for r in _failure_ledger]
def assess_durable_failure_recorded(
*,
require_record: bool,
pending_source_mutation: bool,
) -> dict[str, Any]:
"""Block source mutation when a workflow failure was not recorded first."""
if not require_record or not pending_source_mutation:
return {
"proven": True,
"block": False,
"blocker_kind": None,
"exact_next_action": "proceed",
"reasons": [],
}
with _ledger_lock:
has_record = bool(_failure_ledger)
if has_record:
return {
"proven": True,
"block": False,
"blocker_kind": None,
"exact_next_action": "proceed",
"reasons": [],
}
return {
"proven": False,
"block": True,
"blocker_kind": BLOCKER_UNRECORDED_FAILURE,
"exact_next_action": _NEXT_ACTIONS[BLOCKER_UNRECORDED_FAILURE],
"reasons": [
"workflow/tool failure triggered a need for source changes but no "
"durable failure record exists yet"
],
}
def porcelain_preserves_python_paths(porcelain_status: str) -> bool:
"""Regression helper: dirty ``*.py`` lines must remain visible (#683)."""
text = porcelain_status or ""
for line in text.splitlines():
stripped = line.strip()
if stripped.endswith(".py") or ".py " in stripped or stripped.endswith(".py"):
# Any py path present proves no silent strip of all *.py lines.
if " M " in f" {stripped}" or stripped[:1] in "MADRCTU" or len(line) >= 4:
return True
# Empty porcelain is fine; integrity means we did not strip when present.
return ".py" not in text
def assert_no_pytest_porcelain_filter(source_text: str) -> list[str]:
"""Static check: production reader must not strip ``*.py`` under pytest."""
findings: list[str] = []
lowered = source_text or ""
if "endswith(\".py\")" in lowered or "endswith('.py')" in lowered:
if "pytest" in lowered and "porcelain" in lowered.lower():
findings.append(
"production porcelain reader must not filter *.py under pytest "
"(rejected 300a4ca pattern)"
)
if "if \"pytest\" in sys.modules" in lowered and "porcelain" in lowered.lower():
if ".py" in lowered and ("join" in lowered or "endswith" in lowered):
findings.append(
"test-mode porcelain filtering of source files is forbidden (#683)"
)
return findings
def _scope_ok(
locked: int | None,
target: int | None,
branch_issue: int | None,
) -> dict[str, Any]:
return {
"proven": True,
"block": False,
"blocker_kind": None,
"exact_next_action": "proceed",
"reasons": [],
"locked_issue_number": locked,
"target_issue_number": target,
"branch_issue_number": branch_issue,
}