From 37cf044af3e7ba2b169cb3e52a9b292ff6373f8c Mon Sep 17 00:00:00 2001 From: Your Name Date: Wed, 1 Jul 2026 13:46:28 +0800 Subject: [PATCH] fix(agent): expose ai loop log source tags --- ..._controlled_writeback_executor_readback.py | 137 ++++++++++++++++++ ..._log_controlled_writeback_plan_readback.py | 6 + .../awoooi_priority_work_order_readback.py | 34 +++++ ...trolled_writeback_executor_readback_api.py | 42 ++++++ ..._controlled_writeback_plan_readback_api.py | 12 ++ ...awoooi_priority_work_order_readback_api.py | 43 ++++++ docs/LOGBOOK.md | 13 ++ 7 files changed, 287 insertions(+) diff --git a/apps/api/src/services/ai_agent_log_controlled_writeback_executor_readback.py b/apps/api/src/services/ai_agent_log_controlled_writeback_executor_readback.py index 013c1119a..8d981c2f5 100644 --- a/apps/api/src/services/ai_agent_log_controlled_writeback_executor_readback.py +++ b/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, diff --git a/apps/api/src/services/ai_agent_log_controlled_writeback_plan_readback.py b/apps/api/src/services/ai_agent_log_controlled_writeback_plan_readback.py index 17d4aa4a2..1232f6b29 100644 --- a/apps/api/src/services/ai_agent_log_controlled_writeback_plan_readback.py +++ b/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"), diff --git a/apps/api/src/services/awoooi_priority_work_order_readback.py b/apps/api/src/services/awoooi_priority_work_order_readback.py index 43f55f88a..ce7f0509b 100644 --- a/apps/api/src/services/awoooi_priority_work_order_readback.py +++ b/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 ) diff --git a/apps/api/tests/test_ai_agent_log_controlled_writeback_executor_readback_api.py b/apps/api/tests/test_ai_agent_log_controlled_writeback_executor_readback_api.py index f79890283..68ba8b0cc 100644 --- a/apps/api/tests/test_ai_agent_log_controlled_writeback_executor_readback_api.py +++ b/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" ) diff --git a/apps/api/tests/test_ai_agent_log_controlled_writeback_plan_readback_api.py b/apps/api/tests/test_ai_agent_log_controlled_writeback_plan_readback_api.py index 147756f17..90182d6a6 100644 --- a/apps/api/tests/test_ai_agent_log_controlled_writeback_plan_readback_api.py +++ b/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 diff --git a/apps/api/tests/test_awoooi_priority_work_order_readback_api.py b/apps/api/tests/test_awoooi_priority_work_order_readback_api.py index 1b6ad03ac..0d2ca5815 100644 --- a/apps/api/tests/test_awoooi_priority_work_order_readback_api.py +++ b/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 diff --git a/docs/LOGBOOK.md b/docs/LOGBOOK.md index b0e2827dc..7a1a1a58c 100644 --- a/docs/LOGBOOK.md +++ b/docs/LOGBOOK.md @@ -1,3 +1,16 @@ +## 2026-07-01 — 14:35 AI Loop LOG 來源貼標與分群 contract + +**照主線修正的問題**： +- AI Agent LOG controlled writeback executor 已能把 current P0 blocker 寫成 KM / RAG / PlayBook / MCP / verifier / AI Agent 批次，但缺少統一的 LOG 來源貼標 / 分群 contract，前端與 Work Items 仍不容易直接看出訊號屬於哪個 project / product / service / package / tool / source system / signal lane。 +- 本次把 current blocker recovery 從 plan 層帶上 `project_id`、`product`、`service`、`package`、`tool`、`source_system`，executor 新增 `log_source_tagging_contract`、`log_source_grouping_keys` 與每個 current blocker 的 `log_source_tags`。 +- Priority work order 同步投影 `ai_loop_current_blocker_log_source_tags`、`ai_loop_log_source_grouping_keys` 與 `ai_loop_log_source_tagging_contract`，讓 AwoooP / Work Items / AI Agent surface 可直接依 project、product、site_or_route、service、package、tool、source_system、runtime_component、signal_lane、evidence_boundary 分群。 + +**驗證**： +- `DATABASE_URL=... python3.11 -m pytest apps/api/tests/test_ai_agent_log_controlled_writeback_plan_readback_api.py apps/api/tests/test_ai_agent_log_controlled_writeback_executor_readback_api.py apps/api/tests/test_awoooi_priority_work_order_readback_api.py -q`：`15 passed`。 +- `python3.11 -m py_compile apps/api/src/services/ai_agent_log_controlled_writeback_plan_readback.py apps/api/src/services/ai_agent_log_controlled_writeback_executor_readback.py apps/api/src/services/awoooi_priority_work_order_readback.py`、`git diff --check`：通過。 + +**邊界**：只改 metadata-only LOG tagging contract、AI Loop current blocker evidence 投影、tests 與 LOGBOOK；未使用 GitHub / `gh` / GitHub API；未 workflow_dispatch；未讀 secret / token / `.env` / raw sessions / SQLite / auth；未讀 authorized_keys 內容或 `.runner` 內容；未執行 KM/RAG/PlayBook/MCP live write；未執行 110 runtime apply。 + ## 2026-07-01 — 14:15 Harbor receipt 納入 stale repair running blocker **照主線修正的問題**：