fix(api): reserve db pool for serving readbacks

apps/api/src/main.py

@@ -20,8 +20,9 @@ Date: 2026-03-20
 
 import asyncio
 import os
-from collections.abc import AsyncGenerator
+from collections.abc import AsyncGenerator, Coroutine
 from contextlib import asynccontextmanager
+from typing import Any
 from uuid import uuid4
 
 import sentry_sdk
@@ -123,6 +124,35 @@ logger = get_logger("awoooi.api")
 
 ALERTMANAGER_WEBHOOK_PATH = "/api/v1/webhooks/alertmanager"
 ALERTMANAGER_DEFAULT_PROJECT_ID = "awoooi"
+API_BACKGROUND_TASKS_ENV = "AWOOOI_API_BACKGROUND_TASKS_ENABLED"
+_TRUTHY_ENV_VALUES = {"1", "true", "yes", "on"}
+_FALSY_ENV_VALUES = {"0", "false", "no", "off"}
+
+
+def should_start_api_background_tasks() -> bool:
+    """Keep production API serving responsive when the DB pool has no spare slot."""
+    override = os.getenv(API_BACKGROUND_TASKS_ENV)
+    if override is not None:
+        normalized = override.strip().lower()
+        if normalized in _TRUTHY_ENV_VALUES:
+            return True
+        if normalized in _FALSY_ENV_VALUES:
+            return False
+        logger.warning(
+            "api_background_task_env_invalid",
+            env_name=API_BACKGROUND_TASKS_ENV,
+            value=override,
+            fallback="pool_budget",
+        )
+
+    return settings.DATABASE_POOL_SIZE + settings.DATABASE_MAX_OVERFLOW > 1
+
+
+def _close_unscheduled_coroutine(coro: Coroutine[Any, Any, Any]) -> None:
+    try:
+        coro.close()
+    except Exception as exc:
+        logger.warning("api_background_task_close_failed", error=str(exc))
 
 
 def _resolve_request_project_context(request: Request) -> tuple[str | None, str]:
@@ -237,6 +267,31 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
         four_hosts=settings.four_hosts,
         kubeconfig=settings.KUBECONFIG_PATH,
     )
+    api_background_tasks_enabled = should_start_api_background_tasks()
+    logger.info(
+        "api_background_task_budget",
+        enabled=api_background_tasks_enabled,
+        database_pool_size=settings.DATABASE_POOL_SIZE,
+        database_max_overflow=settings.DATABASE_MAX_OVERFLOW,
+        override=os.getenv(API_BACKGROUND_TASKS_ENV),
+    )
+
+    def schedule_api_background_task(
+        coro: Coroutine[Any, Any, Any],
+    ) -> asyncio.Task[Any] | None:
+        code = getattr(coro, "cr_code", None)
+        task_name = getattr(code, "co_name", coro.__class__.__name__)
+        if not api_background_tasks_enabled:
+            _close_unscheduled_coroutine(coro)
+            logger.info(
+                "api_background_task_skipped",
+                task_name=task_name,
+                reason="db_pool_reserved_for_serving_requests",
+                database_pool_size=settings.DATABASE_POOL_SIZE,
+                database_max_overflow=settings.DATABASE_MAX_OVERFLOW,
+            )
+            return None
+        return asyncio.create_task(coro, name=task_name)
 
     # CTO-201: Initialize PostgreSQL database (統帥鐵律: 禁止 SQLite)
     await init_db()
@@ -374,7 +429,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
     # 必須在 embedding indexing 之前，確保 playbook 表有資料
     try:
         from src.services.playbook_seed_service import seed_playbooks_from_rules
-        asyncio.create_task(seed_playbooks_from_rules())
+        schedule_api_background_task(seed_playbooks_from_rules())
         logger.info("playbook_seed_scheduled")
     except Exception as e:
         logger.warning("playbook_seed_schedule_failed", error=str(e))
@@ -384,7 +439,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
     # 2026-04-15 ogt + Claude Sonnet 4.6（亞太）: Phase 3.5 AI 學習成果持久化
     try:
         from src.repositories.playbook_repository import get_playbook_repository
