diff --git a/apps/api/src/services/ai_agent_autonomous_runtime_control.py b/apps/api/src/services/ai_agent_autonomous_runtime_control.py index 1c97344fc..25e730f85 100644 --- a/apps/api/src/services/ai_agent_autonomous_runtime_control.py +++ b/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, diff --git a/apps/api/tests/test_ai_agent_autonomous_runtime_control.py b/apps/api/tests/test_ai_agent_autonomous_runtime_control.py index edab3cd0c..1ef8105d8 100644 --- a/apps/api/tests/test_ai_agent_autonomous_runtime_control.py +++ b/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",