mirrors/pikasTech-unidesk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
6222744119e52efdbd086a07282ea8357301b571
pikasTech-unidesk/scripts/src/platform-infra-observability/diagnose-code-agent-script.ts
T

2432 lines
110 KiB
TypeScript
Raw Blame History

// 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",
"providerProfile", "model", "modelId", "providerModel",
"code_agent.stage", "agent.chat.session_id",
"agent.chat.provider_profile", "agent.chat.provider_profile_source",
"agent.chat.model", "agent.chat.model_id", "agent.chat.provider_model",
"defaultProviderProfile", "adapter", "adapterEnabled",
"agentRunManagerHost", "agentRunRunnerNamespace",
"agentRunSourceCommitPresent",
"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 "<array>"
if "kvlistValue" in inner:
return "<kvlist>"
if "bytesValue" in inner:
return "<bytes>"
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 "<unknown-service>"
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 ?? "<traceId>"} --full --limit 200",
"grepProviderStream": "bun scripts/cli.ts platform-infra observability trace --target ${target.id} --trace-id ${options.traceId ?? "<traceId>"} --grep provider-stream-disconnected --limit 40",
"raw": "bun scripts/cli.ts platform-infra observability trace --target ${target.id} --trace-id ${options.traceId ?? "<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",
"providerProfile", "model", "modelId", "providerModel",
"code_agent.stage", "agent.chat.session_id",
"agent.chat.provider_profile", "agent.chat.provider_profile_source",
"agent.chat.model", "agent.chat.model_id", "agent.chat.provider_model",
"defaultProviderProfile", "adapter", "adapterEnabled",
"agentRunManagerHost", "agentRunRunnerNamespace",
"agentRunSourceCommitPresent",
"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 "<array>"
if "kvlistValue" in inner:
return "<kvlist>"
if "bytesValue" in inner:
return "<bytes>"
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 "<unknown-service>"
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 "<text>"),
"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 <text> --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 <traceId> --grep <text> --limit 80",
"raw": "bun scripts/cli.ts platform-infra observability search --target ${target.id} --grep <text> --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 = \"<trc_...>\" }";
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",
"providerProfile", "model", "modelId", "providerModel",
"code_agent.stage", "agent.chat.session_id",
"agent.chat.provider_profile", "agent.chat.provider_profile_source",
"agent.chat.model", "agent.chat.model_id", "agent.chat.provider_model",
"defaultProviderProfile", "adapter", "adapterEnabled",
"agentRunManagerHost", "agentRunRunnerNamespace",
"agentRunSourceCommitPresent",
"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 "<array>"
if "kvlistValue" in inner:
return "<kvlist>"
if "bytesValue" in inner:
return "<bytes>"
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",
"providerProfile", "model", "modelId", "providerModel",
"code_agent.stage", "agent.chat.session_id",
"agent.chat.provider_profile", "agent.chat.provider_profile_source",
"agent.chat.model", "agent.chat.model_id", "agent.chat.provider_model",
"defaultProviderProfile", "adapter", "adapterEnabled",
"agentRunManagerHost", "agentRunRunnerNamespace",
"agentRunSourceCommitPresent",
):
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 attr_first(attrs, keys):
if not isinstance(attrs, dict):
return None
for key in keys:
value = attrs.get(key)
if value not in (None, ""):
return value
return None
def provider_decision_summary(spans):
rows = []
for item in spans:
attrs = item.get("attributes", {}) if isinstance(item.get("attributes"), dict) else {}
name = str(item.get("name") or "")
if name != "provider_decision" and attr_first(attrs, ("agent.chat.provider_profile", "providerProfile", "defaultProviderProfile")) in (None, ""):
continue
row = {
"span": name or None,
"service": item.get("service"),
"traceId": attr_first(attrs, ("traceId", "workbench.trace_id", "workbench.turn_id")),
"sessionId": attr_first(attrs, ("agent.chat.session_id", "sessionId", "workbench.session_id")),
"providerProfile": attr_first(attrs, ("agent.chat.provider_profile", "providerProfile", "backendProfile", "defaultProviderProfile")),
"providerProfileSource": attr_first(attrs, ("agent.chat.provider_profile_source", "providerProfileSource")),
"defaultProviderProfile": attr_first(attrs, ("defaultProviderProfile",)),
"model": attr_first(attrs, ("agent.chat.model", "agent.chat.model_id", "agent.chat.provider_model", "model", "modelId", "providerModel")),
"adapter": attr_first(attrs, ("adapter",)),
"adapterEnabled": attr_first(attrs, ("adapterEnabled",)),
"managerHost": attr_first(attrs, ("agentRunManagerHost",)),
"runnerNamespace": attr_first(attrs, ("agentRunRunnerNamespace",)),
"sourceCommitPresent": attr_first(attrs, ("agentRunSourceCommitPresent",)),
}
compact = {key: value for key, value in row.items() if value not in (None, "")}
if compact:
rows.append(compact)
primary = rows[0] if rows else {}
return {
"providerProfile": primary.get("providerProfile"),
"providerProfileSource": primary.get("providerProfileSource"),
"defaultProviderProfile": primary.get("defaultProviderProfile"),
"model": primary.get("model"),
"adapter": primary.get("adapter"),
"adapterEnabled": primary.get("adapterEnabled"),
"managerHost": primary.get("managerHost"),
"runnerNamespace": primary.get("runnerNamespace"),
"sourceCommitPresent": primary.get("sourceCommitPresent"),
"spanCount": len(rows),
"decisions": rows[:5],
}
def span_business_trace_id(item):
attrs = item.get("attributes", {}) if isinstance(item.get("attributes"), dict) else {}
for key in ("traceId", "workbench.trace_id", "workbench.turn_id"):
value = attrs.get(key)
if isinstance(value, str) and value.strip().startswith("trc_"):
return value.strip()
return None
def span_identity_value(item, key):
attrs = item.get("attributes", {}) if isinstance(item.get("attributes"), dict) else {}
value = attrs.get(key)
if value in (None, ""):
return None
return str(value)
def span_matches_scope_identity(item, identity):
if not isinstance(identity, dict):
return False
# Only strong per-turn identities are safe enough for no-trace-id spans.
# runId/sessionId/runnerJobId can cover several turns in the same Workbench
# session and caused terminal status leakage across business trace ids.
for key in ("commandId", "turnId"):
scoped = identity.get(key)
value = span_identity_value(item, key)
if scoped not in (None, "") and value not in (None, ""):
return str(scoped) == str(value)
return False
def scoped_spans_for_business_trace(spans, business_trace_id):
if business_trace_id in (None, ""):
return list(spans), {
"mode": "unscoped-no-business-trace-id",
"requestedBusinessTraceId": None,
"totalSpanCount": len(spans),
"scopedSpanCount": len(spans),
"excludedDifferentTraceSpanCount": 0,
"includedNoTraceIdentitySpanCount": 0,
"crossBusinessTraceIds": [],
}
exact = []
no_trace = []
cross_ids = collections.Counter()
for item in spans:
trace_id = span_business_trace_id(item)
if trace_id == business_trace_id:
exact.append(item)
elif trace_id:
cross_ids[trace_id] += 1
else:
no_trace.append(item)
if not exact:
return list(spans), {
"mode": "unscoped-no-exact-business-trace-span",
"requestedBusinessTraceId": business_trace_id,
"totalSpanCount": len(spans),
"scopedSpanCount": len(spans),
"excludedDifferentTraceSpanCount": 0,
"includedNoTraceIdentitySpanCount": 0,
"crossBusinessTraceIds": [{"traceId": key, "count": count} for key, count in cross_ids.most_common(8)],
}
identity = identity_from_spans(exact)
exact_ids = set(id(item) for item in exact)
scoped = list(exact)
included_no_trace = 0
for item in no_trace:
if id(item) in exact_ids:
continue
if span_matches_scope_identity(item, identity):
scoped.append(item)
included_no_trace += 1
scoped.sort(key=lambda item: (item.get("_start", 0), item.get("_index", 0)))
return scoped, {
"mode": "business-trace-scoped",
"requestedBusinessTraceId": business_trace_id,
"totalSpanCount": len(spans),
"scopedSpanCount": len(scoped),
"excludedDifferentTraceSpanCount": len(spans) - len(scoped),
"includedNoTraceIdentitySpanCount": included_no_trace,
"scopeIdentity": {key: identity.get(key) for key in ("commandId", "turnId", "runId", "sessionId") if identity.get(key) not in (None, "")},
"crossBusinessTraceIds": [{"traceId": key, "count": count} for key, count in cross_ids.most_common(8)],
}
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 "<unknown-service>"
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 "<unknown-route>"
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 = []
session_list_spans = []
session_message_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 == "session_list_read":
session_list_spans.append(public_span(item))
if name == "session_messages_read":
session_message_spans.append(public_span(item))
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,
"sessionListReadCount": len(session_list_spans),
"sessionMessageReadCount": len(session_message_spans),
"sessionListReadSpans": session_list_spans,
"sessionMessageReadSpans": session_message_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 "")
read_model_turn_status = read_model.get("turnStatus")
status_conflict = terminal_status_conflicts(agentrun.get("terminalStatus"), read_model_turn_status)
if status_conflict:
candidates.append({
"code": "otel_business_trace_terminal_conflict",
"label": "OTel terminal conflict",
"confidence": 0.86,
"summary": "Scoped AgentRun terminal status conflicts with HWLAB read-model turn status for the requested business trace; prefer the HWLAB read-model status for Workbench RCA and inspect OTel correlation leakage.",
"evidence": {
"agentrunTerminalStatus": agentrun.get("terminalStatus"),
"hwlabReadModelTurnStatus": read_model_turn_status,
"turnStatusCounts": read_model.get("turnStatusCounts"),
},
})
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") and not status_conflict:
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 agentrun.get("terminalStatus") in (None, "") and read_model_turn_status in ("completed", "failed", "error", "timeout", "blocked", "cancelled", "canceled"):
candidates.append({
"code": "hwlab_read_model_terminal_without_runner_spans",
"label": "HWLAB read model terminal",
"confidence": 0.7,
"summary": "HWLAB read model shows a terminal turn status but scoped AgentRun manager/runner terminal spans are absent; treat missing runner spans as an observability gap, not proof that the turn is still running.",
"evidence": {
"turnStatus": read_model_turn_status,
"turnStatusCounts": read_model.get("turnStatusCounts"),
"sessionListReadCount": read_model.get("sessionListReadCount"),
"sessionMessageReadCount": read_model.get("sessionMessageReadCount"),
},
})
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 comparable_terminal_status(value):
text = str(value or "").strip().lower().replace("_", "-")
if not text:
return None
if text == "cancelled":
return "canceled"
if text == "success" or text == "succeeded" or text == "complete":
return "completed"
if text == "error" or text == "failure":
return "failed"
return text
def terminal_status_conflicts(left, right):
left_status = comparable_terminal_status(left)
right_status = comparable_terminal_status(right)
if left_status in (None, "") or right_status in (None, ""):
return False
terminal_values = {"completed", "failed", "timeout", "blocked", "canceled", "stale", "thread-resume-failed", "interrupted", "expired"}
if left_status not in terminal_values or right_status not in terminal_values:
return False
return left_status != right_status
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)
all_spans = flat["spans"]
spans, business_scope = scoped_spans_for_business_trace(all_spans, BUSINESS_TRACE_ID)
services = set(str(item.get("service") or "") for item in spans)
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)
scoped_span_ids = set(id(item) for item in spans)
error_spans = [item for item in flat["errorSpans"] if id(item) in scoped_span_ids]
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),
"businessTraceScope": business_scope,
"providerDecision": provider_decision_summary(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)
all_spans = flat["spans"]
spans, business_scope = scoped_spans_for_business_trace(all_spans, BUSINESS_TRACE_ID)
services = sorted(set(str(item.get("service") or "") for item in spans))
business_trace_ids = flat["businessTraceIds"]
scoped_span_ids = set(id(item) for item in spans)
error_spans = [item for item in flat["errorSpans"] if id(item) in scoped_span_ids]
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)
provider_decision = provider_decision_summary(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 business_scope.get("mode") == "business-trace-scoped" and business_scope.get("crossBusinessTraceIds"):
facts.append("cross-business-trace spans excluded")
if http_summary.get("actorForbidden"):
facts.append("actor forbidden")
terminal_status = agentrun.get("terminalStatus")
read_model_turn_status = read_model.get("turnStatus")
agentrun_read_model_status_conflict = terminal_status_conflicts(terminal_status, read_model_turn_status)
if agentrun_read_model_status_conflict:
facts.append("OTel AgentRun terminal status conflicts with HWLAB read model")
if terminal_status in ("failed", "error", "timeout", "blocked", "cancelled") and not agentrun_read_model_status_conflict:
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" and not agentrun_read_model_status_conflict:
facts.append("AgentRun completed")
if (terminal_status in (None, "") or agentrun_read_model_status_conflict) and read_model_turn_status in ("completed", "failed", "error", "timeout", "blocked", "cancelled", "canceled"):
facts.append("HWLAB read model terminal " + str(read_model_turn_status))
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,
"readModelTurnStatus": read_model_turn_status,
"agentrunReadModelStatusConflict": agentrun_read_model_status_conflict,
"failureKind": agentrun.get("failureKind"),
"observabilityGap": observability_gap.get("status"),
"businessTraceScope": business_scope.get("mode"),
"providerProfile": provider_decision.get("providerProfile"),
"providerModel": provider_decision.get("model"),
"providerAdapter": provider_decision.get("adapter"),
},
}
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],
"businessTraceScope": business_scope,
"providerDecision": provider_decision,
"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"),
"sessionListReadCount": read_model.get("sessionListReadCount"),
"sessionMessageReadCount": read_model.get("sessionMessageReadCount"),
},
"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,
"sessionReads": "bun scripts/cli.ts platform-infra observability trace --target ${target.id} --trace-id %s --grep session_ --limit 60 --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", []),
"sessionListReadSpans": read_model.get("sessionListReadSpans", []),
"sessionMessageReadSpans": read_model.get("sessionMessageReadSpans", []),
"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", "session_list_read", "session_messages_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
`;
}
Reference in New Issue View Git Blame Copy Permalink