"""Observability incident bridge (#612).

Converts Sentry/GlitchTip **observations** into durable **Gitea issues** and
``incident_links`` rows on the #613 control-plane DB.

Hard rules (ADR):
* Gitea owns work; providers own incidents; the bridge only links them.
* Raw monitoring incidents are **never** assignable ``work_items``.
* The #600 allocator sees bridge work only as normal Gitea issues.
* Phase-1 prefers dry-run / explicit reconcile; apply creates/links issues.
* No tokens, DSNs, cookies, or other secrets in bodies, DB, or logs.
"""

from __future__ import annotations

import json
import os
import re
from dataclasses import dataclass, field
from typing import Any, Callable, Sequence

from control_plane_db import ControlPlaneDB, ControlPlaneError, WORK_KINDS, _norm_scope

PROVIDERS = frozenset({"sentry", "glitchtip"})

OUTCOME_PREVIEW = "preview"
OUTCOME_LINKED = "linked_existing"
OUTCOME_CREATED = "created_issue"
OUTCOME_UPDATED = "updated_link"
OUTCOME_BLOCKED = "blocked"
OUTCOME_NO_ACTION = "no_safe_action"

# Patterns scrubbed from any bridge-facing text (bodies, titles, tags, logs).
_SECRET_PATTERNS: tuple[re.Pattern[str], ...] = (
    re.compile(
        r"(?i)\b(api[_-]?token|token|secret|password|passwd)\s*[:=]\s*\S+"
    ),
    re.compile(
        r"(?i)\bAuthorization\s*[:=]\s*(?:Bearer\s+)?[A-Za-z0-9._\-+=/]{8,}"
    ),
    re.compile(r"(?i)\bBearer\s+[A-Za-z0-9._\-]{8,}"),
    re.compile(r"(?i)\bBasic\s+[A-Za-z0-9+/=]{8,}"),
    re.compile(r"(?i)\b(dsn|sentry_dsn|glitchtip_dsn)\s*[:=]\s*\S+"),
    re.compile(r"https?://[^/\s]+:[^@/\s]+@"),  # user:pass@host
    re.compile(r"(?i)\b(keychain[_-]?id|private[_-]?key)\s*[:=]\s*\S+"),
    re.compile(r"(?i)cookie\s*[:=]\s*[^\s;]+"),
    re.compile(r"(?i)\bsession[_-]?id\s*[:=]\s*\S+"),
)

_SENSITIVE_TAG_KEYS = frozenset(
    {
        "authorization",
        "cookie",
        "set-cookie",
        "password",
        "passwd",
        "secret",
        "token",
        "api_key",
        "apikey",
        "dsn",
        "sentry_dsn",
        "access_token",
        "refresh_token",
    }
)

DEFAULT_LABELS = (
    "type:bug",
    "observability",
    "status:ready",
)


@dataclass(frozen=True)
class ProjectMapping:
    """Maps a monitoring project to a Gitea repository."""

    name: str
    provider: str
    monitor_base_url: str
    monitor_org: str
    monitor_project: str
    gitea_org: str
    gitea_repo: str
    default_labels: tuple[str, ...] = DEFAULT_LABELS
    environment_filters: tuple[str, ...] = ()
    severity_threshold: str | None = None

    def __post_init__(self) -> None:
        prov = (self.provider or "").strip().lower()
        if prov not in PROVIDERS:
            raise ControlPlaneError(
                f"unknown provider '{self.provider}'; expected one of {sorted(PROVIDERS)}"
            )
        object.__setattr__(self, "provider", prov)
        object.__setattr__(self, "monitor_base_url", (self.monitor_base_url or "").rstrip("/"))
        object.__setattr__(self, "monitor_org", (self.monitor_org or "").strip())
        object.__setattr__(self, "monitor_project", (self.monitor_project or "").strip())
        object.__setattr__(self, "gitea_org", (self.gitea_org or "").strip())
        object.__setattr__(self, "gitea_repo", (self.gitea_repo or "").strip())
        object.__setattr__(
            self,
            "default_labels",
            tuple(str(x).strip() for x in (self.default_labels or DEFAULT_LABELS) if str(x).strip()),
        )

    def as_dict(self) -> dict[str, Any]:
        return {
            "name": self.name,
            "provider": self.provider,
            "monitor_base_url": self.monitor_base_url,
            "monitor_org": self.monitor_org,
            "monitor_project": self.monitor_project,
            "gitea_org": self.gitea_org,
            "gitea_repo": self.gitea_repo,
            "default_labels": list(self.default_labels),
            "environment_filters": list(self.environment_filters),
            "severity_threshold": self.severity_threshold,
        }

    def matches_observation(self, obs: dict[str, Any]) -> bool:
        if (obs.get("provider") or "").strip().lower() != self.provider:
            return False
        base = (obs.get("provider_base_url") or obs.get("monitor_base_url") or "").rstrip("/")
        if base and self.monitor_base_url and base != self.monitor_base_url:
            return False
        org = (obs.get("provider_org") or obs.get("monitor_org") or "").strip()
        if org and self.monitor_org and org != self.monitor_org:
            return False
        project = (obs.get("provider_project") or obs.get("monitor_project") or "").strip()
        if project and self.monitor_project and project != self.monitor_project:
            return False
        return True


