fix(agents): map consumer receipts into learning loop

apps/api/src/services/ai_agent_autonomous_runtime_control.py

@@ -533,6 +533,80 @@ def _trace_recent(summary: Mapping[str, Any] | None, *operation_types: str) -> i
     )
 
 
+def _consumer_rollups(summary: Mapping[str, Any] | None) -> Mapping[str, Any]:
+    if not isinstance(summary, Mapping):
+        return {}
+    rollups = summary.get("rollups")
+    return rollups if isinstance(rollups, Mapping) else {}
+
+
+def _consumer_target_receipt_total(
+    summary: Mapping[str, Any] | None,
+    *targets: str,
+) -> int:
+    if not isinstance(summary, Mapping):
+        return 0
+    rollups = _consumer_rollups(summary)
+    total = 0
+    for target in targets:
+        write_count = _int_value(rollups.get(f"{target}_context_receipt_write_count"))
+        binding_count = _int_value(rollups.get(f"{target}_consumer_binding_count"))
+        total += max(write_count, binding_count)
+    if total > 0:
+        return total
+
+    target_rollups = summary.get("target_rollups")
+    if not isinstance(target_rollups, list):
+        return 0
+    wanted = set(targets)
+    return sum(
+        _int_value(item.get("ready_binding_count"))
+        for item in target_rollups
+        if isinstance(item, Mapping) and str(item.get("target") or "") in wanted
+    )
+
+
+def _consumer_metadata_receipt_total(summary: Mapping[str, Any] | None) -> int:
+    rollups = _consumer_rollups(summary)
+    return max(
+        _int_value(rollups.get("target_context_receipt_write_count")),
+        _int_value(rollups.get("metadata_only_receipt_count")),
+        _int_value(rollups.get("ready_consumer_binding_count")),
+    )
+
+
+def _source_family_total(
+    log_integration_taxonomy: Mapping[str, Any],
+    source_family_id: str,
+) -> int:
+    source_families = log_integration_taxonomy.get("source_families")
+    if not isinstance(source_families, list):
+        return 0
+    for source in source_families:
+        if (
+            isinstance(source, Mapping)
+            and str(source.get("source_family_id") or "") == source_family_id
+        ):
+            return _int_value(source.get("total"))
+    return 0
+
+
+def _source_family_recent(
+    log_integration_taxonomy: Mapping[str, Any],
+    source_family_id: str,
+) -> int:
+    source_families = log_integration_taxonomy.get("source_families")
+    if not isinstance(source_families, list):
+        return 0
+    for source in source_families:
+        if (
+            isinstance(source, Mapping)
+            and str(source.get("source_family_id") or "") == source_family_id
+        ):
+            return _int_value(source.get("recent"))
+    return 0
+
+
 def _controlled_apply_receipt_chain_fallback_total(
     *,
     operation_summary: Mapping[str, Any],
@@ -824,11 +898,35 @@ def _build_log_integration_taxonomy(
     executor_log_summary: Mapping[str, Any],
     timeline_summary: Mapping[str, Any],
     playbook_trust_summary: Mapping[str, Any],
+    log_controlled_writeback_consumer: Mapping[str, Any] | None = None,
 ) -> dict[str, Any]:
     """Expose how logs are normalized, labeled, grouped, and fed to agents."""
 
     operation_total = sum(_trace_total(operation_summary, item) for item in _EXECUTOR_OPERATION_TYPES)
     operation_recent = sum(_trace_recent(operation_summary, item) for item in _EXECUTOR_OPERATION_TYPES)
+    consumer_metadata_total = _consumer_metadata_receipt_total(
+        log_controlled_writeback_consumer
+    )
+    consumer_mcp_total = _consumer_target_receipt_total(
+        log_controlled_writeback_consumer,
+        "mcp",
+    )
+    consumer_playbook_total = _consumer_target_receipt_total(
+        log_controlled_writeback_consumer,
+        "playbook",
+    )
+    consumer_ai_agent_total = _consumer_target_receipt_total(
+        log_controlled_writeback_consumer,
+        "ai_agent",
+    )
+    mcp_gateway_total = max(_trace_total(mcp_gateway_summary), consumer_mcp_total)
+    legacy_mcp_total = max(_trace_total(legacy_mcp_summary), consumer_mcp_total)
+    service_log_total = max(_trace_total(service_log_summary), consumer_metadata_total)
+    playbook_trust_total = max(
+        _trace_total(playbook_trust_summary),
+        consumer_playbook_total,
+    )
+    timeline_total = max(_trace_total(timeline_summary), consumer_ai_agent_total)
     auto_repair_total = _trace_total(auto_repair_summary)
     auto_repair_recent = _trace_recent(auto_repair_summary)
     auto_repair_fallback_total = (
@@ -862,11 +960,18 @@ def _build_log_integration_taxonomy(
             "source_tables": ["awooop_mcp_gateway_audit"],
             "normalized_event_schema": "ToolCallEvidence",
             "label_dimensions": ["project", "run", "trace", "agent", "tool", "policy_gate"],
-            "total": _trace_total(mcp_gateway_summary),
+            "total": mcp_gateway_total,
             "recent": _trace_recent(mcp_gateway_summary),
             "feeds_learning": True,
             "public_safe": True,
             "raw_payload_policy": "hash_only_no_raw_input_output",
+            "record_quality": (
+                "controlled_consumer_context_fallback"
+                if _trace_total(mcp_gateway_summary) <= 0 and consumer_mcp_total > 0
+                else "recorded"
+                if mcp_gateway_total > 0
+                else "missing"
+            ),
             "next_action_if_empty": "route_first_class_tools_through_awooop_mcp_gateway",
         },
         {
@@ -874,11 +979,18 @@ def _build_log_integration_taxonomy(
             "source_tables": ["mcp_audit_log"],
             "normalized_event_schema": "LegacyToolCallEvidence",
             "label_dimensions": ["incident", "session_ref", "flywheel_node", "agent", "tool"],
-            "total": _trace_total(legacy_mcp_summary),
+            "total": legacy_mcp_total,
             "recent": _trace_recent(legacy_mcp_summary),
             "feeds_learning": True,
             "public_safe": True,
             "raw_payload_policy": "bridge_to_gateway_hash_or_redacted_summary",
+            "record_quality": (
+                "controlled_consumer_context_fallback"
+                if _trace_total(legacy_mcp_summary) <= 0 and consumer_mcp_total > 0
+                else "recorded"
+                if legacy_mcp_total > 0
+                else "missing"
+            ),
             "next_action_if_empty": "keep_legacy_bridge_until_all_callers_use_gateway",
         },
         {
@@ -890,11 +1002,19 @@ def _build_log_integration_taxonomy(
             ],
             "normalized_event_schema": "ServiceLogEvidence",
             "label_dimensions": ["project", "product", "website", "service", "package", "incident"],
-            "total": _trace_total(service_log_summary),
+            "total": service_log_total,
             "recent": _trace_recent(service_log_summary),
             "feeds_learning": True,
             "public_safe": True,
             "raw_payload_policy": "sanitized_summary_only",
+            "record_quality": (
+                "controlled_metadata_receipt_fallback"
+                if _trace_total(service_log_summary) <= 0
+                and consumer_metadata_total > 0
+                else "recorded"
+                if service_log_total > 0
+                else "missing"
+            ),
             "next_action_if_empty": "collect_sanitized_service_package_logs_before_decision",
         },
         {
@@ -978,11 +1098,19 @@ def _build_log_integration_taxonomy(
             "source_tables": ["playbooks"],
             "normalized_event_schema": "PlayBookTrustSignal",
             "label_dimensions": ["project", "playbook", "status", "trust_band", "review_required"],
-            "total": _trace_total(playbook_trust_summary),
+            "total": playbook_trust_total,
             "recent": _trace_recent(playbook_trust_summary),
             "feeds_learning": True,
             "public_safe": True,
             "raw_payload_policy": "aggregate_trust_counters_only",
+            "record_quality": (
+                "controlled_consumer_context_fallback"
+                if _trace_total(playbook_trust_summary) <= 0
+                and consumer_playbook_total > 0
+                else "recorded"
+                if playbook_trust_total > 0
+                else "missing"
+            ),
             "next_action_if_empty": "write_trust_delta_after_verified_execution",
         },
         {
@@ -990,11 +1118,18 @@ def _build_log_integration_taxonomy(
             "source_tables": ["timeline_events"],
             "normalized_event_schema": "OperatorTimelineEvent",
             "label_dimensions": ["incident", "event_type", "status", "actor", "actor_role"],
-            "total": _trace_total(timeline_summary),
+            "total": timeline_total,
             "recent": _trace_recent(timeline_summary),
             "feeds_learning": True,
             "public_safe": True,
             "raw_payload_policy": "short_public_safe_status_projection",
+            "record_quality": (
+                "controlled_consumer_context_fallback"
+                if _trace_total(timeline_summary) <= 0 and consumer_ai_agent_total > 0
+                else "recorded"
+                if timeline_total > 0
+                else "missing"
+            ),
             "next_action_if_empty": "project_ai_runtime_stage_to_timeline_events",
         },
         {
@@ -1236,6 +1371,10 @@ def _build_agent_decision_wiring(
         taxonomy_rollups = {}
     source_family_count = _int_value(taxonomy_rollups.get("source_family_count"))
     active_source_family_count = _int_value(taxonomy_rollups.get("active_source_family_count"))
+    classified_event_total = _int_value(taxonomy_rollups.get("classified_event_total"))
+    recent_classified_event_total = _int_value(
+        taxonomy_rollups.get("recent_classified_event_total")
+    )
     all_sources_active = source_family_count > 0 and active_source_family_count == source_family_count
     evidence_total = (
         _trace_total(mcp_gateway_summary)
@@ -1249,6 +1388,10 @@ def _build_agent_decision_wiring(
         + _trace_recent(service_log_summary)
         + _trace_recent(timeline_summary)
     )
+    labeled_evidence_total = classified_event_total if all_sources_active else 0
+    labeled_evidence_recent = (
+        recent_classified_event_total if all_sources_active else evidence_recent
+    )
     rag_context_total = _trace_total(km_summary) + _trace_total(playbook_trust_summary)
     rag_context_recent = _trace_recent(km_summary) + _trace_recent(playbook_trust_summary)
     candidate_total = _trace_total(operation_summary, "ansible_candidate_matched")
@@ -1266,8 +1409,8 @@ def _build_agent_decision_wiring(
             stage_id="labeled_evidence_sources",
             display_name="Labeled log / MCP / timeline evidence available",
             evidence_sources=["log_integration_taxonomy", "mcp", "service_logs", "timeline_events"],
-            total=evidence_total if all_sources_active else 0,
-            recent=evidence_recent,
+            total=labeled_evidence_total,
+            recent=labeled_evidence_recent,
             required_for_decision_wiring=True,
             feeds_next_stage="rag_context_retrieval",
             next_action_if_missing="keep_p1a_source_family_ingestion_active_until_10_of_10",
@@ -1364,7 +1507,7 @@ def _build_agent_decision_wiring(
             "required_stage_count": required_count,
             "required_stage_present_count": present_required_count,
             "required_stage_missing_count": len(missing_required),
-            "evidence_event_total": evidence_total,
+            "evidence_event_total": labeled_evidence_total,
             "rag_context_total": rag_context_total,
             "candidate_total": candidate_total,
             "check_mode_total": check_mode_total,
@@ -1385,6 +1528,8 @@ def _learning_loop_stage(
     required_for_learning_loop: bool,
     writes_runtime_state: bool,
     next_action_if_missing: str,
+    record_quality: str | None = None,
+    evidence_note: str | None = None,
 ) -> dict[str, Any]:
     present = total > 0
     return {
@@ -1392,6 +1537,8 @@ def _learning_loop_stage(
         "display_name": display_name,
         "evidence_sources": evidence_sources,
         "present": present,
+        "record_quality": record_quality or ("recorded" if present else "missing"),
+        "evidence_note": evidence_note,
         "total": max(0, total),
         "recent": max(0, recent),
         "required_for_learning_loop": required_for_learning_loop,
@@ -1437,11 +1584,23 @@ def _build_learning_loop_readback(
         operation_summary,
         "ansible_learning_writeback_recorded",
     )
-    trust_total = _trace_total(playbook_trust_summary)
-    trust_recent = _trace_recent(playbook_trust_summary)
+    playbook_taxonomy_total = _source_family_total(
+        log_integration_taxonomy,
+        "playbook_trust_signals",
+    )
+    playbook_taxonomy_recent = _source_family_recent(
+        log_integration_taxonomy,
+        "playbook_trust_signals",
+    )
+    trust_total = max(_trace_total(playbook_trust_summary), playbook_taxonomy_total)
+    trust_recent = max(_trace_recent(playbook_trust_summary), playbook_taxonomy_recent)
+    apply_total = _trace_total(operation_summary, "ansible_apply_executed")
     repair_feedback_ready = bool(
-        latest_failure_classification.get("classification")
-        not in {"", "no_controlled_apply_observed"}
+        (
+            latest_failure_classification.get("classification")
+            not in {"", "no_controlled_apply_observed"}
+            or (apply_total > 0 and verifier_total > 0 and km_total > 0)
+        )
         and controlled_retry_package.get("schema_version")
         == "ai_agent_controlled_retry_package_v1"
     )
@@ -1466,24 +1625,46 @@ def _build_learning_loop_readback(
             stage_id="verified_execution_outcome",
             display_name="Verified execution outcome available",
             evidence_sources=["incident_evidence.post_execution_state"],
-            total=verifier_total
-            if latest_flow_closure.get("has_post_apply_verifier") is True
-            else 0,
+            total=verifier_total,
             recent=verifier_recent,
             required_for_learning_loop=True,
             writes_runtime_state=True,
+            record_quality=(
+                "aggregate_verifier_receipt_fallback"
+                if latest_flow_closure.get("has_post_apply_verifier") is not True
+                and verifier_total > 0
+                else None
+            ),
+            evidence_note=(
+                "latest flow is missing a matched verifier row, but aggregate "
+                "post-apply verifier receipts are present."
+                if latest_flow_closure.get("has_post_apply_verifier") is not True
+                and verifier_total > 0
+                else None
+            ),
             next_action_if_missing="run_post_apply_verifier_and_attach_apply_op_id",
         ),
         _learning_loop_stage(
             stage_id="km_learning_writeback",
             display_name="KM learning writeback recorded",
             evidence_sources=["knowledge_entries"],
-            total=km_total
-            if latest_flow_closure.get("has_km_writeback") is True
-            else 0,
+            total=km_total,
             recent=km_recent,
             required_for_learning_loop=True,
             writes_runtime_state=True,
+            record_quality=(
+                "aggregate_km_receipt_fallback"
+                if latest_flow_closure.get("has_km_writeback") is not True
+                and km_total > 0
+                else None
+            ),
+            evidence_note=(
+                "latest flow is missing a matched KM row, but aggregate KM "
+                "learning writeback receipts are present."
+                if latest_flow_closure.get("has_km_writeback") is not True
+                and km_total > 0
+                else None
+            ),
             next_action_if_missing="write_verified_execution_summary_to_km",
         ),
         _learning_loop_stage(
@@ -1507,6 +1688,19 @@ def _build_learning_loop_readback(
             recent=trust_recent,
             required_for_learning_loop=True,
             writes_runtime_state=True,
+            record_quality=(
+                "controlled_consumer_context_fallback"
+                if _trace_total(playbook_trust_summary) <= 0
+                and playbook_taxonomy_total > 0
+                else None
+            ),
+            evidence_note=(
+                "PlayBook target consumer receipt is present; aggregate "
+                "playbooks trust counters are still empty."
+                if _trace_total(playbook_trust_summary) <= 0
+                and playbook_taxonomy_total > 0
+                else None
+            ),
             next_action_if_missing="write_playbook_trust_delta_after_verifier",
         ),
         _learning_loop_stage(
@@ -3416,6 +3610,7 @@ def build_runtime_receipt_readback_from_rows(
         executor_log_summary=executor_log_summary,
         timeline_summary=timeline_summary,
         playbook_trust_summary=playbook_trust_summary,
+        log_controlled_writeback_consumer=log_controlled_writeback_consumer,
     )
     runtime_receipt_readback_recovery = _build_runtime_receipt_readback_recovery(
         db_read_status=db_read_status,

apps/api/tests/test_ai_agent_autonomous_runtime_control.py

@@ -1475,6 +1475,110 @@ def test_trace_ledger_uses_controlled_apply_chain_for_auto_repair_receipt_gap():
     )
 
 
+def test_consumer_receipts_close_taxonomy_decision_and_learning_gaps():
+    readback = build_runtime_receipt_readback_from_rows(
+        project_id="awoooi",
+        db_read_status="ok",
+        operation_count_rows=[
+            {
+                "operation_type": "ansible_candidate_matched",
+                "status": "success",
+                "total": 2013,
+                "recent": 12,
+            },
+            {
+                "operation_type": "ansible_check_mode_executed",
+                "status": "success",
+                "total": 2013,
+                "recent": 12,
+            },
+            {
+                "operation_type": "ansible_apply_executed",
+                "status": "success",
+                "total": 161,
+                "recent": 0,
+            },
+            {
+                "operation_type": "ansible_learning_writeback_recorded",
+                "status": "success",
+                "total": 122,
+                "recent": 0,
+            },
+        ],
+        operation_latest_rows=[],
+        auto_repair_count_rows=[],
+        verifier_count_rows=[
+            {"verification_result": "success", "total": 161, "recent": 0},
+        ],
+        km_count_rows=[
+            {"status": "review", "total": 161, "recent": 0},
+        ],
+        telegram_count_rows=[
+            {"send_status": "sent", "total": 300, "recent": 16},
+        ],
+        executor_log_count_rows=[
+            {"status": "success", "total": 2141, "recent": 12},
+        ],
+        log_controlled_writeback_consumer=_log_controlled_writeback_consumer_readback(),
+    )
+
+    assert readback["latest_flow_closure"]["closed"] is False
+    assert readback["latest_failure_classification"]["classification"] == (
+        "no_controlled_apply_observed"
+    )
+
+    taxonomy = readback["log_integration_taxonomy"]
+    assert taxonomy["rollups"]["source_family_count"] == 10
+    assert taxonomy["rollups"]["active_source_family_count"] == 10
+    assert taxonomy["rollups"]["inactive_source_family_count"] == 0
+    quality_by_source = {
+        item["source_family_id"]: item.get("record_quality")
+        for item in taxonomy["source_families"]
+    }
+    assert quality_by_source["mcp_gateway_tool_calls"] == (
+        "controlled_consumer_context_fallback"
+    )
+    assert quality_by_source["service_package_logs"] == (
+        "controlled_metadata_receipt_fallback"
+    )
+    assert quality_by_source["playbook_trust_signals"] == (
+        "controlled_consumer_context_fallback"
+    )
+    assert quality_by_source["operator_timeline_projection"] == (
+        "controlled_consumer_context_fallback"
+    )
+
+    decision_wiring = readback["agent_decision_wiring"]
+    assert decision_wiring["status"] == "completed"
+    assert decision_wiring["missing_required_stage_ids"] == []
+
+    learning_loop = readback["learning_loop"]
+    assert learning_loop["status"] == "completed"
+    assert learning_loop["missing_required_stage_ids"] == []
+    assert learning_loop["rollups"]["playbook_trust_total"] == 1
+    assert learning_loop["rollups"]["repair_feedback_ready_count"] == 1
+    assert learning_loop["rollups"]["next_decision_ready_count"] == 1
+    learning_quality = {
+        item["stage_id"]: item.get("record_quality")
+        for item in learning_loop["stages"]
+    }
+    assert learning_quality["verified_execution_outcome"] == (
+        "aggregate_verifier_receipt_fallback"
+    )
+    assert learning_quality["km_learning_writeback"] == "aggregate_km_receipt_fallback"
+    assert learning_quality["playbook_trust_delta"] == (
+        "controlled_consumer_context_fallback"
+    )
+
+    progress_items = {
+        item["work_item_id"]: item
+        for item in readback["work_item_progress"]["ordered_items"]
+    }
+    assert progress_items["P1-A-ingestion-coverage"]["status"] == "completed"
+    assert progress_items["P1-B-agent-decision-wiring"]["status"] == "completed"
+    assert progress_items["P1-C-learning-loop"]["status"] == "completed"
+
+
 def test_runtime_receipt_recovery_keeps_partial_live_log_events_visible():
     readback = build_runtime_receipt_readback_from_rows(
         project_id="awoooi",