fix(agent): expose ai loop log source tags

apps/api/src/services/ai_agent_log_controlled_writeback_executor_readback.py

@@ -20,6 +20,18 @@ _SCHEMA_VERSION = "ai_agent_log_controlled_writeback_executor_readback_v1"
 _PLAN_READY_STATUS = "controlled_writeback_plan_ready"
 _TARGETS = ("km", "rag", "playbook", "mcp", "verifier", "ai_agent")
 _EXECUTOR_ROUTE = "ai_agent_metadata_writeback_executor"
+_LOG_SOURCE_GROUPING_KEYS = (
+    "project_id",
+    "product",
+    "site_or_route",
+    "service",
+    "package",
+    "tool",
+    "source_system",
+    "runtime_component",
+    "signal_lane",
+    "evidence_boundary",
+)
 
 
 def load_latest_ai_agent_log_controlled_writeback_executor_readback() -> dict[str, Any]:
@@ -95,6 +107,9 @@ def load_latest_ai_agent_log_controlled_writeback_executor_readback() -> dict[st
                 for batch in execution_batches
                 if batch["post_apply_verifier"]["required"] is True
             ),
+            "log_source_tagging_contract_count": len(_log_source_tagging_contract()),
+            "log_source_grouping_key_count": len(_LOG_SOURCE_GROUPING_KEYS),
+            "metadata_only_log_source_tagging_ready": True,
             "controlled_executor_dispatch_ready": not active_blockers,
             "controlled_apply_enabled_by_policy": True,
             "current_blocker_execution_queue_count": len(current_blocker_queue),
@@ -252,6 +267,10 @@ def _agent_consumption_context(execution_batches: list[dict[str, Any]]) -> dict[
             "km_rag_playbook_learning_loop",
             "mcp_audit_context_loop",
         ],
