fix(agent): close no-write alert replay receipts
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 2m39s
CD Pipeline / build-and-deploy (push) Failing after 6m12s
CD Pipeline / post-deploy-checks (push) Has been skipped

This commit is contained in:
ogt
2026-07-14 13:18:00 +08:00
parent bff1421a1f
commit 5aa651563c
13 changed files with 2088 additions and 38 deletions

View File

@@ -7235,6 +7235,10 @@ async def build_ai_agent_autonomous_runtime_control_with_live_readback(
)
_attach_runtime_receipt_readback(payload, readback)
payload["executor_trust_boundary"] = boundary_readback
_attach_effective_executor_runtime_switches(
payload,
boundary_readback=boundary_readback,
)
_attach_autonomous_single_writer_readback(
payload,
readback=readback,
@@ -7257,6 +7261,44 @@ async def build_ai_agent_autonomous_runtime_control_with_live_readback(
return payload
def _attach_effective_executor_runtime_switches(
payload: dict[str, Any],
*,
boundary_readback: Mapping[str, Any],
) -> None:
"""Separate the API process flag from the dedicated broker truth."""
runtime_switches = payload.setdefault("runtime_switches", {})
controls = boundary_readback.get("controls")
if not isinstance(controls, Mapping):
controls = {}
dedicated_broker_ready = bool(
boundary_readback.get("status") == "verified_ready"
and boundary_readback.get("production_boundary_verified") is True
and boundary_readback.get("source_sha_matches_deployment") is True
and controls.get("broker_ssh_mount_present") is True
and controls.get("broker_ssh_egress_allowlisted") is True
and controls.get("broker_kubernetes_token_absent") is True
)
runtime_switches.update(
{
"ansible_check_mode_worker_scope": "api_process_only",
"dedicated_execution_broker_ready": dedicated_broker_ready,
"effective_ansible_executor_available": dedicated_broker_ready,
"effective_executor_readback_source": (
boundary_readback.get("receipt_ref")
if dedicated_broker_ready
else None
),
"effective_executor_verified_source_sha": (
boundary_readback.get("verified_source_sha")
if dedicated_broker_ready
else None
),
}
)
def _attach_autonomous_single_writer_readback(
payload: dict[str, Any],
*,