fix(agent): advance program ledger from runtime truth
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 1s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 1m21s
CD Pipeline / build-and-deploy (push) Has started running
CD Pipeline / post-deploy-checks (push) Has been cancelled

This commit is contained in:
ogt
2026-07-10 18:28:02 +08:00
parent da1c836c0b
commit e5417e383f
3 changed files with 141 additions and 7 deletions

View File

@@ -6923,6 +6923,44 @@ def _apply_ai_automation_program_ledger(payload: dict[str, Any]) -> None:
)
work_items = copy.deepcopy(_AI_AUTOMATION_PROGRAM_WORK_ITEMS)
strict_runtime_closed = (
summary.get("ai_agent_autonomous_runtime_strict_closed") is True
)
if strict_runtime_closed:
runtime_work_item = next(
item for item in work_items if item["id"] == "AIA-P0-001"
)
runtime_work_item["status"] = "done"
runtime_work_item["completion_evidence"] = {
"source": "agent-autonomous-runtime-control",
"generated_at": str(
summary.get("ai_agent_autonomous_runtime_generated_at") or ""
),
"automation_run_id": str(
summary.get("ai_agent_autonomous_runtime_automation_run_id") or ""
),
"completion_percent": _int(
summary.get("ai_agent_autonomous_runtime_completion_percent")
),
"required_stage_count": _int(
summary.get("ai_agent_autonomous_runtime_required_stage_count")
),
"present_stage_count": _int(
summary.get("ai_agent_autonomous_runtime_present_stage_count")
),
"same_run_correlation": (
summary.get("ai_agent_autonomous_runtime_same_run_correlation")
is True
),
"latest_flow_closed": (
summary.get("ai_agent_autonomous_runtime_latest_flow_closed")
is True
),
"latest_loop_closed": (
summary.get("ai_agent_autonomous_runtime_latest_loop_closed")
is True
),
}
status_counts = {
status: sum(1 for item in work_items if item["status"] == status)
for status in ("done", "in_progress", "pending", "blocked", "deferred")
@@ -6957,7 +6995,11 @@ def _apply_ai_automation_program_ledger(payload: dict[str, Any]) -> None:
program = {
"schema_version": _AI_AUTOMATION_PROGRAM_SCHEMA_VERSION,
"status": "runtime_closure_in_progress",
"status": (
"p0_001_closed_next_priority_active"
if strict_runtime_closed
else "runtime_closure_in_progress"
),
"authoritative_for_execution_order": True,
"program_goal": "讓所有受管主機、服務、產品、網站、工具、套件、資料與可觀測表面都由專業 AI Agent 完成可驗證、可回滾、可學習的自主閉環。",
"source": _AI_AUTOMATION_PROGRAM_SOURCE,
@@ -7157,12 +7199,55 @@ def apply_ai_automation_live_closure_readbacks(
verifier_readback: dict[str, Any] | None = None,
consumer_readback: dict[str, Any] | None = None,
telegram_alert_context_verifier_readback: dict[str, Any] | None = None,
autonomous_runtime_control: dict[str, Any] | None = None,
) -> None:
"""Overlay live AI-loop executor/verifier/consumer readbacks onto receipts."""
summary = _dict(payload.setdefault("summary", {}))
rollups = _dict(payload.setdefault("rollups", {}))
state = _dict(payload.setdefault("mainline_execution_state", {}))
runtime_control = _dict(autonomous_runtime_control)
strict_runtime = _dict(runtime_control.get("strict_runtime_completion"))
strict_runtime_closed = bool(
strict_runtime.get("closed") is True
and strict_runtime.get("same_run_correlation") is True
and strict_runtime.get("latest_flow_closed") is True
and strict_runtime.get("latest_loop_closed") is True
and _int(strict_runtime.get("present_stage_count"))
== _int(strict_runtime.get("required_stage_count"))
and _int(strict_runtime.get("required_stage_count")) > 0
)
if runtime_control:
summary["ai_agent_autonomous_runtime_generated_at"] = str(
runtime_control.get("generated_at") or ""
)
summary["ai_agent_autonomous_runtime_automation_run_id"] = str(
strict_runtime.get("automation_run_id") or ""
)
summary["ai_agent_autonomous_runtime_completion_percent"] = _int(
strict_runtime.get("completion_percent")
)
summary["ai_agent_autonomous_runtime_required_stage_count"] = _int(
strict_runtime.get("required_stage_count")
)
summary["ai_agent_autonomous_runtime_present_stage_count"] = _int(
strict_runtime.get("present_stage_count")
)
summary["ai_agent_autonomous_runtime_same_run_correlation"] = (
strict_runtime.get("same_run_correlation") is True
)
summary["ai_agent_autonomous_runtime_latest_flow_closed"] = (
strict_runtime.get("latest_flow_closed") is True
)
summary["ai_agent_autonomous_runtime_latest_loop_closed"] = (
strict_runtime.get("latest_loop_closed") is True
)
summary["ai_agent_autonomous_runtime_strict_closed"] = (
strict_runtime_closed
)
rollups["ai_agent_autonomous_runtime_strict_closed_count"] = (
1 if strict_runtime_closed else 0
)
executor = _dict(executor_readback)
executor_rollups = _dict(executor.get("rollups"))