// SPEC: PJ2026-01060307 控制面模块化 draft-2026-06-25-p0. diagnose-code-agent-script module for scripts/src/platform-infra-observability.ts. // Moved mechanically from scripts/src/platform-infra-observability.ts:2137-3030 for #903. // SPEC: PJ2026-01060501 OTel追踪 draft-2026-06-19-p0. // Responsibility: YAML-first platform-infra OpenTelemetry tracing control commands. import { Buffer } from "node:buffer"; import { readFileSync } from "node:fs"; import type { UniDeskConfig } from "../config"; import { rootPath } from "../config"; import type { RenderedCliResult } from "../output"; import { compactCapture, createYamlFieldReader, numberField, parseJsonOutput, redactSensitiveUnknown, shQuote, capture, } from "../platform-infra-ops-library"; import type { DiagnoseCodeAgentOptions, ObservabilityConfig, ObservabilityTarget, SearchOptions, TraceOptions } from "./types"; import { inferSearchTempoQuery } from "./render"; export function traceScript(observability: ObservabilityConfig, target: ObservabilityTarget, tracePath: string, options: TraceOptions): string { const proxyPath = `/api/v1/namespaces/${target.namespace}/services/http:${observability.traceBackend.serviceName}:http/proxy${tracePath}`; const grepLiteral = options.grep === null ? "None" : JSON.stringify(options.grep); return ` set -u tmp="$(mktemp -d)" trap 'rm -rf "$tmp"' EXIT kubectl get --raw ${shQuote(proxyPath)} >"$tmp/trace.out" 2>"$tmp/trace.err" rc=$? python3 - "$rc" "$tmp/trace.out" "$tmp/trace.err" <<'PY' import collections, json, sys FULL = ${options.full ? "True" : "False"} RAW = ${options.raw ? "True" : "False"} GREP = ${grepLiteral} LIMIT = ${options.limit} IMPORTANT_ATTRS = [ "traceId", "otel.trace_id", "agentrun.stage", "runId", "commandId", "sessionId", "turnId", "threadId", "runnerJobId", "failureKind", "willRetry", "retryAttempt", "retryMax", "retryExhausted", "retryBackoffMs", "idleMs", "idleSeconds", "idleThresholdMs", "idleThresholdSeconds", "waitingFor", "lastEventLabel", "lastActivityAt", "lastActivityMs", "lastNotificationAt", "lastNotificationMs", "lastToolCallAt", "upstreamHttpStatus", "upstreamHost", "providerErrorClass", "errorSummary", "terminalStatus", "phase", "message", "http.route", "http.status_code", "http.response.status_code", "http.method", "http.request.method", "http.response.status_class", "workbench.ui.event_type", "workbench.ui.scope", "workbench.ui.state", "workbench.ui.reason", "workbench.ui.route", "workbench.ui.outcome", "workbench.ui.value_ms", "workbench.ui.visibility", "workbench.ui.event_name", "workbench.ui.activity_idle_ms", "workbench.ui.activity_waiting_for", "workbench.ui.activity_last_event_label", "workbench.ui.resource_duration_ms", "workbench.ui.resource_fetch_to_request_ms", "workbench.ui.resource_request_wait_ms", "workbench.ui.resource_response_transfer_ms", "workbench.ui.resource_transfer_size", "workbench.ui.resource_encoded_body_size", "workbench.ui.resource_decoded_body_size", "workbench.ui.resource_next_hop_protocol", "workbench.ui.resource_server_timing", "workbench.session_id", "workbench.trace_id", "workbench.turn_id", "workbench.read_model.route", "workbench.read_model.count", "workbench.read_model.family", "workbench.read_model.status", "workbench.read_model.reason", "workbench.projection.phase", "workbench.projection.session_id", "workbench.projection.turn_id", "workbench.projection.projected_seq", "workbench.projection.source_seq", "workbench.projection.source_event_id", "workbench.projection.event_type", "workbench.projection.message_id", "workbench.projection.terminal", "workbench.projection.suppressed_after_seal", "workbench.projection.terminal_trace_backfill_written", "source", "sessionCount", "fallbackTitleCount", "fallbackTitleRatio", "emptyPreviewCount", "includeSessionId", "sessionIds", "traceIds", "returnedMessages", "totalMessages", "roleSequencePrefix", "consecutiveUserPrefix", "adjacentSameRoleCount", "userCount", "agentCount", "db.system", "db.operation.name", "db.sql.table", "db.query.arg_count", "db.index.expected", "db.pool.max_open", "db.pool.open_connections", "db.pool.in_use", "db.pool.idle", "db.pool.wait_count", "db.pool.wait_duration_ms", "db.pool.max_idle_closed", "db.pool.max_lifetime_closed", "hwlab.runtime.stage", "hwlab.http.stage", "hwlab.http.phase", "hwlab.http.phase.outcome", "hwlab.live_builds.service_id", "hwlab.live_builds.service_kind", "hwlab.live_builds.external", "hwlab.live_builds.deploy_manifest_status", "hwlab.live_builds.artifact_catalog_status", "error.code", "error.category", "error.layer", "stage", "causeStage", "causeCode", "eventType", "runnerId", "attemptId", "backendProfile", "sourceCommit", "jobName", "podName", "logPath", "toolName", "type", "itemType", "itemId", "status", "exitCode", "durationMs", "cwd", "processId", "command", "commandFingerprint", "outputSummary", "outputBytes", "outputTruncated", "storageKind", "storagePvcName", "storagePvcPhase", "resumeMode", "rolloutId", "returnedEvents", "sinceSeq", "afterSeq", "limit", "fromSeq", "toSeq", "totalEvents", "hasMore", "fullTraceLoaded", "rawEventCount", "maxSeq", "traceLastSeq", "endSeq", "commandFiltered", "error", "error.message", "exception.type", "exception.message", "valuesPrinted", ] IMPORTANT_NAMES = { "durable_admission", "billing_preflight", "agentrun_dispatch", "projection_write", "trace_events_read", "turn_status_read", "run_created", "runner_job_created", "runner_job_status", "command_result", "projection_sync", } def text(path, limit=12000): try: return open(path, encoding="utf-8", errors="replace").read()[-limit:] except FileNotFoundError: return "" def read_all(path): try: return open(path, encoding="utf-8", errors="replace").read() except FileNotFoundError: return "" def attr_value(value): if not isinstance(value, dict): return value inner = value.get("value", value) if not isinstance(inner, dict): return inner for key in ("stringValue", "intValue", "doubleValue", "boolValue"): if key in inner: return inner.get(key) if "arrayValue" in inner: return "" if "kvlistValue" in inner: return "" if "bytesValue" in inner: return "" return inner def attrs_to_dict(attrs): if not isinstance(attrs, list): return {} output = {} for item in attrs: if not isinstance(item, dict): continue key = item.get("key") if isinstance(key, str): output[key] = attr_value(item.get("value")) return output def nanos_to_ms(start, end): try: return round((int(end) - int(start)) / 1000000, 3) except Exception: return None def span_status_code(status): if not isinstance(status, dict): return None return status.get("code") def selected_attrs(attrs): output = {} for key in IMPORTANT_ATTRS: if key in attrs: output[key] = attrs[key] return output def is_error_span(span, attrs): status = span.get("status") if isinstance(span, dict) else {} code = span_status_code(status) name = str(span.get("name", "")).lower() if isinstance(span, dict) else "" failure = str(attrs.get("failureKind", "")).strip() terminal = str(attrs.get("terminalStatus", "")).strip() return code in ("STATUS_CODE_ERROR", 2) or "error" in name or bool(failure) or terminal in ("failed", "blocked") def compact_span(span, service, resource_attrs, scope_name): attrs = attrs_to_dict(span.get("attributes")) status = span.get("status") if isinstance(span.get("status"), dict) else {} item = { "name": span.get("name"), "service": service, "scope": scope_name, "statusCode": span_status_code(status), "statusMessage": status.get("message"), "durationMs": nanos_to_ms(span.get("startTimeUnixNano"), span.get("endTimeUnixNano")), "attributes": selected_attrs(attrs), } if "deployment.environment" in resource_attrs: item["environment"] = resource_attrs.get("deployment.environment") if "git.commit" in resource_attrs: item["gitCommit"] = resource_attrs.get("git.commit") return item def batches_from_body(body): if not isinstance(body, dict): return [] for key in ("batches", "resourceSpans"): value = body.get(key) if isinstance(value, list): return value return [] def scope_spans_from_batch(batch): if not isinstance(batch, dict): return [] for key in ("scopeSpans", "instrumentationLibrarySpans"): value = batch.get(key) if isinstance(value, list): return value return [] def grep_matches(item): if GREP is None: return False return GREP.lower() in json.dumps(item, ensure_ascii=False, sort_keys=True).lower() def key_signature(item): attrs = item.get("attributes", {}) if isinstance(item.get("attributes"), dict) else {} return json.dumps({ "name": item.get("name"), "service": item.get("service"), "statusCode": item.get("statusCode"), "runId": attrs.get("runId"), "commandId": attrs.get("commandId"), "failureKind": attrs.get("failureKind"), "terminalStatus": attrs.get("terminalStatus"), "phase": attrs.get("phase"), "http.route": attrs.get("http.route"), }, ensure_ascii=False, sort_keys=True) body = read_all(sys.argv[2]) body_bytes = len(body.encode("utf-8")) try: parsed = json.loads(body) if body else None except Exception: parsed = body[-12000:] if not isinstance(parsed, dict): payload = { "ok": int(sys.argv[1]) == 0, "path": "${tracePath}", "proxyPath": "${proxyPath}", "bodyBytes": body_bytes, "parseOk": False, "bodyTail": text(sys.argv[2], 12000), "stderrTail": text(sys.argv[3], 4000), } print(json.dumps(payload, ensure_ascii=False, indent=2)) sys.exit(0 if payload["ok"] else 1) spans = [] services = set() business_trace_ids = set() identity_values = collections.defaultdict(set) name_counts = collections.Counter() error_spans = [] matched_spans = [] key_spans = [] seen_key_spans = set() for batch in batches_from_body(parsed): resource = batch.get("resource") if isinstance(batch.get("resource"), dict) else {} resource_attrs = attrs_to_dict(resource.get("attributes")) service = resource_attrs.get("service.name") or "" services.add(service) for scope_span in scope_spans_from_batch(batch): scope = scope_span.get("scope") if isinstance(scope_span.get("scope"), dict) else {} scope_name = scope.get("name") raw_spans = scope_span.get("spans") if isinstance(scope_span.get("spans"), list) else [] for span in raw_spans: if not isinstance(span, dict): continue item = compact_span(span, service, resource_attrs, scope_name) attrs = item.get("attributes", {}) if isinstance(attrs.get("traceId"), str): business_trace_ids.add(attrs.get("traceId")) if isinstance(attrs, dict): for key in ("runId", "commandId", "runnerJobId", "runnerId", "backendProfile", "sourceCommit", "jobName", "podName"): value = attrs.get(key) if value not in (None, ""): identity_values[key].add(str(value)) name = str(item.get("name") or "") name_counts[name] += 1 spans.append(item) error = is_error_span(span, attrs if isinstance(attrs, dict) else {}) matched = grep_matches(item) if error: error_spans.append(item) if matched: matched_spans.append(item) if error or name in IMPORTANT_NAMES or name.startswith("runner_error.") or name.startswith("runner_terminal.") or name.startswith("workbench.api_request."): signature = key_signature(item) if signature not in seen_key_spans: seen_key_spans.add(signature) key_spans.append(item) display_source = matched_spans if GREP is not None else key_spans payload = { "ok": int(sys.argv[1]) == 0, "path": "${tracePath}", "proxyPath": "${proxyPath}", "bodyBytes": body_bytes, "parseOk": True, "traceFound": len(spans) > 0, "spanCount": len(spans), "serviceCount": len(services), "services": sorted(services), "businessTraceIds": sorted(business_trace_ids)[:20], "identity": {key: sorted(values)[:8] for key, values in identity_values.items() if values}, "errorSpanCount": len(error_spans), "matchedSpanCount": len(matched_spans) if GREP is not None else None, "grep": GREP, "limit": LIMIT, "spanNameCounts": [{"name": name, "count": count} for name, count in name_counts.most_common(20)], "errorSpans": error_spans[:LIMIT], "spans": (spans if FULL and GREP is None else display_source)[:LIMIT], "truncated": { "errorSpans": len(error_spans) > LIMIT, "spans": (len(spans) if FULL and GREP is None else len(display_source)) > LIMIT, }, "next": { "fullSummary": "bun scripts/cli.ts platform-infra observability trace --target ${target.id} --trace-id ${options.traceId ?? ""} --full --limit 200", "grepProviderStream": "bun scripts/cli.ts platform-infra observability trace --target ${target.id} --trace-id ${options.traceId ?? ""} --grep provider-stream-disconnected --limit 40", "raw": "bun scripts/cli.ts platform-infra observability trace --target ${target.id} --trace-id ${options.traceId ?? ""} --raw", }, "stderrTail": text(sys.argv[3], 4000), } if RAW: payload["rawBody"] = parsed print(json.dumps(payload, ensure_ascii=False, indent=2)) sys.exit(0 if payload["ok"] else 1) PY `; } export function searchScript(observability: ObservabilityConfig, target: ObservabilityTarget, searchPath: string, options: SearchOptions): string { const proxyPrefix = `/api/v1/namespaces/${target.namespace}/services/http:${observability.traceBackend.serviceName}:http/proxy`; const searchProxyPath = `${proxyPrefix}${searchPath}`; const grepLiteral = options.grep === null ? "None" : JSON.stringify(options.grep); const effectiveQuery = options.query ?? inferSearchTempoQuery(options); const queryLiteral = effectiveQuery === null ? "None" : JSON.stringify(effectiveQuery); const pathLiteral = options.path === null ? "None" : JSON.stringify(options.path); const statusLiteral = options.status === null ? "None" : String(options.status); return ` set -u python3 - <<'PY' import collections, json, re, subprocess, time SEARCH_PROXY_PATH = ${JSON.stringify(searchProxyPath)} TRACE_PROXY_PREFIX = ${JSON.stringify(proxyPrefix)} TRACE_PATH_TEMPLATE = ${JSON.stringify(observability.probes.traceQueryPathTemplate)} FULL = ${options.full ? "True" : "False"} RAW = ${options.raw ? "True" : "False"} GREP = ${grepLiteral} QUERY = ${queryLiteral} PATH_FILTER = ${pathLiteral} STATUS_FILTER = ${statusLiteral} LIMIT = ${options.limit} CANDIDATE_LIMIT = ${options.candidateLimit} DEADLINE = time.time() + 50 BUSINESS_TRACE_GREP = bool(GREP is not None and re.search(r"\\btrc_[A-Za-z0-9_-]+\\b", GREP)) IMPORTANT_ATTRS = [ "traceId", "otel.trace_id", "agentrun.stage", "runId", "commandId", "sessionId", "turnId", "threadId", "runnerJobId", "failureKind", "willRetry", "retryAttempt", "retryMax", "retryExhausted", "retryBackoffMs", "idleMs", "idleSeconds", "idleThresholdMs", "idleThresholdSeconds", "waitingFor", "lastEventLabel", "lastActivityAt", "lastActivityMs", "lastNotificationAt", "lastNotificationMs", "lastToolCallAt", "upstreamHttpStatus", "upstreamHost", "providerErrorClass", "errorSummary", "terminalStatus", "phase", "message", "http.route", "http.status_code", "http.response.status_code", "http.method", "http.request.method", "http.response.status_class", "workbench.ui.event_type", "workbench.ui.scope", "workbench.ui.state", "workbench.ui.reason", "workbench.ui.route", "workbench.ui.outcome", "workbench.ui.value_ms", "workbench.ui.visibility", "workbench.ui.event_name", "workbench.ui.activity_idle_ms", "workbench.ui.activity_waiting_for", "workbench.ui.activity_last_event_label", "workbench.ui.resource_duration_ms", "workbench.ui.resource_fetch_to_request_ms", "workbench.ui.resource_request_wait_ms", "workbench.ui.resource_response_transfer_ms", "workbench.ui.resource_transfer_size", "workbench.ui.resource_encoded_body_size", "workbench.ui.resource_decoded_body_size", "workbench.ui.resource_next_hop_protocol", "workbench.ui.resource_server_timing", "workbench.session_id", "workbench.trace_id", "workbench.turn_id", "workbench.read_model.route", "workbench.read_model.count", "workbench.read_model.family", "workbench.read_model.status", "workbench.read_model.reason", "workbench.projection.phase", "workbench.projection.session_id", "workbench.projection.turn_id", "workbench.projection.projected_seq", "workbench.projection.source_seq", "workbench.projection.source_event_id", "workbench.projection.event_type", "workbench.projection.message_id", "workbench.projection.terminal", "workbench.projection.suppressed_after_seal", "workbench.projection.terminal_trace_backfill_written", "source", "sessionCount", "fallbackTitleCount", "fallbackTitleRatio", "emptyPreviewCount", "includeSessionId", "sessionIds", "traceIds", "returnedMessages", "totalMessages", "roleSequencePrefix", "consecutiveUserPrefix", "adjacentSameRoleCount", "userCount", "agentCount", "opencode.proxy.phase", "opencode.proxy.streaming", "opencode.proxy.ticket_accepted", "opencode.proxy.sse.directory_rewrite_enabled", "opencode.proxy.sse.directory_rewrite_from", "opencode.proxy.sse.directory_rewrite_to", "opencode.provider.sse.content_chunks", "opencode.provider.sse.content_chars", "opencode.provider.sse.output_data_lines", "opencode.provider.sse.done_lines", "opencode.provider.sse.json_errors", "opencode.provider.sse.reasoning_only_choices_dropped", "http.target", "http.url", "url.path", "db.system", "db.operation.name", "db.sql.table", "db.query.arg_count", "db.index.expected", "db.pool.max_open", "db.pool.open_connections", "db.pool.in_use", "db.pool.idle", "db.pool.wait_count", "db.pool.wait_duration_ms", "db.pool.max_idle_closed", "db.pool.max_lifetime_closed", "hwlab.runtime.stage", "hwlab.http.stage", "hwlab.http.phase", "hwlab.http.phase.outcome", "hwlab.live_builds.service_id", "hwlab.live_builds.service_kind", "hwlab.live_builds.external", "hwlab.live_builds.deploy_manifest_status", "hwlab.live_builds.artifact_catalog_status", "error.code", "error.category", "error.layer", "stage", "causeStage", "causeCode", "eventType", "runnerId", "attemptId", "backendProfile", "sourceCommit", "jobName", "podName", "logPath", "toolName", "type", "itemType", "itemId", "status", "exitCode", "durationMs", "cwd", "processId", "command", "commandFingerprint", "outputSummary", "outputBytes", "outputTruncated", "storageKind", "storagePvcName", "storagePvcPhase", "resumeMode", "rolloutId", "returnedEvents", "sinceSeq", "afterSeq", "limit", "fromSeq", "toSeq", "totalEvents", "hasMore", "fullTraceLoaded", "rawEventCount", "maxSeq", "traceLastSeq", "endSeq", "commandFiltered", "error", "error.message", "exception.type", "exception.message", "valuesPrinted", ] def run_kubectl(proxy_path, timeout=10): try: proc = subprocess.run( ["kubectl", "get", "--raw", proxy_path], capture_output=True, text=True, timeout=timeout, ) return proc.returncode, proc.stdout, proc.stderr except subprocess.TimeoutExpired as exc: return 124, exc.stdout or "", "timeout while querying tempo" except Exception as exc: return 125, "", str(exc) def parse_json(body): try: return json.loads(body) if body else None except Exception: return None def attr_value(value): if not isinstance(value, dict): return value inner = value.get("value", value) if not isinstance(inner, dict): return inner for key in ("stringValue", "intValue", "doubleValue", "boolValue"): if key in inner: return inner.get(key) if "arrayValue" in inner: return "" if "kvlistValue" in inner: return "" if "bytesValue" in inner: return "" return inner def attrs_to_dict(attrs): if not isinstance(attrs, list): return {} output = {} for item in attrs: if not isinstance(item, dict): continue key = item.get("key") if isinstance(key, str): output[key] = attr_value(item.get("value")) return output def selected_attrs(attrs): output = {} for key in IMPORTANT_ATTRS: if key in attrs: output[key] = attrs[key] return output def nanos_to_ms(start, end): try: return round((int(end) - int(start)) / 1000000, 3) except Exception: return None def span_status_code(status): if not isinstance(status, dict): return None return status.get("code") def to_int(value): if value is None: return None if isinstance(value, bool): return 1 if value else 0 if isinstance(value, int): return value if isinstance(value, float) and value.is_integer(): return int(value) if isinstance(value, str): try: return int(value.strip()) except Exception: return None return None def is_error_span(span, attrs): status = span.get("status") if isinstance(span, dict) else {} code = span_status_code(status) name = str(span.get("name", "")).lower() if isinstance(span, dict) else "" failure = str(attrs.get("failureKind", "")).strip() terminal = str(attrs.get("terminalStatus", "")).strip() http_status = to_int(attrs.get("http.response.status_code")) if http_status is None: http_status = to_int(attrs.get("http.status_code")) return code in ("STATUS_CODE_ERROR", 2) or "error" in name or bool(failure) or terminal in ("failed", "blocked") or (http_status is not None and http_status >= 500) def batches_from_body(body): if not isinstance(body, dict): return [] for key in ("batches", "resourceSpans"): value = body.get(key) if isinstance(value, list): return value return [] def scope_spans_from_batch(batch): if not isinstance(batch, dict): return [] for key in ("scopeSpans", "instrumentationLibrarySpans"): value = batch.get(key) if isinstance(value, list): return value return [] def compact_span(span, service, resource_attrs, scope_name): attrs = attrs_to_dict(span.get("attributes")) status = span.get("status") if isinstance(span.get("status"), dict) else {} item = { "name": span.get("name"), "service": service, "scope": scope_name, "statusCode": span_status_code(status), "statusMessage": status.get("message"), "durationMs": nanos_to_ms(span.get("startTimeUnixNano"), span.get("endTimeUnixNano")), "attributes": selected_attrs(attrs), } if "deployment.environment" in resource_attrs: item["environment"] = resource_attrs.get("deployment.environment") if "git.commit" in resource_attrs: item["gitCommit"] = resource_attrs.get("git.commit") return item def grep_matches_text(text): return GREP is not None and GREP.lower() in text.lower() def parse_grep_key_value(): if GREP is None: return None, None match = re.match(r"^([A-Za-z0-9_.-]+)=(.+)$", GREP) if not match: return None, None key = match.group(1) if key.startswith("span."): key = key[5:] raw = match.group(2).strip() if len(raw) >= 2 and ((raw[0] == raw[-1] == '"') or (raw[0] == raw[-1] == "'")): raw = raw[1:-1] lowered = raw.lower() if lowered == "true": return key, True if lowered == "false": return key, False if re.match(r"^-?(?:0|[1-9][0-9]*)$", raw): try: return key, int(raw) except Exception: pass if re.match(r"^-?(?:0|[1-9][0-9]*)[.][0-9]+$", raw): try: return key, float(raw) except Exception: pass return key, raw def values_equal_for_grep(actual, expected): if actual == expected: return True if isinstance(actual, bool) or isinstance(expected, bool): return str(actual).lower() == str(expected).lower() if isinstance(actual, (int, float)) and isinstance(expected, (int, float)): return float(actual) == float(expected) return str(actual) == str(expected) def grep_matches_item(item): if GREP is None: return False key, expected = parse_grep_key_value() if key is not None: if key == "name" and values_equal_for_grep(item.get("name"), expected): return True if key.startswith("resource.") and values_equal_for_grep(item.get(key.replace("resource.", "", 1)), expected): return True attrs = item.get("attributes", {}) if isinstance(item.get("attributes"), dict) else {} if key in attrs and values_equal_for_grep(attrs.get(key), expected): return True return grep_matches_text(json.dumps(item, ensure_ascii=False, sort_keys=True)) def span_matches_filters(item): attrs = item.get("attributes", {}) if isinstance(item.get("attributes"), dict) else {} if PATH_FILTER is not None: candidates = [ attrs.get("http.route"), attrs.get("http.target"), attrs.get("http.url"), attrs.get("url.path"), item.get("name"), ] if not any(isinstance(value, str) and (value == PATH_FILTER or value.startswith(PATH_FILTER + "?")) for value in candidates): return False if STATUS_FILTER is not None: status_value = to_int(attrs.get("http.response.status_code")) if status_value is None: status_value = to_int(attrs.get("http.status_code")) if status_value != STATUS_FILTER: return False return True def extract_traces(search_body): if not isinstance(search_body, dict): return [] value = search_body.get("traces") if isinstance(value, list): return value value = search_body.get("data") if isinstance(value, list): return value return [] def trace_id_from_meta(meta): if not isinstance(meta, dict): return None for key in ("traceID", "traceId", "trace_id"): value = meta.get(key) normalized = normalize_trace_id(value) if normalized is not None: return normalized return None def normalize_trace_id(value): if not isinstance(value, str): return None text = value.strip().lower() if text.startswith("0x"): text = text[2:] if not text or len(text) > 32: return None if any(ch not in "0123456789abcdef" for ch in text): return None return text.rjust(32, "0") def compact_meta(meta): if not isinstance(meta, dict): return {} output = {} for key in ("traceID", "traceId", "rootServiceName", "rootTraceName", "startTimeUnixNano", "durationMs", "spanSet", "spanSets"): if key in meta: output[key] = meta[key] return output def trace_summary(trace_id, meta, body, rc, stderr): parsed = parse_json(body) raw_matched = grep_matches_text(body) matching_active = GREP is not None or PATH_FILTER is not None or STATUS_FILTER is not None if not isinstance(parsed, dict): return { "traceId": trace_id, "traceCommand": "bun scripts/cli.ts platform-infra observability trace --target ${target.id} --trace-id %s --limit 80" % trace_id, "meta": compact_meta(meta), "queryOk": rc == 0, "parseOk": False, "rawMatched": raw_matched, "matchingActive": matching_active, "bodyBytes": len(body.encode("utf-8")), "stderrTail": (stderr or "")[-1000:], } services = set() business_trace_ids = set() name_counts = collections.Counter() spans = [] error_spans = [] matched_spans = [] for batch in batches_from_body(parsed): resource = batch.get("resource") if isinstance(batch.get("resource"), dict) else {} resource_attrs = attrs_to_dict(resource.get("attributes")) service = resource_attrs.get("service.name") or "" services.add(service) for scope_span in scope_spans_from_batch(batch): scope = scope_span.get("scope") if isinstance(scope_span.get("scope"), dict) else {} scope_name = scope.get("name") raw_spans = scope_span.get("spans") if isinstance(scope_span.get("spans"), list) else [] for span in raw_spans: if not isinstance(span, dict): continue raw_attrs = attrs_to_dict(span.get("attributes")) item = compact_span(span, service, resource_attrs, scope_name) match_item = dict(item) match_item["attributes"] = raw_attrs attrs = item.get("attributes", {}) if isinstance(attrs, dict) and isinstance(attrs.get("traceId"), str): business_trace_ids.add(attrs.get("traceId")) name = str(item.get("name") or "") name_counts[name] += 1 spans.append(item) if is_error_span(span, attrs if isinstance(attrs, dict) else {}): error_spans.append(item) if span_matches_filters(match_item) and (GREP is None or grep_matches_item(match_item)): matched_spans.append(item) return { "traceId": trace_id, "traceCommand": "bun scripts/cli.ts platform-infra observability trace --target ${target.id} --trace-id %s --limit 80" % trace_id, "grepCommand": "bun scripts/cli.ts platform-infra observability trace --target ${target.id} --trace-id %s --grep %s --limit 80" % (trace_id, json.dumps(GREP) if GREP is not None else ""), "meta": compact_meta(meta), "queryOk": rc == 0, "parseOk": True, "rawMatched": raw_matched, "matchingActive": matching_active, "bodyBytes": len(body.encode("utf-8")), "spanCount": len(spans), "serviceCount": len(services), "services": sorted(services), "businessTraceIds": sorted(business_trace_ids)[:20], "errorSpanCount": len(error_spans), "matchedSpanCount": len(matched_spans) if matching_active else None, "spanNameCounts": [{"name": name, "count": count} for name, count in name_counts.most_common(10)], "errorSpans": error_spans[:LIMIT] if FULL else error_spans[: min(3, LIMIT)], "matchedSpans": matched_spans[:LIMIT], "stderrTail": (stderr or "")[-1000:], } def rich_trace_rank(summary): services = set(summary.get("services") or []) service_score = 0 for service in ("agentrun-runner", "agentrun-manager", "hwlab-cloud-api"): if service in services: service_score += 1 return ( service_score, int(summary.get("spanCount") or 0), int(summary.get("errorSpanCount") or 0), int(summary.get("matchedSpanCount") or 0), ) search_rc, search_body, search_err = run_kubectl(SEARCH_PROXY_PATH, timeout=15) search_parsed = parse_json(search_body) search_parse_ok = isinstance(search_parsed, dict) trace_metas = extract_traces(search_parsed) candidate_trace_ids = [] seen = set() for meta in trace_metas: trace_id = trace_id_from_meta(meta) if trace_id is None or trace_id in seen: continue seen.add(trace_id) candidate_trace_ids.append((trace_id, meta)) if len(candidate_trace_ids) >= CANDIDATE_LIMIT: break matched = [] scanned = [] scan_stopped = None for trace_id, meta in candidate_trace_ids: if time.time() > DEADLINE: scan_stopped = "deadline" break trace_path = TRACE_PATH_TEMPLATE.replace("{{traceId}}", trace_id) trace_rc, trace_body, trace_err = run_kubectl(TRACE_PROXY_PREFIX + trace_path, timeout=8) summary = trace_summary(trace_id, meta, trace_body, trace_rc, trace_err) scanned.append(summary) if summary.get("matchingActive") is not True or summary.get("rawMatched") is True or (summary.get("matchedSpanCount") or 0) > 0: matched.append(summary) if len(matched) >= LIMIT and GREP is not None and not BUSINESS_TRACE_GREP: break if BUSINESS_TRACE_GREP: matched.sort(key=rich_trace_rank, reverse=True) payload = { "ok": search_rc == 0 and search_parse_ok, "searchPath": "${searchPath}", "searchProxyPath": SEARCH_PROXY_PATH, "grep": GREP, "tempoQuery": QUERY, "pathFilter": PATH_FILTER, "statusFilter": STATUS_FILTER, "grepCoverage": None if GREP is None else "raw trace body, span name, status message, route and full span attributes inside scanned candidate traces", "grepQueryInference": "tempo-query-present" if QUERY is not None and GREP is not None else None, "businessTraceSearch": BUSINESS_TRACE_GREP, "limit": LIMIT, "candidateLimit": CANDIDATE_LIMIT, "searchParseOk": search_parse_ok, "candidateTraceCount": len(candidate_trace_ids), "scannedTraceCount": len(scanned), "matchedTraceCount": len(matched), "scanStopped": scan_stopped, "matchingActive": GREP is not None or PATH_FILTER is not None or STATUS_FILTER is not None, "traces": matched[:LIMIT] if (GREP is not None or PATH_FILTER is not None or STATUS_FILTER is not None) else scanned[:LIMIT], "truncated": { "candidateTraces": len(trace_metas) > len(candidate_trace_ids), "matchedTraces": len(matched) > LIMIT, }, "next": { "expandWindow": "bun scripts/cli.ts platform-infra observability search --target ${target.id} --grep --lookback-minutes 1440 --candidate-limit 200 --limit 40", "pathStatus": "bun scripts/cli.ts platform-infra observability search --target ${target.id} --path /v1/workbench/sessions --status 502 --lookback-minutes 1440 --candidate-limit 200 --limit 40", "traceDetail": "bun scripts/cli.ts platform-infra observability trace --target ${target.id} --trace-id --grep --limit 80", "raw": "bun scripts/cli.ts platform-infra observability search --target ${target.id} --grep --raw", }, "searchStderrTail": (search_err or "")[-2000:], } if RAW: payload["rawSearchBody"] = search_parsed if search_parse_ok else search_body[-12000:] print(json.dumps(payload, ensure_ascii=False, indent=2)) raise SystemExit(0 if payload["ok"] else 1) PY `; } function cliArg(value: string): string { return /^[A-Za-z0-9_./:=@+-]+$/u.test(value) ? value : shQuote(value); } function diagnoseSearchFilters(options: DiagnoseCodeAgentOptions): string[] { const filters: string[] = []; if (options.businessTraceId !== null) filters.push(`.traceId = ${JSON.stringify(options.businessTraceId)}`); if (options.runId !== null) filters.push(`.runId = ${JSON.stringify(options.runId)}`); if (options.commandId !== null) filters.push(`.commandId = ${JSON.stringify(options.commandId)}`); if (options.sessionId !== null) filters.push(`.sessionId = ${JSON.stringify(options.sessionId)}`); if (options.runnerJobId !== null) filters.push(`.runnerJobId = ${JSON.stringify(options.runnerJobId)}`); return filters; } function diagnoseSearchMode(options: DiagnoseCodeAgentOptions): string { if (options.businessTraceId !== null && options.runId === null && options.commandId === null && options.sessionId === null && options.runnerJobId === null) return "business-trace-id"; if (options.traceId !== null) return "trace-id"; return "trace-attribute-query"; } function buildDiagnoseCodeAgentCommand(target: ObservabilityTarget, options: DiagnoseCodeAgentOptions, full: boolean): string { const parts = ["bun", "scripts/cli.ts", "platform-infra", "observability", "diagnose-code-agent", "--target", target.id]; if (options.businessTraceId !== null) parts.push("--business-trace-id", options.businessTraceId); if (options.traceId !== null) parts.push("--trace-id", options.traceId); if (options.runId !== null) parts.push("--run-id", options.runId); if (options.commandId !== null) parts.push("--command-id", options.commandId); if (options.sessionId !== null) parts.push("--session-id", options.sessionId); if (options.runnerJobId !== null) parts.push("--runner-job-id", options.runnerJobId); parts.push("--lookback-minutes", String(options.lookbackMinutes), "--candidate-limit", String(options.candidateLimit), "--limit", String(options.limit)); if (full) parts.push("--full"); return parts.map(cliArg).join(" "); } function buildDiagnoseSearchCommand(target: ObservabilityTarget, tempoQuery: string | null): string { const query = tempoQuery ?? "{ .traceId = \"\" }"; return [ "bun", "scripts/cli.ts", "platform-infra", "observability", "search", "--target", target.id, "--query", query, "--lookback-minutes", "10080", "--candidate-limit", "500", "--limit", "10", ].map(cliArg).join(" "); } export function diagnoseCodeAgentScript(observability: ObservabilityConfig, target: ObservabilityTarget, searchPath: string | null, options: DiagnoseCodeAgentOptions): string { const proxyPrefix = `/api/v1/namespaces/${target.namespace}/services/http:${observability.traceBackend.serviceName}:http/proxy`; const searchProxyPath = searchPath === null ? null : `${proxyPrefix}${searchPath}`; const traceIdLiteral = options.traceId === null ? "None" : JSON.stringify(options.traceId); const businessTraceIdLiteral = options.businessTraceId === null ? "None" : JSON.stringify(options.businessTraceId); const runIdLiteral = options.runId === null ? "None" : JSON.stringify(options.runId); const commandIdLiteral = options.commandId === null ? "None" : JSON.stringify(options.commandId); const sessionIdLiteral = options.sessionId === null ? "None" : JSON.stringify(options.sessionId); const runnerJobIdLiteral = options.runnerJobId === null ? "None" : JSON.stringify(options.runnerJobId); const searchPathLiteral = searchPath === null ? "None" : JSON.stringify(searchPath); const searchProxyPathLiteral = searchProxyPath === null ? "None" : JSON.stringify(searchProxyPath); const tempoQuery = (() => { const filters = diagnoseSearchFilters(options); return filters.length > 0 ? `{ ${filters.join(" && ")} }` : null; })(); const searchMode = diagnoseSearchMode(options); const diagnoseFullCommand = buildDiagnoseCodeAgentCommand(target, options, true); const searchCommand = buildDiagnoseSearchCommand(target, tempoQuery); return ` set -u python3 - <<'PY' import collections, json, re, subprocess, time, urllib.parse BUSINESS_TRACE_ID = ${businessTraceIdLiteral} TRACE_ID = ${traceIdLiteral} RUN_ID = ${runIdLiteral} COMMAND_ID = ${commandIdLiteral} SESSION_ID = ${sessionIdLiteral} RUNNER_JOB_ID = ${runnerJobIdLiteral} TARGET_ID = ${JSON.stringify(target.id)} SEARCH_MODE = ${JSON.stringify(searchMode)} SEARCH_QUERY = ${tempoQuery === null ? "None" : JSON.stringify(tempoQuery)} SEARCH_PATH = ${searchPathLiteral} SEARCH_PROXY_PATH = ${searchProxyPathLiteral} TRACE_PROXY_PREFIX = ${JSON.stringify(proxyPrefix)} TRACE_PATH_TEMPLATE = ${JSON.stringify(observability.probes.traceQueryPathTemplate)} DIAGNOSE_FULL_COMMAND = ${JSON.stringify(diagnoseFullCommand)} SEARCH_COMMAND = ${JSON.stringify(searchCommand)} FULL = ${options.full ? "True" : "False"} RAW = ${options.raw ? "True" : "False"} LIMIT = ${options.limit} CANDIDATE_LIMIT = ${options.candidateLimit} DEADLINE = time.time() + 50 IMPORTANT_ATTRS = [ "traceId", "otel.trace_id", "agentrun.stage", "runId", "commandId", "sessionId", "turnId", "threadId", "runnerJobId", "failureKind", "willRetry", "retryAttempt", "retryMax", "retryExhausted", "retryBackoffMs", "idleMs", "idleSeconds", "idleThresholdMs", "idleThresholdSeconds", "waitingFor", "lastEventLabel", "lastActivityAt", "lastActivityMs", "lastNotificationAt", "lastNotificationMs", "lastToolCallAt", "upstreamHttpStatus", "upstreamHost", "providerErrorClass", "errorSummary", "terminalStatus", "phase", "message", "http.route", "http.status_code", "http.response.status_code", "http.method", "http.request.method", "http.response.status_class", "workbench.ui.event_type", "workbench.ui.scope", "workbench.ui.state", "workbench.ui.reason", "workbench.ui.route", "workbench.ui.outcome", "workbench.ui.value_ms", "workbench.ui.visibility", "workbench.ui.event_name", "workbench.ui.activity_idle_ms", "workbench.ui.activity_waiting_for", "workbench.ui.activity_last_event_label", "workbench.ui.resource_duration_ms", "workbench.ui.resource_fetch_to_request_ms", "workbench.ui.resource_request_wait_ms", "workbench.ui.resource_response_transfer_ms", "workbench.ui.resource_transfer_size", "workbench.ui.resource_encoded_body_size", "workbench.ui.resource_decoded_body_size", "workbench.ui.resource_next_hop_protocol", "workbench.ui.resource_server_timing", "workbench.session_id", "workbench.trace_id", "workbench.turn_id", "workbench.read_model.route", "workbench.read_model.count", "workbench.read_model.family", "workbench.read_model.status", "workbench.read_model.reason", "workbench.projection.phase", "workbench.projection.session_id", "workbench.projection.turn_id", "workbench.projection.projected_seq", "workbench.projection.source_seq", "workbench.projection.source_event_id", "workbench.projection.event_type", "workbench.projection.message_id", "workbench.projection.terminal", "workbench.projection.suppressed_after_seal", "workbench.projection.terminal_trace_backfill_written", "source", "sessionCount", "fallbackTitleCount", "fallbackTitleRatio", "emptyPreviewCount", "includeSessionId", "sessionIds", "traceIds", "returnedMessages", "totalMessages", "roleSequencePrefix", "consecutiveUserPrefix", "adjacentSameRoleCount", "userCount", "agentCount", "db.system", "db.operation.name", "db.sql.table", "db.query.arg_count", "db.index.expected", "db.pool.max_open", "db.pool.open_connections", "db.pool.in_use", "db.pool.idle", "db.pool.wait_count", "db.pool.wait_duration_ms", "db.pool.max_idle_closed", "db.pool.max_lifetime_closed", "hwlab.runtime.stage", "hwlab.http.stage", "hwlab.http.phase", "hwlab.http.phase.outcome", "hwlab.live_builds.service_id", "hwlab.live_builds.service_kind", "hwlab.live_builds.external", "hwlab.live_builds.deploy_manifest_status", "hwlab.live_builds.artifact_catalog_status", "error.code", "error.category", "error.layer", "stage", "causeStage", "causeCode", "eventType", "runnerId", "attemptId", "backendProfile", "sourceCommit", "jobName", "podName", "logPath", "toolName", "type", "itemType", "itemId", "status", "exitCode", "durationMs", "cwd", "processId", "command", "commandFingerprint", "outputSummary", "outputBytes", "outputTruncated", "storageKind", "storagePvcName", "storagePvcPhase", "resumeMode", "rolloutId", "returnedEvents", "sinceSeq", "afterSeq", "limit", "fromSeq", "toSeq", "totalEvents", "hasMore", "fullTraceLoaded", "rawEventCount", "maxSeq", "traceLastSeq", "endSeq", "commandFiltered", "seq", "eventSeq", "sourceSeq", "sourceLatestSeq", "latestSeq", "lastSeq", "sourceEventCount", "projectedSeq", "lastProjectedSeq", "status", "turnStatus", "error", "error.message", "exception.type", "exception.message", "valuesPrinted", ] IDENTITY_KEYS = [ "runId", "commandId", "sessionId", "turnId", "threadId", "runnerJobId", "runnerId", "attemptId", "backendProfile", "sourceCommit", "jobName", "podName", ] def run_kubectl(proxy_path, timeout=15): try: proc = subprocess.run( ["kubectl", "get", "--raw", proxy_path], capture_output=True, text=True, timeout=timeout, ) return proc.returncode, proc.stdout, proc.stderr except subprocess.TimeoutExpired as exc: return 124, exc.stdout or "", "timeout while querying tempo" 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 preferred_failure_kind(values): kinds = [] for value in values: if value in (None, ""): continue text = str(value).strip() if not text: continue if text not in kinds: kinds.append(text) if not kinds: return None for kind in kinds: lowered = kind.lower() if lowered.startswith("provider-") or lowered.startswith("provider_") or "provider-auth" in lowered: return kind for kind in kinds: lowered = kind.lower() if lowered not in ("infra-failed", "infra_failed", "unknown"): return kind return kinds[-1] 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 except Exception: return None def attr_value(value): if not isinstance(value, dict): return value inner = value.get("value", value) if not isinstance(inner, dict): return inner for key in ("stringValue", "intValue", "doubleValue", "boolValue"): if key in inner: return inner.get(key) if "arrayValue" in inner: return "" if "kvlistValue" in inner: return "" if "bytesValue" in inner: return "" return inner def attrs_to_dict(attrs): if not isinstance(attrs, list): return {} output = {} for item in attrs: if not isinstance(item, dict): continue key = item.get("key") if isinstance(key, str): output[key] = attr_value(item.get("value")) return output def selected_attrs(attrs): output = {} for key in IMPORTANT_ATTRS: if key in attrs: output[key] = attrs[key] return output def to_int(value): if value is None: return None if isinstance(value, bool): return 1 if value else 0 if isinstance(value, int): return value if isinstance(value, float) and value.is_integer(): return int(value) if isinstance(value, str): try: return int(value.strip()) except Exception: return None return None def to_bool(value): if isinstance(value, bool): return value if isinstance(value, str): lowered = value.strip().lower() if lowered in ("true", "1", "yes"): return True if lowered in ("false", "0", "no"): return False return None def nanos_to_ms(start, end): try: return round((int(end) - int(start)) / 1000000, 3) except Exception: return None def span_status_code(status): if not isinstance(status, dict): return None return status.get("code") def is_error_span(span, attrs): status = span.get("status") if isinstance(span, dict) else {} code = span_status_code(status) name = str(span.get("name", "")).lower() if isinstance(span, dict) else "" failure = str(attrs.get("failureKind", "")).strip() terminal = str(attrs.get("terminalStatus", "")).strip() return code in ("STATUS_CODE_ERROR", 2) or "error" in name or bool(failure) or terminal in ("failed", "blocked") def batches_from_body(body): if not isinstance(body, dict): return [] for key in ("batches", "resourceSpans"): value = body.get(key) if isinstance(value, list): return value return [] def scope_spans_from_batch(batch): if not isinstance(batch, dict): return [] for key in ("scopeSpans", "instrumentationLibrarySpans"): value = batch.get(key) if isinstance(value, list): return value return [] def extract_traces(search_body): if not isinstance(search_body, dict): return [] value = search_body.get("traces") if isinstance(value, list): return value value = search_body.get("data") if isinstance(value, list): return value return [] def trace_id_from_meta(meta): if not isinstance(meta, dict): return None for key in ("traceID", "traceId", "trace_id"): value = meta.get(key) normalized = normalize_trace_id(value) if normalized is not None: return normalized return None def normalize_trace_id(value): if not isinstance(value, str): return None text = value.strip().lower() if text.startswith("0x"): text = text[2:] if not text or len(text) > 32: return None if any(ch not in "0123456789abcdef" for ch in text): return None return text.rjust(32, "0") def compact_meta(meta): if not isinstance(meta, dict): return {} output = {} for key in ("traceID", "traceId", "rootServiceName", "rootTraceName", "startTimeUnixNano", "durationMs"): if key in meta: output[key] = meta[key] return output def compact_span(span, service, resource_attrs, scope_name, index): raw_attrs = attrs_to_dict(span.get("attributes")) attrs = selected_attrs(raw_attrs) status = span.get("status") if isinstance(span.get("status"), dict) else {} item = { "name": span.get("name"), "service": service, "scope": scope_name, "statusCode": span_status_code(status), "statusMessage": status.get("message"), "durationMs": nanos_to_ms(span.get("startTimeUnixNano"), span.get("endTimeUnixNano")), "attributes": attrs, "_index": index, "_start": to_int(span.get("startTimeUnixNano")) or index, } if "deployment.environment" in resource_attrs: item["environment"] = resource_attrs.get("deployment.environment") if "git.commit" in resource_attrs: item["gitCommit"] = resource_attrs.get("git.commit") return item, raw_attrs def public_span(item): if not isinstance(item, dict): return item return {key: value for key, value in item.items() if not key.startswith("_")} def tiny_span(item): if not isinstance(item, dict): return item attrs = item.get("attributes", {}) if isinstance(item.get("attributes"), dict) else {} keep_attrs = {} for key in ( "http.route", "http.status_code", "http.method", "status", "terminalStatus", "eventType", "returnedEvents", "sinceSeq", "fromSeq", "toSeq", "totalEvents", "hasMore", "fullTraceLoaded", "afterSeq", "rawEventCount", "failureKind", "willRetry", "retryAttempt", "retryMax", "retryExhausted", "retryBackoffMs", "idleMs", "idleSeconds", "idleThresholdMs", "idleThresholdSeconds", "waitingFor", "lastEventLabel", "lastActivityAt", "lastActivityMs", "lastNotificationAt", "lastNotificationMs", "lastToolCallAt", "upstreamHttpStatus", "upstreamHost", "providerErrorClass", "errorSummary", "toolName", "type", "itemType", "itemId", "status", "exitCode", "durationMs", "cwd", "processId", "command", "commandFingerprint", "outputSummary", "outputBytes", "outputTruncated", "hwlab.http.stage", "hwlab.http.phase", "hwlab.http.phase.outcome", "hwlab.live_builds.service_id", "hwlab.live_builds.service_kind", "hwlab.live_builds.external", "hwlab.live_builds.deploy_manifest_status", "hwlab.live_builds.artifact_catalog_status", ): if key in attrs: keep_attrs[key] = attrs[key] return { "name": item.get("name"), "service": item.get("service"), "attributes": keep_attrs, } def tiny_terminal(terminal): if not isinstance(terminal, dict): return terminal return { "status": terminal.get("status"), "eventType": terminal.get("eventType"), "span": tiny_span(terminal.get("span")), } def identity_from_spans(spans): identity = {} for key in IDENTITY_KEYS: preferred = None fallback = None for item in spans: attrs = item.get("attributes", {}) if isinstance(item.get("attributes"), dict) else {} value = attrs.get(key) if value in (None, ""): continue if fallback is None: fallback = value service = str(item.get("service") or "") if service.startswith("agentrun"): preferred = value break if preferred is not None or fallback is not None: identity[key] = preferred if preferred is not None else fallback return identity def flatten_trace(trace_body): services = set() business_trace_ids = set() name_counts = collections.Counter() spans = [] error_spans = [] raw_index = 0 for batch in batches_from_body(trace_body): resource = batch.get("resource") if isinstance(batch.get("resource"), dict) else {} resource_attrs = attrs_to_dict(resource.get("attributes")) service = resource_attrs.get("service.name") or "" services.add(service) for scope_span in scope_spans_from_batch(batch): scope = scope_span.get("scope") if isinstance(scope_span.get("scope"), dict) else {} scope_name = scope.get("name") raw_spans = scope_span.get("spans") if isinstance(scope_span.get("spans"), list) else [] for span in raw_spans: if not isinstance(span, dict): continue item, raw_attrs = compact_span(span, service, resource_attrs, scope_name, raw_index) raw_index += 1 attrs = item.get("attributes", {}) if isinstance(item.get("attributes"), dict) else {} if isinstance(attrs.get("traceId"), str): business_trace_ids.add(attrs.get("traceId")) name = str(item.get("name") or "") name_counts[name] += 1 spans.append(item) if is_error_span(span, raw_attrs): error_spans.append(item) spans.sort(key=lambda item: (item.get("_start", 0), item.get("_index", 0))) error_spans.sort(key=lambda item: (item.get("_start", 0), item.get("_index", 0))) return { "spans": spans, "services": sorted(services), "businessTraceIds": sorted(business_trace_ids), "spanNameCounts": [{"name": name, "count": count} for name, count in name_counts.most_common(20)], "errorSpans": error_spans, } def raw_trace_shape(trace_body): batches = batches_from_body(trace_body) scope_count = 0 span_count = 0 first_resource_attrs = {} first_scope_keys = [] first_span_keys = [] first_span_attr_keys = [] for batch_index, batch in enumerate(batches): resource = batch.get("resource") if isinstance(batch.get("resource"), dict) else {} resource_attrs = attrs_to_dict(resource.get("attributes")) if batch_index == 0: for key in ("service.name", "deployment.environment", "git.commit"): if key in resource_attrs: first_resource_attrs[key] = resource_attrs[key] for scope_span in scope_spans_from_batch(batch): scope_count += 1 if not first_scope_keys and isinstance(scope_span, dict): first_scope_keys = sorted(scope_span.keys()) raw_spans = scope_span.get("spans") if isinstance(scope_span.get("spans"), list) else [] span_count += len(raw_spans) if not first_span_keys and raw_spans and isinstance(raw_spans[0], dict): first_span_keys = sorted(raw_spans[0].keys()) first_span_attr_keys = sorted(attrs_to_dict(raw_spans[0].get("attributes")).keys())[:50] return { "mode": "bounded-raw-shape", "note": "Full Tempo body is intentionally not printed to avoid secrets/raw transcript and SSH 60s output timeouts.", "topLevelKeys": sorted(trace_body.keys()) if isinstance(trace_body, dict) else [], "resourceSpanCount": len(batches), "scopeSpanCount": scope_count, "spanCount": span_count, "firstResourceAttributes": first_resource_attrs, "firstScopeKeys": first_scope_keys, "firstSpanKeys": first_span_keys, "firstSpanAttributeKeys": first_span_attr_keys, } def http_status_summary(spans): grouped = collections.Counter() problem_spans = [] for item in spans: attrs = item.get("attributes", {}) if isinstance(item.get("attributes"), dict) else {} code = to_int(attrs.get("http.status_code")) if code is None: continue method = attrs.get("http.method") or "" route = attrs.get("http.route") or item.get("name") or "" grouped[(str(method), str(route), code)] += 1 if code in (401, 403) or code >= 500: problem_spans.append(item) status_counts = [ {"method": method or None, "route": route, "status": status, "count": count} for (method, route, status), count in grouped.most_common() ] problem_counts = [ {"method": method or None, "route": route, "status": status, "count": count} for (method, route, status), count in grouped.most_common() if status in (401, 403) or status >= 500 ] return { "statusCounts": status_counts[:20], "problemCounts": problem_counts[:20], "problemSpans": problem_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, authority=None): terminal_spans = [] latest_seq = None failure_kinds = [] failure_message = None terminal_category = None seq_keys = ("maxSeq", "traceLastSeq", "endSeq", "seq", "eventSeq", "sourceSeq", "sourceLatestSeq", "latestSeq", "lastSeq") for item in spans: service = str(item.get("service") or "") name = str(item.get("name") or "") attrs = item.get("attributes", {}) if isinstance(item.get("attributes"), dict) else {} if service.startswith("agentrun"): for key in seq_keys: value = to_int(attrs.get(key)) if value is not None: latest_seq = value if latest_seq is None else max(latest_seq, value) terminal_status = attrs.get("terminalStatus") if terminal_status or name.startswith("runner_terminal."): derived = terminal_status if not derived and name.startswith("runner_terminal."): derived = name.split(".", 1)[1] if attrs.get("failureKind") not in (None, ""): failure_kinds.append(attrs.get("failureKind")) if attrs.get("failureMessage") not in (None, ""): failure_message = attrs.get("failureMessage") if attrs.get("terminalCategory") not in (None, ""): terminal_category = attrs.get("terminalCategory") terminal_spans.append({ "status": derived, "eventType": attrs.get("eventType"), "failureKind": attrs.get("failureKind"), "terminalCategory": attrs.get("terminalCategory"), "span": public_span(item), }) terminal_status = terminal_spans[-1]["status"] if terminal_spans else None terminal_event_type = None for terminal in reversed(terminal_spans): if terminal.get("eventType"): terminal_event_type = terminal.get("eventType") break 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 if authority.get("failureKind") not in (None, ""): failure_kinds.append(authority.get("failureKind")) terminal_source = "agentrun-authority" failure_kind = preferred_failure_kind(failure_kinds) runner_classification = classify_terminal_status(terminal_status, spans) return { "terminalStatus": terminal_status, "failureKind": failure_kind, "observedFailureKinds": failure_kinds, "failureMessage": failure_message, "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, } def hwlab_read_model_summary(spans): trace_read_spans = [] turn_read_spans = [] projection_spans = [] source_event_count = None requested_since_seq = None latest_projected_seq = None turn_statuses = [] turn_status_counts = collections.Counter() for item in spans: if item.get("service") != "hwlab-cloud-api": continue name = str(item.get("name") or "") attrs = item.get("attributes", {}) if isinstance(item.get("attributes"), dict) else {} if name == "trace_events_read": trace_read_spans.append(public_span(item)) for key in ("totalEvents", "sourceEventCount"): value = to_int(attrs.get(key)) if value is not None: source_event_count = value if source_event_count is None else max(source_event_count, value) for key in ("sinceSeq", "fromSeq"): value = to_int(attrs.get(key)) if value is not None: requested_since_seq = value if requested_since_seq is None else max(requested_since_seq, value) if name == "turn_status_read": turn_read_spans.append(public_span(item)) if attrs.get("turnStatus") is not None or attrs.get("status") is not None: status_value = attrs.get("turnStatus") if attrs.get("turnStatus") is not None else attrs.get("status") turn_statuses.append(status_value) turn_status_counts[str(status_value)] += 1 if name == "projection_sync": projection_spans.append(public_span(item)) after_seq = to_int(attrs.get("afterSeq")) raw_count = to_int(attrs.get("rawEventCount")) projected = None for key in ("maxSeq", "traceLastSeq", "endSeq", "projectedSeq", "lastProjectedSeq"): value = to_int(attrs.get(key)) if value is not None: projected = value if projected is None else max(projected, value) if after_seq is not None: projected = after_seq + (raw_count or 0) if projected is not None: latest_projected_seq = projected if latest_projected_seq is None else max(latest_projected_seq, projected) source_event_count = projected if source_event_count is None else max(source_event_count, projected) has_more_values = [] full_loaded_values = [] for item in trace_read_spans: attrs = item.get("attributes", {}) if isinstance(item.get("attributes"), dict) else {} if "hasMore" in attrs: has_more_values.append(to_bool(attrs.get("hasMore"))) if "fullTraceLoaded" in attrs: full_loaded_values.append(to_bool(attrs.get("fullTraceLoaded"))) return { "sourceEventCount": source_event_count, "requestedSinceSeq": requested_since_seq, "latestProjectedSeq": latest_projected_seq, "traceStatus": { "readCount": len(trace_read_spans), "anyHasMore": any(value is True for value in has_more_values), "anyFullTraceLoaded": any(value is True for value in full_loaded_values), }, "turnStatus": turn_statuses[-1] if turn_statuses else None, "turnStatusCounts": [{"status": status, "count": count} for status, count in turn_status_counts.most_common()], "traceEventReadSpans": trace_read_spans, "turnStatusReadSpans": turn_read_spans, "projectionSpans": projection_spans, } def projection_lag_summary(agentrun, read_model): source_event_count = read_model.get("sourceEventCount") requested_since_seq = read_model.get("requestedSinceSeq") latest_projected_seq = read_model.get("latestProjectedSeq") agent_latest_seq = agentrun.get("latestSeq") reasons = [] status = "none" if requested_since_seq is not None and source_event_count is not None and requested_since_seq > source_event_count: status = "confirmed" reasons.append("requested sinceSeq %s is greater than HWLAB sourceEventCount %s" % (requested_since_seq, source_event_count)) if latest_projected_seq is not None and requested_since_seq is not None and latest_projected_seq < requested_since_seq: status = "confirmed" reasons.append("latestProjectedSeq %s is behind requested sinceSeq %s" % (latest_projected_seq, requested_since_seq)) if agent_latest_seq is not None and source_event_count is not None and agent_latest_seq > source_event_count: status = "confirmed" reasons.append("AgentRun latestSeq %s is greater than HWLAB sourceEventCount %s" % (agent_latest_seq, source_event_count)) if status == "none" and agentrun.get("terminalStatus") == "completed" and source_event_count is not None and read_model.get("traceStatus", {}).get("readCount", 0) > 0: status = "suspected" reasons.append("AgentRun completed while HWLAB read model still needs endpoint/status confirmation") return { "status": status, "reasons": reasons, } def root_cause_candidates(http_summary, agentrun, read_model, lag_summary, error_spans, idle_warning_spans): candidates = [] if http_summary.get("actorForbidden"): candidates.append({ "code": "actor_forbidden", "label": "actor forbidden", "confidence": 0.98, "summary": "actor forbidden: HWLAB Code Agent read endpoint returned HTTP 403, likely actor/owner mismatch for this trace or turn.", "evidence": http_summary.get("problemCounts", [])[:5], }) if lag_summary.get("status") == "confirmed": candidates.append({ "code": "projection_read_model_stale", "label": "projection/read-model stale", "confidence": 0.94, "summary": "Projection/read-model stale: read model sequence is behind the caller/requested event window.", "evidence": { "sourceEventCount": read_model.get("sourceEventCount"), "requestedSinceSeq": read_model.get("requestedSinceSeq"), "latestProjectedSeq": read_model.get("latestProjectedSeq"), "reasons": lag_summary.get("reasons", []), }, }) elif lag_summary.get("status") == "suspected": candidates.append({ "code": "projection_read_model_lag_suspected", "label": "projection/read-model stale", "confidence": 0.64, "summary": "Projection/read-model stale is suspected because AgentRun is terminal but HWLAB read-model evidence is incomplete.", "evidence": lag_summary.get("reasons", []), }) failure_kind = str(agentrun.get("failureKind") or "") if failure_kind == "provider-auth-failed": candidates.append({ "code": "provider_auth_failed", "label": "provider auth failed", "confidence": 0.96, "summary": "Provider credential refresh/auth failed before the manager reconciler closed the command; fix provider auth before retrying this Code Agent path.", "evidence": { "terminalStatus": agentrun.get("terminalStatus"), "failureKind": agentrun.get("failureKind"), "observedFailureKinds": agentrun.get("observedFailureKinds"), "terminalCategory": agentrun.get("terminalCategory"), }, }) if agentrun.get("terminalStatus") in ("failed", "error", "timeout", "blocked", "cancelled"): candidates.append({ "code": "agentrun_terminal_failed", "label": "AgentRun terminal failed", "confidence": 0.9, "summary": "AgentRun reached a non-success terminal state; inspect result/events/runnerjob before retrying or continuing the session.", "evidence": { "terminalStatus": agentrun.get("terminalStatus"), "failureKind": agentrun.get("failureKind"), "terminalCategory": agentrun.get("terminalCategory"), "runnerProviderClassification": agentrun.get("runnerProviderClassification"), }, }) if agentrun.get("terminalStatus") == "completed": candidates.append({ "code": "agentrun_completed_not_provider_blocked", "label": "AgentRun completed", "confidence": 0.88, "summary": "AgentRun completed: runner/provider reached terminal completed, so this trace is not explained by an active runner hang.", "evidence": { "terminalStatus": agentrun.get("terminalStatus"), "terminalEventType": agentrun.get("terminalEventType"), "runnerProviderClassification": agentrun.get("runnerProviderClassification"), }, }) if error_spans and agentrun.get("terminalStatus") == "completed": candidates.append({ "code": "tool_call_failures_recovered", "label": "tool call failures recovered", "confidence": 0.42, "summary": "tool call failure spans exist, but AgentRun still completed; treat them as secondary unless correlated with user-visible failure.", "evidence": { "errorSpanCount": len(error_spans), "sampleNames": sorted(set(str(item.get("name") or "") for item in error_spans))[:5], }, }) if idle_warning_spans and agentrun.get("terminalStatus") in (None, ""): candidates.append({ "code": "agentrun_runner_idle_warning_active", "label": "AgentRun runner idle warning active", "confidence": 0.72, "summary": "AgentRun runner is still emitting idle warnings and no runner terminal span is present; inspect waitingFor/lastEventLabel before treating the Workbench UI as terminal.", "evidence": [tiny_span(item) for item in idle_warning_spans[-3:]], }) candidates.sort(key=lambda item: item.get("confidence", 0), reverse=True) return candidates def candidate_score(trace_id, meta, trace_body, trace_rc, trace_err): parsed = parse_json(trace_body) meta_root_service = str(meta.get("rootServiceName") or "") if isinstance(meta, dict) else "" meta_root_name = str(meta.get("rootTraceName") or "") if isinstance(meta, dict) else "" if not isinstance(parsed, dict): return { "traceId": trace_id, "score": -100, "reasons": ["trace parse failed"], "parseOk": False, "queryOk": trace_rc == 0, "meta": compact_meta(meta), "spanCount": 0, "services": [], "servicePath": { "hwlab-cloud-api": "unknown", "agentrun-manager": "unknown", "agentrun-runner": "unknown", "complete": False, }, "rootTraceName": meta_root_name or None, "rootServiceName": meta_root_service or None, "stderrTail": (trace_err or "")[-1000:], }, None flat = flatten_trace(parsed) spans = flat["spans"] services = set(flat["services"]) names = [str(item.get("name") or "") for item in spans] lowered_names = [name.lower() for name in names] identity = identity_from_spans(spans) agentrun = agentrun_summary(spans) error_spans = flat["errorSpans"] score = 0 reasons = [] def add(points, reason): nonlocal score score += points reasons.append("%+d %s" % (points, reason)) if "agentrun-runner" in services: add(120, "contains agentrun-runner") if "agentrun-manager" in services: add(90, "contains agentrun-manager") if "hwlab-cloud-api" in services: add(20, "contains hwlab-cloud-api") span_points = min(len(spans), 80) if span_points: add(span_points, "span count %s" % len(spans)) for key, points in (("runId", 50), ("commandId", 50), ("sessionId", 35), ("runnerJobId", 35), ("runnerId", 30), ("attemptId", 20), ("backendProfile", 15)): if identity.get(key) not in (None, ""): add(points, "identity %s" % key) if any("codex_stdio" in name for name in lowered_names): add(45, "codex stdio span") if any("runner" in name for name in lowered_names): add(30, "runner span") if any("provider" in name for name in lowered_names): add(25, "provider span") if any("tool_call" in name or "command" in name for name in lowered_names): add(20, "tool/command span") if any("turn_completed" in name or "runner_terminal" in name or "command_result" in name for name in lowered_names): add(35, "terminal/result span") if error_spans: add(40 + min(len(error_spans), 20), "error span count %s" % len(error_spans)) if agentrun.get("terminalStatus") not in (None, ""): add(25, "terminal status %s" % agentrun.get("terminalStatus")) is_workbench_get_single_span = ( len(spans) <= 1 and services.issubset({"hwlab-cloud-api"}) and meta_root_name.startswith("GET /v1/workbench/") ) if is_workbench_get_single_span: add(-80, "single-span Workbench GET/SSE trace") candidate_quality = "low-confidence-workbench-single-span" if is_workbench_get_single_span else "normal" service_path = { service: ("reached" if service in services else "missing") for service in ("hwlab-cloud-api", "agentrun-manager", "agentrun-runner") } service_path["complete"] = all(service in services for service in ("hwlab-cloud-api", "agentrun-manager", "agentrun-runner")) return { "traceId": trace_id, "score": score, "reasons": reasons[:12], "parseOk": True, "queryOk": trace_rc == 0, "meta": compact_meta(meta), "spanCount": len(spans), "services": sorted(services), "servicePath": service_path, "identity": {key: identity.get(key) for key in ("runId", "commandId", "sessionId", "runnerJobId", "runnerId", "backendProfile") if identity.get(key) not in (None, "")}, "terminalStatus": agentrun.get("terminalStatus"), "errorSpanCount": len(error_spans), "candidateQuality": candidate_quality, "lowConfidence": candidate_quality != "normal", "rootTraceName": meta_root_name or None, "rootServiceName": meta_root_service or None, "spanNamePreview": names[:8], "traceCommand": "bun scripts/cli.ts platform-infra observability trace --target ${target.id} --trace-id %s --limit 80" % trace_id, }, parsed def resolve_trace(): if TRACE_ID is not None: return TRACE_ID, { "mode": "trace-id", "businessTraceId": BUSINESS_TRACE_ID, "runId": RUN_ID, "commandId": COMMAND_ID, "sessionId": SESSION_ID, "runnerJobId": RUNNER_JOB_ID, "otelTraceId": TRACE_ID, "searchQuery": SEARCH_QUERY, "searchPath": None, "searchOk": None, "searchParseOk": None, "candidateTraceCount": None, "selectedMeta": None, }, None if SEARCH_PROXY_PATH is None: return None, None, "missing trace id and search path" search_rc, search_body, search_err = run_kubectl(SEARCH_PROXY_PATH, timeout=15) search_parsed = parse_json(search_body) metas = extract_traces(search_parsed) candidate_summaries = [] selected = None selected_trace_id = None selected_score = None seen = set() scanned = 0 for meta in metas: trace_id = trace_id_from_meta(meta) if not trace_id or trace_id in seen: continue seen.add(trace_id) if scanned >= CANDIDATE_LIMIT or time.time() > DEADLINE: break scanned += 1 trace_path = TRACE_PATH_TEMPLATE.replace("{{traceId}}", trace_id) trace_rc, trace_body, trace_err = run_kubectl(TRACE_PROXY_PREFIX + trace_path, timeout=8) summary, _parsed = candidate_score(trace_id, meta, trace_body, trace_rc, trace_err) candidate_summaries.append(summary) score = summary.get("score") if isinstance(score, (int, float)) and (selected_score is None or score > selected_score): selected = meta selected_trace_id = trace_id selected_score = score if selected_trace_id is None: for meta in metas: trace_id = trace_id_from_meta(meta) if trace_id: selected = meta selected_trace_id = trace_id selected_score = None break candidate_summaries.sort(key=lambda item: item.get("score", -999999), reverse=True) selected_summary = next((item for item in candidate_summaries if item.get("traceId") == selected_trace_id), None) selected_quality = selected_summary.get("candidateQuality") if isinstance(selected_summary, dict) else None selected_low_confidence = bool(selected_summary.get("lowConfidence")) if isinstance(selected_summary, dict) else False selected_service_path = selected_summary.get("servicePath") if isinstance(selected_summary, dict) else None selected_complete = bool(selected_service_path.get("complete")) if isinstance(selected_service_path, dict) else False selected_rejected_reason = None if selected_low_confidence and not selected_complete: selected_rejected_reason = "best candidate is only a single-span Workbench GET/SSE trace; no high-confidence Code Agent trace was found in the scanned window" mapping = { "mode": SEARCH_MODE, "businessTraceId": BUSINESS_TRACE_ID, "runId": RUN_ID, "commandId": COMMAND_ID, "sessionId": SESSION_ID, "runnerJobId": RUNNER_JOB_ID, "otelTraceId": selected_trace_id, "searchQuery": SEARCH_QUERY, "searchPath": SEARCH_PATH, "searchProxyPath": SEARCH_PROXY_PATH, "searchOk": search_rc == 0, "searchParseOk": isinstance(search_parsed, dict), "candidateTraceCount": len(metas), "scannedCandidateCount": scanned, "candidateSelectionMode": "scored-service-and-span-content", "selectedScore": selected_score, "selectedQuality": selected_quality, "selectedLowConfidence": selected_low_confidence, "selectedRejectedReason": selected_rejected_reason, "selectedReasons": selected_summary.get("reasons", []) if isinstance(selected_summary, dict) else [], "selectedMeta": compact_meta(selected), "candidatePreview": candidate_summaries[:8], "searchStderrTail": (search_err or "")[-2000:], } if RAW: mapping["rawSearchBody"] = search_parsed if isinstance(search_parsed, dict) else search_body[-12000:] if selected_trace_id is None: return None, mapping, "no OTel trace found for diagnosis query" if selected_rejected_reason is not None: return None, mapping, selected_rejected_reason return selected_trace_id, mapping, None resolved_trace_id, mapping, resolve_error = resolve_trace() if resolved_trace_id is None: payload = { "ok": False, "mapping": mapping, "error": resolve_error, "summary": { "rootCause": "otel trace not found", "facts": [], }, "next": { "expandWindow": DIAGNOSE_FULL_COMMAND, "search": SEARCH_COMMAND, }, } print(json.dumps(payload, ensure_ascii=False, indent=2)) raise SystemExit(1) trace_path = TRACE_PATH_TEMPLATE.replace("{{traceId}}", resolved_trace_id) trace_proxy_path = TRACE_PROXY_PREFIX + trace_path trace_rc, trace_body, trace_err = run_kubectl(trace_proxy_path, timeout=25) trace_parsed = parse_json(trace_body) if not isinstance(trace_parsed, dict): payload = { "ok": False, "mapping": mapping, "tracePath": trace_path, "traceProxyPath": trace_proxy_path, "traceQueryOk": trace_rc == 0, "traceParseOk": False, "bodyBytes": len(trace_body.encode("utf-8")), "bodyTail": trace_body[-12000:], "stderrTail": (trace_err or "")[-4000:], } print(json.dumps(payload, ensure_ascii=False, indent=2)) raise SystemExit(1) flat = flatten_trace(trace_parsed) spans = flat["spans"] services = flat["services"] 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) read_model = hwlab_read_model_summary(spans) 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 = { service: ("reached" if service in services else "missing") for service in expected_services } missing_services = [service for service in expected_services if service not in services] service_path["complete"] = len(missing_services) == 0 observability_gap = { "status": "complete" if len(missing_services) == 0 else "missing-service-spans", "expectedServices": expected_services, "seenServices": sorted(services), "missingServices": missing_services, "complete": len(missing_services) == 0, } if missing_services and ("hwlab-cloud-api" in services or identity.get("runId") not in (None, "") or identity.get("commandId") not in (None, "")): candidates.insert(0, { "code": "observability_gap_missing_service_spans", "label": "observability gap", "confidence": 0.76, "summary": "The business trace is correlated to Code Agent context but is missing expected service spans; do not interpret the missing manager/runner spans as proof that those services were not involved.", "evidence": observability_gap, }) facts = [] if missing_services: facts.append("observability gap: missing service spans " + ",".join(missing_services)) if http_summary.get("actorForbidden"): facts.append("actor forbidden") terminal_status = agentrun.get("terminalStatus") if terminal_status in ("failed", "error", "timeout", "blocked", "cancelled"): failure_kind = agentrun.get("failureKind") if failure_kind: facts.append(f"AgentRun terminal failed ({failure_kind})") else: facts.append("AgentRun terminal failed") if terminal_status == "completed": facts.append("AgentRun completed") if lag.get("status") in ("confirmed", "suspected"): facts.append("projection/read-model stale") if idle_warning_spans and terminal_status in (None, ""): facts.append("AgentRun runner idle warnings active") if not facts: facts.append("no dominant Code Agent root cause classified") summary = { "rootCause": "; ".join(facts), "facts": facts, "classification": { "projectionLag": lag.get("status"), "runnerProvider": agentrun.get("runnerProviderClassification"), "actorForbidden": http_summary.get("actorForbidden"), "terminalStatus": terminal_status, "failureKind": agentrun.get("failureKind"), "observabilityGap": observability_gap.get("status"), }, } evidence = { "httpProblemSpanCount": len(http_summary.get("problemSpans", [])), "httpProblemSpanSamples": [tiny_span(item) for item in http_summary.get("problemSpans", [])[:3]], "terminalSpanCount": len(agentrun.get("terminalSpans", [])), "terminalSpanSamples": [tiny_terminal(item) for item in agentrun.get("terminalSpans", [])[:3]], "traceEventReadSpans": [tiny_span(item) for item in read_model.get("traceEventReadSpans", [])[:3]], "turnStatusReadSpanCount": len(read_model.get("turnStatusReadSpans", [])), "turnStatusReadSpanSamples": [tiny_span(item) for item in read_model.get("turnStatusReadSpans", [])[:3]], "projectionSpanCount": len(read_model.get("projectionSpans", [])), "projectionSpanTail": [tiny_span(item) for item in read_model.get("projectionSpans", [])[-3:]], "errorSpanCount": len(error_spans), "errorSpanSampleNames": sorted(set(str(item.get("name") or "") for item in error_spans))[:5], "idleWarningSpanCount": len(idle_warning_spans), "idleWarningSpanTail": [tiny_span(item) for item in idle_warning_spans[-3:]], } payload = { "ok": trace_rc == 0 and len(spans) > 0, "mapping": mapping, "tracePath": trace_path, "traceProxyPath": trace_proxy_path, "bodyBytes": len(trace_body.encode("utf-8")), "traceParseOk": True, "spanCount": len(spans), "services": services, "servicePath": service_path, "observabilityGap": observability_gap, "businessTraceIds": business_trace_ids[:20], "identity": identity, "agentrun": { "terminalStatus": agentrun.get("terminalStatus"), "terminalSource": agentrun.get("terminalSource"), "failureKind": agentrun.get("failureKind"), "observedFailureKinds": agentrun.get("observedFailureKinds"), "latestSeq": agentrun.get("latestSeq"), "terminalEventType": agentrun.get("terminalEventType"), "runnerProviderClassification": agentrun.get("runnerProviderClassification"), "authority": agentrun.get("authority"), }, "hwlabReadModel": { "sourceEventCount": read_model.get("sourceEventCount"), "requestedSinceSeq": read_model.get("requestedSinceSeq"), "latestProjectedSeq": read_model.get("latestProjectedSeq"), "traceStatus": read_model.get("traceStatus"), "turnStatus": read_model.get("turnStatus"), "turnStatusCounts": read_model.get("turnStatusCounts"), }, "http": { "statusCounts": http_summary.get("statusCounts", [])[:20], "problemCounts": http_summary.get("problemCounts", [])[:20], "actorForbidden": http_summary.get("actorForbidden"), }, "projectionLag": lag, "summary": summary, "rootCauseCandidates": candidates, "spanNameCounts": flat["spanNameCounts"][:12], "evidence": evidence, "next": { "diagnoseFull": DIAGNOSE_FULL_COMMAND, "traceSummary": "bun scripts/cli.ts platform-infra observability trace --target ${target.id} --trace-id %s --limit 80" % resolved_trace_id, "traceReads": "bun scripts/cli.ts platform-infra observability trace --target ${target.id} --trace-id %s --grep trace_events_read --limit 20 --full" % resolved_trace_id, "turnStatus": "bun scripts/cli.ts platform-infra observability trace --target ${target.id} --trace-id %s --grep turn_status_read --limit 20 --full" % resolved_trace_id, "projection": "bun scripts/cli.ts platform-infra observability trace --target ${target.id} --trace-id %s --grep projection_sync --limit 120 --full" % resolved_trace_id, "runnerTerminal": "bun scripts/cli.ts platform-infra observability trace --target ${target.id} --trace-id %s --grep runner_terminal --limit 20 --full" % resolved_trace_id, "raw": "bun scripts/cli.ts platform-infra observability diagnose-code-agent --target ${target.id} --trace-id %s --raw" % resolved_trace_id, }, "stderrTail": (trace_err or "")[-4000:], } if FULL: payload["full"] = { "traceEventReadSpans": read_model.get("traceEventReadSpans", []), "turnStatusReadSpans": read_model.get("turnStatusReadSpans", []), "projectionSpans": read_model.get("projectionSpans", [])[:LIMIT], "terminalSpans": agentrun.get("terminalSpans", [])[:LIMIT], "errorSpans": [public_span(item) for item in error_spans[:LIMIT]], "httpProblemSpans": [public_span(item) for item in http_summary.get("problemSpans", [])[:LIMIT]], "selectedSpans": [public_span(item) for item in spans if item.get("name") in ("trace_events_read", "turn_status_read", "projection_sync", "command_result")][:LIMIT], } if RAW: payload["raw"] = raw_trace_shape(trace_parsed) print(json.dumps(payload, ensure_ascii=False, indent=2 if FULL or RAW else None)) raise SystemExit(0 if payload["ok"] else 1) PY `; }