@dataclass
class NormalizedIncident:
    provider: str
    provider_base_url: str
    provider_org: str
    provider_project: str
    provider_issue_id: str
    provider_short_id: str | None = None
    provider_permalink: str | None = None
    fingerprint: str | None = None
    first_seen: str | None = None
    last_seen: str | None = None
    event_count: int | None = None
    status: str = "open"
    environment: str | None = None
    severity: str | None = None
    culprit: str | None = None
    title: str = ""
    summary: str = ""
    tags: dict[str, str] = field(default_factory=dict)
    gitea_org: str = ""
    gitea_repo: str = ""
    default_labels: tuple[str, ...] = DEFAULT_LABELS

    def provider_key(self) -> tuple[str, str, str, str, str]:
        return (
            self.provider,
            _norm_scope(self.provider_base_url),
            _norm_scope(self.provider_org),
            _norm_scope(self.provider_project),
            str(self.provider_issue_id),
        )

    def as_dict(self) -> dict[str, Any]:
        return {
            "provider": self.provider,
            "provider_base_url": self.provider_base_url,
            "provider_org": self.provider_org,
            "provider_project": self.provider_project,
            "provider_issue_id": self.provider_issue_id,
            "provider_short_id": self.provider_short_id,
            "provider_permalink": self.provider_permalink,
            "fingerprint": self.fingerprint,
            "first_seen": self.first_seen,
            "last_seen": self.last_seen,
            "event_count": self.event_count,
            "status": self.status,
            "environment": self.environment,
            "severity": self.severity,
            "culprit": self.culprit,
            "title": self.title,
            "summary": self.summary,
            "tags": dict(self.tags),
            "gitea_org": self.gitea_org,
            "gitea_repo": self.gitea_repo,
            "default_labels": list(self.default_labels),
        }


def redact_text(value: Any) -> str:
    """Redact secret-like material from a string (fail closed to empty)."""
    if value is None:
        return ""
    text = str(value)
    for pat in _SECRET_PATTERNS:
        text = pat.sub("[REDACTED]", text)
    return text


def sanitize_tags(tags: Any) -> dict[str, str]:
    if not isinstance(tags, dict):
        return {}
    out: dict[str, str] = {}
    for k, v in tags.items():
        key = str(k).strip().lower()
        if not key or key in _SENSITIVE_TAG_KEYS:
            continue
        if any(s in key for s in ("token", "secret", "password", "cookie", "auth", "dsn")):
            continue
        val = redact_text(v)
        if "[REDACTED]" in val:
            continue
        if len(val) > 200:
            val = val[:200] + "…"
        out[key] = val
    return out


def project_mapping_from_dict(data: dict[str, Any]) -> ProjectMapping:
    labels = data.get("default_labels") or DEFAULT_LABELS
    envs = data.get("environment_filters") or ()
    return ProjectMapping(
        name=str(data.get("name") or data.get("monitor_project") or "unnamed"),
        provider=str(data.get("provider") or ""),
        monitor_base_url=str(data.get("monitor_base_url") or data.get("provider_base_url") or ""),
        monitor_org=str(data.get("monitor_org") or data.get("provider_org") or ""),
        monitor_project=str(data.get("monitor_project") or data.get("provider_project") or ""),
        gitea_org=str(data.get("gitea_org") or ""),
        gitea_repo=str(data.get("gitea_repo") or ""),
        default_labels=tuple(labels),
        environment_filters=tuple(envs),
        severity_threshold=data.get("severity_threshold"),
    )


