diff --git a/apps/api/src/services/platform_operator_service.py b/apps/api/src/services/platform_operator_service.py index d330746ae..b60850102 100644 --- a/apps/api/src/services/platform_operator_service.py +++ b/apps/api/src/services/platform_operator_service.py @@ -38,7 +38,7 @@ from src.db.awooop_models import ( AwoooPRunStepJournal, ) from src.db.base import get_db_context -from src.db.models import ApprovalRecord, IncidentRecord, MCPAuditLog +from src.db.models import AlertOperationLog, ApprovalRecord, IncidentRecord, MCPAuditLog from src.services.ai_agent_result_capture_owner_release_approval_gate import ( load_latest_ai_agent_result_capture_owner_release_approval_gate, ) @@ -97,6 +97,15 @@ _MAX_STEP_SUMMARY_CHARS = 128 _AI_ROUTE_STATUS_SELECT_TIMEOUT_SECONDS = 12.0 _AI_ROUTE_STATUS_CONNECTIVITY_TIMEOUT_SECONDS = 2.5 _REMEDIATION_HISTORY_LIMIT = 20 +_REMEDIATION_HISTORY_EVENT_TYPES = ( + "PRE_FLIGHT_PASSED", + "PRE_FLIGHT_FAILED", + "APPROVAL_ESCALATED", +) +_REMEDIATION_HISTORY_SCHEMA_VERSIONS = { + "adr100_remediation_dry_run_history_v1", + "adr100_remediation_approval_history_v1", +} _ADR100_GATE5_PROJECTION_TRIGGER = "adr100_runtime_replay_gate5" _CALLBACK_REPLY_CACHE_TTL_SECONDS = int( os.getenv("AWOOOP_CALLBACK_REPLY_CACHE_TTL_SECONDS", "20") @@ -1093,6 +1102,7 @@ async def list_runs( runs=candidate_rows, inbound_by_run=inbound_by_run, outbound_by_run=outbound_by_run, + db=db, ) callback_reply_summaries = { row.run_id: _run_callback_reply_summary(outbound_by_run.get(row.run_id, [])) @@ -1129,6 +1139,7 @@ async def list_runs( runs=rows, inbound_by_run=inbound_by_run, outbound_by_run=outbound_by_run, + db=db, ) callback_reply_summaries = { row.run_id: _run_callback_reply_summary(outbound_by_run.get(row.run_id, [])) @@ -7599,11 +7610,122 @@ def _remediation_summary_matches_incident_id( return isinstance(incident_ids, list) and incident_id in incident_ids +async def _fetch_run_remediation_histories_by_incident( + incident_ids: list[str], + *, + limit: int = _REMEDIATION_HISTORY_LIMIT, + db: Any | None = None, +) -> tuple[dict[str, list[dict[str, Any]]], dict[str, dict[str, str]]]: + """Fetch ADR-100 list evidence with one bounded query for all incidents.""" + normalized_incident_ids = list(dict.fromkeys(incident_ids)) + if not normalized_incident_ids: + return {}, {} + + safe_limit = max(1, min(limit, 200)) + per_event_fetch_limit = min(max(safe_limit * 4, 50), 200) + batch_limit = ( + len(normalized_incident_ids) + * len(_REMEDIATION_HISTORY_EVENT_TYPES) + * per_event_fetch_limit + ) + event_rank = func.row_number().over( + partition_by=( + AlertOperationLog.incident_id, + AlertOperationLog.event_type, + ), + order_by=AlertOperationLog.created_at.desc(), + ).label("event_rank") + ranked_rows = ( + select( + AlertOperationLog.id.label("alert_operation_log_id"), + event_rank, + ) + .where( + AlertOperationLog.incident_id.in_(normalized_incident_ids), + AlertOperationLog.event_type.in_(_REMEDIATION_HISTORY_EVENT_TYPES), + ) + .subquery() + ) + stmt = ( + select(AlertOperationLog) + .join( + ranked_rows, + AlertOperationLog.id == ranked_rows.c.alert_operation_log_id, + ) + .where(ranked_rows.c.event_rank <= per_event_fetch_limit) + .order_by(AlertOperationLog.created_at.desc()) + .limit(batch_limit) + ) + + try: + if db is None: + async with get_db_context() as batch_db: + result = await batch_db.execute(stmt) + else: + result = await db.execute(stmt) + rows = list(result.scalars().all()) + except Exception as exc: + error = str(exc) + logger.warning( + "run_list_remediation_history_batch_fetch_failed", + incident_count=len(normalized_incident_ids), + error=error, + ) + return {}, { + incident_id: {"incident_id": incident_id, "error": error} + for incident_id in normalized_incident_ids + } + + rows_by_incident: dict[str, list[AlertOperationLog]] = defaultdict(list) + for row in rows: + incident_id = str(getattr(row, "incident_id", "") or "") + if incident_id in normalized_incident_ids: + rows_by_incident[incident_id].append(row) + + from src.services.adr100_remediation_service import _history_item + + histories_by_incident: dict[str, list[dict[str, Any]]] = {} + errors_by_incident: dict[str, dict[str, str]] = {} + for incident_id in normalized_incident_ids: + try: + items: list[dict[str, Any]] = [] + incident_rows = sorted( + rows_by_incident.get(incident_id, []), + key=lambda row: str(getattr(row, "created_at", "") or ""), + reverse=True, + ) + for row in incident_rows: + context = getattr(row, "context", None) or {} + if ( + context.get("schema_version") + not in _REMEDIATION_HISTORY_SCHEMA_VERSIONS + ): + continue + items.append(_history_item(row, context)) + if len(items) >= safe_limit: + break + histories_by_incident[incident_id] = items + except Exception as exc: + error = str(exc) + logger.warning( + "run_list_remediation_history_row_decode_failed", + incident_id=incident_id, + error=error, + ) + errors_by_incident[incident_id] = { + "incident_id": incident_id, + "error": error, + } + + return histories_by_incident, errors_by_incident + + async def _build_run_remediation_summaries( *, runs: list[AwoooPRunState], inbound_by_run: dict[UUID, list[AwoooPConversationEvent]], outbound_by_run: dict[UUID, list[AwoooPOutboundMessage]], + db: Any | None = None, ) -> dict[UUID, dict[str, Any]]: """Build remediation summaries for list endpoints without writing state.""" if not runs: @@ -7625,30 +7747,13 @@ async def _build_run_remediation_summaries( legacy_mcp_by_incident: dict[str, list[dict[str, Any]]] = {} errors_by_incident: dict[str, dict[str, str]] = {} if all_incident_ids: - from src.services.adr100_remediation_service import Adr100RemediationService - - service = Adr100RemediationService(record_history=False) - for incident_id in all_incident_ids: - try: - history = await service.history( - limit=_REMEDIATION_HISTORY_LIMIT, - incident_id=incident_id, - ) - histories_by_incident[incident_id] = [ - item - for item in history.get("items", []) - if isinstance(item, dict) - ] - except Exception as exc: - logger.warning( - "run_list_remediation_history_fetch_failed", - incident_id=incident_id, - error=str(exc), - ) - errors_by_incident[incident_id] = { - "incident_id": incident_id, - "error": str(exc), - } + histories_by_incident, errors_by_incident = ( + await _fetch_run_remediation_histories_by_incident( + all_incident_ids, + limit=_REMEDIATION_HISTORY_LIMIT, + db=db, + ) + ) legacy_mcp_by_incident = await _fetch_legacy_mcp_by_incident_ids( all_incident_ids, limit=min(max(len(all_incident_ids) * _REMEDIATION_HISTORY_LIMIT, 100), 5_000), @@ -8291,12 +8396,12 @@ async def list_approvals( rows = list(result.scalars().all()) inbound_by_run, outbound_by_run = await _load_run_message_context(db, rows) - - remediation_summaries = await _build_run_remediation_summaries( - runs=rows, - inbound_by_run=inbound_by_run, - outbound_by_run=outbound_by_run, - ) + remediation_summaries = await _build_run_remediation_summaries( + runs=rows, + inbound_by_run=inbound_by_run, + outbound_by_run=outbound_by_run, + db=db, + ) if remediation_status: rows = [ row diff --git a/apps/api/tests/test_platform_operator_run_remediation_batch.py b/apps/api/tests/test_platform_operator_run_remediation_batch.py new file mode 100644 index 000000000..01ed35019 --- /dev/null +++ b/apps/api/tests/test_platform_operator_run_remediation_batch.py @@ -0,0 +1,216 @@ +from datetime import UTC, datetime, timedelta +from types import SimpleNamespace +from uuid import UUID + +import pytest + +import src.services.adr100_remediation_service as adr100_remediation_service +import src.services.platform_operator_service as platform_operator_service + + +class _FakeScalars: + def __init__(self, rows: list[SimpleNamespace]) -> None: + self._rows = rows + + def all(self) -> list[SimpleNamespace]: + return self._rows + + +class _FakeResult: + def __init__(self, rows: list[SimpleNamespace]) -> None: + self._rows = rows + + def scalars(self) -> _FakeScalars: + return _FakeScalars(self._rows) + + +class _FakeDb: + def __init__( + self, + rows: list[SimpleNamespace], + *, + error: Exception | None = None, + ) -> None: + self._rows = rows + self._error = error + self.execute_calls = 0 + self.statements: list[object] = [] + + async def execute(self, statement: object) -> _FakeResult: + self.execute_calls += 1 + self.statements.append(statement) + if self._error is not None: + raise self._error + return _FakeResult(self._rows) + + +class _FakeDbContext: + def __init__(self, db: _FakeDb, enter_count: list[int]) -> None: + self._db = db + self._enter_count = enter_count + + async def __aenter__(self) -> _FakeDb: + self._enter_count[0] += 1 + return self._db + + async def __aexit__(self, *_args: object) -> None: + return None + + +def _history_row(incident_id: str, sequence: int) -> SimpleNamespace: + return SimpleNamespace( + id=f"{incident_id}-{sequence}", + incident_id=incident_id, + auto_repair_id=f"repair-{sequence}", + event_type="PRE_FLIGHT_PASSED", + actor="pytest", + success=True, + created_at=datetime(2026, 7, 14, tzinfo=UTC) + timedelta(minutes=sequence), + context={ + "schema_version": "adr100_remediation_dry_run_history_v1", + "work_item_id": f"work-{incident_id}", + "mode": "reverify", + "allowed": True, + "verification_result_preview": "success", + "post_state_summary": {"tool_count": 1, "tools": ["current_state"]}, + "mcp_route": { + "agent_id": "pytest", + "tool_name": "current_state", + "required_scope": "read", + }, + "writes_incident_state": False, + "writes_auto_repair_result": False, + }, + ) + + +async def _empty_legacy_history( + _incident_ids: list[str], + *, + limit: int, +) -> dict[str, list[dict[str, object]]]: + assert limit > 0 + return {} + + +@pytest.mark.asyncio +async def test_run_remediation_summaries_use_one_bounded_batch_query( + monkeypatch: pytest.MonkeyPatch, +) -> None: + run_a = SimpleNamespace( + run_id=UUID("11111111-1111-1111-1111-111111111111"), + state="completed", + ) + run_b = SimpleNamespace( + run_id=UUID("22222222-2222-2222-2222-222222222222"), + state="completed", + ) + incident_by_run = { + run_a.run_id: "INC-20260714-A001", + run_b.run_id: "INC-20260714-B001", + } + rows = [ + _history_row(incident_id, sequence) + for incident_id in incident_by_run.values() + for sequence in range(25) + ] + db = _FakeDb(rows) + enter_count = [0] + + monkeypatch.setattr( + platform_operator_service, + "get_db_context", + lambda *_args, **_kwargs: _FakeDbContext(db, enter_count), + ) + monkeypatch.setattr( + platform_operator_service, + "_collect_run_incident_ids", + lambda *, run, inbound_events, outbound_messages: [incident_by_run[run.run_id]], + ) + monkeypatch.setattr( + platform_operator_service, + "_fetch_legacy_mcp_by_incident_ids", + _empty_legacy_history, + ) + + async def forbidden_per_incident_history( + *_args: object, **_kwargs: object + ) -> object: + raise AssertionError( + "per-incident Adr100RemediationService.history must not run" + ) + + monkeypatch.setattr( + adr100_remediation_service.Adr100RemediationService, + "history", + forbidden_per_incident_history, + ) + + summaries = await platform_operator_service._build_run_remediation_summaries( + runs=[run_a, run_b], + inbound_by_run={}, + outbound_by_run={}, + db=db, + ) + + assert enter_count == [0] + assert db.execute_calls == 1 + assert len(db.statements) == 1 + assert summaries[run_a.run_id]["total"] == 20 + assert summaries[run_b.run_id]["total"] == 20 + assert summaries[run_a.run_id]["status"] == "read_only_dry_run" + assert summaries[run_b.run_id]["status"] == "read_only_dry_run" + assert summaries[run_a.run_id]["errors"] == [] + assert summaries[run_b.run_id]["errors"] == [] + + +@pytest.mark.asyncio +async def test_run_remediation_batch_failure_is_reported_for_each_incident( + monkeypatch: pytest.MonkeyPatch, +) -> None: + run_a = SimpleNamespace( + run_id=UUID("11111111-1111-1111-1111-111111111111"), + state="completed", + ) + run_b = SimpleNamespace( + run_id=UUID("22222222-2222-2222-2222-222222222222"), + state="completed", + ) + incident_by_run = { + run_a.run_id: "INC-20260714-A001", + run_b.run_id: "INC-20260714-B001", + } + db = _FakeDb([], error=RuntimeError("database pool exhausted")) + enter_count = [0] + + monkeypatch.setattr( + platform_operator_service, + "get_db_context", + lambda *_args, **_kwargs: _FakeDbContext(db, enter_count), + ) + monkeypatch.setattr( + platform_operator_service, + "_collect_run_incident_ids", + lambda *, run, inbound_events, outbound_messages: [incident_by_run[run.run_id]], + ) + monkeypatch.setattr( + platform_operator_service, + "_fetch_legacy_mcp_by_incident_ids", + _empty_legacy_history, + ) + + summaries = await platform_operator_service._build_run_remediation_summaries( + runs=[run_a, run_b], + inbound_by_run={}, + outbound_by_run={}, + db=db, + ) + + assert enter_count == [0] + assert db.execute_calls == 1 + for run in (run_a, run_b): + incident_id = incident_by_run[run.run_id] + assert summaries[run.run_id]["status"] == "no_evidence" + assert summaries[run.run_id]["errors"] == [ + {"incident_id": incident_id, "error": "database pool exhausted"} + ]