// 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 {
  compactCapture,
  createYamlFieldReader,
  numberField,
  parseJsonOutput,
  redactSensitiveUnknown,
  shQuote,
  capture,
} from "./platform-infra-ops-library";

const configFile = rootPath("config", "platform-infra", "observability.yaml");
const configLabel = "config/platform-infra/observability.yaml";
const fieldManager = "unidesk-platform-observability";
const {
  asRecord,
  objectField,
  arrayOfRecords,
  stringField,
  integerField,
  booleanField,
  stringArrayField,
  numberArrayField,
  enumField,
  kubernetesNameField,
  portField,
  apiPathField,
} = createYamlFieldReader(configLabel);

interface ObservabilityConfig {
  version: number;
  kind: "platform-infra-observability";
  metadata: { id: string; owner: string; spec: string; relatedIssues: number[] };
  defaults: { targetId: string };
  images: {
    collector: ImageSpec;
    tempo: ImageSpec;
  };
  targets: ObservabilityTarget[];
  collector: {
    deploymentName: string;
    serviceName: string;
    configMapName: string;
    replicas: number;
    healthPort: number;
    otlp: OtlpPorts;
  };
  traceBackend: {
    type: "tempo";
    deploymentName: string;
    serviceName: string;
    configMapName: string;
    replicas: number;
    httpPort: number;
    otlp: OtlpPorts;
    storage: { mode: "emptyDir"; retention: string };
  };
  sampling: { mode: "parentbased_traceidratio"; ratio: number };
  instrumentation: {
    contextPropagation: string[];
    serviceConnections: ServiceConnection[];
  };
  resourceAttributes: {
    required: string[];
    businessCorrelationAttributes: string[];
  };
  probes: {
    readinessPath: string;
    traceQueryPathTemplate: string;
    statusEndpoints: StatusEndpoint[];
  };
}

interface ImageSpec {
  repository: string;
  tag: string;
  pullPolicy: "Always" | "IfNotPresent" | "Never";
}

interface ObservabilityTarget {
  id: string;
  route: string;
  namespace: string;
  role: "active" | "standby";
  enabled: boolean;
  createNamespace: boolean;
}

interface OtlpPorts {
  grpcPort: number;
  httpPort: number;
}

interface ServiceConnection {
  serviceName: string;
  owningRepo: string;
  targetNode: string;
  lane: string;
  namespace: string;
  requiredSpans: string[];
}

interface StatusEndpoint {
  name: string;
  service: string;
  portName: string;
  path: string;
}

interface CommonOptions {
  targetId: string | null;
  full: boolean;
  raw: boolean;
}

interface ApplyOptions extends CommonOptions {
  confirm: boolean;
  dryRun: boolean;
  wait: boolean;
}

interface TraceOptions extends CommonOptions {
  traceId: string | null;
  grep: string | null;
  limit: number;
}

interface SearchOptions extends CommonOptions {
  grep: string | null;
  query: string | null;
  path: string | null;
  status: number | null;
  limit: number;
  candidateLimit: number;
  lookbackMinutes: number;
}

interface DiagnoseCodeAgentOptions extends CommonOptions {
  businessTraceId: string | null;
  traceId: string | null;
  limit: number;
  candidateLimit: number;
  lookbackMinutes: number;
}

export function observabilityHelp(): Record<string, unknown> {
  return {
    command: "platform-infra observability plan|apply|status|validate|trace|search|diagnose-code-agent",
    output: "json",
    configTruth: "config/platform-infra/observability.yaml",
    spec: "PJ2026-01060501 OTel追踪 draft-2026-06-19-p0",
    usage: [
      "bun scripts/cli.ts platform-infra observability plan --target D601",
      "bun scripts/cli.ts platform-infra observability apply --target D601 --dry-run",
      "bun scripts/cli.ts platform-infra observability apply --target D601 --confirm",
      "bun scripts/cli.ts platform-infra observability status --target D601 [--full|--raw]",
      "bun scripts/cli.ts platform-infra observability validate --target D601 [--full|--raw]",
      "bun scripts/cli.ts platform-infra observability trace --target D601 --trace-id <traceId> [--grep provider-stream-disconnected] [--limit 40] [--full|--raw]",
      "bun scripts/cli.ts platform-infra observability search --target D601 --grep 'no rollout found' [--lookback-minutes 360] [--candidate-limit 80] [--limit 20] [--full|--raw]",
      "bun scripts/cli.ts platform-infra observability search --target D601 --path /v1/workbench/sessions --status 502 [--lookback-minutes 120] [--full|--raw]",
      "bun scripts/cli.ts platform-infra observability diagnose-code-agent --target D601 --business-trace-id <trc_...> [--full|--raw]",
    ],
    boundary: "Prometheus remains the metrics source; this command owns only platform-infra OTel Collector, trace backend readiness, and trace lookup.",
  };
}

export async function runPlatformObservabilityCommand(config: UniDeskConfig, args: string[]): Promise<Record<string, unknown>> {
  const [action = "plan"] = args;
  if (action === "plan") return plan(parseCommonOptions(args.slice(1)));
  if (action === "apply") return await apply(config, parseApplyOptions(args.slice(1)));
  if (action === "status") return await status(config, parseCommonOptions(args.slice(1)));
  if (action === "validate") return await validate(config, parseCommonOptions(args.slice(1)));
  if (action === "trace") return await trace(config, parseTraceOptions(args.slice(1)));
  if (action === "search") return await search(config, parseSearchOptions(args.slice(1)));
  if (action === "diagnose-code-agent") return await diagnoseCodeAgent(config, parseDiagnoseCodeAgentOptions(args.slice(1)));
  return { ok: false, error: "unsupported-platform-infra-observability-command", args, help: observabilityHelp() };
}

function parseCommonOptions(args: string[]): CommonOptions {
  let targetId: string | null = null;
  let full = false;
  let raw = false;
  for (let index = 0; index < args.length; index += 1) {
    const arg = args[index];
    if (arg === "--target") {
      const value = args[index + 1];
      if (value === undefined || value.startsWith("--")) throw new Error("--target requires a value");
      if (!/^[A-Za-z0-9._-]+$/u.test(value)) throw new Error("--target must be a simple target id");
      targetId = value;
      index += 1;
    } else if (arg === "--full") {
      full = true;
    } else if (arg === "--raw") {
      raw = true;
      full = true;
    } else {
      throw new Error(`unsupported observability option: ${arg}`);
    }
  }
  return { targetId, full, raw };
}

function parseApplyOptions(args: string[]): ApplyOptions {
  const commonArgs: string[] = [];
  let confirm = false;
  let dryRun = false;
  let wait = false;
  for (let index = 0; index < args.length; index += 1) {
    const arg = args[index];
    if (arg === "--confirm") confirm = true;
    else if (arg === "--dry-run") dryRun = true;
    else if (arg === "--wait") wait = true;
    else {
      commonArgs.push(arg);
      if (arg === "--target") {
        commonArgs.push(args[index + 1] ?? "");
        index += 1;
      }
    }
  }
  if (confirm && dryRun) throw new Error("observability apply accepts only one of --confirm or --dry-run");
  return { ...parseCommonOptions(commonArgs), confirm, dryRun: dryRun || !confirm, wait };
}

function parseTraceOptions(args: string[]): TraceOptions {
  const commonArgs: string[] = [];
  let traceId: string | null = null;
  let grep: string | null = null;
  let limit = 40;
  for (let index = 0; index < args.length; index += 1) {
    const arg = args[index];
    if (arg === "--trace-id") {
      const value = args[index + 1];
      if (value === undefined || value.startsWith("--")) throw new Error("--trace-id requires a value");
      if (!/^[0-9a-fA-F]{32}$/u.test(value)) throw new Error("--trace-id must be a 32-character OpenTelemetry trace id encoded as hex");
      traceId = value;
      index += 1;
    } else if (arg === "--grep") {
      const value = args[index + 1];
      if (value === undefined || value.startsWith("--")) throw new Error("--grep requires a value");
      grep = value;
      index += 1;
    } else if (arg === "--limit") {
      const value = args[index + 1];
      if (value === undefined || value.startsWith("--")) throw new Error("--limit requires a value");
      const parsed = Number(value);
      if (!Number.isInteger(parsed) || parsed < 1 || parsed > 500) throw new Error("--limit must be an integer from 1 to 500");
      limit = parsed;
      index += 1;
    } else {
      commonArgs.push(arg);
      if (arg === "--target") {
        commonArgs.push(args[index + 1] ?? "");
        index += 1;
      }
    }
  }
  return { ...parseCommonOptions(commonArgs), traceId, grep, limit };
}

