wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 3 Packages Projects Releases Wiki Activity

fix(api): batch run remediation evidence reads

Browse Source
This commit is contained in:
ogt
2026-07-14 16:53:08 +08:00
parent 5ec6fd3c30
commit ceae823172
2 changed files with 352 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

167
apps/api/src/services/platform_operator_service.py
View File

@@ -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

216
apps/api/tests/test_platform_operator_run_remediation_batch.py Normal file
View File

@@ -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"}
]