add CI crash diagnostics, API Dockerfile, value set seeds, rustfs policies, loch theme, sigv4 client, and provider attribution table

api/Dockerfile

@@ -0,0 +1,23 @@
+# syntax=docker/dockerfile:1
+FROM ghcr.io/astral-sh/uv:python3.13-bookworm-slim
+
+WORKDIR /app
+
+# Copy project files for install
+COPY pyproject.toml uv.lock ./
+COPY src/ src/
+
+# Install the package (no dev deps)
+ENV UV_PYTHON_PREFERENCE=only-system \
+    UV_LINK_MODE=copy \
+    UV_PROJECT_ENVIRONMENT=.venv
+RUN uv sync --no-dev && uv pip install -e .
+
+# Config
+COPY stack.toml ./
+
+EXPOSE 8000
+
+CMD ["uv", "run", "--no-sync", "uvicorn", "api.server:app", \
+     "--host", "0.0.0.0", "--port", "8000", \
+     "--workers", "1", "--log-level", "info"]

rustfs/policies/gitea.json

@@ -0,0 +1,33 @@
+{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Effect": "Allow",
+      "Action": [
+        "s3:GetBucketLocation",
+        "s3:ListBucket",
+        "s3:ListBucketMultipartUploads"
+      ],
+      "Resource": [
+        "arn:aws:s3:::gitea",
+        "arn:aws:s3:::gitea-lfs",
+        "arn:aws:s3:::gitea-packages"
+      ]
+    },
+    {
+      "Effect": "Allow",
+      "Action": [
+        "s3:GetObject",
+        "s3:PutObject",
+        "s3:DeleteObject",
+        "s3:ListMultipartUploadParts",
+        "s3:AbortMultipartUpload"
+      ],
+      "Resource": [
+        "arn:aws:s3:::gitea/*",
+        "arn:aws:s3:::gitea-lfs/*",
+        "arn:aws:s3:::gitea-packages/*"
+      ]
+    }
+  ]
+}

rustfs/policies/nessie.json

@@ -0,0 +1,29 @@
+{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Effect": "Allow",
+      "Action": [
+        "s3:GetBucketLocation",
+        "s3:ListBucket",
+        "s3:ListBucketMultipartUploads"
+      ],
+      "Resource": [
+        "arn:aws:s3:::lakehouse"
+      ]
+    },
+    {
+      "Effect": "Allow",
+      "Action": [
+        "s3:GetObject",
+        "s3:PutObject",
+        "s3:DeleteObject",
+        "s3:ListMultipartUploadParts",
+        "s3:AbortMultipartUpload"
+      ],
+      "Resource": [
+        "arn:aws:s3:::lakehouse/*"
+      ]
+    }
+  ]
+}

rustfs/policies/polaris.json

@@ -0,0 +1,29 @@
+{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Effect": "Allow",
+      "Action": [
+        "s3:GetBucketLocation",
+        "s3:ListBucket",
+        "s3:ListBucketMultipartUploads"
+      ],
+      "Resource": [
+        "arn:aws:s3:::lakehouse"
+      ]
+    },
+    {
+      "Effect": "Allow",
+      "Action": [
+        "s3:GetObject",
+        "s3:PutObject",
+        "s3:DeleteObject",
+        "s3:ListMultipartUploadParts",
+        "s3:AbortMultipartUpload"
+      ],
+      "Resource": [
+        "arn:aws:s3:::lakehouse/*"
+      ]
+    }
+  ]
+}

src/aco/table/tuva_provider_attribution.py

@@ -0,0 +1,214 @@
+from __future__ import annotations
+
+from datetime import date
+from decimal import Decimal
+
+from aco.table.base import SQLTable
+
+
+class TuvaProviderAttributionIntCurrentSteps(SQLTable):
+    """Schema: tuva_provider_attribution / Table: _int_current_steps"""
+
+    __schema__ = "tuva_provider_attribution"
+    __tablename__ = "_int_current_steps"
+
+    person_id: str | None = None
+    provider_id: str | None = None
+    provider_bucket: str | None = None
+    prov_specialty: str | None = None
+    step: int | None = None
+    step_description: str | None = None
+    allowed_amount: Decimal | None = None
+    visits: int | None = None
+
+
+class TuvaProviderAttributionIntPersonYears(SQLTable):
+    """Schema: tuva_provider_attribution / Table: _int_person_years"""
+
+    __schema__ = "tuva_provider_attribution"
+    __tablename__ = "_int_person_years"
+
+    person_id: str | None = None
+    performance_year: int | None = None
+
+
+class TuvaProviderAttributionIntPrimaryCareClaims(SQLTable):
+    """Schema: tuva_provider_attribution / Table: _int_primary_care_claims"""
+
+    __schema__ = "tuva_provider_attribution"
+    __tablename__ = "_int_primary_care_claims"
+
+    person_id: str | None = None
+    claim_id: str | None = None
+    claim_line_number: int | None = None
+    encounter_id: str | None = None
+    claim_start_date: date | None = None
+    claim_end_date: date | None = None
+    claim_year: int | None = None
+    claim_month: int | None = None
+    claim_year_month: str | None = None
+    claim_year_month_int: int | None = None
+    allowed_amount: Decimal | None = None
+    provider_id: str | None = None
+    hcpcs_code: str | None = None
+    provider_bucket: str | None = None
+    prov_specialty: str | None = None
+
+
+class TuvaProviderAttributionIntProviderClassification(SQLTable):
+    """Schema: tuva_provider_attribution / Table: _int_provider_classification"""
+
+    __schema__ = "tuva_provider_attribution"
+    __tablename__ = "_int_provider_classification"
+
+    provider_id: str | None = None
+    prov_specialty: str | None = None
+    provider_bucket: str | None = None
+
+
+class TuvaProviderAttributionIntYearlySteps(SQLTable):
+    """Schema: tuva_provider_attribution / Table: _int_yearly_steps"""
+
+    __schema__ = "tuva_provider_attribution"
+    __tablename__ = "_int_yearly_steps"
+
+    person_id: str | None = None
+    performance_year: int | None = None
+    provider_id: str | None = None
+    provider_bucket: str | None = None
+    prov_specialty: str | None = None
+    step: int | None = None
+    step_description: str | None = None
+    allowed_amount: Decimal | None = None
+    visits: int | None = None
+
+
+class TuvaProviderAttributionStgCoreClaimsMedicalClaim(SQLTable):
+    """Schema: tuva_provider_attribution / Table: _stg_core__claims_medical_claim"""
+
+    __schema__ = "tuva_provider_attribution"
+    __tablename__ = "_stg_core__claims_medical_claim"
+
+    claim_id: str | None = None
+    claim_line_number: int | None = None
+    data_source: str | None = None
+    encounter_id: str | None = None
+
+
+class TuvaProviderAttributionStgCoreMedicalClaim(SQLTable):
+    """Schema: tuva_provider_attribution / Table: _stg_core__medical_claim"""
+
+    __schema__ = "tuva_provider_attribution"
+    __tablename__ = "_stg_core__medical_claim"
+
+    claim_id: str | None = None
+    claim_line_number: int | None = None
+    person_id: str | None = None
+    claim_start_date: date | None = None
+    claim_end_date: date | None = None
+    allowed_amount: Decimal | None = None
+    paid_amount: Decimal | None = None
+    rendering_npi: str | None = None
+    hcpcs_code: str | None = None
+    data_source: str | None = None
+    encounter_id: str | None = None
+
+
+class TuvaProviderAttributionStgCoreMemberMonths(SQLTable):
+    """Schema: tuva_provider_attribution / Table: _stg_core__member_months"""
+
+    __schema__ = "tuva_provider_attribution"
+    __tablename__ = "_stg_core__member_months"
+
+    person_id: str | None = None
+    year_month: str | None = None
+
+
+class TuvaProviderAttributionStgReferenceDataCalendar(SQLTable):
+    """Schema: tuva_provider_attribution / Table: _stg_reference_data__calendar"""
+
+    __schema__ = "tuva_provider_attribution"
+    __tablename__ = "_stg_reference_data__calendar"
+
+    full_date: date | None = None
+    year: int | None = None
+    month: int | None = None
+    year_month: str | None = None
+    first_day_of_month: date | None = None
+    last_day_of_month: date | None = None
+    year_month_int: int | None = None
+
+
+class TuvaProviderAttributionStgTerminologyProvider(SQLTable):
+    """Schema: tuva_provider_attribution / Table: _stg_terminology__provider"""
+
+    __schema__ = "tuva_provider_attribution"
+    __tablename__ = "_stg_terminology__provider"
+
+    npi: str | None = None
+    primary_taxonomy_code: str | None = None
+    primary_specialty_description: str | None = None
+    entity_type_description: str | None = None
+
+
+class TuvaProviderAttributionAssignedBeneficiariesCurrent(SQLTable):
+    """Schema: tuva_provider_attribution / Table: assigned_beneficiaries_current"""
+
+    __schema__ = "tuva_provider_attribution"
+    __tablename__ = "assigned_beneficiaries_current"
+
+    person_id: str | None = None
+    as_of_date: date | None = None
+    provider_id: str | None = None
+    provider_bucket: str | None = None
+    prov_specialty: str | None = None
+    assigned_step: int | None = None
+    step_description: str | None = None
+    allowed_amount: Decimal | None = None
+    visits: int | None = None
+    lookback_start_date: date | None = None
+    lookback_end_date: date | None = None
+    attribution_key: str | None = None
+
+
+class TuvaProviderAttributionAssignedBeneficiariesYearly(SQLTable):
+    """Schema: tuva_provider_attribution / Table: assigned_beneficiaries_yearly"""
+
+    __schema__ = "tuva_provider_attribution"
+    __tablename__ = "assigned_beneficiaries_yearly"
+
+    person_id: str | None = None
+    performance_year: int | None = None
+    provider_id: str | None = None
+    provider_bucket: str | None = None
+    prov_specialty: str | None = None
+    assigned_step: int | None = None
+    step_description: str | None = None
+    allowed_amount: Decimal | None = None
+    visits: int | None = None
+    lookback_start_date: date | None = None
+    lookback_end_date: date | None = None
+    attribution_key: str | None = None
+
+
+class TuvaProviderAttributionProviderRanking(SQLTable):
+    """Schema: tuva_provider_attribution / Table: provider_ranking"""
+
+    __schema__ = "tuva_provider_attribution"
+    __tablename__ = "provider_ranking"
+
+    person_id: str | None = None
+    performance_year: int | None = None
+    as_of_date: date | None = None
+    provider_id: str | None = None
+    provider_bucket: str | None = None
+    prov_specialty: str | None = None
+    step: int | None = None
+    step_description: str | None = None
+    allowed_amount: Decimal | None = None
+    visits: int | None = None
+    scope: str | None = None
+    lookback_start_date: date | None = None
+    lookback_end_date: date | None = None
+    ranking: int | None = None
+    attribution_key: str | None = None