function parseSearchOptions(args: string[]): SearchOptions {
  const commonArgs: string[] = [];
  let grep: string | null = null;
  let query: string | null = null;
  let path: string | null = null;
  let status: number | null = null;
  let limit = 20;
  let candidateLimit = 80;
  let lookbackMinutes = 360;
  for (let index = 0; index < args.length; index += 1) {
    const arg = args[index];
    if (arg === "--grep") {
      const value = args[index + 1];
      if (value === undefined || value.startsWith("--")) throw new Error("--grep requires a value");
      grep = value;
      index += 1;
    } else if (arg === "--query" || arg === "--q") {
      const value = args[index + 1];
      if (value === undefined || value.startsWith("--")) throw new Error(`${arg} requires a value`);
      query = value;
      index += 1;
    } else if (arg === "--path") {
      const value = args[index + 1];
      if (value === undefined || value.startsWith("--")) throw new Error("--path requires a value");
      if (!value.startsWith("/") || value.includes("\n") || value.length > 300) throw new Error("--path must be an absolute HTTP path up to 300 characters");
      path = value;
      index += 1;
    } else if (arg === "--status") {
      const value = args[index + 1];
      if (value === undefined || value.startsWith("--")) throw new Error("--status requires a value");
      const parsed = Number(value);
      if (!Number.isInteger(parsed) || parsed < 100 || parsed > 599) throw new Error("--status must be an integer HTTP status from 100 to 599");
      status = parsed;
      index += 1;
    } else if (arg === "--limit") {
      const value = args[index + 1];
      if (value === undefined || value.startsWith("--")) throw new Error("--limit requires a value");
      const parsed = Number(value);
      if (!Number.isInteger(parsed) || parsed < 1 || parsed > 100) throw new Error("--limit must be an integer from 1 to 100");
      limit = parsed;
      index += 1;
    } else if (arg === "--candidate-limit") {
      const value = args[index + 1];
      if (value === undefined || value.startsWith("--")) throw new Error("--candidate-limit requires a value");
      const parsed = Number(value);
      if (!Number.isInteger(parsed) || parsed < 1 || parsed > 500) throw new Error("--candidate-limit must be an integer from 1 to 500");
      candidateLimit = parsed;
      index += 1;
    } else if (arg === "--lookback-minutes" || arg === "--since-minutes") {
      const value = args[index + 1];
      if (value === undefined || value.startsWith("--")) throw new Error(`${arg} requires a value`);
      const parsed = Number(value);
      if (!Number.isInteger(parsed) || parsed < 1 || parsed > 10080) throw new Error(`${arg} must be an integer from 1 to 10080`);
      lookbackMinutes = parsed;
      index += 1;
    } else {
      commonArgs.push(arg);
      if (arg === "--target") {
        commonArgs.push(args[index + 1] ?? "");
        index += 1;
      }
    }
  }
  if (grep === null && query === null && path === null && status === null) throw new Error("observability search requires --grep <text>, --query <tempo-query>, --path <route>, or --status <code>");
  return { ...parseCommonOptions(commonArgs), grep, query, path, status, limit, candidateLimit, lookbackMinutes };
}

function parseDiagnoseCodeAgentOptions(args: string[]): DiagnoseCodeAgentOptions {
  const commonArgs: string[] = [];
  let businessTraceId: string | null = null;
  let traceId: string | null = null;
  let limit = 40;
  let candidateLimit = 300;
  let lookbackMinutes = 10080;
  for (let index = 0; index < args.length; index += 1) {
    const arg = args[index];
    if (arg === "--business-trace-id" || arg === "--trace") {
      const value = args[index + 1];
      if (value === undefined || value.startsWith("--")) throw new Error(`${arg} requires a value`);
      if (!/^trc_[A-Za-z0-9_-]+$/u.test(value)) throw new Error(`${arg} must look like trc_<id>`);
      businessTraceId = value;
      index += 1;
    } else if (arg === "--trace-id" || arg === "--otel-trace-id") {
      const value = args[index + 1];
      if (value === undefined || value.startsWith("--")) throw new Error(`${arg} requires a value`);
      if (!/^[0-9a-fA-F]{32}$/u.test(value)) throw new Error(`${arg} must be a 32-character OpenTelemetry trace id encoded as hex`);
      traceId = value.toLowerCase();
      index += 1;
    } else if (arg === "--limit") {
      const value = args[index + 1];
      if (value === undefined || value.startsWith("--")) throw new Error("--limit requires a value");
      const parsed = Number(value);
      if (!Number.isInteger(parsed) || parsed < 1 || parsed > 200) throw new Error("--limit must be an integer from 1 to 200");
      limit = parsed;
      index += 1;
    } else if (arg === "--candidate-limit") {
      const value = args[index + 1];
      if (value === undefined || value.startsWith("--")) throw new Error("--candidate-limit requires a value");
      const parsed = Number(value);
      if (!Number.isInteger(parsed) || parsed < 1 || parsed > 500) throw new Error("--candidate-limit must be an integer from 1 to 500");
      candidateLimit = parsed;
      index += 1;
    } else if (arg === "--lookback-minutes" || arg === "--since-minutes") {
      const value = args[index + 1];
      if (value === undefined || value.startsWith("--")) throw new Error(`${arg} requires a value`);
      const parsed = Number(value);
      if (!Number.isInteger(parsed) || parsed < 1 || parsed > 10080) throw new Error(`${arg} must be an integer from 1 to 10080`);
      lookbackMinutes = parsed;
      index += 1;
    } else {
      commonArgs.push(arg);
      if (arg === "--target") {
        commonArgs.push(args[index + 1] ?? "");
        index += 1;
      }
    }
  }
  if (businessTraceId === null && traceId === null) throw new Error("observability diagnose-code-agent requires --business-trace-id <trc_...> or --trace-id <otelTraceId>");
  return { ...parseCommonOptions(commonArgs), businessTraceId, traceId, limit, candidateLimit, lookbackMinutes };
}

