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

.env.example

@@ -39,27 +39,6 @@ GITEA_AUDIT_LOG=/path/to/gitea-mcp-audit.log
 # only — never the token value. Surfaced by gitea_get_profile.
 GITEA_TOKEN_SOURCE=GITEA_TOKEN
 
-# ── Optional self-hosted Sentry observability (#606) ────────────────────────
-# Emits runtime errors, fail-closed workflow blockers, lease/terminal-lock/
-# stale-runtime collisions, and watchdog cron check-ins to a SELF-HOSTED Sentry
-# (https://sentry.prgs.cc/) — never Sentry Cloud. Gitea stays the source of
-# truth; Sentry is observe-only. OFF by default: with MCP_SENTRY_ENABLED unset
-# or SENTRY_DSN empty, nothing is initialised and no events are sent.
-#
-# Master gate. Truthy = 1/true/yes/on. Both this AND SENTRY_DSN are required.
-MCP_SENTRY_ENABLED=0
-# DSN for the self-hosted project (create a `gitea-tools-mcp` project in
-# https://sentry.prgs.cc/ and copy its DSN). Never commit a real DSN.
-SENTRY_DSN=
-# Deployment environment tag (local/dev/prod). Defaults to "development".
-SENTRY_ENVIRONMENT=development
-# Optional release identifier (e.g. a git SHA or version string).
-SENTRY_RELEASE=
-# Performance-trace sample rate, 0.0–1.0 (clamped). Default 0.0 (traces off).
-MCP_SENTRY_TRACES_SAMPLE_RATE=0.0
-# Set to 1 to forward Python logs to Sentry as structured logs. Default off.
-MCP_SENTRY_ENABLE_LOGS=0
-
 # Optional canonical runtime-profile config (#19). Instead of the fields above,
 # point every LLM launcher at ONE JSON file of named profiles and select one.
 # Secrets are referenced (keychain id / env var name), never inlined. See

anti_stomp_preflight.py

@@ -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, []

docs/observability/sentry-integration.md

