// SPEC: PJ2026-01060307 控制面模块化 draft-2026-06-25-p0. diagnose-code-agent-script module for scripts/src/platform-infra-observability.ts.

// Moved mechanically from scripts/src/platform-infra-observability.ts:2137-3030 for #903.

// SPEC: PJ2026-01060501 OTel追踪 draft-2026-06-19-p0.
// Responsibility: YAML-first platform-infra OpenTelemetry tracing control commands.
import { Buffer } from "node:buffer";
import { readFileSync } from "node:fs";
import type { UniDeskConfig } from "../config";
import { rootPath } from "../config";
import type { RenderedCliResult } from "../output";
import {
  compactCapture,
  createYamlFieldReader,
  numberField,
  parseJsonOutput,
  redactSensitiveUnknown,
  shQuote,
  capture,
} from "../platform-infra-ops-library";

import type { DiagnoseCodeAgentOptions, ObservabilityConfig, ObservabilityTarget, SearchOptions, TraceOptions } from "./types";
import { inferSearchTempoQuery } from "./render";

export function traceScript(observability: ObservabilityConfig, target: ObservabilityTarget, tracePath: string, options: TraceOptions): string {
  const proxyPath = `/api/v1/namespaces/${target.namespace}/services/http:${observability.traceBackend.serviceName}:http/proxy${tracePath}`;
  const grepLiteral = options.grep === null ? "None" : JSON.stringify(options.grep);
  return `
set -u
tmp="$(mktemp -d)"
trap 'rm -rf "$tmp"' EXIT
kubectl get --raw ${shQuote(proxyPath)} >"$tmp/trace.out" 2>"$tmp/trace.err"
rc=$?
python3 - "$rc" "$tmp/trace.out" "$tmp/trace.err" <<'PY'
import collections, json, sys
FULL = ${options.full ? "True" : "False"}
RAW = ${options.raw ? "True" : "False"}
GREP = ${grepLiteral}
LIMIT = ${options.limit}
IMPORTANT_ATTRS = [
    "traceId", "otel.trace_id", "agentrun.stage", "runId", "commandId",
    "sessionId", "turnId", "threadId", "runnerJobId", "failureKind", "willRetry",
    "retryAttempt", "retryMax", "retryExhausted", "retryBackoffMs",
    "idleMs", "idleSeconds", "idleThresholdMs", "idleThresholdSeconds",
    "waitingFor", "lastEventLabel", "lastActivityAt", "lastActivityMs",
    "lastNotificationAt", "lastNotificationMs", "lastToolCallAt",
    "upstreamHttpStatus", "upstreamHost", "providerErrorClass", "errorSummary",
    "terminalStatus", "phase", "message", "http.route", "http.status_code",
    "http.response.status_code", "http.method", "http.request.method",
    "http.response.status_class", "workbench.ui.event_type",
    "workbench.ui.scope", "workbench.ui.state", "workbench.ui.reason",
    "workbench.ui.route", "workbench.ui.outcome", "workbench.ui.value_ms",
    "workbench.ui.visibility", "workbench.ui.event_name",
    "workbench.ui.activity_idle_ms", "workbench.ui.activity_waiting_for",
    "workbench.ui.activity_last_event_label",
    "workbench.ui.resource_duration_ms",
    "workbench.ui.resource_fetch_to_request_ms",
    "workbench.ui.resource_request_wait_ms",
    "workbench.ui.resource_response_transfer_ms",
    "workbench.ui.resource_transfer_size",
    "workbench.ui.resource_encoded_body_size",
    "workbench.ui.resource_decoded_body_size",
    "workbench.ui.resource_next_hop_protocol",
    "workbench.ui.resource_server_timing",
    "workbench.session_id", "workbench.trace_id", "workbench.turn_id",
    "workbench.read_model.route", "workbench.read_model.count",
    "workbench.read_model.family", "workbench.read_model.status",
    "workbench.read_model.reason",
    "workbench.projection.phase", "workbench.projection.session_id",
    "workbench.projection.turn_id", "workbench.projection.projected_seq",
    "workbench.projection.source_seq", "workbench.projection.source_event_id",
    "workbench.projection.event_type", "workbench.projection.message_id",
    "workbench.projection.terminal", "workbench.projection.suppressed_after_seal",
    "workbench.projection.terminal_trace_backfill_written",
    "source", "sessionCount", "fallbackTitleCount", "fallbackTitleRatio",
    "emptyPreviewCount", "includeSessionId", "sessionIds", "traceIds",
    "returnedMessages", "totalMessages", "roleSequencePrefix",
    "consecutiveUserPrefix", "adjacentSameRoleCount", "userCount",
    "agentCount",
    "db.system", "db.operation.name", "db.sql.table", "db.query.arg_count",
    "db.index.expected", "db.pool.max_open", "db.pool.open_connections",
    "db.pool.in_use", "db.pool.idle", "db.pool.wait_count",
    "db.pool.wait_duration_ms", "db.pool.max_idle_closed",
    "db.pool.max_lifetime_closed",
    "hwlab.runtime.stage", "hwlab.http.stage", "hwlab.http.phase",
    "hwlab.http.phase.outcome", "hwlab.live_builds.service_id",
    "hwlab.live_builds.service_kind", "hwlab.live_builds.external",
    "hwlab.live_builds.deploy_manifest_status",
    "hwlab.live_builds.artifact_catalog_status",
    "error.code", "error.category", "error.layer",
    "stage", "causeStage", "causeCode",
    "eventType", "runnerId", "attemptId", "backendProfile",
    "sourceCommit", "jobName", "podName", "logPath",
    "toolName", "type", "itemType", "itemId", "status", "exitCode",
    "durationMs", "cwd", "processId", "command", "commandFingerprint",
    "outputSummary", "outputBytes", "outputTruncated", "storageKind",
    "storagePvcName", "storagePvcPhase", "resumeMode", "rolloutId",
    "returnedEvents", "sinceSeq", "afterSeq", "limit", "fromSeq", "toSeq",
    "totalEvents", "hasMore", "fullTraceLoaded", "rawEventCount",
    "maxSeq", "traceLastSeq", "endSeq", "commandFiltered",
    "error", "error.message", "exception.type", "exception.message",
    "valuesPrinted",
]
IMPORTANT_NAMES = {
    "durable_admission", "billing_preflight", "agentrun_dispatch",
    "projection_write", "trace_events_read", "turn_status_read",
    "run_created", "runner_job_created", "runner_job_status",
    "command_result", "projection_sync",
}
def text(path, limit=12000):
    try:
        return open(path, encoding="utf-8", errors="replace").read()[-limit:]
    except FileNotFoundError:
        return ""
def read_all(path):
    try:
        return open(path, encoding="utf-8", errors="replace").read()
    except FileNotFoundError:
        return ""
def attr_value(value):
    if not isinstance(value, dict):
        return value
    inner = value.get("value", value)
    if not isinstance(inner, dict):
        return inner
    for key in ("stringValue", "intValue", "doubleValue", "boolValue"):
        if key in inner:
            return inner.get(key)
    if "arrayValue" in inner:
        return "<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 = 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",
    "http.target", "http.url", "url.path",
    "db.system", "db.operation.name", "db.sql.table", "db.query.arg_count",
    "db.index.expected", "db.pool.max_open", "db.pool.open_connections",
    "db.pool.in_use", "db.pool.idle", "db.pool.wait_count",
    "db.pool.wait_duration_ms", "db.pool.max_idle_closed",
    "db.pool.max_lifetime_closed",
    "hwlab.runtime.stage", "hwlab.http.stage", "hwlab.http.phase",
    "hwlab.http.phase.outcome", "hwlab.live_builds.service_id",
    "hwlab.live_builds.service_kind", "hwlab.live_builds.external",
    "hwlab.live_builds.deploy_manifest_status",
    "hwlab.live_builds.artifact_catalog_status",
    "error.code", "error.category", "error.layer",
    "stage", "causeStage", "causeCode",
    "eventType", "runnerId", "attemptId", "backendProfile",
    "sourceCommit", "jobName", "podName", "logPath",
    "toolName", "type", "itemType", "itemId", "status", "exitCode",
    "durationMs", "cwd", "processId", "command", "commandFingerprint",
    "outputSummary", "outputBytes", "outputTruncated", "storageKind",
    "storagePvcName", "storagePvcPhase", "resumeMode", "rolloutId",
    "returnedEvents", "sinceSeq", "afterSeq", "limit", "fromSeq", "toSeq",
    "totalEvents", "hasMore", "fullTraceLoaded", "rawEventCount",
    "maxSeq", "traceLastSeq", "endSeq", "commandFiltered",
    "error", "error.message", "exception.type", "exception.message",
    "valuesPrinted",
]

def run_kubectl(proxy_path, timeout=10):
    try:
        proc = subprocess.run(
            ["kubectl", "get", "--raw", proxy_path],
            capture_output=True,
            text=True,
            timeout=timeout,
        )
        return proc.returncode, proc.stdout, proc.stderr
    except subprocess.TimeoutExpired as exc:
        return 124, exc.stdout or "", "timeout while querying tempo"
    except Exception as exc:
        return 125, "", str(exc)

def parse_json(body):
    try:
        return json.loads(body) if body else None
    except Exception:
        return None

def attr_value(value):
    if not isinstance(value, dict):
        return value
    inner = value.get("value", value)
    if not isinstance(inner, dict):
        return inner
    for key in ("stringValue", "intValue", "doubleValue", "boolValue"):
        if key in inner:
            return inner.get(key)
    if "arrayValue" in inner:
        return "<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 grep_matches_item(item):
    return GREP is not None and 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
                item = compact_span(span, service, resource_attrs, scope_name)
                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(item) and (GREP is None or grep_matches_item(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,
    "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.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.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.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 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}
RUNNER_JOB_ID = ${runnerJobIdLiteral}
TARGET_ID = ${JSON.stringify(target.id)}
SEARCH_MODE = ${JSON.stringify(searchMode)}
SEARCH_QUERY = ${tempoQuery === null ? "None" : JSON.stringify(tempoQuery)}
SEARCH_PATH = ${searchPathLiteral}
SEARCH_PROXY_PATH = ${searchProxyPathLiteral}
TRACE_PROXY_PREFIX = ${JSON.stringify(proxyPrefix)}
TRACE_PATH_TEMPLATE = ${JSON.stringify(observability.probes.traceQueryPathTemplate)}
DIAGNOSE_FULL_COMMAND = ${JSON.stringify(diagnoseFullCommand)}
SEARCH_COMMAND = ${JSON.stringify(searchCommand)}
FULL = ${options.full ? "True" : "False"}
RAW = ${options.raw ? "True" : "False"}
LIMIT = ${options.limit}
CANDIDATE_LIMIT = ${options.candidateLimit}
DEADLINE = time.time() + 50
IMPORTANT_ATTRS = [
    "traceId", "otel.trace_id", "agentrun.stage", "runId", "commandId",
    "sessionId", "turnId", "threadId", "runnerJobId", "failureKind", "willRetry",
    "retryAttempt", "retryMax", "retryExhausted", "retryBackoffMs",
    "idleMs", "idleSeconds", "idleThresholdMs", "idleThresholdSeconds",
    "waitingFor", "lastEventLabel", "lastActivityAt", "lastActivityMs",
    "lastNotificationAt", "lastNotificationMs", "lastToolCallAt",
    "upstreamHttpStatus", "upstreamHost", "providerErrorClass", "errorSummary",
    "terminalStatus", "phase", "message", "http.route", "http.status_code",
    "http.response.status_code", "http.method", "http.request.method",
    "http.response.status_class", "workbench.ui.event_type",
    "workbench.ui.scope", "workbench.ui.state", "workbench.ui.reason",
    "workbench.ui.route", "workbench.ui.outcome", "workbench.ui.value_ms",
    "workbench.ui.visibility", "workbench.ui.event_name",
    "workbench.ui.activity_idle_ms", "workbench.ui.activity_waiting_for",
    "workbench.ui.activity_last_event_label",
    "workbench.ui.resource_duration_ms",
    "workbench.ui.resource_fetch_to_request_ms",
    "workbench.ui.resource_request_wait_ms",
    "workbench.ui.resource_response_transfer_ms",
    "workbench.ui.resource_transfer_size",
    "workbench.ui.resource_encoded_body_size",
    "workbench.ui.resource_decoded_body_size",
    "workbench.ui.resource_next_hop_protocol",
    "workbench.ui.resource_server_timing",
    "workbench.session_id", "workbench.trace_id", "workbench.turn_id",
    "workbench.read_model.route", "workbench.read_model.count",
    "workbench.read_model.family", "workbench.read_model.status",
    "workbench.read_model.reason",
    "workbench.projection.phase", "workbench.projection.session_id",
    "workbench.projection.turn_id", "workbench.projection.projected_seq",
    "workbench.projection.source_seq", "workbench.projection.source_event_id",
    "workbench.projection.event_type", "workbench.projection.message_id",
    "workbench.projection.terminal", "workbench.projection.suppressed_after_seal",
    "workbench.projection.terminal_trace_backfill_written",
    "source", "sessionCount", "fallbackTitleCount", "fallbackTitleRatio",
    "emptyPreviewCount", "includeSessionId", "sessionIds", "traceIds",
    "returnedMessages", "totalMessages", "roleSequencePrefix",
    "consecutiveUserPrefix", "adjacentSameRoleCount", "userCount",
    "agentCount",
    "db.system", "db.operation.name", "db.sql.table", "db.query.arg_count",
    "db.index.expected", "db.pool.max_open", "db.pool.open_connections",
    "db.pool.in_use", "db.pool.idle", "db.pool.wait_count",
    "db.pool.wait_duration_ms", "db.pool.max_idle_closed",
    "db.pool.max_lifetime_closed",
    "hwlab.runtime.stage", "hwlab.http.stage", "hwlab.http.phase",
    "hwlab.http.phase.outcome", "hwlab.live_builds.service_id",
    "hwlab.live_builds.service_kind", "hwlab.live_builds.external",
    "hwlab.live_builds.deploy_manifest_status",
    "hwlab.live_builds.artifact_catalog_status",
    "error.code", "error.category", "error.layer",
    "stage", "causeStage", "causeCode",
    "eventType", "runnerId", "attemptId", "backendProfile",
    "sourceCommit", "jobName", "podName", "logPath",
    "toolName", "type", "itemType", "itemId", "status", "exitCode",
    "durationMs", "cwd", "processId", "command", "commandFingerprint",
    "outputSummary", "outputBytes", "outputTruncated", "storageKind",
    "storagePvcName", "storagePvcPhase", "resumeMode", "rolloutId",
    "returnedEvents", "sinceSeq", "afterSeq", "limit", "fromSeq", "toSeq",
    "totalEvents", "hasMore", "fullTraceLoaded", "rawEventCount",
    "maxSeq", "traceLastSeq", "endSeq", "commandFiltered",
    "seq", "eventSeq", "sourceSeq", "sourceLatestSeq", "latestSeq",
    "lastSeq", "sourceEventCount", "projectedSeq", "lastProjectedSeq",
    "status", "turnStatus", "error", "error.message", "exception.type",
    "exception.message", "valuesPrinted",
]
IDENTITY_KEYS = [
    "runId", "commandId", "sessionId", "turnId", "threadId",
    "runnerJobId", "runnerId", "attemptId", "backendProfile",
    "sourceCommit", "jobName", "podName",
]

def run_kubectl(proxy_path, timeout=15):
    try:
        proc = subprocess.run(
            ["kubectl", "get", "--raw", proxy_path],
            capture_output=True,
            text=True,
            timeout=timeout,
        )
        return proc.returncode, proc.stdout, proc.stderr
    except subprocess.TimeoutExpired as exc:
        return 124, exc.stdout or "", "timeout while querying tempo"
    except Exception as exc:
        return 125, "", str(exc)

def run_kubectl_json(proxy_path, timeout=15):
    rc, body, stderr = run_kubectl(proxy_path, timeout=timeout)
    parsed = parse_json(body)
    return {
        "ok": rc == 0 and isinstance(parsed, dict),
        "rc": rc,
        "json": parsed if isinstance(parsed, dict) else None,
        "stderrTail": (stderr or "")[-1000:],
        "bodyBytes": len((body or "").encode("utf-8")),
    }

def first_present(*values):
    for value in values:
        if value not in (None, ""):
            return value
    return None

def shell_quote(value):
    return "'" + str(value).replace("'", "'\\''") + "'"

def unwrap_payload(value):
    if not isinstance(value, dict):
        return {}
    data = value.get("data")
    if isinstance(data, dict):
        return data
    result = value.get("result")
    if isinstance(result, dict):
        return result
    return value

def guess_agentrun_namespace(identity):
    if isinstance(identity, dict):
        for key in ("jobName", "podName", "runnerJobId"):
            text = str(identity.get(key) or "")
            match = re.search(r"\\bagentrun-v[0-9]+\\b", text)
            if match:
                return match.group(0)
    if str(TARGET_ID).upper() == "D601":
        return "agentrun-v02"
    return "agentrun-v01"

def agentrun_service_proxy_paths(namespace, api_path):
    return [
        "/api/v1/namespaces/%s/services/http:agentrun-mgr:http/proxy%s" % (namespace, api_path),
        "/api/v1/namespaces/%s/services/http:agentrun-mgr:8080/proxy%s" % (namespace, api_path),
    ]

def run_agentrun_mgr_json(namespace, api_path, timeout=12):
    url = "http://127.0.0.1:8080%s" % api_path
    script_parts = []
    script_parts.append("url=" + shell_quote(url) + "; ")
    script_parts.append("api_key=\\\"\${AGENTRUN_API_KEY:-}\\\"; ")
    script_parts.append("if [ -z \\\"$api_key\\\" ]; then api_key=\\\"\${HWLAB_API_KEY:-}\\\"; fi; ")
    script_parts.append("if command -v wget >/dev/null 2>&1; then ")
    script_parts.append("wget -qO- --header=\\\"Authorization: Bearer $api_key\\\" \\\"$url\\\"; ")
    script_parts.append("else ")
    script_parts.append("curl -fsS -H \\\"Authorization: Bearer $api_key\\\" \\\"$url\\\"; ")
    script_parts.append("fi")
    script = "".join(script_parts)
    try:
        proc = subprocess.run(
            ["kubectl", "-n", namespace, "exec", "deploy/agentrun-mgr", "--", "sh", "-lc", script],
            capture_output=True,
            text=True,
            timeout=timeout,
        )
        parsed = parse_json(proc.stdout)
        return {
            "ok": proc.returncode == 0 and isinstance(parsed, dict),
            "rc": proc.returncode,
            "json": parsed if isinstance(parsed, dict) else None,
            "stderrTail": (proc.stderr or "")[-1000:],
            "bodyBytes": len((proc.stdout or "").encode("utf-8")),
            "method": "manager-exec-localhost",
        }
    except subprocess.TimeoutExpired as exc:
        return {"ok": False, "rc": 124, "json": None, "stderrTail": "timeout while querying agentrun manager", "bodyBytes": len((exc.stdout or "").encode("utf-8")), "method": "manager-exec-localhost"}
    except Exception as exc:
        return {"ok": False, "rc": 125, "json": None, "stderrTail": str(exc), "bodyBytes": 0, "method": "manager-exec-localhost"}

def get_agentrun_json(namespace, api_path, timeout=10):
    attempts = []
    exec_response = run_agentrun_mgr_json(namespace, api_path, timeout=timeout)
    attempts.append(exec_response)
    if exec_response.get("ok"):
        return exec_response, attempts
    for proxy_path in agentrun_service_proxy_paths(namespace, api_path):
        response = run_kubectl_json(proxy_path, timeout=timeout)
        response["proxyPath"] = proxy_path
        response["method"] = "service-proxy"
        attempts.append(response)
        if response.get("ok"):
            return response, attempts
    return attempts[-1] if attempts else {"ok": False, "json": None}, attempts

def event_items(value):
    if not isinstance(value, dict):
        return []
    for key in ("items", "events", "data"):
        items = value.get(key)
        if isinstance(items, list):
            return [item for item in items if isinstance(item, dict)]
    return []

def terminal_from_events(events):
    latest = None
    for item in events:
        payload = item.get("payload") if isinstance(item.get("payload"), dict) else {}
        event_type = item.get("type") or item.get("eventType") or payload.get("type")
        status = first_present(payload.get("terminalStatus"), payload.get("status"), item.get("terminalStatus"), item.get("status"))
        if event_type == "terminal_status" or status in ("completed", "failed", "blocked", "cancelled", "canceled"):
            latest = {
                "seq": item.get("seq"),
                "type": event_type,
                "status": status,
                "failureKind": first_present(payload.get("failureKind"), item.get("failureKind")),
            }
    return latest

def classify_terminal_status(status, spans):
    if status in ("completed", "succeeded", "success"):
        return "completed"
    if status in ("failed", "error", "timeout", "cancelled", "canceled"):
        return "failed"
    if status == "blocked":
        return "blocked"
    if any(str(item.get("service") or "") == "agentrun-runner" for item in spans):
        return "running"
    return "unknown"

def preferred_failure_kind(values):
    kinds = []
    for value in values:
        if value in (None, ""):
            continue
        text = str(value).strip()
        if not text:
            continue
        if text not in kinds:
            kinds.append(text)
    if not kinds:
        return None
    for kind in kinds:
        lowered = kind.lower()
        if lowered.startswith("provider-") or lowered.startswith("provider_") or "provider-auth" in lowered:
            return kind
    for kind in kinds:
        lowered = kind.lower()
        if lowered not in ("infra-failed", "infra_failed", "unknown"):
            return kind
    return kinds[-1]

def agentrun_authority_summary(identity):
    run_id = identity.get("runId") if isinstance(identity, dict) else None
    command_id = identity.get("commandId") if isinstance(identity, dict) else None
    if not run_id or not command_id:
        return {
            "queried": False,
            "reason": "missing-run-or-command-id",
        }
    namespace = guess_agentrun_namespace(identity)
    run_q = urllib.parse.quote(str(run_id), safe="")
    command_q = urllib.parse.quote(str(command_id), safe="")
    command_response, command_attempts = get_agentrun_json(namespace, "/api/v1/runs/%s/commands/%s" % (run_q, command_q), timeout=8)
    result_response, result_attempts = get_agentrun_json(namespace, "/api/v1/runs/%s/commands/%s/result" % (run_q, command_q), timeout=12)
    events_response, events_attempts = get_agentrun_json(namespace, "/api/v1/runs/%s/events?afterSeq=0&limit=1000" % run_q, timeout=12)
    command_payload = unwrap_payload(command_response.get("json"))
    result_payload = unwrap_payload(result_response.get("json"))
    events = event_items(events_response.get("json"))
    terminal_event = terminal_from_events(events)
    command_state = first_present(command_payload.get("state"), command_payload.get("status"))
    result_status = first_present(
        result_payload.get("terminalStatus"),
        result_payload.get("terminal_status"),
        result_payload.get("state"),
        result_payload.get("status"),
    )
    terminal_status = first_present(
        terminal_event.get("status") if isinstance(terminal_event, dict) else None,
        result_status,
        command_state if command_state in ("completed", "failed", "blocked", "cancelled", "canceled") else None,
    )
    return {
        "queried": True,
        "namespace": namespace,
        "ok": bool(command_response.get("ok") or result_response.get("ok") or events_response.get("ok")),
        "commandOk": command_response.get("ok"),
        "resultOk": result_response.get("ok"),
        "eventsOk": events_response.get("ok"),
        "commandState": command_state,
        "terminalStatus": terminal_status,
        "resultTerminalStatus": result_status,
        "resultOkValue": result_payload.get("ok"),
        "authority": result_payload.get("authority"),
        "fallback": result_payload.get("fallback"),
        "failureKind": first_present(
            result_payload.get("failureKind"),
            result_payload.get("terminalFailureKind"),
            terminal_event.get("failureKind") if isinstance(terminal_event, dict) else None,
        ),
        "terminalEventType": terminal_event.get("type") if isinstance(terminal_event, dict) else None,
        "terminalEventSeq": terminal_event.get("seq") if isinstance(terminal_event, dict) else None,
        "eventCount": len(events),
        "attempts": {
            "command": [{"ok": item.get("ok"), "rc": item.get("rc"), "method": item.get("method"), "proxyPath": item.get("proxyPath"), "stderrTail": item.get("stderrTail")} for item in command_attempts],
            "result": [{"ok": item.get("ok"), "rc": item.get("rc"), "method": item.get("method"), "proxyPath": item.get("proxyPath"), "stderrTail": item.get("stderrTail")} for item in result_attempts],
            "events": [{"ok": item.get("ok"), "rc": item.get("rc"), "method": item.get("method"), "proxyPath": item.get("proxyPath"), "stderrTail": item.get("stderrTail")} for item in events_attempts],
        },
    }

def parse_json(body):
    try:
        return json.loads(body) if body else None
    except Exception:
        return None

def attr_value(value):
    if not isinstance(value, dict):
        return value
    inner = value.get("value", value)
    if not isinstance(inner, dict):
        return inner
    for key in ("stringValue", "intValue", "doubleValue", "boolValue"):
        if key in inner:
            return inner.get(key)
    if "arrayValue" in inner:
        return "<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",
    ):
        if key in attrs:
            keep_attrs[key] = attrs[key]
    return {
        "name": item.get("name"),
        "service": item.get("service"),
        "attributes": keep_attrs,
    }

def tiny_terminal(terminal):
    if not isinstance(terminal, dict):
        return terminal
    return {
        "status": terminal.get("status"),
        "eventType": terminal.get("eventType"),
        "span": tiny_span(terminal.get("span")),
    }

def identity_from_spans(spans):
    identity = {}
    for key in IDENTITY_KEYS:
        preferred = None
        fallback = None
        for item in spans:
            attrs = item.get("attributes", {}) if isinstance(item.get("attributes"), dict) else {}
            value = attrs.get(key)
            if value in (None, ""):
                continue
            if fallback is None:
                fallback = value
            service = str(item.get("service") or "")
            if service.startswith("agentrun"):
                preferred = value
                break
        if preferred is not None or fallback is not None:
            identity[key] = preferred if preferred is not None else fallback
    return identity

def flatten_trace(trace_body):
    services = set()
    business_trace_ids = set()
    name_counts = collections.Counter()
    spans = []
    error_spans = []
    raw_index = 0
    for batch in batches_from_body(trace_body):
        resource = batch.get("resource") if isinstance(batch.get("resource"), dict) else {}
        resource_attrs = attrs_to_dict(resource.get("attributes"))
        service = resource_attrs.get("service.name") or "<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 = []
    projection_spans = []
    source_event_count = None
    requested_since_seq = None
    latest_projected_seq = None
    turn_statuses = []
    turn_status_counts = collections.Counter()
    for item in spans:
        if item.get("service") != "hwlab-cloud-api":
            continue
        name = str(item.get("name") or "")
        attrs = item.get("attributes", {}) if isinstance(item.get("attributes"), dict) else {}
        if name == "trace_events_read":
            trace_read_spans.append(public_span(item))
            for key in ("totalEvents", "sourceEventCount"):
                value = to_int(attrs.get(key))
                if value is not None:
                    source_event_count = value if source_event_count is None else max(source_event_count, value)
            for key in ("sinceSeq", "fromSeq"):
                value = to_int(attrs.get(key))
                if value is not None:
                    requested_since_seq = value if requested_since_seq is None else max(requested_since_seq, value)
        if name == "turn_status_read":
            turn_read_spans.append(public_span(item))
            if attrs.get("turnStatus") is not None or attrs.get("status") is not None:
                status_value = attrs.get("turnStatus") if attrs.get("turnStatus") is not None else attrs.get("status")
                turn_statuses.append(status_value)
                turn_status_counts[str(status_value)] += 1
        if name == "projection_sync":
            projection_spans.append(public_span(item))
            after_seq = to_int(attrs.get("afterSeq"))
            raw_count = to_int(attrs.get("rawEventCount"))
            projected = None
            for key in ("maxSeq", "traceLastSeq", "endSeq", "projectedSeq", "lastProjectedSeq"):
                value = to_int(attrs.get(key))
                if value is not None:
                    projected = value if projected is None else max(projected, value)
            if after_seq is not None:
                projected = after_seq + (raw_count or 0)
            if projected is not None:
                latest_projected_seq = projected if latest_projected_seq is None else max(latest_projected_seq, projected)
                source_event_count = projected if source_event_count is None else max(source_event_count, projected)
    has_more_values = []
    full_loaded_values = []
    for item in trace_read_spans:
        attrs = item.get("attributes", {}) if isinstance(item.get("attributes"), dict) else {}
        if "hasMore" in attrs:
            has_more_values.append(to_bool(attrs.get("hasMore")))
        if "fullTraceLoaded" in attrs:
            full_loaded_values.append(to_bool(attrs.get("fullTraceLoaded")))
    return {
        "sourceEventCount": source_event_count,
        "requestedSinceSeq": requested_since_seq,
        "latestProjectedSeq": latest_projected_seq,
        "traceStatus": {
            "readCount": len(trace_read_spans),
            "anyHasMore": any(value is True for value in has_more_values),
            "anyFullTraceLoaded": any(value is True for value in full_loaded_values),
        },
        "turnStatus": turn_statuses[-1] if turn_statuses else None,
        "turnStatusCounts": [{"status": status, "count": count} for status, count in turn_status_counts.most_common()],
        "traceEventReadSpans": trace_read_spans,
        "turnStatusReadSpans": turn_read_spans,
        "projectionSpans": projection_spans,
    }

def projection_lag_summary(agentrun, read_model):
    source_event_count = read_model.get("sourceEventCount")
    requested_since_seq = read_model.get("requestedSinceSeq")
    latest_projected_seq = read_model.get("latestProjectedSeq")
    agent_latest_seq = agentrun.get("latestSeq")
    reasons = []
    status = "none"
    if requested_since_seq is not None and source_event_count is not None and requested_since_seq > source_event_count:
        status = "confirmed"
        reasons.append("requested sinceSeq %s is greater than HWLAB sourceEventCount %s" % (requested_since_seq, source_event_count))
    if latest_projected_seq is not None and requested_since_seq is not None and latest_projected_seq < requested_since_seq:
        status = "confirmed"
        reasons.append("latestProjectedSeq %s is behind requested sinceSeq %s" % (latest_projected_seq, requested_since_seq))
    if agent_latest_seq is not None and source_event_count is not None and agent_latest_seq > source_event_count:
        status = "confirmed"
        reasons.append("AgentRun latestSeq %s is greater than HWLAB sourceEventCount %s" % (agent_latest_seq, source_event_count))
    if status == "none" and agentrun.get("terminalStatus") == "completed" and source_event_count is not None and read_model.get("traceStatus", {}).get("readCount", 0) > 0:
        status = "suspected"
        reasons.append("AgentRun completed while HWLAB read model still needs endpoint/status confirmation")
    return {
        "status": status,
        "reasons": reasons,
    }

def root_cause_candidates(http_summary, agentrun, read_model, lag_summary, error_spans, idle_warning_spans):
    candidates = []
    if http_summary.get("actorForbidden"):
        candidates.append({
            "code": "actor_forbidden",
            "label": "actor forbidden",
            "confidence": 0.98,
            "summary": "actor forbidden: HWLAB Code Agent read endpoint returned HTTP 403, likely actor/owner mismatch for this trace or turn.",
            "evidence": http_summary.get("problemCounts", [])[:5],
        })
    if lag_summary.get("status") == "confirmed":
        candidates.append({
            "code": "projection_read_model_stale",
            "label": "projection/read-model stale",
            "confidence": 0.94,
            "summary": "Projection/read-model stale: read model sequence is behind the caller/requested event window.",
            "evidence": {
                "sourceEventCount": read_model.get("sourceEventCount"),
                "requestedSinceSeq": read_model.get("requestedSinceSeq"),
                "latestProjectedSeq": read_model.get("latestProjectedSeq"),
                "reasons": lag_summary.get("reasons", []),
            },
        })
    elif lag_summary.get("status") == "suspected":
        candidates.append({
            "code": "projection_read_model_lag_suspected",
            "label": "projection/read-model stale",
            "confidence": 0.64,
            "summary": "Projection/read-model stale is suspected because AgentRun is terminal but HWLAB read-model evidence is incomplete.",
            "evidence": lag_summary.get("reasons", []),
        })
    failure_kind = str(agentrun.get("failureKind") or "")
    if failure_kind == "provider-auth-failed":
        candidates.append({
            "code": "provider_auth_failed",
            "label": "provider auth failed",
            "confidence": 0.96,
            "summary": "Provider credential refresh/auth failed before the manager reconciler closed the command; fix provider auth before retrying this Code Agent path.",
            "evidence": {
                "terminalStatus": agentrun.get("terminalStatus"),
                "failureKind": agentrun.get("failureKind"),
                "observedFailureKinds": agentrun.get("observedFailureKinds"),
                "terminalCategory": agentrun.get("terminalCategory"),
            },
        })
    if agentrun.get("terminalStatus") in ("failed", "error", "timeout", "blocked", "cancelled"):
        candidates.append({
            "code": "agentrun_terminal_failed",
            "label": "AgentRun terminal failed",
            "confidence": 0.9,
            "summary": "AgentRun reached a non-success terminal state; inspect result/events/runnerjob before retrying or continuing the session.",
            "evidence": {
                "terminalStatus": agentrun.get("terminalStatus"),
                "failureKind": agentrun.get("failureKind"),
                "terminalCategory": agentrun.get("terminalCategory"),
                "runnerProviderClassification": agentrun.get("runnerProviderClassification"),
            },
        })
    if agentrun.get("terminalStatus") == "completed":
        candidates.append({
            "code": "agentrun_completed_not_provider_blocked",
            "label": "AgentRun completed",
            "confidence": 0.88,
            "summary": "AgentRun completed: runner/provider reached terminal completed, so this trace is not explained by an active runner hang.",
            "evidence": {
                "terminalStatus": agentrun.get("terminalStatus"),
                "terminalEventType": agentrun.get("terminalEventType"),
                "runnerProviderClassification": agentrun.get("runnerProviderClassification"),
            },
        })
    if error_spans and agentrun.get("terminalStatus") == "completed":
        candidates.append({
            "code": "tool_call_failures_recovered",
            "label": "tool call failures recovered",
            "confidence": 0.42,
            "summary": "tool call failure spans exist, but AgentRun still completed; treat them as secondary unless correlated with user-visible failure.",
            "evidence": {
                "errorSpanCount": len(error_spans),
                "sampleNames": sorted(set(str(item.get("name") or "") for item in error_spans))[:5],
            },
        })
    if idle_warning_spans and agentrun.get("terminalStatus") in (None, ""):
        candidates.append({
            "code": "agentrun_runner_idle_warning_active",
            "label": "AgentRun runner idle warning active",
            "confidence": 0.72,
            "summary": "AgentRun runner is still emitting idle warnings and no runner terminal span is present; inspect waitingFor/lastEventLabel before treating the Workbench UI as terminal.",
            "evidence": [tiny_span(item) for item in idle_warning_spans[-3:]],
        })
    candidates.sort(key=lambda item: item.get("confidence", 0), reverse=True)
    return candidates

def candidate_score(trace_id, meta, trace_body, trace_rc, trace_err):
    parsed = parse_json(trace_body)
    meta_root_service = str(meta.get("rootServiceName") or "") if isinstance(meta, dict) else ""
    meta_root_name = str(meta.get("rootTraceName") or "") if isinstance(meta, dict) else ""
    if not isinstance(parsed, dict):
        return {
            "traceId": trace_id,
            "score": -100,
            "reasons": ["trace parse failed"],
            "parseOk": False,
            "queryOk": trace_rc == 0,
            "meta": compact_meta(meta),
            "spanCount": 0,
            "services": [],
            "servicePath": {
                "hwlab-cloud-api": "unknown",
                "agentrun-manager": "unknown",
                "agentrun-runner": "unknown",
                "complete": False,
            },
            "rootTraceName": meta_root_name or None,
            "rootServiceName": meta_root_service or None,
            "stderrTail": (trace_err or "")[-1000:],
        }, None
    flat = flatten_trace(parsed)
    spans = flat["spans"]
    services = set(flat["services"])
    names = [str(item.get("name") or "") for item in spans]
    lowered_names = [name.lower() for name in names]
    identity = identity_from_spans(spans)
    agentrun = agentrun_summary(spans)
    error_spans = flat["errorSpans"]
    score = 0
    reasons = []
    def add(points, reason):
        nonlocal score
        score += points
        reasons.append("%+d %s" % (points, reason))
    if "agentrun-runner" in services:
        add(120, "contains agentrun-runner")
    if "agentrun-manager" in services:
        add(90, "contains agentrun-manager")
    if "hwlab-cloud-api" in services:
        add(20, "contains hwlab-cloud-api")
    span_points = min(len(spans), 80)
    if span_points:
        add(span_points, "span count %s" % len(spans))
    for key, points in (("runId", 50), ("commandId", 50), ("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", "runnerJobId", "runnerId", "backendProfile") if identity.get(key) not in (None, "")},
        "terminalStatus": agentrun.get("terminalStatus"),
        "errorSpanCount": len(error_spans),
        "candidateQuality": candidate_quality,
        "lowConfidence": candidate_quality != "normal",
        "rootTraceName": meta_root_name or None,
        "rootServiceName": meta_root_service or None,
        "spanNamePreview": names[:8],
        "traceCommand": "bun scripts/cli.ts platform-infra observability trace --target ${target.id} --trace-id %s --limit 80" % trace_id,
    }, parsed

def resolve_trace():
    if TRACE_ID is not None:
        return TRACE_ID, {
            "mode": "trace-id",
            "businessTraceId": BUSINESS_TRACE_ID,
            "runId": RUN_ID,
            "commandId": COMMAND_ID,
            "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,
        "runnerJobId": RUNNER_JOB_ID,
        "otelTraceId": selected_trace_id,
        "searchQuery": SEARCH_QUERY,
        "searchPath": SEARCH_PATH,
        "searchProxyPath": SEARCH_PROXY_PATH,
        "searchOk": search_rc == 0,
        "searchParseOk": isinstance(search_parsed, dict),
        "candidateTraceCount": len(metas),
        "scannedCandidateCount": scanned,
        "candidateSelectionMode": "scored-service-and-span-content",
        "selectedScore": selected_score,
        "selectedQuality": selected_quality,
        "selectedLowConfidence": selected_low_confidence,
        "selectedRejectedReason": selected_rejected_reason,
        "selectedReasons": selected_summary.get("reasons", []) if isinstance(selected_summary, dict) else [],
        "selectedMeta": compact_meta(selected),
        "candidatePreview": candidate_summaries[:8],
        "searchStderrTail": (search_err or "")[-2000:],
    }
    if RAW:
        mapping["rawSearchBody"] = search_parsed if isinstance(search_parsed, dict) else search_body[-12000:]
    if selected_trace_id is None:
        return None, mapping, "no OTel trace found for diagnosis query"
    if selected_rejected_reason is not None:
        return None, mapping, selected_rejected_reason
    return selected_trace_id, mapping, None

resolved_trace_id, mapping, resolve_error = resolve_trace()
if resolved_trace_id is None:
    payload = {
        "ok": False,
        "mapping": mapping,
        "error": resolve_error,
        "summary": {
            "rootCause": "otel trace not found",
            "facts": [],
        },
        "next": {
            "expandWindow": DIAGNOSE_FULL_COMMAND,
            "search": SEARCH_COMMAND,
        },
    }
    print(json.dumps(payload, ensure_ascii=False, indent=2))
    raise SystemExit(1)

trace_path = TRACE_PATH_TEMPLATE.replace("{{traceId}}", resolved_trace_id)
trace_proxy_path = TRACE_PROXY_PREFIX + trace_path
trace_rc, trace_body, trace_err = run_kubectl(trace_proxy_path, timeout=25)
trace_parsed = parse_json(trace_body)
if not isinstance(trace_parsed, dict):
    payload = {
        "ok": False,
        "mapping": mapping,
        "tracePath": trace_path,
        "traceProxyPath": trace_proxy_path,
        "traceQueryOk": trace_rc == 0,
        "traceParseOk": False,
        "bodyBytes": len(trace_body.encode("utf-8")),
        "bodyTail": trace_body[-12000:],
        "stderrTail": (trace_err or "")[-4000:],
    }
    print(json.dumps(payload, ensure_ascii=False, indent=2))
    raise SystemExit(1)

flat = flatten_trace(trace_parsed)
spans = flat["spans"]
services = flat["services"]
business_trace_ids = flat["businessTraceIds"]
error_spans = flat["errorSpans"]
idle_warning_spans = [item for item in spans if str(item.get("name") or "") == "codex_stdio.idle_warning"]
http_summary = http_status_summary(spans)
read_model = hwlab_read_model_summary(spans)
identity = identity_from_spans(spans)
agentrun_authority = agentrun_authority_summary(identity)
agentrun = agentrun_summary(spans, agentrun_authority)
lag = projection_lag_summary(agentrun, read_model)
candidates = root_cause_candidates(http_summary, agentrun, read_model, lag, error_spans, idle_warning_spans)
expected_services = ["hwlab-cloud-api", "agentrun-manager", "agentrun-runner"]
service_path = {
    service: ("reached" if service in services else "missing")
    for service in expected_services
}
service_path["complete"] = all(service in services for service in expected_services)
facts = []
if http_summary.get("actorForbidden"):
    facts.append("actor forbidden")
terminal_status = agentrun.get("terminalStatus")
if terminal_status in ("failed", "error", "timeout", "blocked", "cancelled"):
    failure_kind = agentrun.get("failureKind")
    if failure_kind:
        facts.append(f"AgentRun terminal failed ({failure_kind})")
    else:
        facts.append("AgentRun terminal failed")
if terminal_status == "completed":
    facts.append("AgentRun completed")
if lag.get("status") in ("confirmed", "suspected"):
    facts.append("projection/read-model stale")
if idle_warning_spans and terminal_status in (None, ""):
    facts.append("AgentRun runner idle warnings active")
if not facts:
    facts.append("no dominant Code Agent root cause classified")
summary = {
    "rootCause": "; ".join(facts),
    "facts": facts,
    "classification": {
        "projectionLag": lag.get("status"),
        "runnerProvider": agentrun.get("runnerProviderClassification"),
        "actorForbidden": http_summary.get("actorForbidden"),
        "terminalStatus": terminal_status,
        "failureKind": agentrun.get("failureKind"),
    },
}
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,
    "businessTraceIds": business_trace_ids[:20],
    "identity": identity,
    "agentrun": {
        "terminalStatus": agentrun.get("terminalStatus"),
        "terminalSource": agentrun.get("terminalSource"),
        "failureKind": agentrun.get("failureKind"),
        "observedFailureKinds": agentrun.get("observedFailureKinds"),
        "latestSeq": agentrun.get("latestSeq"),
        "terminalEventType": agentrun.get("terminalEventType"),
        "runnerProviderClassification": agentrun.get("runnerProviderClassification"),
        "authority": agentrun.get("authority"),
    },
    "hwlabReadModel": {
        "sourceEventCount": read_model.get("sourceEventCount"),
        "requestedSinceSeq": read_model.get("requestedSinceSeq"),
        "latestProjectedSeq": read_model.get("latestProjectedSeq"),
        "traceStatus": read_model.get("traceStatus"),
        "turnStatus": read_model.get("turnStatus"),
        "turnStatusCounts": read_model.get("turnStatusCounts"),
    },
    "http": {
        "statusCounts": http_summary.get("statusCounts", [])[:20],
        "problemCounts": http_summary.get("problemCounts", [])[:20],
        "actorForbidden": http_summary.get("actorForbidden"),
    },
    "projectionLag": lag,
    "summary": summary,
    "rootCauseCandidates": candidates,
    "spanNameCounts": flat["spanNameCounts"][:12],
    "evidence": evidence,
    "next": {
        "diagnoseFull": DIAGNOSE_FULL_COMMAND,
        "traceSummary": "bun scripts/cli.ts platform-infra observability trace --target ${target.id} --trace-id %s --limit 80" % resolved_trace_id,
        "traceReads": "bun scripts/cli.ts platform-infra observability trace --target ${target.id} --trace-id %s --grep trace_events_read --limit 20 --full" % resolved_trace_id,
        "turnStatus": "bun scripts/cli.ts platform-infra observability trace --target ${target.id} --trace-id %s --grep turn_status_read --limit 20 --full" % resolved_trace_id,
        "projection": "bun scripts/cli.ts platform-infra observability trace --target ${target.id} --trace-id %s --grep projection_sync --limit 120 --full" % resolved_trace_id,
        "runnerTerminal": "bun scripts/cli.ts platform-infra observability trace --target ${target.id} --trace-id %s --grep runner_terminal --limit 20 --full" % resolved_trace_id,
        "raw": "bun scripts/cli.ts platform-infra observability diagnose-code-agent --target ${target.id} --trace-id %s --raw" % resolved_trace_id,
    },
    "stderrTail": (trace_err or "")[-4000:],
}
if FULL:
    payload["full"] = {
        "traceEventReadSpans": read_model.get("traceEventReadSpans", []),
        "turnStatusReadSpans": read_model.get("turnStatusReadSpans", []),
        "projectionSpans": read_model.get("projectionSpans", [])[:LIMIT],
        "terminalSpans": agentrun.get("terminalSpans", [])[:LIMIT],
        "errorSpans": [public_span(item) for item in error_spans[:LIMIT]],
        "httpProblemSpans": [public_span(item) for item in http_summary.get("problemSpans", [])[:LIMIT]],
        "selectedSpans": [public_span(item) for item in spans if item.get("name") in ("trace_events_read", "turn_status_read", "projection_sync", "command_result")][:LIMIT],
    }
if RAW:
    payload["raw"] = raw_trace_shape(trace_parsed)
print(json.dumps(payload, ensure_ascii=False, indent=2 if FULL or RAW else None))
raise SystemExit(0 if payload["ok"] else 1)
PY
`;
}