function readObservabilityConfig(): ObservabilityConfig {
  const parsed = Bun.YAML.parse(readFileSync(configFile, "utf8")) as unknown;
  const root = asRecord(parsed, configLabel);
  const version = integerField(root, "version", "");
  const kind = stringField(root, "kind", "");
  if (kind !== "platform-infra-observability") throw new Error(`${configLabel}.kind must be platform-infra-observability`);
  const metadata = objectField(root, "metadata", "");
  const defaults = objectField(root, "defaults", "");
  const images = objectField(root, "images", "");
  const collector = objectField(root, "collector", "");
  const collectorOtlp = objectField(collector, "otlp", "collector");
  const traceBackend = objectField(root, "traceBackend", "");
  const traceBackendOtlp = objectField(traceBackend, "otlp", "traceBackend");
  const traceBackendStorage = objectField(traceBackend, "storage", "traceBackend");
  const sampling = objectField(root, "sampling", "");
  const instrumentation = objectField(root, "instrumentation", "");
  const resourceAttributes = objectField(root, "resourceAttributes", "");
  const probes = objectField(root, "probes", "");
  const config: ObservabilityConfig = {
    version,
    kind,
    metadata: {
      id: stringField(metadata, "id", "metadata"),
      owner: stringField(metadata, "owner", "metadata"),
      spec: stringField(metadata, "spec", "metadata"),
      relatedIssues: numberArrayField(metadata, "relatedIssues", "metadata"),
    },
    defaults: { targetId: stringField(defaults, "targetId", "defaults") },
    images: {
      collector: imageSpec(objectField(images, "collector", "images"), "images.collector"),
      tempo: imageSpec(objectField(images, "tempo", "images"), "images.tempo"),
    },
    targets: arrayOfRecords(root.targets, "targets").map(parseTarget),
    collector: {
      deploymentName: kubernetesNameField(collector, "deploymentName", "collector"),
      serviceName: kubernetesNameField(collector, "serviceName", "collector"),
      configMapName: kubernetesNameField(collector, "configMapName", "collector"),
      replicas: integerField(collector, "replicas", "collector"),
      healthPort: portField(collector, "healthPort", "collector"),
      otlp: parseOtlpPorts(collectorOtlp, "collector.otlp"),
    },
    traceBackend: {
      type: enumField(traceBackend, "type", "traceBackend", ["tempo"] as const),
      deploymentName: kubernetesNameField(traceBackend, "deploymentName", "traceBackend"),
      serviceName: kubernetesNameField(traceBackend, "serviceName", "traceBackend"),
      configMapName: kubernetesNameField(traceBackend, "configMapName", "traceBackend"),
      replicas: integerField(traceBackend, "replicas", "traceBackend"),
      httpPort: portField(traceBackend, "httpPort", "traceBackend"),
      otlp: parseOtlpPorts(traceBackendOtlp, "traceBackend.otlp"),
      storage: {
        mode: enumField(traceBackendStorage, "mode", "traceBackend.storage", ["emptyDir"] as const),
        retention: stringField(traceBackendStorage, "retention", "traceBackend.storage"),
      },
    },
    sampling: {
      mode: enumField(sampling, "mode", "sampling", ["parentbased_traceidratio"] as const),
      ratio: numberField(sampling, "ratio", "sampling"),
    },
    instrumentation: {
      contextPropagation: stringArrayField(instrumentation, "contextPropagation", "instrumentation"),
      serviceConnections: arrayOfRecords(instrumentation.serviceConnections, "instrumentation.serviceConnections").map(parseServiceConnection),
    },
    resourceAttributes: {
      required: stringArrayField(resourceAttributes, "required", "resourceAttributes"),
      businessCorrelationAttributes: stringArrayField(resourceAttributes, "businessCorrelationAttributes", "resourceAttributes"),
    },
    probes: {
      readinessPath: apiPathField(probes, "readinessPath", "probes"),
      traceQueryPathTemplate: stringField(probes, "traceQueryPathTemplate", "probes"),
      statusEndpoints: arrayOfRecords(probes.statusEndpoints, "probes.statusEndpoints").map(parseStatusEndpoint),
    },
  };
  if (config.targets.length === 0) throw new Error(`${configLabel}.targets must not be empty`);
  assertKnownEnabledTarget(config.targets, config.defaults.targetId, "defaults.targetId");
  if (config.collector.replicas < 0 || config.traceBackend.replicas < 0) throw new Error(`${configLabel} replicas must be >= 0`);
  if (config.sampling.ratio < 0 || config.sampling.ratio > 1) throw new Error(`${configLabel}.sampling.ratio must be between 0 and 1`);
  if (!config.probes.traceQueryPathTemplate.includes("{{traceId}}")) throw new Error(`${configLabel}.probes.traceQueryPathTemplate must include {{traceId}}`);
  return config;
}

function imageSpec(record: Record<string, unknown>, path: string): ImageSpec {
  const image = {
    repository: stringField(record, "repository", path),
    tag: stringField(record, "tag", path),
    pullPolicy: enumField(record, "pullPolicy", path, ["Always", "IfNotPresent", "Never"] as const),
  };
  if (!/^[A-Za-z0-9._/:@-]+$/u.test(`${image.repository}:${image.tag}`)) throw new Error(`${configLabel}.${path} must render a valid image reference`);
  return image;
}

function parseTarget(record: Record<string, unknown>, index: number): ObservabilityTarget {
  const path = `targets[${index}]`;
  return {
    id: stringField(record, "id", path),
    route: stringField(record, "route", path),
    namespace: kubernetesNameField(record, "namespace", path),
    role: enumField(record, "role", path, ["active", "standby"] as const),
    enabled: booleanField(record, "enabled", path),
    createNamespace: booleanField(record, "createNamespace", path),
  };
}

function parseOtlpPorts(record: Record<string, unknown>, path: string): OtlpPorts {
  return {
    grpcPort: portField(record, "grpcPort", path),
    httpPort: portField(record, "httpPort", path),
  };
}

function parseServiceConnection(record: Record<string, unknown>, index: number): ServiceConnection {
  const path = `instrumentation.serviceConnections[${index}]`;
  return {
    serviceName: stringField(record, "serviceName", path),
    owningRepo: stringField(record, "owningRepo", path),
    targetNode: stringField(record, "targetNode", path),
    lane: stringField(record, "lane", path),
    namespace: kubernetesNameField(record, "namespace", path),
    requiredSpans: stringArrayField(record, "requiredSpans", path),
  };
}

function parseStatusEndpoint(record: Record<string, unknown>, index: number): StatusEndpoint {
  const path = `probes.statusEndpoints[${index}]`;
  return {
    name: stringField(record, "name", path),
    service: kubernetesNameField(record, "service", path),
    portName: stringField(record, "portName", path),
    path: apiPathField(record, "path", path),
  };
}

function assertKnownEnabledTarget(targets: ObservabilityTarget[], targetId: string, path: string): void {
  const target = targets.find((item) => item.id.toLowerCase() === targetId.toLowerCase());
  if (target === undefined) throw new Error(`${configLabel}.${path} references unknown target ${targetId}; known targets: ${targets.map((item) => item.id).join(", ")}`);
  if (!target.enabled) throw new Error(`${configLabel}.${path} references disabled target ${target.id}`);
}

function resolveTarget(observability: ObservabilityConfig, targetId: string | null): ObservabilityTarget {
  const resolved = targetId ?? observability.defaults.targetId;
  const target = observability.targets.find((item) => item.id.toLowerCase() === resolved.toLowerCase());
  if (target === undefined) throw new Error(`unknown observability target ${resolved}; known targets: ${observability.targets.map((item) => item.id).join(", ")}`);
  if (!target.enabled) throw new Error(`observability target ${target.id} is disabled in ${configLabel}`);
  return target;
}

function plan(options: CommonOptions): Record<string, unknown> {
  const observability = readObservabilityConfig();
  const target = resolveTarget(observability, options.targetId);
  const yaml = renderManifest(observability, target);
  const policy = policyChecks(yaml, target);
  return {
    ok: policy.every((check) => check.ok),
    action: "platform-infra-observability-plan",
    mutation: false,
    config: configSummary(observability, target),
    renderPlan: {
      target: targetSummary(target),
      objects: manifestObjectSummary(yaml),
      otlp: {
        collectorGrpcEndpoint: `${observability.collector.serviceName}.${target.namespace}.svc.cluster.local:${observability.collector.otlp.grpcPort}`,
        collectorHttpEndpoint: `http://${observability.collector.serviceName}.${target.namespace}.svc.cluster.local:${observability.collector.otlp.httpPort}`,
        backendGrpcEndpoint: `${observability.traceBackend.serviceName}.${target.namespace}.svc.cluster.local:${observability.traceBackend.otlp.grpcPort}`,
      },
      instrumentation: observability.instrumentation.serviceConnections,
      resourceAttributes: observability.resourceAttributes,
    },
    policy,
    next: {
      dryRun: `bun scripts/cli.ts platform-infra observability apply --target ${target.id} --dry-run`,
      apply: `bun scripts/cli.ts platform-infra observability apply --target ${target.id} --confirm`,
      status: `bun scripts/cli.ts platform-infra observability status --target ${target.id}`,
      validate: `bun scripts/cli.ts platform-infra observability validate --target ${target.id}`,
      trace: `bun scripts/cli.ts platform-infra observability trace --target ${target.id} --trace-id <traceId>`,
    },
  };
}

async function apply(config: UniDeskConfig, options: ApplyOptions): Promise<Record<string, unknown>> {
  const observability = readObservabilityConfig();
  const target = resolveTarget(observability, options.targetId);
  const yaml = renderManifest(observability, target);
  const policy = policyChecks(yaml, target);
  if (!policy.every((check) => check.ok)) return { ok: false, action: "platform-infra-observability-apply", mode: "policy-blocked", policy };
  const result = await capture(config, target.route, ["sh"], applyScript({
    yaml,
    target,
    dryRun: options.dryRun,
    wait: options.wait,
    collectorDeploymentName: observability.collector.deploymentName,
    backendDeploymentName: observability.traceBackend.deploymentName,
  }));
  const parsed = parseJsonOutput(result.stdout);
  return {
    ok: result.exitCode === 0 && parsed?.ok === true,
    action: "platform-infra-observability-apply",
    mode: options.dryRun ? "dry-run" : "confirmed",
    mutation: !options.dryRun,
    target: targetSummary(target),
    policy,
    remote: parsed ?? compactCapture(result, { full: true }),
  };
}

