src/lib/background.py

"""Process-lifetime fire-and-forget asyncio task helper.

【為什麼存在？】
  某些 endpoint 想 kick off 慢的 side-effect 卻不要 block response。直接
  `asyncio.create_task(coro)` 會回 Task，但 event loop 只 weakref 著它；
  caller 沒留 strong ref 的話 GC 可能在 await 中途 finalize 它，工作
  silently 消失（CPython 實作層細節，文件有寫）。

【pattern】
  建 task 後立刻塞進 process-lifetime 的 set；task 結束時 done callback
  把它從 set 拿掉。caller 完全不需要 await 或檢查狀態 — 真正的
  fire-and-forget。

【目前 caller】
  - /api/v1/ml/valuation/predict (F4 v2 (A) CR-3 fix) — observation pipeline
    做 ~3s SSRF guard + httpx download + torch forward + psycopg INSERT。
    await 它讓每個帶 image_url 的 predict call 多 ~3s latency 在 critical path。
    record_observation 已經 universally fail-open（每環節 try/except → return
    without raising），把 await 換成 fire_and_forget 的失誤代價是「worker
    SIGKILL 時 in-flight task 丟失」，跟原 sync 版的 worker SIGKILL 失誤代價
    一樣（兩者都不 durable）。

【為什麼不用 FastAPI BackgroundTasks？】
  BackgroundTasks 的語意是「response 送出後執行」，會等 response 寫完 socket
  才開始 — 我們要的是「prediction 算完同時並行寫 observation」，create_task
  能讓 record_observation 的 download 部分在 prediction return 之前先跑。
"""

from __future__ import annotations

import asyncio
import logging
from collections.abc import Coroutine
from typing import Any

logger = logging.getLogger(__name__)

# Strong-ref store: 沒有它 GC 可能在 task await 中途 collect。
# 模組級 set，process lifetime 共用。
_BACKGROUND_TASKS: set[asyncio.Task[Any]] = set()


def fire_and_forget(
    coro: Coroutine[Any, Any, Any],
    *,
    name: str | None = None,
) -> asyncio.Task[Any]:
    """排程 `coro` 為 background task，保留 strong ref 直到結束。

    回傳的 Task 給 test / introspection 用；prod caller 可以丟掉不接。
    """
    task = asyncio.create_task(coro, name=name)
    _BACKGROUND_TASKS.add(task)
    task.add_done_callback(_BACKGROUND_TASKS.discard)
    task.add_done_callback(_log_unhandled_exception)
    return task


def _log_unhandled_exception(task: asyncio.Task[Any]) -> None:
    """把 task 內未處理的例外寫到 log + Sentry，不然會被 asyncio runtime 默默吞掉。

    record_observation 文件保證不 raise，但未來新 caller 可能傳會 raise
    的 coroutine。done callback 在 task 結束後跑；呼叫 .exception() 會
    consume 例外（避免 "Task exception was never retrieved" warning），
    同時 log warning 讓 observability 看得到。

    Sentry capture 是 batch-3 post-review 補的（security-reviewer P2 conf 80）：
    sentry-sdk 2.x 的 AsyncioIntegration 只在「task exception 沒被 retrieve」
    時 hook 進來。我們在這個 callback 裡顯式呼叫 task.exception() 把例外
    consume 掉了，等於主動關掉 AsyncioIntegration 的捕獲路徑。
    需要顯式 capture 才能讓 fire-and-forget escapes 進到 Sentry 告警鏈
    （ml-service.error-rate-spike / ml-service.5xx-rate）。

    F3 deferred sweep（2026-05-03）—— 直接 capture_exception(exc) 會把
    exc.args（psycopg OperationalError 的 Neon DSN / huggingface_hub 401 的
    hf_ token）原樣寫進 event 的 exception.values[].value。Sentry-only
    reader 拿到 plaintext credentials。改走 capture_message +
    scrub_exc_for_audit() opaque label。Sibling site: main.py:_on_preload_done。
    memory: feedback_python_sentry_capture_needs_scrub。

    sentry_sdk import 是 lazy 的：DSN 沒設時 init_sentry 從沒呼叫過
    sentry_sdk.init，capture_message 是 no-op；DSN 有設時就會 capture。
    都不會 raise，所以這條 path 對 caller 完全透明。
    """
    if task.cancelled():
        return
    exc = task.exception()
    if exc is not None:
        # Local logger 仍寫 unredacted exc — operator 從 Render / HF Spaces
        # stdout 撈 trace。Sentry 只見 scrubbed label。
        logger.warning(
            "background task %s raised: %s",
            task.get_name(),
            exc,
        )
        try:
            import sentry_sdk

            from src.lib.error_scrub import scrub_exc_for_audit

            # F2/F5 closer (c-series wave-2, hardened in wave-3)：必須用
            # isolation_scope，不是 new_scope。sentry-sdk Python 2.x scope 模型
            # 拆三層（global / isolation / current），breadcrumb 寫在 isolation
            # scope；上面 logger.warning("...%s", exc) 經 LoggingIntegration
            # level=INFO 把 str(exc) 塞進 ambient isolation breadcrumb buffer。
            # `new_scope()` 只 fork current scope，clear_breadcrumbs 對 isolation
            # scope 沒效（empirically verified sentry-sdk 2.58.0：event 仍夾
            # ambient breadcrumb）；`isolation_scope()` fork isolation 才真清。
            # 鏡像 TS `Sentry.withScope`（TS scope 不分層）。
            # memory: feedback_sentry_breadcrumb_bypasses_message_scrub。
            with sentry_sdk.isolation_scope() as scope:
                scope.clear_breadcrumbs()
                sentry_sdk.capture_message(
                    f"background-task-failed[{task.get_name()}]: "
                    f"{scrub_exc_for_audit(exc)}",
                    level="error",
                )
        except Exception:
            # Sentry 失敗不 escalate（done_callback 內部）— logger 已經
            # 留下 trail，operator 仍可從 Render/HF Spaces stdout 撈到。
            pass


def background_task_count() -> int:
    """Test-only — 回 pending background task 數量。"""
    return len(_BACKGROUND_TASKS)