fix(otel): read AgentRun authority in code-agent diagnosis

2026-06-24 10:45:30 +00:00
parent 8e6fa7f816
commit 87cead4d38
1 changed files with 208 additions and 15 deletions
@@ -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<string, unknown>
    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"),