async function status(config: UniDeskConfig, options: CommonOptions): Promise<Record<string, unknown>> {
  const observability = readObservabilityConfig();
  const target = resolveTarget(observability, options.targetId);
  const result = await capture(config, target.route, ["sh"], statusScript(observability, target, options.full));
  const parsed = parseJsonOutput(result.stdout);
  const summary = parsed === null ? null : statusSummary(parsed);
  return {
    ok: result.exitCode === 0 && summary?.ready === true,
    action: "platform-infra-observability-status",
    mutation: false,
    target: targetSummary(target),
    summary,
    remote: options.raw ? parsed : compactStatus(parsed, options.full) ?? compactCapture(result, { full: true }),
    next: {
      plan: `bun scripts/cli.ts platform-infra observability plan --target ${target.id}`,
      apply: `bun scripts/cli.ts platform-infra observability apply --target ${target.id} --confirm`,
      validate: `bun scripts/cli.ts platform-infra observability validate --target ${target.id}`,
    },
  };
}

async function validate(config: UniDeskConfig, options: CommonOptions): Promise<Record<string, unknown>> {
  const observability = readObservabilityConfig();
  const target = resolveTarget(observability, options.targetId);
  const result = await capture(config, target.route, ["sh"], statusScript(observability, target, options.full, true));
  const parsed = parseJsonOutput(result.stdout);
  const summary = parsed === null ? null : statusSummary(parsed);
  const ready = summary?.ready === true;
  const validationTrace = parsed !== null && typeof parsed.validationTrace === "object" && parsed.validationTrace !== null && !Array.isArray(parsed.validationTrace)
    ? parsed.validationTrace as Record<string, unknown>
    : null;
  const validationTraceId = typeof validationTrace?.traceId === "string" ? validationTrace.traceId : null;
  const validationTraceOk = validationTrace?.ok === true;
  return {
    ok: result.exitCode === 0 && ready && validationTraceOk,
    action: "platform-infra-observability-validate",
    mutation: false,
    target: targetSummary(target),
    summary,
    validation: {
      readiness: ready ? "passed" : "failed",
      testTrace: validationTrace ?? "not-generated-runtime-not-ready",
      traceQuery: validationTraceId !== null
        ? `bun scripts/cli.ts platform-infra observability trace --target ${target.id} --trace-id ${validationTraceId}`
        : "blocked-until-validation-trace-generated",
      metricsBoundary: "Prometheus/RUM remains outside this trace readiness check.",
    },
    remote: options.raw ? parsed : compactStatus(parsed, options.full) ?? compactCapture(result, { full: true }),
  };
}

async function trace(config: UniDeskConfig, options: TraceOptions): Promise<Record<string, unknown>> {
  if (options.traceId === null) throw new Error("observability trace requires --trace-id <traceId>");
  const observability = readObservabilityConfig();
  const target = resolveTarget(observability, options.targetId);
  const tracePath = observability.probes.traceQueryPathTemplate.replaceAll("{{traceId}}", encodeURIComponent(options.traceId));
  const result = await capture(config, target.route, ["sh"], traceScript(observability, target, tracePath, options));
  const parsed = parseJsonOutput(result.stdout);
  return {
    ok: result.exitCode === 0 && parsed?.ok === true,
    action: "platform-infra-observability-trace",
    mutation: false,
    target: targetSummary(target),
    traceId: options.traceId,
    query: {
      backend: observability.traceBackend.type,
      service: observability.traceBackend.serviceName,
      path: tracePath,
    },
    result: redactSensitiveUnknown(parsed),
  };
}

async function search(config: UniDeskConfig, options: SearchOptions): Promise<Record<string, unknown>> {
  const observability = readObservabilityConfig();
  const target = resolveTarget(observability, options.targetId);
  const endSeconds = Math.floor(Date.now() / 1000);
  const startSeconds = endSeconds - (options.lookbackMinutes * 60);
  const tempoQuery = options.query ?? inferSearchTempoQuery(options);
  const params = new URLSearchParams({
    start: String(startSeconds),
    end: String(endSeconds),
    limit: String(options.candidateLimit),
  });
  if (tempoQuery !== null) params.set("q", tempoQuery);
  const searchPath = `/api/search?${params.toString()}`;
  const result = await capture(config, target.route, ["sh"], searchScript(observability, target, searchPath, options));
  const parsed = parseJsonOutput(result.stdout);
  return {
    ok: result.exitCode === 0 && parsed?.ok === true,
    action: "platform-infra-observability-search",
    mutation: false,
    target: targetSummary(target),
    query: {
      backend: observability.traceBackend.type,
      service: observability.traceBackend.serviceName,
      path: searchPath,
      grep: options.grep,
      tempoQuery,
      explicitTempoQuery: options.query,
      pathFilter: options.path,
      statusFilter: options.status,
      lookbackMinutes: options.lookbackMinutes,
      candidateLimit: options.candidateLimit,
      limit: options.limit,
    },
    result: redactSensitiveUnknown(parsed) ?? compactCapture(result, { full: true }),
  };
}

function inferSearchTempoQuery(options: SearchOptions): string | null {
  if (options.path !== null) return `{ .http.route = ${JSON.stringify(options.path)} }`;
  if (options.status !== null) return `{ .http.response.status_code = ${options.status} }`;
  return null;
}

async function diagnoseCodeAgent(config: UniDeskConfig, options: DiagnoseCodeAgentOptions): Promise<Record<string, unknown>> {
  const observability = readObservabilityConfig();
  const target = resolveTarget(observability, options.targetId);
  const endSeconds = Math.floor(Date.now() / 1000);
  const startSeconds = endSeconds - (options.lookbackMinutes * 60);
  let searchPath: string | null = null;
  if (options.traceId === null && options.businessTraceId !== null) {
    const params = new URLSearchParams({
      start: String(startSeconds),
      end: String(endSeconds),
      limit: String(options.candidateLimit),
      q: `{ .traceId = "${options.businessTraceId}" }`,
    });
    searchPath = `/api/search?${params.toString()}`;
  }
  const result = await capture(config, target.route, ["sh"], diagnoseCodeAgentScript(observability, target, searchPath, options));
  const parsed = parseJsonOutput(result.stdout);
  return {
    ok: result.exitCode === 0 && parsed?.ok === true,
    action: "platform-infra-observability-diagnose-code-agent",
    mutation: false,
    target: targetSummary(target),
    query: {
      backend: observability.traceBackend.type,
      service: observability.traceBackend.serviceName,
      businessTraceId: options.businessTraceId,
      traceId: options.traceId,
      path: searchPath,
      lookbackMinutes: options.lookbackMinutes,
      candidateLimit: options.candidateLimit,
      limit: options.limit,
    },
    result: redactSensitiveUnknown(parsed) ?? compactCapture(result, { full: true }),
  };
}

function renderManifest(observability: ObservabilityConfig, target: ObservabilityTarget): string {
  const collectorImage = imageReference(observability.images.collector);
  const tempoImage = imageReference(observability.images.tempo);
  return [
    target.createNamespace ? namespaceManifest(target) : "",
    allowAllNetworkPolicy(target),
    collectorConfigMap(observability, target),
    collectorDeployment(observability, target, collectorImage),
    collectorService(observability, target),
    tempoConfigMap(observability, target),
    tempoDeployment(observability, target, tempoImage),
    tempoService(observability, target),
  ].filter((item) => item.trim().length > 0).join("\n---\n");
}

function namespaceManifest(target: ObservabilityTarget): string {
  return `apiVersion: v1
kind: Namespace
metadata:
  name: ${target.namespace}
  labels:
    app.kubernetes.io/part-of: platform-infra
    app.kubernetes.io/managed-by: unidesk
    unidesk.ai/runtime-node: ${target.id}
`;
}

function allowAllNetworkPolicy(target: ObservabilityTarget): string {
  return `apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: allow-all
  namespace: ${target.namespace}
  labels:
    app.kubernetes.io/part-of: platform-infra
    app.kubernetes.io/managed-by: unidesk
spec:
  podSelector: {}
  policyTypes:
    - Ingress
    - Egress
  ingress:
    - {}
  egress:
    - {}
`;
}