src/api/clients/rustfs/sigv4.py

@@ -0,0 +1,148 @@
+"""AWS Signature Version 4 signing — stdlib only (hmac, hashlib, datetime)."""
+
+from __future__ import annotations
+
+import hashlib
+import hmac
+import urllib.parse
+from datetime import datetime, timezone
+
+
+def _sign(key: bytes, msg: str) -> bytes:
+    return hmac.new(key, msg.encode(), hashlib.sha256).digest()
+
+
+def _get_signature_key(
+    secret_key: str, date_stamp: str, region: str, service: str
+) -> bytes:
+    k_date = _sign(f"AWS4{secret_key}".encode(), date_stamp)
+    k_region = _sign(k_date, region)
+    k_service = _sign(k_region, service)
+    return _sign(k_service, "aws4_request")
+
+
+def _canonical_query_string(params: dict[str, str] | None) -> str:
+    if not params:
+        return ""
+    return "&".join(
+        f"{urllib.parse.quote(k, safe='')}={urllib.parse.quote(v, safe='')}"
+        for k, v in sorted(params.items())
+    )
+
+
+def sign_request(
+    method: str,
+    url: str,
+    headers: dict[str, str],
+    body: bytes | None,
+    access_key: str,
+    secret_key: str,
+    region: str = "us-east-1",
+    service: str = "s3",
+    *,
+    now: datetime | None = None,
+) -> dict[str, str]:
+    """Add SigV4 Authorization header to a request.
+
+    Parameters
+    ----------
+    method : str
+        HTTP method (GET, PUT, etc.).
+    url : str
+        Full URL including scheme, host, path, and optional query string.
+    headers : dict
+        Existing headers (Host is required or derived from URL).
+    body : bytes | None
+        Request body (use b"" for empty).
+    access_key, secret_key : str
+        AWS credentials.
+    region, service : str
+        Signing scope.
+    now : datetime | None
+        Override timestamp for testing.
+
+    Returns
+    -------
+    dict
+        Updated headers dict with Authorization, x-amz-date, and
+        x-amz-content-sha256 added.
+    """
+    if now is None:
+        now = datetime.now(timezone.utc)
+
+    amz_date = now.strftime("%Y%m%dT%H%M%SZ")
+    date_stamp = now.strftime("%Y%m%d")
+
+    parsed = urllib.parse.urlparse(url)
+    host = parsed.hostname or ""
+    if parsed.port and parsed.port not in (80, 443):
+        host = f"{host}:{parsed.port}"
+    canonical_uri = urllib.parse.quote(parsed.path or "/", safe="/")
+
+    # Parse query string
+    qs_params = dict(urllib.parse.parse_qsl(parsed.query, keep_blank_values=True))
+    canonical_querystring = _canonical_query_string(qs_params)
+
+    payload_hash = hashlib.sha256(body or b"").hexdigest()
+
+    # Build headers to sign
+    headers_to_sign = dict(headers)
+    headers_to_sign["host"] = host
+    headers_to_sign["x-amz-date"] = amz_date
+    headers_to_sign["x-amz-content-sha256"] = payload_hash
+
+    # Canonical headers: sorted, lowercased, trimmed
+    signed_header_keys = sorted(k.lower() for k in headers_to_sign)
+    canonical_headers = "".join(
+        f"{k}:{headers_to_sign[k].strip()}\n"
+        for k in signed_header_keys
+        if k.lower() in {h.lower() for h in headers_to_sign}
+    )
+    # Re-resolve: canonical headers must use lowercased keys
+    canonical_headers = ""
+    lower_map: dict[str, str] = {}
+    for k, v in headers_to_sign.items():
+        lk = k.lower()
+        lower_map[lk] = v.strip()
+    signed_header_keys = sorted(lower_map)
+    canonical_headers = "".join(f"{k}:{lower_map[k]}\n" for k in signed_header_keys)
+    signed_headers = ";".join(signed_header_keys)
+
+    canonical_request = "\n".join(
+        [
+            method.upper(),
+            canonical_uri,
+            canonical_querystring,
+            canonical_headers,
+            signed_headers,
+            payload_hash,
+        ]
+    )
+
+    credential_scope = f"{date_stamp}/{region}/{service}/aws4_request"
+    string_to_sign = "\n".join(
+        [
+            "AWS4-HMAC-SHA256",
+            amz_date,
+            credential_scope,
+            hashlib.sha256(canonical_request.encode()).hexdigest(),
+        ]
+    )
+
+    signing_key = _get_signature_key(secret_key, date_stamp, region, service)
+    signature = hmac.new(
+        signing_key, string_to_sign.encode(), hashlib.sha256
+    ).hexdigest()
+
+    authorization = (
+        f"AWS4-HMAC-SHA256 "
+        f"Credential={access_key}/{credential_scope}, "
+        f"SignedHeaders={signed_headers}, "
+        f"Signature={signature}"
+    )
+
+    result = dict(headers)
+    result["x-amz-date"] = amz_date
+    result["x-amz-content-sha256"] = payload_hash
+    result["Authorization"] = authorization
+    return result