@@ -1,138 +0,0 @@
-# Self-hosted Sentry observability for the Gitea MCP server (#606)
-
-Optional, **off-by-default** instrumentation that reports MCP runtime errors,
-fail-closed workflow blockers, lease / terminal-lock / stale-runtime
-collisions, and recurring watchdog check-ins to a **self-hosted** Sentry at
-`https://sentry.prgs.cc/`.
-
-> **Gitea remains the source of truth.** Sentry is observe-only. It never
-> approves, merges, closes, or otherwise mutates Gitea workflow state, and it
-> never bypasses leases, #332, workflow roles, or the MCP gates. Sentry alerts
-> may only feed the *sanctioned* Gitea issue/comment path via the #612 incident
-> bridge — never a direct write.
-
-Implemented by [`sentry_observability.py`](../../sentry_observability.py).
-
----
-
-## 1. Create the Sentry project
-
-1. Sign in to the self-hosted Sentry at **`https://sentry.prgs.cc/`** (this is
-   **not** Sentry Cloud — do not use `*.ingest.sentry.io`).
-2. Create a new **Python** project named **`gitea-tools-mcp`**.
-3. Open **Settings → Projects → gitea-tools-mcp → Client Keys (DSN)** and copy
-   the DSN. It looks like `https://<publickey>@sentry.prgs.cc/<project-id>`.
-4. **Never commit the DSN.** It is a runtime secret supplied via env var only.
-
-## 2. Configure the environment
-
-All configuration is env-var driven (see [`.env.example`](../../.env.example)):
-
-| Variable | Purpose | Default |
-|----------|---------|---------|
-| `MCP_SENTRY_ENABLED` | Master gate (`1/true/yes/on`). Required. | off |
-| `SENTRY_DSN` | Self-hosted DSN. Required. | *(empty)* |
-| `SENTRY_ENVIRONMENT` | `local` / `dev` / `prod` tag. | `development` |
-| `SENTRY_RELEASE` | Release id (git SHA or version). | *(none)* |
-| `MCP_SENTRY_TRACES_SAMPLE_RATE` | Perf-trace sample rate `0.0–1.0` (clamped). | `0.0` |
-| `MCP_SENTRY_ENABLE_LOGS` | Forward Python logs as structured logs. | off |
-
-**The feature stays completely off unless `MCP_SENTRY_ENABLED` is truthy *and*
-`SENTRY_DSN` is non-empty.** With either missing, `init_sentry()` is a no-op,
-the SDK is never initialised, and no events are sent — existing tool behaviour
-and API-call patterns are unchanged.
-
-### Per-environment examples
-
-```bash
-# local (quiet: capture errors/blockers, no traces)
-export MCP_SENTRY_ENABLED=1
-export SENTRY_DSN="https://<key>@sentry.prgs.cc/<id>"
-export SENTRY_ENVIRONMENT=local
-
-# dev (light tracing + logs)
-export MCP_SENTRY_ENABLED=1
-export SENTRY_DSN="https://<key>@sentry.prgs.cc/<id>"
-export SENTRY_ENVIRONMENT=dev
-export MCP_SENTRY_TRACES_SAMPLE_RATE=0.2
-export MCP_SENTRY_ENABLE_LOGS=1
-
-# prod (errors/blockers + low-rate tracing, release-tagged)
-export MCP_SENTRY_ENABLED=1
-export SENTRY_DSN="https://<key>@sentry.prgs.cc/<id>"
-export SENTRY_ENVIRONMENT=prod
-export SENTRY_RELEASE="$(git rev-parse --short HEAD)"
-export MCP_SENTRY_TRACES_SAMPLE_RATE=0.05
-```
-
-The optional SDK is pinned in [`requirements.txt`](../../requirements.txt)
-(`sentry-sdk==2.20.0`). It is imported lazily: if the package is absent, the
-module still imports and every entry point is a safe no-op.
-
-## 3. What is instrumented
-
-| Signal | Where | Notes |
-|--------|-------|-------|
-| Startup init | `gitea_mcp_server.py` `__main__`, before `mcp.run` | Prints a redaction-safe status line to stderr. |
-| Failing mutations (exceptions) | `_audited(...)` context manager | `capture_exception` with scrubbed tags. |
-| Fail-closed blockers / failed mutations | `_audit_pr_result(...)` (BLOCKED/FAILED) | Structured `capture_workflow_blocker` event incl. the canonical next action when available (criterion 7). |
-| Allocator watchdog check-ins | `gitea_allocate_next_work` tool | `allocator_health`, `stale_lease_scan`, `terminal_lock_scan`. |
-| Namespace-health check-in | `gitea_assess_mcp_namespace_health` tool | `namespace_health`. |
-
-All capture paths are **best-effort / fail open**: a Sentry outage or capture
-error never breaks an MCP tool success path.
-
-## 4. Cron / watchdog monitors
-
-`sentry_observability.MONITOR_SLUGS` defines stable check-in slugs:
-
-| Registry key | Sentry monitor slug | Wired at |
-|--------------|--------------------|----------|
-| `stale_lease_scan` | `gitea-mcp-stale-lease-scan` | allocator run (global lease expiry) |
-| `terminal_lock_scan` | `gitea-mcp-terminal-lock-scan` | allocator run (terminal-lock lookup) |
-| `allocator_health` | `gitea-mcp-allocator-health` | allocator run |
-| `namespace_health` | `gitea-mcp-namespace-health` | namespace-health probe |
-| `dashboard_freshness` | `gitea-mcp-dashboard-freshness` | call `monitor_checkin("dashboard_freshness", ...)` from the dashboard refresh job (#605) |
-| `reconciler_cleanup` | `gitea-mcp-reconciler-cleanup` | call `monitor_checkin("reconciler_cleanup", ...)` from the reconciler cleanup entrypoint |
-
-Create matching Cron monitors in Sentry with those slugs. Emit an
-`in_progress` check-in at job start and `ok`/`error` at completion via
-`sentry_observability.monitor_checkin(slug_key, status)`.
-
-## 5. Redaction guarantees (fail closed)
-
-Redaction fails *closed*: if a field cannot be proven safe it is dropped rather
-than sent. The `before_send` (and `before_send_log`) hook `scrub_event`
-recursively redacts every outgoing event; on any error it drops the event
-entirely. Guarantees, proven by `tests/test_sentry_observability.py`:
-
-- **No** tokens, passwords, keychain IDs, DSNs, cookies, or `user:pass@host`.
-- **No** raw session-state or full prompt/comment bodies — `session_id` is only
-  ever surfaced as a 12-char `session_id_hash`.
-- **No** private config contents or raw credential headers.
-- **No** full local filesystem paths — a worktree path collapses to a coarse
-  `worktree_category` (`author` / `reviewer` / `merger` / `reconciler` /
-  `branches` / `root` / `other`).
-- Only the allowlisted tag keys in `ALLOWED_TAG_KEYS` are ever attached.
-
-## 6. Coexistence with GlitchTip / the #612 incident bridge
-
-This is the **outbound** path (MCP → Sentry SDK). It complements — it does not
-replace — the **inbound** [`incident_bridge.py`](../../incident_bridge.py)
-(#612), which turns Sentry/GlitchTip *observations* into durable Gitea issues
-and `incident_links` rows.
-
-- Prefer **one** observability path per environment. Point the MCP server's
-  `SENTRY_DSN` at the same self-hosted `gitea-tools-mcp` project that the #612
-  bridge reconciles from, so an MCP-reported error and its Gitea issue line up.
-- GlitchTip is Sentry-protocol compatible; if an existing GlitchTip DSN is in
-  use, either migrate it to `https://sentry.prgs.cc/` or document the split
-  (MCP → Sentry, legacy → GlitchTip) explicitly for operators.
-- The bridge remains the **only** sanctioned route from an alert back into
-  Gitea workflow state.
-
-## 7. Non-goals
-
-- Sentry must **not** become the workflow source of truth.
-- Sentry must **not** approve, merge, close, or mutate Gitea workflow state.
-- Sentry must **not** bypass leases, #332, workflow roles, or the MCP gates.

gitea_mcp_server.py

@@ -619,18 +619,219 @@ def _enforce_branches_only_author_mutation(worktree_path: str | None = None) ->
         )
 
 
+def _anti_stomp_in_test_mode() -> bool:
+    """Whether the #604 anti-stomp gate is skipped under pytest.
+
+    Mirrors remote-repo / stable-contamination test bypasses so the existing
+    suite (bare remotes, mocked APIs) stays green. Set
+    ``GITEA_TEST_FORCE_ANTI_STOMP=1`` to exercise the live gate under tests.
+    """
+    if not _preflight_in_test_mode():
+        return False
+    return not bool(os.environ.get("GITEA_TEST_FORCE_ANTI_STOMP"))
+
+
+def _run_anti_stomp_preflight(
+    task: str | None,
+    *,
+    remote: str | None = None,
+    worktree_path: str | None = None,
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    expected_head_sha: str | None = None,
+    live_head_sha: str | None = None,
+    require_head_sha: bool = False,
+    lease_required: bool = False,
+    foreign_lease: bool | None = None,
+    lease_reasons: list[str] | None = None,
+    terminal_lock_blocks: bool | None = None,
+    terminal_lock_reasons: list[str] | None = None,
+    workflow_hash_valid: bool | None = None,
+    workflow_hash_reasons: list[str] | None = None,
+    raise_on_block: bool = True,
+) -> dict | None:
+    """Shared #604 anti-stomp preflight for MCP mutation entrypoints.
+
+    Gathers live workspace/parity/role/repo facts and invokes the pure
+    :func:`anti_stomp_preflight.assess_anti_stomp_preflight` assessor. On
+    block, raises ``RuntimeError`` (default) or returns a structured
+    :func:`anti_stomp_preflight.block_response` when *raise_on_block* is
+    False. Returns ``None`` when the mutation may proceed.
+    """
+    if _anti_stomp_in_test_mode():
+        return None
+
+    # Only enforce when the caller names a known mutation task. Read-only
+    # paths and legacy verify_preflight_purity calls without a task stay
+    # on the pre-#604 gate chain (whoami/capability/root/worktree).
+    if not task or not anti_stomp_preflight.is_mutation_task(task):
+        return None
+
+    profile = get_profile()
+    profile_name = profile.get("profile_name")
+    profile_role = _actual_profile_role()
+    allowed_operations = list(profile.get("allowed_operations") or [])
+    req_role = None
+    req_perm = None
+    if task:
+        try:
+            req_role = task_capability_map.required_role(task)
+        except KeyError:
+            req_role = _preflight_resolved_role
+        try:
+            req_perm = task_capability_map.required_permission(task)
+        except KeyError:
+            req_perm = None
+
+    ctx = _resolve_namespace_mutation_context(worktree_path)
+    workspace = ctx["workspace_path"]
+    canonical_root = ctx["canonical_repo_root"]
+    git_state = issue_lock_worktree.read_worktree_git_state(canonical_root)
+    remote_master_sha = root_checkout_guard.resolve_remote_master_sha(canonical_root)
+
+    # Repo/org facts (best-effort; explicit org/repo when provided).
+    resolved_org = org
+    resolved_repo = repo
+    local_remote_url = None
+    org_explicit = org is not None
+    repo_explicit = repo is not None
+    if remote:
+        try:
+            local_remote_url = _local_git_remote_url(remote)
+        except Exception:
+            local_remote_url = None
+        if resolved_org is None or resolved_repo is None:
+            try:
+                rem = _effective_remote(remote)
+                if rem in REMOTES:
+                    if resolved_org is None:
+                        resolved_org = REMOTES[rem]["org"]
+                    if resolved_repo is None:
+                        resolved_repo = REMOTES[rem]["repo"]
+            except Exception:
+                pass
+
+    parity = _current_master_parity()
+    startup_head = parity.get("startup_head")
+    current_code_head = parity.get("current_head")
+
+    # Source contamination from durable #671 marker (role-aware).
+    source_contaminated = None
+    contamination_reasons = None
+    try:
+        marker = _load_stable_contamination_marker(remote)
+        contaminated, cont_reasons = anti_stomp_preflight.contamination_from_stable_marker(
+            marker,
+            task=task,
+            actual_role=profile_role,
+        )
+        if marker is not None:
+            source_contaminated = contaminated
+            contamination_reasons = cont_reasons
+    except Exception:
+        # Fail soft on marker I/O: existing #671 enforcer still runs separately.
+        source_contaminated = None
+
+    # Workflow-hash facts when the task is review/merge oriented.
+    if workflow_hash_valid is None and task in {
+        "review_pr",
+        "submit_pr_review",
+        "approve_pr",
+        "request_changes_pr",
+        "merge_pr",
+    }:
+        try:
+            wf_reasons = _review_workflow_load_gate_reasons()
+            workflow_hash_valid = not bool(wf_reasons)
+            workflow_hash_reasons = list(wf_reasons) if wf_reasons else None
+        except Exception:
+            pass
+
+    # Mirror remote_repo_guard's pytest bypass: under the unit suite bare
+    # remote=prgs still defaults to Timesheet, and re-checking here would
+    # break happy-path reconciler/author tests that already rely on the
+    # #530 gate being opt-in via GITEA_FORCE_REMOTE_REPO_CHECK.
+    check_repo = True
+    if "pytest" in sys.modules and not os.environ.get(
+        "GITEA_FORCE_REMOTE_REPO_CHECK"
+    ):
+        check_repo = False
+
+    assessment = anti_stomp_preflight.assess_anti_stomp_preflight(
+        task=task,
+        remote=remote,
+        resolved_org=resolved_org,
+        resolved_repo=resolved_repo,
+        local_remote_url=local_remote_url,
+        org_explicit=org_explicit,
+        repo_explicit=repo_explicit,
+        check_repo=check_repo,
+        profile_name=profile_name,
+        profile_role=profile_role,
+        required_role=req_role or _preflight_resolved_role,
+        required_permission=req_perm,
+        allowed_operations=allowed_operations,
+        workspace_path=workspace,
+        project_root=canonical_root,
+        current_branch=git_state.get("current_branch"),
+        root_head_sha=git_state.get("head_sha"),
+        root_porcelain=git_state.get("porcelain_status") or "",
+        remote_master_sha=remote_master_sha,
+        startup_head=startup_head,
+        current_code_head=current_code_head,
+        lease_required=lease_required,
+        foreign_lease=foreign_lease,
+        lease_reasons=lease_reasons,
+        terminal_lock_blocks=terminal_lock_blocks,
+        terminal_lock_reasons=terminal_lock_reasons,
+        require_head_sha=require_head_sha,
+        expected_head_sha=expected_head_sha,
+        live_head_sha=live_head_sha,
+        workflow_hash_valid=workflow_hash_valid,
+        workflow_hash_reasons=workflow_hash_reasons,
+        source_contaminated=source_contaminated,
+        contamination_reasons=contamination_reasons,
+        manual_bypass_attempted=False,
+    )
+    if not assessment.get("block"):
+        return None
+    if raise_on_block:
+        raise RuntimeError(anti_stomp_preflight.format_anti_stomp_error(assessment))
+    return anti_stomp_preflight.block_response(assessment)
+
+
 def verify_preflight_purity(
     remote: str | None = None,
     worktree_path: str | None = None,
     task: str | None = None,
+    *,
+    expected_head_sha: str | None = None,
+    live_head_sha: str | None = None,
+    require_head_sha: bool = False,
+    lease_required: bool = False,
+    foreign_lease: bool | None = None,
+    lease_reasons: list[str] | None = None,
+    terminal_lock_blocks: bool | None = None,
+    terminal_lock_reasons: list[str] | None = None,
+    workflow_hash_valid: bool | None = None,
+    workflow_hash_reasons: list[str] | None = None,
+    org: str | None = None,
+    repo: str | None = None,
 ):
-    """Verify that identity and capability were verified prior to session edits."""
+    """Verify that identity and capability were verified prior to session edits.
+
+    Also runs the shared #604 anti-stomp preflight for mutation tasks (typed
+    blocker + exact next action). Extended lease/head/terminal facts may be
+    supplied by review/merge callers.
+    """
     global _preflight_reviewer_violation_files
 
     in_test = _preflight_in_test_mode()
     if in_test and not (
         os.environ.get("GITEA_TEST_FORCE_DIRTY")
         or os.environ.get("GITEA_TEST_PORCELAIN") is not None
+        or os.environ.get("GITEA_TEST_FORCE_ANTI_STOMP")
     ):
         return
 
@@ -717,6 +918,30 @@ def verify_preflight_purity(
 
     _enforce_root_checkout_guard(worktree_path)
     _enforce_branches_only_author_mutation(worktree_path)
+
+    # #604: common anti-stomp preflight (typed blocker + exact next action).
+    # Runs after the legacy enforcers so existing happy paths stay green and
+    # the shared module remains the single fail-closed composition point for
+    # repo/role/worktree/lease/terminal-lock/head/stale/workflow/contamination.
+    _run_anti_stomp_preflight(
+        task,
+        remote=remote,
+        worktree_path=worktree_path,
+        org=org,
+        repo=repo,
+        expected_head_sha=expected_head_sha,
+        live_head_sha=live_head_sha,
+        require_head_sha=require_head_sha,
+        lease_required=lease_required,
+        foreign_lease=foreign_lease,
+        lease_reasons=lease_reasons,
+        terminal_lock_blocks=terminal_lock_blocks,
+        terminal_lock_reasons=terminal_lock_reasons,
+        workflow_hash_valid=workflow_hash_valid,
+        workflow_hash_reasons=workflow_hash_reasons,
+        raise_on_block=True,
+    )
+
     _clear_preflight_capability_state()
 
 
@@ -918,7 +1143,6 @@ import allocator_service  # noqa: E402
 import control_plane_db  # noqa: E402
 import lease_lifecycle  # noqa: E402
 import incident_bridge  # noqa: E402
-import sentry_observability  # noqa: E402  (#606 optional Sentry observability)
 import agent_temp_artifacts
 import issue_lock_worktree  # noqa: E402
 import issue_lock_provenance  # noqa: E402
@@ -931,6 +1155,7 @@ import author_mutation_worktree  # noqa: E402
 import root_checkout_guard  # noqa: E402
 import stable_branch_push_guard  # noqa: E402
 import remote_repo_guard  # noqa: E402
+import anti_stomp_preflight  # noqa: E402
 import issue_claim_heartbeat  # noqa: E402
 import issue_work_duplicate_gate  # noqa: E402
 import issue_workflow_labels  # noqa: E402
@@ -1778,18 +2003,6 @@ def _audited(action: str, *, host, remote, org=None, repo=None,
                result=gitea_audit.FAILED, reason=_redact(str(exc)),
                request_metadata=request_metadata, issue_number=issue_number,
                pr_number=pr_number, target_branch=target_branch)
-        # #606: best-effort Sentry capture of the failing mutation (fail open).
-        sentry_observability.capture_exception(
-            exc,
-            tags={
-                "mutation_tool": action,
-                "remote": remote,
-                "repo": repo,
-                "org": org,
-                "issue_number": issue_number,
-                "pr_number": pr_number,
-            },
-        )
         raise
     _audit(action, host=host, remote=remote, org=org, repo=repo,
            result=gitea_audit.SUCCEEDED, request_metadata=request_metadata,
@@ -1836,20 +2049,6 @@ def _audit_pr_result(action: str):
                             "merge_method": result.get("merge_method"),
                         },
                     )
-                    # #606: surface fail-closed blockers / failed mutations to
-                    # Sentry as structured events (best-effort, fail open).
-                    if status in (gitea_audit.BLOCKED, gitea_audit.FAILED):
-                        sentry_observability.capture_workflow_blocker(
-                            action,
-                            message="; ".join(reasons) or action,
-                            next_action=result.get("safe_next_action"),
-                            level="error" if status == gitea_audit.FAILED else "warning",
-                            tags={
-                                "mutation_tool": action,
-                                "pr_number": result.get("pr_number"),
-                                "current_head_sha": result.get("head_sha"),
-                            },
-                        )
             except Exception:
                 pass  # best-effort; never break the tool
             return result
@@ -3860,6 +4059,30 @@ def _evaluate_pr_review_submission(
         result["pr_work_lease"] = lease_block
         return result
 
+    if live:
+        # #604: common anti-stomp with live head + lease proof (typed blocker).
+        anti_task = {
+            "approve": "approve_pr",
+            "request_changes": "request_changes_pr",
+            "comment": "submit_pr_review",
+        }.get(action, "submit_pr_review")
+        try:
+            _run_anti_stomp_preflight(
+                anti_task,
+                remote=remote,
+                worktree_path=worktree_path,
+                org=org,
+                repo=repo,
+                expected_head_sha=pinned_sha,
+                live_head_sha=actual_sha,
+                require_head_sha=True,
+                foreign_lease=False,
+                terminal_lock_blocks=False,
+            )
+        except RuntimeError as exc:
+            reasons.append(str(exc))
+            return result
+
     if (body or "").strip():
         gate = _canonical_comment_gate(body, context="pr_review")
         if gate["blocked"]:
@@ -4814,7 +5037,12 @@ def gitea_edit_pr(
         raise ValueError("At least one field to edit (title, body, state, base) must be provided.")
 
     closing = payload.get("state") == "closed"
-    verify_preflight_purity(remote, task="close_pr" if closing else None)
+    # Closing uses close_pr; non-closing title/body/base edits use edit_pr so
+    # the shared #604 anti-stomp inventory stays consistent with wiring.
+    if closing:
+        verify_preflight_purity(remote, task="close_pr")
+    else:
+        verify_preflight_purity(remote, task="edit_pr")
 
     # PR closure is a first-class capability, distinct from retitling or
     # rebasing edits (#216). Gate BEFORE auth/API setup so a blocked close
@@ -5296,6 +5524,25 @@ def gitea_merge_pr(
         reasons.extend(lease_block.get("reasons") or [])
         result["pr_work_lease"] = lease_block
         return result
+
+    # #604: common anti-stomp with live head + lease proof (typed blocker).
+    try:
+        _run_anti_stomp_preflight(
+            "merge_pr",
+            remote=remote,
+            worktree_path=worktree_path,
+            org=org,
+            repo=repo,
+            expected_head_sha=expected_head_sha,
+            live_head_sha=actual_sha,
+            require_head_sha=True,
+            foreign_lease=False,
+            terminal_lock_blocks=False,
+        )
+    except RuntimeError as exc:
+        reasons.append(str(exc))
+        return result
+
     if expected_head_sha and actual_sha and expected_head_sha != actual_sha:
         reasons.append(
             "expected head SHA does not match current PR head (fail closed)"
@@ -7415,7 +7662,11 @@ def gitea_acquire_reviewer_pr_lease(
             "permission_report": _permission_block_report("gitea.pr.comment"),
         }
 
-    _verify_role_mutation_workspace(remote, worktree=worktree, task="review_pr")
+    # task=acquire_reviewer_pr_lease so verify_preflight_purity runs shared #604
+    # anti-stomp for the declared lease-acquire mutation inventory entry.
+    _verify_role_mutation_workspace(
+        remote, worktree=worktree, task="acquire_reviewer_pr_lease"
+    )
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     profile = get_profile()
@@ -7556,6 +7807,7 @@ def gitea_adopt_merger_pr_lease(
             "permission_report": _permission_block_report("gitea.pr.merge"),
         }
 
+    # task=adopt_merger_pr_lease so verify_preflight_purity runs shared #604 anti-stomp.
     _verify_role_mutation_workspace(
         remote, worktree=worktree, task="adopt_merger_pr_lease"
     )
@@ -9981,11 +10233,6 @@ def gitea_assess_mcp_namespace_health(
         probe_source=probe_source,
     )
     _record_live_namespace_health(result)
-    # #606: namespace-health watchdog check-in (best-effort, fail open).
-    sentry_observability.monitor_checkin(
-        "namespace_health",
-        "ok" if result.get("healthy", result.get("callable", True)) else "error",
-    )
     return result
 
 
@@ -11913,18 +12160,6 @@ def gitea_allocate_next_work(
     result["inventory_source"] = (
         "candidates_json" if candidates_json else "gitea_live"
     )
-    # #606: watchdog check-ins for the recurring jobs this allocator run
-    # performs — global stale-lease expiry, terminal-lock lookup, and the
-    # allocator itself. Best-effort; a failed selection reports "error".
-    _alloc_ok = bool(result.get("success"))
-    sentry_observability.monitor_checkin(
-        "allocator_health", "ok" if _alloc_ok else "error"
-    )
-    if _alloc_ok:
-        # These two scans complete inside allocate_next_work before selection;
-        # a successful result proves both ran.
-        sentry_observability.monitor_checkin("stale_lease_scan", "ok")
-        sentry_observability.monitor_checkin("terminal_lock_scan", "ok")
     return result
 
 
@@ -12249,10 +12484,4 @@ if __name__ == "__main__":
     # processes (e.g. review_pr.py) can detect and refuse profile
     # side-channel overrides (#199).
     _export_session_profile_lock()
-    # #606: optional self-hosted Sentry observability. No-op unless
-    # MCP_SENTRY_ENABLED is truthy and SENTRY_DSN is set; never blocks startup.
-    _sentry_status = sentry_observability.init_sentry()
-    sys.stderr.write(
-        f"--- Sentry observability: {_sentry_status.get('reason')} ---\n"
-    )
     mcp.run(transport="stdio")

requirements.txt

@@ -31,7 +31,6 @@ python-multipart==0.0.32
 referencing==0.37.0
 rich==15.0.0
 rpds-py==2026.5.1
-sentry-sdk==2.20.0
 shellingham==1.5.4
 sse-starlette==3.4.5
 starlette==1.3.1

sentry_observability.py

@@ -1,535 +0,0 @@
-"""Optional self-hosted Sentry observability for the Gitea MCP server (#606).
-
-Adds env-var-gated Sentry SDK instrumentation so runtime errors, fail-closed
-workflow blockers, lease/terminal-lock/stale-runtime collisions, and recurring
-watchdog check-ins are visible in a *self-hosted* Sentry at
-``https://sentry.prgs.cc/`` — never Sentry Cloud, and never as the workflow
-source of truth (Gitea stays canonical).
-
-Design constraints (mirror ``gitea_audit`` and the #612 incident bridge):
-
-- **Off by default.** With ``MCP_SENTRY_ENABLED`` false/unset *or* ``SENTRY_DSN``
-  empty, ``init_sentry`` is a no-op and no events are ever sent — existing tool
-  behaviour and API-call patterns are unchanged (acceptance criterion 1).
-- **Fail *open* for observability.** A Sentry outage, a missing ``sentry_sdk``
-  package, or any capture error must never break an MCP tool success path. Every
-  public entry point swallows its own exceptions.
-- **Fail *closed* for redaction.** If a field cannot be proven safe it is dropped
-  rather than sent. Tokens, passwords, keychain IDs, DSNs, private config, raw
-  session-state, full prompt bodies, and full filesystem paths never leave here.
-- **No hard dependency.** ``sentry_sdk`` is imported lazily; the module is fully
-  importable and testable without it installed.
-
-Sentry is observe-only: it must not approve, merge, close, or otherwise mutate
-Gitea workflow state, nor bypass leases, #332, or MCP gates. Alerts may only feed
-the sanctioned Gitea issue/comment path via the #612 incident bridge.
-"""
-
-from __future__ import annotations
-
-import hashlib
-import os
-import re
-from dataclasses import dataclass
-from typing import Any
-
-# Reuse the most comprehensive existing scrubber so redaction stays consistent
-# with the #612 incident bridge (tokens, DSNs, cookies, bearer/basic, keychain
-# ids, session ids, user:pass@host).
-from incident_bridge import redact_text as _redact_text
-
-# Second, complementary scrubber: catches bare ``token <value>`` /
-# ``Bearer <value>`` / ``Basic <value>`` prefixes and raw URLs that the
-# incident-bridge delimiter patterns miss.
-from gitea_audit import _redact_str as _redact_prefixes
-
-# ── Optional SDK (lazy, never a hard dependency) ────────────────────────────
-try:  # pragma: no cover - trivial import guard
-    import sentry_sdk  # type: ignore
-except Exception:  # pragma: no cover - absence is a supported state
-    sentry_sdk = None  # type: ignore
-
-
-# ── Env var names (single source of truth) ──────────────────────────────────
-ENV_ENABLED = "MCP_SENTRY_ENABLED"
-ENV_DSN = "SENTRY_DSN"
-ENV_ENVIRONMENT = "SENTRY_ENVIRONMENT"
-ENV_RELEASE = "SENTRY_RELEASE"
-ENV_TRACES_SAMPLE_RATE = "MCP_SENTRY_TRACES_SAMPLE_RATE"
-ENV_ENABLE_LOGS = "MCP_SENTRY_ENABLE_LOGS"
-
-_TRUTHY = frozenset({"1", "true", "yes", "on"})
-
-REDACTED = "[REDACTED]"
-REDACTED_PATH = "[REDACTED_PATH]"
-
-
-# ── Cron / watchdog monitor slugs (acceptance criterion 6) ──────────────────
-# Stable slugs for the recurring/watchdog jobs #606 wants check-ins for. The
-# slug is the durable monitor identity in Sentry; the wiring call sites pass one
-# of these keys (or an explicit slug) to ``monitor_checkin``.
-MONITOR_SLUGS: dict[str, str] = {
-    "stale_lease_scan": "gitea-mcp-stale-lease-scan",
-    "terminal_lock_scan": "gitea-mcp-terminal-lock-scan",
-    "allocator_health": "gitea-mcp-allocator-health",
-    "namespace_health": "gitea-mcp-namespace-health",
-    "dashboard_freshness": "gitea-mcp-dashboard-freshness",
-    "reconciler_cleanup": "gitea-mcp-reconciler-cleanup",
-}
-
-_CHECKIN_STATUSES = frozenset({"in_progress", "ok", "error"})
-
-
-# ── Tag allowlist (issue "Suggested Sentry tags/context") ───────────────────
-# Only these keys are ever attached as Sentry tags. Anything else is dropped so
-# a caller cannot accidentally leak a sensitive value through a tag.
-ALLOWED_TAG_KEYS = frozenset({
-    "role",
-    "profile",
-    "namespace",
-    "repo",
-    "org",
-    "issue_number",
-    "pr_number",
-    "blocker_type",
-    "workflow_hash",
-    "session_id_hash",  # hash only — never the raw session id
-    "pid",
-    "worktree_category",  # category, never the full sensitive path
-    "lease_comment_id",
-    "expected_head_sha",
-    "current_head_sha",
-    "terminal_lock_state",
-    "capability",
-    "mutation_tool",
-})
-
-# Absolute-path shapes that must never be sent verbatim (macOS/Linux + temp).
-_PATH_RE = re.compile(r"(?:/private)?/(?:Users|home|tmp|var|opt|Volumes)/[^\s\"']*")
-
-# ``extra`` keys whose *full* contents are forbidden by the redaction rules
-# (raw session-state, full prompt/comment bodies, private config blobs, raw
-# headers).
-_FORBIDDEN_EXTRA_KEYS = frozenset({
-    "prompt",
-    "prompt_body",
-    "next_prompt",
-    "body",
-    "raw_body",
-    "session_state",
-    "session_state_contents",
-    "config",
-    "config_contents",
-    "private_config",
-    "headers",
-    "authorization",
-})
-
-
-# ── Configuration ───────────────────────────────────────────────────────────
-@dataclass(frozen=True)
-class SentryConfig:
-    """Immutable snapshot of the Sentry env configuration."""
-
-    enabled: bool = False
-    dsn: str | None = None
-    environment: str = "development"
-    release: str | None = None
-    traces_sample_rate: float = 0.0
-    enable_logs: bool = False
-
-    @property
-    def active(self) -> bool:
-        """True only when the operator both opted in *and* supplied a DSN.
-
-        This is the single gate that keeps the feature off by default: enabling
-        the flag without a DSN (or vice versa) sends nothing.
-        """
-        return bool(self.enabled and self.dsn)
-
-    def safe_summary(self) -> dict[str, Any]:
-        """Operator-facing status with **no** DSN value (only presence)."""
-        return {
-            "enabled": self.enabled,
-            "dsn_present": bool(self.dsn),
-            "environment": self.environment,
-            "release": self.release,
-            "traces_sample_rate": self.traces_sample_rate,
-            "enable_logs": self.enable_logs,
-            "active": self.active,
-        }
-
-
-def _env_bool(name: str, env: dict[str, str]) -> bool:
-    return (env.get(name) or "").strip().lower() in _TRUTHY
-
-
-def _env_float(name: str, default: float, env: dict[str, str]) -> float:
-    raw = (env.get(name) or "").strip()
-    if not raw:
-        return default
-    try:
-        val = float(raw)
-    except (TypeError, ValueError):
-        return default
-    # Clamp to Sentry's valid [0.0, 1.0] sample-rate range.
-    if val < 0.0:
-        return 0.0
-    if val > 1.0:
-        return 1.0
-    return val
-
-
-def load_config(env: dict[str, str] | None = None) -> SentryConfig:
-    """Build a :class:`SentryConfig` from the environment (read at call time)."""
-    env = dict(os.environ if env is None else env)
-    dsn = (env.get(ENV_DSN) or "").strip() or None
-    return SentryConfig(
-        enabled=_env_bool(ENV_ENABLED, env),
-        dsn=dsn,
-        environment=(env.get(ENV_ENVIRONMENT) or "").strip() or "development",
-        release=(env.get(ENV_RELEASE) or "").strip() or None,
-        traces_sample_rate=_env_float(ENV_TRACES_SAMPLE_RATE, 0.0, env),
-        enable_logs=_env_bool(ENV_ENABLE_LOGS, env),
-    )
-
-
-def sdk_available() -> bool:
-    """True when the optional ``sentry_sdk`` package is importable."""
-    return sentry_sdk is not None
-
-
-# ── Redaction (fail closed) ─────────────────────────────────────────────────
-def sanitize_path(value: Any) -> str:
-    """Reduce a filesystem path to a non-sensitive *category* token.
-
-    Full local paths must never be sent. We keep only a coarse worktree
-    category derived from the path shape (author/reviewer/merger/reconciler/
-    branches/root/other).
-    """
-    text = "" if value is None else str(value)
-    low = text.lower()
-    if not text:
-        return "unknown"
-    # Order matters: more specific role markers before the generic "branches".
-    if "reconcile" in low:
-        return "reconciler"
-    if "review" in low:
-        return "reviewer"
-    if "merge" in low or "merger" in low:
-        return "merger"
-    if "author" in low or re.search(r"/branches/(?:feat|fix|docs|chore|issue)", low):
-        return "author"
-    if "/branches/" in low:
-        return "branches"
-    if low.rstrip("/").endswith("gitea-tools"):
-        return "root"
-    return "other"
-
-
-def redact_value(value: Any) -> Any:
-    """Recursively redact a JSON-able value: secret text, absolute paths, and
-    known-sensitive dict keys are removed. Fail closed — any error drops the
-    value entirely rather than risk leaking it."""
-    try:
-        if isinstance(value, dict):
-            out: dict[str, Any] = {}
-            for k, v in value.items():
-                key = str(k)
-                low = key.lower()
-                if low in _FORBIDDEN_EXTRA_KEYS or any(
-                    s in low
-                    for s in ("token", "secret", "password", "cookie", "auth", "dsn", "keychain")
-                ):
-                    out[key] = REDACTED
-                    continue
-                out[key] = redact_value(v)
-            return out
-        if isinstance(value, (list, tuple)):
-            return [redact_value(v) for v in value]
-        if isinstance(value, str):
-            scrubbed = _redact_text(value)
-            scrubbed = _redact_prefixes(scrubbed)
-            scrubbed = _PATH_RE.sub(REDACTED_PATH, scrubbed)
-            return scrubbed
-        return value
-    except Exception:
-        return REDACTED
-
-
-def hash_session_id(session_id: Any) -> str:
-    """Short, stable, non-reversible fingerprint of a session id."""
-    digest = hashlib.sha256(str(session_id).encode("utf-8", "replace")).hexdigest()
-    return digest[:12]
-
-
-def build_tags(**kwargs: Any) -> dict[str, str]:
-    """Return a scrubbed, allowlisted tag dict.
-
-    ``session_id`` is accepted but only ever surfaced as ``session_id_hash``.
-    ``worktree_path`` collapses to ``worktree_category``. Any non-allowlisted
-    key, or a value that still contains redacted material after scrubbing, is
-    dropped.
-    """
-    raw: dict[str, Any] = dict(kwargs)
-
-    # Hash the session id — never emit it raw.
-    session_id = raw.pop("session_id", None)
-    if session_id and "session_id_hash" not in raw:
-        raw["session_id_hash"] = hash_session_id(session_id)
-
-    # A full worktree path collapses to a category tag.
-    wt = raw.pop("worktree_path", None)
-    if wt and "worktree_category" not in raw:
-        raw["worktree_category"] = sanitize_path(wt)
-
-    out: dict[str, str] = {}
-    for key, val in raw.items():
-        if key not in ALLOWED_TAG_KEYS:
-            continue
-        if val is None:
-            continue
-        scrubbed = redact_value(val)
-        text = str(scrubbed)
-        if not text or REDACTED in text or REDACTED_PATH in text:
-            continue
-        if len(text) > 200:
-            text = text[:200] + "…"
-        out[key] = text
-    return out
-
-
-def scrub_event(event: Any, hint: Any = None) -> dict[str, Any] | None:
-    """Sentry ``before_send`` / ``before_send_log`` hook.
-
-    Recursively redacts the outgoing event. On *any* failure it returns ``None``
-    so the event is dropped rather than sent unscrubbed (fail closed for
-    redaction).
-    """
-    try:
-        if not isinstance(event, dict):
-            return None
-        scrubbed = redact_value(event)
-        # Drop server_name if it leaked a hostname/path; PID is kept via tags.
-        scrubbed.pop("server_name", None)
-        return scrubbed
-    except Exception:
-        return None
-
-
-# ── Event builders (pure, independently testable) ───────────────────────────
-def build_blocker_event(
-    blocker_type: str,
-    *,
-    message: str | None = None,
-    next_action: str | None = None,
-    level: str = "warning",
-    tags: dict[str, Any] | None = None,
-    extra: dict[str, Any] | None = None,
-) -> dict[str, Any]:
-    """Build a redacted, structured Sentry event for a workflow blocker.
-
-    ``next_action`` maps the issue's "canonical next action when available"
-    requirement (acceptance criterion 7).
-    """
-    merged_tags = dict(tags or {})
-    merged_tags.setdefault("blocker_type", blocker_type)
-    safe_tags = build_tags(**merged_tags)
-
-    safe_extra = redact_value(dict(extra or {}))
-    if next_action:
-        # A short canonical next action is allowed (it is not a full prompt).
-        safe_extra["canonical_next_action"] = redact_value(str(next_action)[:500])
-
-    event: dict[str, Any] = {
-        "message": redact_value(message or blocker_type),
-        "level": level if level in ("debug", "info", "warning", "error", "fatal") else "warning",
-        "logger": "gitea-mcp.workflow",
-        "tags": safe_tags,
-        "extra": safe_extra,
-        "fingerprint": ["workflow-blocker", blocker_type],
-    }
-    return event
-
-
-def build_checkin_payload(
-    monitor: str,
-    status: str,
-    *,
-    check_in_id: str | None = None,
-    duration: float | None = None,
-) -> dict[str, Any]:
-    """Build a Sentry cron check-in payload for one of :data:`MONITOR_SLUGS`.
-
-    ``monitor`` may be a registry key (e.g. ``"stale_lease_scan"``) or an
-    explicit slug. Raises ``ValueError`` on an unknown status so callers cannot
-    silently send a malformed check-in.
-    """
-    if status not in _CHECKIN_STATUSES:
-        raise ValueError(
-            f"invalid check-in status {status!r}; expected one of {sorted(_CHECKIN_STATUSES)}"
-        )
-    slug = MONITOR_SLUGS.get(monitor, monitor)
-    payload: dict[str, Any] = {"monitor_slug": slug, "status": status}
-    if check_in_id:
-        payload["check_in_id"] = str(check_in_id)
-    if duration is not None:
-        try:
-            payload["duration"] = float(duration)
-        except (TypeError, ValueError):
-            pass
-    return payload
-
-
-# ── Runtime init + capture (fail open) ──────────────────────────────────────
-_STATE: dict[str, Any] = {"initialized": False, "config": None}
-
-
-def is_initialized() -> bool:
-    return bool(_STATE.get("initialized"))
-
-
-def active_config() -> SentryConfig | None:
-    return _STATE.get("config")
-
-
-def reset_for_tests() -> None:
-    """Clear module init state. Test-only helper (never called in production)."""
-    _STATE["initialized"] = False
-    _STATE["config"] = None
-
-
-def init_sentry(config: SentryConfig | None = None) -> dict[str, Any]:
-    """Initialise the Sentry SDK if (and only if) enabled + DSN + SDK present.
-
-    Idempotent and never raises. Returns an operator-safe status dict (no DSN
-    value). Behaviour is unchanged when the feature is off.
-    """
-    cfg = config or load_config()
-    status: dict[str, Any] = {"initialized": False, **cfg.safe_summary()}
-    try:
-        if not cfg.active:
-            status["reason"] = "disabled (MCP_SENTRY_ENABLED false or SENTRY_DSN empty)"
-            _STATE["config"] = cfg
-            return status
-        if not sdk_available():
-            status["reason"] = "sentry_sdk not installed"
-            _STATE["config"] = cfg
-            return status
-
-        init_kwargs: dict[str, Any] = {
-            "dsn": cfg.dsn,
-            "environment": cfg.environment,
-            "release": cfg.release,
-            "traces_sample_rate": cfg.traces_sample_rate,
-            "before_send": scrub_event,
-            "send_default_pii": False,
-        }
-        if cfg.enable_logs:
-            # sentry-sdk 2.x captures Python logs as structured logs when the
-            # experimental logs feature is enabled; scrub those too.
-            init_kwargs["_experiments"] = {
-                "enable_logs": True,
-                "before_send_log": scrub_event,
-            }
-        sentry_sdk.init(**init_kwargs)  # type: ignore[union-attr]
-        _STATE["initialized"] = True
-        _STATE["config"] = cfg
-        status["initialized"] = True
-        status["reason"] = "sentry initialised"
-    except Exception as exc:  # fail open: observability must not block startup
-        status["reason"] = f"init failed (ignored): {type(exc).__name__}"
-        _STATE["initialized"] = False
-    return status
-
-
-def _set_scope_tags(scope: Any, tags: dict[str, str]) -> None:
-    for key, val in tags.items():
-        try:
-            scope.set_tag(key, val)
-        except Exception:
-            pass
-
-
-def capture_workflow_blocker(
-    blocker_type: str,
-    *,
-    message: str | None = None,
-    next_action: str | None = None,
-    level: str = "warning",
-    tags: dict[str, Any] | None = None,
-    extra: dict[str, Any] | None = None,
-) -> dict[str, Any]:
-    """Capture a fail-closed workflow blocker as a structured Sentry event.
-
-    Always returns the redacted event dict (so callers/tests can inspect it),
-    and sends it to Sentry only when initialised. Fail open.
-    """
-    event = build_blocker_event(
-        blocker_type,
-        message=message,
-        next_action=next_action,
-        level=level,
-        tags=tags,
-        extra=extra,
-    )
-    try:
-        if is_initialized() and sdk_available():
-            sentry_sdk.capture_event(event)  # type: ignore[union-attr]
-    except Exception:
-        pass
-    return event
-
-
-def capture_exception(
-    exc: BaseException,
-    *,
-    tags: dict[str, Any] | None = None,
-    extra: dict[str, Any] | None = None,
-) -> bool:
-    """Capture a runtime exception with scrubbed tags. Fail open.
-
-    Returns True only when the event was handed to an initialised SDK.
-    """
-    try:
-        if not (is_initialized() and sdk_available()):
-            return False
-        safe_tags = build_tags(**(tags or {}))
-        safe_extra = redact_value(dict(extra or {}))
-        with sentry_sdk.push_scope() as scope:  # type: ignore[union-attr]
-            _set_scope_tags(scope, safe_tags)
-            for key, val in safe_extra.items():
-                try:
-                    scope.set_extra(key, val)
-                except Exception:
-                    pass
-            sentry_sdk.capture_exception(exc)  # type: ignore[union-attr]
-        return True
-    except Exception:
-        return False
-
-
-def monitor_checkin(
-    monitor: str,
-    status: str,
-    *,
-    check_in_id: str | None = None,
-    duration: float | None = None,
-) -> dict[str, Any] | None:
-    """Send a Sentry cron check-in for a watchdog job. Fail open.
-
-    Returns the payload (for inspection/tests), or ``None`` if the status was
-    invalid. Only transmits when initialised.
-    """
-    try:
-        payload = build_checkin_payload(
-            monitor, status, check_in_id=check_in_id, duration=duration
-        )
-    except ValueError:
-        return None
-    try:
-        if is_initialized() and sdk_available() and hasattr(sentry_sdk, "capture_checkin"):
-            sentry_sdk.capture_checkin(**payload)  # type: ignore[union-attr]
-    except Exception:
-        pass
-    return payload

task_capability_map.py

@@ -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,10 +77,22 @@ 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",

tests/test_sentry_observability.py

@@ -1,412 +0,0 @@
-"""Tests for optional self-hosted Sentry observability (#606).
-
-Covers the pure module (config, redaction, event/check-in builders) and the
-runtime capture paths using a fake ``sentry_sdk``, so nothing ever touches the
-network. Critically proves the feature is a no-op when disabled or DSN-less
-(acceptance criterion 1) and that secrets/paths/session-state are never sent
-(criterion 5).
-"""
-
-import sys
-import contextlib
-from pathlib import Path
-
-import pytest
-
-sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
-
-import sentry_observability as so  # noqa: E402
-
-
-# ── Fake SDK ────────────────────────────────────────────────────────────────
-class _FakeScope:
-    def __init__(self):
-        self.tags = {}
-        self.extras = {}
-
-    def set_tag(self, k, v):
-        self.tags[k] = v
-
-    def set_extra(self, k, v):
-        self.extras[k] = v
-
-
-class FakeSentrySDK:
-    """Minimal stand-in exposing the SDK surface sentry_observability uses."""
-
-    def __init__(self):
-        self.init_kwargs = None
-        self.events = []
-        self.exceptions = []
-        self.checkins = []
-        self.last_scope = None
-
-    def init(self, **kwargs):
-        self.init_kwargs = kwargs
-
-    def capture_event(self, event):
-        self.events.append(event)
-
-    def capture_exception(self, exc):
-        self.exceptions.append(exc)
-
-    def capture_checkin(self, **payload):
-        self.checkins.append(payload)
-
-    @contextlib.contextmanager
-    def push_scope(self):
-        self.last_scope = _FakeScope()
-        yield self.last_scope
-
-
-@pytest.fixture(autouse=True)
-def _reset_state():
-    """Isolate module init state between tests."""
-    so.reset_for_tests()
-    yield
-    so.reset_for_tests()
-
-
-@pytest.fixture
-def fake_sdk(monkeypatch):
-    sdk = FakeSentrySDK()
-    monkeypatch.setattr(so, "sentry_sdk", sdk)
-    return sdk
-
-
-# ── Config / gating ─────────────────────────────────────────────────────────
-def test_disabled_by_default_empty_env():
-    cfg = so.load_config(env={})
-    assert cfg.enabled is False
-    assert cfg.active is False
-
-
-def test_enabled_flag_without_dsn_is_not_active():
-    cfg = so.load_config(env={"MCP_SENTRY_ENABLED": "1"})
-    assert cfg.enabled is True
-    assert cfg.dsn is None
-    assert cfg.active is False  # DSN required
-
-
-def test_dsn_without_enabled_flag_is_not_active():
-    cfg = so.load_config(env={"SENTRY_DSN": "https://[email protected]/1"})
-    assert cfg.active is False
-
-
-def test_active_requires_enabled_and_dsn():
-    cfg = so.load_config(
-        env={"MCP_SENTRY_ENABLED": "true", "SENTRY_DSN": "https://[email protected]/1"}
-    )
-    assert cfg.active is True
-
-
-def test_truthy_variants():
-    for val in ("1", "true", "YES", "On"):
-        cfg = so.load_config(env={"MCP_SENTRY_ENABLED": val})
-        assert cfg.enabled is True
-    for val in ("0", "false", "no", "", "off"):
-        cfg = so.load_config(env={"MCP_SENTRY_ENABLED": val})
-        assert cfg.enabled is False
-
-
-def test_traces_sample_rate_parsed_and_clamped():
-    assert so.load_config(env={"MCP_SENTRY_TRACES_SAMPLE_RATE": "0.25"}).traces_sample_rate == 0.25
-    assert so.load_config(env={"MCP_SENTRY_TRACES_SAMPLE_RATE": "5"}).traces_sample_rate == 1.0
-    assert so.load_config(env={"MCP_SENTRY_TRACES_SAMPLE_RATE": "-1"}).traces_sample_rate == 0.0
-    assert so.load_config(env={"MCP_SENTRY_TRACES_SAMPLE_RATE": "junk"}).traces_sample_rate == 0.0
-
-
-def test_safe_summary_has_no_dsn_value():
-    cfg = so.load_config(
-        env={"MCP_SENTRY_ENABLED": "1", "SENTRY_DSN": "https://[email protected]/1"}
-    )
-    summary = cfg.safe_summary()
-    assert summary["dsn_present"] is True
-    assert "secret" not in repr(summary)
-    assert "dsn" not in summary  # only presence, never the value
-
-
-# ── init_sentry ─────────────────────────────────────────────────────────────
-def test_init_noop_when_disabled(fake_sdk):
-    status = so.init_sentry(so.load_config(env={}))
-    assert status["initialized"] is False
-    assert fake_sdk.init_kwargs is None  # no SDK init
-    assert so.is_initialized() is False
-
-
-def test_init_noop_when_enabled_but_missing_dsn(fake_sdk):
-    status = so.init_sentry(so.load_config(env={"MCP_SENTRY_ENABLED": "1"}))
-    assert status["initialized"] is False
-    assert fake_sdk.init_kwargs is None
-
-
-def test_init_reports_missing_sdk(monkeypatch):
-    monkeypatch.setattr(so, "sentry_sdk", None)
-    status = so.init_sentry(
-        so.load_config(
-            env={"MCP_SENTRY_ENABLED": "1", "SENTRY_DSN": "https://[email protected]/1"}
-        )
-    )
-    assert status["initialized"] is False
-    assert "not installed" in status["reason"]
-
-
-def test_init_configures_sdk_with_scrubber(fake_sdk):
-    status = so.init_sentry(
-        so.load_config(
-            env={
-                "MCP_SENTRY_ENABLED": "1",
-                "SENTRY_DSN": "https://[email protected]/1",
-                "SENTRY_ENVIRONMENT": "prod",
-                "MCP_SENTRY_TRACES_SAMPLE_RATE": "0.1",
-            }
-        )
-    )
-    assert status["initialized"] is True
-    assert so.is_initialized() is True
-    kw = fake_sdk.init_kwargs
-    assert kw["dsn"] == "https://[email protected]/1"
-    assert kw["environment"] == "prod"
-    assert kw["traces_sample_rate"] == 0.1
-    assert kw["before_send"] is so.scrub_event
-    assert kw["send_default_pii"] is False
-
-
-def test_init_enable_logs_wires_log_scrubber(fake_sdk):
-    so.init_sentry(
-        so.load_config(
-            env={
-                "MCP_SENTRY_ENABLED": "1",
-                "SENTRY_DSN": "https://[email protected]/1",
-                "MCP_SENTRY_ENABLE_LOGS": "1",
-            }
-        )
-    )
-    exp = fake_sdk.init_kwargs["_experiments"]
-    assert exp["enable_logs"] is True
-    assert exp["before_send_log"] is so.scrub_event
-
-
-def test_init_never_raises_on_sdk_failure(monkeypatch):
-    class Boom:
-        def init(self, **kwargs):
-            raise RuntimeError("sentry down")
-
-    monkeypatch.setattr(so, "sentry_sdk", Boom())
-    status = so.init_sentry(
-        so.load_config(
-            env={"MCP_SENTRY_ENABLED": "1", "SENTRY_DSN": "https://[email protected]/1"}
-        )
-    )
-    assert status["initialized"] is False
-    assert "init failed" in status["reason"]
-
-
-# ── Redaction (fail closed) ─────────────────────────────────────────────────
-def test_build_tags_allowlist_only():
-    tags = so.build_tags(role="author", secret_thing="leak", pid=123)
-    assert tags["role"] == "author"
-    assert tags["pid"] == "123"
-    assert "secret_thing" not in tags
-
-
-def test_build_tags_hashes_session_id():
-    tags = so.build_tags(session_id="prgs-author-20479-cf9ac178")
-    assert "session_id" not in tags
-    assert "session_id_hash" in tags
-    assert tags["session_id_hash"] != "prgs-author-20479-cf9ac178"
-    assert len(tags["session_id_hash"]) == 12
-
-
-def test_build_tags_collapses_worktree_path():
-    tags = so.build_tags(
-        worktree_path="/Users/x/Development/Gitea-Tools/branches/issue-606-sentry-observability"
-    )
-    assert "worktree_path" not in tags
-    assert tags["worktree_category"] == "author"
-
-
-def test_build_tags_drops_value_that_scrubs_to_redacted():
-    # A tag value that is itself a token gets scrubbed then dropped.
-    tags = so.build_tags(capability="token abcdef1234567890")
-    assert "capability" not in tags
-
-
-def test_sanitize_path_categories():
-    assert so.sanitize_path("/repo/branches/review-pr-654") == "reviewer"
-    assert so.sanitize_path("/repo/branches/merge-pr-1") == "merger"
-    assert so.sanitize_path("/repo/branches/reconcile-pr-1") == "reconciler"
-    assert so.sanitize_path("/repo/branches/feat-issue-606") == "author"
-    assert so.sanitize_path("/x/y/Gitea-Tools") == "root"
-
-
-def test_redact_value_scrubs_secrets_and_paths():
-    out = so.redact_value(
-        {
-            "token": "abc123",
-            "note": "Authorization: Bearer sk_live_abcdefgh12345",
-            "path": "/Users/jasonwalker/Development/Gitea-Tools/secret",
-            "dsn": "https://[email protected]/1",
-            "safe": "hello",
-        }
-    )
-    assert out["token"] == so.REDACTED
-    assert out["dsn"] == so.REDACTED
-    assert "sk_live" not in out["note"]
-    assert so.REDACTED_PATH in out["path"]
-    assert "/Users/" not in out["path"]
-    assert out["safe"] == "hello"
-
-
-def test_redact_value_forbidden_prompt_and_session_state():
-    out = so.redact_value(
-        {"prompt": "full body", "session_state": "{...}", "keep": "ok"}
-    )
-    assert out["prompt"] == so.REDACTED
-    assert out["session_state"] == so.REDACTED
-    assert out["keep"] == "ok"
-
-
-def test_scrub_event_redacts_nested_and_drops_server_name():
-    event = {
-        "server_name": "some-host",
-        "message": "boom",
-        "extra": {"token": "leak", "ok": "1"},
-    }
-    scrubbed = so.scrub_event(event)
-    assert "server_name" not in scrubbed
-    assert scrubbed["extra"]["token"] == so.REDACTED
-    assert scrubbed["extra"]["ok"] == "1"
-
-
-def test_scrub_event_drops_non_dict():
-    assert so.scrub_event("not a dict") is None
-    assert so.scrub_event(None) is None
-
-
-# ── Event builders ──────────────────────────────────────────────────────────
-def test_build_blocker_event_structure_and_next_action():
-    event = so.build_blocker_event(
-        "active_foreign_lease",
-        message="blocked by foreign lease",
-        next_action="wait or adopt via allocator",
-        level="warning",
-        tags={"pr_number": 606, "session_id": "s-123"},
-    )
-    assert event["tags"]["blocker_type"] == "active_foreign_lease"
-    assert event["tags"]["pr_number"] == "606"
-    assert "session_id" not in event["tags"]
-    assert event["tags"]["session_id_hash"]
-    assert event["extra"]["canonical_next_action"] == "wait or adopt via allocator"
-    assert event["fingerprint"] == ["workflow-blocker", "active_foreign_lease"]
-
-
-def test_build_blocker_event_invalid_level_defaults_warning():
-    event = so.build_blocker_event("x", level="nonsense")
-    assert event["level"] == "warning"
-
-
-def test_build_checkin_payload_maps_slugs():
-    for key, slug in so.MONITOR_SLUGS.items():
-        payload = so.build_checkin_payload(key, "ok")
-        assert payload["monitor_slug"] == slug
-        assert payload["status"] == "ok"
-
-
-def test_build_checkin_payload_explicit_slug_passthrough():
-    payload = so.build_checkin_payload("custom-slug", "in_progress", duration=1.5)
-    assert payload["monitor_slug"] == "custom-slug"
-    assert payload["duration"] == 1.5
-
-
-def test_build_checkin_payload_rejects_bad_status():
-    with pytest.raises(ValueError):
-        so.build_checkin_payload("allocator_health", "bogus")
-
-
-def test_all_six_monitors_registered():
-    assert set(so.MONITOR_SLUGS) == {
-        "stale_lease_scan",
-        "terminal_lock_scan",
-        "allocator_health",
-        "namespace_health",
-        "dashboard_freshness",
-        "reconciler_cleanup",
-    }
-
-
-# ── Capture paths (fail open) ───────────────────────────────────────────────
-def test_capture_workflow_blocker_noop_when_disabled(fake_sdk):
-    # not initialised
-    event = so.capture_workflow_blocker("some_blocker", message="x")
-    assert isinstance(event, dict)  # still returns redacted event
-    assert fake_sdk.events == []  # but nothing sent
-
-
-def test_capture_workflow_blocker_sends_when_initialised(fake_sdk):
-    so.init_sentry(
-        so.load_config(
-            env={"MCP_SENTRY_ENABLED": "1", "SENTRY_DSN": "https://[email protected]/1"}
-        )
-    )
-    so.capture_workflow_blocker("terminal_lock_occupied", message="held")
-    assert len(fake_sdk.events) == 1
-    assert fake_sdk.events[0]["tags"]["blocker_type"] == "terminal_lock_occupied"
-
-
-def test_capture_exception_noop_when_disabled(fake_sdk):
-    assert so.capture_exception(ValueError("x")) is False
-    assert fake_sdk.exceptions == []
-
-
-def test_capture_exception_sends_scrubbed_tags(fake_sdk):
-    so.init_sentry(
-        so.load_config(
-            env={"MCP_SENTRY_ENABLED": "1", "SENTRY_DSN": "https://[email protected]/1"}
-        )
-    )
-    ok = so.capture_exception(
-        RuntimeError("bad"), tags={"mutation_tool": "gitea_merge_pr", "leaky": "x"}
-    )
-    assert ok is True
-    assert len(fake_sdk.exceptions) == 1
-    assert fake_sdk.last_scope.tags["mutation_tool"] == "gitea_merge_pr"
-    assert "leaky" not in fake_sdk.last_scope.tags
-
-
-def test_capture_exception_never_raises(monkeypatch, fake_sdk):
-    so.init_sentry(
-        so.load_config(
-            env={"MCP_SENTRY_ENABLED": "1", "SENTRY_DSN": "https://[email protected]/1"}
-        )
-    )
-
-    def boom(exc):
-        raise RuntimeError("sdk exploded")
-
-    monkeypatch.setattr(fake_sdk, "capture_exception", boom)
-    # Must swallow the SDK failure (fail open).
-    assert so.capture_exception(ValueError("y")) is False
-
-
-def test_monitor_checkin_noop_when_disabled(fake_sdk):
-    payload = so.monitor_checkin("allocator_health", "ok")
-    assert payload["monitor_slug"] == "gitea-mcp-allocator-health"
-    assert fake_sdk.checkins == []  # not sent while disabled
-
-
-def test_monitor_checkin_sends_when_initialised(fake_sdk):
-    so.init_sentry(
-        so.load_config(
-            env={"MCP_SENTRY_ENABLED": "1", "SENTRY_DSN": "https://[email protected]/1"}
-        )
-    )
-    so.monitor_checkin("stale_lease_scan", "ok")
-    assert fake_sdk.checkins == [
-        {"monitor_slug": "gitea-mcp-stale-lease-scan", "status": "ok"}
-    ]
-
-
-def test_monitor_checkin_invalid_status_returns_none(fake_sdk):
-    assert so.monitor_checkin("allocator_health", "bogus") is None
-    assert fake_sdk.checkins == []