-        asyncio.create_task(get_playbook_repository().backfill_redis_to_pg())
+        schedule_api_background_task(get_playbook_repository().backfill_redis_to_pg())
         logger.info("playbook_pg_backfill_scheduled")
     except Exception as e:
         logger.warning("playbook_pg_backfill_schedule_failed", error=str(e))
@@ -393,21 +448,28 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
         from src.services.playbook_embedding_service import (
             ensure_playbook_embeddings_indexed,
         )
-        asyncio.create_task(ensure_playbook_embeddings_indexed())
+        schedule_api_background_task(ensure_playbook_embeddings_indexed())
         logger.info("playbook_embedding_indexing_scheduled")
     except Exception as e:
         logger.warning("playbook_embedding_schedule_failed", error=str(e))
 
     # Phase 6.1: 啟動 Signal Worker (Redis Streams Consumer)
     # 統帥鐵律: Event Bus 解耦告警接收與處理
-    await init_signal_worker()
-    logger.info("signal_worker_initialized")
+    if api_background_tasks_enabled:
+        await init_signal_worker()
+        logger.info("signal_worker_initialized")
+    else:
+        logger.info(
+            "signal_worker_skipped",
+            reason="api_db_pool_reserved_for_serving_requests",
+            handoff="awoooi-worker deployment owns Redis stream consumption",
+        )
 
     # BUG-005 修復 2026-04-11: 啟動時掃描 Redis 中所有 state=ready 但未送 Telegram 的 token
     # dedup TTL 10 分鐘過期後，ready decisions 就沒有補送機制 → 長期卡在 ready 無人審核
     try:
         from src.services.decision_manager import get_decision_manager
-        asyncio.create_task(get_decision_manager().resend_stale_ready_tokens())
+        schedule_api_background_task(get_decision_manager().resend_stale_ready_tokens())
         logger.info("stale_ready_tokens_resend_scheduled")
     except Exception as e:
         logger.warning("stale_ready_tokens_resend_schedule_failed", error=str(e))
@@ -418,7 +480,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
     # Sweeper 定期掃描無 decision token 的 INVESTIGATING incidents → 背景觸發
     try:
         from src.jobs.incident_analysis_sweeper import run_incident_analysis_sweeper
-        asyncio.create_task(run_incident_analysis_sweeper())
+        schedule_api_background_task(run_incident_analysis_sweeper())
         logger.info("incident_analysis_sweeper_scheduled", interval_sec=90)
     except Exception as e:
         logger.warning("incident_analysis_sweeper_schedule_failed", error=str(e))
@@ -428,7 +490,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
     # 解開 8 張 0 writer 表的第一個 (asset_inventory / asset_discovery_run / asset_coverage_snapshot)
     try:
         from src.jobs.asset_scanner_job import run_asset_scanner_loop
-        asyncio.create_task(run_asset_scanner_loop())
+        schedule_api_background_task(run_asset_scanner_loop())
         logger.info("asset_scanner_loop_scheduled", interval_sec=3600)
     except Exception as e:
         logger.warning("asset_scanner_loop_schedule_failed", error=str(e))
@@ -438,7 +500,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
     # 解鎖 E3 Hermes 自動建規則: AI 需要 alert_rule_catalog 作為 baseline 才能提案修正
     try:
         from src.jobs.rule_catalog_sync_job import run_rule_catalog_sync_loop
-        asyncio.create_task(run_rule_catalog_sync_loop())
+        schedule_api_background_task(run_rule_catalog_sync_loop())
         logger.info("rule_catalog_sync_loop_scheduled", interval_sec=3600)
     except Exception as e:
         logger.warning("rule_catalog_sync_loop_schedule_failed", error=str(e))
@@ -448,7 +510,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
     # 解鎖: Phase 4 Holt-Winters 預測有歷史資料 / 容量趨勢分析
     try:
         from src.jobs.capacity_scanner_job import run_capacity_scanner_loop
-        asyncio.create_task(run_capacity_scanner_loop())
+        schedule_api_background_task(run_capacity_scanner_loop())
         logger.info("capacity_scanner_loop_scheduled", daily_trigger_hour_taipei=2)
     except Exception as e:
         logger.warning("capacity_scanner_loop_schedule_failed", error=str(e))