src/api/diag/__init__.py

@@ -0,0 +1,5 @@
+"""Crash diagnostics — auto-capture, blame, and issue filing."""
+
+from api.diag.guard import cli_guard as cli_guard
+from api.diag.guard import guarded as guarded
+from api.diag.hook import install as install

src/api/diag/__main__.py

@@ -0,0 +1,171 @@
+"""CI crash reporter — scrape failed Woodpecker step logs and file issues.
+
+Usage (in a Woodpecker ``failure`` step)::
+
+    uv run python -m api.diag
+
+Reads CI environment variables to locate the failed pipeline, pulls
+step logs via the Woodpecker API, parses any Python tracebacks, runs
+``git blame`` on the offending lines, and files a Gitea issue.
+
+For non-Python failures (docker build errors, shell script crashes),
+it still files an issue with the raw log output.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import sys
+
+log = logging.getLogger(__name__)
+
+
+def _get_env(name: str) -> str:
+    val = os.environ.get(name, "")
+    if not val:
+        log.warning("Missing CI env var: %s", name)
+    return val
+
+
+def main() -> int:
+    logging.basicConfig(
+        level=logging.INFO,
+        format="%(levelname)s: %(message)s",
+    )
+
+    repo_id = _get_env("CI_REPO_ID")
+    pipeline_number = _get_env("CI_PIPELINE_NUMBER")
+    commit_sha = _get_env("CI_COMMIT_SHA")
+    repo_name = _get_env("CI_REPO")  # e.g. "homelab/stack"
+
+    if not repo_id or not pipeline_number:
+        log.error("CI_REPO_ID and CI_PIPELINE_NUMBER are required")
+        return 1
+
+    gitea_token = os.environ.get("GITEA_TOKEN", "")
+    if not gitea_token:
+        log.error("GITEA_TOKEN is required to file issues")
+        return 1
+
+    # Parse owner/repo from CI_REPO
+    if "/" in repo_name:
+        owner, repo = repo_name.split("/", 1)
+    else:
+        owner, repo = "homelab", "stack"
+
+    # Fetch pipeline steps and find failures
+    from api.clients.woodpecker import WoodpeckerClient
+
+    wp = WoodpeckerClient(
+        gitea_token,
+        base_url=os.environ.get(
+            "CI_WOODPECKER_URL", "http://woodpecker-server:8000/api"
+        ),
+    )
+
+    rid = int(repo_id)
+    pnum = int(pipeline_number)
+
+    pipeline = wp.get_pipeline(rid, pnum)
+    steps = pipeline.get("steps", [])
+
+    failed_steps = [s for s in steps if s.get("state") == "failure"]
+
+    if not failed_steps:
+        log.info("No failed steps in pipeline %d — nothing to report", pnum)
+        wp.close()
+        return 0
+
+    # Collect logs from all failed steps
+    from api.clients.gitea import GiteaClient
+    from api.diag.blame import blame_report
+    from api.diag.issue import _truncate, build_issue_body
+    from api.diag.trace import parse_traceback_text
+
+    gitea = GiteaClient(gitea_token)
+
+    for step in failed_steps:
+        step_name = step.get("name", "unknown")
+        step_id = step.get("id", 0)
+
+        log.info("Processing failed step: %s (id=%d)", step_name, step_id)
+
+        # Fetch logs — Woodpecker returns a list of log line dicts
+        try:
+            log_entries = wp.get_logs(rid, pnum, step_id)
+        except Exception:
+            log.exception("Failed to fetch logs for step %s", step_name)
+            continue
+
+        # Concatenate log lines
+        if isinstance(log_entries, list):
+            log_text = "\n".join(
+                entry.get("data", "") if isinstance(entry, dict) else str(entry)
+                for entry in log_entries
+            )
+        else:
+            log_text = str(log_entries)
+
+        # Try to parse Python tracebacks
+        reports = parse_traceback_text(log_text)
+
+        if reports:
+            # File one issue per traceback found
+            for report in reports:
+                blames = blame_report(report)
+                body = build_issue_body(
+                    report,
+                    blames,
+                    {f"step:{step_name}": log_text[-3000:]},
+                    commit_sha or "unknown",
+                )
+                body = (
+                    f"**Pipeline:** #{pnum} step `{step_name}`\n"
+                    f"**Status:** `failure`\n\n" + body
+                )
+                title = (
+                    f"ci: {step_name} — {report.exc_type}: "
+                    f"{_truncate(report.exc_value, 50)}"
+                )
+                result = gitea.create_issue(
+                    owner,
+                    repo,
+                    {
+                        "title": title,
+                        "body": body,
+                        "labels": [],
+                    },
+                )
+                log.info("Filed issue #%s: %s", result.get("number"), title)
+        else:
+            # Non-Python failure — file with raw log
+            title = f"ci: {step_name} failed (pipeline #{pnum})"
+            body = (
+                f"**Pipeline:** #{pnum} step `{step_name}`\n"
+                f"**Commit:** `{commit_sha}`\n"
+                f"**Status:** `failure`\n\n"
+                f"## Step Log\n\n"
+                f"<details>\n"
+                f"<summary>{step_name} output (last 3000 chars)</summary>\n\n"
+                f"```\n{log_text[-3000:]}\n```\n\n"
+                f"</details>\n"
+            )
+            result = gitea.create_issue(
+                owner,
+                repo,
+                {
+                    "title": title,
+                    "body": body,
+                    "labels": [],
+                },
+            )
+            log.info("Filed issue #%s: %s", result.get("number"), title)
+
+    wp.close()
+    gitea.close()
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

src/api/diag/blame.py

@@ -0,0 +1,92 @@
+"""Git blame integration for crash frames.
+
+Runs ``git blame -L <line>,<line> --porcelain`` on each project-local
+frame to identify the commit that last touched the offending line.
+"""
+
+from __future__ import annotations
+
+import logging
+import subprocess
+from dataclasses import dataclass
+from pathlib import Path
+
+from api.diag.trace import CrashReport
+
+log = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True, slots=True)
+class BlameInfo:
+    sha: str
+    author: str
+    summary: str
+    timestamp: str
+
+
+def blame_line(
+    filepath: str, lineno: int, *, cwd: str | None = None
+) -> BlameInfo | None:
+    """Run git blame on a single line, return structured result."""
+    try:
+        result = subprocess.run(
+            ["git", "blame", "-L", f"{lineno},{lineno}", "--porcelain", filepath],
+            capture_output=True,
+            text=True,
+            cwd=cwd,
+            timeout=10,
+        )
+    except (subprocess.TimeoutExpired, FileNotFoundError):
+        return None
+
+    if result.returncode != 0:
+        return None
+
+    sha = ""
+    author = ""
+    summary = ""
+    timestamp = ""
+
+    for raw_line in result.stdout.splitlines():
+        if not sha and len(raw_line) >= 40 and raw_line[0] != "\t":
+            parts = raw_line.split()
+            if len(parts[0]) == 40:
+                sha = parts[0]
+        elif raw_line.startswith("author "):
+            author = raw_line[7:]
+        elif raw_line.startswith("summary "):
+            summary = raw_line[8:]
+        elif raw_line.startswith("author-time "):
+            timestamp = raw_line[12:]
+
+    if not sha:
+        return None
+
+    return BlameInfo(sha=sha, author=author, summary=summary, timestamp=timestamp)
+
+
+def _project_root() -> Path:
+    return Path(__file__).resolve().parents[3]
+
+
+def blame_report(report: CrashReport) -> dict[str, BlameInfo]:
+    """Run blame on all project-local frames in the crash report.
+
+    Returns a dict keyed by ``filepath:lineno``.
+    """
+    root = _project_root()
+    results: dict[str, BlameInfo] = {}
+
+    for frame in report.frames:
+        try:
+            rel = Path(frame.filepath).resolve().relative_to(root)
+        except ValueError:
+            continue
+
+        key = f"{rel}:{frame.lineno}"
+        info = blame_line(str(rel), frame.lineno, cwd=str(root))
+        if info:
+            results[key] = info
+            log.debug("Blamed %s → %s (%s)", key, info.sha[:8], info.summary)
+
+    return results

src/api/diag/guard.py

@@ -0,0 +1,76 @@
+"""CLI crash guard — wraps entry points to auto-file issues on failure.
+
+Usage::
+
+    from api.diag.guard import cli_guard
+
+    def main() -> int:
+        ...
+
+    if __name__ == "__main__":
+        sys.exit(cli_guard(main))
+
+Or as a decorator::
+
+    @guarded
+    def main() -> int:
+        ...
+"""
+
+from __future__ import annotations
+
+import logging
+import sys
+import traceback
+from typing import Callable
+
+log = logging.getLogger(__name__)
+
+
+def cli_guard(fn: Callable[[], int], **issue_kw) -> int:
+    """Run *fn* and auto-file a Gitea issue if it raises.
+
+    Returns the int return code from *fn*, or 2 on unhandled exception.
+    The exception is still printed to stderr.
+    """
+    try:
+        return fn()
+    except (KeyboardInterrupt, SystemExit) as exc:
+        if isinstance(exc, SystemExit):
+            return exc.code if isinstance(exc.code, int) else 1
+        return 130
+    except Exception:
+        # Print the traceback to stderr as normal
+        traceback.print_exc()
+
+        # File the issue
+        exc_type, exc_value, exc_tb = sys.exc_info()
+        try:
+            from api.diag.issue import file_issue
+            from api.diag.trace import parse_exception
+
+            report = parse_exception(exc_type, exc_value, exc_tb)
+            result = file_issue(report, **issue_kw)
+            if result:
+                log.error(
+                    "Crash filed as issue #%s: %s",
+                    result.get("number"),
+                    result.get("html_url", ""),
+                )
+            else:
+                log.error("Could not file crash issue")
+        except Exception:
+            log.exception("Crash diagnostics failed")
+
+        return 2
+
+
+def guarded(fn: Callable[[], int]) -> Callable[[], int]:
+    """Decorator form of ``cli_guard``."""
+
+    def wrapper() -> int:
+        return cli_guard(fn)
+
+    wrapper.__name__ = fn.__name__
+    wrapper.__doc__ = fn.__doc__
+    return wrapper

src/api/diag/hook.py

@@ -0,0 +1,91 @@
+"""sys.excepthook integration — auto-capture unhandled exceptions.
+
+Call ``install()`` once at application startup (e.g. in ``api.server``
+or ``cli.__init__``) to wire the crash diagnostics into Python's
+unhandled exception path.
+
+The hook:
+1. Parses the traceback via AST
+2. Runs ``git blame`` on project frames
+3. Collects container logs from related services
+4. Posts a Gitea issue with full context
+5. Calls the original ``sys.excepthook`` so the traceback still prints
+"""
+
+from __future__ import annotations
+
+import logging
+import sys
+import types
+
+log = logging.getLogger(__name__)
+
+_original_hook: types.FunctionType | None = None
+_installed = False
+
+
+def _excepthook(
+    exc_type: type[BaseException],
+    exc_value: BaseException,
+    exc_tb: types.TracebackType | None,
+) -> None:
+    """Custom excepthook that files a Gitea issue on crash."""
+    # Always print the traceback first
+    if _original_hook:
+        _original_hook(exc_type, exc_value, exc_tb)
+    else:
+        sys.__excepthook__(exc_type, exc_value, exc_tb)
+
+    # Skip keyboard interrupts and system exits
+    if issubclass(exc_type, (KeyboardInterrupt, SystemExit)):
+        return
+
+    try:
+        from api.diag.trace import parse_exception
+
+        report = parse_exception(exc_type, exc_value, exc_tb)
+
+        from api.diag.issue import file_issue
+
+        result = file_issue(report)
+        if result:
+            number = result.get("number", "?")
+            log.info("Crash filed as issue #%s", number)
+        else:
+            log.warning("Could not file crash issue (no token or API error)")
+    except Exception:
+        # The diagnostics module itself crashed — don't recurse
+        log.exception("Crash diagnostics failed")
+
+
+def install() -> None:
+    """Install the crash diagnostics excepthook.
+
+    Safe to call multiple times — only installs once.
+    """
+    global _original_hook, _installed
+
+    if _installed:
+        return
+
+    _original_hook = sys.excepthook
+    sys.excepthook = _excepthook
+    _installed = True
+    log.debug("Crash diagnostics hook installed")
+
+
+def uninstall() -> None:
+    """Restore the original excepthook."""
+    global _original_hook, _installed
+
+    if not _installed:
+        return
+
+    if _original_hook:
+        sys.excepthook = _original_hook
+    else:
+        sys.excepthook = sys.__excepthook__
+
+    _original_hook = None
+    _installed = False
+    log.debug("Crash diagnostics hook uninstalled")

src/api/diag/issue.py

@@ -0,0 +1,184 @@
+"""Gitea issue generation from crash reports.
+
+Builds a markdown issue body from a CrashReport, blame data, and
+container logs, then posts it to Gitea via the API.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import subprocess
+from pathlib import Path
+
+from api.diag.blame import BlameInfo, blame_report
+from api.diag.logs import collect_logs
+from api.diag.trace import CrashReport
+
+log = logging.getLogger(__name__)
+
+
+def _current_commit() -> str:
+    """Read COMMIT_SHA from .env, fall back to git rev-parse."""
+    env_path = Path(__file__).resolve().parents[3] / ".env"
+    if env_path.exists():
+        for line in env_path.read_text().splitlines():
+            if line.startswith("COMMIT_SHA="):
+                return line.partition("=")[2].strip()
+
+    try:
+        result = subprocess.run(
+            ["git", "rev-parse", "HEAD"],
+            capture_output=True,
+            text=True,
+            timeout=5,
+        )
+        if result.returncode == 0:
+            return result.stdout.strip()
+    except (subprocess.TimeoutExpired, FileNotFoundError):
+        pass
+
+    return "unknown"
+
+
+def _format_frame(frame, blame: BlameInfo | None) -> str:
+    """Format a single traceback frame as a markdown table row."""
+    ctx = frame.ast_context or frame.name
+    blame_str = ""
+    if blame:
+        blame_str = f"`{blame.sha[:8]}` {blame.author} — {blame.summary}"
+    return f"| `{frame.filepath}` | {frame.lineno} | `{ctx}` | {blame_str} |"
+
+
+def build_issue_body(
+    report: CrashReport,
+    blames: dict[str, BlameInfo],
+    container_logs: dict[str, str],
+    commit: str,
+) -> str:
+    """Build a markdown issue body from crash data."""
+    lines: list[str] = []
+
+    # Header
+    lines.append(f"**Commit:** `{commit}`")
+    lines.append(f"**Exception:** `{report.exc_type}: {report.exc_value}`")
+    lines.append("")
+
+    # Stack trace table
+    lines.append("## Stack Trace")
+    lines.append("")
+    lines.append("| File | Line | Context | Blame |")
+    lines.append("|------|------|---------|-------|")
+
+    for frame in report.frames:
+        try:
+            root = Path(__file__).resolve().parents[3]
+            rel = Path(frame.filepath).resolve().relative_to(root)
+            key = f"{rel}:{frame.lineno}"
+        except ValueError:
+            key = f"{frame.filepath}:{frame.lineno}"
+
+        blame = blames.get(key)
+        lines.append(_format_frame(frame, blame))
+
+    lines.append("")
+
+    # Offending code
+    project_frames = [f for f in report.frames if f.ast_context]
+    if project_frames:
+        lines.append("## Offending Code")
+        lines.append("")
+        for frame in project_frames:
+            lines.append(
+                f"**`{frame.filepath}:{frame.lineno}`** in `{frame.ast_context}`:"
+            )
+            lines.append("```python")
+            lines.append(frame.line)
+            lines.append("```")
+            lines.append("")
+
+    # Blame summary — unique commits that touched crash frames
+    unique_shas: dict[str, BlameInfo] = {}
+    for info in blames.values():
+        if info.sha not in unique_shas:
+            unique_shas[info.sha] = info
+
+    if unique_shas:
+        lines.append("## Commits Involved")
+        lines.append("")
+        for sha, info in unique_shas.items():
+            lines.append(f"- `{sha[:8]}` by **{info.author}** — {info.summary}")
+        lines.append("")
+
+    # Container logs
+    if container_logs:
+        lines.append("## Container Logs")
+        lines.append("")
+        for container, output in container_logs.items():
+            lines.append(f"### `{container}`")
+            lines.append("")
+            lines.append("<details>")
+            lines.append(f"<summary>{container} logs (last 50 lines)</summary>")
+            lines.append("")
+            lines.append("```")
+            lines.append(output)
+            lines.append("```")
+            lines.append("")
+            lines.append("</details>")
+            lines.append("")
+
+    return "\n".join(lines)
+
+
+def file_issue(
+    report: CrashReport,
+    *,
+    owner: str = "homelab",
+    repo: str = "stack",
+    labels: list[str] | None = None,
+) -> dict | None:
+    """Collect blame + logs, build issue body, and post to Gitea.
+
+    Returns the created issue dict, or None if posting fails.
+    """
+    token = os.environ.get("GITEA_TOKEN", "")
+    if not token:
+        from conf import secret
+
+        token = secret("gitea.token", "GITEA_TOKEN")
+
+    if not token:
+        log.error("No GITEA_TOKEN available — cannot file issue")
+        return None
+
+    commit = _current_commit()
+    blames = blame_report(report)
+    container_logs = collect_logs(report.exc_value)
+    body = build_issue_body(report, blames, container_logs, commit)
+
+    title = f"crash: {report.exc_type}: {_truncate(report.exc_value, 60)}"
+
+    issue_body: dict = {
+        "title": title,
+        "body": body,
+    }
+    if labels:
+        issue_body["labels"] = labels
+
+    try:
+        from api.clients.gitea import GiteaClient
+
+        client = GiteaClient(token)
+        result = client.create_issue(owner, repo, issue_body)
+        log.info("Filed issue #%s: %s", result.get("number"), title)
+        client.close()
+        return result
+    except Exception:
+        log.exception("Failed to file Gitea issue")
+        return None
+
+
+def _truncate(s: str, max_len: int) -> str:
+    if len(s) <= max_len:
+        return s
+    return s[: max_len - 3] + "..."

