feat(pr-sync): native assess and author update-by-merge lifecycle

Prevent approved PRs from stalling when master advances. Add
gitea_assess_pr_sync_status and gitea_update_pr_branch_by_merge with
head/base pinning, author-only updates, conflict handoff, and approval
invalidation after head changes. Wire task capability map, sequential
controller routing in review-merge workflow, and hermetic AC tests.

.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

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

@@ -1104,6 +1104,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
@@ -1133,6 +1134,7 @@ import review_merge_state_machine  # noqa: E402
 import pr_work_lease  # noqa: E402
 import workflow_skill  # noqa: E402
 import conflict_fix_classification  # noqa: E402
+import pr_sync_status  # noqa: E402  # PR sync / update-by-merge lifecycle
 import native_mcp_preference  # noqa: E402
 import branch_cleanup_guard  # noqa: E402
 import thread_state_ledger_validator  # noqa: E402
@@ -2005,6 +2007,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,
@@ -2051,6 +2065,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
@@ -12730,6 +12758,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
 
 
@@ -13163,6 +13196,616 @@ def gitea_assess_conflict_fix_classification(
     return result
 
 
+def _count_commits_behind(
+    base_url: str,
+    auth: dict,
+    *,
+    pr_head_sha: str,
+    base_head_sha: str,
+) -> int | None:
+    """Return how many base commits are missing from the PR head, if knowable."""
+    # compare old...new returns commits reachable from new not from old.
+    # pr_head...base_head ≈ commits on base not in PR (behind count).
+    try:
+        cmp_url = (
+            f"{base_url}/compare/{pr_head_sha}...{base_head_sha}"
+        )
+        data = api_request("GET", cmp_url, auth) or {}
+        total = data.get("total_commits")
+        if isinstance(total, int) and total >= 0:
+            return total
+        commits = data.get("commits")
+        if isinstance(commits, list):
+            return len(commits)
+    except Exception:
+        pass
+    return None
+
+
+def _branch_protection_requires_current_base(
+    base_url: str,
+    auth: dict,
+    *,
+    base_branch: str,
+) -> bool | None:
+    """Read Gitea branch protection ``block_on_outdated_branch`` when present."""
+    branch = (base_branch or "").strip()
+    if not branch:
+        return None
+    try:
+        # Prefer the named protection rule matching the base branch.
+        protections = api_request(
+            "GET", f"{base_url}/branch_protections", auth
+        ) or []
+        if isinstance(protections, list):
+            for rule in protections:
+                if not isinstance(rule, dict):
+                    continue
+                name = (rule.get("branch_name") or rule.get("rule_name") or "").strip()
+                # Exact match or glob-ish contains for common patterns.
+                if name == branch or name in ("*", f"{branch}"):
+                    if "block_on_outdated_branch" in rule:
+                        return bool(rule.get("block_on_outdated_branch"))
+            # Fallback: any protection that mentions the base branch.
+            for rule in protections:
+                if not isinstance(rule, dict):
+                    continue
+                name = (rule.get("branch_name") or rule.get("rule_name") or "").strip()
+                if branch in name or name.endswith(branch):
+                    if "block_on_outdated_branch" in rule:
+                        return bool(rule.get("block_on_outdated_branch"))
+        # Branch payload may embed effective protection.
+        br = api_request("GET", f"{base_url}/branches/{branch}", auth) or {}
+        prot = br.get("protection") or br.get("effective_branch_protection") or {}
+        if isinstance(prot, dict) and "block_on_outdated_branch" in prot:
+            return bool(prot.get("block_on_outdated_branch"))
+    except Exception:
+        return None
+    return None
+
+
+@mcp.tool()
+def gitea_assess_pr_sync_status(
+    pr_number: int,
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    prepared_verdict_head_sha: str | None = None,
+    branch_protection_requires_current_base: bool | None = None,
+    checks_status: str | None = None,
+) -> dict:
+    """Read-only: assess whether an approved/open PR is merge-ready or needs sync.
+
+    Distinguishes ``merge_now``, ``update_branch_by_merge``,
+    ``author_conflict_remediation``, ``fresh_review_required``, and ``blocked``.
+
+    Pins exact PR head and live base head. Never mutates. Never returns tokens.
+    """
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "success": False,
+            "performed": False,
+            "recommended_next_action": pr_sync_status.ACTION_BLOCKED,
+            "reasons": read_block,
+            "permission_report": _permission_block_report("gitea.read"),
+        }
+
+    try:
+        h, o, r = _resolve(remote, host, org, repo)
+    except Exception as exc:
+        return {
+            "success": False,
+            "performed": False,
+            "recommended_next_action": pr_sync_status.ACTION_BLOCKED,
+            "reasons": [f"repository identity resolve failed: {_redact(str(exc))}"],
+        }
+
+    auth = _auth(h)
+    if not auth:
+        return {
+            "success": False,
+            "performed": False,
+            "recommended_next_action": pr_sync_status.ACTION_BLOCKED,
+            "reasons": ["credentials unavailable (fail closed)"],
+            "host": h,
+            "org": o,
+            "repo": r,
+        }
+
+    base = repo_api_url(h, o, r)
+    try:
+        pr = api_request("GET", f"{base}/pulls/{pr_number}", auth) or {}
+    except Exception as exc:
+        return {
+            "success": False,
+            "performed": False,
+            "recommended_next_action": pr_sync_status.ACTION_BLOCKED,
+            "reasons": [f"PR details could not be retrieved: {_redact(str(exc))}"],
+            "host": h,
+            "org": o,
+            "repo": r,
+            "pr_number": pr_number,
+        }
+
+    pr_state = (pr.get("state") or "").strip().lower() or None
+    head_obj = pr.get("head") or {}
+    base_obj = pr.get("base") or {}
+    pr_head_sha = (head_obj.get("sha") or "").strip() or None
+    source_branch = (head_obj.get("ref") or "").strip() or None
+    base_branch = (base_obj.get("ref") or "").strip() or "master"
+    base_head_sha = (base_obj.get("sha") or "").strip() or None
+    mergeable = pr.get("mergeable")
+    # Gitea sometimes exposes conflict via mergeable=false only.
+    has_conflicts = False if mergeable is True else (True if mergeable is False else None)
+
+    # Prefer live base branch tip over the PR's stored base SHA (may lag).
+    try:
+        br = api_request("GET", f"{base}/branches/{base_branch}", auth) or {}
+        live_base = ((br.get("commit") or {}).get("id") or "").strip()
+        if live_base:
+            base_head_sha = live_base
+    except Exception:
+        pass
+
+    commits_behind = None
+    if pr_head_sha and base_head_sha:
+        commits_behind = _count_commits_behind(
+            base, auth, pr_head_sha=pr_head_sha, base_head_sha=base_head_sha
+        )
+        if commits_behind is None:
+            # Fail closed on unknown behind-count only when SHAs differ.
+            commits_behind = 0 if pr_head_sha == base_head_sha else None
+
+    if branch_protection_requires_current_base is None:
+        branch_protection_requires_current_base = (
+            _branch_protection_requires_current_base(
+                base, auth, base_branch=base_branch
+            )
+        )
+    # When protection cannot be read, fail closed by requiring current base
+    # whenever the PR is behind (safer for protected repos). Callers may pass
+    # an explicit False when policy is known not to require it.
+    if branch_protection_requires_current_base is None:
+        if commits_behind is not None and commits_behind > 0:
+            branch_protection_requires_current_base = True
+        else:
+            branch_protection_requires_current_base = False
+
+    approval_at_current_head = None
+    try:
+        feedback = gitea_get_pr_review_feedback(
+            pr_number=pr_number, remote=remote, host=host, org=org, repo=repo,
+        )
+        if feedback.get("success"):
+            approval_at_current_head = bool(feedback.get("approval_at_current_head"))
+        else:
+            approval_at_current_head = False
+    except Exception:
+        approval_at_current_head = None
+
+    # Locks / leases (best-effort; absence is reported as None/False).
+    active_author_lock = None
+    try:
+        lock = issue_lock_store.load_issue_lock(
+            remote=remote if remote in REMOTES else (h or "unknown"),
+            org=o,
+            repo=r,
+            issue_number=pr_number,
+        )
+        if lock:
+            active_author_lock = issue_lock_store.is_lease_live(lock)
+        else:
+            active_author_lock = False
+    except Exception:
+        active_author_lock = None
+
+    active_reviewer_lease = None
+    active_merger_lease = None
+    try:
+        comments = api_request(
+            "GET", f"{base}/issues/{pr_number}/comments", auth
+        ) or []
+        if isinstance(comments, list):
+            rev = pr_work_lease.find_active_reviewer_lease(
+                comments, pr_number=pr_number
+            )
+            active_reviewer_lease = bool(rev)
+            # Merger lease comments share reviewer_pr_lease session store when present.
+            try:
+                sess = reviewer_pr_lease.get_session_lease() or {}
+                active_merger_lease = bool(
+                    sess.get("active")
+                    and int(sess.get("pr_number") or 0) == int(pr_number)
+                    and (sess.get("phase") or "").startswith("merger")
+                )
+            except Exception:
+                active_merger_lease = False
+    except Exception:
+        pass
+
+    if checks_status is None:
+        checks_status = "unknown"
+        # Combined status on PR head when Actions/status API is available.
+        if pr_head_sha:
+            try:
+                st = api_request(
+                    "GET",
+                    f"{base}/commits/{pr_head_sha}/status",
+                    auth,
+                ) or {}
+                state = (st.get("state") or "").strip().lower()
+                if state:
+                    checks_status = state
+                elif not st:
+                    checks_status = "none"
+            except Exception:
+                checks_status = "unknown"
+
+    assessment = pr_sync_status.assess_pr_sync_status(
+        host=h,
+        org=o,
+        repo=r,
+        pr_number=pr_number,
+        pr_state=pr_state,
+        source_branch=source_branch,
+        pr_head_sha=pr_head_sha,
+        base_head_sha=base_head_sha,
+        commits_behind=commits_behind,
+        mergeable=mergeable,
+        has_conflicts=has_conflicts,
+        branch_protection_requires_current_base=branch_protection_requires_current_base,
+        approval_at_current_head=approval_at_current_head,
+        checks_status=checks_status,
+        active_author_lock=active_author_lock,
+        active_reviewer_lease=active_reviewer_lease,
+        active_merger_lease=active_merger_lease,
+        prepared_verdict_head_sha=prepared_verdict_head_sha,
+    )
+    assessment["remote"] = remote if remote in REMOTES else None
+    assessment["base_branch"] = base_branch
+    assessment["success"] = True
+    assessment["performed"] = False
+    return assessment
+
+
+@mcp.tool()
+def gitea_update_pr_branch_by_merge(
+    pr_number: int,
+    expected_pr_head_sha: str,
+    expected_base_head_sha: str,
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    worktree_path: str | None = None,
+) -> dict:
+    """Author-only: merge live base into the PR branch (Gitea Update-by-merge).
+
+    Pins both expected PR head and expected base head; fails closed on either
+    race. Never rebases or force-pushes. On conflicts, returns a structured
+    author-remediation handoff without partial remote mutation.
+
+    Requires author profile, existing issue/PR lock, PR worktree under
+    branches/, clean control checkout, and master parity.
+    """
+    profile = get_profile()
+    allowed = profile.get("allowed_operations") or []
+    forbidden = profile.get("forbidden_operations") or []
+    role = _role_kind(allowed, forbidden)
+
+    # Permission: author branch push / PR mutation surface.
+    push_block = _profile_operation_gate("gitea.branch.push")
+    if push_block:
+        return {
+            "success": False,
+            "performed": False,
+            "mutation_allowed": False,
+            "reasons": push_block,
+            "permission_report": _permission_block_report("gitea.branch.push"),
+            "role_kind": role,
+        }
+
+    if role != "author":
+        pre = pr_sync_status.assess_update_pr_branch_preflight(
+            role_kind=role,
+            expected_pr_head_sha=expected_pr_head_sha,
+            live_pr_head_sha=expected_pr_head_sha,
+            expected_base_head_sha=expected_base_head_sha,
+            live_base_head_sha=expected_base_head_sha,
+            style="merge",
+        )
+        pre["success"] = False
+        return pre
+
+    try:
+        h, o, r = _resolve(remote, host, org, repo)
+    except Exception as exc:
+        return {
+            "success": False,
+            "performed": False,
+            "mutation_allowed": False,
+            "reasons": [f"repository identity resolve failed: {_redact(str(exc))}"],
+            "role_kind": role,
+        }
+
+    # Workspace / control / parity gates.
+    try:
+        _verify_role_mutation_workspace(
+            remote, worktree_path=worktree_path, task="push_branch"
+        )
+    except Exception as exc:
+        return {
+            "success": False,
+            "performed": False,
+            "mutation_allowed": False,
+            "reasons": [f"workspace gate failed: {_redact(str(exc))}"],
+            "role_kind": role,
+            "host": h,
+            "org": o,
+            "repo": r,
+            "pr_number": pr_number,
+        }
+
+    control_clean = True
+    try:
+        porcelain = _get_workspace_porcelain(PROJECT_ROOT)
+        # Untracked-only dirt is ignored; tracked edits contaminate control.
+        tracked_dirty = [
+            line for line in (porcelain or "").splitlines()
+            if line.strip() and not line.startswith("??")
+        ]
+        control_clean = not bool(tracked_dirty)
+    except Exception:
+        control_clean = False
+
+    master_ok = True
+    try:
+        stale = _master_parity_block("gitea.branch.push")
+        master_ok = not bool(stale)
+    except Exception:
+        master_ok = False
+
+    wt = (worktree_path or "").strip() or (
+        os.environ.get("GITEA_AUTHOR_WORKTREE")
+        or os.environ.get("GITEA_ACTIVE_WORKTREE")
+        or ""
+    ).strip()
+
+    has_lock = False
+    try:
+        lock = issue_lock_store.read_session_issue_lock()
+        if lock and int(lock.get("issue_number") or 0) == int(pr_number):
+            has_lock = issue_lock_store.is_lease_live(lock)
+        if not has_lock:
+            lock2 = issue_lock_store.load_issue_lock(
+                remote=remote if remote in REMOTES else (h or "unknown"),
+                org=o,
+                repo=r,
+                issue_number=pr_number,
+            )
+            has_lock = bool(lock2 and issue_lock_store.is_lease_live(lock2))
+    except Exception:
+        has_lock = False
+
+    auth = _auth(h)
+    if not auth:
+        return {
+            "success": False,
+            "performed": False,
+            "mutation_allowed": False,
+            "reasons": ["credentials unavailable (fail closed)"],
+            "role_kind": role,
+            "host": h,
+            "org": o,
+            "repo": r,
+            "pr_number": pr_number,
+        }
+
+    base = repo_api_url(h, o, r)
+    try:
+        pr = api_request("GET", f"{base}/pulls/{pr_number}", auth) or {}
+    except Exception as exc:
+        return {
+            "success": False,
+            "performed": False,
+            "mutation_allowed": False,
+            "reasons": [f"PR details could not be retrieved: {_redact(str(exc))}"],
+            "role_kind": role,
+            "host": h,
+            "org": o,
+            "repo": r,
+            "pr_number": pr_number,
+        }
+
+    if (pr.get("state") or "").strip().lower() != "open":
+        return {
+            "success": False,
+            "performed": False,
+            "mutation_allowed": False,
+            "reasons": [
+                f"PR is not open (state={pr.get('state')}); refuse update"
+            ],
+            "role_kind": role,
+            "host": h,
+            "org": o,
+            "repo": r,
+            "pr_number": pr_number,
+        }
+
+    head_obj = pr.get("head") or {}
+    base_obj = pr.get("base") or {}
+    live_pr_head = (head_obj.get("sha") or "").strip() or None
+    source_branch = (head_obj.get("ref") or "").strip() or None
+    base_branch = (base_obj.get("ref") or "").strip() or "master"
+    live_base_head = (base_obj.get("sha") or "").strip() or None
+    try:
+        br = api_request("GET", f"{base}/branches/{base_branch}", auth) or {}
+        tip = ((br.get("commit") or {}).get("id") or "").strip()
+        if tip:
+            live_base_head = tip
+    except Exception:
+        pass
+
+    mergeable = pr.get("mergeable")
+    has_conflicts = False if mergeable is True else (True if mergeable is False else None)
+
+    owns_branch = True
+    if wt and source_branch:
+        try:
+            # Best-effort: worktree branch should match PR source branch.
+            import subprocess as _sp
+            out = _sp.check_output(
+                ["git", "-C", wt, "rev-parse", "--abbrev-ref", "HEAD"],
+                text=True,
+                stderr=_sp.DEVNULL,
+            ).strip()
+            if out and out != source_branch:
+                owns_branch = False
+        except Exception:
+            owns_branch = None  # unknown — do not hard-fail solely on probe
+
+    preflight = pr_sync_status.assess_update_pr_branch_preflight(
+        role_kind=role,
+        expected_pr_head_sha=expected_pr_head_sha,
+        live_pr_head_sha=live_pr_head,
+        expected_base_head_sha=expected_base_head_sha,
+        live_base_head_sha=live_base_head,
+        has_conflicts=has_conflicts,
+        mergeable=mergeable,
+        style="merge",
+        has_author_lock=has_lock,
+        worktree_path=wt or None,
+        worktree_owns_source_branch=owns_branch,
+        control_checkout_clean=control_clean,
+        master_parity_ok=master_ok,
+        force_push=False,
+        rebase=False,
+    )
+    preflight["host"] = h
+    preflight["org"] = o
+    preflight["repo"] = r
+    preflight["pr_number"] = pr_number
+    preflight["source_branch"] = source_branch
+    preflight["base_branch"] = base_branch
+    preflight["worktree_path"] = wt or None
+    preflight["profile_name"] = profile.get("profile_name")
+
+    if not preflight.get("mutation_allowed"):
+        preflight["success"] = False
+        preflight["performed"] = False
+        if preflight.get("author_remediation_handoff"):
+            preflight["recommended_next_action"] = (
+                pr_sync_status.ACTION_AUTHOR_CONFLICT_REMEDIATION
+            )
+        return preflight
+
+    # Native Gitea update: POST .../pulls/{index}/update?style=merge
+    update_url = f"{base}/pulls/{pr_number}/update?style=merge"
+    try:
+        api_request("POST", update_url, auth)
+    except Exception as exc:
+        msg = _redact(str(exc)).lower()
+        conflictish = any(
+            token in msg
+            for token in ("conflict", "409", "merge conflict", "not mergeable")
+        )
+        if conflictish:
+            return {
+                "success": False,
+                "performed": False,
+                "mutation_allowed": False,
+                "author_remediation_handoff": True,
+                "recommended_next_action": (
+                    pr_sync_status.ACTION_AUTHOR_CONFLICT_REMEDIATION
+                ),
+                "reasons": [
+                    "Gitea refused update-by-merge due to conflicts; "
+                    "no partial remote update applied",
+                    _redact(str(exc)),
+                ],
+                "host": h,
+                "org": o,
+                "repo": r,
+                "pr_number": pr_number,
+                "expected_pr_head_sha": expected_pr_head_sha,
+                "expected_base_head_sha": expected_base_head_sha,
+                "live_pr_head_sha_before": live_pr_head,
+                "style": "merge",
+                "role_kind": role,
+            }
+        return {
+            "success": False,
+            "performed": False,
+            "mutation_allowed": False,
+            "reasons": [
+                f"update-by-merge API failed (fail closed): {_redact(str(exc))}"
+            ],
+            "host": h,
+            "org": o,
+            "repo": r,
+            "pr_number": pr_number,
+            "style": "merge",
+            "role_kind": role,
+        }
+
+    # Re-read new head after successful update.
+    new_head = None
+    try:
+        pr_after = api_request("GET", f"{base}/pulls/{pr_number}", auth) or {}
+        new_head = ((pr_after.get("head") or {}).get("sha") or "").strip() or None
+        mergeable_after = pr_after.get("mergeable")
+    except Exception:
+        pr_after = {}
+        mergeable_after = None
+
+    transition = pr_sync_status.assess_post_update_head_transition(
+        former_pr_head_sha=live_pr_head,
+        new_pr_head_sha=new_head,
+        former_approval_head_sha=live_pr_head,
+        prepared_verdict_head_sha=live_pr_head,
+    )
+
+    return {
+        "success": True,
+        "performed": True,
+        "mutation_allowed": True,
+        "style": "merge",
+        "force_push": False,
+        "rebase": False,
+        "host": h,
+        "org": o,
+        "repo": r,
+        "pr_number": pr_number,
+        "source_branch": source_branch,
+        "base_branch": base_branch,
+        "former_pr_head_sha": live_pr_head,
+        "new_pr_head_sha": new_head,
+        "base_head_sha": live_base_head,
+        "mergeable_after": mergeable_after,
+        "approval_invalidated_for_former_head": transition.get(
+            "approval_invalidated", True
+        ),
+        "reviewer_lease_superseded": transition.get("reviewer_lease_superseded"),
+        "merger_lease_superseded": transition.get("merger_lease_superseded"),
+        "prepared_verdict_invalidated": transition.get(
+            "prepared_verdict_invalidated"
+        ),
+        "recommended_next_action": transition.get(
+            "recommended_next_action",
+            pr_sync_status.ACTION_FRESH_REVIEW_REQUIRED,
+        ),
+        "transition": transition,
+        "role_kind": role,
+        "profile_name": profile.get("profile_name"),
+        "worktree_path": wt or None,
+        "reasons": list(transition.get("reasons") or []) + [
+            "update-by-merge completed via native Gitea API (style=merge only)"
+        ],
+    }
+
+
 @mcp.tool()
 def gitea_assess_conflict_fix_push(
     pr_number: int,
@@ -13835,6 +14478,8 @@ def gitea_resolve_task_capability(
         "commit_files",
         "gitea_commit_files",
         "address_pr_change_requests",
+        "update_pr_branch_by_merge",
+        "gitea_update_pr_branch_by_merge",
         "delete_branch",
         "cleanup_merged_pr_branch",
         "reconciliation_cleanup",
@@ -14660,6 +15305,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
 
 
@@ -15221,4 +15878,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")

pr_sync_status.py

@@ -0,0 +1,648 @@
+"""PR synchronization and conflict-remediation lifecycle (#PR-SYNC).
+
+Pure assessment helpers for:
+
+* ``gitea_assess_pr_sync_status`` — recommend the next sanctioned action for an
+  open PR when the base branch advances, approvals go stale, or conflicts appear.
+* ``gitea_update_pr_branch_by_merge`` preflight — author-only, fail-closed pin
+  of both PR head and live base head; never rebase/force-push.
+
+All recommendation logic is hermetic (no network) so unit tests cover every
+acceptance path without Gitea credentials.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+_FULL_SHA = re.compile(r"^[0-9a-f]{40}$", re.IGNORECASE)
+
+# Recommended next actions (controller / skill vocabulary).
+ACTION_MERGE_NOW = "merge_now"
+ACTION_UPDATE_BRANCH_BY_MERGE = "update_branch_by_merge"
+ACTION_AUTHOR_CONFLICT_REMEDIATION = "author_conflict_remediation"
+ACTION_FRESH_REVIEW_REQUIRED = "fresh_review_required"
+ACTION_BLOCKED = "blocked"
+
+_VALID_ACTIONS = frozenset({
+    ACTION_MERGE_NOW,
+    ACTION_UPDATE_BRANCH_BY_MERGE,
+    ACTION_AUTHOR_CONFLICT_REMEDIATION,
+    ACTION_FRESH_REVIEW_REQUIRED,
+    ACTION_BLOCKED,
+})
+
+# Author-only mutation; reviewer/merger must never update author branches.
+_AUTHOR_UPDATE_ROLES = frozenset({"author"})
+_DENIED_UPDATE_ROLES = frozenset({"reviewer", "merger", "reconciler", "mixed", "limited"})
+
+# Allowed Gitea update style — merge only (never rebase).
+UPDATE_STYLE_MERGE = "merge"
+_FORBIDDEN_UPDATE_STYLES = frozenset({"rebase", "rebase-merge", "squash", "force"})
+
+
+def _normalize_sha(value: str | None) -> str | None:
+    text = (value or "").strip().lower()
+    if not text:
+        return None
+    return text if _FULL_SHA.match(text) else None
+
+
+def _normalize_role(role: str | None) -> str:
+    return (role or "").strip().lower() or "unknown"
+
+
+def assess_pr_sync_status(
+    *,
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    pr_number: int,
+    pr_state: str | None = None,
+    source_branch: str | None = None,
+    pr_head_sha: str | None = None,
+    base_head_sha: str | None = None,
+    commits_behind: int | None = None,
+    mergeable: bool | None = None,
+    has_conflicts: bool | None = None,
+    branch_protection_requires_current_base: bool | None = None,
+    approval_at_current_head: bool | None = None,
+    checks_status: str | None = None,
+    active_author_lock: bool | None = None,
+    active_reviewer_lease: bool | None = None,
+    active_merger_lease: bool | None = None,
+    prepared_verdict_head_sha: str | None = None,
+    checks_required: bool = True,
+) -> dict[str, Any]:
+    """Recommend the next sanctioned PR lifecycle action.
+
+    Decision order (fail closed):
+
+    1. Incomplete identity / open state / head SHAs → ``blocked``
+    2. Conflicts (or mergeable false with conflict signal) →
+       ``author_conflict_remediation``
+    3. Head changed after approval / prepared verdict for former head →
+       ``fresh_review_required`` (unless conflicts already routed author work)
+    4. Behind live base + branch protection requires current base +
+       auto-mergeable → ``update_branch_by_merge``
+    5. Approval at current head + mergeable + checks ok (+ update not
+       required) → ``merge_now``
+    6. Otherwise ``blocked`` with structured reasons
+    """
+    reasons: list[str] = []
+    pr_head = _normalize_sha(pr_head_sha)
+    base_head = _normalize_sha(base_head_sha)
+    prepared_head = _normalize_sha(prepared_verdict_head_sha)
+    state = (pr_state or "").strip().lower()
+    checks = (checks_status or "unknown").strip().lower()
+    behind = commits_behind if isinstance(commits_behind, int) and commits_behind >= 0 else None
+    requires_current = bool(branch_protection_requires_current_base)
+    approval_ok = bool(approval_at_current_head)
+
+    # Derive conflict when caller only supplies mergeable=False.
+    if has_conflicts is None and mergeable is False:
+        has_conflicts = True
+    conflicts = bool(has_conflicts) if has_conflicts is not None else None
+
+    result: dict[str, Any] = {
+        "host": (host or "").strip() or None,
+        "org": (org or "").strip() or None,
+        "repo": (repo or "").strip() or None,
+        "pr_number": pr_number,
+        "pr_state": state or None,
+        "source_branch": (source_branch or "").strip() or None,
+        "pr_head_sha": pr_head,
+        "base_head_sha": base_head,
+        "commits_behind": behind,
+        "mergeable": mergeable,
+        "has_conflicts": conflicts,
+        "branch_protection_requires_current_base": requires_current,
+        "approval_at_current_head": approval_ok if approval_at_current_head is not None else None,
+        "checks_status": checks,
+        "active_locks_and_leases": {
+            "author_lock": bool(active_author_lock) if active_author_lock is not None else None,
+            "reviewer_lease": bool(active_reviewer_lease) if active_reviewer_lease is not None else None,
+            "merger_lease": bool(active_merger_lease) if active_merger_lease is not None else None,
+        },
+        "prepared_verdict_head_sha": prepared_head,
+        "recommended_next_action": ACTION_BLOCKED,
+        "approval_valid_for_merge": False,
+        "stale_approval": False,
+        "stale_prepared_verdict": False,
+        "reasons": reasons,
+        "success": True,
+        "performed": False,
+    }
+
+    if not isinstance(pr_number, int) or pr_number <= 0:
+        reasons.append("pr_number must be a positive integer (fail closed)")
+        return result
+
+    if state and state not in ("open",):
+        reasons.append(f"PR is not open (state={state}); cannot synchronize or merge")
+        result["recommended_next_action"] = ACTION_BLOCKED
+        return result
+    if not state:
+        reasons.append("PR state missing (fail closed)")
+        return result
+
+    if pr_head is None:
+        reasons.append(
+            "exact PR head SHA missing or not a full 40-char hex SHA (fail closed)"
+        )
+    if base_head is None:
+        reasons.append(
+            "exact live base head SHA missing or not a full 40-char hex SHA (fail closed)"
+        )
+    if behind is None:
+        reasons.append("commits_behind missing or invalid (fail closed)")
+    if mergeable is None:
+        reasons.append("mergeable signal missing (fail closed)")
+    if approval_at_current_head is None:
+        reasons.append("approval_at_current_head missing (fail closed)")
+    if branch_protection_requires_current_base is None:
+        reasons.append(
+            "branch_protection_requires_current_base missing (fail closed)"
+        )
+
+    if reasons:
+        result["recommended_next_action"] = ACTION_BLOCKED
+        return result
+
+    # Stale prepared verdict / approval across heads.
+    stale_prepared = bool(prepared_head and prepared_head != pr_head)
+    result["stale_prepared_verdict"] = stale_prepared
+    stale_approval = not approval_ok
+    result["stale_approval"] = stale_approval
+    result["approval_valid_for_merge"] = bool(approval_ok and not stale_prepared)
+
+    # ── Conflicts: author remediation in dedicated worktree ──────────────
+    if conflicts is True or mergeable is False:
+        if conflicts is True or mergeable is False:
+            # Distinguish pure non-mergeable without explicit conflict flag
+            # still routes author remediation (Gitea cannot auto-update).
+            reasons.append(
+                f"PR #{pr_number} has conflicts or is not mergeable at head "
+                f"{pr_head}; route author conflict remediation in the existing "
+                "issue/PR worktree under branches/ (never force-push or rebase)"
+            )
+            if stale_approval:
+                reasons.append(
+                    "any approval at a former head is invalid; after remediation "
+                    "require a fresh independent review at the new exact head"
+                )
+            result["recommended_next_action"] = ACTION_AUTHOR_CONFLICT_REMEDIATION
+            result["approval_valid_for_merge"] = False
+            return result
+
+    # ── Stale approval / prepared verdict at former head ─────────────────
+    if not approval_ok or stale_prepared:
+        if behind and behind > 0 and requires_current and mergeable is True:
+            # Must update first; update will also invalidate approval.
+            reasons.append(
+                f"PR is {behind} commit(s) behind live base and branch protection "
+                "requires current base; automatic merge-from-base is available"
+            )
+            reasons.append(
+                "approval is not valid at the current head (or prepared verdict "
+                "is head-scoped to a former SHA); after update require fresh review"
+            )
+            result["recommended_next_action"] = ACTION_UPDATE_BRANCH_BY_MERGE
+            result["approval_valid_for_merge"] = False
+            return result
+        reasons.append(
+            "approval is not valid at the exact current PR head "
+            f"({pr_head}); never reuse a former-head approval for merge"
+        )
+        if stale_prepared:
+            reasons.append(
+                f"prepared verdict pinned to former head {prepared_head} cannot "
+                f"authorize review/merge at current head {pr_head}"
+            )
+        if active_reviewer_lease:
+            reasons.append(
+                "active reviewer lease must be released or superseded for the "
+                "new head before a fresh independent review"
+            )
+        if active_merger_lease:
+            reasons.append(
+                "active merger lease cannot cross heads; release or re-acquire "
+                "at the new exact head"
+            )
+        result["recommended_next_action"] = ACTION_FRESH_REVIEW_REQUIRED
+        result["approval_valid_for_merge"] = False
+        return result
+
+    # ── Outdated + protection requires current base, auto-updatable ──────
+    if behind and behind > 0 and requires_current:
+        if mergeable is True and conflicts is not True:
+            reasons.append(
+                f"PR is {behind} commit(s) behind live base {base_head}; "
+                "branch protection requires the PR branch to contain current base; "
+                "Gitea can merge base into the PR branch without conflicts"
+            )
+            reasons.append(
+                "route author-only update_branch_by_merge; never rebase or force-push; "
+                "new head invalidates prior approval and requires fresh independent review"
+            )
+            result["recommended_next_action"] = ACTION_UPDATE_BRANCH_BY_MERGE
+            # Approval today is at old head that will change — not mergeable yet
+            # for final merge until re-reviewed, but assess marks update path.
+            result["approval_valid_for_merge"] = False
+            result["stale_approval"] = True  # will become stale after update
+            return result
+        reasons.append(
+            "update required by branch protection but automatic merge is not available"
+        )
+        result["recommended_next_action"] = ACTION_BLOCKED
+        return result
+
+    # ── Checks gate for merge_now ────────────────────────────────────────
+    if checks_required and checks not in ("success", "passed", "ok", "none", "skipped", "not_required"):
+        if checks in ("pending", "running", "queued"):
+            reasons.append(f"required checks are not finished (status={checks})")
+            result["recommended_next_action"] = ACTION_BLOCKED
+            return result
+        if checks in ("failure", "failed", "error", "cancelled"):
+            reasons.append(f"required checks failed (status={checks})")
+            result["recommended_next_action"] = ACTION_BLOCKED
+            return result
+        # unknown — fail closed when checks_required
+        if checks == "unknown":
+            reasons.append("checks status unknown (fail closed)")
+            result["recommended_next_action"] = ACTION_BLOCKED
+            return result
+
+    # ── Ready to merge without update ────────────────────────────────────
+    # Includes: current with approval; outdated when update is NOT required.
+    if approval_ok and mergeable is True and conflicts is not True:
+        if behind and behind > 0 and not requires_current:
+            reasons.append(
+                f"PR is {behind} commit(s) behind live base but branch protection "
+                "does not require the latest base; preserve the approved head"
+            )
+        else:
+            reasons.append(
+                "PR has a valid approval at the exact current head, is conflict-free "
+                "and mergeable; do not update the branch unnecessarily"
+            )
+        reasons.append("route directly to the sanctioned merger workflow (merge_now)")
+        result["recommended_next_action"] = ACTION_MERGE_NOW
+        result["approval_valid_for_merge"] = True
+        return result
+
+    reasons.append("no sanctioned next action matched the observed PR state (fail closed)")
+    result["recommended_next_action"] = ACTION_BLOCKED
+    return result
+
+
+def assess_update_pr_branch_preflight(
+    *,
+    role_kind: str | None,
+    expected_pr_head_sha: str | None,
+    live_pr_head_sha: str | None,
+    expected_base_head_sha: str | None,
+    live_base_head_sha: str | None,
+    has_conflicts: bool | None = None,
+    mergeable: bool | None = None,
+    style: str | None = UPDATE_STYLE_MERGE,
+    has_author_lock: bool | None = None,
+    worktree_path: str | None = None,
+    worktree_owns_source_branch: bool | None = None,
+    control_checkout_clean: bool | None = None,
+    master_parity_ok: bool | None = None,
+    force_push: bool = False,
+    rebase: bool = False,
+) -> dict[str, Any]:
+    """Author-only preflight for update-by-merge; fail closed on any race.
+
+    When conflicts exist, returns a structured author-remediation handoff and
+    sets ``mutation_allowed=False`` so no partial remote update is performed.
+    """
+    reasons: list[str] = []
+    role = _normalize_role(role_kind)
+    exp_pr = _normalize_sha(expected_pr_head_sha)
+    live_pr = _normalize_sha(live_pr_head_sha)
+    exp_base = _normalize_sha(expected_base_head_sha)
+    live_base = _normalize_sha(live_base_head_sha)
+    style_norm = (style or "").strip().lower() or UPDATE_STYLE_MERGE
+
+    conflicts = has_conflicts
+    if conflicts is None and mergeable is False:
+        conflicts = True
+
+    result: dict[str, Any] = {
+        "mutation_allowed": False,
+        "performed": False,
+        "style": style_norm,
+        "role_kind": role,
+        "expected_pr_head_sha": exp_pr,
+        "live_pr_head_sha": live_pr,
+        "expected_base_head_sha": exp_base,
+        "live_base_head_sha": live_base,
+        "head_race": False,
+        "base_race": False,
+        "has_conflicts": conflicts,
+        "author_remediation_handoff": False,
+        "approval_invalidated_for_former_head": False,
+        "force_push": bool(force_push),
+        "rebase": bool(rebase),
+        "reasons": reasons,
+        "success": True,
+    }
+
+    # Role gate — author only.
+    if role not in _AUTHOR_UPDATE_ROLES:
+        reasons.append(
+            f"role '{role}' cannot update an author PR branch "
+            "(author-only; reviewer/merger denial)"
+        )
+        return result
+
+    if force_push:
+        reasons.append("force-push is forbidden for PR branch update (fail closed)")
+        return result
+    if rebase or style_norm in _FORBIDDEN_UPDATE_STYLES:
+        reasons.append(
+            f"update style '{style_norm}' forbidden; only style=merge is allowed "
+            "(never rebase)"
+        )
+        return result
+    if style_norm != UPDATE_STYLE_MERGE:
+        reasons.append(
+            f"unknown update style '{style_norm}'; expected '{UPDATE_STYLE_MERGE}'"
+        )
+        return result
+
+    if not exp_pr or not live_pr:
+        reasons.append(
+            "expected and live PR head SHAs must be full 40-char hex (fail closed)"
+        )
+    if not exp_base or not live_base:
+        reasons.append(
+            "expected and live base head SHAs must be full 40-char hex (fail closed)"
+        )
+    if reasons:
+        return result
+
+    if exp_pr != live_pr:
+        result["head_race"] = True
+        reasons.append(
+            f"PR head race: expected {exp_pr} but live is {live_pr} "
+            "(fail closed; no partial mutation)"
+        )
+        return result
+    if exp_base != live_base:
+        result["base_race"] = True
+        reasons.append(
+            f"base head race: expected {exp_base} but live is {live_base} "
+            "(fail closed; no partial mutation)"
+        )
+        return result
+
+    if has_author_lock is not True:
+        reasons.append(
+            "existing issue/PR lock required before update_branch_by_merge (fail closed)"
+        )
+    wt = (worktree_path or "").strip()
+    if not wt:
+        reasons.append("existing PR worktree path required under branches/ (fail closed)")
+    else:
+        norm = wt.replace("\\", "/")
+        if "/branches/" not in norm and not norm.startswith("branches/"):
+            reasons.append(
+                f"worktree_path '{wt}' must be under branches/ (fail closed)"
+            )
+    if worktree_owns_source_branch is False:
+        reasons.append(
+            "worktree does not own the PR source branch (fail closed)"
+        )
+    if control_checkout_clean is False:
+        reasons.append("control checkout is not clean (fail closed)")
+    if master_parity_ok is False:
+        reasons.append("runtime/master parity failed (fail closed)")
+
+    if conflicts is True:
+        result["author_remediation_handoff"] = True
+        reasons.append(
+            "conflicts present: return structured author-remediation handoff "
+            "without creating a partial remote update"
+        )
+        reasons.append(
+            "resolve conflicts explicitly in the existing dedicated worktree, "
+            "run focused and regression tests, commit/push via sanctioned author "
+            "workflow, then route the new exact head to independent review"
+        )
+        return result
+
+    if mergeable is False and conflicts is not False:
+        result["author_remediation_handoff"] = True
+        reasons.append(
+            "PR is not mergeable; refuse automatic update and hand off to "
+            "author conflict remediation (fail closed)"
+        )
+        return result
+
+    if reasons:
+        return result
+
+    result["mutation_allowed"] = True
+    reasons.append(
+        "preflight passed: author may merge live base into the PR branch via "
+        "native MCP update (style=merge only)"
+    )
+    return result
+
+
+def assess_post_update_head_transition(
+    *,
+    former_pr_head_sha: str | None,
+    new_pr_head_sha: str | None,
+    former_approval_head_sha: str | None = None,
+    former_reviewer_lease_head_sha: str | None = None,
+    former_merger_lease_head_sha: str | None = None,
+    prepared_verdict_head_sha: str | None = None,
+) -> dict[str, Any]:
+    """After a successful update-by-merge, invalidate cross-head artifacts.
+
+    The new head never inherits approval, reviewer/merger leases, or prepared
+    verdicts from the former head.
+    """
+    former = _normalize_sha(former_pr_head_sha)
+    new = _normalize_sha(new_pr_head_sha)
+    reasons: list[str] = []
+    result: dict[str, Any] = {
+        "former_pr_head_sha": former,
+        "new_pr_head_sha": new,
+        "head_changed": bool(former and new and former != new),
+        "approval_invalidated": False,
+        "reviewer_lease_superseded": False,
+        "merger_lease_superseded": False,
+        "prepared_verdict_invalidated": False,
+        "recommended_next_action": ACTION_BLOCKED,
+        "reasons": reasons,
+        "success": True,
+    }
+
+    if not former or not new:
+        reasons.append("former and new PR head SHAs required (fail closed)")
+        return result
+    if former == new:
+        reasons.append(
+            "PR head unchanged after update; unexpected for merge-from-base"
+        )
+        result["recommended_next_action"] = ACTION_BLOCKED
+        return result
+
+    result["head_changed"] = True
+    # Approval at former head is always void at new head.
+    former_approval = _normalize_sha(former_approval_head_sha) or former
+    if former_approval != new:
+        result["approval_invalidated"] = True
+        reasons.append(
+            f"approval for former head {former_approval} is invalid at new head "
+            f"{new}; never preserve approval across a head change"
+        )
+
+    rev_lease = _normalize_sha(former_reviewer_lease_head_sha)
+    if rev_lease and rev_lease != new:
+        result["reviewer_lease_superseded"] = True
+        reasons.append(
+            f"reviewer lease for head {rev_lease} cannot cross to {new}; "
+            "release or supersede before fresh review"
+        )
+    elif rev_lease is None:
+        # Unknown lease head still must not authorize at new head.
+        result["reviewer_lease_superseded"] = True
+        reasons.append(
+            "any reviewer lease scoped to the former head is superseded at the new head"
+        )
+
+    mer_lease = _normalize_sha(former_merger_lease_head_sha)
+    if mer_lease and mer_lease != new:
+        result["merger_lease_superseded"] = True
+        reasons.append(
+            f"merger lease for head {mer_lease} cannot cross to {new}"
+        )
+    else:
+        result["merger_lease_superseded"] = True
+        reasons.append(
+            "any merger lease scoped to the former head is superseded at the new head"
+        )
+
+    prepared = _normalize_sha(prepared_verdict_head_sha)
+    if prepared and prepared != new:
+        result["prepared_verdict_invalidated"] = True
+        reasons.append(
+            f"prepared verdict for head {prepared} cannot authorize the new head {new}"
+        )
+    elif prepared is None:
+        result["prepared_verdict_invalidated"] = True
+        reasons.append(
+            "prepared verdicts associated with the former head are invalidated"
+        )
+
+    result["recommended_next_action"] = ACTION_FRESH_REVIEW_REQUIRED
+    reasons.append(
+        "route the new exact head to independent review "
+        f"({ACTION_FRESH_REVIEW_REQUIRED})"
+    )
+    return result
+
+
+def assess_sequential_queue_step(
+    *,
+    just_merged: bool,
+    master_refreshed: bool,
+    next_pr_number: int | None = None,
+) -> dict[str, Any]:
+    """Controller sequential processing: reassess only after merge + master refresh."""
+    reasons: list[str] = []
+    if just_merged and not master_refreshed:
+        reasons.append(
+            "master must be refreshed after each merge before reassessing the next PR "
+            "(do not update every PR simultaneously)"
+        )
+        return {
+            "reassess_allowed": False,
+            "process_next": False,
+            "next_pr_number": next_pr_number,
+            "reasons": reasons,
+            "success": True,
+        }
+    if not just_merged:
+        reasons.append("no merge completed; sequential step does not advance")
+        return {
+            "reassess_allowed": False,
+            "process_next": False,
+            "next_pr_number": next_pr_number,
+            "reasons": reasons,
+            "success": True,
+        }
+    if next_pr_number:
+        reasons.append(
+            "merge completed and live master refreshed; reassess the next PR "
+            f"#{next_pr_number}"
+        )
+    else:
+        reasons.append(
+            "merge completed and live master refreshed; reassess the next PR"
+        )
+    return {
+        "reassess_allowed": True,
+        "process_next": True,
+        "next_pr_number": next_pr_number,
+        "post_merge_cleanup_handoff": True,
+        "reasons": reasons,
+        "success": True,
+    }
+
+
+def merge_approval_usable_at_head(
+    *,
+    current_head_sha: str | None,
+    approved_head_sha: str | None,
+) -> dict[str, Any]:
+    """Stale approval cannot authorize merge (head-scoped only)."""
+    current = _normalize_sha(current_head_sha)
+    approved = _normalize_sha(approved_head_sha)
+    ok = bool(current and approved and current == approved)
+    reasons: list[str] = []
+    if not ok:
+        reasons.append(
+            f"stale approval: approved head '{approved or '(none)'}' does not "
+            f"match current head '{current or '(none)'}'; cannot authorize merge"
+        )
+    return {
+        "approval_usable": ok,
+        "current_head_sha": current,
+        "approved_head_sha": approved,
+        "reasons": reasons,
+    }
+
+
+def lease_usable_at_head(
+    *,
+    lease_kind: str,
+    lease_head_sha: str | None,
+    current_head_sha: str | None,
+) -> dict[str, Any]:
+    """Reviewer/merger leases cannot cross heads."""
+    current = _normalize_sha(current_head_sha)
+    lease_head = _normalize_sha(lease_head_sha)
+    kind = (lease_kind or "").strip().lower() or "unknown"
+    ok = bool(current and lease_head and current == lease_head)
+    reasons: list[str] = []
+    if not ok:
+        reasons.append(
+            f"stale {kind} lease: lease head '{lease_head or '(none)'}' cannot "
+            f"authorize work at current head '{current or '(none)'}'"
+        )
+    return {
+        "lease_usable": ok,
+        "lease_kind": kind,
+        "lease_head_sha": lease_head,
+        "current_head_sha": current,
+        "reasons": reasons,
+    }

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

skills/llm-project-workflow/templates/merge-pr.md

@@ -33,22 +33,34 @@ Steps:
    *If the current identity does not match the required role (or is the PR author), STOP. Relaunch/switch to the correct profile first.*
 2. Verify authenticated identity + active profile.
 3. Confirm PR #<pr>: author (not you), state open, mergeable, review approved. Check if PR body uses `Closes #N` or `Fixes #N`; if it uses `Implements #N` or `Refs #N`, manual closing will be needed in step 29.
-4. Capability evidence (#179): cite the exact gitea_resolve_task_capability
+4. **PR sync assess (required):** call `gitea_assess_pr_sync_status` with
+   explicit `remote`/`org`/`repo`. Route on `recommended_next_action`:
+   - `merge_now` → continue merger path (do **not** update the branch)
+   - `update_branch_by_merge` → STOP; hand off to **author** for
+     `gitea_update_pr_branch_by_merge` (pins expected PR head + base head)
+   - `author_conflict_remediation` → STOP; author worktree conflict fix
+   - `fresh_review_required` → STOP; independent re-review at current head
+   - `blocked` → STOP and diagnose
+   Never merge on a former-head approval after update/remediation.
+5. Capability evidence (#179): cite the exact gitea_resolve_task_capability
    output (or runtime context) proving merge_pr is allowed — a bare
    "capability checks passed" claim is downgraded.
-5. Final live-state recheck (#179), immediately before the merge mutation —
+6. Final live-state recheck (#179), immediately before the merge mutation —
    re-read the live PR and prove:
    - PR still open
    - live head SHA still equals the pinned/reviewed head SHA
    - base branch unchanged
    - no undismissed REQUEST_CHANGES / blocking review state remains
    If any recheck fails → STOP, re-pin, re-validate.
-6. If any gate fails → STOP and report.
-7. Merge with explicit confirmation (e.g. confirmation="MERGE PR <pr>"),
+7. If any gate fails → STOP and report.
+8. Merge with explicit confirmation (e.g. confirmation="MERGE PR <pr>"),
    pinning the reviewed head SHA (expected_head_sha) and, where supported,
    the changed-file set.
-8. Confirm remote master now contains the merge commit (or the expected changes if squash merged).
+9. Confirm remote master now contains the merge commit (or the expected changes if squash merged).
    *Note: Gitea PR "closed" state is NOT equivalent to "merged". Do not assume a closed PR succeeded without verifying the actual landed changes.*
+10. **Sequential queue:** after merge, refresh live `master`, run post-merge
+    cleanup handoff, then reassess the **next** PR (do not batch-update all
+    open PRs).
 
 Post-merge cleanup (#517): merger sessions must NOT perform ad hoc cleanup.
 - Record merge mutations separately from cleanup mutations in the controller handoff.

skills/llm-project-workflow/workflows/review-merge-pr.md

@@ -950,9 +950,53 @@ Final reports must state:
 * final live head SHA before merge
 * whether any push occurred during validation
 
+## 26D. PR synchronization and conflict-remediation lifecycle
+
+**Do not treat “approved” as the final readiness state.** For every approved
+open PR, call `gitea_assess_pr_sync_status` (native MCP only) and route by
+`recommended_next_action`:
+
+| Action | Meaning | Next role / tools |
+|--------|---------|-------------------|
+| `merge_now` | Valid approval at exact current head; conflict-free; update not required; checks ok | Merger: sanctioned merge workflow only — **do not** update the branch |
+| `update_branch_by_merge` | Behind live base; protection requires current base; Gitea can merge base without conflicts | **Author only:** `gitea_update_pr_branch_by_merge` with pinned `expected_pr_head_sha` + `expected_base_head_sha` |
+| `author_conflict_remediation` | Conflicts / not auto-updatable | **Author only:** existing `branches/` worktree, issue lock, merge master, resolve, test, push; never force-push/rebase |
+| `fresh_review_required` | Head changed after approval, or update/remediation produced a new head | Independent reviewer at the **new exact head**; old approvals/leases/verdicts are void |
+| `blocked` | Other gate (checks, incomplete facts, closed PR, …) | Diagnose; do not merge or update |
+
+### Hard rules
+
+* Pin both expected PR head and expected base head for every update. Either
+  race → fail closed with no partial mutation.
+* Never rebase or force-push author branches.
+* Never update an author branch from a reviewer or merger profile.
+* Never preserve approval, reviewer lease, merger lease, or prepared verdict
+  across a head change.
+* Process PRs **sequentially**: synchronize/remediate one → review new head →
+  merge → refresh live `master` → reassess the next PR. Do not update every
+  open PR at once (each merge restales the rest).
+* Conflict remediation only in the existing dedicated issue/PR worktree under
+  `branches/`. Do not delete that worktree until the updated PR is merged and
+  cleanup eligibility is proven.
+* Post-merge: canonical cleanup handoff to reconciler (section 28).
+
+### After `gitea_update_pr_branch_by_merge` succeeds
+
+1. Treat former-head approval as invalidated.
+2. Release/supersede obsolete reviewer and merger leases.
+3. Route `recommended_next_action=fresh_review_required` at the new head.
+4. Independent re-review, then merger lease/adopt + merge at the new head only.
+
+All Gitea reads/mutations use native MCP. Never substitute direct API, curl,
+tea/gh, Web UI mutation by an LLM, database changes, raw Git push, or token
+access.
+
 ## 27. Merge rules
 
-Before merge, rerun fresh live checks:
+Before merge, call `gitea_assess_pr_sync_status` when the PR is approved/open
+and may be behind base. Only proceed with merge when
+`recommended_next_action` is `merge_now` and approval remains at the current
+head. Then rerun fresh live checks:
 
 * whoami
 * active profile/runtime

task_capability_map.py

@@ -64,6 +64,24 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.branch.push",
         "role": "author",
     },
+    # PR synchronization lifecycle: assess is read-only (any role with gitea.read);
+    # update-by-merge is author-only and mutates the PR head via Gitea API.
+    "assess_pr_sync_status": {
+        "permission": "gitea.read",
+        "role": "author",
+    },
+    "gitea_assess_pr_sync_status": {
+        "permission": "gitea.read",
+        "role": "author",
+    },
+    "update_pr_branch_by_merge": {
+        "permission": "gitea.branch.push",
+        "role": "author",
+    },
+    "gitea_update_pr_branch_by_merge": {
+        "permission": "gitea.branch.push",
+        "role": "author",
+    },
     "review_pr": {
         "permission": "gitea.pr.review",
         "role": "reviewer",

tests/test_pr_sync_status.py

@@ -0,0 +1,338 @@
+"""Hermetic tests for PR synchronization / conflict-remediation lifecycle."""
+
+from __future__ import annotations
+
+import unittest
+
+from pr_sync_status import (
+    ACTION_AUTHOR_CONFLICT_REMEDIATION,
+    ACTION_BLOCKED,
+    ACTION_FRESH_REVIEW_REQUIRED,
+    ACTION_MERGE_NOW,
+    ACTION_UPDATE_BRANCH_BY_MERGE,
+    assess_post_update_head_transition,
+    assess_pr_sync_status,
+    assess_sequential_queue_step,
+    assess_update_pr_branch_preflight,
+    lease_usable_at_head,
+    merge_approval_usable_at_head,
+)
+
+
+def _sha(prefix: str) -> str:
+    return (prefix + "0" * 40)[:40]
+
+
+PR_HEAD = _sha("aaaaaaaa")
+BASE_HEAD = _sha("bbbbbbbb")
+NEW_HEAD = _sha("cccccccc")
+OLD_HEAD = _sha("dddddddd")
+
+
+def _base_kwargs(**overrides):
+    data = {
+        "host": "gitea.prgs.cc",
+        "org": "Scaled-Tech-Consulting",
+        "repo": "Gitea-Tools",
+        "pr_number": 100,
+        "pr_state": "open",
+        "source_branch": "fix/issue-100",
+        "pr_head_sha": PR_HEAD,
+        "base_head_sha": BASE_HEAD,
+        "commits_behind": 0,
+        "mergeable": True,
+        "has_conflicts": False,
+        "branch_protection_requires_current_base": False,
+        "approval_at_current_head": True,
+        "checks_status": "success",
+        "active_author_lock": True,
+        "active_reviewer_lease": False,
+        "active_merger_lease": False,
+    }
+    data.update(overrides)
+    return data
+
+
+class TestAssessPrSyncStatus(unittest.TestCase):
+    def test_approved_current_mergeable_merge_now(self):
+        result = assess_pr_sync_status(**_base_kwargs())
+        self.assertEqual(result["recommended_next_action"], ACTION_MERGE_NOW)
+        self.assertTrue(result["approval_valid_for_merge"])
+        self.assertFalse(result["stale_approval"])
+        self.assertEqual(result["pr_head_sha"], PR_HEAD)
+        self.assertEqual(result["base_head_sha"], BASE_HEAD)
+
+    def test_approved_outdated_update_not_required_merge_now(self):
+        result = assess_pr_sync_status(
+            **_base_kwargs(
+                commits_behind=3,
+                branch_protection_requires_current_base=False,
+            )
+        )
+        self.assertEqual(result["recommended_next_action"], ACTION_MERGE_NOW)
+        self.assertTrue(result["approval_valid_for_merge"])
+        self.assertTrue(any("does not require" in r for r in result["reasons"]))
+
+    def test_outdated_update_required_no_conflict(self):
+        result = assess_pr_sync_status(
+            **_base_kwargs(
+                commits_behind=2,
+                branch_protection_requires_current_base=True,
+                mergeable=True,
+                has_conflicts=False,
+            )
+        )
+        self.assertEqual(
+            result["recommended_next_action"], ACTION_UPDATE_BRANCH_BY_MERGE
+        )
+        self.assertFalse(result["approval_valid_for_merge"])
+        self.assertTrue(any("update_branch_by_merge" in r for r in result["reasons"]))
+
+    def test_outdated_has_conflicts_author_remediation(self):
+        result = assess_pr_sync_status(
+            **_base_kwargs(
+                commits_behind=5,
+                branch_protection_requires_current_base=True,
+                mergeable=False,
+                has_conflicts=True,
+            )
+        )
+        self.assertEqual(
+            result["recommended_next_action"], ACTION_AUTHOR_CONFLICT_REMEDIATION
+        )
+        self.assertFalse(result["approval_valid_for_merge"])
+
+    def test_stale_approval_fresh_review_required(self):
+        result = assess_pr_sync_status(
+            **_base_kwargs(
+                approval_at_current_head=False,
+                commits_behind=0,
+            )
+        )
+        self.assertEqual(
+            result["recommended_next_action"], ACTION_FRESH_REVIEW_REQUIRED
+        )
+        self.assertTrue(result["stale_approval"])
+        self.assertFalse(result["approval_valid_for_merge"])
+
+    def test_stale_prepared_verdict_cannot_cross_heads(self):
+        result = assess_pr_sync_status(
+            **_base_kwargs(
+                approval_at_current_head=True,
+                prepared_verdict_head_sha=OLD_HEAD,
+            )
+        )
+        self.assertEqual(
+            result["recommended_next_action"], ACTION_FRESH_REVIEW_REQUIRED
+        )
+        self.assertTrue(result["stale_prepared_verdict"])
+        self.assertFalse(result["approval_valid_for_merge"])
+
+    def test_missing_heads_fail_closed(self):
+        result = assess_pr_sync_status(
+            **_base_kwargs(pr_head_sha="short", base_head_sha=None)
+        )
+        self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
+        self.assertTrue(any("PR head" in r for r in result["reasons"]))
+
+    def test_closed_pr_blocked(self):
+        result = assess_pr_sync_status(**_base_kwargs(pr_state="closed"))
+        self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
+
+    def test_failed_checks_block_merge_now(self):
+        result = assess_pr_sync_status(**_base_kwargs(checks_status="failure"))
+        self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
+
+    def test_stale_approval_with_update_required_routes_update(self):
+        result = assess_pr_sync_status(
+            **_base_kwargs(
+                approval_at_current_head=False,
+                commits_behind=1,
+                branch_protection_requires_current_base=True,
+                mergeable=True,
+                has_conflicts=False,
+            )
+        )
+        self.assertEqual(
+            result["recommended_next_action"], ACTION_UPDATE_BRANCH_BY_MERGE
+        )
+
+
+class TestUpdatePrBranchPreflight(unittest.TestCase):
+    def _ok_kwargs(self, **overrides):
+        data = {
+            "role_kind": "author",
+            "expected_pr_head_sha": PR_HEAD,
+            "live_pr_head_sha": PR_HEAD,
+            "expected_base_head_sha": BASE_HEAD,
+            "live_base_head_sha": BASE_HEAD,
+            "has_conflicts": False,
+            "mergeable": True,
+            "style": "merge",
+            "has_author_lock": True,
+            "worktree_path": "/Users/x/Development/Gitea-Tools/branches/issue-100",
+            "worktree_owns_source_branch": True,
+            "control_checkout_clean": True,
+            "master_parity_ok": True,
+            "force_push": False,
+            "rebase": False,
+        }
+        data.update(overrides)
+        return data
+
+    def test_author_allowed_when_preflight_clean(self):
+        result = assess_update_pr_branch_preflight(**self._ok_kwargs())
+        self.assertTrue(result["mutation_allowed"])
+        self.assertFalse(result["head_race"])
+        self.assertFalse(result["base_race"])
+
+    def test_head_race_fail_closed(self):
+        result = assess_update_pr_branch_preflight(
+            **self._ok_kwargs(live_pr_head_sha=NEW_HEAD)
+        )
+        self.assertFalse(result["mutation_allowed"])
+        self.assertTrue(result["head_race"])
+        self.assertTrue(any("PR head race" in r for r in result["reasons"]))
+
+    def test_base_race_fail_closed(self):
+        result = assess_update_pr_branch_preflight(
+            **self._ok_kwargs(live_base_head_sha=NEW_HEAD)
+        )
+        self.assertFalse(result["mutation_allowed"])
+        self.assertTrue(result["base_race"])
+        self.assertTrue(any("base head race" in r for r in result["reasons"]))
+
+    def test_conflicts_structured_handoff_no_mutation(self):
+        result = assess_update_pr_branch_preflight(
+            **self._ok_kwargs(has_conflicts=True, mergeable=False)
+        )
+        self.assertFalse(result["mutation_allowed"])
+        self.assertTrue(result["author_remediation_handoff"])
+        self.assertTrue(any("conflicts" in r.lower() for r in result["reasons"]))
+
+    def test_reviewer_denied(self):
+        result = assess_update_pr_branch_preflight(
+            **self._ok_kwargs(role_kind="reviewer")
+        )
+        self.assertFalse(result["mutation_allowed"])
+        self.assertTrue(any("author-only" in r for r in result["reasons"]))
+
+    def test_merger_denied(self):
+        result = assess_update_pr_branch_preflight(
+            **self._ok_kwargs(role_kind="merger")
+        )
+        self.assertFalse(result["mutation_allowed"])
+        self.assertTrue(any("author-only" in r for r in result["reasons"]))
+
+    def test_no_force_push(self):
+        result = assess_update_pr_branch_preflight(
+            **self._ok_kwargs(force_push=True)
+        )
+        self.assertFalse(result["mutation_allowed"])
+        self.assertTrue(any("force-push" in r for r in result["reasons"]))
+
+    def test_no_rebase(self):
+        result = assess_update_pr_branch_preflight(
+            **self._ok_kwargs(style="rebase", rebase=True)
+        )
+        self.assertFalse(result["mutation_allowed"])
+        self.assertTrue(any("rebase" in r for r in result["reasons"]))
+
+    def test_missing_lock_fail_closed(self):
+        result = assess_update_pr_branch_preflight(
+            **self._ok_kwargs(has_author_lock=False)
+        )
+        self.assertFalse(result["mutation_allowed"])
+
+    def test_worktree_not_under_branches(self):
+        result = assess_update_pr_branch_preflight(
+            **self._ok_kwargs(worktree_path="/tmp/not-a-branches-path")
+        )
+        self.assertFalse(result["mutation_allowed"])
+        self.assertTrue(any("branches/" in r for r in result["reasons"]))
+
+
+class TestPostUpdateHeadTransition(unittest.TestCase):
+    def test_update_invalidates_approval_and_requires_fresh_review(self):
+        result = assess_post_update_head_transition(
+            former_pr_head_sha=PR_HEAD,
+            new_pr_head_sha=NEW_HEAD,
+            former_approval_head_sha=PR_HEAD,
+            former_reviewer_lease_head_sha=PR_HEAD,
+            former_merger_lease_head_sha=PR_HEAD,
+            prepared_verdict_head_sha=PR_HEAD,
+        )
+        self.assertTrue(result["head_changed"])
+        self.assertTrue(result["approval_invalidated"])
+        self.assertTrue(result["reviewer_lease_superseded"])
+        self.assertTrue(result["merger_lease_superseded"])
+        self.assertTrue(result["prepared_verdict_invalidated"])
+        self.assertEqual(
+            result["recommended_next_action"], ACTION_FRESH_REVIEW_REQUIRED
+        )
+
+    def test_same_head_unexpected(self):
+        result = assess_post_update_head_transition(
+            former_pr_head_sha=PR_HEAD,
+            new_pr_head_sha=PR_HEAD,
+        )
+        self.assertFalse(result["head_changed"])
+        self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
+
+
+class TestStaleArtifacts(unittest.TestCase):
+    def test_stale_approval_cannot_authorize_merge(self):
+        result = merge_approval_usable_at_head(
+            current_head_sha=NEW_HEAD,
+            approved_head_sha=PR_HEAD,
+        )
+        self.assertFalse(result["approval_usable"])
+
+    def test_fresh_approval_usable(self):
+        result = merge_approval_usable_at_head(
+            current_head_sha=NEW_HEAD,
+            approved_head_sha=NEW_HEAD,
+        )
+        self.assertTrue(result["approval_usable"])
+
+    def test_stale_reviewer_lease_cannot_cross_heads(self):
+        result = lease_usable_at_head(
+            lease_kind="reviewer",
+            lease_head_sha=PR_HEAD,
+            current_head_sha=NEW_HEAD,
+        )
+        self.assertFalse(result["lease_usable"])
+
+    def test_stale_merger_lease_cannot_cross_heads(self):
+        result = lease_usable_at_head(
+            lease_kind="merger",
+            lease_head_sha=PR_HEAD,
+            current_head_sha=NEW_HEAD,
+        )
+        self.assertFalse(result["lease_usable"])
+
+
+class TestSequentialQueue(unittest.TestCase):
+    def test_reassess_after_merge_and_master_refresh(self):
+        result = assess_sequential_queue_step(
+            just_merged=True,
+            master_refreshed=True,
+            next_pr_number=101,
+        )
+        self.assertTrue(result["reassess_allowed"])
+        self.assertTrue(result["process_next"])
+        self.assertTrue(result["post_merge_cleanup_handoff"])
+        self.assertEqual(result["next_pr_number"], 101)
+
+    def test_block_reassess_without_master_refresh(self):
+        result = assess_sequential_queue_step(
+            just_merged=True,
+            master_refreshed=False,
+            next_pr_number=101,
+        )
+        self.assertFalse(result["reassess_allowed"])
+        self.assertTrue(any("master must be refreshed" in r for r in result["reasons"]))
+
+
+if __name__ == "__main__":
+    unittest.main()

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 == []