@@ -458,7 +520,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
     # MVP: secret_rotated 真實檢查,其他 6 維占位 'unknown',後續 agent 補
     try:
         from src.jobs.compliance_scanner_job import run_compliance_scanner_loop
-        asyncio.create_task(run_compliance_scanner_loop())
+        schedule_api_background_task(run_compliance_scanner_loop())
         logger.info("compliance_scanner_loop_scheduled", daily_trigger_hour_taipei=3)
     except Exception as e:
         logger.warning("compliance_scanner_loop_schedule_failed", error=str(e))
@@ -468,7 +530,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
     try:
         from src.workers.aider_event_processor import run_aider_event_processor_loop
         logger.info("aider_event_processor: starting Redis stream consumer")
-        asyncio.create_task(run_aider_event_processor_loop())
+        schedule_api_background_task(run_aider_event_processor_loop())
     except Exception as e:
         logger.warning("aider_event_processor_schedule_failed", error=str(e))
 
@@ -477,7 +539,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
     # 依據: Prometheus targets / alert_rule_catalog labels / knowledge_entries 覆蓋
     try:
         from src.jobs.coverage_evaluator_job import run_coverage_evaluator_loop
-        asyncio.create_task(run_coverage_evaluator_loop())
+        schedule_api_background_task(run_coverage_evaluator_loop())
         logger.info("coverage_evaluator_loop_scheduled", interval_sec=3600)
     except Exception as e:
         logger.warning("coverage_evaluator_loop_schedule_failed", error=str(e))
@@ -487,7 +549,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
     # 解鎖 E3 Hermes: noise_rate > 0.5 的 rule 可被 AI 提案 deprecate
     try:
         from src.jobs.rule_stats_updater_job import run_rule_stats_updater_loop
-        asyncio.create_task(run_rule_stats_updater_loop())
+        schedule_api_background_task(run_rule_stats_updater_loop())
         logger.info("rule_stats_updater_loop_scheduled", interval_sec=3600)
     except Exception as e:
         logger.warning("rule_stats_updater_loop_schedule_failed", error=str(e))
@@ -497,7 +559,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
     # 解鎖: 資產變化歷史 (added/removed/lifecycle_changed),AI 可追蹤集群演進
     try:
         from src.jobs.asset_change_tracker_job import run_asset_change_tracker_loop
-        asyncio.create_task(run_asset_change_tracker_loop())
+        schedule_api_background_task(run_asset_change_tracker_loop())
         logger.info("asset_change_tracker_loop_scheduled", interval_sec=3600)
     except Exception as e:
         logger.warning("asset_change_tracker_loop_schedule_failed", error=str(e))
@@ -507,7 +569,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
     # 統帥鐵律: AI 只推建議不自動改 review_status,人工決策 deprecate
     try:
         from src.jobs.hermes_rule_quality_job import run_hermes_rule_quality_loop
-        asyncio.create_task(run_hermes_rule_quality_loop())
+        schedule_api_background_task(run_hermes_rule_quality_loop())
         logger.info("hermes_rule_quality_loop_scheduled", daily_trigger_hour_taipei=4)
     except Exception as e:
         logger.warning("hermes_rule_quality_loop_schedule_failed", error=str(e))
@@ -517,7 +579,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
     # 高風險 host 寫 aol(capacity_recommendation) + Telegram 建議
     try:
         from src.jobs.capacity_forecaster_job import run_capacity_forecaster_loop
-        asyncio.create_task(run_capacity_forecaster_loop())
+        schedule_api_background_task(run_capacity_forecaster_loop())
         logger.info("capacity_forecaster_loop_scheduled", daily_trigger_hour_taipei=5)
     except Exception as e:
         logger.warning("capacity_forecaster_loop_schedule_failed", error=str(e))
@@ -530,9 +592,9 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
             run_monthly_report_loop,
             run_weekly_report_loop,
         )
