From 47586431c0f3f93a43aca4d7f7528f5bb80cf411 Mon Sep 17 00:00:00 2001 From: ogt Date: Wed, 15 Jul 2026 08:18:17 +0800 Subject: [PATCH] fix(automation): run asset reconciliation on worker --- .../asset_capability_reconciliation_job.py | 144 ++++++++++------ apps/api/src/main.py | 157 +++++++++++++----- .../ai_automation_asset_capability_matrix.py | 1 + apps/api/src/workers/signal_worker.py | 35 ++-- ...t_ai_automation_asset_capability_matrix.py | 81 +++++++++ ..._signal_worker_ansible_executor_binding.py | 63 +++++-- 6 files changed, 365 insertions(+), 116 deletions(-) diff --git a/apps/api/src/jobs/asset_capability_reconciliation_job.py b/apps/api/src/jobs/asset_capability_reconciliation_job.py index 33625980f..8073fbed3 100644 --- a/apps/api/src/jobs/asset_capability_reconciliation_job.py +++ b/apps/api/src/jobs/asset_capability_reconciliation_job.py @@ -33,6 +33,8 @@ logger = structlog.get_logger(__name__) _FIRST_DELAY_SECONDS = 90 _LOOP_BACKOFF_SECONDS = 30 * 60 +_MAX_RECONCILIATION_CANDIDATES = 20_000 +_RETRYABLE_RESULT_STATUSES = {"blocked_safe_no_write", "degraded_no_write"} _DAILY_TRIGGER_HOUR_TAIPEI = 3 _DAILY_TRIGGER_MINUTE_TAIPEI = 30 _TAIPEI = ZoneInfo("Asia/Taipei") @@ -187,17 +189,30 @@ async def run_asset_capability_reconciliation_loop() -> None: daily_trigger_minute_taipei=_DAILY_TRIGGER_MINUTE_TAIPEI, ) await asyncio.sleep(_FIRST_DELAY_SECONDS) + triggered_by = "startup" while True: try: - await reconcile_once() + result = await reconcile_once(triggered_by=triggered_by) except Exception as exc: logger.exception( "asset_capability_reconciliation_loop_error", error_type=type(exc).__name__, ) await asyncio.sleep(_LOOP_BACKOFF_SECONDS) + triggered_by = "retry" + continue + if str(result.get("status") or "") in _RETRYABLE_RESULT_STATUSES: + logger.warning( + "asset_capability_reconciliation_retry_scheduled", + status=result.get("status"), + reason=result.get("reason"), + retry_seconds=_LOOP_BACKOFF_SECONDS, + ) + await asyncio.sleep(_LOOP_BACKOFF_SECONDS) + triggered_by = "retry" continue await asyncio.sleep(_seconds_until_next_trigger()) + triggered_by = "cron" async def reconcile_once( @@ -231,6 +246,19 @@ async def reconcile_once( } candidates = build_reconciliation_candidates(matrix) + if len(candidates) > _MAX_RECONCILIATION_CANDIDATES: + logger.error( + "asset_capability_reconciliation_candidate_bound_exceeded", + candidate_count=len(candidates), + max_candidate_count=_MAX_RECONCILIATION_CANDIDATES, + external_runtime_write_performed=False, + ) + return { + "status": "blocked_safe_no_write", + "reason": "candidate_count_above_controlled_execution_bound", + "candidate_count": len(candidates), + "max_candidate_count": _MAX_RECONCILIATION_CANDIDATES, + } receipt = await _persist_reconciliation( matrix, candidates, @@ -284,6 +312,8 @@ async def _persist_reconciliation( AND COALESCE(input ->> 'semantic_operation_type', operation_type) = :semantic_operation_type AND input ->> 'idempotency_key' = :idempotency_key + AND status = 'success' + AND output ->> 'closed' = 'true' ORDER BY created_at DESC LIMIT 1 """ @@ -298,28 +328,38 @@ async def _persist_reconciliation( if isinstance(prior, dict): return prior - for candidate_raw in candidates: - candidate = dict(candidate_raw) + candidate_fingerprints = [ + str(candidate.get("candidate_fingerprint") or "") + for candidate in candidates + ] + existing_fingerprints: set[str] = set() + if candidate_fingerprints: existing = await db.execute( text( """ - SELECT op_id::text + SELECT DISTINCT input ->> 'candidate_fingerprint' FROM automation_operation_log WHERE actor = :actor AND COALESCE(input ->> 'semantic_operation_type', operation_type) = :semantic_operation_type - AND input ->> 'candidate_fingerprint' = :candidate_fingerprint - ORDER BY created_at DESC - LIMIT 1 + AND input ->> 'candidate_fingerprint' + = ANY(CAST(:candidate_fingerprints AS text[])) """ ), { "actor": RECONCILIATION_ACTOR, "semantic_operation_type": RECONCILIATION_WORK_ITEM_OPERATION, - "candidate_fingerprint": candidate["candidate_fingerprint"], + "candidate_fingerprints": candidate_fingerprints, }, ) - if existing.scalar_one_or_none(): + existing_fingerprints = { + str(row[0]) for row in existing.fetchall() if row[0] + } + + insert_rows: list[dict[str, Any]] = [] + for candidate_raw in candidates: + candidate = dict(candidate_raw) + if candidate["candidate_fingerprint"] in existing_fingerprints: continue input_payload = build_reconciliation_work_item_input( candidate, @@ -334,6 +374,29 @@ async def _persist_reconciliation( "post_verifier": "pending_batch_readback", "rollback_or_no_write_terminal": "no_external_runtime_write", } + insert_rows.append( + { + "asset_id": candidate.get("asset_inventory_id"), + "discovery_run_id": discovery_run_id, + "actor": RECONCILIATION_ACTOR, + "input": _canonical_json(input_payload), + "output": _canonical_json(output_payload), + "dry_run_result": _canonical_json( + { + "would_create_work_item": True, + "external_runtime_write": False, + "candidate_fingerprint": candidate["candidate_fingerprint"], + } + ), + "tags": [ + "asset_capability_reconciliation", + "work_item", + "aia_p0_006", + ], + } + ) + + if insert_rows: await db.execute( text( """ @@ -360,50 +423,37 @@ async def _persist_reconciliation( ) """ ), - { - "asset_id": candidate.get("asset_inventory_id"), - "discovery_run_id": discovery_run_id, - "actor": RECONCILIATION_ACTOR, - "input": _canonical_json(input_payload), - "output": _canonical_json(output_payload), - "dry_run_result": _canonical_json( - { - "would_create_work_item": True, - "external_runtime_write": False, - "candidate_fingerprint": candidate["candidate_fingerprint"], - } - ), - "tags": [ - "asset_capability_reconciliation", - "work_item", - "aia_p0_006", - ], - }, + insert_rows, ) - created_count += 1 + created_count = len(insert_rows) - rows_result = await db.execute( - text( - """ - SELECT input ->> 'candidate_fingerprint' AS candidate_fingerprint - FROM automation_operation_log - WHERE actor = :actor - AND COALESCE(input ->> 'semantic_operation_type', operation_type) - = :semantic_operation_type - """ - ), - { - "actor": RECONCILIATION_ACTOR, - "semantic_operation_type": RECONCILIATION_WORK_ITEM_OPERATION, - }, - ) - persisted_fingerprints = { - str(row[0]) for row in rows_result.fetchall() if row[0] - } expected_fingerprints = { str(candidate.get("candidate_fingerprint") or "") for candidate in candidates } + persisted_fingerprints: set[str] = set() + if candidate_fingerprints: + rows_result = await db.execute( + text( + """ + SELECT DISTINCT input ->> 'candidate_fingerprint' + FROM automation_operation_log + WHERE actor = :actor + AND COALESCE(input ->> 'semantic_operation_type', operation_type) + = :semantic_operation_type + AND input ->> 'candidate_fingerprint' + = ANY(CAST(:candidate_fingerprints AS text[])) + """ + ), + { + "actor": RECONCILIATION_ACTOR, + "semantic_operation_type": RECONCILIATION_WORK_ITEM_OPERATION, + "candidate_fingerprints": candidate_fingerprints, + }, + ) + persisted_fingerprints = { + str(row[0]) for row in rows_result.fetchall() if row[0] + } repository_readback_count = len(expected_fingerprints & persisted_fingerprints) summary_output = build_reconciliation_summary_payload( matrix, diff --git a/apps/api/src/main.py b/apps/api/src/main.py index f9ca12fce..ed734b331 100644 --- a/apps/api/src/main.py +++ b/apps/api/src/main.py @@ -191,6 +191,7 @@ def _resolve_request_project_context(request: Request) -> tuple[str | None, str] return None, "request.project_id.missing" + # ============================================================================= # Sentry SDK Initialization (Error Tracking - 補強 SignOz) # Self-Hosted @ 192.168.0.110 @@ -274,6 +275,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: # AwoooP Phase 2.4 (2026-05-04 ogt): 設定 startup handler 的 project_id context # asyncio.create_task() 自動繼承父任務的 ContextVar → 31 個 background loop 全部標記為 awoooi from src.core.context import PROJECT_ID + PROJECT_ID.set("awoooi") # Startup @@ -345,6 +347,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: # ADR-015: MCP Provider 註冊 (DI 模式) from src.plugins.mcp.providers import register_all_providers + register_all_providers() logger.info("mcp_providers_registered") @@ -352,6 +355,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: # 2026-04-16 ogt + Claude Sonnet 4.6: 修復 sensors=0/0 根因 — init 從未在 startup 被呼叫 try: from src.services.mcp_tool_registry import init_mcp_tool_registry + await init_mcp_tool_registry() logger.info("mcp_tool_registry_initialized") except Exception as e: @@ -367,7 +371,9 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: ) logger.info("telegram_heartbeat_monitor_started", interval_minutes=30) else: - logger.warning("telegram_heartbeat_monitor_skipped", reason="OPENCLAW_TG_BOT_TOKEN not set") + logger.warning( + "telegram_heartbeat_monitor_skipped", reason="OPENCLAW_TG_BOT_TOKEN not set" + ) # Reboot Recovery: Warm-up Redis Working Memory from PostgreSQL # 2026-04-05 ogt: 重開機後 Redis 清空，從 DB restore 未解決的 incidents @@ -392,10 +398,12 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: async with get_db_context() as db: result = await db.execute( select(IncidentRecord).where( - IncidentRecord.status.in_([ - IncidentStatus.INVESTIGATING, - IncidentStatus.MITIGATING, - ]) + IncidentRecord.status.in_( + [ + IncidentStatus.INVESTIGATING, + IncidentStatus.MITIGATING, + ] + ) ) ) records = result.scalars().all() @@ -446,6 +454,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: # 必須在 embedding indexing 之前，確保 playbook 表有資料 try: from src.services.playbook_seed_service import seed_playbooks_from_rules + schedule_api_background_task(seed_playbooks_from_rules()) logger.info("playbook_seed_scheduled") except Exception as e: @@ -456,6 +465,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 + schedule_api_background_task(get_playbook_repository().backfill_redis_to_pg()) logger.info("playbook_pg_backfill_scheduled") except Exception as e: @@ -465,6 +475,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: from src.services.playbook_embedding_service import ( ensure_playbook_embeddings_indexed, ) + schedule_api_background_task(ensure_playbook_embeddings_indexed()) logger.info("playbook_embedding_indexing_scheduled") except Exception as e: @@ -486,6 +497,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: # dedup TTL 10 分鐘過期後，ready decisions 就沒有補送機制 → 長期卡在 ready 無人審核 try: from src.services.decision_manager import get_decision_manager + schedule_api_background_task(get_decision_manager().resend_stale_ready_tokens()) logger.info("stale_ready_tokens_resend_scheduled") except Exception as e: @@ -497,6 +509,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 + schedule_api_background_task(run_incident_analysis_sweeper()) logger.info("incident_analysis_sweeper_scheduled", interval_sec=90) except Exception as e: @@ -507,6 +520,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 + schedule_api_background_task(run_asset_scanner_loop()) logger.info("asset_scanner_loop_scheduled", interval_sec=3600) except Exception as e: @@ -517,6 +531,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 + schedule_api_background_task(run_rule_catalog_sync_loop()) logger.info("rule_catalog_sync_loop_scheduled", interval_sec=3600) except Exception as e: @@ -527,6 +542,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 + schedule_api_background_task(run_capacity_scanner_loop()) logger.info("capacity_scanner_loop_scheduled", daily_trigger_hour_taipei=2) except Exception as e: @@ -537,6 +553,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 + schedule_api_background_task(run_compliance_scanner_loop()) logger.info("compliance_scanner_loop_scheduled", daily_trigger_hour_taipei=3) except Exception as e: @@ -546,6 +563,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: # 消費 signals:aider:events stream → 建 incident + 寫 aider_events 表 try: from src.workers.aider_event_processor import run_aider_event_processor_loop + logger.info("aider_event_processor: starting Redis stream consumer") schedule_api_background_task(run_aider_event_processor_loop()) except Exception as e: @@ -556,6 +574,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 + schedule_api_background_task(run_coverage_evaluator_loop()) logger.info("coverage_evaluator_loop_scheduled", interval_sec=3600) except Exception as e: @@ -566,6 +585,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 + schedule_api_background_task(run_rule_stats_updater_loop()) logger.info("rule_stats_updater_loop_scheduled", interval_sec=3600) except Exception as e: @@ -576,35 +596,26 @@ 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 + 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)) - # AIA-P0-006: reconcile the declared all-asset catalog with live ADR-090 - # inventory and project added/removed/drift capability gaps as durable work items. - try: - from src.jobs.asset_capability_reconciliation_job import ( - run_asset_capability_reconciliation_loop, - ) - - schedule_api_background_task(run_asset_capability_reconciliation_loop()) - logger.info( - "asset_capability_reconciliation_loop_scheduled", - daily_trigger_hour_taipei=3, - daily_trigger_minute_taipei=30, - ) - except Exception as e: - logger.warning( - "asset_capability_reconciliation_loop_schedule_failed", - error=str(e), - ) + # AIA-P0-006 has one runtime owner. Production API pods reserve their DB + # budget for requests, so the singleton signal worker owns reconciliation. + logger.info( + "asset_capability_reconciliation_owner_delegated", + owner="signal_worker", + reason="api_db_pool_reserved_for_serving_requests", + ) # ADR-090 § Hermes Rule Quality Advisor (2026-04-19 ogt + Claude Opus 4.7 Asia/Taipei) # 每日 04:00 Taipei 分析 alert_rule_catalog.noise_rate,對高噪音規則推 Telegram 建議 # 統帥鐵律: AI 只推建議不自動改 review_status,人工決策 deprecate try: from src.jobs.hermes_rule_quality_job import 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: @@ -615,6 +626,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 + schedule_api_background_task(run_capacity_forecaster_loop()) logger.info("capacity_forecaster_loop_scheduled", daily_trigger_hour_taipei=5) except Exception as e: @@ -628,6 +640,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: run_monthly_report_loop, run_weekly_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()) @@ -644,6 +657,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 + schedule_api_background_task(run_approval_timeout_resolver()) logger.info("approval_timeout_resolver_scheduled", interval_sec=3600) except Exception as e: @@ -657,6 +671,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: from src.jobs.incident_lifecycle_reconciler import ( INTERVAL_SECONDS as INCIDENT_LIFECYCLE_RECONCILER_INTERVAL, ) + logger.info( "incident_lifecycle_reconciler_owner_delegated", interval_sec=INCIDENT_LIFECYCLE_RECONCILER_INTERVAL, @@ -676,6 +691,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: from src.jobs.awooop_ansible_candidate_backfill_job import ( run_awooop_ansible_candidate_backfill_loop, ) + scheduled_task = schedule_api_background_task( run_awooop_ansible_candidate_backfill_loop() ) @@ -687,7 +703,9 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: interval_seconds=settings.AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_INTERVAL_SECONDS, ) except Exception as e: - logger.warning("awooop_ansible_candidate_backfill_worker_schedule_failed", error=str(e)) + logger.warning( + "awooop_ansible_candidate_backfill_worker_schedule_failed", error=str(e) + ) # AwoooP Ansible check-mode worker. # 先執行 ansible-playbook --check --diff 並回寫 automation_operation_log； @@ -696,6 +714,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: from src.jobs.awooop_ansible_check_mode_job import ( run_awooop_ansible_check_mode_loop, ) + scheduled_task = schedule_api_background_task( run_awooop_ansible_check_mode_loop() ) @@ -713,6 +732,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 + schedule_api_background_task(run_evolver_loop()) logger.info("evolver_loop_scheduled", interval_sec=86400) except Exception as e: @@ -723,18 +743,22 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: from src.jobs.playbook_generation_governance_job import ( 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, ) except Exception as e: - logger.warning("playbook_generation_governance_loop_schedule_failed", error=str(e)) + logger.warning( + "playbook_generation_governance_loop_schedule_failed", error=str(e) + ) # ADR-083 Phase 3: 知識遺忘 Job（每日）— 30d 未引用 KB entry 標記 archived # 2026-04-15 ogt + Claude Sonnet 4.6（亞太）: Phase 3 初始建立 try: from src.jobs.knowledge_decay_job import 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: @@ -745,6 +769,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 + schedule_api_background_task(run_km_backfill_reconciler_loop()) logger.info("km_backfill_reconciler_loop_scheduled", interval_sec=300) except Exception as e: @@ -756,6 +781,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 + schedule_api_background_task(run_aol_writeback_loop()) logger.info("aol_to_catalog_writeback_loop_scheduled", interval_sec=3600) except Exception as e: @@ -771,28 +797,48 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: import asyncio as _asyncio from src.jobs.kb_rot_cleaner import get_kb_rot_cleaner + while True: try: now = now_taipei() # 計算下次月初 3 點（台北時間） if now.day == 1 and now.hour < 3: - next_run = now.replace(hour=3, minute=0, second=0, microsecond=0) + next_run = now.replace( + hour=3, minute=0, second=0, microsecond=0 + ) elif now.month == 12: next_run = now.replace( - year=now.year + 1, month=1, day=1, - hour=3, minute=0, second=0, microsecond=0, + year=now.year + 1, + month=1, + day=1, + hour=3, + minute=0, + second=0, + microsecond=0, ) else: next_run = now.replace( - month=now.month + 1, day=1, - hour=3, minute=0, second=0, microsecond=0, + month=now.month + 1, + day=1, + hour=3, + minute=0, + second=0, + microsecond=0, ) sleep_sec = (next_run - now).total_seconds() - logger.info("kb_rot_cleaner_next_run", next_run=next_run.isoformat(), sleep_sec=int(sleep_sec)) + logger.info( + "kb_rot_cleaner_next_run", + next_run=next_run.isoformat(), + sleep_sec=int(sleep_sec), + ) await _asyncio.sleep(sleep_sec) try: result = await get_kb_rot_cleaner().run() - logger.info("kb_rot_cleaner_completed", stale_count=result.stale_count, total=result.total_scanned) + logger.info( + "kb_rot_cleaner_completed", + stale_count=result.stale_count, + total=result.total_scanned, + ) except Exception as _e: logger.exception("kb_rot_cleaner_failed", error=str(_e)) except _asyncio.CancelledError: @@ -810,6 +856,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 + schedule_api_background_task(run_finetune_export_loop()) logger.info("finetune_export_loop_scheduled", interval_sec=604800) except Exception as e: @@ -821,6 +868,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 + schedule_api_background_task(run_proactive_inspector_loop()) logger.info("proactive_inspector_loop_scheduled", interval_sec=300) except Exception as e: @@ -830,6 +878,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 + schedule_api_background_task(run_offline_replay_loop()) logger.info("offline_replay_loop_scheduled", interval_sec=604800) except Exception as e: @@ -839,6 +888,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 + schedule_api_background_task(run_ai_slo_watchdog_loop()) logger.info("ai_slo_watchdog_scheduled", interval_sec=900) except Exception as e: @@ -848,6 +898,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 + schedule_api_background_task(run_governance_loop()) logger.info("governance_agent_scheduled", interval_sec=3600) except Exception as e: @@ -856,6 +907,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 + schedule_api_background_task(run_governance_dispatcher_loop()) logger.info("governance_dispatcher_scheduled", interval_sec=30) except Exception as e: @@ -866,6 +918,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 + schedule_api_background_task(run_hermes_kb_growth_loop()) logger.info("hermes_kb_growth_worker_scheduled", interval_sec=300) except Exception as e: @@ -893,6 +946,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: try: from src.core.redis_client import get_redis from src.services.failover_alerter import configure_alerter + configure_alerter(get_redis()) logger.info("failover_alerter_configured") except Exception as _alerter_err: @@ -909,8 +963,10 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: # 探測 5 Provider（ollama/ollama_local/gemini/claude/openclaw_nemo）版本 # 寫入 ai_provider_version_history；版本變更時 log warning，P3.2.3 alerter 後續整合 try: + async def _run_model_version_tracker_loop() -> None: from src.services.model_version_tracker import get_model_version_tracker + tracker = get_model_version_tracker() while True: try: @@ -924,7 +980,9 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: except asyncio.CancelledError: break except Exception as _loop_err: - logger.exception("model_version_tracker_loop_error", error=str(_loop_err)) + logger.exception( + "model_version_tracker_loop_error", error=str(_loop_err) + ) await asyncio.sleep(60) # 錯誤後 1 分鐘重試 schedule_api_background_task(_run_model_version_tracker_loop()) @@ -937,6 +995,7 @@ 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 + if api_background_tasks_enabled: await start_platform_worker() logger.info("platform_worker_started", mode="shadow") @@ -955,6 +1014,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: # 必須在 Redis pool 關閉之前停止（recovery service 可能仍在寫 Redis） try: from src.services.ollama_auto_recovery import get_ollama_auto_recovery_service + await get_ollama_auto_recovery_service().stop() logger.info("ollama_failover_system_stopped") except Exception as e: @@ -965,6 +1025,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: # 確保 _verify_and_learn / runbook_generator 寫入不被 SIGTERM cancel try: from src.services.auto_repair_service import get_auto_repair_service + _repair_svc = get_auto_repair_service() if hasattr(_repair_svc, "drain_pending_tasks"): _drain_result = await _repair_svc.drain_pending_tasks(timeout=60.0) @@ -975,6 +1036,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: # AwoooP Phase 4: Platform Worker 優雅停機（2026-05-04 ogt） try: from src.workers.platform_worker import stop_platform_worker + await stop_platform_worker() logger.info("platform_worker_stopped") except Exception as e: @@ -991,6 +1053,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: await telegram_gw.close() # Phase 33: Close RAG Service httpx client (ADR-067) from src.services.knowledge_rag_service import get_knowledge_rag_service + await get_knowledge_rag_service().close() # Phase 5: Close HTTP Clients (統帥鐵律: 連線池回收) await close_all_http_clients() @@ -1178,7 +1241,9 @@ async def http_exception_handler(_request: Request, exc: HTTPException) -> JSONR This is critical for P1-1 fail-closed evidence; without it, all HTTPException is swallowed by the generic exception handler and downgraded to 500. """ - return JSONResponse(status_code=exc.status_code, content={"detail": exc.detail}, headers=exc.headers) + return JSONResponse( + status_code=exc.status_code, content={"detail": exc.detail}, headers=exc.headers + ) @app.exception_handler(Exception) @@ -1219,15 +1284,25 @@ app.include_router(csrf_v1.router, prefix="/api/v1", tags=["Security"]) # Phase app.include_router(dashboard_v1.router, prefix="/api/v1", tags=["Dashboard"]) app.include_router(approvals_v1.router, prefix="/api/v1", tags=["HITL Approvals"]) app.include_router(ai_v1.router, prefix="/api/v1", tags=["AI Decision"]) -app.include_router(ai_governance_v1.router, prefix="/api/v1", tags=["AI Governance"]) # 2026-05-02: /governance 頁面 -app.include_router(ai_slo_v1.router, prefix="/api/v1", tags=["AI SLO"]) # Phase 6 ADR-087 -app.include_router(aiops_kpi_v1.router, prefix="/api/v1", tags=["AIOps KPI"]) # ADR-090 § Phase 7 Dashboard -app.include_router(aiops_timeline_v1.router, prefix="/api/v1", tags=["AIOps Timeline"]) # 2026-04-27 Wave8-X3 B4 +app.include_router( + ai_governance_v1.router, prefix="/api/v1", tags=["AI Governance"] +) # 2026-05-02: /governance 頁面 +app.include_router( + ai_slo_v1.router, prefix="/api/v1", tags=["AI SLO"] +) # Phase 6 ADR-087 +app.include_router( + aiops_kpi_v1.router, prefix="/api/v1", tags=["AIOps KPI"] +) # ADR-090 § Phase 7 Dashboard +app.include_router( + aiops_timeline_v1.router, prefix="/api/v1", tags=["AIOps Timeline"] +) # 2026-04-27 Wave8-X3 B4 app.include_router(webhooks_v1.router, prefix="/api/v1", tags=["Webhooks"]) app.include_router(timeline_v1.router, prefix="/api/v1", tags=["Timeline"]) app.include_router(audit_logs_v1.router, prefix="/api/v1", tags=["Audit Logs"]) # 2026-04-09 Claude Sonnet 4.6: alert_operation_log 查詢 API (Sprint 5.2) -app.include_router(alert_operation_logs_v1.router, prefix="/api/v1", tags=["Alert Operation Logs"]) +app.include_router( + alert_operation_logs_v1.router, prefix="/api/v1", tags=["Alert Operation Logs"] +) app.include_router( aider_events_v1.router, prefix="/api/v1", @@ -1310,7 +1385,9 @@ app.include_router( notifications.router, prefix="/api/v1/notifications", tags=["Notifications"] ) # AwoooP Phase 4 (2026-05-04 ogt): Platform Shell — Shadow Mode Run API -app.include_router(platform_v1.router, prefix="/api/v1/platform", tags=["AwoooP Platform"]) +app.include_router( + platform_v1.router, prefix="/api/v1/platform", tags=["AwoooP Platform"] +) # ============================================================================= diff --git a/apps/api/src/services/ai_automation_asset_capability_matrix.py b/apps/api/src/services/ai_automation_asset_capability_matrix.py index 27549f312..2f35dba01 100644 --- a/apps/api/src/services/ai_automation_asset_capability_matrix.py +++ b/apps/api/src/services/ai_automation_asset_capability_matrix.py @@ -631,6 +631,7 @@ def _receipt_controls( fresh = bool( row and str(row.get("status") or "") == "success" + and output.get("closed") is True and _is_fresh( created_at, now=now, diff --git a/apps/api/src/workers/signal_worker.py b/apps/api/src/workers/signal_worker.py index ef9124ded..55b84a433 100644 --- a/apps/api/src/workers/signal_worker.py +++ b/apps/api/src/workers/signal_worker.py @@ -71,6 +71,7 @@ GRACEFUL_SHUTDOWN_TIMEOUT_S = 75 # K8s terminationGracePeriodSeconds: 90 # Signal Worker # ============================================================================= + class SignalWorker: """ Redis Streams 訊號消費者 @@ -366,7 +367,9 @@ class SignalWorker: span.set_attribute("messaging.destination", STREAM_KEY) span.set_attribute("messaging.message_id", message_id) span.set_attribute("signal.source", data.get("source", "unknown")) - span.set_attribute("signal.alert_name", data.get("alert_name", "unknown")) + span.set_attribute( + "signal.alert_name", data.get("alert_name", "unknown") + ) span.set_attribute("signal.severity", data.get("severity", "unknown")) logger.info( @@ -400,7 +403,9 @@ class SignalWorker: severity=incident.severity.value, signal_count=len(incident.signals), affected_services=incident.affected_services, - persisted_to_pg=getattr(incident, "persisted_to_pg", False), # 2026-04-01 ogt: BrainIncident 無此欄位 (ADR-046 P2-01) + persisted_to_pg=getattr( + incident, "persisted_to_pg", False + ), # 2026-04-01 ogt: BrainIncident 無此欄位 (ADR-046 P2-01) ) try: from src.services.signal_observation_service import ( @@ -505,6 +510,7 @@ def get_signal_worker() -> SignalWorker: # Standalone Entry Point (for K8s Worker Deployment) # ============================================================================= + async def _write_health_files() -> None: """ Write health check files for K8s probes. @@ -542,10 +548,7 @@ async def _heartbeat_loop(shutdown_event: asyncio.Event) -> None: # 等待下次心跳或收到關閉信號 try: - await asyncio.wait_for( - shutdown_event.wait(), - timeout=HEARTBEAT_INTERVAL - ) + await asyncio.wait_for(shutdown_event.wait(), timeout=HEARTBEAT_INTERVAL) break # 收到關閉信號 except TimeoutError: continue # 超時，繼續下次心跳 @@ -575,7 +578,9 @@ async def _main() -> None: "signal_worker_standalone_starting", environment=settings.ENVIRONMENT, redis_url=settings.REDIS_URL.split("@")[-1] if settings.REDIS_URL else "N/A", - database_url=settings.DATABASE_URL.split("@")[-1] if settings.DATABASE_URL else "N/A", + database_url=settings.DATABASE_URL.split("@")[-1] + if settings.DATABASE_URL + else "N/A", ) # Initialize Redis (API pool + Worker 專屬長連線池) @@ -695,6 +700,9 @@ async def _main() -> None: telegram_dispatch_performed=False, ) if settings.ENABLE_SECURITY_CONTROL_PLANE_MAINTENANCE_WORKER: + from src.jobs.asset_capability_reconciliation_job import ( + run_asset_capability_reconciliation_loop, + ) from src.jobs.asset_change_tracker_job import run_asset_change_tracker_loop from src.jobs.asset_scanner_job import run_asset_scanner_loop from src.jobs.compliance_scanner_job import run_compliance_scanner_loop @@ -707,6 +715,10 @@ async def _main() -> None: ("run_compliance_scanner_loop", run_compliance_scanner_loop), ("run_coverage_evaluator_loop", run_coverage_evaluator_loop), ("run_asset_change_tracker_loop", run_asset_change_tracker_loop), + ( + "run_asset_capability_reconciliation_loop", + run_asset_capability_reconciliation_loop, + ), ) security_maintenance_tasks = [ asyncio.create_task(loop(), name=task_name) @@ -791,12 +803,8 @@ async def _main() -> None: "signal_worker_agent99_controlled_dispatch_reconciler_skipped_fail_closed", owner="signal_worker", reason="authenticated_relay_not_configured", - relay_url_configured=bool( - settings.AGENT99_SRE_ALERT_RELAY_URL.strip() - ), - relay_token_configured=bool( - settings.AGENT99_SRE_ALERT_RELAY_TOKEN.strip() - ), + relay_url_configured=bool(settings.AGENT99_SRE_ALERT_RELAY_URL.strip()), + relay_token_configured=bool(settings.AGENT99_SRE_ALERT_RELAY_TOKEN.strip()), ) # Setup graceful shutdown shutdown_event = asyncio.Event() @@ -873,6 +881,7 @@ async def _main() -> None: # Remove health files from pathlib import Path + Path("/tmp/worker-healthy").unlink(missing_ok=True) Path("/tmp/worker-ready").unlink(missing_ok=True) diff --git a/apps/api/tests/test_ai_automation_asset_capability_matrix.py b/apps/api/tests/test_ai_automation_asset_capability_matrix.py index e36e7d95f..ad2680394 100644 --- a/apps/api/tests/test_ai_automation_asset_capability_matrix.py +++ b/apps/api/tests/test_ai_automation_asset_capability_matrix.py @@ -1,12 +1,15 @@ from __future__ import annotations import asyncio +import inspect from datetime import UTC, datetime, timedelta from pathlib import Path from typing import Any +import src.jobs.asset_capability_reconciliation_job as reconciliation_job import src.services.ai_automation_asset_capability_matrix as matrix_service from src.jobs.asset_capability_reconciliation_job import ( + _persist_reconciliation, build_reconciliation_summary_payload, build_reconciliation_work_item_input, ) @@ -212,6 +215,7 @@ def test_live_asset_has_per_stage_covered_status_and_closes_after_receipt() -> N "candidate_count": 0, "repository_readback_count": 0, "repository_readback_verified": True, + "closed": True, }, } @@ -232,6 +236,38 @@ def test_live_asset_has_per_stage_covered_status_and_closes_after_receipt() -> N assert matrix["summary"]["gap_asset_count"] == 0 +def test_reconciliation_receipt_requires_closed_terminal() -> None: + now = datetime(2026, 7, 11, 12, 0, tzinfo=UTC) + initial = build_asset_capability_matrix( + _single_host_scope(), + live_assets=[_fully_covered_host(now)], + latest_run={"run_id": "run-1", "status": "success", "ended_at": now}, + repo_root=_REPO_ROOT, + generated_at=now, + ) + matrix = build_asset_capability_matrix( + _single_host_scope(), + live_assets=[_fully_covered_host(now)], + latest_run={"run_id": "run-1", "status": "success", "ended_at": now}, + reconciliation_receipt={ + "status": "success", + "created_at": now, + "output": { + "matrix_fingerprint": initial["matrix_fingerprint"], + "candidate_count": 0, + "repository_readback_count": 0, + "repository_readback_verified": True, + "closed": False, + }, + }, + repo_root=_REPO_ROOT, + generated_at=now, + ) + + assert matrix["controls"]["daily_reconciliation_receipt"] is False + assert matrix["closed"] is False + + def test_outdated_live_receipts_are_classified_stale_per_stage() -> None: now = datetime(2026, 7, 11, 12, 0, tzinfo=UTC) old = now - timedelta(hours=40) @@ -342,6 +378,51 @@ def test_reconciliation_work_item_uses_same_trace_and_run_as_summary() -> None: ) +def test_reconciliation_persistence_batches_candidate_queries_and_inserts() -> None: + source = inspect.getsource(_persist_reconciliation) + loop_start = source.index("for candidate_raw in candidates:") + loop_end = source.index("if insert_rows:") + + assert "ANY(CAST(:candidate_fingerprints AS text[]))" in source + assert "insert_rows" in source + assert "await db.execute" not in source[loop_start:loop_end] + assert "AND output ->> 'closed' = 'true'" in source + + +def test_reconciliation_loop_retries_degraded_result_before_daily_wait( + monkeypatch: Any, +) -> None: + triggered_by_values: list[str] = [] + sleep_values: list[int] = [] + + async def _fake_reconcile_once(*, triggered_by: str) -> dict[str, object]: + triggered_by_values.append(triggered_by) + if len(triggered_by_values) == 1: + return {"status": "degraded_no_write", "reason": "transient_db_pressure"} + return {"status": "success", "closed": True} + + async def _fake_sleep(seconds: int) -> None: + sleep_values.append(seconds) + if len(sleep_values) == 3: + raise asyncio.CancelledError + + monkeypatch.setattr(reconciliation_job, "reconcile_once", _fake_reconcile_once) + monkeypatch.setattr(reconciliation_job.asyncio, "sleep", _fake_sleep) + monkeypatch.setattr(reconciliation_job, "_FIRST_DELAY_SECONDS", 90) + monkeypatch.setattr(reconciliation_job, "_LOOP_BACKOFF_SECONDS", 1_800) + monkeypatch.setattr(reconciliation_job, "_seconds_until_next_trigger", lambda: 1) + + try: + asyncio.run(reconciliation_job.run_asset_capability_reconciliation_loop()) + except asyncio.CancelledError: + pass + else: + raise AssertionError("loop should stop at the test cancellation point") + + assert triggered_by_values == ["startup", "retry"] + assert sleep_values == [90, 1_800, 1] + + def test_priority_overlay_projects_aia_p0_006_runtime_controls() -> None: matrix = build_asset_capability_matrix( _single_host_scope(), diff --git a/apps/api/tests/test_signal_worker_ansible_executor_binding.py b/apps/api/tests/test_signal_worker_ansible_executor_binding.py index 21245814e..1d5802b05 100644 --- a/apps/api/tests/test_signal_worker_ansible_executor_binding.py +++ b/apps/api/tests/test_signal_worker_ansible_executor_binding.py @@ -21,6 +21,7 @@ def test_signal_worker_produces_candidates_and_owns_security_maintenance() -> No assert "run_compliance_scanner_loop" in source assert "run_coverage_evaluator_loop" in source assert "run_asset_change_tracker_loop" in source + assert "run_asset_capability_reconciliation_loop" in source assert "signal_worker_security_control_plane_maintenance_started" in source assert "task.cancel()" in source assert "await asyncio.gather(*security_maintenance_tasks" in source @@ -33,6 +34,25 @@ def test_signal_worker_produces_candidates_and_owns_security_maintenance() -> No assert "signal_worker_ansible_check_mode_loop_started" not in source +def test_signal_worker_is_single_asset_capability_reconciliation_owner() -> None: + from src import main as api_main + + api_source = inspect.getsource(api_main.lifespan) + worker_source = inspect.getsource(signal_worker._main) + + assert "run_asset_capability_reconciliation_loop" not in api_source + assert "asset_capability_reconciliation_owner_delegated" in api_source + assert "run_asset_capability_reconciliation_loop" in worker_source + maintenance_start = worker_source.index("security_maintenance_loops =") + maintenance_end = worker_source.index( + "security_maintenance_tasks =", maintenance_start + ) + assert ( + "run_asset_capability_reconciliation_loop" + in worker_source[maintenance_start:maintenance_end] + ) + + def test_dedicated_broker_owns_ansible_executor_loop() -> None: source = inspect.getsource(ansible_executor_broker._main) @@ -198,7 +218,10 @@ def test_cd_prunes_and_verifies_api_executor_boundary_in_production() -> None: assert "legacy cluster-wide executor binding still exists" in workflow assert "awoooi-executor-boundary-verification" in workflow assert "awoooi_executor_boundary_verification_v4" in workflow - assert "single_writer_source_verifier=cd_tests_and_production_source_sha_v1" in workflow + assert ( + "single_writer_source_verifier=cd_tests_and_production_source_sha_v1" + in workflow + ) assert "decision_manager_queue_only=true" in workflow assert "webhook_router_queue_only=true" in workflow assert "failure_watcher_queue_only=true" in workflow @@ -206,24 +229,27 @@ def test_cd_prunes_and_verifies_api_executor_boundary_in_production() -> None: assert "candidate_idempotency_contract_verified=true" in workflow assert "apply_idempotency_contract_verified=true" in workflow assert "critical_break_glass_contract_verified=true" in workflow - assert "independent_post_verifier_source_verifier=cd_tests_and_production_source_sha_v1" in workflow + assert ( + "independent_post_verifier_source_verifier=cd_tests_and_production_source_sha_v1" + in workflow + ) assert "asset_postcondition_registry_complete=true" in workflow assert "independent_post_verifier_read_only=true" in workflow assert "executor_returncode_not_success_source=true" in workflow assert "verifier_failure_marks_apply_failed=true" in workflow assert "verifier_failure_retry_first=true" in workflow assert "post_verifier_raw_output_not_persisted=true" in workflow - assert "canonical_learning_source_verifier=cd_tests_and_production_source_sha_v1" in workflow + assert ( + "canonical_learning_source_verifier=cd_tests_and_production_source_sha_v1" + in workflow + ) assert "canonical_playbook_resolver_complete=true" in workflow assert "km_row_readback_required=true" in workflow assert "playbook_row_readback_required=true" in workflow assert "learning_failure_fail_closed=true" in workflow assert "trust_failure_fail_closed=true" in workflow assert "learning_receipts_public_safe=true" in workflow - assert ( - "k8s/awoooi-prod/08-deployment-ansible-executor-broker.yaml" - in workflow - ) + assert "k8s/awoooi-prod/08-deployment-ansible-executor-broker.yaml" in workflow assert "executor_boundary_public_readback=verified_ready" in workflow assert "EXECUTOR_BOUNDARY_READBACK_ATTEMPTS" in workflow assert "range(1, attempts + 1)" in workflow @@ -234,9 +260,7 @@ def test_cd_prunes_and_verifies_api_executor_boundary_in_production() -> None: assert "autonomous_single_writer_source_boundary" in workflow assert "independent_post_verifier_source_boundary" in workflow assert "canonical_learning_source_boundary" in workflow - worker_env = worker_deployment["spec"]["template"]["spec"]["containers"][0][ - "env" - ] + worker_env = worker_deployment["spec"]["template"]["spec"]["containers"][0]["env"] assert any(item.get("name") == "AWOOOI_BUILD_COMMIT_SHA" for item in worker_env) assert "k8s/awoooi-prod/08-deployment-worker.yaml" in workflow assert '"source_boundary_verified"' in workflow @@ -254,9 +278,18 @@ def test_cd_prunes_and_verifies_api_executor_boundary_in_production() -> None: assert "current_user AS role_name" in workflow assert "hashlib.sha256" in workflow assert "representative_db_preflights_verified" in workflow - assert 'api_required_connection_budget="$API_DB_REQUIRED_CONNECTION_BUDGET"' in workflow - assert 'worker_required_connection_budget="$WORKER_DB_REQUIRED_CONNECTION_BUDGET"' in workflow - assert 'broker_required_connection_budget="$BROKER_DB_REQUIRED_CONNECTION_BUDGET"' in workflow + assert ( + 'api_required_connection_budget="$API_DB_REQUIRED_CONNECTION_BUDGET"' + in workflow + ) + assert ( + 'worker_required_connection_budget="$WORKER_DB_REQUIRED_CONNECTION_BUDGET"' + in workflow + ) + assert ( + 'broker_required_connection_budget="$BROKER_DB_REQUIRED_CONNECTION_BUDGET"' + in workflow + ) assert "api_representative_connection_probe_count" in workflow assert "worker_representative_connection_probe_count" in workflow assert "broker_representative_connection_probe_count" in workflow @@ -330,7 +363,5 @@ def test_api_log_does_not_claim_skipped_executor_was_scheduled() -> None: assert "awooop_ansible_check_mode_worker_schedule_evaluated" in main_source assert "scheduled_in_api_process=scheduled_task is not None" in main_source assert 'production_process_owner="awoooi-worker"' in main_source - assert ( - 'production_process_owner="awoooi-ansible-executor-broker"' in main_source - ) + assert 'production_process_owner="awoooi-ansible-executor-broker"' in main_source assert "awooop_ansible_candidate_backfill_worker_schedule_evaluated" in main_source