src/api/diag/logs.py

@@ -0,0 +1,89 @@
+"""Container log collection for crash diagnostics.
+
+Pulls recent ``docker logs`` output from the service container that
+crashed, plus any related containers (e.g. postgres if the crash
+mentions a DB connection error).
+"""
+
+from __future__ import annotations
+
+import logging
+import subprocess
+
+log = logging.getLogger(__name__)
+
+# Containers to always collect logs from on crash
+DEFAULT_CONTAINERS = ("api",)
+
+# Map exception substrings → extra containers to pull logs from
+RELATED_CONTAINERS: dict[str, tuple[str, ...]] = {
+    "postgres": ("postgres",),
+    "psycopg": ("postgres",),
+    "sqlalchemy": ("postgres",),
+    "rustfs": ("rustfs",),
+    "s3": ("rustfs",),
+    "gitea": ("gitea",),
+    "woodpecker": ("woodpecker-server",),
+    "grafana": ("grafana",),
+    "nessie": ("nessie",),
+    "trino": ("trino",),
+}
+
+
+def collect_logs(
+    exc_value: str,
+    *,
+    containers: tuple[str, ...] = DEFAULT_CONTAINERS,
+    tail: int = 50,
+    since: str = "5m",
+) -> dict[str, str]:
+    """Collect recent logs from relevant containers.
+
+    Parameters
+    ----------
+    exc_value : str
+        The exception message — used to detect related services.
+    containers : tuple[str, ...]
+        Base containers to always query.
+    tail : int
+        Number of trailing lines per container.
+    since : str
+        Docker ``--since`` duration (e.g. ``5m``, ``1h``).
+
+    Returns
+    -------
+    dict[str, str]
+        Container name → log output.
+    """
+    targets = set(containers)
+    exc_lower = exc_value.lower()
+
+    for keyword, extra in RELATED_CONTAINERS.items():
+        if keyword in exc_lower:
+            targets.update(extra)
+
+    results: dict[str, str] = {}
+    for name in sorted(targets):
+        output = _docker_logs(name, tail=tail, since=since)
+        if output:
+            results[name] = output
+
+    return results
+
+
+def _docker_logs(container: str, *, tail: int = 50, since: str = "5m") -> str:
+    """Fetch recent logs from a Docker container."""
+    try:
+        result = subprocess.run(
+            ["docker", "logs", container, "--tail", str(tail), "--since", since],
+            capture_output=True,
+            text=True,
+            timeout=10,
+        )
+    except (subprocess.TimeoutExpired, FileNotFoundError):
+        log.warning("Failed to collect logs from %s", container)
+        return ""
+
+    # docker logs outputs to stderr for some containers
+    output = result.stdout or result.stderr
+    return output.strip()