+        "log_source_tagging_contract": _log_source_tagging_contract(),
+        "log_source_tagging_contract_count": len(_log_source_tagging_contract()),
+        "log_source_grouping_keys": list(_LOG_SOURCE_GROUPING_KEYS),
+        "log_source_grouping_key_count": len(_LOG_SOURCE_GROUPING_KEYS),
         "evidence_chain": [
             "/api/v1/agents/agent-log-intelligence-integration-readback",
             "/api/v1/agents/agent-log-feedback-receipt-dry-run",
@@ -335,6 +354,7 @@ def _current_blocker_queue_item(recovery: dict[str, Any]) -> dict[str, Any]:
     control_path_pressure_blocker = (
         "node_load_high" if node_load_classifier == "high_load" else ""
     )
+    log_source_tags = _current_blocker_log_source_tags(recovery)
     return {
         "queue_item_id": f"current-p0-blocker::{blocker_id}",
         "source_sample_id": source_sample_id,
@@ -346,6 +366,9 @@ def _current_blocker_queue_item(recovery: dict[str, Any]) -> dict[str, Any]:
         "node_load_classifier": node_load_classifier,
         "node_load_high": node_load_classifier == "high_load",
         "registry_v2_status": recovery.get("registry_v2_status"),
+        "log_source_tags": log_source_tags,
+        "log_source_tag_count": len(log_source_tags),
+        "log_grouping_keys": list(_LOG_SOURCE_GROUPING_KEYS),
         "controlled_recovery_package": str(
             recovery.get("controlled_recovery_package") or ""
         ),
@@ -392,6 +415,120 @@ def _current_blocker_queue_item(recovery: dict[str, Any]) -> dict[str, Any]:
     }
 
 
+def _log_source_tagging_contract() -> list[dict[str, Any]]:
+    metadata_boundary = {
+        "metadata_only": True,
+        "raw_output_allowed": False,
+        "secret_value_allowed": False,
+        "writeback_targets": ["km", "rag", "playbook", "mcp", "verifier", "ai_agent"],
+    }
+    return [
+        {
+            "tag_key": "project_id",
+            "source_fields": ["target_selector.project_id", "receipt.project_id"],
+            "purpose": "group all LOG evidence by project before KM/RAG/PlayBook writeback",
+            **metadata_boundary,
+        },
+        {
+            "tag_key": "product",
+            "source_fields": ["target_selector.product", "receipt.product"],
+            "purpose": "group LOG evidence by product surface across AWOOOI/AwoooP/IwoooS",
+            **metadata_boundary,
+        },
+        {
+            "tag_key": "site_or_route",
+            "source_fields": [
+                "target_selector.service",
+                "post_recovery_readback_commands",
+                "harbor_recovery_receipt_inputs.source",
+            ],
+            "purpose": "group website, API, registry, and route verifier signals",
+            **metadata_boundary,
+        },
+        {
+            "tag_key": "service",
+            "source_fields": ["target_selector.service", "receipt.service"],
+            "purpose": "group service runtime logs and queue evidence by service",
+            **metadata_boundary,
+        },
+        {
+            "tag_key": "package",
+            "source_fields": ["target_selector.package", "receipt.package"],
+            "purpose": "group code package or repo path evidence without storing raw paths",
+            **metadata_boundary,
+        },
+        {
+            "tag_key": "tool",
+            "source_fields": ["target_selector.tool", "receipt.tool"],
+            "purpose": "group LOG evidence by tool, script, verifier, or classifier",
+            **metadata_boundary,
+        },
+        {
+            "tag_key": "source_system",
+            "source_fields": ["target_selector.source_system", "receipt.source_system"],
+            "purpose": "group raw signal origin such as k8s logs, Gitea queue, registry verifier, or runner readback",
+            **metadata_boundary,
+        },
+        {
+            "tag_key": "runtime_component",
+            "source_fields": [
+                "harbor_recovery_receipt_inputs.input_id",
+                "harbor_recovery_receipt_output_contract.output_id",
+                "queue_readback_normalizer_contract.field_id",
+            ],
+            "purpose": "group runtime components such as runner, registry, SSH control path, CD, verifier, and deploy marker",
+            **metadata_boundary,
+        },
+        {
+            "tag_key": "signal_lane",
+            "source_fields": [
+                "classification.current_blocker",
+                "classification.ssh_auth_classification",
+                "classification.risk_tier",
+            ],
+            "purpose": "group evidence into AI repair, rollback, verifier, security, host resource, and controlled apply lanes",
+            **metadata_boundary,
+        },
+        {
+            "tag_key": "evidence_boundary",
+            "source_fields": [
+                "redaction_state",
+                "metadata_only",
+                "raw_payload_required",
+                "secret_value_allowed",
+            ],
+            "purpose": "keep learning evidence metadata-only and prevent raw LOG or secret ingestion",
+            **metadata_boundary,
+        },
+    ]
+
+
+def _current_blocker_log_source_tags(recovery: dict[str, Any]) -> list[dict[str, Any]]:
+    tag_values = {
+        "project_id": str(recovery.get("project_id") or ""),
+        "product": str(recovery.get("product") or ""),
+        "site_or_route": str(recovery.get("service") or ""),
+        "service": str(recovery.get("service") or ""),
+        "package": str(recovery.get("package") or ""),
+        "tool": str(recovery.get("tool") or ""),
+        "source_system": str(recovery.get("source_system") or ""),
+        "runtime_component": "harbor_registry_110_control_path",
+        "signal_lane": str(recovery.get("blocker_id") or ""),
+        "evidence_boundary": "metadata_only_no_raw_payload",
+    }
+    return [
+        {
+            "tag_key": key,
+            "tag_value": tag_values[key],
+            "metadata_only": True,
+            "raw_output_allowed": False,
+            "secret_value_allowed": False,
+            "writeback_targets": ["km", "rag", "playbook", "mcp", "verifier", "ai_agent"],
+        }
+        for key in _LOG_SOURCE_GROUPING_KEYS
+    ]
+
+
 def _harbor_recovery_receipt_inputs() -> list[dict[str, Any]]:
     metadata_boundary = {
         "metadata_only": True,

apps/api/src/services/ai_agent_log_controlled_writeback_plan_readback.py

@@ -206,6 +206,12 @@ def _current_blocker_recovery(receipt: dict[str, Any]) -> dict[str, Any] | None:
     source_sample_id = str(receipt.get("source_sample_id") or "")
     recovery = {
         "source_sample_id": source_sample_id,
+        "project_id": str(receipt.get("project_id") or ""),
+        "product": str(receipt.get("product") or ""),
+        "service": str(receipt.get("service") or ""),
+        "package": str(receipt.get("package") or ""),
+        "tool": str(receipt.get("tool") or ""),
+        "source_system": str(receipt.get("source_system") or ""),
         "incident_id": str(classification.get("incident_id") or ""),
         "blocker_id": blocker_id,
         "risk_tier": str(classification.get("risk_tier") or "unknown"),

apps/api/src/services/awoooi_priority_work_order_readback.py

@@ -561,6 +561,18 @@ def apply_ai_loop_current_blocker_execution_queue(
         for item in queue_readback_normalizer_contract
         if item.get("field_id")
     ]
+    log_source_tags = [
+        _dict(item) for item in _list(first_item.get("log_source_tags"))
+    ]
+    log_source_tag_keys = [
+        str(item.get("tag_key") or "")
+        for item in log_source_tags
+        if item.get("tag_key")
+    ]
+    log_source_grouping_keys = _strings(context.get("log_source_grouping_keys"))
+    log_source_tagging_contract = [
+        _dict(item) for item in _list(context.get("log_source_tagging_contract"))
+    ]
     forbidden_runtime_actions = _strings(first_item.get("forbidden_runtime_actions"))
     external_blocker = str(first_item.get("external_control_path_blocker") or "")
     pressure_blocker = str(first_item.get("control_path_pressure_blocker") or "")
@@ -611,6 +623,13 @@ def apply_ai_loop_current_blocker_execution_queue(
     state["ai_loop_current_blocker_queue_readback_normalizer_field_ids"] = (
         queue_readback_normalizer_field_ids
     )
+    state["ai_loop_current_blocker_log_source_tag_count"] = len(log_source_tags)
+    state["ai_loop_current_blocker_log_source_tag_keys"] = log_source_tag_keys
+    state["ai_loop_log_source_grouping_key_count"] = len(log_source_grouping_keys)
+    state["ai_loop_log_source_grouping_keys"] = log_source_grouping_keys
+    state["ai_loop_log_source_tagging_contract_count"] = len(
+        log_source_tagging_contract
+    )
     state["ai_loop_current_blocker_forbidden_runtime_action_count"] = len(
         forbidden_runtime_actions
     )
@@ -696,6 +715,14 @@ def apply_ai_loop_current_blocker_execution_queue(
         evidence["ai_loop_current_blocker_queue_readback_normalizer_field_ids"] = (
             queue_readback_normalizer_field_ids
         )
+        evidence["ai_loop_current_blocker_log_source_tags"] = log_source_tags
+        evidence["ai_loop_current_blocker_log_source_tag_keys"] = (
+            log_source_tag_keys
+        )
+        evidence["ai_loop_log_source_grouping_keys"] = log_source_grouping_keys
+        evidence["ai_loop_log_source_tagging_contract"] = (
+            log_source_tagging_contract
+        )
         evidence["ai_loop_current_blocker_forbidden_runtime_actions"] = (
             forbidden_runtime_actions
         )
@@ -797,6 +824,13 @@ def apply_ai_loop_current_blocker_execution_queue(
     summary["ai_loop_current_blocker_queue_readback_normalizer_field_ids"] = (
         queue_readback_normalizer_field_ids
     )
+    summary["ai_loop_current_blocker_log_source_tag_count"] = len(log_source_tags)
+    summary["ai_loop_current_blocker_log_source_tag_keys"] = log_source_tag_keys
+    summary["ai_loop_log_source_grouping_key_count"] = len(log_source_grouping_keys)
+    summary["ai_loop_log_source_grouping_keys"] = log_source_grouping_keys
+    summary["ai_loop_log_source_tagging_contract_count"] = len(
+        log_source_tagging_contract
+    )
     summary["ai_loop_current_blocker_forbidden_runtime_action_count"] = len(
         forbidden_runtime_actions
     )

apps/api/tests/test_ai_agent_log_controlled_writeback_executor_readback_api.py

@@ -61,6 +61,9 @@ def _assert_executor_readback(payload: dict, *, public_endpoint: bool = False):
     assert payload["rollups"]["check_mode_ready_count"] == expected_target_count
     assert payload["rollups"]["rollback_ready_count"] == expected_target_count
     assert payload["rollups"]["post_apply_verifier_ready_count"] == expected_target_count
+    assert payload["rollups"]["log_source_tagging_contract_count"] == 10
+    assert payload["rollups"]["log_source_grouping_key_count"] == 10
+    assert payload["rollups"]["metadata_only_log_source_tagging_ready"] is True
     assert payload["rollups"]["controlled_executor_dispatch_ready"] is True
     assert payload["rollups"]["controlled_apply_enabled_by_policy"] is True
     assert payload["rollups"]["current_blocker_execution_queue_count"] == 1
@@ -117,6 +120,22 @@ def _assert_executor_readback(payload: dict, *, public_endpoint: bool = False):
     assert context["context_id"] == "ai-agent-log-controlled-writeback-consumption-v1"
     assert "ai_agent_autonomous_runtime_control" in context["consumable_by"]
     assert "km_rag_playbook_learning_loop" in context["consumable_by"]
+    assert context["log_source_tagging_contract_count"] == 10
+    assert context["log_source_grouping_keys"] == [
+        "project_id",
+        "product",
+        "site_or_route",
+        "service",
+        "package",
+        "tool",
+        "source_system",
+        "runtime_component",
+        "signal_lane",
+        "evidence_boundary",
+    ]
+    assert {
+        item["tag_key"] for item in context["log_source_tagging_contract"]
+    } == set(context["log_source_grouping_keys"])
     assert len(context["evidence_chain"]) == 4
     assert len(context["next_action_queue"]) == 6
     assert {item["target"] for item in context["next_action_queue"]} == set(batches)
@@ -132,6 +151,29 @@ def _assert_executor_readback(payload: dict, *, public_endpoint: bool = False):
     assert current_queue[0]["node_load_classifier"] == "load_not_high"
     assert current_queue[0]["node_load_high"] is False
     assert current_queue[0]["registry_v2_status"] == 502
+    assert current_queue[0]["log_source_tag_count"] == 10
+    source_tags = {
+        item["tag_key"]: item for item in current_queue[0]["log_source_tags"]
+    }
+    assert source_tags["project_id"]["tag_value"] == "awoooi"
+    assert source_tags["product"]["tag_value"] == "awoooi"
+    assert source_tags["service"]["tag_value"] == "awoooi-110-control-path"
+    assert source_tags["package"]["tag_value"] == "ops/runner"
+    assert source_tags["tool"]["tag_value"] == (
+        "read_public_gitea_queue_and_remote_control_classifier"
+    )
+    assert source_tags["source_system"]["tag_value"] == (
+        "gitea_queue_and_remote_control_classifier_readback"
+    )
+    assert source_tags["signal_lane"]["tag_value"] == (
+        "harbor_110_remote_ssh_publickey_auth_stalled"
+    )
+    assert all(
+        item["metadata_only"] is True
+        and item["raw_output_allowed"] is False
+        and item["secret_value_allowed"] is False
+        for item in source_tags.values()
+    )
     assert current_queue[0]["controlled_recovery_package"] == (
         "recover-110-control-path-and-harbor-local.sh --check"
     )

apps/api/tests/test_ai_agent_log_controlled_writeback_plan_readback_api.py

@@ -125,6 +125,18 @@ def _assert_controlled_writeback_plan(payload: dict, *, public_endpoint: bool =
     assert {
         plan["current_blocker_recovery"]["blocker_id"] for plan in p0_plans
     } == {"harbor_110_remote_ssh_publickey_auth_stalled"}
+    assert {
+        plan["current_blocker_recovery"]["project_id"] for plan in p0_plans
+    } == {"awoooi"}
+    assert {
+        plan["current_blocker_recovery"]["service"] for plan in p0_plans
+    } == {"awoooi-110-control-path"}
+    assert {
+        plan["current_blocker_recovery"]["package"] for plan in p0_plans
+    } == {"ops/runner"}
+    assert {
+        plan["current_blocker_recovery"]["tool"] for plan in p0_plans
+    } == {"read_public_gitea_queue_and_remote_control_classifier"}
     assert {
         plan["current_blocker_recovery"]["controlled_recovery_package"]
         for plan in p0_plans

apps/api/tests/test_awoooi_priority_work_order_readback_api.py

@@ -379,6 +379,43 @@ def test_awoooi_priority_work_order_readback_overlays_ai_loop_current_blocker_qu
         "gitea_queue_cd_jobs_run_id_mismatch",
         "gitea_queue_cd_jobs_stale_or_mismatched",
     ]
+    assert evidence["ai_loop_current_blocker_log_source_tag_keys"] == [
+        "project_id",
+        "product",
+        "site_or_route",
+        "service",
+        "package",
+        "tool",
+        "source_system",
+        "runtime_component",
+        "signal_lane",
+        "evidence_boundary",
+    ]
+    source_tags = {
+        item["tag_key"]: item
+        for item in evidence["ai_loop_current_blocker_log_source_tags"]
+    }
+    assert source_tags["project_id"]["tag_value"] == "awoooi"
+    assert source_tags["service"]["tag_value"] == "awoooi-110-control-path"
+    assert source_tags["tool"]["tag_value"] == (
+        "read_public_gitea_queue_and_remote_control_classifier"
+    )
+    assert source_tags["signal_lane"]["tag_value"] == (
+        "harbor_110_remote_ssh_publickey_auth_stalled"
+    )
+    assert evidence["ai_loop_log_source_grouping_keys"] == [
+        "project_id",
+        "product",
+        "site_or_route",
+        "service",
+        "package",
+        "tool",
+        "source_system",
+        "runtime_component",
+        "signal_lane",
+        "evidence_boundary",
+    ]
+    assert len(evidence["ai_loop_log_source_tagging_contract"]) == 10
     assert all(
         item["metadata_only"] is True
         and item["raw_output_allowed"] is False
@@ -432,6 +469,12 @@ def test_awoooi_priority_work_order_readback_overlays_ai_loop_current_blocker_qu
     assert payload["summary"][
         "ai_loop_current_blocker_queue_readback_normalizer_field_ids"
     ][0] == "cd_run_jobs_payload_classifier"
+    assert payload["summary"]["ai_loop_current_blocker_log_source_tag_count"] == 10
+    assert payload["summary"]["ai_loop_current_blocker_log_source_tag_keys"][0] == (
+        "project_id"
+    )
+    assert payload["summary"]["ai_loop_log_source_grouping_key_count"] == 10
+    assert payload["summary"]["ai_loop_log_source_tagging_contract_count"] == 10
     assert payload["summary"][
         "ai_loop_current_blocker_forbidden_runtime_action_count"
     ] == 6