Merge remote-tracking branch 'origin/main' into codex/mcp-control-plane-20260715

apps/api/src/services/ai_agent_log_controlled_writeback_consumer_readback.py

@@ -11,8 +11,11 @@ prove this route has rolled out.
 from __future__ import annotations
 
 import asyncio
+import copy
 import json
+import time
 from collections.abc import Mapping
+from datetime import UTC, datetime
 from typing import Any
 
 from sqlalchemy import text
@@ -51,6 +54,13 @@ _CONSUMER_DIRECT_CONNECT_TIMEOUT_SECONDS = 1.5
 _CONSUMER_DIRECT_QUERY_TIMEOUT_SECONDS = 2.0
 _CONSUMER_TELEGRAM_CONTEXT_TIMEOUT_SECONDS = 1.5
 _CONSUMER_STATEMENT_TIMEOUT_MS = 1500
+_CONSUMER_READBACK_CACHE_TTL_SECONDS = 20.0
+_consumer_readback_cache: dict[
+    tuple[int, int, str], tuple[float, dict[str, Any]]
+] = {}
+_consumer_readback_inflight: dict[
+    tuple[int, int, str], asyncio.Task[dict[str, Any]]
+] = {}
 logger = get_logger(__name__)
 
 
@@ -58,7 +68,48 @@ async def load_latest_ai_agent_log_controlled_writeback_consumer_readback(
     *,
     project_id: str = DEFAULT_PROJECT_ID,
 ) -> dict[str, Any]:
-    """Return live consumer bindings for LOG controlled writeback receipts."""
+    """Return live consumer bindings without duplicating the same DB read burst."""
+
+    loop = asyncio.get_running_loop()
+    cache_key = (id(loop), id(get_db_context), project_id)
+    cached = _consumer_readback_cache_get(cache_key=cache_key)
+    if cached is not None:
+        return cached
+
+    task = _consumer_readback_inflight.get(cache_key)
+    coalesced = task is not None
+    if task is None:
+        task = loop.create_task(
+            _load_latest_ai_agent_log_controlled_writeback_consumer_readback_uncached(
+                project_id=project_id
+            )
+        )
+        _consumer_readback_inflight[cache_key] = task
+
+    try:
+        payload = await asyncio.shield(task)
+        payload = _with_consumer_readback_freshness(
+            payload,
+            cache_hit=False,
+            cache_age_seconds=0.0,
+            inflight_coalesced=coalesced,
+        )
+        if _consumer_readback_cacheable(payload):
+            _consumer_readback_cache[cache_key] = (
+                time.monotonic(),
+                copy.deepcopy(payload),
+            )
+        return payload
+    finally:
+        if task.done() and _consumer_readback_inflight.get(cache_key) is task:
+            _consumer_readback_inflight.pop(cache_key, None)
+
+
+async def _load_latest_ai_agent_log_controlled_writeback_consumer_readback_uncached(
+    *,
+    project_id: str,
+) -> dict[str, Any]:
+    """Read current receipts once; the public loader owns cache/coalescing."""
 
     try:
         rows, consumer_rows = await _load_consumer_rows_with_retry(project_id=project_id)
@@ -94,6 +145,68 @@ async def load_latest_ai_agent_log_controlled_writeback_consumer_readback(
     )
 
 
+def _consumer_readback_cache_get(
+    *,
+    cache_key: tuple[int, int, str],
+) -> dict[str, Any] | None:
+    now = time.monotonic()
+    for key, (stored_at, _payload) in list(_consumer_readback_cache.items()):
+        if now - stored_at > _CONSUMER_READBACK_CACHE_TTL_SECONDS:
+            _consumer_readback_cache.pop(key, None)
+
+    cached = _consumer_readback_cache.get(cache_key)
+    if cached is None:
+        return None
+    stored_at, payload = cached
+    return _with_consumer_readback_freshness(
+        payload,
+        cache_hit=True,
+        cache_age_seconds=max(0.0, now - stored_at),
+        inflight_coalesced=False,
+    )
+
+
+def _consumer_readback_cacheable(payload: Mapping[str, Any]) -> bool:
+    boundaries = payload.get("operation_boundaries")
+    return bool(
+        isinstance(boundaries, Mapping)
+        and boundaries.get("metadata_ledger_read_performed") is True
+        and boundaries.get("consumer_apply_receipt_read_performed") is True
+    )
+
+
+def _with_consumer_readback_freshness(
+    payload: Mapping[str, Any],
+    *,
+    cache_hit: bool,
+    cache_age_seconds: float,
+    inflight_coalesced: bool,
+) -> dict[str, Any]:
+    result = copy.deepcopy(dict(payload))
+    prior = result.get("readback_freshness")
+    observed_at = (
+        str(prior.get("observed_at") or "")
+        if isinstance(prior, Mapping)
+        else ""
+    )
+    result["readback_freshness"] = {
+        "observed_at": observed_at or datetime.now(UTC).isoformat(),
+        "cache_ttl_seconds": _CONSUMER_READBACK_CACHE_TTL_SECONDS,
+        "cache_hit": cache_hit,
+        "cache_age_seconds": round(cache_age_seconds, 3),
+        "stale": False,
+        "inflight_coalesced": inflight_coalesced,
+    }
+    return result
+
+
+def _reset_consumer_readback_cache_for_tests() -> None:
+    """Reset process-local readback state between focused unit tests."""
+
+    _consumer_readback_cache.clear()
+    _consumer_readback_inflight.clear()
+
+
 async def _load_consumer_rows_with_retry(
     *,
     project_id: str,

apps/api/src/services/awooop_ansible_check_mode_service.py

@@ -7733,9 +7733,18 @@ async def claim_catalog_drift_failed_check_modes(
             {
                 "project_id": project_id,
                 "candidate_max_age_hours": max(1, max_age_hours),
-                "scan_limit": min(100, max(10, limit * 10)),
+                "scan_limit": 100,
             },
         )
+        ranked_rows: list[
+            tuple[
+                int,
+                dict[str, Any],
+                AnsibleCheckModeClaim,
+                str,
+                str,
+            ]
+        ] = []
         for row_value in result.mappings().all():
             row = dict(row_value)
             previous_input = _json_loads(row.get("input"))
@@ -7744,6 +7753,9 @@ async def claim_catalog_drift_failed_check_modes(
                 or previous_input.get("playbook_path")
                 or ""
             )
+            previous_catalog_id = str(
+                previous_input.get("catalog_id") or ""
+            )
             previous_catalog_revision = str(
                 previous_input.get("catalog_revision") or ""
             )
@@ -7768,6 +7780,33 @@ async def claim_catalog_drift_failed_check_modes(
                 or not (path_changed or revision_changed)
             ):
                 continue
+            catalog_route_changed = bool(
+                previous_catalog_id
+                and canonical_claim.catalog_id != previous_catalog_id
+            )
+            replay_priority = (
+                0 if catalog_route_changed else 1 if path_changed else 2
+            )
+            ranked_rows.append(
+                (
+                    replay_priority,
+                    row,
+                    canonical_claim,
+                    previous_check_path,
+                    previous_catalog_revision,
+                )
+            )
+
+        for (
+            _,
+            row,
+            canonical_claim,
+            previous_check_path,
+            previous_catalog_revision,
+        ) in sorted(ranked_rows, key=lambda item: item[0]):
+            current_catalog_revision = str(
+                canonical_claim.input_payload.get("catalog_revision") or ""
+            )
             catalog_replay_revision = (
                 current_catalog_revision
                 or f"path:{canonical_claim.playbook_path}"