src/api/diag/trace.py

@@ -0,0 +1,191 @@
+"""AST-driven traceback parsing.
+
+Extracts structured frame data from exception tracebacks using the
+``traceback`` module (not regex) for live exceptions, and a line-by-line
+state machine for parsing tracebacks from log text (CI step output).
+
+For each frame that points to a file inside the project ``src/`` tree,
+the surrounding AST node (function/class) is resolved so the issue body
+can show *what* broke, not just a line number.
+"""
+
+from __future__ import annotations
+
+import ast
+import re
+import traceback
+import types
+from dataclasses import dataclass, field
+from pathlib import Path
+
+
+@dataclass(frozen=True, slots=True)
+class Frame:
+    filepath: str
+    lineno: int
+    name: str
+    line: str
+    ast_context: str = ""
+
+
+@dataclass(frozen=True, slots=True)
+class CrashReport:
+    exc_type: str
+    exc_value: str
+    frames: list[Frame] = field(default_factory=list)
+
+
+def _resolve_ast_context(filepath: str, lineno: int) -> str:
+    """Walk the AST to find the enclosing function/class for *lineno*."""
+    try:
+        source = Path(filepath).read_text()
+    except (OSError, UnicodeDecodeError):
+        return ""
+
+    try:
+        tree = ast.parse(source, filename=filepath)
+    except SyntaxError:
+        return ""
+
+    best: str = ""
+    best_line: int = 0
+
+    for node in ast.walk(tree):
+        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
+            if node.lineno <= lineno and node.lineno > best_line:
+                best = f"{type(node).__name__} {node.name}"
+                best_line = node.lineno
+
+    return best
+
+
+def _is_project_file(filepath: str, root: Path | None = None) -> bool:
+    """Return True if *filepath* lives under the project src/ tree."""
+    if root is None:
+        root = Path(__file__).resolve().parents[3]
+    try:
+        Path(filepath).resolve().relative_to(root / "src")
+        return True
+    except ValueError:
+        return False
+
+
+def parse_exception(
+    exc_type: type[BaseException],
+    exc_value: BaseException,
+    exc_tb: types.TracebackType | None,
+) -> CrashReport:
+    """Build a ``CrashReport`` from a live exception triple."""
+    frames: list[Frame] = []
+
+    for fs in traceback.extract_tb(exc_tb):
+        ast_ctx = ""
+        if _is_project_file(fs.filename):
+            ast_ctx = _resolve_ast_context(fs.filename, fs.lineno)
+
+        frames.append(
+            Frame(
+                filepath=fs.filename,
+                lineno=fs.lineno,
+                name=fs.name,
+                line=fs.line or "",
+                ast_context=ast_ctx,
+            )
+        )
+
+    return CrashReport(
+        exc_type=exc_type.__qualname__,
+        exc_value=str(exc_value),
+        frames=frames,
+    )
+
+
+# ── Text-based traceback parser (for CI logs) ─────────────────
+
+# Matches:  File "/path/to/file.py", line 42, in func_name
+_TB_FILE_RE = re.compile(r'^\s*File "([^"]+)", line (\d+), in (.+)$')
+# Matches:  ExceptionType: message  or  ExceptionType
+_TB_EXC_RE = re.compile(
+    r"^([A-Za-z_][\w.]*(?:Error|Exception|Warning|Exit|Interrupt))\s*(?::\s*(.*))?$"
+)
+
+
+def parse_traceback_text(text: str) -> list[CrashReport]:
+    """Parse Python tracebacks from raw log text.
+
+    Handles multiple tracebacks in a single log dump.  Uses a simple
+    state machine — no regex on the frame lines themselves beyond the
+    standard ``File "...", line N, in name`` pattern.
+
+    Returns a list of ``CrashReport`` objects (one per traceback found).
+    """
+    reports: list[CrashReport] = []
+    lines = text.splitlines()
+
+    i = 0
+    while i < len(lines):
+        line = lines[i]
+
+        # Look for "Traceback (most recent call last):"
+        if "Traceback (most recent call last)" in line:
+            frames: list[Frame] = []
+            i += 1
+
+            # Parse frames
+            while i < len(lines):
+                m = _TB_FILE_RE.match(lines[i])
+                if m:
+                    filepath, lineno_str, name = m.group(1), m.group(2), m.group(3)
+                    lineno = int(lineno_str)
+                    # Next line is the code line (if present and indented)
+                    code_line = ""
+                    if i + 1 < len(lines) and lines[i + 1].startswith("    "):
+                        code_line = lines[i + 1].strip()
+                        i += 1
+
+                    ast_ctx = ""
+                    if _is_project_file(filepath):
+                        ast_ctx = _resolve_ast_context(filepath, lineno)
+
+                    frames.append(
+                        Frame(
+                            filepath=filepath,
+                            lineno=lineno,
+                            name=name,
+                            line=code_line,
+                            ast_context=ast_ctx,
+                        )
+                    )
+                    i += 1
+                elif _TB_EXC_RE.match(lines[i]):
+                    # Hit the exception line — end of this traceback
+                    break
+                elif lines[i].strip() == "":
+                    # Skip blank lines within traceback
+                    i += 1
+                else:
+                    # Not a frame line and not the exception — might be
+                    # a "During handling..." separator; skip it
+                    i += 1
+
+            # Parse exception line
+            exc_type = "UnknownError"
+            exc_value = ""
+            if i < len(lines):
+                exc_m = _TB_EXC_RE.match(lines[i])
+                if exc_m:
+                    exc_type = exc_m.group(1)
+                    exc_value = (exc_m.group(2) or "").strip()
+
+            if frames or exc_type != "UnknownError":
+                reports.append(
+                    CrashReport(
+                        exc_type=exc_type,
+                        exc_value=exc_value,
+                        frames=frames,
+                    )
+                )
+
+        i += 1
+
+    return reports