def load_project_mappings(
    *,
    config_path: str | None = None,
    mappings_json: str | None = None,
) -> list[ProjectMapping]:
    """Load project mappings from JSON env, file, or explicit JSON string.

    Env: ``GITEA_OBSERVABILITY_PROJECTS_JSON`` or path
    ``GITEA_OBSERVABILITY_PROJECTS_FILE``.
    """
    raw: Any = None
    if mappings_json:
        raw = json.loads(mappings_json)
    else:
        env_json = (os.environ.get("GITEA_OBSERVABILITY_PROJECTS_JSON") or "").strip()
        path = (
            (config_path or "").strip()
            or (os.environ.get("GITEA_OBSERVABILITY_PROJECTS_FILE") or "").strip()
        )
        if env_json:
            raw = json.loads(env_json)
        elif path and os.path.isfile(path):
            with open(path, encoding="utf-8") as fh:
                raw = json.load(fh)
    if raw is None:
        return []
    if isinstance(raw, dict) and "projects" in raw:
        raw = raw["projects"]
    if not isinstance(raw, list):
        raise ControlPlaneError("observability project mappings must be a list")
    return [project_mapping_from_dict(item) for item in raw if isinstance(item, dict)]


def resolve_mapping(
    observation: dict[str, Any],
    mappings: Sequence[ProjectMapping],
    *,
    explicit: ProjectMapping | None = None,
) -> ProjectMapping | None:
    if explicit is not None:
        return explicit
    matches = [m for m in mappings if m.matches_observation(observation)]
    if len(matches) == 1:
        return matches[0]
    if len(matches) > 1:
        raise ControlPlaneError(
            "ambiguous project mapping for observation: "
            + ", ".join(m.name for m in matches)
            + " (fail closed, #612)"
        )
    # Fallback: observation itself may carry gitea targets
    g_org = (observation.get("gitea_org") or "").strip()
    g_repo = (observation.get("gitea_repo") or "").strip()
    provider = (observation.get("provider") or "").strip().lower()
    if g_org and g_repo and provider in PROVIDERS:
        return ProjectMapping(
            name=f"inline-{provider}",
            provider=provider,
            monitor_base_url=str(
                observation.get("provider_base_url")
                or observation.get("monitor_base_url")
                or ""
            ),
            monitor_org=str(
                observation.get("provider_org") or observation.get("monitor_org") or ""
            ),
            monitor_project=str(
                observation.get("provider_project")
                or observation.get("monitor_project")
                or ""
            ),
            gitea_org=g_org,
            gitea_repo=g_repo,
            default_labels=tuple(observation.get("default_labels") or DEFAULT_LABELS),
        )
    return None