function collectorConfigMap(observability: ObservabilityConfig, target: ObservabilityTarget): string {
  const config = `receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:${observability.collector.otlp.grpcPort}
      http:
        endpoint: 0.0.0.0:${observability.collector.otlp.httpPort}
processors:
  batch: {}
exporters:
  otlp/tempo:
    endpoint: ${observability.traceBackend.serviceName}.${target.namespace}.svc.cluster.local:${observability.traceBackend.otlp.grpcPort}
    tls:
      insecure: true
extensions:
  health_check:
    endpoint: 0.0.0.0:${observability.collector.healthPort}
service:
  extensions: [health_check]
  pipelines:
    traces:
      receivers: [otlp]
      processors: [batch]
      exporters: [otlp/tempo]
`;
  return `apiVersion: v1
kind: ConfigMap
metadata:
  name: ${observability.collector.configMapName}
  namespace: ${target.namespace}
  labels:
    app.kubernetes.io/name: ${observability.collector.deploymentName}
    app.kubernetes.io/component: tracing
    app.kubernetes.io/part-of: platform-infra
    app.kubernetes.io/managed-by: unidesk
  annotations:
    unidesk.ai/spec: "${observability.metadata.spec}"
data:
  collector.yaml: |
${indent(config, 4)}
`;
}

function collectorDeployment(observability: ObservabilityConfig, target: ObservabilityTarget, image: string): string {
  return `apiVersion: apps/v1
kind: Deployment
metadata:
  name: ${observability.collector.deploymentName}
  namespace: ${target.namespace}
  labels:
    app.kubernetes.io/name: ${observability.collector.deploymentName}
    app.kubernetes.io/component: tracing
    app.kubernetes.io/part-of: platform-infra
    app.kubernetes.io/managed-by: unidesk
spec:
  replicas: ${observability.collector.replicas}
  strategy:
    type: RollingUpdate
    rollingUpdate:
      maxSurge: 0
      maxUnavailable: 1
  selector:
    matchLabels:
      app.kubernetes.io/name: ${observability.collector.deploymentName}
      app.kubernetes.io/component: tracing
  template:
    metadata:
      labels:
        app.kubernetes.io/name: ${observability.collector.deploymentName}
        app.kubernetes.io/component: tracing
        app.kubernetes.io/part-of: platform-infra
      annotations:
        unidesk.ai/spec: "${observability.metadata.spec}"
    spec:
      containers:
        - name: collector
          image: ${image}
          imagePullPolicy: ${observability.images.collector.pullPolicy}
          args:
            - --config=/etc/otelcol/collector.yaml
          ports:
            - name: otlp-grpc
              containerPort: ${observability.collector.otlp.grpcPort}
            - name: otlp-http
              containerPort: ${observability.collector.otlp.httpPort}
            - name: health
              containerPort: ${observability.collector.healthPort}
          readinessProbe:
            httpGet:
              path: /
              port: health
          volumeMounts:
            - name: config
              mountPath: /etc/otelcol/collector.yaml
              subPath: collector.yaml
              readOnly: true
      volumes:
        - name: config
          configMap:
            name: ${observability.collector.configMapName}
`;
}

function collectorService(observability: ObservabilityConfig, target: ObservabilityTarget): string {
  return `apiVersion: v1
kind: Service
metadata:
  name: ${observability.collector.serviceName}
  namespace: ${target.namespace}
  labels:
    app.kubernetes.io/name: ${observability.collector.deploymentName}
    app.kubernetes.io/component: tracing
    app.kubernetes.io/part-of: platform-infra
    app.kubernetes.io/managed-by: unidesk
spec:
  type: ClusterIP
  selector:
    app.kubernetes.io/name: ${observability.collector.deploymentName}
    app.kubernetes.io/component: tracing
  ports:
    - name: otlp-grpc
      port: ${observability.collector.otlp.grpcPort}
      targetPort: otlp-grpc
    - name: otlp-http
      port: ${observability.collector.otlp.httpPort}
      targetPort: otlp-http
    - name: health
      port: ${observability.collector.healthPort}
      targetPort: health
`;
}

function tempoConfigMap(observability: ObservabilityConfig, _target: ObservabilityTarget): string {
  const config = `server:
  http_listen_port: ${observability.traceBackend.httpPort}
distributor:
  receivers:
    otlp:
      protocols:
        grpc:
          endpoint: 0.0.0.0:${observability.traceBackend.otlp.grpcPort}
        http:
          endpoint: 0.0.0.0:${observability.traceBackend.otlp.httpPort}
ingester:
  trace_idle_period: 10s
  max_block_duration: 5m
compactor:
  compaction:
    block_retention: ${observability.traceBackend.storage.retention}
storage:
  trace:
    backend: local
    wal:
      path: /var/tempo/wal
    local:
      path: /var/tempo/traces
`;
  return `apiVersion: v1
kind: ConfigMap
metadata:
  name: ${observability.traceBackend.configMapName}
  namespace: ${_target.namespace}
  labels:
    app.kubernetes.io/name: ${observability.traceBackend.deploymentName}
    app.kubernetes.io/component: trace-backend
    app.kubernetes.io/part-of: platform-infra
    app.kubernetes.io/managed-by: unidesk
  annotations:
    unidesk.ai/spec: "${observability.metadata.spec}"
data:
  tempo.yaml: |
${indent(config, 4)}
`;
}

function tempoDeployment(observability: ObservabilityConfig, target: ObservabilityTarget, image: string): string {
  return `apiVersion: apps/v1
kind: Deployment
metadata:
  name: ${observability.traceBackend.deploymentName}
  namespace: ${target.namespace}
  labels:
    app.kubernetes.io/name: ${observability.traceBackend.deploymentName}
    app.kubernetes.io/component: trace-backend
    app.kubernetes.io/part-of: platform-infra
    app.kubernetes.io/managed-by: unidesk
spec:
  replicas: ${observability.traceBackend.replicas}
  strategy:
    type: RollingUpdate
    rollingUpdate:
      maxSurge: 0
      maxUnavailable: 1
  selector:
    matchLabels:
      app.kubernetes.io/name: ${observability.traceBackend.deploymentName}
      app.kubernetes.io/component: trace-backend
  template:
    metadata:
      labels:
        app.kubernetes.io/name: ${observability.traceBackend.deploymentName}
        app.kubernetes.io/component: trace-backend
        app.kubernetes.io/part-of: platform-infra
      annotations:
        unidesk.ai/spec: "${observability.metadata.spec}"
    spec:
      containers:
        - name: tempo
          image: ${image}
          imagePullPolicy: ${observability.images.tempo.pullPolicy}
          args:
            - -config.file=/etc/tempo/tempo.yaml
          ports:
            - name: http
              containerPort: ${observability.traceBackend.httpPort}
            - name: otlp-grpc
              containerPort: ${observability.traceBackend.otlp.grpcPort}
            - name: otlp-http
              containerPort: ${observability.traceBackend.otlp.httpPort}
          readinessProbe:
            httpGet:
              path: ${observability.probes.readinessPath}
              port: http
          volumeMounts:
            - name: config
              mountPath: /etc/tempo/tempo.yaml
              subPath: tempo.yaml
              readOnly: true
            - name: data
              mountPath: /var/tempo
      volumes:
        - name: config
          configMap:
            name: ${observability.traceBackend.configMapName}
        - name: data
          emptyDir: {}
`;
}

function tempoService(observability: ObservabilityConfig, target: ObservabilityTarget): string {
  return `apiVersion: v1
kind: Service
metadata:
  name: ${observability.traceBackend.serviceName}
  namespace: ${target.namespace}
  labels:
    app.kubernetes.io/name: ${observability.traceBackend.deploymentName}
    app.kubernetes.io/component: trace-backend
    app.kubernetes.io/part-of: platform-infra
    app.kubernetes.io/managed-by: unidesk
spec:
  type: ClusterIP
  selector:
    app.kubernetes.io/name: ${observability.traceBackend.deploymentName}
    app.kubernetes.io/component: trace-backend
  ports:
    - name: http
      port: ${observability.traceBackend.httpPort}
      targetPort: http
    - name: otlp-grpc
      port: ${observability.traceBackend.otlp.grpcPort}
      targetPort: otlp-grpc
    - name: otlp-http
      port: ${observability.traceBackend.otlp.httpPort}
      targetPort: otlp-http
`;
}