styles/move.svg

@@ -0,0 +1,57 @@
+<svg viewBox="0 0 373 373" xmlns="http://www.w3.org/2000/svg">
+
+<style>
+
+#camera {
+  transform-origin: 186.5px 186.5px;
+  animation: zoomOut 3s ease forwards;
+}
+
+#pan {
+  animation: panOut 3s ease forwards;
+}
+
+#man {
+  transform-origin: 32px 364px;
+  animation: pivot 3s ease forwards;
+}
+
+/* camera zoom out */
+@keyframes zoomOut {
+  0%   { transform: scale(1.35); }
+  100% { transform: scale(1); }
+}
+
+/* camera pan */
+@keyframes panOut {
+  0%   { transform: translate(70px,0); }
+  100% { transform: translate(0,0); }
+}
+
+/* subtle pivot */
+@keyframes pivot {
+  0%   { transform: rotate(18deg); }
+  100% { transform: rotate(0deg); }
+}
+
+</style>
+
+<g id="camera">
+  <g id="pan">
+
+    <!-- fire layer -->
+    <image href="fire_layer.png"
+           x="0" y="0"
+           width="373" height="373"/>
+
+    <!-- F silhouette -->
+    <g id="man">
+      <image href="f_layer.png"
+             x="0" y="0"
+             width="373" height="373"/>
+    </g>
+
+  </g>
+</g>
+
+</svg>

tests/api/test_diag.py