def normalize_incident(
    observation: dict[str, Any],
    mapping: ProjectMapping,
) -> NormalizedIncident:
    """Normalize + sanitize a raw provider observation (fail closed)."""
    if not isinstance(observation, dict):
        raise ControlPlaneError("observation must be a dict (fail closed, #612)")

    provider = (observation.get("provider") or mapping.provider or "").strip().lower()
    if provider not in PROVIDERS:
        raise ControlPlaneError(
            f"unknown provider '{provider}'; expected {sorted(PROVIDERS)} (fail closed)"
        )
    if provider != mapping.provider:
        raise ControlPlaneError(
            f"observation provider '{provider}' does not match mapping "
            f"'{mapping.provider}' (fail closed, #612)"
        )

    issue_id = observation.get("provider_issue_id") or observation.get("id")
    if issue_id is None or str(issue_id).strip() == "":
        raise ControlPlaneError(
            "observation missing provider_issue_id (fail closed, #612)"
        )

    base = (
        observation.get("provider_base_url")
        or observation.get("monitor_base_url")
        or mapping.monitor_base_url
        or ""
    )
    org = (
        observation.get("provider_org")
        or observation.get("monitor_org")
        or mapping.monitor_org
        or ""
    )
    project = (
        observation.get("provider_project")
        or observation.get("monitor_project")
        or mapping.monitor_project
        or ""
    )

    title = redact_text(
        observation.get("title")
        or observation.get("culprit")
        or f"{provider} incident {issue_id}"
    )
    meta = observation.get("metadata")
    meta_value = meta.get("value") if isinstance(meta, dict) else None
    raw_sum = (
        observation.get("summary")
        or observation.get("message")
        or meta_value
        or title
    )
    summary = redact_text(raw_sum)

    permalink = observation.get("provider_permalink") or observation.get("permalink")
    if permalink:
        permalink = redact_text(permalink)
        if "[REDACTED]" in permalink:
            permalink = None

    tags = sanitize_tags(observation.get("tags") or {})
    event_count = observation.get("event_count") or observation.get("count")
    try:
        event_count_i = int(event_count) if event_count is not None else None
    except (TypeError, ValueError):
        event_count_i = None

    return NormalizedIncident(
        provider=provider,
        provider_base_url=str(base).rstrip("/"),
        provider_org=str(org).strip(),
        provider_project=str(project).strip(),
        provider_issue_id=str(issue_id).strip(),
        provider_short_id=redact_text(observation.get("provider_short_id") or observation.get("shortId") or "")
        or None,
        provider_permalink=permalink,
        fingerprint=redact_text(observation.get("fingerprint") or "") or None,
        first_seen=str(observation.get("first_seen") or observation.get("firstSeen") or "")
        or None,
        last_seen=str(observation.get("last_seen") or observation.get("lastSeen") or "")
        or None,
        event_count=event_count_i,
        status=str(observation.get("status") or "open").strip().lower() or "open",
        environment=redact_text(observation.get("environment") or "") or None,
        severity=redact_text(observation.get("severity") or observation.get("level") or "")
        or None,
        culprit=redact_text(observation.get("culprit") or "") or None,
        title=title[:200],
        summary=summary[:2000],
        tags=tags,
        gitea_org=mapping.gitea_org,
        gitea_repo=mapping.gitea_repo,
        default_labels=mapping.default_labels,
    )


def build_gitea_issue_title(inc: NormalizedIncident) -> str:
    env = f" [{inc.environment}]" if inc.environment else ""
    sev = f" ({inc.severity})" if inc.severity else ""
    return redact_text(f"[obs:{inc.provider}]{env}{sev} {inc.title}")[:180]