function applyScript(params: {
  yaml: string;
  target: ObservabilityTarget;
  dryRun: boolean;
  wait: boolean;
  collectorDeploymentName: string;
  backendDeploymentName: string;
}): string {
  const encoded = Buffer.from(params.yaml, "utf8").toString("base64");
  const dryRunArg = params.dryRun ? "--dry-run=server" : "";
  const waitDisposition = params.wait ? "skipped-short-connection-use-status" : "skipped";
  return `
set -u
tmp="$(mktemp -d)"
trap 'rm -rf "$tmp"' EXIT
manifest="$tmp/platform-infra-observability.yaml"
printf '%s' '${encoded}' | base64 -d > "$manifest"
kubectl apply --server-side --field-manager=${fieldManager} ${dryRunArg} -f "$manifest" >"$tmp/apply.out" 2>"$tmp/apply.err"
apply_rc=$?
collector_rollout_rc=0
backend_rollout_rc=0
wait_disposition=${waitDisposition}
python3 - "$apply_rc" "$collector_rollout_rc" "$backend_rollout_rc" "$wait_disposition" "$tmp/apply.out" "$tmp/apply.err" "$tmp/collector-rollout.out" "$tmp/collector-rollout.err" "$tmp/backend-rollout.out" "$tmp/backend-rollout.err" <<'PY'
import json, os, sys
def text(path, limit=6000):
    try:
        return open(path, encoding="utf-8", errors="replace").read()[-limit:]
    except FileNotFoundError:
        return ""
apply_rc = int(sys.argv[1])
collector_rc = int(sys.argv[2])
backend_rc = int(sys.argv[3])
payload = {
  "ok": apply_rc == 0 and collector_rc == 0 and backend_rc == 0,
  "target": "${params.target.id}",
  "namespace": "${params.target.namespace}",
  "dryRun": ${params.dryRun ? "True" : "False"},
  "apply": {"exitCode": apply_rc, "stdout": text(sys.argv[5]), "stderr": text(sys.argv[6])},
  "rollout": {
    "disposition": sys.argv[4],
    "collector": {"exitCode": collector_rc, "stdout": text(sys.argv[7]), "stderr": text(sys.argv[8])},
    "backend": {"exitCode": backend_rc, "stdout": text(sys.argv[9]), "stderr": text(sys.argv[10])},
  },
}
print(json.dumps(payload, ensure_ascii=False, indent=2))
sys.exit(0 if payload["ok"] else 1)
PY
`;
}

function statusScript(observability: ObservabilityConfig, target: ObservabilityTarget, full: boolean, generateValidationTrace = false): string {
  const endpointsJson = JSON.stringify(observability.probes.statusEndpoints);
  return `
set -u
tmp="$(mktemp -d)"
trap 'rm -rf "$tmp"' EXIT
capture_json() {
  name="$1"
  shift
  "$@" >"$tmp/$name.json" 2>"$tmp/$name.err"
  echo $? >"$tmp/$name.rc"
}
capture_raw() {
  name="$1"
  shift
  "$@" >"$tmp/$name.out" 2>"$tmp/$name.err"
  echo $? >"$tmp/$name.rc"
}
capture_json namespace kubectl get namespace ${shQuote(target.namespace)} -o json
capture_json deployments kubectl -n ${shQuote(target.namespace)} get deployment ${shQuote(observability.collector.deploymentName)} ${shQuote(observability.traceBackend.deploymentName)} -o json
capture_json services kubectl -n ${shQuote(target.namespace)} get service ${shQuote(observability.collector.serviceName)} ${shQuote(observability.traceBackend.serviceName)} -o json
capture_json pods kubectl -n ${shQuote(target.namespace)} get pods -l ${shQuote(`app.kubernetes.io/name in (${observability.collector.deploymentName},${observability.traceBackend.deploymentName})`)} -o json
capture_json events kubectl -n ${shQuote(target.namespace)} get events -o json
python3 - "$tmp" '${endpointsJson.replaceAll("'", "'\"'\"'")}' <<'PY'
import json, random, socket, subprocess, sys, time, urllib.error, urllib.request
tmp = sys.argv[1]
endpoints = json.loads(sys.argv[2])
namespace = "${target.namespace}"
def read(path, binary=False, limit=8000):
    try:
        mode = "rb" if binary else "r"
        with open(path, mode, encoding=None if binary else "utf-8", errors=None if binary else "replace") as fh:
            data = fh.read()
        if binary:
            data = data.decode("utf-8", errors="replace")
        return data[-limit:]
    except FileNotFoundError:
        return ""
def rc(name):
    try:
        return int(read(f"{tmp}/{name}.rc").strip() or "1")
    except ValueError:
        return 1
def parsed_json(name):
    raw = read(f"{tmp}/{name}.json", limit=1000000)
    try:
        return json.loads(raw) if raw else None
    except Exception:
        return None
def compact_section(name):
    return {
        "exitCode": rc(name),
        "stderrTail": read(f"{tmp}/{name}.err", limit=2000),
        "json": parsed_json(name),
    }
probe_results = []
for ep in endpoints:
    path = ep.get("path") or "/"
    if not path.startswith("/"):
        path = "/" + path
    proxy_path = f"/api/v1/namespaces/{namespace}/services/http:{ep['service']}:{ep['portName']}/proxy{path}"
    proc = subprocess.run(["kubectl", "get", "--raw", proxy_path], text=True, capture_output=True, timeout=20)
    probe_results.append({
        "name": ep["name"],
        "service": ep["service"],
        "portName": ep["portName"],
        "path": path,
        "exitCode": proc.returncode,
        "stdoutTail": proc.stdout[-1000:],
        "stderrTail": proc.stderr[-1000:],
        "ok": proc.returncode == 0,
    })
def free_port():
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.bind(("127.0.0.1", 0))
    port = sock.getsockname()[1]
    sock.close()
    return port
def parse_body(raw):
    try:
        return json.loads(raw) if raw else None
    except Exception:
        return raw[-2000:]
def http_request(method, url, body=None, timeout=10):
    data = None
    headers = {}
    if body is not None:
        data = json.dumps(body).encode("utf-8")
        headers["Content-Type"] = "application/json"
    req = urllib.request.Request(url, data=data, headers=headers, method=method)
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            raw = resp.read(200000).decode("utf-8", errors="replace")
            status = int(getattr(resp, "status", 0) or 0)
            return {"ok": 200 <= status < 300, "status": status, "body": parse_body(raw)}
    except urllib.error.HTTPError as exc:
        raw = exc.read(200000).decode("utf-8", errors="replace")
        return {"ok": False, "status": exc.code, "body": parse_body(raw)}
    except Exception as exc:
        return {"ok": False, "error": type(exc).__name__ + ": " + str(exc)}
def wait_proxy(base_url):
    last = None
    for _ in range(30):
        last = http_request("GET", base_url + "/version", timeout=1)
        if last.get("ok") is True:
            return last
        time.sleep(0.2)
    return last or {"ok": False, "error": "kubectl proxy did not answer"}
def compact_http(result):
    if not isinstance(result, dict):
        return {"ok": False, "error": "missing-result"}
    compact = {"ok": result.get("ok") is True}
    if "status" in result:
        compact["status"] = result.get("status")
    if "error" in result:
        compact["error"] = result.get("error")
    body = result.get("body")
    if isinstance(body, dict):
        compact["bodyKeys"] = sorted(list(body.keys()))[:12]
    elif isinstance(body, str) and body:
        compact["bodyTail"] = body[-1000:]
    return compact
def validation_trace_payload(trace_id, span_id):
    now = time.time_ns()
    return {
        "resourceSpans": [{
            "resource": {"attributes": [
                {"key": "service.name", "value": {"stringValue": "unidesk-observability-validate"}},
                {"key": "deployment.environment", "value": {"stringValue": "${target.id}"}},
                {"key": "unidesk.node", "value": {"stringValue": "${target.id}"}},
                {"key": "k8s.namespace.name", "value": {"stringValue": namespace}},
            ]},
            "scopeSpans": [{
                "scope": {"name": "unidesk-cli-platform-infra-observability"},
                "spans": [{
                    "traceId": trace_id,
                    "spanId": span_id,
                    "name": "unidesk.observability.validate",
                    "kind": 1,
                    "startTimeUnixNano": str(now),
                    "endTimeUnixNano": str(now + 1000000),
                    "attributes": [
                        {"key": "unidesk.issue", "value": {"stringValue": "489"}},
                        {"key": "unidesk.spec", "value": {"stringValue": "${observability.metadata.spec}"}},
                        {"key": "unidesk.validation", "value": {"stringValue": "platform-infra-observability"}},
                    ],
                    "status": {"code": 1},
                }],
            }],
        }],
    }
def generate_test_trace():
    trace_id = ("%032x" % random.getrandbits(128))
    span_id = ("%016x" % random.getrandbits(64))
    port = free_port()
    base_url = "http://127.0.0.1:%s" % port
    stdout_path = f"{tmp}/kubectl-proxy.out"
    stderr_path = f"{tmp}/kubectl-proxy.err"
    with open(stdout_path, "w", encoding="utf-8") as stdout, open(stderr_path, "w", encoding="utf-8") as stderr:
        proc = subprocess.Popen(
            ["kubectl", "proxy", "--address=127.0.0.1", "--port", str(port), "--accept-hosts=^127\\\\.0\\\\.0\\\\.1$"],
            stdout=stdout,
            stderr=stderr,
            text=True,
        )
    try:
        proxy_ready = wait_proxy(base_url)
        if proxy_ready.get("ok") is not True:
            return {
                "ok": False,
                "stage": "proxy-start",
                "traceId": trace_id,
                "proxy": compact_http(proxy_ready),
                "proxyStderrTail": read(stderr_path, limit=2000),
            }
        collector_url = base_url + "/api/v1/namespaces/%s/services/http:${observability.collector.serviceName}:otlp-http/proxy/v1/traces" % namespace
        tempo_url = base_url + "/api/v1/namespaces/%s/services/http:${observability.traceBackend.serviceName}:http/proxy/api/traces/%s" % (namespace, trace_id)
        ingest = http_request("POST", collector_url, validation_trace_payload(trace_id, span_id), timeout=15)
        query = None
        if ingest.get("ok") is True:
            for _ in range(20):
                query = http_request("GET", tempo_url, timeout=10)
                if query.get("ok") is True:
                    break
                time.sleep(1)
        return {
            "ok": ingest.get("ok") is True and isinstance(query, dict) and query.get("ok") is True,
            "stage": "query" if ingest.get("ok") is True else "ingest",
            "traceId": trace_id,
            "spanId": span_id,
            "collectorEndpoint": "${observability.collector.serviceName}:otlp-http",
            "backendEndpoint": "${observability.traceBackend.serviceName}:http",
            "ingest": compact_http(ingest),
            "query": compact_http(query),
            "valuesPrinted": False,
        }
    finally:
        proc.terminate()
        try:
            proc.wait(timeout=3)
        except subprocess.TimeoutExpired:
            proc.kill()
            proc.wait(timeout=3)
runtime_ready = rc("namespace") == 0 and rc("deployments") == 0 and rc("services") == 0 and all(item["ok"] for item in probe_results)
validation_trace = generate_test_trace() if (${generateValidationTrace ? "True" : "False"} and runtime_ready) else None
payload = {
    "ok": runtime_ready and (validation_trace is None or validation_trace.get("ok") is True),
    "target": "${target.id}",
    "namespace": namespace,
    "sections": {
        "namespace": compact_section("namespace"),
        "deployments": compact_section("deployments"),
        "services": compact_section("services"),
        "pods": compact_section("pods"),
        "events": compact_section("events"),
    },
    "probes": probe_results,
    "validationTrace": validation_trace,
}
print(json.dumps(payload, ensure_ascii=False, indent=2 if ${full ? "True" : "False"} else None))
sys.exit(0 if payload["ok"] else 1)
PY
`;
}

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",
    "db.system", "db.operation.name", "db.sql.table", "db.query.arg_count",
    "hwlab.runtime.stage", "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()
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"))
            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."):
                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],
  "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