@@ -0,0 +1,465 @@
+"""Tests for api.diag — crash diagnostics."""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+from api.diag.trace import (
+    CrashReport,
+    Frame,
+    _resolve_ast_context,
+    parse_exception,
+    parse_traceback_text,
+)
+
+
+class TestParseException:
+    def test_parses_basic_exception(self):
+        try:
+            raise ValueError("test error")
+        except ValueError:
+            exc_type, exc_value, exc_tb = sys.exc_info()
+            report = parse_exception(exc_type, exc_value, exc_tb)
+
+        assert report.exc_type == "ValueError"
+        assert report.exc_value == "test error"
+        assert len(report.frames) > 0
+        # The last frame should be this test file
+        assert "test_diag.py" in report.frames[-1].filepath
+
+    def test_identifies_project_frames(self):
+        try:
+            raise RuntimeError("boom")
+        except RuntimeError:
+            exc_type, exc_value, exc_tb = sys.exc_info()
+            report = parse_exception(exc_type, exc_value, exc_tb)
+
+        # At least one frame should have ast_context (this file is under tests/)
+        # but _is_project_file checks src/ — so test frames won't have it
+        assert report.frames[-1].ast_context == ""
+
+    def test_handles_none_tb(self):
+        report = parse_exception(ValueError, ValueError("test"), None)
+        assert report.exc_type == "ValueError"
+        assert report.frames == []
+
+
+class TestResolveAstContext:
+    def test_finds_function(self, tmp_path):
+        src = tmp_path / "example.py"
+        src.write_text("def foo():\n    x = 1\n    return x\n")
+        result = _resolve_ast_context(str(src), 2)
+        assert result == "FunctionDef foo"
+
+    def test_finds_class(self, tmp_path):
+        src = tmp_path / "example.py"
+        src.write_text("class Bar:\n    def method(self):\n        pass\n")
+        result = _resolve_ast_context(str(src), 3)
+        assert result == "FunctionDef method"
+
+    def test_returns_empty_for_missing_file(self):
+        result = _resolve_ast_context("/nonexistent/file.py", 1)
+        assert result == ""
+
+    def test_returns_empty_for_syntax_error(self, tmp_path):
+        src = tmp_path / "bad.py"
+        src.write_text("def foo(:\n")
+        result = _resolve_ast_context(str(src), 1)
+        assert result == ""
+
+
+class TestBlame:
+    def test_blame_line_runs_git(self):
+        from api.diag.blame import blame_line
+
+        # Blame a known file in the repo
+        root = Path(__file__).resolve().parents[2]
+        info = blame_line("src/api/diag/__init__.py", 1, cwd=str(root))
+        # This file was just created so blame should return something
+        if info:
+            assert len(info.sha) == 40
+            assert info.author != ""
+
+    def test_blame_line_returns_none_for_bad_file(self):
+        from api.diag.blame import blame_line
+
+        info = blame_line("/nonexistent/file.py", 1)
+        assert info is None
+
+    def test_blame_report_filters_project_files(self):
+        from api.diag.blame import blame_report
+
+        report = CrashReport(
+            exc_type="ValueError",
+            exc_value="test",
+            frames=[
+                Frame(
+                    filepath="/usr/lib/python3/something.py",
+                    lineno=1,
+                    name="test",
+                    line="pass",
+                ),
+            ],
+        )
+        results = blame_report(report)
+        assert results == {}
+
+
+class TestLogs:
+    def test_collect_logs_includes_default_container(self):
+        from api.diag.logs import collect_logs
+
+        with patch("api.diag.logs._docker_logs", return_value="some output") as mock:
+            results = collect_logs("generic error")
+
+        assert "api" in results
+        mock.assert_called()
+
+    def test_collect_logs_detects_postgres(self):
+        from api.diag.logs import collect_logs
+
+        with patch("api.diag.logs._docker_logs", return_value="log output"):
+            results = collect_logs("psycopg2.OperationalError: connection refused")
+
+        assert "postgres" in results
+        assert "api" in results
+
+    def test_collect_logs_skips_empty(self):
+        from api.diag.logs import collect_logs
+
+        with patch("api.diag.logs._docker_logs", return_value=""):
+            results = collect_logs("generic error")
+
+        assert results == {}
+
+
+class TestIssueBody:
+    def test_build_issue_body(self):
+        from api.diag.blame import BlameInfo
+        from api.diag.issue import build_issue_body
+
+        report = CrashReport(
+            exc_type="ValueError",
+            exc_value="bad value",
+            frames=[
+                Frame(
+                    filepath="src/api/routes/auth.py",
+                    lineno=42,
+                    name="issue_token",
+                    line="raise ValueError('bad value')",
+                    ast_context="FunctionDef issue_token",
+                ),
+            ],
+        )
+        blames = {
+            "src/api/routes/auth.py:42": BlameInfo(
+                sha="a" * 40,
+                author="kert",
+                summary="add auth endpoint",
+                timestamp="1234567890",
+            ),
+        }
+        body = build_issue_body(report, blames, {"api": "error log"}, "abc123")
+
+        assert "abc123" in body
+        assert "ValueError" in body
+        assert "bad value" in body
+        assert "kert" in body
+        assert "add auth endpoint" in body
+        assert "error log" in body
+        assert "FunctionDef issue_token" in body
+
+    def test_build_issue_body_no_blames(self):
+        from api.diag.issue import build_issue_body
+
+        report = CrashReport(
+            exc_type="RuntimeError",
+            exc_value="oops",
+            frames=[],
+        )
+        body = build_issue_body(report, {}, {}, "def456")
+        assert "RuntimeError" in body
+        assert "def456" in body
+
+
+class TestFileIssue:
+    def test_skips_without_token(self, monkeypatch):
+        from api.diag.issue import file_issue
+
+        monkeypatch.delenv("GITEA_TOKEN", raising=False)
+        report = CrashReport(exc_type="E", exc_value="e", frames=[])
+        with patch("api.diag.issue.collect_logs", return_value={}):
+            with patch("conf.secret", return_value=""):
+                result = file_issue(report)
+        assert result is None
+
+    def test_posts_issue_with_token(self, monkeypatch):
+        from api.diag.issue import file_issue
+
+        monkeypatch.setenv("GITEA_TOKEN", "test-token")
+        report = CrashReport(exc_type="ValueError", exc_value="boom", frames=[])
+
+        mock_client = MagicMock()
+        mock_client.create_issue.return_value = {"number": 99, "html_url": "http://..."}
+
+        with (
+            patch("api.diag.issue.collect_logs", return_value={}),
+            patch("api.diag.issue.blame_report", return_value={}),
+            patch("api.clients.gitea.GiteaClient", return_value=mock_client),
+        ):
+            result = file_issue(report)
+
+        assert result["number"] == 99
+        mock_client.create_issue.assert_called_once()
+        call_args = mock_client.create_issue.call_args
+        assert call_args[0][0] == "homelab"
+        assert call_args[0][1] == "stack"
+        assert "ValueError" in call_args[0][2]["title"]
+
+
+class TestHook:
+    def test_install_and_uninstall(self):
+        from api.diag.hook import install, uninstall
+
+        original = sys.excepthook
+        install()
+        assert sys.excepthook is not original
+
+        uninstall()
+        # After uninstall, hook should be restored
+        # (may be the original or sys.__excepthook__)
+
+    def test_install_is_idempotent(self):
+        from api.diag.hook import install, uninstall
+
+        install()
+        hook_after_first = sys.excepthook
+        install()
+        assert sys.excepthook is hook_after_first
+        uninstall()
+
+    def test_hook_skips_keyboard_interrupt(self):
+        from api.diag.hook import _excepthook
+
+        # Should not try to file an issue for KeyboardInterrupt
+        with patch("api.diag.hook._original_hook"):
+            _excepthook(KeyboardInterrupt, KeyboardInterrupt(), None)
+
+
+class TestParseTracebackText:
+    SAMPLE_TB = (
+        "Traceback (most recent call last):\n"
+        '  File "/home/kert/stack/src/api/auth/provision.py", line 55, in provision\n'
+        "    values = derive_all(root_key, commit_sha)\n"
+        '  File "/home/kert/stack/src/api/auth/derive.py", line 28, in derive\n'
+        "    prk = _hkdf_extract(salt, root_key)\n"
+        "ValueError: bad key material\n"
+    )
+
+    def test_parses_single_traceback(self):
+        reports = parse_traceback_text(self.SAMPLE_TB)
+        assert len(reports) == 1
+        r = reports[0]
+        assert r.exc_type == "ValueError"
+        assert r.exc_value == "bad key material"
+        assert len(r.frames) == 2
+        assert r.frames[0].lineno == 55
+        assert r.frames[0].name == "provision"
+        assert r.frames[0].line == "values = derive_all(root_key, commit_sha)"
+        assert r.frames[1].lineno == 28
+
+    def test_parses_multiple_tracebacks(self):
+        text = (
+            self.SAMPLE_TB
+            + "\n"
+            + (
+                "Traceback (most recent call last):\n"
+                '  File "script.py", line 1, in <module>\n'
+                "    import foo\n"
+                "ModuleNotFoundError: No module named 'foo'\n"
+            )
+        )
+        reports = parse_traceback_text(text)
+        assert len(reports) == 2
+        assert reports[0].exc_type == "ValueError"
+        assert reports[1].exc_type == "ModuleNotFoundError"
+
+    def test_returns_empty_for_no_tracebacks(self):
+        reports = parse_traceback_text("just some regular log output\nno errors here")
+        assert reports == []
+
+    def test_handles_exception_without_message(self):
+        text = (
+            "Traceback (most recent call last):\n"
+            '  File "x.py", line 1, in f\n'
+            "    pass\n"
+            "RuntimeError\n"
+        )
+        reports = parse_traceback_text(text)
+        assert len(reports) == 1
+        assert reports[0].exc_type == "RuntimeError"
+        assert reports[0].exc_value == ""
+
+    def test_handles_woodpecker_log_prefix(self):
+        # Woodpecker log lines may have timestamps or prefixes
+        text = (
+            "some setup output\n"
+            "Traceback (most recent call last):\n"
+            '  File "/app/main.py", line 10, in run\n'
+            "    do_thing()\n"
+            "OSError: disk full\n"
+            "step exited with code 1\n"
+        )
+        reports = parse_traceback_text(text)
+        assert len(reports) == 1
+        assert reports[0].exc_type == "OSError"
+        assert reports[0].exc_value == "disk full"
+
+
+class TestCliGuard:
+    def test_returns_fn_result(self):
+        from api.diag.guard import cli_guard
+
+        assert cli_guard(lambda: 0) == 0
+        assert cli_guard(lambda: 42) == 42
+
+    def test_catches_exception_and_returns_2(self):
+        from api.diag.guard import cli_guard
+
+        def boom():
+            raise RuntimeError("kaboom")
+
+        with (
+            patch("api.diag.guard.traceback.print_exc"),
+            patch("api.diag.issue.file_issue", return_value=None),
+            patch("api.diag.issue.collect_logs", return_value={}),
+            patch("api.diag.issue.blame_report", return_value={}),
+        ):
+            result = cli_guard(boom)
+
+        assert result == 2
+
+    def test_handles_keyboard_interrupt(self):
+        from api.diag.guard import cli_guard
+
+        def interrupted():
+            raise KeyboardInterrupt()
+
+        assert cli_guard(interrupted) == 130
+
+    def test_handles_system_exit(self):
+        from api.diag.guard import cli_guard
+
+        def exits():
+            raise SystemExit(7)
+
+        assert cli_guard(exits) == 7
+
+    def test_guarded_decorator(self):
+        from api.diag.guard import guarded
+
+        @guarded
+        def good():
+            return 0
+
+        assert good() == 0
+
+
+class TestCiMain:
+    def test_missing_env_vars(self, monkeypatch):
+        monkeypatch.delenv("CI_REPO_ID", raising=False)
+        monkeypatch.delenv("CI_PIPELINE_NUMBER", raising=False)
+
+        from api.diag.__main__ import main
+
+        assert main() == 1
+
+    def test_no_failed_steps(self, monkeypatch):
+        monkeypatch.setenv("CI_REPO_ID", "1")
+        monkeypatch.setenv("CI_PIPELINE_NUMBER", "42")
+        monkeypatch.setenv("CI_COMMIT_SHA", "abc123")
+        monkeypatch.setenv("CI_REPO", "homelab/stack")
+        monkeypatch.setenv("GITEA_TOKEN", "tok")
+
+        mock_wp = MagicMock()
+        mock_wp.get_pipeline.return_value = {
+            "steps": [{"name": "build", "state": "success"}],
+        }
+
+        with patch("api.clients.woodpecker.WoodpeckerClient", return_value=mock_wp):
+            from api.diag.__main__ import main
+
+            assert main() == 0
+
+    def test_files_issue_for_python_failure(self, monkeypatch):
+        monkeypatch.setenv("CI_REPO_ID", "1")
+        monkeypatch.setenv("CI_PIPELINE_NUMBER", "42")
+        monkeypatch.setenv("CI_COMMIT_SHA", "abc123")
+        monkeypatch.setenv("CI_REPO", "homelab/stack")
+        monkeypatch.setenv("GITEA_TOKEN", "tok")
+
+        mock_wp = MagicMock()
+        mock_wp.get_pipeline.return_value = {
+            "steps": [
+                {"name": "provision", "state": "failure", "id": 5},
+            ],
+        }
+        mock_wp.get_logs.return_value = [
+            {"data": "Traceback (most recent call last):"},
+            {"data": '  File "src/api/auth/provision.py", line 10, in provision'},
+            {"data": "    do_stuff()"},
+            {"data": "RuntimeError: oops"},
+        ]
+
+        mock_gitea = MagicMock()
+        mock_gitea.create_issue.return_value = {"number": 55}
+
+        with (
+            patch("api.clients.woodpecker.WoodpeckerClient", return_value=mock_wp),
+            patch("api.clients.gitea.GiteaClient", return_value=mock_gitea),
+            patch("api.diag.blame.blame_report", return_value={}),
+        ):
+            from api.diag.__main__ import main
+
+            assert main() == 0
+
+        mock_gitea.create_issue.assert_called_once()
+        issue = mock_gitea.create_issue.call_args[0][2]
+        assert "RuntimeError" in issue["title"]
+        assert "provision" in issue["title"]
+
+    def test_files_issue_for_non_python_failure(self, monkeypatch):
+        monkeypatch.setenv("CI_REPO_ID", "1")
+        monkeypatch.setenv("CI_PIPELINE_NUMBER", "42")
+        monkeypatch.setenv("CI_COMMIT_SHA", "abc123")
+        monkeypatch.setenv("CI_REPO", "homelab/stack")
+        monkeypatch.setenv("GITEA_TOKEN", "tok")
+
+        mock_wp = MagicMock()
+        mock_wp.get_pipeline.return_value = {
+            "steps": [
+                {"name": "build-docs", "state": "failure", "id": 3},
+            ],
+        }
+        mock_wp.get_logs.return_value = [
+            {"data": "COPY failed: file not found"},
+            {"data": "ERROR: build failed"},
+        ]
+
+        mock_gitea = MagicMock()
+        mock_gitea.create_issue.return_value = {"number": 56}
+
+        with (
+            patch("api.clients.woodpecker.WoodpeckerClient", return_value=mock_wp),
+            patch("api.clients.gitea.GiteaClient", return_value=mock_gitea),
+        ):
+            from api.diag.__main__ import main
+
+            assert main() == 0
+
+        mock_gitea.create_issue.assert_called_once()
+        issue = mock_gitea.create_issue.call_args[0][2]
+        assert "build-docs" in issue["title"]
+        assert "COPY failed" in issue["body"]