wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 3 Packages Projects Releases Wiki Activity

feat(api): expose log consumer readback
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 0s
Details
CD Pipeline / cancel-stale-cd (push) Has been skipped
Details
CD Pipeline / tests (push) Successful in 29s
Details
CD Pipeline / post-deploy-checks (push) Has been cancelled
Details
CD Pipeline / build-and-deploy (push) Has been cancelled
Details

Browse Source
This commit is contained in:
Your Name
2026-06-30 07:43:06 +08:00
parent 43396f5603
commit 1de80f86cb
8 changed files with 866 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

182
apps/api/src/services/ai_agent_autonomous_runtime_control.py
View File

@@ -18,6 +18,9 @@ from sqlalchemy import text
from src.core.config import settings
from src.core.logging import get_logger
from src.db.base import get_db_context
from src.services.ai_agent_log_controlled_writeback_consumer_readback import (
load_latest_ai_agent_log_controlled_writeback_consumer_readback,
)
from src.services.ai_agent_log_controlled_writeback_dispatch import (
OPERATION_TYPE as LOG_CONTROLLED_WRITEBACK_DISPATCH_OPERATION_TYPE,
)
@@ -269,6 +272,91 @@ def _load_log_controlled_writeback_executor_readback() -> dict[str, Any]:
}
def _fallback_log_controlled_writeback_consumer_readback(
error_type: str | None = None,
) -> dict[str, Any]:
readback = {
"schema_version": "ai_agent_log_controlled_writeback_consumer_readback_v1",
"priority": "P1-LOG-KM-RAG-MCP-PLAYBOOK",
"scope": "ai_agent_log_controlled_writeback_consumer_readback",
"status": "blocked_waiting_controlled_writeback_consumer_receipts",
"readback": {
"workplan_id": "P1-LOG-CONTROLLED-WRITEBACK-CONSUMER-READBACK",
"workplan_title": (
"LOG metadata ledger receipts consumable by KM / RAG / PlayBook / "
"MCP / verifier / AI Agent context"
),
"source_operation_type": LOG_CONTROLLED_WRITEBACK_DISPATCH_OPERATION_TYPE,
"source_executor_route": "ai_agent_metadata_writeback_executor",
"safe_next_step": "repair_log_controlled_writeback_consumer_readback_then_retry",
},
"controlled_consume": {
"mode": "blocked_waiting_consumer_readback",
"controlled_consume_allowed": False,
"owner_review_required_for_low_medium_high": False,
"critical_break_glass_required": True,
"target_selector_required": True,
"source_of_truth_diff_required": True,
"check_mode_required": True,
"rollback_required": True,
"post_apply_verifier_required": True,
"runtime_target_write_performed": False,
},
"consumer_bindings": [],
"target_rollups": [],
"rollups": {
"target_count": 6,
"dispatch_ledger_row_count": 0,
"consumer_binding_count": 0,
"ready_consumer_binding_count": 0,
"ready_target_count": 0,
"metadata_only_receipt_count": 0,
"post_apply_verifier_ref_count": 0,
"controlled_consumer_readback_ready": False,
"runtime_target_write_performed": False,
},
"active_blockers": ["log_controlled_writeback_consumer_readback_unavailable"],
"operation_boundaries": {
"consumer_readback_only": True,
"metadata_ledger_read_performed": False,
"km_write_performed": False,
"rag_index_write_performed": False,
"playbook_trust_write_performed": False,
"mcp_tool_call_performed": False,
"agent_runtime_action_performed": False,
"telegram_send_performed": False,
"workflow_trigger_performed": False,
"raw_log_payload_persisted": False,
"secret_value_collection_allowed": False,
"github_api_used": False,
},
}
if error_type:
readback["readback"]["error_type"] = error_type
return readback
async def _load_log_controlled_writeback_consumer_readback(
*,
project_id: str,
) -> dict[str, Any]:
"""Attach LOG consumer bindings without writing KM/RAG/PlayBook/MCP targets."""
try:
return await load_latest_ai_agent_log_controlled_writeback_consumer_readback(
project_id=project_id,
)
except Exception as exc: # pragma: no cover - keeps runtime control API visible
logger.warning(
"log_controlled_writeback_consumer_readback_failed",
project_id=project_id,
error_type=type(exc).__name__,
)
return _fallback_log_controlled_writeback_consumer_readback(
error_type=type(exc).__name__,
)
def _trace_stage(
*,
stage_id: str,
@@ -1625,6 +1713,7 @@ def _build_work_item_progress(
trace_ledger: Mapping[str, Any],
log_integration_taxonomy: Mapping[str, Any],
log_controlled_writeback_executor: Mapping[str, Any],
log_controlled_writeback_consumer: Mapping[str, Any],
agent_decision_wiring: Mapping[str, Any],
learning_loop: Mapping[str, Any],
alert_noise_reduction: Mapping[str, Any],
@@ -1687,6 +1776,20 @@ def _build_work_item_progress(
and log_executor_rollups.get("controlled_executor_dispatch_ready") is True
and not log_executor_blockers
)
log_consumer_rollups = log_controlled_writeback_consumer.get("rollups")
if not isinstance(log_consumer_rollups, Mapping):
log_consumer_rollups = {}
log_consumer_blockers = log_controlled_writeback_consumer.get("active_blockers")
if not isinstance(log_consumer_blockers, list):
log_consumer_blockers = []
log_consumer_ready = (
log_controlled_writeback_consumer.get("schema_version")
== "ai_agent_log_controlled_writeback_consumer_readback_v1"
and log_controlled_writeback_consumer.get("status")
== "controlled_writeback_consumer_readback_ready"
and log_consumer_rollups.get("controlled_consumer_readback_ready") is True
and not log_consumer_blockers
)
ui_rollups = ui_productization.get("rollups")
if not isinstance(ui_rollups, Mapping):
ui_rollups = {}
@@ -1799,11 +1902,24 @@ def _build_work_item_progress(
),
"active_blocker_count": len(log_executor_blockers),
},
{
"work_item_id": "P1-F-log-controlled-writeback-consumer",
"priority": "P1-F",
"title": "LOG metadata receipts consumable by KM / RAG / MCP / PlayBook / AI Agent",
"status": "completed" if log_consumer_ready else "in_progress" if log_executor_ready else "pending",
"exit_criteria": "runtime-control exposes ready consumer bindings for all LOG metadata writeback targets",
"remaining_consumer_binding_count": max(
0,
_int_value(log_consumer_rollups.get("target_count"))
- _int_value(log_consumer_rollups.get("ready_target_count")),
),
"active_blocker_count": len(log_consumer_blockers),
},
{
"work_item_id": "P2-A-ui-ux-productization",
"priority": "P2-A",
"title": "Professional product UI replacing text-heavy surfaces",
"status": "completed" if p2a_completed else "in_progress" if log_executor_ready else "pending",
"status": "completed" if p2a_completed else "in_progress" if log_consumer_ready else "pending",
"exit_criteria": "AI automation status is shown as dense dashboard controls, filters, counters, and action rails",
"remaining_ui_surface_count": ui_surface_missing,
},
@@ -2584,6 +2700,7 @@ def build_runtime_receipt_readback_from_rows(
alert_operation_count_rows: Iterable[Mapping[str, Any] | Any] = (),
alertmanager_event_count_rows: Iterable[Mapping[str, Any] | Any] = (),
grouped_alert_event_count_rows: Iterable[Mapping[str, Any] | Any] = (),
log_controlled_writeback_consumer: Mapping[str, Any] | None = None,
error_type: str | None = None,
) -> dict[str, Any]:
"""Build the live executor receipt readback from already-fetched rows."""
@@ -2709,10 +2826,15 @@ def build_runtime_receipt_readback_from_rows(
ui_productization = _build_ui_productization_readback()
multi_product_taxonomy = _build_multi_product_taxonomy_contract(log_integration_taxonomy)
log_controlled_writeback_executor = _load_log_controlled_writeback_executor_readback()
if not isinstance(log_controlled_writeback_consumer, Mapping):
log_controlled_writeback_consumer = (
_fallback_log_controlled_writeback_consumer_readback()
)
work_item_progress = _build_work_item_progress(
trace_ledger=trace_ledger,
log_integration_taxonomy=log_integration_taxonomy,
log_controlled_writeback_executor=log_controlled_writeback_executor,
log_controlled_writeback_consumer=log_controlled_writeback_consumer,
agent_decision_wiring=agent_decision_wiring,
learning_loop=learning_loop,
alert_noise_reduction=alert_noise_reduction,
@@ -2840,6 +2962,7 @@ def build_runtime_receipt_readback_from_rows(
"trace_ledger": trace_ledger,
"log_integration_taxonomy": log_integration_taxonomy,
"log_controlled_writeback_executor": log_controlled_writeback_executor,
"log_controlled_writeback_consumer": dict(log_controlled_writeback_consumer),
"agent_decision_wiring": agent_decision_wiring,
"learning_loop": learning_loop,
"alert_noise_reduction": alert_noise_reduction,
@@ -2876,6 +2999,15 @@ def _attach_runtime_receipt_readback(
log_executor_blockers = log_executor.get("active_blockers")
if not isinstance(log_executor_blockers, list):
log_executor_blockers = []
log_consumer = readback.get("log_controlled_writeback_consumer")
if not isinstance(log_consumer, Mapping):
log_consumer = {}
log_consumer_rollups = log_consumer.get("rollups")
if not isinstance(log_consumer_rollups, Mapping):
log_consumer_rollups = {}
log_consumer_blockers = log_consumer.get("active_blockers")
if not isinstance(log_consumer_blockers, list):
log_consumer_blockers = []
operation_counts = (readback.get("ansible_operations") or {}).get("counts")
if not isinstance(operation_counts, Mapping):
operation_counts = {}
@@ -2992,6 +3124,48 @@ def _attach_runtime_receipt_readback(
"live_log_controlled_writeback_recent_dispatch_count": _int_value(
log_dispatch_summary.get("recent")
),
"live_log_controlled_writeback_consumer_binding_count": _int_value(
log_consumer_rollups.get("consumer_binding_count")
),
"live_log_controlled_writeback_consumer_ready_binding_count": _int_value(
log_consumer_rollups.get("ready_consumer_binding_count")
),
"live_log_controlled_writeback_consumer_ready_target_count": _int_value(
log_consumer_rollups.get("ready_target_count")
),
"live_log_controlled_writeback_consumer_ready_count": (
1
if log_consumer.get("status")
== "controlled_writeback_consumer_readback_ready"
else 0
),
"live_log_controlled_writeback_consumer_blocker_count": len(
log_consumer_blockers
),
"live_log_controlled_writeback_consumer_metadata_only_count": _int_value(
log_consumer_rollups.get("metadata_only_receipt_count")
),
"live_log_controlled_writeback_consumer_verifier_ref_count": _int_value(
log_consumer_rollups.get("post_apply_verifier_ref_count")
),
"live_log_controlled_writeback_km_consumer_binding_count": _int_value(
log_consumer_rollups.get("km_consumer_binding_count")
),
"live_log_controlled_writeback_rag_consumer_binding_count": _int_value(
log_consumer_rollups.get("rag_consumer_binding_count")
),
"live_log_controlled_writeback_playbook_consumer_binding_count": _int_value(
log_consumer_rollups.get("playbook_consumer_binding_count")
),
"live_log_controlled_writeback_mcp_consumer_binding_count": _int_value(
log_consumer_rollups.get("mcp_consumer_binding_count")
),
"live_log_controlled_writeback_verifier_consumer_binding_count": _int_value(
log_consumer_rollups.get("verifier_consumer_binding_count")
),
"live_log_controlled_writeback_ai_agent_consumer_binding_count": _int_value(
log_consumer_rollups.get("ai_agent_consumer_binding_count")
),
"live_agent_decision_wiring_stage_count": _int_value(
((readback.get("agent_decision_wiring") or {}).get("rollups") or {}).get(
"stage_count"
@@ -3483,6 +3657,11 @@ async def load_ai_agent_autonomous_runtime_receipt_readback(
"grouped_alert_event_counts",
_RUNTIME_GROUPED_ALERT_EVENT_COUNTS_SQL,
)
log_controlled_writeback_consumer = (
await _load_log_controlled_writeback_consumer_readback(
project_id=project_id,
)
)
except Exception as exc:
logger.warning(
"ai_agent_autonomous_runtime_receipt_readback_failed",
@@ -3519,6 +3698,7 @@ async def load_ai_agent_autonomous_runtime_receipt_readback(
alert_operation_count_rows=alert_operation_counts,
alertmanager_event_count_rows=alertmanager_event_counts,
grouped_alert_event_count_rows=grouped_alert_event_counts,
log_controlled_writeback_consumer=log_controlled_writeback_consumer,
)

315
apps/api/src/services/ai_agent_log_controlled_writeback_consumer_readback.py Normal file
View File

@@ -0,0 +1,315 @@
"""AI Agent LOG controlled writeback consumer readback.
Reads live metadata-only dispatch receipts from automation_operation_log and
turns them into KM / RAG / PlayBook / MCP / verifier / AI Agent consumer
bindings. This endpoint does not write those target systems, call MCP tools,
trigger workflows, or persist raw log payloads.
"""
from __future__ import annotations
from collections.abc import Mapping
from typing import Any
from sqlalchemy import text
from src.db.base import get_db_context
from src.services.ai_agent_log_controlled_writeback_dispatch import (
EXECUTOR_ROUTE,
OPERATION_TYPE,
)
SCHEMA_VERSION = "ai_agent_log_controlled_writeback_consumer_readback_v1"
DEFAULT_PROJECT_ID = "awoooi"
_TARGETS = ("km", "rag", "playbook", "mcp", "verifier", "ai_agent")
_CONSUMER_SURFACES = {
"km": "knowledge_memory_context",
"rag": "rag_index_candidate_context",
"playbook": "playbook_trust_candidate_context",
"mcp": "mcp_audit_feedback_context",
"verifier": "post_apply_verifier_feedback_context",
"ai_agent": "autonomous_runtime_decision_context",
}
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."""
async with get_db_context(project_id) as db:
result = await db.execute(
text("""
SELECT
op_id::text AS op_id,
operation_type,
actor,
status,
created_at,
input ->> 'semantic_operation_type' AS semantic_operation_type,
input ->> 'ledger_operation_type' AS ledger_operation_type,
input ->> 'dispatch_receipt_id' AS dispatch_receipt_id,
input ->> 'batch_id' AS batch_id,
input ->> 'target' AS target,
input ->> 'target_surface' AS target_surface,
input ->> 'risk_tier' AS risk_tier,
input ->> 'project_id' AS project_id,
input ->> 'raw_payload_included' AS raw_payload_included,
output ->> 'next_action' AS next_action,
output ->> 'km_write_performed' AS km_write_performed,
output ->> 'rag_index_write_performed' AS rag_index_write_performed,
output ->> 'playbook_trust_write_performed' AS playbook_trust_write_performed,
output ->> 'mcp_tool_call_performed' AS mcp_tool_call_performed,
output ->> 'telegram_send_performed' AS telegram_send_performed,
output -> 'post_apply_verifier_refs' AS post_apply_verifier_refs
FROM automation_operation_log
WHERE coalesce(input ->> 'semantic_operation_type', operation_type)
= :operation_type
ORDER BY created_at DESC, op_id DESC
LIMIT 50
"""),
{"operation_type": OPERATION_TYPE},
)
rows = _result_rows(result)
bindings = [_consumer_binding(row) for row in rows]
active_blockers = _active_blockers(bindings)
return {
"schema_version": SCHEMA_VERSION,
"priority": "P1-LOG-KM-RAG-MCP-PLAYBOOK",
"scope": "ai_agent_log_controlled_writeback_consumer_readback",
"status": (
"controlled_writeback_consumer_readback_ready"
if not active_blockers
else "blocked_waiting_controlled_writeback_consumer_receipts"
),
"readback": {
"workplan_id": "P1-LOG-CONTROLLED-WRITEBACK-CONSUMER-READBACK",
"workplan_title": (
"LOG metadata ledger receipts consumable by KM / RAG / PlayBook / "
"MCP / verifier / AI Agent context"
),
"source_operation_type": OPERATION_TYPE,
"source_executor_route": EXECUTOR_ROUTE,
"safe_next_step": (
"consume_metadata_receipts_in_km_rag_playbook_agent_context_with_post_apply_verifier"
),
},
"controlled_consume": {
"mode": "controlled_consumer_readback_ready"
if not active_blockers
else "blocked_waiting_consumer_readback",
"controlled_consume_allowed": not active_blockers,
"owner_review_required_for_low_medium_high": False,
"critical_break_glass_required": True,
"target_selector_required": True,
"source_of_truth_diff_required": True,
"check_mode_required": True,
"rollback_required": True,
"post_apply_verifier_required": True,
"runtime_target_write_performed": False,
},
"consumer_bindings": bindings,
"target_rollups": _target_rollups(bindings),
"rollups": {
"target_count": len(_TARGETS),
"dispatch_ledger_row_count": len(rows),
"consumer_binding_count": len(bindings),
"ready_consumer_binding_count": sum(
1 for binding in bindings if binding["status"] == "ready_for_consumer_context"
),
"ready_target_count": sum(
1 for item in _target_rollups(bindings) if item["ready_binding_count"] > 0
),
"metadata_only_receipt_count": sum(
1 for binding in bindings if binding["raw_payload_included"] is False
),
"post_apply_verifier_ref_count": sum(
len(binding["post_apply_verifier_refs"]) for binding in bindings
),
"controlled_consumer_readback_ready": not active_blockers,
"km_consumer_binding_count": _target_count(bindings, "km"),
"rag_consumer_binding_count": _target_count(bindings, "rag"),
"playbook_consumer_binding_count": _target_count(bindings, "playbook"),
"mcp_consumer_binding_count": _target_count(bindings, "mcp"),
"verifier_consumer_binding_count": _target_count(bindings, "verifier"),
"ai_agent_consumer_binding_count": _target_count(bindings, "ai_agent"),
"runtime_target_write_performed": False,
},
"active_blockers": active_blockers,
"operation_boundaries": {
"consumer_readback_only": True,
"metadata_ledger_read_performed": True,
"km_write_performed": False,
"rag_index_write_performed": False,
"playbook_trust_write_performed": False,
"mcp_tool_call_performed": False,
"agent_runtime_action_performed": False,
"telegram_send_performed": False,
"workflow_trigger_performed": False,
"raw_log_payload_persisted": False,
"secret_value_collection_allowed": False,
"github_api_used": False,
},
}
def _consumer_binding(row: Mapping[str, Any]) -> dict[str, Any]:
target = str(row.get("target") or "")
verifier_refs = _strings(row.get("post_apply_verifier_refs"))
raw_payload_included = str(row.get("raw_payload_included") or "").lower() == "true"
return {
"consumer_binding_id": f"consumer::{row.get('dispatch_receipt_id') or row.get('op_id')}",
"dispatch_receipt_id": str(row.get("dispatch_receipt_id") or ""),
"ledger_op_id": str(row.get("op_id") or ""),
"target": target,
"target_surface": str(row.get("target_surface") or ""),
"consumer_surface": _CONSUMER_SURFACES.get(target, "unknown_consumer_surface"),
"risk_tier": str(row.get("risk_tier") or ""),
"status": (
"ready_for_consumer_context"
if _binding_ready(row, raw_payload_included)
else "blocked_waiting_consumer_binding_controls"
),
"project_id": str(row.get("project_id") or ""),
"executor_route": str(row.get("actor") or ""),
"semantic_operation_type": str(row.get("semantic_operation_type") or ""),
"ledger_operation_type": str(row.get("ledger_operation_type") or ""),
"ledger_status": str(row.get("status") or ""),
"raw_payload_included": raw_payload_included,
"next_action": str(row.get("next_action") or ""),
"post_apply_verifier_refs": verifier_refs,
"target_selector": {
"dispatch_receipt_id": str(row.get("dispatch_receipt_id") or ""),
"target": target,
"target_surface": str(row.get("target_surface") or ""),
"consumer_surface": _CONSUMER_SURFACES.get(target, "unknown_consumer_surface"),
},
"source_of_truth_diff": {
"current_state": "metadata_dispatch_receipt_in_ledger",
"desired_state": "metadata_receipt_available_to_consumer_context",
"delta_kind": f"{target}_consumer_context_binding",
"raw_payload_included": raw_payload_included,
},
"check_mode": {
"enabled": True,
"checks": [
"dispatch_receipt_id_present",
"target_surface_present",
"metadata_only_raw_payload_absent",
"post_apply_verifier_refs_present",
"ledger_status_success",
],
},
"rollback": {
"required": True,
"rollback_ref": (
"rollback://ai-agent-log-controlled-writeback-consumer/"
f"{row.get('dispatch_receipt_id') or row.get('op_id')}"
),
"strategy": "ignore_consumer_binding_and_mark_receipt_superseded",
},
"post_apply_verifier": {
"required": True,
"verifier_refs": verifier_refs,
},
"target_write_performed": False,
}
def _binding_ready(row: Mapping[str, Any], raw_payload_included: bool) -> bool:
if str(row.get("semantic_operation_type") or "") != OPERATION_TYPE:
return False
if str(row.get("actor") or "") != EXECUTOR_ROUTE:
return False
if str(row.get("status") or "") != "success":
return False
if str(row.get("target") or "") not in _TARGETS:
return False
if not str(row.get("dispatch_receipt_id") or ""):
return False
if not str(row.get("target_surface") or ""):
return False
if raw_payload_included:
return False
return bool(_strings(row.get("post_apply_verifier_refs")))
def _active_blockers(bindings: list[dict[str, Any]]) -> list[str]:
blockers: list[str] = []
if not bindings:
blockers.append("log_controlled_writeback_dispatch_receipts_missing")
missing_targets = [target for target in _TARGETS if _target_count(bindings, target) == 0]
for target in missing_targets:
blockers.append(f"{target}_consumer_binding_missing")
for binding in bindings:
if binding["status"] != "ready_for_consumer_context":
blockers.append(f"{binding['target'] or 'unknown'}_consumer_binding_not_ready")
if binding["raw_payload_included"] is not False:
blockers.append(f"{binding['target'] or 'unknown'}_raw_payload_included")
return _unique(blockers)
def _target_rollups(bindings: list[dict[str, Any]]) -> list[dict[str, Any]]:
return [
{
"target": target,
"consumer_surface": _CONSUMER_SURFACES[target],
"binding_count": _target_count(bindings, target),
"ready_binding_count": sum(
1
for binding in bindings
if binding["target"] == target
and binding["status"] == "ready_for_consumer_context"
),
"metadata_only": all(
binding["raw_payload_included"] is False
for binding in bindings
if binding["target"] == target
),
}
for target in _TARGETS
]
def _target_count(bindings: list[dict[str, Any]], target: str) -> int:
return sum(1 for binding in bindings if binding["target"] == target)
def _result_rows(result: Any) -> list[dict[str, Any]]:
mappings = getattr(result, "mappings", None)
if callable(mappings):
return [dict(row) for row in mappings().all()]
all_rows = getattr(result, "all", None)
if callable(all_rows):
return [_row_mapping(row) for row in all_rows()]
return [_row_mapping(row) for row in result]
def _row_mapping(row: Any) -> dict[str, Any]:
if isinstance(row, Mapping):
return dict(row)
mapping = getattr(row, "_mapping", None)
if mapping is not None:
return dict(mapping)
return dict(row)
def _strings(value: Any) -> list[str]:
if isinstance(value, list):
return [str(item) for item in value]
if isinstance(value, tuple):
return [str(item) for item in value]
return []
def _unique(values: list[str]) -> list[str]:
seen = set()
result = []
for value in values:
if value in seen:
continue
seen.add(value)
result.append(value)
return result