-        asyncio.create_task(run_daily_report_loop())
-        asyncio.create_task(run_weekly_report_loop())
-        asyncio.create_task(run_monthly_report_loop())
+        schedule_api_background_task(run_daily_report_loop())
+        schedule_api_background_task(run_weekly_report_loop())
+        schedule_api_background_task(run_monthly_report_loop())
         logger.info(
             "report_delivery_loops_scheduled",
             daily_hour_taipei=8,
@@ -546,7 +608,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
     # 確保 PENDING approval 超過 48h 後觸發 resolve_incident → KM 學習鏈閉環
     try:
         from src.jobs.approval_timeout_resolver import run_approval_timeout_resolver
-        asyncio.create_task(run_approval_timeout_resolver())
+        schedule_api_background_task(run_approval_timeout_resolver())
         logger.info("approval_timeout_resolver_scheduled", interval_sec=3600)
     except Exception as e:
         logger.warning("approval_timeout_resolver_schedule_failed", error=str(e))
@@ -561,7 +623,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
         from src.jobs.incident_lifecycle_reconciler import (
             run_incident_lifecycle_reconciler_loop,
         )
-        asyncio.create_task(run_incident_lifecycle_reconciler_loop())
+        schedule_api_background_task(run_incident_lifecycle_reconciler_loop())
         logger.info(
             "incident_lifecycle_reconciler_scheduled",
             interval_sec=INCIDENT_LIFECYCLE_RECONCILER_INTERVAL,
@@ -576,7 +638,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
         from src.jobs.awooop_ansible_candidate_backfill_job import (
             run_awooop_ansible_candidate_backfill_loop,
         )
-        asyncio.create_task(run_awooop_ansible_candidate_backfill_loop())
+        schedule_api_background_task(run_awooop_ansible_candidate_backfill_loop())
         logger.info(
             "awooop_ansible_candidate_backfill_worker_scheduled",
             enabled=settings.ENABLE_AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WORKER,
@@ -592,7 +654,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
         from src.jobs.awooop_ansible_check_mode_job import (
             run_awooop_ansible_check_mode_loop,
         )
-        asyncio.create_task(run_awooop_ansible_check_mode_loop())
+        schedule_api_background_task(run_awooop_ansible_check_mode_loop())
         logger.info(
             "awooop_ansible_check_mode_worker_scheduled",
             enabled=settings.ENABLE_AWOOOP_ANSIBLE_CHECK_MODE_WORKER,
@@ -605,7 +667,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
     # 2026-04-15 ogt + Claude Sonnet 4.6（亞太）: Phase 3 初始建立
     try:
         from src.services.playbook_evolver import run_evolver_loop
-        asyncio.create_task(run_evolver_loop())
+        schedule_api_background_task(run_evolver_loop())
         logger.info("evolver_loop_scheduled", interval_sec=86400)
     except Exception as e:
         logger.warning("evolver_loop_schedule_failed", error=str(e))
@@ -615,7 +677,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
         from src.jobs.playbook_generation_governance_job import (
             run_playbook_generation_governance_loop,
         )
-        asyncio.create_task(run_playbook_generation_governance_loop())
+        schedule_api_background_task(run_playbook_generation_governance_loop())
         logger.info(
             "playbook_generation_governance_loop_scheduled",
             interval_sec=settings.PLAYBOOK_DRAFT_GOVERNANCE_INTERVAL_SECONDS,
@@ -627,7 +689,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
     # 2026-04-15 ogt + Claude Sonnet 4.6（亞太）: Phase 3 初始建立
     try:
         from src.jobs.knowledge_decay_job import run_knowledge_decay_loop
-        asyncio.create_task(run_knowledge_decay_loop())
+        schedule_api_background_task(run_knowledge_decay_loop())
         logger.info("knowledge_decay_loop_scheduled", interval_sec=86400)
     except Exception as e:
         logger.warning("knowledge_decay_loop_schedule_failed", error=str(e))
@@ -637,7 +699,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
     # Feature Flag: ENABLE_KM_BACKFILL_RECONCILER=false 停用（回滾用）
     try:
         from src.jobs.km_backfill_reconciler_job import run_km_backfill_reconciler_loop
-        asyncio.create_task(run_km_backfill_reconciler_loop())
+        schedule_api_background_task(run_km_backfill_reconciler_loop())
         logger.info("km_backfill_reconciler_loop_scheduled", interval_sec=300)
     except Exception as e:
         logger.warning("km_backfill_reconciler_loop_schedule_failed", error=str(e))
@@ -648,7 +710,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
     # ADR-091 Task T2
     try:
         from src.jobs.aol_to_catalog_writeback_job import run_aol_writeback_loop
-        asyncio.create_task(run_aol_writeback_loop())
+        schedule_api_background_task(run_aol_writeback_loop())
         logger.info("aol_to_catalog_writeback_loop_scheduled", interval_sec=3600)
     except Exception as e:
         logger.warning("aol_to_catalog_writeback_loop_schedule_failed", error=str(e))
@@ -693,7 +755,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
                     logger.exception("kb_rot_cleaner_loop_error", error=str(_e))
                     await _asyncio.sleep(3600)  # 1h 後重試
 
-        asyncio.create_task(_run_kb_rot_cleaner_loop())
+        schedule_api_background_task(_run_kb_rot_cleaner_loop())
         logger.info("kb_rot_cleaner_loop_scheduled", trigger="monthly_day1_03h_taipei")
     except Exception as e:
         logger.warning("kb_rot_cleaner_loop_schedule_failed", error=str(e))
@@ -702,7 +764,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
     # 2026-04-15 ogt + Claude Sonnet 4.6（亞太）: Phase 3 初始建立
     try:
         from src.services.finetune_exporter import run_finetune_export_loop
-        asyncio.create_task(run_finetune_export_loop())
+        schedule_api_background_task(run_finetune_export_loop())
         logger.info("finetune_export_loop_scheduled", interval_sec=604800)
     except Exception as e:
         logger.warning("finetune_export_loop_schedule_failed", error=str(e))
@@ -713,7 +775,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
     # 2026-04-15 ogt + Claude Sonnet 4.6（亞太）: Phase 4 初始建立
     try:
         from src.services.proactive_inspector import run_proactive_inspector_loop
-        asyncio.create_task(run_proactive_inspector_loop())
+        schedule_api_background_task(run_proactive_inspector_loop())
         logger.info("proactive_inspector_loop_scheduled", interval_sec=300)
     except Exception as e:
         logger.warning("proactive_inspector_schedule_failed", error=str(e))
@@ -722,7 +784,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
     # 2026-04-15 ogt + Claude Sonnet 4.6（亞太）: Phase 6 初始建立
     try:
         from src.jobs.offline_replay_service import run_offline_replay_loop
-        asyncio.create_task(run_offline_replay_loop())
+        schedule_api_background_task(run_offline_replay_loop())
         logger.info("offline_replay_loop_scheduled", interval_sec=604800)
     except Exception as e:
         logger.warning("offline_replay_loop_schedule_failed", error=str(e))
@@ -731,7 +793,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
     # MASTER §1.1：系統必須能感知自身故障，W-1 SLO + W-2 Telegram靜默 + W-3 飛輪成功率
     try:
         from src.jobs.ai_slo_watchdog_job import run_ai_slo_watchdog_loop
-        asyncio.create_task(run_ai_slo_watchdog_loop())
+        schedule_api_background_task(run_ai_slo_watchdog_loop())
         logger.info("ai_slo_watchdog_scheduled", interval_sec=900)
     except Exception as e:
         logger.warning("ai_slo_watchdog_schedule_failed", error=str(e))
@@ -740,7 +802,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
     # MASTER P2.2：trust_drift / knowledge_degradation / llm_hallucination / execution_blast_radius
     try:
         from src.services.governance_agent import run_governance_loop
-        asyncio.create_task(run_governance_loop())
+        schedule_api_background_task(run_governance_loop())
         logger.info("governance_agent_scheduled", interval_sec=3600)
     except Exception as e:
         logger.warning("governance_agent_schedule_failed", error=str(e))
@@ -748,7 +810,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
     # 2026-05-03 ogt + Claude Sonnet 4.6（亞太）: GovernanceDispatcher Wave 2E（每 30s poll）
     try:
         from src.services.governance_dispatcher import run_governance_dispatcher_loop
-        asyncio.create_task(run_governance_dispatcher_loop())
+        schedule_api_background_task(run_governance_dispatcher_loop())
         logger.info("governance_dispatcher_scheduled", interval_sec=30)
     except Exception as e:
         logger.warning("governance_dispatcher_schedule_failed", error=str(e))
@@ -758,7 +820,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
     # 只產生 REVIEW 草稿並停在 owner review，不直接批准或發布 KM。
     try:
         from src.jobs.hermes_kb_growth_worker import run_hermes_kb_growth_loop
-        asyncio.create_task(run_hermes_kb_growth_loop())
+        schedule_api_background_task(run_hermes_kb_growth_loop())
         logger.info("hermes_kb_growth_worker_scheduled", interval_sec=300)
     except Exception as e:
         logger.warning("hermes_kb_growth_worker_schedule_failed", error=str(e))
@@ -819,7 +881,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
                     logger.exception("model_version_tracker_loop_error", error=str(_loop_err))
                     await asyncio.sleep(60)  # 錯誤後 1 分鐘重試
 
-        asyncio.create_task(_run_model_version_tracker_loop())
+        schedule_api_background_task(_run_model_version_tracker_loop())
         logger.info("model_version_tracker_scheduled", interval_sec=3600)
     except Exception as e:
         logger.warning("model_version_tracker_schedule_failed", error=str(e))
@@ -829,8 +891,14 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
     # Shadow mode：is_shadow=True，0 user-visible response，0 destructive tool call
     try:
         from src.workers.platform_worker import start_platform_worker
-        await start_platform_worker()
-        logger.info("platform_worker_started", mode="shadow")
+        if api_background_tasks_enabled:
+            await start_platform_worker()
+            logger.info("platform_worker_started", mode="shadow")
+        else:
+            logger.info(
+                "platform_worker_skipped",
+                reason="api_db_pool_reserved_for_serving_requests",
+            )
     except Exception as e:
         logger.warning("platform_worker_start_failed", error=str(e))
 

apps/api/tests/test_runtime_bootstrap_guards.py

@@ -257,6 +257,51 @@ def test_api_lifespan_closes_worker_redis_pool_after_signal_worker() -> None:
     assert source.index("close_worker_redis_pool") < source.index("close_redis_pool")
 
 
+def test_api_background_tasks_default_off_when_db_pool_has_no_spare_slot(monkeypatch) -> None:
+    from src import main as api_main
+
+    monkeypatch.delenv(api_main.API_BACKGROUND_TASKS_ENV, raising=False)
+    monkeypatch.setattr(api_main.settings, "DATABASE_POOL_SIZE", 1)
+    monkeypatch.setattr(api_main.settings, "DATABASE_MAX_OVERFLOW", 0)
+
+    assert api_main.should_start_api_background_tasks() is False
+
+
+def test_api_background_tasks_default_on_when_db_pool_has_spare_slot(monkeypatch) -> None:
+    from src import main as api_main
+
+    monkeypatch.delenv(api_main.API_BACKGROUND_TASKS_ENV, raising=False)
+    monkeypatch.setattr(api_main.settings, "DATABASE_POOL_SIZE", 2)
+    monkeypatch.setattr(api_main.settings, "DATABASE_MAX_OVERFLOW", 0)
+
+    assert api_main.should_start_api_background_tasks() is True
+
+
+def test_api_background_tasks_env_override_controls_serving_budget(monkeypatch) -> None:
+    from src import main as api_main
+
+    monkeypatch.setattr(api_main.settings, "DATABASE_POOL_SIZE", 1)
+    monkeypatch.setattr(api_main.settings, "DATABASE_MAX_OVERFLOW", 0)
+
+    monkeypatch.setenv(api_main.API_BACKGROUND_TASKS_ENV, "true")
+    assert api_main.should_start_api_background_tasks() is True
+
+    monkeypatch.setenv(api_main.API_BACKGROUND_TASKS_ENV, "false")
+    assert api_main.should_start_api_background_tasks() is False
+
+
+def test_api_lifespan_routes_background_loops_through_serving_budget_gate() -> None:
+    import inspect
+
+    from src import main as api_main
+
+    source = inspect.getsource(api_main.lifespan)
+    assert "schedule_api_background_task(run_incident_analysis_sweeper())" in source
+    assert "schedule_api_background_task(run_km_backfill_reconciler_loop())" in source
+    assert "signal_worker_skipped" in source
+    assert "platform_worker_skipped" in source
+
+
 def test_init_db_backfills_all_optional_knowledge_read_columns() -> None:
     import inspect