def build_gitea_issue_body(inc: NormalizedIncident) -> str:
    """Sanitized durable issue body (no secrets)."""
    lines = [
        "## Observability incident (bridge #612)",
        "",
        "<!-- mcp-incident-bridge:v1 -->",
        f"<!-- provider={inc.provider} issue_id={inc.provider_issue_id} -->",
        "",
        f"- **provider:** `{inc.provider}`",
        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"- **provider_org/project:** `{inc.provider_org}` / `{inc.provider_project}`",
            f"- **base_url:** `{inc.provider_base_url}`",
            f"- **fingerprint:** `{inc.fingerprint or ''}`",
            f"- **first_seen:** `{inc.first_seen or ''}`",
            f"- **last_seen:** `{inc.last_seen or ''}`",
            f"- **event_count:** `{inc.event_count if inc.event_count is not None else ''}`",
            f"- **environment:** `{inc.environment or ''}`",
            f"- **severity:** `{inc.severity or ''}`",
            f"- **culprit:** `{inc.culprit or ''}`",
            f"- **status:** `{inc.status}`",
            "",
            "### Summary",
            "",
            redact_text(inc.summary) or "(no summary)",
            "",
            "### Safe tags",
            "",
        ]
    )
    if inc.tags:
        for k, v in sorted(inc.tags.items()):
            lines.append(f"- `{k}`: `{v}`")
    else:
        lines.append("- (none)")
    lines.extend(
        [
            "",
            "### Canonical next action",
            "",
            "Author: investigate and fix under normal Gitea workflow. "
            "Allocator may select this issue as ordinary Gitea work "
            "(never as a raw provider incident).",
            "",
            "### Bridge notes",
            "",
            "- Raw Sentry/GlitchTip incidents are **not** assignable work items.",
            "- Gitea remains the durable work record; DB stores `incident_links` only.",
            "- Provider tokens must never appear in this issue.",
        ]
    )
    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 "")
    eg_repo = str(existing.get("gitea_repo") or "")
    eg_num = existing.get("gitea_issue_number")
    if eg_org and eg_org != inc.gitea_org:
        return (
            f"existing link points to org '{eg_org}', mapping wants "
            f"'{inc.gitea_org}' (fail closed, #612)"
        )
    if eg_repo and eg_repo != inc.gitea_repo:
        return (
            f"existing link points to repo '{eg_repo}', mapping wants "
            f"'{inc.gitea_repo}' (fail closed, #612)"
        )
    # fingerprint conflict when both present and disagree
    ef = (existing.get("fingerprint") or "").strip()
    nf = (inc.fingerprint or "").strip()
    if ef and nf and ef != nf:
        # Same provider issue id with different fingerprints is ambiguous.
        return (
            f"fingerprint conflict for provider issue {inc.provider_issue_id}: "
            f"link has '{ef}', observation has '{nf}' (fail closed, #612)"
        )
    return None


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,
    *,
    observation: dict[str, Any],
    mappings: Sequence[ProjectMapping] | None = None,
    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.

    *apply=False* (default): preview only — no DB mutation, no Gitea mutation.
    *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] = {
        "success": False,
        "apply": bool(apply),
        "outcome": OUTCOME_NO_ACTION,
        "performed": False,
        "gitea_mutated": False,
        "db_mutated": False,
        "raw_incident_assignable": False,
        "work_item_kind_would_be": None,
        "reasons": [],
        "skipped": None,
        "incident": None,
        "existing_link": None,
        "gitea_issue": None,
        "action": None,
        "mapping": None,
        "recurrence_comment": None,
        "substrate": "control_plane_db.incident_links",
        "durable_work_system": "gitea_issues",
    }

    if db is None:
        base["reasons"] = [
            "control-plane DB substrate unavailable (fail closed, #613/#612)"
        ]
        base["outcome"] = OUTCOME_BLOCKED
        return base

    try:
        maps = list(mappings or [])
        resolved = resolve_mapping(observation, maps, explicit=mapping)
        if resolved is None:
            base["outcome"] = OUTCOME_BLOCKED
            base["reasons"] = [
                "no project mapping for observation (fail closed, #612); "
                "configure GITEA_OBSERVABILITY_PROJECTS_JSON or pass mapping"
            ]
            return base
        base["mapping"] = resolved.as_dict()
        inc = normalize_incident(observation, resolved)
        base["incident"] = inc.as_dict()
    except (ControlPlaneError, json.JSONDecodeError, TypeError, ValueError) as exc:
        base["outcome"] = OUTCOME_BLOCKED
        base["reasons"] = [str(exc)]
        return base

    # Environment filter (optional)
    if resolved.environment_filters and inc.environment:
        allowed = {e.lower() for e in resolved.environment_filters}
        if inc.environment.lower() not in allowed:
            base["outcome"] = OUTCOME_NO_ACTION
            base["reasons"] = [
                f"environment '{inc.environment}' not in filters "
                f"{sorted(allowed)}; skip (policy)"
            ]
            base["success"] = True
            base["action"] = "skip_environment_filter"
            return base

    existing = db.get_incident_link_by_provider(
        provider=inc.provider,
        provider_issue_id=inc.provider_issue_id,
        provider_base_url=inc.provider_base_url,
        provider_org=inc.provider_org,
        provider_project=inc.provider_project,
    )
    base["existing_link"] = existing

    if existing:
        conflict = _link_conflict(existing, inc)
        if conflict:
            base["outcome"] = OUTCOME_BLOCKED
            base["reasons"] = [conflict]
            return base

    title = build_gitea_issue_title(inc)
    body = build_gitea_issue_body(inc)
    labels = list(inc.default_labels)
    if inc.provider and f"{inc.provider}" not in labels:
        labels.append(inc.provider)
    if "observability" not in labels:
        labels.append("observability")

    preview_issue = {
        "title": title,
        "body_preview": body[:500] + ("…" if len(body) > 500 else ""),
        "labels": labels,
        "gitea_org": inc.gitea_org,
        "gitea_repo": inc.gitea_repo,
        "would_create": existing is None and force_gitea_issue_number is None,
        "would_reuse_issue": int(existing["gitea_issue_number"])
        if existing
        else force_gitea_issue_number,
    }
    base["gitea_issue_preview"] = preview_issue

    if not apply:
        base["success"] = True
        base["outcome"] = OUTCOME_PREVIEW
        base["action"] = (
            "preview_reuse_link" if existing else "preview_create_issue"
        )
        base["reasons"] = [
            "dry-run only (apply=false); no Gitea mutation and no DB write — "
            "call again with apply=true to create/link"
        ]
        if existing:
            base["gitea_issue"] = {
                "number": existing.get("gitea_issue_number"),
                "org": existing.get("gitea_org"),
                "repo": existing.get("gitea_repo"),
            }
        # Prove we never treat this as work_item kind incident
        base["raw_incident_assignable"] = False
        base["work_item_kind_would_be"] = "issue"  # only after Gitea issue exists
        return base

    # --- 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"
        outcome = OUTCOME_LINKED
    else:
        if create_issue_fn is None:
            base["outcome"] = OUTCOME_BLOCKED
            base["reasons"] = [
                "apply=true requires create_issue_fn when no existing link "
                "(fail closed, #612)"
            ]
            return base
        try:
            created_res = create_issue_fn(
                title, body, labels, inc.gitea_org, inc.gitea_repo
            )
        except Exception as exc:  # noqa: BLE001
            base["outcome"] = OUTCOME_BLOCKED
            base["reasons"] = [
                f"Gitea issue creation failed: {redact_text(exc)} (fail closed)"
            ]
            return base
        number = created_res.get("number") if isinstance(created_res, dict) else None
        if number is None:
            base["outcome"] = OUTCOME_BLOCKED
            base["reasons"] = [
                "Gitea issue creation returned no issue number (fail closed, #612)"
            ]
            base["create_result"] = created_res
            return base
        issue_number = int(number)
        created = True
        action = "created_gitea_issue"
        outcome = OUTCOME_CREATED
        base["gitea_mutated"] = True
        base["create_result"] = {
            k: created_res.get(k)
            for k in ("number", "success", "performed", "reasons")
            if isinstance(created_res, dict)
        }

    assert issue_number is not None
    try:
        link = db.upsert_incident_link(
            provider=inc.provider,
            provider_issue_id=inc.provider_issue_id,
            gitea_org=inc.gitea_org,
            gitea_repo=inc.gitea_repo,
            gitea_issue_number=issue_number,
            provider_base_url=inc.provider_base_url,
            provider_org=inc.provider_org,
            provider_project=inc.provider_project,
            provider_short_id=inc.provider_short_id,
            provider_permalink=inc.provider_permalink,
            fingerprint=inc.fingerprint,
            first_seen=inc.first_seen,
            last_seen=inc.last_seen,
            event_count=inc.event_count,
            status=inc.status if not created else "open",
        )
    except Exception as exc:  # noqa: BLE001
        base["outcome"] = OUTCOME_BLOCKED
        base["reasons"] = [
            f"incident_links upsert failed: {redact_text(exc)} (fail closed, #613)"
        ]
        base["gitea_issue"] = {
            "number": issue_number,
            "org": inc.gitea_org,
            "repo": inc.gitea_repo,
            "created": created,
        }
        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
    base["outcome"] = outcome
    base["action"] = action
    base["gitea_issue"] = {
        "number": issue_number,
        "org": inc.gitea_org,
        "repo": inc.gitea_repo,
        "created": created,
    }
    base["link"] = {
        k: link.get(k)
        for k in (
            "link_id",
            "provider",
            "provider_issue_id",
            "gitea_org",
            "gitea_repo",
            "gitea_issue_number",
            "fingerprint",
            "event_count",
            "status",
            "last_sync_at",
        )
    }
    base["reasons"] = [
        (
            f"reused Gitea issue #{issue_number} for provider "
            f"{inc.provider}:{inc.provider_issue_id}"
            if not created
            else f"created Gitea issue #{issue_number} and stored incident_links row"
        ),
        "raw provider incident is not an assignable work_item "
        f"(WORK_KINDS={sorted(WORK_KINDS)})",
    ]
    base["raw_incident_assignable"] = False
    base["work_item_kind_would_be"] = "issue"
    base["allocator_visible_as"] = {
        "kind": "issue",
        "number": issue_number,
        "org": inc.gitea_org,
        "repo": inc.gitea_repo,
        "note": "allocator selects normal Gitea issues only (#600/#612)",
    }
    return base


def assert_not_raw_incident_work_item(kind: str) -> None:
    """Guard used by tests / callers: incidents must not enter WORK_KINDS."""
    k = (kind or "").strip().lower()
    if k not in WORK_KINDS:
        if k in ("sentry_incident", "glitchtip_incident", "incident", "observation"):
            raise ControlPlaneError(
                f"kind '{k}' is not assignable; bridge must create Gitea issues first (#612)"
            )
        raise ControlPlaneError(f"kind '{k}' is not assignable")