`;
}

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, 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
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.target", "http.url", "url.path",
    "db.system", "db.operation.name", "db.sql.table", "db.query.arg_count",
    "hwlab.runtime.stage", "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:],
    }

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 summary.get("matchingActive") is True:
        break

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,
    "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 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 searchPathLiteral = searchPath === null ? "None" : JSON.stringify(searchPath);
  const searchProxyPathLiteral = searchProxyPath === null ? "None" : JSON.stringify(searchProxyPath);
  const businessTraceIdForNext = options.businessTraceId ?? "<trc_...>";
  return `
set -u
python3 - <<'PY'
import collections, json, subprocess, time

BUSINESS_TRACE_ID = ${businessTraceIdLiteral}
TRACE_ID = ${traceIdLiteral}
SEARCH_PATH = ${searchPathLiteral}
SEARCH_PROXY_PATH = ${searchProxyPathLiteral}
TRACE_PROXY_PREFIX = ${JSON.stringify(proxyPrefix)}
TRACE_PATH_TEMPLATE = ${JSON.stringify(observability.probes.traceQueryPathTemplate)}
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",
    "db.system", "db.operation.name", "db.sql.table", "db.query.arg_count",
    "hwlab.runtime.stage", "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 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",
    ):
        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):
    terminal_spans = []
    latest_seq = 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]
            terminal_spans.append({
                "status": derived,
                "eventType": attrs.get("eventType"),
                "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
    if terminal_status in ("completed", "succeeded", "success"):
        runner_classification = "completed"
    elif terminal_status in ("failed", "error", "timeout", "cancelled"):
        runner_classification = "failed"
    elif terminal_status in ("blocked",):
        runner_classification = "blocked"
    elif any(str(item.get("service") or "") == "agentrun-runner" for item in spans):
        runner_classification = "running"
    else:
        runner_classification = "unknown"
    return {
        "terminalStatus": terminal_status,
        "latestSeq": latest_seq,
        "terminalEventType": terminal_event_type,
        "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": "hwlab_projection_stale",
            "label": "HWLAB projection stale",
            "confidence": 0.94,
            "summary": "HWLAB projection 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": "hwlab_projection_lag_suspected",
            "label": "HWLAB projection stale",
            "confidence": 0.64,
            "summary": "HWLAB projection stale is suspected because AgentRun is terminal but HWLAB read-model evidence is incomplete.",
            "evidence": lag_summary.get("reasons", []),
        })
    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,
            "otelTraceId": TRACE_ID,
            "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": "business-trace-id",
        "businessTraceId": BUSINESS_TRACE_ID,
        "otelTraceId": selected_trace_id,
        "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 business trace"
    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": "bun scripts/cli.ts platform-infra observability diagnose-code-agent --target ${target.id} --business-trace-id ${businessTraceIdForNext} --lookback-minutes 10080 --candidate-limit 500 --full",
            "search": "bun scripts/cli.ts platform-infra observability search --target ${target.id} --query '{ .traceId = \\"${businessTraceIdForNext}\\" }' --lookback-minutes 10080 --candidate-limit 500 --limit 10",
        },
    }
    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)
agentrun = agentrun_summary(spans)
read_model = hwlab_read_model_summary(spans)
lag = projection_lag_summary(agentrun, read_model)
identity = identity_from_spans(spans)
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")
if agentrun.get("terminalStatus") == "completed":
    facts.append("AgentRun completed")
if lag.get("status") in ("confirmed", "suspected"):
    facts.append("HWLAB projection stale")
if idle_warning_spans and agentrun.get("terminalStatus") 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"),
    },
}
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"),
        "latestSeq": agentrun.get("latestSeq"),
        "terminalEventType": agentrun.get("terminalEventType"),
        "runnerProviderClassification": agentrun.get("runnerProviderClassification"),
    },
    "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": "bun scripts/cli.ts platform-infra observability diagnose-code-agent --target ${target.id} --business-trace-id ${businessTraceIdForNext} --full",
        "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
