diff --git a/scripts/src/platform-infra-observability.ts b/scripts/src/platform-infra-observability.ts index 7ed73cc0..5b60a01a 100644 --- a/scripts/src/platform-infra-observability.ts +++ b/scripts/src/platform-infra-observability.ts @@ -1000,6 +1000,7 @@ function renderDiagnoseCodeAgentTable(input: { "", "Summary:", ` target=${input.target.id} spanCount=${textValue(input.result.spanCount)} servicePath=${joinValues(input.result.servicePath, 60)}`, + ` agentrun=${textValue(agentrun?.terminalStatus)} source=${textValue(agentrun?.terminalSource)} authority=${shortenEnd(JSON.stringify((asPlainRecord(agentrun?.authority) ?? {})), 120)}`, ` readModel=${shortenEnd(JSON.stringify(input.result.hwlabReadModel ?? null), 140)}`, ]; const next = asPlainRecord(input.result.next); @@ -1369,7 +1370,7 @@ function compactDiagnoseCodeAgentResult(value: unknown): Record servicePath: source.servicePath ?? null, businessTraceIds: source.businessTraceIds ?? null, identity: source.identity ?? null, - agentrun: compactRecord(source.agentrun, ["terminalStatus", "latestSeq", "terminalEventType", "runnerProviderClassification"]), + agentrun: compactRecord(source.agentrun, ["terminalStatus", "terminalSource", "latestSeq", "terminalEventType", "runnerProviderClassification", "authority"]), hwlabReadModel: source.hwlabReadModel ?? null, http: source.http ?? null, projectionLag: source.projectionLag ?? null, @@ -2807,10 +2808,11 @@ function diagnoseCodeAgentScript(observability: ObservabilityConfig, target: Obs return ` set -u python3 - <<'PY' -import collections, json, subprocess, time +import collections, json, re, subprocess, time, urllib.parse BUSINESS_TRACE_ID = ${businessTraceIdLiteral} TRACE_ID = ${traceIdLiteral} +TARGET_ID = ${JSON.stringify(target.id)} SEARCH_PATH = ${searchPathLiteral} SEARCH_PROXY_PATH = ${searchProxyPathLiteral} TRACE_PROXY_PREFIX = ${JSON.stringify(proxyPrefix)} @@ -2895,6 +2897,195 @@ def run_kubectl(proxy_path, timeout=15): except Exception as exc: return 125, "", str(exc) +def run_kubectl_json(proxy_path, timeout=15): + rc, body, stderr = run_kubectl(proxy_path, timeout=timeout) + parsed = parse_json(body) + return { + "ok": rc == 0 and isinstance(parsed, dict), + "rc": rc, + "json": parsed if isinstance(parsed, dict) else None, + "stderrTail": (stderr or "")[-1000:], + "bodyBytes": len((body or "").encode("utf-8")), + } + +def first_present(*values): + for value in values: + if value not in (None, ""): + return value + return None + +def shell_quote(value): + return "'" + str(value).replace("'", "'\\''") + "'" + +def unwrap_payload(value): + if not isinstance(value, dict): + return {} + data = value.get("data") + if isinstance(data, dict): + return data + result = value.get("result") + if isinstance(result, dict): + return result + return value + +def guess_agentrun_namespace(identity): + if isinstance(identity, dict): + for key in ("jobName", "podName", "runnerJobId"): + text = str(identity.get(key) or "") + match = re.search(r"\\bagentrun-v[0-9]+\\b", text) + if match: + return match.group(0) + if str(TARGET_ID).upper() == "D601": + return "agentrun-v02" + return "agentrun-v01" + +def agentrun_service_proxy_paths(namespace, api_path): + return [ + "/api/v1/namespaces/%s/services/http:agentrun-mgr:http/proxy%s" % (namespace, api_path), + "/api/v1/namespaces/%s/services/http:agentrun-mgr:8080/proxy%s" % (namespace, api_path), + ] + +def run_agentrun_mgr_json(namespace, api_path, timeout=12): + url = "http://127.0.0.1:8080%s" % api_path + script_parts = [] + script_parts.append("url=" + shell_quote(url) + "; ") + script_parts.append("api_key=\\\"\${AGENTRUN_API_KEY:-}\\\"; ") + script_parts.append("if [ -z \\\"$api_key\\\" ]; then api_key=\\\"\${HWLAB_API_KEY:-}\\\"; fi; ") + script_parts.append("if command -v wget >/dev/null 2>&1; then ") + script_parts.append("wget -qO- --header=\\\"Authorization: Bearer $api_key\\\" \\\"$url\\\"; ") + script_parts.append("else ") + script_parts.append("curl -fsS -H \\\"Authorization: Bearer $api_key\\\" \\\"$url\\\"; ") + script_parts.append("fi") + script = "".join(script_parts) + try: + proc = subprocess.run( + ["kubectl", "-n", namespace, "exec", "deploy/agentrun-mgr", "--", "sh", "-lc", script], + capture_output=True, + text=True, + timeout=timeout, + ) + parsed = parse_json(proc.stdout) + return { + "ok": proc.returncode == 0 and isinstance(parsed, dict), + "rc": proc.returncode, + "json": parsed if isinstance(parsed, dict) else None, + "stderrTail": (proc.stderr or "")[-1000:], + "bodyBytes": len((proc.stdout or "").encode("utf-8")), + "method": "manager-exec-localhost", + } + except subprocess.TimeoutExpired as exc: + return {"ok": False, "rc": 124, "json": None, "stderrTail": "timeout while querying agentrun manager", "bodyBytes": len((exc.stdout or "").encode("utf-8")), "method": "manager-exec-localhost"} + except Exception as exc: + return {"ok": False, "rc": 125, "json": None, "stderrTail": str(exc), "bodyBytes": 0, "method": "manager-exec-localhost"} + +def get_agentrun_json(namespace, api_path, timeout=10): + attempts = [] + exec_response = run_agentrun_mgr_json(namespace, api_path, timeout=timeout) + attempts.append(exec_response) + if exec_response.get("ok"): + return exec_response, attempts + for proxy_path in agentrun_service_proxy_paths(namespace, api_path): + response = run_kubectl_json(proxy_path, timeout=timeout) + response["proxyPath"] = proxy_path + response["method"] = "service-proxy" + attempts.append(response) + if response.get("ok"): + return response, attempts + return attempts[-1] if attempts else {"ok": False, "json": None}, attempts + +def event_items(value): + if not isinstance(value, dict): + return [] + for key in ("items", "events", "data"): + items = value.get(key) + if isinstance(items, list): + return [item for item in items if isinstance(item, dict)] + return [] + +def terminal_from_events(events): + latest = None + for item in events: + payload = item.get("payload") if isinstance(item.get("payload"), dict) else {} + event_type = item.get("type") or item.get("eventType") or payload.get("type") + status = first_present(payload.get("terminalStatus"), payload.get("status"), item.get("terminalStatus"), item.get("status")) + if event_type == "terminal_status" or status in ("completed", "failed", "blocked", "cancelled", "canceled"): + latest = { + "seq": item.get("seq"), + "type": event_type, + "status": status, + "failureKind": first_present(payload.get("failureKind"), item.get("failureKind")), + } + return latest + +def classify_terminal_status(status, spans): + if status in ("completed", "succeeded", "success"): + return "completed" + if status in ("failed", "error", "timeout", "cancelled", "canceled"): + return "failed" + if status == "blocked": + return "blocked" + if any(str(item.get("service") or "") == "agentrun-runner" for item in spans): + return "running" + return "unknown" + +def agentrun_authority_summary(identity): + run_id = identity.get("runId") if isinstance(identity, dict) else None + command_id = identity.get("commandId") if isinstance(identity, dict) else None + if not run_id or not command_id: + return { + "queried": False, + "reason": "missing-run-or-command-id", + } + namespace = guess_agentrun_namespace(identity) + run_q = urllib.parse.quote(str(run_id), safe="") + command_q = urllib.parse.quote(str(command_id), safe="") + command_response, command_attempts = get_agentrun_json(namespace, "/api/v1/runs/%s/commands/%s" % (run_q, command_q), timeout=8) + result_response, result_attempts = get_agentrun_json(namespace, "/api/v1/runs/%s/commands/%s/result" % (run_q, command_q), timeout=12) + events_response, events_attempts = get_agentrun_json(namespace, "/api/v1/runs/%s/events?afterSeq=0&limit=1000" % run_q, timeout=12) + command_payload = unwrap_payload(command_response.get("json")) + result_payload = unwrap_payload(result_response.get("json")) + events = event_items(events_response.get("json")) + terminal_event = terminal_from_events(events) + command_state = first_present(command_payload.get("state"), command_payload.get("status")) + result_status = first_present( + result_payload.get("terminalStatus"), + result_payload.get("terminal_status"), + result_payload.get("state"), + result_payload.get("status"), + ) + terminal_status = first_present( + terminal_event.get("status") if isinstance(terminal_event, dict) else None, + result_status, + command_state if command_state in ("completed", "failed", "blocked", "cancelled", "canceled") else None, + ) + return { + "queried": True, + "namespace": namespace, + "ok": bool(command_response.get("ok") or result_response.get("ok") or events_response.get("ok")), + "commandOk": command_response.get("ok"), + "resultOk": result_response.get("ok"), + "eventsOk": events_response.get("ok"), + "commandState": command_state, + "terminalStatus": terminal_status, + "resultTerminalStatus": result_status, + "resultOkValue": result_payload.get("ok"), + "authority": result_payload.get("authority"), + "fallback": result_payload.get("fallback"), + "failureKind": first_present( + result_payload.get("failureKind"), + result_payload.get("terminalFailureKind"), + terminal_event.get("failureKind") if isinstance(terminal_event, dict) else None, + ), + "terminalEventType": terminal_event.get("type") if isinstance(terminal_event, dict) else None, + "terminalEventSeq": terminal_event.get("seq") if isinstance(terminal_event, dict) else None, + "eventCount": len(events), + "attempts": { + "command": [{"ok": item.get("ok"), "rc": item.get("rc"), "method": item.get("method"), "proxyPath": item.get("proxyPath"), "stderrTail": item.get("stderrTail")} for item in command_attempts], + "result": [{"ok": item.get("ok"), "rc": item.get("rc"), "method": item.get("method"), "proxyPath": item.get("proxyPath"), "stderrTail": item.get("stderrTail")} for item in result_attempts], + "events": [{"ok": item.get("ok"), "rc": item.get("rc"), "method": item.get("method"), "proxyPath": item.get("proxyPath"), "stderrTail": item.get("stderrTail")} for item in events_attempts], + }, + } + def parse_json(body): try: return json.loads(body) if body else None @@ -3234,7 +3425,7 @@ def http_status_summary(spans): "actorForbidden": any(to_int((item.get("attributes") or {}).get("http.status_code")) == 403 and item.get("service") == "hwlab-cloud-api" for item in problem_spans), } -def agentrun_summary(spans): +def agentrun_summary(spans, authority=None): terminal_spans = [] latest_seq = None failure_kind = None @@ -3274,16 +3465,13 @@ def agentrun_summary(spans): if terminal.get("eventType"): terminal_event_type = terminal.get("eventType") break - if terminal_status in ("completed", "succeeded", "success"): - runner_classification = "completed" - elif terminal_status in ("failed", "error", "timeout", "cancelled"): - runner_classification = "failed" - elif terminal_status in ("blocked",): - runner_classification = "blocked" - elif any(str(item.get("service") or "") == "agentrun-runner" for item in spans): - runner_classification = "running" - else: - runner_classification = "unknown" + terminal_source = "otel-span" if terminal_status not in (None, "") else None + if isinstance(authority, dict) and authority.get("terminalStatus") not in (None, ""): + terminal_status = authority.get("terminalStatus") + terminal_event_type = authority.get("terminalEventType") or terminal_event_type + failure_kind = authority.get("failureKind") or failure_kind + terminal_source = "agentrun-authority" + runner_classification = classify_terminal_status(terminal_status, spans) return { "terminalStatus": terminal_status, "failureKind": failure_kind, @@ -3291,6 +3479,8 @@ def agentrun_summary(spans): "terminalCategory": terminal_category, "latestSeq": latest_seq, "terminalEventType": terminal_event_type, + "terminalSource": terminal_source, + "authority": authority if isinstance(authority, dict) else None, "terminalSpans": terminal_spans, "runnerProviderClassification": runner_classification, } @@ -3692,10 +3882,11 @@ business_trace_ids = flat["businessTraceIds"] error_spans = flat["errorSpans"] idle_warning_spans = [item for item in spans if str(item.get("name") or "") == "codex_stdio.idle_warning"] http_summary = http_status_summary(spans) -agentrun = agentrun_summary(spans) read_model = hwlab_read_model_summary(spans) -lag = projection_lag_summary(agentrun, read_model) identity = identity_from_spans(spans) +agentrun_authority = agentrun_authority_summary(identity) +agentrun = agentrun_summary(spans, agentrun_authority) +lag = projection_lag_summary(agentrun, read_model) candidates = root_cause_candidates(http_summary, agentrun, read_model, lag, error_spans, idle_warning_spans) expected_services = ["hwlab-cloud-api", "agentrun-manager", "agentrun-runner"] service_path = { @@ -3761,9 +3952,11 @@ payload = { "identity": identity, "agentrun": { "terminalStatus": agentrun.get("terminalStatus"), + "terminalSource": agentrun.get("terminalSource"), "latestSeq": agentrun.get("latestSeq"), "terminalEventType": agentrun.get("terminalEventType"), "runnerProviderClassification": agentrun.get("runnerProviderClassification"), + "authority": agentrun.get("authority"), }, "hwlabReadModel": { "sourceEventCount": read_model.get("sourceEventCount"),