29072e4004
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 2m8s
CD Pipeline / build-and-deploy (push) Successful in 13m33s
CD Pipeline / post-deploy-checks (push) Has been cancelled
396 lines
15 KiB
Python
396 lines
15 KiB
Python
"""Fail-closed, no-write Agent99 same-run Status reconciliation.
|
|
|
|
This module deliberately keeps the production source verifier separate from
|
|
the Agent99 verifier. A missing historical Recover outcome may be refreshed
|
|
only after the live cold-start source reports a fresh, unblocked state with no
|
|
firing ColdStart alert. The refresh keeps the existing deterministic
|
|
dispatch identity and asks the authenticated relay for ``Status`` in
|
|
``reconcileOnly`` mode; it never reserves a new run, advances a generation,
|
|
or authorizes controlled apply.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import math
|
|
import re
|
|
import time
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
from typing import Any
|
|
|
|
from src.core.config import settings
|
|
from src.core.logging import get_logger
|
|
from src.services.agent99_controlled_dispatch_ledger import (
|
|
Agent99DispatchIdentity,
|
|
)
|
|
from src.services.agent99_public_receipts import (
|
|
normalize_agent99_public_receipt_ref,
|
|
)
|
|
|
|
logger = get_logger("awoooi.agent99_same_run_reconcile")
|
|
|
|
AGENT99_SAME_RUN_RECONCILE_SCHEMA = "agent99_same_run_status_reconcile_v1"
|
|
AGENT99_COLD_START_ROUTE_ID = "agent99:host_recovery:Recover"
|
|
AGENT99_COLD_START_INCIDENT_ID = "INC-20260711-11C751"
|
|
AGENT99_COLD_START_RUN_ID = "bf30ca01-1080-5725-b2d1-3e1534dfc811"
|
|
AGENT99_SAME_RUN_STATUS_VERIFIER = (
|
|
"agent99-status-same-run-no-write-post-condition-v1"
|
|
)
|
|
AGENT99_COLD_START_METRIC_HOST = "110"
|
|
AGENT99_COLD_START_METRIC_SCOPE = "110_112_120_121_188"
|
|
AGENT99_COLD_START_MAX_AGE_SECONDS = 900
|
|
|
|
_COLD_START_QUERIES = {
|
|
"monitor_up": (
|
|
f'awoooi_cold_start_monitor_up{{host="110",scope="{AGENT99_COLD_START_METRIC_SCOPE}",'
|
|
'mode="read_only"}'
|
|
),
|
|
"pass_gates": (
|
|
f'awoooi_cold_start_pass_gates{{host="110",scope="{AGENT99_COLD_START_METRIC_SCOPE}"}}'
|
|
),
|
|
"warn_gates": (
|
|
f'awoooi_cold_start_warn_gates{{host="110",scope="{AGENT99_COLD_START_METRIC_SCOPE}"}}'
|
|
),
|
|
"blocked_gates": (
|
|
f'awoooi_cold_start_blocked_gates{{host="110",scope="{AGENT99_COLD_START_METRIC_SCOPE}"}}'
|
|
),
|
|
"last_run_timestamp": (
|
|
f'awoooi_cold_start_last_run_timestamp{{host="110",'
|
|
f'scope="{AGENT99_COLD_START_METRIC_SCOPE}"}}'
|
|
),
|
|
"green_result": (
|
|
f'awoooi_cold_start_last_result{{host="110",scope="{AGENT99_COLD_START_METRIC_SCOPE}",'
|
|
'result="green"}'
|
|
),
|
|
"degraded_result": (
|
|
f'awoooi_cold_start_last_result{{host="110",scope="{AGENT99_COLD_START_METRIC_SCOPE}",'
|
|
'result="degraded"}'
|
|
),
|
|
"blocked_result": (
|
|
f'awoooi_cold_start_last_result{{host="110",scope="{AGENT99_COLD_START_METRIC_SCOPE}",'
|
|
'result="blocked"}'
|
|
),
|
|
"check_failed_result": (
|
|
f'awoooi_cold_start_last_result{{host="110",scope="{AGENT99_COLD_START_METRIC_SCOPE}",'
|
|
'result="check_failed"}'
|
|
),
|
|
"firing_blocking_alerts": (
|
|
'sum(ALERTS{alertname=~"ColdStart.*",alertstate="firing",'
|
|
'severity="critical"}) or vector(0)'
|
|
),
|
|
}
|
|
|
|
|
|
def is_cold_start_same_run_identity(
|
|
identity: Agent99DispatchIdentity,
|
|
receipt: dict[str, Any] | None = None,
|
|
) -> bool:
|
|
"""Admit only a canonical cold-start Recover dispatch to no-write Status."""
|
|
|
|
legacy_identity = bool(
|
|
identity.route_id == AGENT99_COLD_START_ROUTE_ID
|
|
and identity.incident_id == AGENT99_COLD_START_INCIDENT_ID
|
|
and str(identity.run_id) == AGENT99_COLD_START_RUN_ID
|
|
and identity.source_fingerprint.startswith("legacy-cold-start:")
|
|
and identity.execution_generation == "1"
|
|
)
|
|
if legacy_identity:
|
|
return True
|
|
|
|
scope = (
|
|
receipt.get("dispatch_scope")
|
|
if isinstance(receipt, dict)
|
|
and isinstance(receipt.get("dispatch_scope"), dict)
|
|
else {}
|
|
)
|
|
try:
|
|
generation = int(identity.execution_generation)
|
|
except ValueError:
|
|
return False
|
|
expected_work_item_id = (
|
|
f"agent99-dispatch:awoooi:{identity.incident_id}:"
|
|
f"{AGENT99_COLD_START_ROUTE_ID}"
|
|
)
|
|
return bool(
|
|
identity.project_id == "awoooi"
|
|
and identity.route_id == AGENT99_COLD_START_ROUTE_ID
|
|
and re.fullmatch(r"INC-[0-9]{8}-[0-9A-F]{6}", identity.incident_id)
|
|
is not None
|
|
and identity.work_item_id == expected_work_item_id
|
|
and re.fullmatch(r"[0-9a-f]{32,64}", identity.source_fingerprint)
|
|
is not None
|
|
and 1 <= generation <= 3
|
|
and scope.get("schema_version") == "agent99_dispatch_scope_v1"
|
|
and scope.get("kind") == "host_recovery"
|
|
and scope.get("suggested_mode") == "Recover"
|
|
and scope.get("target_resource") == "cold-start-gate"
|
|
and scope.get("controlled_apply_requested") is True
|
|
)
|
|
|
|
|
|
def agent99_same_run_evidence_id(identity: Agent99DispatchIdentity) -> str:
|
|
"""Return the only evidence identifier accepted for this bounded run."""
|
|
|
|
digest = identity.public_dict()["canonical_digest"]
|
|
return f"same-run-status-{identity.run_id}-{digest[:12]}"
|
|
|
|
|
|
def _query_prometheus_scalar(
|
|
query_url: str,
|
|
query: str,
|
|
*,
|
|
timeout_seconds: float,
|
|
) -> float | None:
|
|
target = f"{query_url}?{urllib.parse.urlencode({'query': query})}"
|
|
request = urllib.request.Request(
|
|
target,
|
|
headers={
|
|
"Accept": "application/json",
|
|
"User-Agent": "awoooi-agent99-source-verifier/1",
|
|
},
|
|
method="GET",
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=timeout_seconds) as response:
|
|
if int(response.getcode()) != 200:
|
|
return None
|
|
body = response.read(65536).decode("utf-8", errors="replace")
|
|
except (urllib.error.HTTPError, urllib.error.URLError, TimeoutError):
|
|
return None
|
|
try:
|
|
payload = json.loads(body)
|
|
data = payload.get("data") if isinstance(payload, dict) else None
|
|
result = data.get("result") if isinstance(data, dict) else None
|
|
if payload.get("status") != "success" or not isinstance(result, list):
|
|
return None
|
|
if len(result) != 1 or not isinstance(result[0], dict):
|
|
return None
|
|
sample = result[0].get("value")
|
|
if not isinstance(sample, list) or len(sample) != 2:
|
|
return None
|
|
value = float(sample[1])
|
|
return value if math.isfinite(value) else None
|
|
except (AttributeError, TypeError, ValueError, json.JSONDecodeError):
|
|
return None
|
|
|
|
|
|
def read_cold_start_source_resolution(
|
|
*,
|
|
prometheus_url: str | None = None,
|
|
timeout_seconds: float = 3.0,
|
|
now_epoch: float | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Read a bounded production source receipt; missing evidence fails closed."""
|
|
|
|
base_url = str(prometheus_url or settings.PROMETHEUS_URL).rstrip("/")
|
|
query_url = f"{base_url}/api/v1/query"
|
|
values: dict[str, float] = {}
|
|
missing: list[str] = []
|
|
for name, query in _COLD_START_QUERIES.items():
|
|
value = _query_prometheus_scalar(
|
|
query_url,
|
|
query,
|
|
timeout_seconds=timeout_seconds,
|
|
)
|
|
if value is None:
|
|
missing.append(name)
|
|
else:
|
|
values[name] = value
|
|
|
|
blockers: list[str] = []
|
|
if missing:
|
|
blockers.append("source_metrics_missing:" + ",".join(sorted(missing)))
|
|
observed_at = float(now_epoch if now_epoch is not None else time.time())
|
|
last_run = values.get("last_run_timestamp")
|
|
age_seconds: float | None = None
|
|
if last_run is not None:
|
|
age_seconds = observed_at - last_run
|
|
if age_seconds < -60:
|
|
blockers.append("source_clock_skew_future")
|
|
elif age_seconds > AGENT99_COLD_START_MAX_AGE_SECONDS:
|
|
blockers.append("source_metrics_stale")
|
|
if values.get("monitor_up") != 1:
|
|
blockers.append("cold_start_monitor_not_up")
|
|
for count_name in ("pass_gates", "warn_gates", "blocked_gates"):
|
|
count_value = values.get(count_name)
|
|
if count_value is not None and (
|
|
count_value < 0 or not count_value.is_integer()
|
|
):
|
|
blockers.append(f"cold_start_{count_name}_invalid")
|
|
if values.get("pass_gates", 0) <= 0:
|
|
blockers.append("cold_start_pass_gates_missing")
|
|
if values.get("blocked_gates") != 0:
|
|
blockers.append("cold_start_blocked_gates_present")
|
|
if values.get("blocked_result") != 0:
|
|
blockers.append("cold_start_result_blocked")
|
|
if values.get("check_failed_result") != 0:
|
|
blockers.append("cold_start_result_check_failed")
|
|
warn_gates = values.get("warn_gates")
|
|
expected_result = "degraded" if warn_gates is not None and warn_gates > 0 else "green"
|
|
if expected_result == "degraded":
|
|
if values.get("degraded_result") != 1 or values.get("green_result") != 0:
|
|
blockers.append("cold_start_degraded_result_not_one_hot")
|
|
elif values.get("green_result") != 1 or values.get("degraded_result") != 0:
|
|
blockers.append("cold_start_green_result_not_one_hot")
|
|
if values.get("firing_blocking_alerts") != 0:
|
|
blockers.append("cold_start_blocking_alert_still_firing")
|
|
|
|
resolved = bool(not blockers and last_run is not None)
|
|
receipt_ref: str | None = None
|
|
if resolved:
|
|
receipt_ref = normalize_agent99_public_receipt_ref(
|
|
"prometheus:cold-start:"
|
|
f"{int(last_run)}:pass-{int(values['pass_gates'])}:"
|
|
f"warn-{int(values['warn_gates'])}:blocked-0:"
|
|
f"result-{expected_result}:blocking-alerts-0"
|
|
)
|
|
if receipt_ref is None:
|
|
blockers.append("source_receipt_ref_invalid")
|
|
resolved = False
|
|
|
|
return {
|
|
"schema_version": "agent99_cold_start_source_resolution_v1",
|
|
"status": "source_resolved" if resolved else "source_not_resolved",
|
|
"resolved": resolved,
|
|
"source_result": expected_result if resolved else "unresolved",
|
|
"degraded_provenance": bool(resolved and expected_result == "degraded"),
|
|
"receipt_ref": receipt_ref,
|
|
"observed_at_epoch": int(observed_at),
|
|
"age_seconds": round(age_seconds, 3) if age_seconds is not None else None,
|
|
"values": {
|
|
key: int(value) if float(value).is_integer() else value
|
|
for key, value in sorted(values.items())
|
|
},
|
|
"blockers": blockers,
|
|
"stores_raw_evidence": False,
|
|
}
|
|
|
|
|
|
def request_agent99_same_run_status_reconcile(
|
|
identity: Agent99DispatchIdentity,
|
|
source_receipt: dict[str, Any],
|
|
dispatch_receipt: dict[str, Any] | None = None,
|
|
*,
|
|
relay_url: str | None = None,
|
|
relay_token: str | None = None,
|
|
timeout_seconds: float | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Ask Agent99 for one idempotent, no-write Status on the existing run."""
|
|
|
|
if not is_cold_start_same_run_identity(identity, dispatch_receipt):
|
|
return {
|
|
"status": "identity_not_eligible_fail_closed",
|
|
"accepted": False,
|
|
"runtime_write_authorized": False,
|
|
}
|
|
receipt_ref = normalize_agent99_public_receipt_ref(
|
|
source_receipt.get("receipt_ref")
|
|
)
|
|
if source_receipt.get("resolved") is not True or not receipt_ref:
|
|
return {
|
|
"status": "source_not_resolved_fail_closed",
|
|
"accepted": False,
|
|
"runtime_write_authorized": False,
|
|
}
|
|
if not receipt_ref.startswith("prometheus:cold-start:"):
|
|
return {
|
|
"status": "source_receipt_not_cold_start_fail_closed",
|
|
"accepted": False,
|
|
"runtime_write_authorized": False,
|
|
}
|
|
|
|
target_url = str(
|
|
relay_url
|
|
if relay_url is not None
|
|
else settings.AGENT99_SRE_ALERT_RELAY_URL
|
|
)
|
|
token = str(
|
|
relay_token
|
|
if relay_token is not None
|
|
else settings.AGENT99_SRE_ALERT_RELAY_TOKEN
|
|
)
|
|
if not target_url or not token:
|
|
return {
|
|
"status": "relay_auth_not_configured_fail_closed",
|
|
"accepted": False,
|
|
"runtime_write_authorized": False,
|
|
}
|
|
timeout = float(
|
|
timeout_seconds
|
|
if timeout_seconds is not None
|
|
else settings.AGENT99_SRE_ALERT_RELAY_TIMEOUT_SECONDS
|
|
)
|
|
payload = {
|
|
"schemaVersion": AGENT99_SAME_RUN_RECONCILE_SCHEMA,
|
|
"action": "same_run_status_reconcile",
|
|
"identity": identity.public_dict(),
|
|
"mode": "Status",
|
|
"controlledApply": False,
|
|
"reconcileOnly": True,
|
|
"sourceEventResolved": True,
|
|
"sourceEventEvidence": receipt_ref,
|
|
"sourceEventResolutionPolicy": "external_source_receipt_required",
|
|
"suppressTelegram": True,
|
|
"evidenceReceiptId": agent99_same_run_evidence_id(identity),
|
|
}
|
|
request = urllib.request.Request(
|
|
target_url,
|
|
data=json.dumps(payload, ensure_ascii=False, sort_keys=True).encode("utf-8"),
|
|
headers={
|
|
"Accept": "application/json",
|
|
"Content-Type": "application/json; charset=utf-8",
|
|
"User-Agent": "awoooi-agent99-same-run-reconciler/1",
|
|
"X-Agent99-Relay-Token": token,
|
|
},
|
|
method="POST",
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=timeout) as response:
|
|
status_code = int(response.getcode())
|
|
body = response.read(65536).decode("utf-8", errors="replace")
|
|
except (urllib.error.HTTPError, urllib.error.URLError, TimeoutError) as exc:
|
|
logger.info(
|
|
"agent99_same_run_status_reconcile_pending",
|
|
run_id=str(identity.run_id),
|
|
error_type=type(exc).__name__,
|
|
)
|
|
return {
|
|
"status": "relay_request_failed_fail_closed",
|
|
"accepted": False,
|
|
"runtime_write_authorized": False,
|
|
}
|
|
try:
|
|
response_payload = json.loads(body)
|
|
except json.JSONDecodeError:
|
|
response_payload = {}
|
|
accepted = bool(
|
|
200 <= status_code < 300
|
|
and isinstance(response_payload, dict)
|
|
and response_payload.get("ok") is True
|
|
and response_payload.get("reconcileOnly") is True
|
|
and response_payload.get("controlledApply") is False
|
|
and str(response_payload.get("automationRunId") or "")
|
|
== str(identity.run_id)
|
|
and str(response_payload.get("evidenceReceiptId") or "")
|
|
== agent99_same_run_evidence_id(identity)
|
|
)
|
|
return {
|
|
"schema_version": "agent99_same_run_status_reconcile_request_receipt_v1",
|
|
"status": (
|
|
str(response_payload.get("status") or "accepted")
|
|
if accepted
|
|
else "relay_response_rejected_fail_closed"
|
|
),
|
|
"accepted": accepted,
|
|
"http_status": status_code,
|
|
"identity": identity.public_dict(),
|
|
"source_event_evidence_ref": receipt_ref,
|
|
"evidence_receipt_id": agent99_same_run_evidence_id(identity),
|
|
"reconcile_only": True,
|
|
"controlled_apply": False,
|
|
"runtime_write_authorized": False,
|
|
"stores_raw_response": False,
|
|
}
|