`;
}

function configSummary(observability: ObservabilityConfig, target: ObservabilityTarget): Record<string, unknown> {
  return {
    configPath: configLabel,
    spec: observability.metadata.spec,
    target: targetSummary(target),
    images: {
      collector: imageReference(observability.images.collector),
      traceBackend: imageReference(observability.images.tempo),
    },
    collector: observability.collector,
    traceBackend: observability.traceBackend,
    sampling: observability.sampling,
  };
}

function targetSummary(target: ObservabilityTarget): Record<string, unknown> {
  return {
    id: target.id,
    route: target.route,
    namespace: target.namespace,
    role: target.role,
    createNamespace: target.createNamespace,
  };
}

function policyChecks(yaml: string, target: ObservabilityTarget): Array<Record<string, unknown>> {
  return [
    { name: "yaml-source-of-truth", ok: true, detail: "All concrete images, routes, namespace, ports, retention and sampling values are read from config/platform-infra/observability.yaml." },
    { name: "clusterip-only", ok: !/^\s*type:\s*(NodePort|LoadBalancer)\s*$/mu.test(yaml), detail: "Collector and trace backend stay ClusterIP-only." },
    { name: "no-ingress", ok: !/^\s*kind:\s*Ingress\s*$/mu.test(yaml), detail: "No public ingress is rendered for the first tracing backend." },
    { name: "no-host-network", ok: !/^\s*hostNetwork:\s*true\s*$/mu.test(yaml), detail: "Pods must not use host network." },
    { name: "allow-all-network-policy", ok: yaml.includes("kind: NetworkPolicy") && yaml.includes("name: allow-all") && yaml.includes(`namespace: ${target.namespace}`), detail: `NetworkPolicy/allow-all is rendered in ${target.namespace}.` },
  ];
}

function statusSummary(payload: Record<string, unknown>): Record<string, unknown> {
  const sections = asRecord(payload.sections, "status.sections");
  const deployments = objectList(sectionJson(sections, "deployments"));
  const services = objectList(sectionJson(sections, "services"));
  const pods = objectList(sectionJson(sections, "pods"));
  const events = objectList(sectionJson(sections, "events"));
  const podEvents = latestPodEvents(events);
  const probes = Array.isArray(payload.probes) ? payload.probes as Array<Record<string, unknown>> : [];
  const readyDeployments = deployments.map((item) => deploymentSummary(item));
  return {
    ready: payload.ok === true,
    namespace: payload.namespace,
    deployments: readyDeployments,
    services: services.map((item) => metadataName(item)),
    pods: pods.map((item) => podSummary(item, podEvents)),
    probes: probes.map((item) => ({
      name: item.name,
      ok: item.ok === true,
      service: item.service,
      path: item.path,
      stderrTail: item.ok === true ? "" : item.stderrTail,
    })),
  };
}

function compactStatus(payload: Record<string, unknown> | null, full: boolean): Record<string, unknown> | null {
  if (payload === null) return null;
  if (full) return redactSensitiveUnknown(payload) as Record<string, unknown>;
  return statusSummary(payload);
}

function sectionJson(sections: Record<string, unknown>, name: string): unknown {
  const section = asRecord(sections[name], `sections.${name}`);
  return section.json;
}

function objectList(value: unknown): Record<string, unknown>[] {
  if (typeof value !== "object" || value === null) return [];
  const items = (value as Record<string, unknown>).items;
  if (!Array.isArray(items)) return [];
  return items.filter((item): item is Record<string, unknown> => typeof item === "object" && item !== null && !Array.isArray(item));
}

function deploymentSummary(item: Record<string, unknown>): Record<string, unknown> {
  const spec = item.spec as Record<string, unknown> | undefined;
  const status = item.status as Record<string, unknown> | undefined;
  const replicas = typeof spec?.replicas === "number" ? spec.replicas : null;
  const available = typeof status?.availableReplicas === "number" ? status.availableReplicas : 0;
  return {
    name: metadataName(item),
    replicas,
    availableReplicas: available,
    ready: replicas !== null && available >= replicas,
  };
}

function podSummary(item: Record<string, unknown>, podEvents: Map<string, Record<string, unknown>>): Record<string, unknown> {
  const status = item.status as Record<string, unknown> | undefined;
  const waiting = firstWaitingContainer(status);
  const scheduled = conditionSummary(status, "PodScheduled");
  const ready = conditionSummary(status, "Ready");
  const name = metadataName(item);
  const showEvent = status?.phase !== "Running" || waiting !== null || scheduled !== null || ready !== null;
  const latestEvent = showEvent && name !== null ? podEvents.get(name) ?? null : null;
  return {
    name,
    phase: status?.phase ?? null,
    reason: waiting?.reason ?? scheduled?.reason ?? ready?.reason ?? null,
    message: waiting?.message ?? scheduled?.message ?? ready?.message ?? null,
    latestEvent: latestEvent === null ? null : {
      reason: stringValue(latestEvent.reason),
      message: stringValue(latestEvent.message),
      type: stringValue(latestEvent.type),
      count: typeof latestEvent.count === "number" ? latestEvent.count : null,
      timestamp: eventTimestamp(latestEvent),
    },
  };
}

function latestPodEvents(events: Record<string, unknown>[]): Map<string, Record<string, unknown>> {
  const byPod = new Map<string, Record<string, unknown>>();
  for (const item of events) {
    const involved = item.involvedObject;
    if (typeof involved !== "object" || involved === null || Array.isArray(involved)) continue;
    const podName = (involved as Record<string, unknown>).name;
    const kind = (involved as Record<string, unknown>).kind;
    if (kind !== "Pod" || typeof podName !== "string") continue;
    const current = byPod.get(podName);
    if (current === undefined || eventTimestamp(item) > eventTimestamp(current)) byPod.set(podName, item);
  }
  return byPod;
}

function eventTimestamp(item: Record<string, unknown>): string | null {
  const candidate = stringValue(item.eventTime)
    ?? stringValue(item.lastTimestamp)
    ?? stringValue(item.deprecatedLastTimestamp)
    ?? (typeof item.metadata === "object" && item.metadata !== null && !Array.isArray(item.metadata)
      ? stringValue((item.metadata as Record<string, unknown>).creationTimestamp)
      : null);
  return candidate;
}

function stringValue(value: unknown): string | null {
  return typeof value === "string" && value.length > 0 ? value : null;
}

function firstWaitingContainer(status: Record<string, unknown> | undefined): { reason: string | null; message: string | null } | null {
  const statuses = Array.isArray(status?.containerStatuses) ? status.containerStatuses : [];
  for (const item of statuses) {
    if (typeof item !== "object" || item === null || Array.isArray(item)) continue;
    const state = (item as Record<string, unknown>).state;
    if (typeof state !== "object" || state === null || Array.isArray(state)) continue;
    const waiting = (state as Record<string, unknown>).waiting;
    if (typeof waiting !== "object" || waiting === null || Array.isArray(waiting)) continue;
    const record = waiting as Record<string, unknown>;
    return {
      reason: typeof record.reason === "string" ? record.reason : null,
      message: typeof record.message === "string" ? record.message : null,
    };
  }
  return null;
}

function conditionSummary(status: Record<string, unknown> | undefined, type: string): { reason: string | null; message: string | null } | null {
  const conditions = Array.isArray(status?.conditions) ? status.conditions : [];
  for (const item of conditions) {
    if (typeof item !== "object" || item === null || Array.isArray(item)) continue;
    const record = item as Record<string, unknown>;
    if (record.type !== type || record.status === "True") continue;
    return {
      reason: typeof record.reason === "string" ? record.reason : null,
      message: typeof record.message === "string" ? record.message : null,
    };
  }
  return null;
}

function metadataName(item: Record<string, unknown>): string | null {
  const metadata = item.metadata as Record<string, unknown> | undefined;
  return typeof metadata?.name === "string" ? metadata.name : null;
}

function manifestObjectSummary(yaml: string): Array<Record<string, string>> {
  const docs = yaml.split(/^---$/mu);
  return docs.map((doc) => {
    const kind = doc.match(/^\s*kind:\s*(.+)$/mu)?.[1]?.trim() ?? "unknown";
    const name = doc.match(/^\s*name:\s*(.+)$/mu)?.[1]?.trim() ?? "unknown";
    return { kind, name };
  });
}

function imageReference(image: ImageSpec): string {
  return `${image.repository}:${image.tag}`;
}

function indent(value: string, spaces: number): string {
  const prefix = " ".repeat(spaces);
  return value.split("\n").map((line) => `${prefix}${line}`).join("\n");
}