feat(observability): optional self-hosted Sentry instrumentation for MCP workflow failures (Closes #606)

Add an env-var-gated, off-by-default Sentry SDK integration so MCP 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/. Gitea stays the source of truth; Sentry is
observe-only.

New module `sentry_observability.py` mirrors the `gitea_audit` conventions
(env-gated, best-effort, redacting):

- Config from MCP_SENTRY_ENABLED / SENTRY_DSN / SENTRY_ENVIRONMENT /
  SENTRY_RELEASE / MCP_SENTRY_TRACES_SAMPLE_RATE / MCP_SENTRY_ENABLE_LOGS.
  Active only when enabled AND a DSN is present; otherwise a hard no-op.
- Fail OPEN for observability (never blocks a tool success path) and fail
  CLOSED for redaction (drop a field rather than risk leaking it).
- `scrub_event` before_send/before_send_log hook + allowlisted tags: no
  tokens/passwords/keychain ids/DSNs/cookies, no raw session-state or full
  prompt bodies (session_id -> 12-char hash), no full filesystem paths
  (worktree path -> coarse category). Reuses incident_bridge + gitea_audit
  scrubbers.
- capture_exception, capture_workflow_blocker (with canonical next action),
  and monitor_checkin with six stable cron slugs (stale lease scan, terminal
  lock scan, allocator health, namespace health, dashboard freshness,
  reconciler cleanup).
- `sentry_sdk` is a lazily-imported optional dependency; the module imports
  and no-ops cleanly when the package is absent.

Wiring in gitea_mcp_server.py (additive, guarded, best-effort):
- init_sentry() in __main__ before mcp.run.
- capture_exception in the `_audited` failure path; capture_workflow_blocker
  in `_audit_pr_result` BLOCKED/FAILED path.
- allocator + namespace-health watchdog check-ins at their MCP tool sites
  (domain modules left pure).

Also: pin `sentry-sdk==2.20.0` (optional), document the six env vars in
`.env.example`, and add `docs/observability/sentry-integration.md` covering
project creation in https://sentry.prgs.cc/, DSN handling, local/dev/prod
config, redaction guarantees, and coexistence with the #612 incident bridge.

Tests: tests/test_sentry_observability.py (36 cases) cover disabled / enabled /
missing-DSN / missing-SDK, redaction, exception capture, workflow-blocker
capture, and cron check-in behaviour. Full suite: 2632 passed, 6 skipped.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

.env.example

@@ -39,6 +39,27 @@ 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

@@ -1,807 +0,0 @@
-"""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

@@ -0,0 +1,138 @@
+# 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,219 +619,18 @@ 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.
-
-    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.
-    """
+    """Verify that identity and capability were verified prior to session edits."""
     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
 
@@ -918,30 +717,6 @@ 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()
 
 
@@ -1143,6 +918,7 @@ 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
@@ -1155,7 +931,6 @@ 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
@@ -2003,6 +1778,18 @@ 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,
@@ -2049,6 +1836,20 @@ 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
@@ -4059,30 +3860,6 @@ 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"]:
@@ -5037,12 +4814,7 @@ def gitea_edit_pr(
         raise ValueError("At least one field to edit (title, body, state, base) must be provided.")
 
     closing = payload.get("state") == "closed"
-    # 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")
+    verify_preflight_purity(remote, task="close_pr" if closing else None)
 
     # PR closure is a first-class capability, distinct from retitling or
     # rebasing edits (#216). Gate BEFORE auth/API setup so a blocked close
@@ -5524,25 +5296,6 @@ 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)"
@@ -7662,11 +7415,7 @@ def gitea_acquire_reviewer_pr_lease(
             "permission_report": _permission_block_report("gitea.pr.comment"),
         }
 
-    # 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"
-    )
+    _verify_role_mutation_workspace(remote, worktree=worktree, task="review_pr")
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     profile = get_profile()
@@ -7807,7 +7556,6 @@ 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"
     )
@@ -10233,6 +9981,11 @@ 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
 
 
@@ -12160,6 +11913,18 @@ 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
 
 
@@ -12484,4 +12249,10 @@ 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,6 +31,7 @@ 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

@@ -0,0 +1,535 @@
+"""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,15 +60,6 @@ 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",
@@ -77,22 +68,10 @@ 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

@@ -0,0 +1,412 @@
+"""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 == []