From f156b17b3ecfb692b67e1d7b91d8a46980aa88f3 Mon Sep 17 00:00:00 2001 From: ogt Date: Thu, 9 Jul 2026 19:11:39 +0800 Subject: [PATCH] fix(api): reserve db pool for serving readbacks --- apps/api/src/main.py | 150 +++++++++++++----- .../tests/test_runtime_bootstrap_guards.py | 45 ++++++ 2 files changed, 154 insertions(+), 41 deletions(-) diff --git a/apps/api/src/main.py b/apps/api/src/main.py index 899e53279..cbb1758b5 100644 --- a/apps/api/src/main.py +++ b/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)) diff --git a/apps/api/tests/test_runtime_bootstrap_guards.py b/apps/api/tests/test_runtime_bootstrap_guards.py index 8d1723747..0e75ddaa4 100644 --- a/apps/api/tests/test_runtime_bootstrap_guards.py +++ b/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