Merge pull request 'feat: add Sentry-to-Gitea incident bridge for MCP workflow failures (Closes #607)' (#767) from feat/issue-607-sentry-incident-bridge into master

docs/observability/sentry-integration.md

@@ -131,7 +131,59 @@ and `incident_links` rows.
 - The bridge remains the **only** sanctioned route from an alert back into
   Gitea workflow state.
 
-## 7. Non-goals
+## 7. Reading Sentry back into Gitea (#607)
+
+[`sentry_incident_bridge.py`](../../sentry_incident_bridge.py) supplies the
+**read** half of the inbound path: it pulls unresolved issues/events from the
+self-hosted Sentry API, normalizes them into #612 observations, and hands them
+to `incident_bridge.reconcile_incident`. It never adds a second linking store —
+`incident_links` on the #613 control-plane DB stays canonical, which is what
+makes the mapping survive restarts.
+
+### Configuration
+
+| Variable | Purpose | Default |
+| --- | --- | --- |
+| `SENTRY_BASE_URL` | Self-hosted Sentry root | `https://sentry.prgs.cc` |
+| `SENTRY_AUTH_TOKEN` | API token — **env only**, never logged or returned | _(unset)_ |
+| `SENTRY_ORG` | Sentry organization slug | _(unset)_ |
+| `SENTRY_PROJECT` | Sentry project slug | _(unset)_ |
+| `MCP_SENTRY_ISSUE_BRIDGE_ENABLED` | Required for `apply=true` | `false` |
+| `MCP_SENTRY_MIN_EVENTS_FOR_ISSUE` | Recurrence threshold before an issue is worth filing | `2` |
+| `MCP_SENTRY_LOOKBACK` | Scan window (`statsPeriod`, e.g. `24h`) | `24h` |
+
+Missing org/project fails closed as `not_configured`; a missing token fails
+closed as `missing_token` **before** any HTTP call is made.
+
+### Tools
+
+| Tool | Mode | Purpose |
+| --- | --- | --- |
+| `gitea_sentry_list_issues` | read-only | Unresolved issues, `Link`-header pagination |
+| `gitea_sentry_get_issue_events` | read-only | Sanitized recent + latest event for one issue |
+| `gitea_sentry_reconcile_issue` | dry-run default | One Sentry issue → durable Gitea issue |
+| `gitea_sentry_link_gitea_issue` | dry-run default | Link a Sentry issue to an existing Gitea issue |
+| `gitea_sentry_watchdog` | dry-run default | Scan + create/update issues for active incidents |
+
+### Policy
+
+- **Dedupe:** one Sentry issue maps to exactly one Gitea issue, keyed by
+  provider + base URL + org + project + issue id. Recurrence updates the link
+  (and its `event_count`) instead of filing a duplicate.
+- **No reopen:** a Sentry issue that is no longer `unresolved` is skipped; the
+  bridge never reopens or re-files a closed Gitea issue.
+- **Threshold:** issues below `MCP_SENTRY_MIN_EVENTS_FOR_ISSUE` are skipped, so
+  one-off noise does not become durable work.
+- **Apply is explicit:** `apply=true` requires both
+  `MCP_SENTRY_ISSUE_BRIDGE_ENABLED` and issue-create permission on the profile.
+- **Outages fail closed:** an unreachable Sentry returns `sentry_unavailable`
+  and creates nothing.
+- **Redaction:** secrets are scrubbed and absolute local paths are reduced to a
+  category token (`[path:author]`, `[path:root]`, …) before any value reaches a
+  Gitea issue body. Sensitive tag keys (`authorization`, `cookie`, …) are
+  dropped, and permalinks carrying embedded credentials are discarded entirely.
+
+## 8. Non-goals
 
 - Sentry must **not** become the workflow source of truth.
 - Sentry must **not** approve, merge, close, or mutate Gitea workflow state.

gitea_mcp_server.py

@@ -1800,6 +1800,7 @@ import lease_lifecycle  # noqa: E402
 import workflow_dashboard  # noqa: E402  # #605 live queue/lease dashboard
 import incident_bridge  # noqa: E402
 import sentry_observability  # noqa: E402  (#606 optional Sentry observability)
+import sentry_incident_bridge  # noqa: E402  (#607 Sentry→Gitea incident bridge)
 import agent_temp_artifacts
 import issue_lock_worktree  # noqa: E402
 import issue_lock_provenance  # noqa: E402
@@ -18157,6 +18158,41 @@ def gitea_observability_list_projects(
     }
 
 
+def _incident_recurrence_comment_fn(
+    *,
+    apply: bool,
+    remote: str,
+    host: str | None,
+    org: str | None,
+    repo: str | None,
+    worktree_path: str | None = None,
+):
+    """Sanctioned issue-comment route for AC4 recurrence comments (#607).
+
+    Returns ``None`` on dry runs (never comment) and when the active profile
+    lacks ``gitea.issue.comment``. Withholding the callable degrades to "link
+    updated, no comment" rather than failing the durable mapping, and the
+    reconcile result records why the comment was withheld.
+    """
+    if not apply:
+        return None
+    if _profile_operation_gate("gitea.issue.comment"):
+        return None
+
+    def comment_fn(issue_number, body, g_org, g_repo):
+        return gitea_create_issue_comment(
+            issue_number=int(issue_number),
+            body=body,
+            remote=remote,
+            host=host,
+            org=g_org or org,
+            repo=g_repo or repo,
+            worktree_path=worktree_path,
+        )
+
+    return comment_fn
+
+
 @mcp.tool()
 def gitea_observability_reconcile_incident(
     observation_json: str,
@@ -18283,12 +18319,22 @@ def gitea_observability_reconcile_incident(
                 worktree_path=worktree_path,
             )
 
+    comment_fn = _incident_recurrence_comment_fn(
+        apply=bool(apply),
+        remote=remote,
+        host=host,
+        org=org,
+        repo=repo,
+        worktree_path=worktree_path,
+    )
+
     result = incident_bridge.reconcile_incident(
         db,
         observation=observation if isinstance(observation, dict) else {},
         mappings=mappings,
         apply=bool(apply),
         create_issue_fn=create_fn,
+        comment_issue_fn=comment_fn,
         force_gitea_issue_number=force_gitea_issue_number,
     )
     result["remote"] = remote
@@ -18326,6 +18372,426 @@ def gitea_observability_link_issue(
     )
 
 
+def _sentry_bridge_config(
+    base_url: str | None = None,
+    sentry_org: str | None = None,
+    sentry_project: str | None = None,
+    lookback: str | None = None,
+    min_events: int | None = None,
+) -> "sentry_incident_bridge.SentryBridgeConfig":
+    """Env-backed Sentry bridge config with explicit per-call overrides."""
+    return sentry_incident_bridge.config_with_overrides(
+        sentry_incident_bridge.load_bridge_config(),
+        base_url=base_url,
+        org=sentry_org,
+        project=sentry_project,
+        lookback=lookback,
+        min_events_for_issue=min_events,
+    )
+
+
+def _sentry_error_result(exc: "sentry_incident_bridge.SentryApiError") -> dict:
+    """Convert a Sentry read failure into a redacted, fail-closed result."""
+    payload = exc.as_dict()
+    payload.update({"success": False, "reasons": [str(exc)]})
+    return payload
+
+
+def _sentry_find_issue(
+    config: "sentry_incident_bridge.SentryBridgeConfig",
+    sentry_issue_id: str,
+    token: str,
+) -> dict | None:
+    """Locate one live Sentry issue by id within the configured project."""
+    wanted = str(sentry_issue_id).strip()
+    listing = sentry_incident_bridge.list_issues(
+        config,
+        token=token,
+        query=f"issue.id:{wanted}",
+        limit=1,
+        max_pages=1,
+    )
+    for item in listing.get("issues", []):
+        if str(item.get("id")) == wanted:
+            return item
+    return None
+
+
+@mcp.tool()
+def gitea_sentry_list_issues(
+    query: str = "is:unresolved",
+    limit: int = 25,
+    max_pages: int = 10,
+    cursor: str | None = None,
+    environment: str | None = None,
+    base_url: str | None = None,
+    sentry_org: str | None = None,
+    sentry_project: str | None = None,
+    lookback: str | None = None,
+) -> dict:
+    """List unresolved Sentry issues from the self-hosted server (#607).
+
+    Read-only. The auth token is read from ``SENTRY_AUTH_TOKEN`` in the
+    environment only and is never returned. Results are sanitized (secrets
+    redacted, local paths reduced to a category) before leaving this tool.
+    """
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "success": False,
+            "issues": [],
+            "reasons": read_block,
+            "permission_report": _permission_block_report("gitea.read"),
+        }
+    config = _sentry_bridge_config(base_url, sentry_org, sentry_project, lookback)
+    try:
+        return sentry_incident_bridge.list_issues(
+            config,
+            token=sentry_incident_bridge.resolve_token(),
+            query=query,
+            limit=limit,
+            max_pages=max_pages,
+            cursor=cursor,
+            environment=environment,
+        )
+    except sentry_incident_bridge.SentryApiError as exc:
+        result = _sentry_error_result(exc)
+        result["issues"] = []
+        return result
+
+
+@mcp.tool()
+def gitea_sentry_get_issue_events(
+    sentry_issue_id: str,
+    limit: int = 10,
+    base_url: str | None = None,
+    sentry_org: str | None = None,
+    sentry_project: str | None = None,
+) -> dict:
+    """Fetch sanitized recent events for one Sentry issue (#607).
+
+    Read-only. Returns the latest event plus sanitized tags, environment,
+    release, and timestamps. Sensitive tag keys are dropped.
+    """
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "success": False,
+            "events": [],
+            "reasons": read_block,
+            "permission_report": _permission_block_report("gitea.read"),
+        }
+    config = _sentry_bridge_config(base_url, sentry_org, sentry_project)
+    try:
+        return sentry_incident_bridge.get_issue_events(
+            config,
+            sentry_issue_id,
+            token=sentry_incident_bridge.resolve_token(),
+            limit=limit,
+        )
+    except sentry_incident_bridge.SentryApiError as exc:
+        result = _sentry_error_result(exc)
+        result["events"] = []
+        return result
+
+
+@mcp.tool()
+def gitea_sentry_reconcile_issue(
+    sentry_issue_id: str,
+    apply: bool = False,
+    mappings_json: str | None = None,
+    config_path: str | None = None,
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    base_url: str | None = None,
+    sentry_org: str | None = None,
+    sentry_project: str | None = None,
+    worktree_path: str | None = None,
+) -> dict:
+    """Reconcile one live Sentry issue into a durable Gitea issue (#607).
+
+    Fetches the Sentry issue plus its latest event, converts it into a #612
+    observation, then delegates dedupe/link/create to the sanctioned
+    ``gitea_observability_reconcile_incident`` path. Dry-run by default.
+
+    Dedupe basis: provider identity — provider, base URL, org, project, and
+    Sentry issue id. This bridge does not populate ``fingerprint``, so
+    fingerprint plays no part in its dedupe decisions.
+
+    Gitea remains the source of truth; the raw Sentry incident is never an
+    assignable control-plane work item.
+    """
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "success": False,
+            "apply": bool(apply),
+            "outcome": incident_bridge.OUTCOME_BLOCKED,
+            "reasons": read_block,
+            "permission_report": _permission_block_report("gitea.read"),
+            "raw_incident_assignable": False,
+        }
+
+    config = _sentry_bridge_config(base_url, sentry_org, sentry_project)
+    token = sentry_incident_bridge.resolve_token()
+    try:
+        issue = _sentry_find_issue(config, sentry_issue_id, token)
+        if issue is None:
+            return {
+                "success": False,
+                "apply": bool(apply),
+                "outcome": incident_bridge.OUTCOME_BLOCKED,
+                "reasons": [
+                    f"Sentry issue {sentry_issue_id} not found in the configured "
+                    "project/window (fail closed)"
+                ],
+                "raw_incident_assignable": False,
+            }
+        latest_event = sentry_incident_bridge.get_issue_events(
+            config, issue["id"], token=token, limit=1
+        ).get("latest_event")
+    except sentry_incident_bridge.SentryApiError as exc:
+        result = _sentry_error_result(exc)
+        result.update(
+            {
+                "apply": bool(apply),
+                "outcome": incident_bridge.OUTCOME_BLOCKED,
+                "raw_incident_assignable": False,
+            }
+        )
+        return result
+
+    observation = sentry_incident_bridge.observation_from_issue(
+        issue, config, latest_event=latest_event
+    )
+    return gitea_observability_reconcile_incident(
+        observation_json=json.dumps(observation),
+        apply=apply,
+        mappings_json=mappings_json,
+        config_path=config_path,
+        remote=remote,
+        host=host,
+        org=org,
+        repo=repo,
+        worktree_path=worktree_path,
+    )
+
+
+@mcp.tool()
+def gitea_sentry_link_gitea_issue(
+    sentry_issue_id: str,
+    gitea_issue_number: int,
+    apply: bool = False,
+    mappings_json: str | None = None,
+    config_path: str | None = None,
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    base_url: str | None = None,
+    sentry_org: str | None = None,
+    sentry_project: str | None = None,
+) -> dict:
+    """Link a live Sentry issue to an existing Gitea issue (#607).
+
+    Dry-run by default. ``apply=true`` upserts the durable ``incident_links``
+    mapping only — it never creates a new Gitea issue and never reopens one.
+    """
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "success": False,
+            "apply": bool(apply),
+            "outcome": incident_bridge.OUTCOME_BLOCKED,
+            "reasons": read_block,
+            "permission_report": _permission_block_report("gitea.read"),
+            "raw_incident_assignable": False,
+        }
+
+    config = _sentry_bridge_config(base_url, sentry_org, sentry_project)
+    try:
+        issue = _sentry_find_issue(
+            config, sentry_issue_id, sentry_incident_bridge.resolve_token()
+        )
+    except sentry_incident_bridge.SentryApiError as exc:
+        result = _sentry_error_result(exc)
+        result.update(
+            {
+                "apply": bool(apply),
+                "outcome": incident_bridge.OUTCOME_BLOCKED,
+                "raw_incident_assignable": False,
+            }
+        )
+        return result
+
+    if issue is None:
+        return {
+            "success": False,
+            "apply": bool(apply),
+            "outcome": incident_bridge.OUTCOME_BLOCKED,
+            "reasons": [
+                f"Sentry issue {sentry_issue_id} not found in the configured "
+                "project/window (fail closed)"
+            ],
+            "raw_incident_assignable": False,
+        }
+
+    observation = sentry_incident_bridge.observation_from_issue(issue, config)
+    return gitea_observability_link_issue(
+        observation_json=json.dumps(observation),
+        gitea_issue_number=int(gitea_issue_number),
+        apply=apply,
+        mappings_json=mappings_json,
+        config_path=config_path,
+        remote=remote,
+        host=host,
+        org=org,
+        repo=repo,
+    )
+
+
+@mcp.tool()
+def gitea_sentry_watchdog(
+    apply: bool = False,
+    query: str = "is:unresolved",
+    limit: int = 25,
+    max_pages: int = 10,
+    min_events: int | None = None,
+    lookback: str | None = None,
+    mappings_json: str | None = None,
+    config_path: str | None = None,
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    base_url: str | None = None,
+    sentry_org: str | None = None,
+    sentry_project: str | None = None,
+    worktree_path: str | None = None,
+) -> dict:
+    """Scan Sentry and create/update durable Gitea issues for incidents (#607).
+
+    Dry-run by default. ``apply=true`` additionally requires
+    ``MCP_SENTRY_ISSUE_BRIDGE_ENABLED`` and issue-create permission. Each
+    incident is deduped through the #612 ``incident_links`` substrate by
+    provider identity (provider, base URL, org, project, Sentry issue id), so
+    a recurring failure updates one issue instead of creating duplicates.
+
+    When a linked issue sees continued events, a sanitized recurrence comment
+    is posted through ``gitea_create_issue_comment`` (AC4). That requires
+    ``gitea.issue.comment``; without it the link still updates and the result
+    records that the comment was withheld.
+    """
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "success": False,
+            "apply": bool(apply),
+            "reasons": read_block,
+            "permission_report": _permission_block_report("gitea.read"),
+            "raw_incident_assignable": False,
+        }
+
+    if apply:
+        create_block = _profile_permission_block(
+            task_capability_map.required_permission("create_issue"),
+            number=None,
+            remote=remote,
+            host=host,
+            org=org,
+            repo=repo,
+            org_explicit=org is not None,
+            repo_explicit=repo is not None,
+        )
+        if create_block:
+            return {
+                "success": False,
+                "apply": True,
+                "reasons": create_block.get("reasons")
+                if isinstance(create_block, dict)
+                else [str(create_block)],
+                "raw_incident_assignable": False,
+            }
+
+    config = _sentry_bridge_config(
+        base_url, sentry_org, sentry_project, lookback, min_events
+    )
+
+    try:
+        mappings = incident_bridge.load_project_mappings(
+            config_path=config_path, mappings_json=mappings_json
+        )
+    except Exception as exc:  # noqa: BLE001
+        return {
+            "success": False,
+            "apply": bool(apply),
+            "reasons": [incident_bridge.redact_text(exc)],
+            "raw_incident_assignable": False,
+        }
+
+    try:
+        db = control_plane_db.ControlPlaneDB()
+    except Exception as exc:  # noqa: BLE001
+        return {
+            "success": False,
+            "apply": bool(apply),
+            "reasons": [
+                f"control-plane DB unavailable: {incident_bridge.redact_text(exc)} "
+                "(fail closed, #613)"
+            ],
+            "raw_incident_assignable": False,
+        }
+
+    try:
+        _, o_def, r_def = _resolve(remote, host, org, repo)
+    except ValueError:
+        o_def, r_def = org, repo
+
+    create_fn = None
+    if apply:
+
+        def create_fn(title, body, labels, g_org, g_repo):
+            return gitea_create_issue(
+                title=title,
+                body=body,
+                remote=remote,
+                host=host,
+                org=g_org,
+                repo=g_repo,
+                labels=list(labels),
+                issue_type="bug",
+                initial_status="ready",
+                require_workflow_labels=False,
+                worktree_path=worktree_path,
+            )
+
+    result = sentry_incident_bridge.watchdog(
+        db,
+        config,
+        token=sentry_incident_bridge.resolve_token(),
+        apply=bool(apply),
+        mappings=mappings,
+        gitea_org=o_def,
+        gitea_repo=r_def,
+        query=query,
+        limit=limit,
+        max_pages=max_pages,
+        create_issue_fn=create_fn,
+        comment_issue_fn=_incident_recurrence_comment_fn(
+            apply=bool(apply),
+            remote=remote,
+            host=host,
+            org=org,
+            repo=repo,
+            worktree_path=worktree_path,
+        ),
+    )
+    result["remote"] = remote
+    return result
+
+
 @mcp.tool()
 def gitea_allocate_next_work(
     apply: bool = False,

incident_bridge.py

@@ -484,6 +484,78 @@ def build_gitea_issue_body(inc: NormalizedIncident) -> str:
     return "\n".join(lines)
 
 
+def incident_recurred(
+    existing: dict[str, Any], inc: NormalizedIncident
+) -> tuple[bool, str]:
+    """Did new provider events arrive since the existing link was last synced?
+
+    AC4 asks for a recurrence comment when events *continue*, so a scan that
+    observes no new events must stay silent instead of re-posting the same
+    state on every pass.
+    """
+    old_count = existing.get("event_count")
+    new_count = inc.event_count
+    if (
+        isinstance(old_count, int)
+        and isinstance(new_count, int)
+        and new_count > old_count
+    ):
+        return True, f"event_count advanced {old_count} -> {new_count}"
+    old_seen = str(existing.get("last_seen") or "").strip()
+    new_seen = str(inc.last_seen or "").strip()
+    if new_seen and new_seen != old_seen:
+        return True, f"last_seen advanced '{old_seen}' -> '{new_seen}'"
+    return False, "no new provider events since the last sync"
+
+
+def build_recurrence_comment_body(
+    inc: NormalizedIncident, existing: dict[str, Any], *, reason: str = ""
+) -> str:
+    """Sanitized recurrence comment for an already-linked Gitea issue (AC4).
+
+    Uses the same redaction path as :func:`build_gitea_issue_body`; never
+    carries tokens, raw paths, or session state.
+    """
+    lines = [
+        "## Observability incident recurrence (bridge #612)",
+        "",
+        "<!-- mcp-incident-bridge:recurrence:v1 -->",
+        f"<!-- provider={inc.provider} issue_id={inc.provider_issue_id} -->",
+        "",
+        f"Continued `{inc.provider}` events for this linked incident.",
+        "",
+        f"- **provider_issue_id:** `{inc.provider_issue_id}`",
+    ]
+    if inc.provider_short_id:
+        lines.append(f"- **provider_short_id:** `{inc.provider_short_id}`")
+    if inc.provider_permalink:
+        lines.append(f"- **provider_url:** {inc.provider_permalink}")
+    lines.extend(
+        [
+            f"- **event_count:** `{existing.get('event_count')}` -> "
+            f"`{inc.event_count if inc.event_count is not None else ''}`",
+            f"- **first_seen:** `{inc.first_seen or ''}`",
+            f"- **last_seen:** `{inc.last_seen or ''}`",
+            f"- **environment:** `{inc.environment or ''}`",
+            f"- **severity:** `{inc.severity or ''}`",
+            f"- **culprit:** `{inc.culprit or ''}`",
+            f"- **status:** `{inc.status}`",
+            f"- **recurrence_basis:** `{reason}`",
+            "",
+            "### Latest summary",
+            "",
+            redact_text(inc.summary) or "(no summary)",
+            "",
+            "### Canonical next action",
+            "",
+            "Author: this incident is still firing — investigate under the "
+            "normal Gitea workflow. This comment records observability "
+            "recurrence only and changes no workflow state.",
+        ]
+    )
+    return "\n".join(lines)
+
+
 def _link_conflict(existing: dict[str, Any], inc: NormalizedIncident) -> str | None:
     """Fail closed if existing link targets a different Gitea issue/repo."""
     eg_org = str(existing.get("gitea_org") or "")
@@ -514,6 +586,9 @@ def _link_conflict(existing: dict[str, Any], inc: NormalizedIncident) -> str | N
 CreateIssueFn = Callable[[str, str, list[str], str, str], dict[str, Any]]
 # create_issue_fn(title, body, labels, gitea_org, gitea_repo) -> {"number": int, ...}
 
+CommentIssueFn = Callable[[int, str, str, str], dict[str, Any]]
+# comment_issue_fn(gitea_issue_number, body, gitea_org, gitea_repo) -> {"success": bool, ...}
+
 
 def reconcile_incident(
     db: ControlPlaneDB | None,
@@ -523,6 +598,7 @@ def reconcile_incident(
     mapping: ProjectMapping | None = None,
     apply: bool = False,
     create_issue_fn: CreateIssueFn | None = None,
+    comment_issue_fn: CommentIssueFn | None = None,
     force_gitea_issue_number: int | None = None,
 ) -> dict[str, Any]:
     """Reconcile one observation into incident_links + optional Gitea issue.
@@ -531,6 +607,11 @@ def reconcile_incident(
     *apply=True*: upsert link; create Gitea issue when none linked (requires
     ``create_issue_fn``) or use ``force_gitea_issue_number`` for explicit link.
 
+    When an existing link is reused and the provider reports *new* events,
+    ``comment_issue_fn`` posts a sanitized recurrence comment on the linked
+    Gitea issue (AC4). Dry runs never comment, and a missing
+    ``comment_issue_fn`` withholds the comment without failing the link.
+
     Never creates control-plane ``work_items`` for raw incidents.
     """
     base: dict[str, Any] = {
@@ -549,6 +630,7 @@ def reconcile_incident(
         "gitea_issue": None,
         "action": None,
         "mapping": None,
+        "recurrence_comment": None,
         "substrate": "control_plane_db.incident_links",
         "durable_work_system": "gitea_issues",
     }
@@ -652,10 +734,14 @@ def reconcile_incident(
     # --- apply path ---
     issue_number: int | None = None
     created = False
+    recurrence: tuple[bool, str] | None = None
     if existing:
         issue_number = int(existing["gitea_issue_number"])
         action = "updated_existing_link"
         outcome = OUTCOME_UPDATED
+        # Compare against the pre-upsert link row: the upsert below overwrites
+        # event_count/last_seen, which would erase the recurrence signal.
+        recurrence = incident_recurred(existing, inc)
     elif force_gitea_issue_number is not None:
         issue_number = int(force_gitea_issue_number)
         action = "link_explicit_issue"
@@ -729,6 +815,63 @@ def reconcile_incident(
         }
         return base
 
+    # AC4: continued provider events post a recurrence comment on the linked
+    # Gitea issue. The durable incident_links row is already written above, so
+    # a comment failure never rolls back or blocks the mapping — the next scan
+    # retries while the link stays authoritative.
+    if outcome == OUTCOME_UPDATED and recurrence is not None:
+        recurred, why = recurrence
+        if not recurred:
+            base["recurrence_comment"] = {"posted": False, "reason": why}
+        elif comment_issue_fn is None:
+            base["recurrence_comment"] = {
+                "posted": False,
+                "reason": (
+                    "no comment_issue_fn supplied; recurrence comment withheld "
+                    "(link remains durable)"
+                ),
+                "recurrence_basis": why,
+            }
+        else:
+            try:
+                comment_res = comment_issue_fn(
+                    issue_number,
+                    build_recurrence_comment_body(inc, existing, reason=why),
+                    inc.gitea_org,
+                    inc.gitea_repo,
+                )
+            except Exception as exc:  # noqa: BLE001 - never break the link write
+                base["recurrence_comment"] = {
+                    "posted": False,
+                    "reason": (
+                        f"recurrence comment failed: {redact_text(exc)} "
+                        "(link remains durable)"
+                    ),
+                    "recurrence_basis": why,
+                }
+            else:
+                posted = (
+                    bool(comment_res.get("success"))
+                    if isinstance(comment_res, dict)
+                    else bool(comment_res)
+                )
+                base["recurrence_comment"] = {
+                    "posted": posted,
+                    "recurrence_basis": why,
+                    "gitea_issue_number": issue_number,
+                    "comment_id": (
+                        comment_res.get("comment_id")
+                        if isinstance(comment_res, dict)
+                        else None
+                    ),
+                }
+                if posted:
+                    base["gitea_mutated"] = True
+                elif isinstance(comment_res, dict):
+                    base["recurrence_comment"]["reasons"] = [
+                        redact_text(r) for r in (comment_res.get("reasons") or [])
+                    ]
+
     base["success"] = True
     base["performed"] = True
     base["db_mutated"] = True

sentry_incident_bridge.py

@@ -0,0 +1,718 @@
+"""Sentry → Gitea incident bridge (#607).
+
+Reads unresolved issues/events from a **self-hosted** Sentry, normalizes them
+into #612 observations, and reconciles them into durable Gitea issues.
+
+Hard rules (inherited from #612 and restated here):
+* Gitea owns workflow state; Sentry is observability **input only**.
+* Raw Sentry incidents are never assignable control-plane ``work_items``.
+* Dedupe/link/create is delegated to :mod:`incident_bridge` — this module
+  never invents a second linking substrate.
+* Tokens, DSNs, and raw headers never appear in returns, bodies, or logs.
+* The watchdog defaults to dry-run; ``apply`` is explicit.
+
+Network access is injected as ``http_fn`` so the whole surface is testable
+without a live Sentry.
+"""
+
+from __future__ import annotations
+
+import dataclasses
+import json
+import os
+import re
+import urllib.error
+import urllib.parse
+import urllib.request
+from dataclasses import dataclass
+from typing import Any, Callable, Sequence
+
+import incident_bridge
+import sentry_observability
+
+PROVIDER = "sentry"
+
+ENV_BASE_URL = "SENTRY_BASE_URL"
+ENV_AUTH_TOKEN = "SENTRY_AUTH_TOKEN"
+ENV_ORG = "SENTRY_ORG"
+ENV_PROJECT = "SENTRY_PROJECT"
+ENV_ENVIRONMENT = "SENTRY_ENVIRONMENT"
+ENV_BRIDGE_ENABLED = "MCP_SENTRY_ISSUE_BRIDGE_ENABLED"
+ENV_MIN_EVENTS = "MCP_SENTRY_MIN_EVENTS_FOR_ISSUE"
+ENV_LOOKBACK = "MCP_SENTRY_LOOKBACK"
+
+DEFAULT_BASE_URL = "https://sentry.prgs.cc"
+DEFAULT_LOOKBACK = "24h"
+DEFAULT_MIN_EVENTS = 2
+DEFAULT_TIMEOUT = 15.0
+DEFAULT_PAGE_SIZE = 25
+MAX_PAGE_SIZE = 100
+DEFAULT_MAX_PAGES = 10
+
+_TRUTHY = frozenset({"1", "true", "yes", "on"})
+# Absolute local paths embedded in free text. Mirrors the shape matched by
+# sentry_observability's internal path detector; each hit is replaced by the
+# coarse category from sentry_observability.sanitize_path.
+_ABS_PATH_RE = re.compile(
+    r"(?:/private)?/(?:Users|home|tmp|var|opt|Volumes)/[^\s\"']*"
+)
+_LOOKBACK_RE = re.compile(r"^\d+[mhd]$")
+_CURSOR_RE = re.compile(r'cursor="([^"]+)"')
+_RESULTS_RE = re.compile(r'results="([^"]+)"')
+_REL_RE = re.compile(r'rel="([^"]+)"')
+
+# Error kinds surfaced to callers (stable strings; safe to branch on).
+ERROR_NOT_CONFIGURED = "not_configured"
+ERROR_MISSING_TOKEN = "missing_token"
+ERROR_UNAVAILABLE = "sentry_unavailable"
+ERROR_HTTP = "sentry_http_error"
+ERROR_INVALID_RESPONSE = "invalid_response"
+ERROR_BRIDGE_DISABLED = "bridge_disabled"
+
+# Watchdog per-issue dispositions.
+ACTION_RECONCILED = "reconciled"
+ACTION_SKIPPED_THRESHOLD = "skipped_below_event_threshold"
+ACTION_SKIPPED_STATUS = "skipped_not_unresolved"
+ACTION_FAILED = "failed"
+
+
+class SentryApiError(RuntimeError):
+    """Sentry read failure with a stable, redacted classification."""
+
+    def __init__(self, message: str, *, kind: str, status: int | None = None):
+        super().__init__(incident_bridge.redact_text(message))
+        self.kind = kind
+        self.status = status
+
+    def as_dict(self) -> dict[str, Any]:
+        return {
+            "error_kind": self.kind,
+            "status": self.status,
+            "message": str(self),
+        }
+
+
+@dataclass(frozen=True)
+class SentryBridgeConfig:
+    """Resolved bridge configuration. Never carries the auth token."""
+
+    base_url: str
+    org: str
+    project: str
+    environment: str | None = None
+    lookback: str = DEFAULT_LOOKBACK
+    min_events_for_issue: int = DEFAULT_MIN_EVENTS
+    bridge_enabled: bool = False
+    timeout: float = DEFAULT_TIMEOUT
+
+    def issues_path(self) -> str:
+        return f"/api/0/projects/{self.org}/{self.project}/issues/"
+
+    def issue_events_path(self, issue_id: str) -> str:
+        return f"/api/0/issues/{issue_id}/events/"
+
+    def as_dict(self) -> dict[str, Any]:
+        """Safe projection. The auth token is never included by construction."""
+        return {
+            "base_url": self.base_url,
+            "org": self.org,
+            "project": self.project,
+            "environment": self.environment,
+            "lookback": self.lookback,
+            "min_events_for_issue": self.min_events_for_issue,
+            "bridge_enabled": self.bridge_enabled,
+            "self_hosted": not self.base_url.rstrip("/").endswith("sentry.io"),
+        }
+
+
+def _env_bool(name: str, env: dict[str, str], default: bool = False) -> bool:
+    raw = (env.get(name) or "").strip().lower()
+    if not raw:
+        return default
+    return raw in _TRUTHY
+
+
+def _env_int(name: str, env: dict[str, str], default: int) -> int:
+    raw = (env.get(name) or "").strip()
+    if not raw:
+        return default
+    try:
+        value = int(raw)
+    except ValueError:
+        return default
+    return value if value >= 1 else default
+
+
+def load_bridge_config(env: dict[str, str] | None = None) -> SentryBridgeConfig:
+    """Build config from environment. Never reads or returns the token value."""
+    source = dict(env if env is not None else os.environ)
+    base_url = (source.get(ENV_BASE_URL) or DEFAULT_BASE_URL).strip().rstrip("/")
+    lookback = (source.get(ENV_LOOKBACK) or DEFAULT_LOOKBACK).strip()
+    if not _LOOKBACK_RE.match(lookback):
+        lookback = DEFAULT_LOOKBACK
+    environment = (source.get(ENV_ENVIRONMENT) or "").strip() or None
+    return SentryBridgeConfig(
+        base_url=base_url,
+        org=(source.get(ENV_ORG) or "").strip(),
+        project=(source.get(ENV_PROJECT) or "").strip(),
+        environment=environment,
+        lookback=lookback,
+        min_events_for_issue=_env_int(ENV_MIN_EVENTS, source, DEFAULT_MIN_EVENTS),
+        bridge_enabled=_env_bool(ENV_BRIDGE_ENABLED, source, False),
+    )
+
+
+def config_with_overrides(
+    config: SentryBridgeConfig,
+    *,
+    base_url: str | None = None,
+    org: str | None = None,
+    project: str | None = None,
+    lookback: str | None = None,
+    min_events_for_issue: int | None = None,
+) -> SentryBridgeConfig:
+    """Return *config* with explicit per-call overrides applied."""
+    overrides: dict[str, Any] = {}
+    if base_url:
+        overrides["base_url"] = str(base_url).strip().rstrip("/")
+    if org:
+        overrides["org"] = str(org).strip()
+    if project:
+        overrides["project"] = str(project).strip()
+    if lookback:
+        candidate = str(lookback).strip()
+        overrides["lookback"] = candidate if _LOOKBACK_RE.match(candidate) else config.lookback
+    if min_events_for_issue is not None:
+        overrides["min_events_for_issue"] = max(1, int(min_events_for_issue))
+    return dataclasses.replace(config, **overrides) if overrides else config
+
+
+def resolve_token(env: dict[str, str] | None = None) -> str:
+    """Return the Sentry auth token from env only (never logged or returned)."""
+    source = env if env is not None else os.environ
+    return (source.get(ENV_AUTH_TOKEN) or "").strip()
+
+
+def assert_configured(config: SentryBridgeConfig, token: str) -> None:
+    """Fail closed before any network call."""
+    missing = [
+        name
+        for name, value in (
+            (ENV_BASE_URL, config.base_url),
+            (ENV_ORG, config.org),
+            (ENV_PROJECT, config.project),
+        )
+        if not value
+    ]
+    if missing:
+        raise SentryApiError(
+            "Sentry bridge is not configured; missing " + ", ".join(sorted(missing)),
+            kind=ERROR_NOT_CONFIGURED,
+        )
+    if not token:
+        raise SentryApiError(
+            f"{ENV_AUTH_TOKEN} is not set; refusing to call Sentry (fail closed)",
+            kind=ERROR_MISSING_TOKEN,
+        )
+
+
+# --------------------------------------------------------------------------
+# HTTP layer (injectable)
+# --------------------------------------------------------------------------
+
+# http_fn(url, headers, timeout) -> (status, body_bytes, response_headers)
+HttpFn = Callable[[str, dict[str, str], float], "tuple[int, bytes, dict[str, str]]"]
+
+
+def _default_http_fn(
+    url: str, headers: dict[str, str], timeout: float
+) -> tuple[int, bytes, dict[str, str]]:
+    request = urllib.request.Request(url, headers=headers, method="GET")
+    try:
+        with urllib.request.urlopen(request, timeout=timeout) as response:
+            return (
+                int(response.status),
+                response.read(),
+                {k.lower(): v for k, v in response.headers.items()},
+            )
+    except urllib.error.HTTPError as exc:  # status is meaningful
+        try:
+            body = exc.read()
+        except Exception:  # noqa: BLE001 - body is best-effort only
+            body = b""
+        return (
+            int(exc.code),
+            body,
+            {k.lower(): v for k, v in (exc.headers or {}).items()},
+        )
+    except urllib.error.URLError as exc:
+        raise SentryApiError(
+            f"Sentry unreachable: {exc.reason}", kind=ERROR_UNAVAILABLE
+        ) from exc
+    except TimeoutError as exc:
+        raise SentryApiError("Sentry request timed out", kind=ERROR_UNAVAILABLE) from exc
+
+
+def parse_next_cursor(link_header: str | None) -> str | None:
+    """Extract the ``rel="next"`` cursor when more results exist."""
+    if not link_header:
+        return None
+    for part in link_header.split(","):
+        rel = _REL_RE.search(part)
+        if not rel or rel.group(1) != "next":
+            continue
+        results = _RESULTS_RE.search(part)
+        if results and results.group(1).lower() != "true":
+            return None
+        cursor = _CURSOR_RE.search(part)
+        if cursor:
+            return cursor.group(1)
+    return None
+
+
+def _get_json(
+    config: SentryBridgeConfig,
+    path: str,
+    params: dict[str, Any],
+    *,
+    token: str,
+    http_fn: HttpFn | None = None,
+) -> tuple[Any, dict[str, str]]:
+    caller = http_fn or _default_http_fn
+    query = urllib.parse.urlencode(
+        {k: v for k, v in params.items() if v not in (None, "")}
+    )
+    url = f"{config.base_url}{path}"
+    if query:
+        url = f"{url}?{query}"
+    headers = {
+        "Authorization": f"Bearer {token}",
+        "Accept": "application/json",
+        "User-Agent": "gitea-tools-sentry-bridge/1.0",
+    }
+    status, body, response_headers = caller(url, headers, config.timeout)
+    if status in (401, 403):
+        raise SentryApiError(
+            "Sentry rejected the auth token (unauthorized)",
+            kind=ERROR_MISSING_TOKEN,
+            status=status,
+        )
+    if status >= 500:
+        raise SentryApiError(
+            f"Sentry server error (HTTP {status})",
+            kind=ERROR_UNAVAILABLE,
+            status=status,
+        )
+    if status >= 400:
+        raise SentryApiError(
+            f"Sentry request failed (HTTP {status})", kind=ERROR_HTTP, status=status
+        )
+    try:
+        payload = json.loads(body.decode("utf-8") or "null")
+    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise SentryApiError(
+            f"Sentry returned an unparseable response: {exc}",
+            kind=ERROR_INVALID_RESPONSE,
+            status=status,
+        ) from exc
+    return payload, response_headers
+
+
+# --------------------------------------------------------------------------
+# Sanitization
+# --------------------------------------------------------------------------
+
+
+def _clean(value: Any) -> str:
+    """Redact secrets, then replace embedded local paths with a category token.
+
+    ``sentry_observability.sanitize_path`` categorizes a string that *is* a
+    path; it must never be applied to whole free-text fields (it would collapse
+    a title or timestamp to ``"other"``). Here it is applied only to substrings
+    that actually match an absolute path.
+    """
+    text = incident_bridge.redact_text(value)
+    if not text:
+        return ""
+    return _ABS_PATH_RE.sub(
+        lambda m: f"[path:{sentry_observability.sanitize_path(m.group(0))}]", text
+    )
+
+
+def sanitize_issue(raw: Any) -> dict[str, Any]:
+    """Project one raw Sentry issue into a sanitized, LLM-safe summary."""
+    if not isinstance(raw, dict):
+        raise SentryApiError(
+            "Sentry issue payload is not an object", kind=ERROR_INVALID_RESPONSE
+        )
+    issue_id = raw.get("id")
+    if issue_id is None or str(issue_id).strip() == "":
+        raise SentryApiError(
+            "Sentry issue payload is missing 'id'", kind=ERROR_INVALID_RESPONSE
+        )
+    metadata = raw.get("metadata") if isinstance(raw.get("metadata"), dict) else {}
+    try:
+        count = int(raw.get("count"))
+    except (TypeError, ValueError):
+        count = None
+    permalink = _clean(raw.get("permalink"))
+    if "[REDACTED]" in permalink:
+        permalink = ""
+    user_count = raw.get("userCount")
+    return {
+        "id": str(issue_id).strip(),
+        "short_id": _clean(raw.get("shortId")) or None,
+        "title": _clean(raw.get("title"))[:200],
+        "culprit": _clean(raw.get("culprit")) or None,
+        "level": _clean(raw.get("level")) or None,
+        "status": str(raw.get("status") or "unresolved").strip().lower() or "unresolved",
+        "count": count,
+        "user_count": user_count if isinstance(user_count, int) else None,
+        "first_seen": _clean(raw.get("firstSeen")) or None,
+        "last_seen": _clean(raw.get("lastSeen")) or None,
+        "permalink": permalink or None,
+        "metadata_value": _clean(metadata.get("value"))[:500] or None,
+        "metadata_type": _clean(metadata.get("type")) or None,
+    }
+
+
+def sanitize_event(raw: Any) -> dict[str, Any]:
+    """Project one raw Sentry event into a sanitized summary."""
+    if not isinstance(raw, dict):
+        raise SentryApiError(
+            "Sentry event payload is not an object", kind=ERROR_INVALID_RESPONSE
+        )
+    tags: dict[str, str] = {}
+    raw_tags = raw.get("tags")
+    if isinstance(raw_tags, list):
+        # Sentry events return tags as [{"key": ..., "value": ...}, ...]
+        tags = incident_bridge.sanitize_tags(
+            {
+                t.get("key"): t.get("value")
+                for t in raw_tags
+                if isinstance(t, dict) and t.get("key")
+            }
+        )
+    elif isinstance(raw_tags, dict):
+        tags = incident_bridge.sanitize_tags(raw_tags)
+    return {
+        "event_id": _clean(raw.get("eventID") or raw.get("id")) or None,
+        "message": _clean(raw.get("message") or raw.get("title"))[:2000] or None,
+        "date_created": _clean(raw.get("dateCreated")) or None,
+        "platform": _clean(raw.get("platform")) or None,
+        "environment": _clean(raw.get("environment")) or None,
+        "release": _clean(raw.get("release")) or None,
+        "tags": tags,
+    }
+
+
+# --------------------------------------------------------------------------
+# Reads
+# --------------------------------------------------------------------------
+
+
+def list_issues(
+    config: SentryBridgeConfig,
+    *,
+    token: str,
+    query: str = "is:unresolved",
+    limit: int = DEFAULT_PAGE_SIZE,
+    max_pages: int = DEFAULT_MAX_PAGES,
+    cursor: str | None = None,
+    environment: str | None = None,
+    http_fn: HttpFn | None = None,
+) -> dict[str, Any]:
+    """List sanitized unresolved Sentry issues, following ``Link`` pagination."""
+    assert_configured(config, token)
+    page_size = max(1, min(int(limit or DEFAULT_PAGE_SIZE), MAX_PAGE_SIZE))
+    pages_allowed = max(1, int(max_pages or 1))
+
+    issues: list[dict[str, Any]] = []
+    next_cursor = cursor
+    pages_fetched = 0
+    for _ in range(pages_allowed):
+        payload, headers = _get_json(
+            config,
+            config.issues_path(),
+            {
+                "query": query,
+                "statsPeriod": config.lookback,
+                "limit": page_size,
+                "cursor": next_cursor,
+                "environment": environment or config.environment,
+            },
+            token=token,
+            http_fn=http_fn,
+        )
+        pages_fetched += 1
+        if payload is None:
+            payload = []
+        if not isinstance(payload, list):
+            raise SentryApiError(
+                "Sentry issue list response was not a JSON array",
+                kind=ERROR_INVALID_RESPONSE,
+            )
+        issues.extend(sanitize_issue(item) for item in payload)
+        next_cursor = parse_next_cursor(headers.get("link"))
+        if not next_cursor:
+            break
+
+    return {
+        "success": True,
+        "issues": issues,
+        "count": len(issues),
+        "pages_fetched": pages_fetched,
+        "next_cursor": next_cursor,
+        "inventory_complete": next_cursor is None,
+        "config": config.as_dict(),
+        "query": query,
+    }
+
+
+def get_issue_events(
+    config: SentryBridgeConfig,
+    issue_id: str,
+    *,
+    token: str,
+    limit: int = 10,
+    http_fn: HttpFn | None = None,
+) -> dict[str, Any]:
+    """Fetch sanitized recent events plus the latest event for one issue."""
+    assert_configured(config, token)
+    if not str(issue_id or "").strip():
+        raise SentryApiError("issue_id is required", kind=ERROR_INVALID_RESPONSE)
+    issue_key = str(issue_id).strip()
+
+    payload, _ = _get_json(
+        config,
+        config.issue_events_path(issue_key),
+        {"limit": max(1, min(int(limit or 10), MAX_PAGE_SIZE))},
+        token=token,
+        http_fn=http_fn,
+    )
+    if payload is None:
+        payload = []
+    if not isinstance(payload, list):
+        raise SentryApiError(
+            "Sentry event list response was not a JSON array",
+            kind=ERROR_INVALID_RESPONSE,
+        )
+    events = [sanitize_event(item) for item in payload]
+    return {
+        "success": True,
+        "issue_id": issue_key,
+        "events": events,
+        "count": len(events),
+        "latest_event": events[0] if events else None,
+        "config": config.as_dict(),
+    }
+
+
+# --------------------------------------------------------------------------
+# Observation mapping + policy
+# --------------------------------------------------------------------------
+
+
+def observation_from_issue(
+    issue: dict[str, Any],
+    config: SentryBridgeConfig,
+    *,
+    gitea_org: str | None = None,
+    gitea_repo: str | None = None,
+    latest_event: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Convert a sanitized Sentry issue into a #612 observation dict."""
+    tags = dict(latest_event.get("tags") or {}) if isinstance(latest_event, dict) else {}
+    environment = None
+    if isinstance(latest_event, dict):
+        environment = latest_event.get("environment")
+    environment = environment or config.environment
+
+    observation: dict[str, Any] = {
+        "provider": PROVIDER,
+        "provider_base_url": config.base_url,
+        "provider_org": config.org,
+        "provider_project": config.project,
+        "provider_issue_id": issue.get("id"),
+        "provider_short_id": issue.get("short_id"),
+        "provider_permalink": issue.get("permalink"),
+        "title": issue.get("title"),
+        "culprit": issue.get("culprit"),
+        "summary": issue.get("metadata_value") or issue.get("title"),
+        "level": issue.get("level"),
+        "status": issue.get("status") or "unresolved",
+        "event_count": issue.get("count"),
+        "first_seen": issue.get("first_seen"),
+        "last_seen": issue.get("last_seen"),
+        "environment": environment,
+        "tags": tags,
+    }
+    if gitea_org:
+        observation["gitea_org"] = gitea_org
+    if gitea_repo:
+        observation["gitea_repo"] = gitea_repo
+    return observation
+
+
+def should_bridge_issue(
+    issue: dict[str, Any], config: SentryBridgeConfig
+) -> tuple[bool, str]:
+    """Policy gate: is this Sentry issue worth a durable Gitea issue?"""
+    status = str(issue.get("status") or "").strip().lower()
+    if status and status != "unresolved":
+        return False, f"status '{status}' is not unresolved"
+    count = issue.get("count")
+    threshold = int(config.min_events_for_issue or 1)
+    if isinstance(count, int) and count < threshold:
+        return (
+            False,
+            f"event count {count} below {ENV_MIN_EVENTS} threshold {threshold}",
+        )
+    return True, "meets bridge policy"
+
+
+# --------------------------------------------------------------------------
+# Watchdog
+# --------------------------------------------------------------------------
+
+
+def watchdog(
+    db: Any,
+    config: SentryBridgeConfig,
+    *,
+    token: str,
+    apply: bool = False,
+    mappings: Sequence[Any] | None = None,
+    gitea_org: str | None = None,
+    gitea_repo: str | None = None,
+    query: str = "is:unresolved",
+    limit: int = DEFAULT_PAGE_SIZE,
+    max_pages: int = DEFAULT_MAX_PAGES,
+    http_fn: HttpFn | None = None,
+    create_issue_fn: Any = None,
+    comment_issue_fn: Any = None,
+    reconcile_fn: Callable[..., dict[str, Any]] | None = None,
+    fetch_events: bool = True,
+) -> dict[str, Any]:
+    """Scan Sentry and reconcile active incidents into Gitea issues.
+
+    Dry-run by default. ``apply=True`` additionally requires the bridge to be
+    explicitly enabled via ``MCP_SENTRY_ISSUE_BRIDGE_ENABLED``.
+
+    ``comment_issue_fn`` carries the sanctioned issue-comment route used for
+    AC4 recurrence comments on already-linked issues; dry runs never comment.
+    """
+    result: dict[str, Any] = {
+        "success": False,
+        "apply": bool(apply),
+        "scanned": 0,
+        "reconciled": 0,
+        "skipped": 0,
+        "failed": 0,
+        "results": [],
+        "reasons": [],
+        "config": config.as_dict(),
+        "raw_incident_assignable": False,
+        "durable_work_system": "gitea_issues",
+    }
+
+    if apply and not config.bridge_enabled:
+        result["reasons"].append(
+            f"{ENV_BRIDGE_ENABLED} is not enabled; apply refused (fail closed)"
+        )
+        result["error_kind"] = ERROR_BRIDGE_DISABLED
+        return result
+
+    try:
+        listing = list_issues(
+            config,
+            token=token,
+            query=query,
+            limit=limit,
+            max_pages=max_pages,
+            http_fn=http_fn,
+        )
+    except SentryApiError as exc:
+        result["reasons"].append(str(exc))
+        result.update(exc.as_dict())
+        return result
+
+    reconciler = reconcile_fn or incident_bridge.reconcile_incident
+    result["inventory_complete"] = listing.get("inventory_complete", False)
+    result["pages_fetched"] = listing.get("pages_fetched", 0)
+
+    for issue in listing.get("issues", []):
+        result["scanned"] += 1
+        eligible, reason = should_bridge_issue(issue, config)
+        if not eligible:
+            result["skipped"] += 1
+            result["results"].append(
+                {
+                    "sentry_issue_id": issue.get("id"),
+                    "action": (
+                        ACTION_SKIPPED_THRESHOLD
+                        if "threshold" in reason
+                        else ACTION_SKIPPED_STATUS
+                    ),
+                    "reason": reason,
+                }
+            )
+            continue
+
+        latest_event = None
+        if fetch_events:
+            try:
+                events = get_issue_events(
+                    config, issue["id"], token=token, limit=1, http_fn=http_fn
+                )
+                latest_event = events.get("latest_event")
+            except SentryApiError as exc:
+                # Event enrichment is best-effort; the issue itself still bridges.
+                result["reasons"].append(
+                    f"event fetch failed for {issue.get('id')}: {exc}"
+                )
+
+        observation = observation_from_issue(
+            issue,
+            config,
+            gitea_org=gitea_org,
+            gitea_repo=gitea_repo,
+            latest_event=latest_event,
+        )
+        try:
+            reconciled = reconciler(
+                db,
+                observation=observation,
+                mappings=list(mappings or []),
+                apply=bool(apply),
+                create_issue_fn=create_issue_fn,
+                comment_issue_fn=comment_issue_fn,
+            )
+        except Exception as exc:  # noqa: BLE001 - one bad issue must not abort the scan
+            result["failed"] += 1
+            result["results"].append(
+                {
+                    "sentry_issue_id": issue.get("id"),
+                    "action": ACTION_FAILED,
+                    "reason": incident_bridge.redact_text(exc),
+                }
+            )
+            continue
+
+        result["reconciled"] += 1
+        result["results"].append(
+            {
+                "sentry_issue_id": issue.get("id"),
+                "action": ACTION_RECONCILED,
+                "outcome": reconciled.get("outcome"),
+                "gitea_issue": reconciled.get("gitea_issue"),
+                "existing_link": reconciled.get("existing_link"),
+                "recurrence_comment": reconciled.get("recurrence_comment"),
+                "reasons": reconciled.get("reasons"),
+            }
+        )
+
+    result["success"] = result["failed"] == 0
+    if not result["results"]:
+        result["reasons"].append("no Sentry issues matched the scan window/policy")
+    return result

tests/test_sentry_incident_bridge.py

@@ -0,0 +1,665 @@
+"""Tests for the Sentry → Gitea incident bridge (#607).
+
+Covers AC9: create, update, dedupe, closed-linked issue, redaction,
+pagination, missing token, unavailable Sentry server, and self-hosted base URL.
+
+No live Sentry: the HTTP layer is injected via ``http_fn``.
+"""
+
+from __future__ import annotations
+
+import sys as _sys
+from pathlib import Path as _Path
+
+_sys.path.insert(0, str(_Path(__file__).resolve().parent))
+from mutation_profile_fixture import shared_mutation_env  # noqa: E402
+
+import json
+import os
+import tempfile
+import unittest
+import unittest.mock
+import urllib.error
+import urllib.request
+
+from control_plane_db import ControlPlaneDB
+from incident_bridge import (
+    OUTCOME_CREATED,
+    OUTCOME_PREVIEW,
+    OUTCOME_UPDATED,
+    ProjectMapping,
+)
+
+import sentry_incident_bridge as bridge
+
+BASE_URL = "https://sentry.prgs.cc"
+SENTRY_ORG = "prgs"
+SENTRY_PROJECT = "gitea-tools-mcp"
+GITEA_ORG = "Scaled-Tech-Consulting"
+GITEA_REPO = "Gitea-Tools"
+TOKEN = "synthetic-test-token"
+
+
+def _config(**kwargs) -> bridge.SentryBridgeConfig:
+    base = dict(
+        base_url=BASE_URL,
+        org=SENTRY_ORG,
+        project=SENTRY_PROJECT,
+        lookback="24h",
+        min_events_for_issue=2,
+        bridge_enabled=True,
+    )
+    base.update(kwargs)
+    return bridge.SentryBridgeConfig(**base)
+
+
+def _mapping() -> ProjectMapping:
+    return ProjectMapping(
+        name="gitea-tools-mcp",
+        provider="sentry",
+        monitor_base_url=BASE_URL,
+        monitor_org=SENTRY_ORG,
+        monitor_project=SENTRY_PROJECT,
+        gitea_org=GITEA_ORG,
+        gitea_repo=GITEA_REPO,
+        default_labels=("type:bug", "observability", "sentry", "status:ready"),
+    )
+
+
+def _raw_issue(issue_id: str = "4001", **kwargs) -> dict:
+    payload = {
+        "id": issue_id,
+        "shortId": "GITEA-TOOLS-1A",
+        "title": "RuntimeError: lease acquisition failed",
+        "culprit": "lease_lifecycle in acquire",
+        "level": "error",
+        "status": "unresolved",
+        "count": "7",
+        "userCount": 1,
+        "firstSeen": "2026-07-18T04:11:02.000000Z",
+        "lastSeen": "2026-07-19T22:40:17.000000Z",
+        "permalink": f"{BASE_URL}/organizations/{SENTRY_ORG}/issues/{issue_id}/",
+        "metadata": {"type": "RuntimeError", "value": "lease acquisition failed"},
+    }
+    payload.update(kwargs)
+    return payload
+
+
+def _raw_event(event_id: str = "ev-1", **kwargs) -> dict:
+    payload = {
+        "eventID": event_id,
+        "message": "lease acquisition failed",
+        "dateCreated": "2026-07-19T22:40:17.000000Z",
+        "platform": "python",
+        "environment": "prod",
+        "release": "1.2.3",
+        "tags": [{"key": "role", "value": "author"}],
+    }
+    payload.update(kwargs)
+    return payload
+
+
+class FakeHttp:
+    """Routes synthetic Sentry responses and records requested URLs."""
+
+    def __init__(self, routes: list[tuple[int, object, dict[str, str]]] | None = None):
+        # routes: sequential responses for the issues endpoint
+        self.routes = routes or []
+        self.calls: list[str] = []
+        self.headers_seen: list[dict[str, str]] = []
+        self.issue_page = 0
+
+    def __call__(self, url, headers, timeout):
+        self.calls.append(url)
+        self.headers_seen.append(dict(headers))
+        if "/events/" in url:
+            return 200, json.dumps([_raw_event()]).encode(), {}
+        if self.routes:
+            index = min(self.issue_page, len(self.routes) - 1)
+            self.issue_page += 1
+            status, payload, resp_headers = self.routes[index]
+            body = payload if isinstance(payload, bytes) else json.dumps(payload).encode()
+            return status, body, resp_headers
+        return 200, json.dumps([_raw_issue()]).encode(), {}
+
+
+class SentryBridgeTestCase(unittest.TestCase):
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self._tmp.cleanup)
+        self.db_path = os.path.join(self._tmp.name, "cp.sqlite3")
+        self.db = ControlPlaneDB(self.db_path)
+        self.created: list[dict] = []
+        self.comments: list[dict] = []
+        self._next_issue_number = 900
+        self._next_comment_id = 5000
+
+    def _create_issue_fn(self):
+        def create_fn(title, body, labels, g_org, g_repo):
+            self._next_issue_number += 1
+            self.created.append(
+                {
+                    "title": title,
+                    "body": body,
+                    "labels": list(labels),
+                    "org": g_org,
+                    "repo": g_repo,
+                    "number": self._next_issue_number,
+                }
+            )
+            return {"success": True, "number": self._next_issue_number}
+
+        return create_fn
+
+    def _comment_issue_fn(self):
+        def comment_fn(issue_number, body, g_org, g_repo):
+            self._next_comment_id += 1
+            self.comments.append(
+                {
+                    "issue_number": issue_number,
+                    "body": body,
+                    "org": g_org,
+                    "repo": g_repo,
+                    "comment_id": self._next_comment_id,
+                }
+            )
+            return {"success": True, "comment_id": self._next_comment_id}
+
+        return comment_fn
+
+    def _link(self, issue_id: str = "4001"):
+        return self.db.get_incident_link_by_provider(
+            provider="sentry",
+            provider_issue_id=issue_id,
+            provider_base_url=BASE_URL,
+            provider_org=SENTRY_ORG,
+            provider_project=SENTRY_PROJECT,
+        )
+
+    def _watchdog(self, http, *, apply=True, config=None, **kwargs):
+        return bridge.watchdog(
+            self.db,
+            config or _config(),
+            token=TOKEN,
+            apply=apply,
+            mappings=[_mapping()],
+            http_fn=http,
+            create_issue_fn=self._create_issue_fn(),
+            # Always supplied, including dry runs: the bridge itself must
+            # withhold the comment when apply=False (AC4 + AC8).
+            comment_issue_fn=self._comment_issue_fn(),
+            **kwargs,
+        )
+
+
+class TestConfigAndSelfHosted(SentryBridgeTestCase):
+    def test_self_hosted_base_url_is_used_and_flagged(self):
+        config = _config()
+        self.assertTrue(config.as_dict()["self_hosted"])
+        http = FakeHttp()
+        bridge.list_issues(config, token=TOKEN, http_fn=http)
+        self.assertTrue(http.calls[0].startswith(f"{BASE_URL}/api/0/projects/"))
+        self.assertIn(f"/projects/{SENTRY_ORG}/{SENTRY_PROJECT}/issues/", http.calls[0])
+        self.assertIn("statsPeriod=24h", http.calls[0])
+
+    def test_config_never_exposes_token(self):
+        config = bridge.load_bridge_config(
+            {
+                bridge.ENV_BASE_URL: BASE_URL,
+                bridge.ENV_ORG: SENTRY_ORG,
+                bridge.ENV_PROJECT: SENTRY_PROJECT,
+                bridge.ENV_AUTH_TOKEN: "super-secret-value",
+            }
+        )
+        serialized = json.dumps(config.as_dict())
+        self.assertNotIn("super-secret-value", serialized)
+        self.assertNotIn("token", serialized.lower())
+
+    def test_invalid_lookback_falls_back_to_default(self):
+        config = bridge.load_bridge_config({bridge.ENV_LOOKBACK: "not-a-window"})
+        self.assertEqual(config.lookback, bridge.DEFAULT_LOOKBACK)
+
+
+class TestMissingTokenAndUnavailable(SentryBridgeTestCase):
+    def test_missing_token_fails_closed_without_http_call(self):
+        http = FakeHttp()
+        with self.assertRaises(bridge.SentryApiError) as ctx:
+            bridge.list_issues(_config(), token="", http_fn=http)
+        self.assertEqual(ctx.exception.kind, bridge.ERROR_MISSING_TOKEN)
+        self.assertEqual(http.calls, [], "no HTTP call may be made without a token")
+
+    def test_unconfigured_project_fails_closed(self):
+        with self.assertRaises(bridge.SentryApiError) as ctx:
+            bridge.list_issues(_config(project=""), token=TOKEN, http_fn=FakeHttp())
+        self.assertEqual(ctx.exception.kind, bridge.ERROR_NOT_CONFIGURED)
+
+    def test_unauthorized_status_maps_to_missing_token(self):
+        http = FakeHttp(routes=[(401, {"detail": "Invalid token"}, {})])
+        with self.assertRaises(bridge.SentryApiError) as ctx:
+            bridge.list_issues(_config(), token=TOKEN, http_fn=http)
+        self.assertEqual(ctx.exception.kind, bridge.ERROR_MISSING_TOKEN)
+
+    def test_server_error_maps_to_unavailable(self):
+        http = FakeHttp(routes=[(502, {"detail": "bad gateway"}, {})])
+        with self.assertRaises(bridge.SentryApiError) as ctx:
+            bridge.list_issues(_config(), token=TOKEN, http_fn=http)
+        self.assertEqual(ctx.exception.kind, bridge.ERROR_UNAVAILABLE)
+
+    def test_urlerror_from_default_handler_maps_to_unavailable(self):
+        """The real urllib handler must translate URLError, not leak it."""
+
+        def boom(request, timeout=None):
+            raise urllib.error.URLError("connection refused")
+
+        with unittest.mock.patch.object(urllib.request, "urlopen", boom):
+            with self.assertRaises(bridge.SentryApiError) as ctx:
+                bridge._default_http_fn("https://sentry.prgs.cc/api/0/x/", {}, 1.0)
+        self.assertEqual(ctx.exception.kind, bridge.ERROR_UNAVAILABLE)
+
+    def test_watchdog_reports_unavailable_without_mutating(self):
+        def failing(url, headers, timeout):
+            raise bridge.SentryApiError(
+                "Sentry unreachable", kind=bridge.ERROR_UNAVAILABLE
+            )
+
+        result = self._watchdog(failing)
+        self.assertFalse(result["success"])
+        self.assertEqual(result["error_kind"], bridge.ERROR_UNAVAILABLE)
+        self.assertEqual(self.created, [], "no Gitea issue on Sentry outage")
+
+    def test_invalid_json_fails_closed(self):
+        http = FakeHttp(routes=[(200, b"<html>not json</html>", {})])
+        with self.assertRaises(bridge.SentryApiError) as ctx:
+            bridge.list_issues(_config(), token=TOKEN, http_fn=http)
+        self.assertEqual(ctx.exception.kind, bridge.ERROR_INVALID_RESPONSE)
+
+
+class TestPagination(SentryBridgeTestCase):
+    def test_link_header_cursor_is_followed(self):
+        page1 = (
+            200,
+            [_raw_issue("4001")],
+            {
+                "link": (
+                    f'<{BASE_URL}/api/0/x/?cursor=c1>; rel="previous"; results="false", '
+                    f'<{BASE_URL}/api/0/x/?cursor=c2>; rel="next"; results="true"; cursor="c2"'
+                )
+            },
+        )
+        page2 = (
+            200,
+            [_raw_issue("4002")],
+            {
+                "link": (
+                    f'<{BASE_URL}/api/0/x/?cursor=c3>; rel="next"; '
+                    'results="false"; cursor="c3"'
+                )
+            },
+        )
+        http = FakeHttp(routes=[page1, page2])
+        result = bridge.list_issues(_config(), token=TOKEN, http_fn=http)
+        self.assertEqual(result["pages_fetched"], 2)
+        self.assertEqual([i["id"] for i in result["issues"]], ["4001", "4002"])
+        self.assertTrue(result["inventory_complete"])
+        self.assertIn("cursor=c2", http.calls[1])
+
+    def test_max_pages_caps_traversal_and_reports_incomplete(self):
+        page = (
+            200,
+            [_raw_issue("4001")],
+            {"link": f'<{BASE_URL}/x>; rel="next"; results="true"; cursor="cN"'},
+        )
+        http = FakeHttp(routes=[page])
+        result = bridge.list_issues(_config(), token=TOKEN, http_fn=http, max_pages=3)
+        self.assertEqual(result["pages_fetched"], 3)
+        self.assertFalse(result["inventory_complete"])
+
+    def test_parse_next_cursor_ignores_exhausted_results(self):
+        self.assertIsNone(
+            bridge.parse_next_cursor('<u>; rel="next"; results="false"; cursor="c"')
+        )
+        self.assertEqual(
+            bridge.parse_next_cursor('<u>; rel="next"; results="true"; cursor="c9"'),
+            "c9",
+        )
+        self.assertIsNone(bridge.parse_next_cursor(None))
+
+
+class TestRedaction(SentryBridgeTestCase):
+    def test_secrets_and_paths_are_scrubbed(self):
+        raw = _raw_issue(
+            title="RuntimeError: token=abc123supersecret failed",
+            culprit="/Users/jasonwalker/Development/Gitea-Tools/lease_lifecycle.py",
+            metadata={"type": "RuntimeError", "value": "password=hunter2"},
+        )
+        sanitized = bridge.sanitize_issue(raw)
+        blob = json.dumps(sanitized)
+        self.assertNotIn("abc123supersecret", blob)
+        self.assertNotIn("hunter2", blob)
+        self.assertNotIn("/Users/jasonwalker", blob)
+        self.assertIn("[REDACTED]", sanitized["title"])
+
+    def test_permalink_with_embedded_credentials_is_dropped(self):
+        raw = _raw_issue(permalink="https://user:[email protected]/issues/4001/")
+        self.assertIsNone(bridge.sanitize_issue(raw)["permalink"])
+
+    def test_sensitive_event_tags_are_removed(self):
+        event = _raw_event(
+            tags=[
+                {"key": "authorization", "value": "Bearer abc123secrettoken"},
+                {"key": "role", "value": "author"},
+            ]
+        )
+        sanitized = bridge.sanitize_event(event)
+        blob = json.dumps(sanitized)
+        self.assertNotIn("abc123secrettoken", blob)
+        self.assertEqual(sanitized["tags"].get("role"), "author")
+
+    def test_token_never_appears_in_watchdog_output(self):
+        result = self._watchdog(FakeHttp())
+        self.assertNotIn(TOKEN, json.dumps(result))
+
+    def test_issue_without_id_fails_closed(self):
+        with self.assertRaises(bridge.SentryApiError) as ctx:
+            bridge.sanitize_issue({"title": "no id"})
+        self.assertEqual(ctx.exception.kind, bridge.ERROR_INVALID_RESPONSE)
+
+
+class TestCreateUpdateDedupe(SentryBridgeTestCase):
+    def test_dry_run_creates_nothing(self):
+        result = self._watchdog(FakeHttp(), apply=False)
+        self.assertTrue(result["success"])
+        self.assertEqual(result["reconciled"], 1)
+        self.assertEqual(result["results"][0]["outcome"], OUTCOME_PREVIEW)
+        self.assertEqual(self.created, [], "dry-run must not create Gitea issues")
+
+    def test_apply_creates_one_durable_gitea_issue(self):
+        result = self._watchdog(FakeHttp())
+        self.assertTrue(result["success"])
+        self.assertEqual(result["results"][0]["outcome"], OUTCOME_CREATED)
+        self.assertEqual(len(self.created), 1)
+        created = self.created[0]
+        self.assertEqual(created["org"], GITEA_ORG)
+        self.assertEqual(created["repo"], GITEA_REPO)
+        self.assertIn("sentry", created["labels"])
+        # AC5: body carries the Sentry id and the first-seen window.
+        self.assertIn("4001", created["body"])
+        self.assertIn("2026-07-18T04:11:02", created["body"])
+
+    def test_repeat_scan_dedupes_to_a_single_issue(self):
+        first = self._watchdog(FakeHttp())
+        second = self._watchdog(FakeHttp())
+        self.assertEqual(first["results"][0]["outcome"], OUTCOME_CREATED)
+        self.assertEqual(second["results"][0]["outcome"], OUTCOME_UPDATED)
+        self.assertEqual(len(self.created), 1, "recurrence must not create a duplicate")
+
+    def test_recurrence_updates_link_event_count(self):
+        self._watchdog(FakeHttp())
+        recurring = FakeHttp(routes=[(200, [_raw_issue("4001", count="42")], {})])
+        result = self._watchdog(recurring)
+        self.assertEqual(result["results"][0]["outcome"], OUTCOME_UPDATED)
+        self.assertEqual(self._link()["event_count"], 42)
+
+    def test_recurrence_posts_a_comment_on_the_second_scan(self):
+        """AC4: continued Sentry events comment on the linked Gitea issue."""
+        first = self._watchdog(FakeHttp())
+        self.assertEqual(first["results"][0]["outcome"], OUTCOME_CREATED)
+        self.assertEqual(self.comments, [], "creation must not post a recurrence comment")
+
+        recurring = FakeHttp(routes=[(200, [_raw_issue("4001", count="42")], {})])
+        second = self._watchdog(recurring)
+
+        self.assertEqual(second["results"][0]["outcome"], OUTCOME_UPDATED)
+        self.assertEqual(len(self.comments), 1, "recurrence must post exactly one comment")
+        comment = self.comments[0]
+        linked_number = int(self._link()["gitea_issue_number"])
+        self.assertEqual(comment["issue_number"], linked_number)
+        self.assertEqual(comment["org"], GITEA_ORG)
+        self.assertEqual(comment["repo"], GITEA_REPO)
+        # AC5 fields carried on the recurrence record.
+        self.assertIn("4001", comment["body"])
+        self.assertIn("42", comment["body"])
+        self.assertIn("recurrence_basis", comment["body"])
+        reported = second["results"][0]["recurrence_comment"]
+        self.assertTrue(reported["posted"])
+        self.assertEqual(reported["comment_id"], comment["comment_id"])
+        self.assertEqual(len(self.created), 1, "recurrence must not create a duplicate issue")
+
+    def test_dry_run_scan_posts_no_recurrence_comment(self):
+        """AC4 + AC8: dry run never comments, even on a linked recurrence."""
+        self._watchdog(FakeHttp())
+        recurring = FakeHttp(routes=[(200, [_raw_issue("4001", count="42")], {})])
+        result = self._watchdog(recurring, apply=False)
+
+        self.assertEqual(result["results"][0]["outcome"], OUTCOME_PREVIEW)
+        self.assertEqual(self.comments, [], "dry-run must not post recurrence comments")
+
+    def test_repeat_scan_without_new_events_posts_no_comment(self):
+        """A scan that observes no new events must stay silent."""
+        self._watchdog(FakeHttp())
+        result = self._watchdog(FakeHttp())
+
+        self.assertEqual(result["results"][0]["outcome"], OUTCOME_UPDATED)
+        self.assertEqual(self.comments, [], "unchanged event state must not comment")
+        self.assertFalse(result["results"][0]["recurrence_comment"]["posted"])
+
+    def test_recurrence_comment_failure_keeps_the_link_durable(self):
+        """A failed comment must not roll back or block the incident_links row."""
+        self._watchdog(FakeHttp())
+
+        def failing_comment(issue_number, body, g_org, g_repo):
+            raise RuntimeError("gitea comment route unavailable")
+
+        recurring = FakeHttp(routes=[(200, [_raw_issue("4001", count="42")], {})])
+        result = bridge.watchdog(
+            self.db,
+            _config(),
+            token=TOKEN,
+            apply=True,
+            mappings=[_mapping()],
+            http_fn=recurring,
+            create_issue_fn=self._create_issue_fn(),
+            comment_issue_fn=failing_comment,
+        )
+
+        entry = result["results"][0]
+        self.assertEqual(entry["outcome"], OUTCOME_UPDATED)
+        self.assertFalse(entry["recurrence_comment"]["posted"])
+        self.assertEqual(self._link()["event_count"], 42, "link must still be updated")
+
+    def test_recurrence_comment_is_redacted(self):
+        """AC2: recurrence comments pass through the same redaction path."""
+        self._watchdog(FakeHttp())
+        recurring = FakeHttp(
+            routes=[
+                (
+                    200,
+                    [
+                        _raw_issue(
+                            "4001",
+                            count="42",
+                            metadata={
+                                "type": "RuntimeError",
+                                "value": "token=abc123supersecret",
+                            },
+                        )
+                    ],
+                    {},
+                )
+            ]
+        )
+        self._watchdog(recurring)
+
+        self.assertEqual(len(self.comments), 1)
+        body = self.comments[0]["body"]
+        self.assertNotIn("abc123supersecret", body)
+        self.assertNotIn(TOKEN, body)
+
+    def test_link_survives_a_new_db_handle(self):
+        """AC6: bridge mapping survives process restarts."""
+        self._watchdog(FakeHttp())
+        reopened = ControlPlaneDB(self.db_path)
+        link = reopened.get_incident_link_by_provider(
+            provider="sentry",
+            provider_issue_id="4001",
+            provider_base_url=BASE_URL,
+            provider_org=SENTRY_ORG,
+            provider_project=SENTRY_PROJECT,
+        )
+        self.assertIsNotNone(link)
+        self.assertEqual(int(link["gitea_issue_number"]), 901)
+
+    def test_resolved_issue_is_not_recreated_or_reopened(self):
+        """AC7: a resolved Sentry issue never creates or reopens Gitea work."""
+        self._watchdog(FakeHttp())
+        linked_number = int(self._link()["gitea_issue_number"])
+        closed = FakeHttp(routes=[(200, [_raw_issue("4001", status="resolved")], {})])
+        result = self._watchdog(closed)
+        self.assertEqual(result["skipped"], 1)
+        self.assertEqual(result["results"][0]["action"], bridge.ACTION_SKIPPED_STATUS)
+        self.assertEqual(len(self.created), 1)
+        self.assertEqual(linked_number, 901)
+
+
+class TestPolicyGates(SentryBridgeTestCase):
+    def test_below_threshold_issue_is_skipped(self):
+        http = FakeHttp(routes=[(200, [_raw_issue("4001", count="1")], {})])
+        result = self._watchdog(http)
+        self.assertEqual(result["skipped"], 1)
+        self.assertEqual(result["results"][0]["action"], bridge.ACTION_SKIPPED_THRESHOLD)
+        self.assertEqual(self.created, [])
+
+    def test_apply_refused_when_bridge_disabled(self):
+        result = self._watchdog(FakeHttp(), config=_config(bridge_enabled=False))
+        self.assertFalse(result["success"])
+        self.assertEqual(result["error_kind"], bridge.ERROR_BRIDGE_DISABLED)
+        self.assertEqual(self.created, [])
+
+    def test_dry_run_allowed_while_bridge_disabled(self):
+        result = self._watchdog(
+            FakeHttp(), apply=False, config=_config(bridge_enabled=False)
+        )
+        self.assertTrue(result["success"])
+
+    def test_raw_incident_is_never_assignable_work(self):
+        result = self._watchdog(FakeHttp())
+        self.assertFalse(result["raw_incident_assignable"])
+        self.assertEqual(result["durable_work_system"], "gitea_issues")
+
+    def test_reconcile_failure_is_isolated_and_redacted(self):
+        def exploding(db, **kwargs):
+            raise RuntimeError("token=abc123 boom")
+
+        result = self._watchdog(FakeHttp(), reconcile_fn=exploding)
+        self.assertFalse(result["success"])
+        self.assertEqual(result["failed"], 1)
+        self.assertNotIn("abc123", json.dumps(result))
+
+
+class TestObservationMapping(SentryBridgeTestCase):
+    def test_observation_carries_provider_identity_and_targets(self):
+        issue = bridge.sanitize_issue(_raw_issue())
+        event = bridge.sanitize_event(_raw_event())
+        obs = bridge.observation_from_issue(
+            issue,
+            _config(),
+            gitea_org=GITEA_ORG,
+            gitea_repo=GITEA_REPO,
+            latest_event=event,
+        )
+        self.assertEqual(obs["provider"], "sentry")
+        self.assertEqual(obs["provider_base_url"], BASE_URL)
+        self.assertEqual(obs["provider_issue_id"], "4001")
+        self.assertEqual(obs["event_count"], 7)
+        self.assertEqual(obs["environment"], "prod")
+        self.assertEqual(obs["gitea_repo"], GITEA_REPO)
+        self.assertTrue(_mapping().matches_observation(obs))
+
+    def test_events_fetch_returns_latest_first(self):
+        result = bridge.get_issue_events(
+            _config(), "4001", token=TOKEN, http_fn=FakeHttp()
+        )
+        self.assertEqual(result["count"], 1)
+        self.assertEqual(result["latest_event"]["environment"], "prod")
+
+
+class TestMcpToolWrappers(unittest.TestCase):
+    """The registered MCP tools must fail closed, never raise, never leak."""
+
+    def setUp(self):
+        # Read-capable profile, fully configured Sentry target, but
+        # deliberately no SENTRY_AUTH_TOKEN — the token gap is the only fault.
+        self.env = shared_mutation_env(
+            "test-author-prgs",
+            **{
+                bridge.ENV_BASE_URL: BASE_URL,
+                bridge.ENV_ORG: SENTRY_ORG,
+                bridge.ENV_PROJECT: SENTRY_PROJECT,
+            },
+        )
+        self.env.pop(bridge.ENV_AUTH_TOKEN, None)
+
+    def _server(self):
+        import gitea_mcp_server
+
+        return gitea_mcp_server
+
+    def test_all_five_tools_are_registered(self):
+        import asyncio
+
+        tools = asyncio.run(self._server().mcp.list_tools())
+        registered = {t.name for t in tools if t.name.startswith("gitea_sentry_")}
+        self.assertEqual(
+            registered,
+            {
+                "gitea_sentry_list_issues",
+                "gitea_sentry_get_issue_events",
+                "gitea_sentry_reconcile_issue",
+                "gitea_sentry_link_gitea_issue",
+                "gitea_sentry_watchdog",
+            },
+        )
+
+    def test_list_issues_without_token_fails_closed(self):
+        with unittest.mock.patch.dict(os.environ, self.env, clear=True):
+            result = self._server().gitea_sentry_list_issues()
+        self.assertFalse(result["success"])
+        self.assertEqual(result.get("error_kind"), bridge.ERROR_MISSING_TOKEN)
+        self.assertEqual(result["issues"], [])
+
+    def test_get_issue_events_without_token_fails_closed(self):
+        with unittest.mock.patch.dict(os.environ, self.env, clear=True):
+            result = self._server().gitea_sentry_get_issue_events("4001")
+        self.assertFalse(result["success"])
+        self.assertEqual(result.get("error_kind"), bridge.ERROR_MISSING_TOKEN)
+
+    def test_reconcile_without_token_fails_closed_without_mutation(self):
+        with unittest.mock.patch.dict(os.environ, self.env, clear=True):
+            result = self._server().gitea_sentry_reconcile_issue("4001", apply=True)
+        self.assertFalse(result["success"])
+        self.assertFalse(result["raw_incident_assignable"])
+        self.assertEqual(result.get("error_kind"), bridge.ERROR_MISSING_TOKEN)
+
+    def test_watchdog_without_token_fails_closed(self):
+        with unittest.mock.patch.dict(os.environ, self.env, clear=True):
+            result = self._server().gitea_sentry_watchdog()
+        self.assertFalse(result["success"])
+        self.assertEqual(result.get("error_kind"), bridge.ERROR_MISSING_TOKEN)
+
+    def test_unconfigured_target_reports_not_configured_before_token(self):
+        env = {k: v for k, v in self.env.items() if not k.startswith("SENTRY_")}
+        with unittest.mock.patch.dict(os.environ, env, clear=True):
+            result = self._server().gitea_sentry_list_issues()
+        self.assertFalse(result["success"])
+        self.assertEqual(result.get("error_kind"), bridge.ERROR_NOT_CONFIGURED)
+
+    def test_tool_output_never_contains_a_token_value(self):
+        env = dict(self.env)
+        env[bridge.ENV_AUTH_TOKEN] = "leaky-token-value"
+        with unittest.mock.patch.dict(os.environ, env, clear=True):
+            result = self._server().gitea_sentry_list_issues()
+        self.assertNotIn("leaky-token-value", json.dumps(result))
+
+
+if __name__ == "__main__":
+    unittest.main()