fix(security): reconcile asset control plane truth

apps/api/src/jobs/asset_scanner_job.py

@@ -92,7 +92,7 @@ async def scan_once(
     triggered_by: str = "cron",
     scope: tuple[str, ...] = ("k8s", "prometheus", "database_catalog"),
     scan_depth: str = "shallow",
-) -> str | None:
+) -> str:
     """
     執行一次資產盤點。
 
@@ -102,15 +102,14 @@ async def scan_once(
         scan_depth:   shallow (僅 list) / deep (含 describe) / full
 
     Returns:
-        run_id (UUID 字串) 或 None (寫 header 失敗時)
+        run_id (UUID 字串)。無法建立 durable header 時會拋出例外。
     """
     started_ms = _time.time()
     run_id = await _start_discovery_run(triggered_by, list(scope), scan_depth)
-    if not run_id:
-        return None
 
     new_count = 0
     modified_count = 0
+    disappeared_count = 0
     total_count = 0
     error_msg: str | None = None
 
@@ -119,11 +118,29 @@ async def scan_once(
         # 資源類型: pods (container), deployments/statefulsets/daemonsets (k8s_workload),
         #          services (k8s_resource), nodes (host), configmaps (k8s_resource)
         # 跳過: secrets (awoooi-executor RBAC 不允許 list)
-        runtime_assets, relationships = await _collect_runtime_assets()
+        (
+            runtime_assets,
+            relationships,
+            reconciled_prefixes,
+            collector_errors,
+        ) = await _collect_runtime_assets()
         total_count = len(runtime_assets)
 
         # UPSERT inventory
         new_count, modified_count = await _upsert_assets(runtime_assets, run_id)
+        if new_count + modified_count != total_count:
+            raise RuntimeError(
+                "asset_upsert_count_mismatch "
+                f"expected={total_count} persisted={new_count + modified_count}"
+            )
+
+        # Only reconcile sources that completed collection in this run. This
+        # prevents a failed/empty collector from deprecating healthy assets.
+        disappeared_count = await _deprecate_missing_assets(
+            runtime_assets,
+            reconciled_prefixes,
+            run_id,
+        )
 
         # 建立 asset_relationship (OwnerReference + Service selector + Pod volumes)
         rel_written = await _upsert_relationships(relationships)
@@ -132,6 +149,10 @@ async def scan_once(
         await _write_coverage_snapshots(run_id)
 
         logger.info("asset_scan_relationships_written", run_id=run_id, relationships=rel_written)
+        if collector_errors:
+            raise RuntimeError(
+                "asset_collector_failures=" + ",".join(collector_errors)
+            )
 
     except Exception as e:
         error_msg = f"{type(e).__name__}: {e}"[:1000]
@@ -140,26 +161,18 @@ async def scan_once(
     duration_ms = int((_time.time() - started_ms) * 1000)
     final_status = "failed" if error_msg else "success"
 
-    await _finish_discovery_run(
+    # The discovery terminal and AOL receipt share one DB transaction. A run
+    # cannot be marked successful without its durable audit receipt.
+    await _finalize_discovery_run(
         run_id=run_id,
+        triggered_by=triggered_by,
         status=final_status,
         total_assets=total_count,
         new_assets=new_count,
         modified_assets=modified_count,
+        disappeared_assets=disappeared_count,
         duration_ms=duration_ms,
         error=error_msg,
-    )
-
-    # ADR-090 § aol 留痕 — asset_discovered 是合法 op_type
-    await _log_aol_asset_discovered(
-        run_id=run_id,
-        triggered_by=triggered_by,
-        total=total_count,
-        new=new_count,
-        modified=modified_count,
-        duration_ms=duration_ms,
-        status=final_status,
-        error=error_msg,
         scope=list(scope),
         scan_depth=scan_depth,
     )
@@ -171,6 +184,7 @@ async def scan_once(
         total=total_count,
         new=new_count,
         modified=modified_count,
+        disappeared=disappeared_count,
         duration_ms=duration_ms,
     )
     return run_id
@@ -345,7 +359,12 @@ def _build_configmap_asset(item: dict[str, Any]) -> dict[str, Any]:
     }
 
 
-async def _collect_runtime_assets() -> tuple[list[dict[str, Any]], list[dict[str, str]]]:
+async def _collect_runtime_assets() -> tuple[
+    list[dict[str, Any]],
+    list[dict[str, str]],
+    tuple[str, ...],
+    tuple[str, ...],
+]:
     """
     掃多種 K8s 資源類型 + 提取 relationship.
 
@@ -360,10 +379,16 @@ async def _collect_runtime_assets() -> tuple[list[dict[str, Any]], list[dict[str
       原 code 跳過 ReplicaSet → Pod→Deployment 全部漏掉.
       修: 先掃 ReplicaSet 建 rs_to_deployment map,Pod 用 rs_name 反查 Deployment.
 
-    回傳 (assets, relationships) tuple.
+    回傳 (assets, relationships, reconciled_asset_key_prefixes, collector_errors) tuple.
+
+    Only prefixes whose source query completed successfully are returned. The
+    caller may safely deprecate missing assets within those prefixes without
+    treating collector failures as asset deletion.
     """
     assets: list[dict[str, Any]] = []
     relationships: list[dict[str, str]] = []
+    reconciled_prefixes: set[str] = set()
+    collector_errors: list[str] = []
 
     # 0. ReplicaSets — 僅作為 Pod→Deployment 橋樑,不寫入 asset_inventory
     rs_to_deployment: dict[str, str] = {}  # "ns/rs_name" -> "deployment_name"
@@ -378,14 +403,17 @@ async def _collect_runtime_assets() -> tuple[list[dict[str, Any]], list[dict[str
                     rs_to_deployment[f"{rs_ns}/{rs_name}"] = ref.get("name", "")
     except Exception as e:
         logger.warning("collect_replicasets_failed", error=str(e))
+        collector_errors.append("k8s_replicasets")
 
     # 1. Nodes (不帶 ns)
     try:
         payload = await _fetch_kubectl_json("nodes", all_namespaces=False)
         for item in payload.get("items", []) or []:
             assets.append(_build_node_asset(item))
+        reconciled_prefixes.add("k8s/node/")
     except Exception as e:
         logger.warning("collect_nodes_failed", error=str(e))
+        collector_errors.append("k8s_nodes")
 
     # 2. Pods — 主體 + 從 ownerReferences 建 relationship
     pod_by_key: dict[str, dict[str, Any]] = {}
@@ -436,8 +464,10 @@ async def _collect_runtime_assets() -> tuple[list[dict[str, Any]], list[dict[str
                         "to_key": f"k8s/configmap/{ns}/{cm}",
                         "relationship_type": "depends_on",
                     })
+        reconciled_prefixes.add("k8s/pod/")
     except Exception as e:
         logger.warning("collect_pods_failed", error=str(e))
+        collector_errors.append("k8s_pods")
 
     # 3. Deployments / StatefulSets / DaemonSets
     for kind, resource in (("Deployment", "deployments"), ("StatefulSet", "statefulsets"), ("DaemonSet", "daemonsets")):
@@ -445,8 +475,10 @@ async def _collect_runtime_assets() -> tuple[list[dict[str, Any]], list[dict[str
             payload = await _fetch_kubectl_json(resource)
             for item in payload.get("items", []) or []:
                 assets.append(_build_workload_asset(item, kind))
+            reconciled_prefixes.add(f"k8s/{kind.lower()}/")
         except Exception as e:
             logger.warning(f"collect_{resource}_failed", error=str(e))
+            collector_errors.append(f"k8s_{resource}")
 
     # 4. Services + routes_to Pod (via selector match)
     try:
@@ -471,16 +503,20 @@ async def _collect_runtime_assets() -> tuple[list[dict[str, Any]], list[dict[str
                         "to_key": pod_key,
                         "relationship_type": "routes_to",
                     })
+        reconciled_prefixes.add("k8s/service/")
     except Exception as e:
         logger.warning("collect_services_failed", error=str(e))
+        collector_errors.append("k8s_services")
 
     # 5. ConfigMaps
     try:
         payload = await _fetch_kubectl_json("configmaps")
         for item in payload.get("items", []) or []:
             assets.append(_build_configmap_asset(item))
+        reconciled_prefixes.add("k8s/configmap/")
     except Exception as e:
         logger.warning("collect_configmaps_failed", error=str(e))
+        collector_errors.append("k8s_configmaps")
 
     # 6. Prometheus targets — 補齊 host-install services (110/112/188/125 等非 K8s)
     #    Gap 1 修補 (2026-04-19 audit): 原本 asset_inventory 只涵蓋 K8s,
@@ -490,18 +526,30 @@ async def _collect_runtime_assets() -> tuple[list[dict[str, Any]], list[dict[str
         prom_assets, host_relationships = await _collect_prometheus_targets()
         assets.extend(prom_assets)
         relationships.extend(host_relationships)
+        reconciled_prefixes.update(("host/", "prometheus_target/"))
     except Exception as e:
         logger.warning("collect_prometheus_targets_failed", error=str(e))
+        collector_errors.append("prometheus_targets")
 
     # 7. PostgreSQL catalog — database/table/view metadata only, never row data.
     try:
         database_assets, database_relationships = await _collect_database_catalog_assets()
         assets.extend(database_assets)
         relationships.extend(database_relationships)
+        if database_assets:
+            reconciled_prefixes.update(
+                ("postgres/database/", "postgres/relation/")
+            )
     except Exception as e:
         logger.warning("collect_database_catalog_failed", error=str(e))
+        collector_errors.append("database_catalog")
 
-    return assets, relationships
+    return (
+        assets,
+        relationships,
+        tuple(sorted(reconciled_prefixes)),
+        tuple(collector_errors),
+    )
 
 
 def _is_valid_ipv4(s: str) -> bool:
@@ -543,14 +591,10 @@ async def _collect_prometheus_targets() -> tuple[list[dict[str, Any]], list[dict
     seen_hosts: set[str] = set()
 
     url = f"{settings.PROMETHEUS_URL.rstrip('/')}/api/v1/targets"
-    try:
-        async with httpx.AsyncClient(timeout=10.0, trust_env=False) as client:
-            resp = await client.get(url, params={"state": "active"})
-            resp.raise_for_status()
-            data = resp.json()
-    except Exception as e:
-        logger.warning("prometheus_targets_fetch_failed", error=str(e))
-        return assets, relationships
+    async with httpx.AsyncClient(timeout=10.0, trust_env=False) as client:
+        resp = await client.get(url, params={"state": "active"})
+        resp.raise_for_status()
+        data = resp.json()
 
     for t in (data.get("data", {}) or {}).get("activeTargets", []) or []:
         labels = t.get("labels", {}) or {}
@@ -744,11 +788,11 @@ async def _start_discovery_run(
     triggered_by: str,
     scope: list[str],
     scan_depth: str,
-) -> str | None:
+) -> str:
     """
     寫 asset_discovery_run header (status='running'), 回傳 run_id。
 
-    失敗 → log warning + 回 None,主流程靜默跳過本次 scan。
+    失敗會向排程器傳播並進入 bounded backoff，不可靜默跳過整輪掃描。
     """
     try:
         from sqlalchemy import text as _sql
@@ -783,51 +827,97 @@ async def _start_discovery_run(
                 },
             )
             run_id = row.scalar()
-            return str(run_id) if run_id else None
+            if not run_id:
+                raise RuntimeError("asset_discovery_run_missing_run_id")
+            return str(run_id)
     except Exception as e:
         logger.warning("asset_discovery_run_start_failed", error=str(e))
-        return None
+        raise
 
 
-async def _finish_discovery_run(
+async def _finalize_discovery_run(
     run_id: str,
+    triggered_by: str,
     status: str,
     total_assets: int,
     new_assets: int,
     modified_assets: int,
+    disappeared_assets: int,
     duration_ms: int,
     error: str | None,
+    scope: list[str],
+    scan_depth: str,
 ) -> None:
-    try:
-        from sqlalchemy import text as _sql
+    """Atomically persist the discovery terminal and its AOL receipt."""
+    from sqlalchemy import text as _sql
 
-        from src.db.base import get_db_context
+    from src.db.base import get_db_context
 
-        async with get_db_context(_PROJECT_ID) as db:
-            await db.execute(
-                _sql("""
-                    UPDATE asset_discovery_run
-                    SET status = :st,
-                        ended_at = NOW(),
-                        total_assets = :total,
-                        new_assets = :new,
-                        modified_assets = :mod,
-                        duration_ms = :dur,
-                        error = :err
-                    WHERE run_id = CAST(:rid AS uuid)
-                """),
-                {
-                    "st": status,
-                    "total": total_assets,
-                    "new": new_assets,
-                    "mod": modified_assets,
-                    "dur": duration_ms,
-                    "err": error,
-                    "rid": run_id,
-                },
-            )
-    except Exception as e:
-        logger.warning("asset_discovery_run_finish_failed", run_id=run_id, error=str(e))
+    input_payload = {
+        "run_id": run_id,
+        "triggered_by": triggered_by,
+        "scope": scope,
+        "scan_depth": scan_depth,
+    }
+    output_payload = {
+        "run_id": run_id,
+        "total_assets": total_assets,
+        "new_assets": new_assets,
+        "modified_assets": modified_assets,
+        "disappeared_assets": disappeared_assets,
+    }
+
+    async with get_db_context(_PROJECT_ID) as db:
+        await db.execute(
+            _sql("""
+                UPDATE asset_discovery_run
+                SET status = :st,
+                    ended_at = NOW(),
+                    total_assets = :total,
+                    new_assets = :new,
+                    modified_assets = :mod,
+                    disappeared_assets = :gone,
+                    duration_ms = :dur,
+                    error = :err
+                WHERE run_id = CAST(:rid AS uuid)
+            """),
+            {
+                "st": status,
+                "total": total_assets,
+                "new": new_assets,
+                "mod": modified_assets,
+                "gone": disappeared_assets,
+                "dur": duration_ms,
+                "err": error,
+                "rid": run_id,
+            },
+        )
+        await db.execute(
+            _sql("""
+                INSERT INTO automation_operation_log (
+                    operation_type, actor, status,
+                    input, output, run_id, duration_ms, error, tags
+                ) VALUES (
+                    'asset_discovered',
+                    'asset_scanner',
+                    :st,
+                    CAST(:input AS jsonb),
+                    CAST(:output AS jsonb),
+                    CAST(:rid AS uuid),
+                    :dur, :err,
+                    :tags
+                )
+            """),
+            {
+                "st": "success" if status == "success" else "failed",
+                "input": _json.dumps(input_payload, ensure_ascii=False),
+                "output": _json.dumps(output_payload, ensure_ascii=False),
+                "rid": run_id,
+                "dur": duration_ms,
+                "err": (error or "")[:2000] if error else None,
+                "tags": ["asset_scanner", "discovery", *scope],
+            },
+        )
 
 
 async def _upsert_assets(
@@ -891,10 +981,62 @@ async def _upsert_assets(
                     modified_count += 1
     except Exception as e:
         logger.exception("asset_upsert_failed", run_id=run_id, error=str(e))
+        raise
 
     return new_count, modified_count
 
 
+async def _deprecate_missing_assets(
+    assets: list[dict[str, Any]],
+    reconciled_prefixes: tuple[str, ...],
+    run_id: str,
+) -> int:
+    """Deprecate assets missing from collectors that succeeded in this run."""
+    prefixes = sorted({prefix for prefix in reconciled_prefixes if prefix})
+    if not prefixes:
+        return 0
+
+    seen_keys = sorted(
+        {
+            str(asset["asset_key"])
+            for asset in assets
+            if any(str(asset["asset_key"]).startswith(prefix) for prefix in prefixes)
+        }
+    )
+
+    from sqlalchemy import text as _sql
+
+    from src.db.base import get_db_context
+
+    async with get_db_context(_PROJECT_ID) as db:
+        result = await db.execute(
+            _sql(
+                """
+                UPDATE asset_inventory
+                SET lifecycle_state = 'deprecated',
+                    decommissioned_at = NOW(),
+                    updated_at = NOW()
+                WHERE lifecycle_state = 'active'
+                  AND asset_key LIKE ANY(CAST(:patterns AS text[]))
+                  AND last_seen_at < (
+                      SELECT started_at
+                      FROM asset_discovery_run
+                      WHERE run_id = CAST(:rid AS uuid)
+                  )
+                  AND NOT (asset_key = ANY(CAST(:seen_keys AS text[])))
+                """
+            ),
+            {
+                "patterns": [f"{prefix}%" for prefix in prefixes],
+                "seen_keys": seen_keys,
+                "rid": run_id,
+            },
+        )
+
+    rowcount = getattr(result, "rowcount", 0) or 0
+    return max(int(rowcount), 0)
+
+
 async def _upsert_relationships(relationships: list[dict[str, str]]) -> int:
     """
     UPSERT asset_relationship (from_asset_id/to_asset_id/relationship_type 為 UNIQUE).
@@ -939,8 +1081,10 @@ async def _upsert_relationships(relationships: list[dict[str, str]]) -> int:
                 except Exception as e:
                     logger.debug("relationship_upsert_skipped",
                                  from_key=rel["from_key"], to_key=rel["to_key"], error=str(e))
+                    raise
     except Exception as e:
         logger.warning("relationship_upsert_failed", error=str(e))
+        raise
     return written
 
 
@@ -950,99 +1094,33 @@ async def _write_coverage_snapshots(run_id: str) -> None:
 
     後續 service (rule_catalog / playbook_extractor / km_writer) 會 UPDATE 對應維度。
     """
-    try:
-        from sqlalchemy import text as _sql
+    from sqlalchemy import text as _sql
 
-        from src.db.base import get_db_context
+    from src.db.base import get_db_context
 
-        async with get_db_context(_PROJECT_ID) as db:
-            # 一次性 INSERT: 取所有 active asset × 7 dimensions
-            await db.execute(
-                _sql("""
-                    INSERT INTO asset_coverage_snapshot (
-                        run_id, asset_id, dimension, coverage_status,
-                        evidence, detected_by
-                    )
-                    SELECT
-                        CAST(:rid AS uuid),
-                        ai.asset_id,
-                        d.dimension,
-                        'unknown' AS coverage_status,
-                        '{}'::jsonb,
-                        'asset_scanner' AS detected_by
-                    FROM asset_inventory ai
-                    CROSS JOIN (
-                        VALUES ('auto_monitoring'),('auto_alerting'),('auto_rule_creation'),
-                               ('auto_rule_matching'),('auto_playbook'),('auto_remediation'),
-                               ('auto_km_creation')
-                    ) AS d(dimension)
-                    WHERE ai.lifecycle_state = 'active'
-                    ON CONFLICT (run_id, asset_id, dimension) DO NOTHING
-                """),
-                {"rid": run_id},
-            )
-    except Exception as e:
-        logger.warning("asset_coverage_write_failed", run_id=run_id, error=str(e))
-
-
-async def _log_aol_asset_discovered(
-    run_id: str,
-    triggered_by: str,
-    total: int,
-    new: int,
-    modified: int,
-    duration_ms: int,
-    status: str,
-    error: str | None,
-    scope: list[str],
-    scan_depth: str,
-) -> None:
-    """寫 automation_operation_log(asset_discovered)。失敗只 log 不阻塞。"""
-    try:
-        from sqlalchemy import text as _sql
-
-        from src.db.base import get_db_context
-
-        aol_status = "success" if status == "success" else "failed"
-        input_payload = {
-            "run_id": run_id,
-            "triggered_by": triggered_by,
-            "scope": scope,
-            "scan_depth": scan_depth,
-        }
-        output_payload = {
-            "run_id": run_id,
-            "total_assets": total,
-            "new_assets": new,
-            "modified_assets": modified,
-        }
-
-        async with get_db_context(_PROJECT_ID) as db:
-            await db.execute(
-                _sql("""
-                    INSERT INTO automation_operation_log (
-                        operation_type, actor, status,
-                        input, output, run_id, duration_ms, error, tags
-                    ) VALUES (
-                        'asset_discovered',
-                        'asset_scanner',
-                        :st,
-                        CAST(:input AS jsonb),
-                        CAST(:output AS jsonb),
-                        CAST(:rid AS uuid),
-                        :dur, :err,
-                        :tags
-                    )
-                """),
-                {
-                    "st": aol_status,
-                    "input": _json.dumps(input_payload, ensure_ascii=False),
-                    "output": _json.dumps(output_payload, ensure_ascii=False),
-                    "rid": run_id,
-                    "dur": duration_ms,
-                    "err": (error or "")[:2000] if error else None,
-                    "tags": ["asset_scanner", "discovery", *scope],
-                },
-            )
-    except Exception as e:
-        logger.warning("asset_scanner_aol_write_failed", run_id=run_id, error=str(e))
+    async with get_db_context(_PROJECT_ID) as db:
+        # 一次性 INSERT: 取所有 active asset × 7 dimensions
+        await db.execute(
+            _sql("""
+                INSERT INTO asset_coverage_snapshot (
+                    run_id, asset_id, dimension, coverage_status,
+                    evidence, detected_by
+                )
+                SELECT
+                    CAST(:rid AS uuid),
+                    ai.asset_id,
+                    d.dimension,
+                    'unknown' AS coverage_status,
+                    '{}'::jsonb,
+                    'asset_scanner' AS detected_by
+                FROM asset_inventory ai
+                CROSS JOIN (
+                    VALUES ('auto_monitoring'),('auto_alerting'),('auto_rule_creation'),
+                           ('auto_rule_matching'),('auto_playbook'),('auto_remediation'),
+                           ('auto_km_creation')
+                ) AS d(dimension)
+                WHERE ai.lifecycle_state = 'active'
+                ON CONFLICT (run_id, asset_id, dimension) DO NOTHING
+            """),
+            {"rid": run_id},
+        )