// SPEC: PJ2026-01060307 控制面模块化 draft-2026-06-25-p0. runtime-common module for scripts/src/hwlab-node-impl.ts.

// Moved mechanically from scripts/src/hwlab-node-impl.ts:1600-1885 for #903.

// SPEC: PJ2026-01060505 Workbench Performance draft-2026-06-17-p0.
// SPEC: PJ2026-01060508 Web哨兵 draft-2026-06-25-p0-web-probe-sentinel.
// Responsibility: YAML-first node/lane operations, including Workbench observability control commands.
import { createHash, randomBytes } from "node:crypto";
import { existsSync, mkdirSync, readFileSync, renameSync, writeFileSync } from "node:fs";
import { dirname, join } from "node:path";
import { repoRoot, rootPath, type Config } from "../config";
import { runCommand, type CommandResult } from "../command";
import { startJob } from "../jobs";
import { classifySshTcpPoolFailure } from "../ssh";
import { HWLAB_NODE_CONTROL_PLANE_CONFIG_PATH, hwlabNodeControlPlaneInfraHelp, runHwlabNodeControlPlaneInfra } from "../hwlab-node-control-plane";
import { hwlabRuntimeLaneConfigPath, hwlabRuntimeLaneIds, hwlabRuntimeLaneSpec, hwlabRuntimeLaneSpecForNode, hwlabRuntimeNodeIds, isHwlabRuntimeLane, type HwlabRuntimeLane, type HwlabRuntimeLaneSpec, type HwlabRuntimeObservabilityRecordingRuleSpec, type HwlabRuntimeObservabilitySpec, type HwlabRuntimeObservabilityWarningAlertSpec, type HwlabRuntimePublicExposureSpec, type HwlabRuntimeWebProbeAlertThresholdsSpec, type HwlabRuntimeWebProbeProjectManagementSpec } from "../hwlab-node-lanes";
import { nodeWebProbeScriptRunnerSource } from "../hwlab-node-web-probe-runner-source";
import { nodeWebObserveAnalyzerSource } from "../hwlab-node-web-observe-analyzer-source";
import { nodeWebObserveRunnerSource } from "../hwlab-node-web-observe-runner-source";
import { nodeWebObserveCollectViewNodeScript, parseNodeWebProbeObserveCollectView, type NodeWebProbeObserveCollectView } from "../hwlab-node-web-observe-collect";
import { withWebObserveCollectRendered, withWebObserveCommandRendered, withWebObserveStatusRendered } from "../hwlab-node-web-observe-render";
import { buildWebObserveWrapperForObserveOptions, webObserveWrapperStateDirFromStatus } from "../hwlab-node-web-observe-wrapper";
import { renderWebObserveWrapperContract } from "../hwlab-node-web-observe-wrapper-render";
import { runWebProbeSentinelCommand, type WebProbeSentinelOptions } from "../hwlab-node-web-sentinel-cicd";
import { hwlabNodeHelp, hwlabNodeObservabilityHelp, hwlabNodeWebProbeHelp } from "../hwlab-node-help";
import { compactWebProbeResult, compactWebProbeScriptResult } from "../hwlab-node-web-probe-summary";
import { nodeObservabilityRecordingRuleExpression, nodeObservabilityRecordingRuleSummaries, nodeObservabilityWarningAlertExpression, nodeObservabilityWarningAlertSummaries } from "../hwlab-node-observability-promql";
import { runDelegatedHwlabNodeCommand, type DelegatedNodeDomain } from "../hwlab-node-transport";
import type { RenderedCliResult } from "../output";

import type { NodeRuntimeGitMirrorTargetSpec } from "./entry";
import { commandResultFromAsync, isCommandSuccess } from "./cleanup";
import { nodeRuntimeExpected, parseNodeScopedDelegatedOptions } from "./plan";
import { nodeRuntimeRenderToken } from "./render";
import { parseJsonObject, record, shellQuote } from "./utils";

export function metricNameFromPrometheusLine(line: string): string | null {
  const match = /^([A-Za-z_:][A-Za-z0-9_:]*)/u.exec(line);
  return match?.[1] ?? null;
}

export function prometheusLineMatchesMetric(line: string, name: string): boolean {
  if (!line.startsWith(name)) return false;
  const next = line.charAt(name.length);
  return next === "" || next === "{" || /\s/u.test(next);
}

export function prometheusLineHasLabel(line: string, name: string, value: string): boolean {
  return line.includes(`${name}="${value.replace(/\\/gu, "\\\\").replace(/"/gu, '\\"')}"`);
}

export function isWorkbenchBackendEventMetric(name: string): boolean {
  return name.startsWith("hwlab_workbench_event_") || name.startsWith("hwlab_workbench_backend_event_");
}

export function compactPrometheusLines(lines: readonly string[]): string[] {
  return lines.slice(0, 16).map((line) => line.length > 320 ? `${line.slice(0, 317)}...` : line);
}

export function parseJsonRecordFromText(text: string): Record<string, unknown> {
  const trimmed = text.trim();
  if (trimmed.length === 0) return {};
  try {
    const parsed = JSON.parse(trimmed) as unknown;
    return record(parsed);
  } catch {
    const start = trimmed.indexOf("{");
    const end = trimmed.lastIndexOf("}");
    if (start >= 0 && end > start) {
      try {
        return record(JSON.parse(trimmed.slice(start, end + 1)) as unknown);
      } catch {}
    }
  }
  return {};
}

export function k8sServiceSelector(serviceJson: Record<string, unknown>): Record<string, string> {
  const spec = record(serviceJson.spec);
  const selector = record(spec.selector);
  return Object.fromEntries(Object.entries(selector).filter((entry): entry is [string, string] => typeof entry[1] === "string" && entry[1].length > 0));
}

export function labelSelectorText(selector: Record<string, string>): string {
  return Object.entries(selector).map(([key, value]) => `${key}=${value}`).join(",");
}

export function selectK8sPod(podsJson: Record<string, unknown>, containerName: string): { name: string; phase: string | null; ready: boolean; containerReady: boolean } | null {
  const items = Array.isArray(podsJson.items) ? podsJson.items.map(record) : [];
  const pods = items.map((item) => {
    const metadata = record(item.metadata);
    const status = record(item.status);
    const name = typeof metadata.name === "string" ? metadata.name : "";
    const phase = typeof status.phase === "string" ? status.phase : null;
    const containerStatuses = Array.isArray(status.containerStatuses) ? status.containerStatuses.map(record) : [];
    const targetContainer = containerStatuses.find((container) => container.name === containerName);
    const containerReady = targetContainer?.ready === true;
    return {
      name,
      phase,
      ready: name.length > 0 && phase === "Running" && containerReady,
      containerReady,
    };
  }).filter((pod) => pod.name.length > 0);
  return pods.find((pod) => pod.ready) ?? pods.find((pod) => pod.phase === "Running") ?? pods[0] ?? null;
}

export function compactRuntimeCommand(result: CommandResult): Record<string, unknown> {
  return {
    exitCode: result.exitCode,
    stdoutBytes: Buffer.byteLength(result.stdout),
    stderrBytes: Buffer.byteLength(result.stderr),
    stdoutTail: result.stdout.trim().slice(-2000),
    stderrTail: result.stderr.trim().slice(-2000),
    timedOut: result.timedOut,
  };
}

export function compactRuntimeCommandStats(result: CommandResult): Record<string, unknown> {
  return {
    exitCode: result.exitCode,
    stdoutBytes: Buffer.byteLength(result.stdout),
    stderrBytes: Buffer.byteLength(result.stderr),
    timedOut: result.timedOut,
  };
}

export function parseLastJsonLineObject(text: string): Record<string, unknown> {
  for (const line of text.split(/\r?\n/u).reverse()) {
    const trimmed = line.trim();
    if (!trimmed.startsWith("{") || !trimmed.endsWith("}")) continue;
    try {
      return record(JSON.parse(trimmed) as unknown);
    } catch {
      // Keep scanning: command tails can mix kubectl status, git output, and JSON payloads.
    }
  }
  return {};
}

export function escapeRegExp(value: string): string {
  return value.replace(/[.*+?^${}()|[\]\\]/gu, "\\$&");
}

export function nodeRuntimeGitMirrorRefSources(scoped: ReturnType<typeof parseNodeScopedDelegatedOptions>, mirror: NodeRuntimeGitMirrorTargetSpec): Record<string, unknown> {
  return {
    localSource: `refs/heads/${mirror.sourceBranch}`,
    githubSource: `refs/mirror-stage/heads/${mirror.sourceBranch}`,
    localGitops: `refs/heads/${mirror.gitopsBranch}`,
    githubGitops: `refs/mirror-stage/heads/${mirror.gitopsBranch}`,
    githubFieldsAreMirrorStageCache: true,
    cacheRefresh: `bun scripts/cli.ts hwlab nodes git-mirror sync --node ${scoped.node} --lane ${scoped.lane} --confirm --wait`,
  };
}

export function nodeRuntimeGitMirrorFlushPartialSuccess(scoped: ReturnType<typeof parseNodeScopedDelegatedOptions>, mirror: NodeRuntimeGitMirrorTargetSpec, result: CommandResult): Record<string, unknown> | null {
  if (scoped.action !== "flush") return null;
  const text = `${result.stdout}\n${result.stderr}`;
  const payload = parseLastJsonLineObject(text);
  const partialSuccess = typeof payload.partialSuccess === "string" && payload.partialSuccess.length > 0
    ? payload.partialSuccess
    : null;
  const payloadStatus = typeof payload.status === "string" && payload.status.length > 0 ? payload.status : null;
  const branch = escapeRegExp(mirror.gitopsBranch);
  const pushSucceededInGitOutput = new RegExp(`\\b${branch}\\s*->\\s*${branch}\\b`, "u").test(text);
  const postPushFetchFailed = /kex_exchange_identification|Connection closed by remote host|Could not read from remote repository|fatal:.*fetch|fetch-pack|early EOF/iu.test(text);
  if (partialSuccess !== "push-succeeded-fetch-failed" && !(result.exitCode !== 0 && pushSucceededInGitOutput && postPushFetchFailed)) {
    return null;
  }
  return {
    status: payloadStatus ?? "partial-success",
    partialSuccess: "push-succeeded-fetch-failed",
    degradedReason: "node-runtime-git-mirror-flush-post-push-fetch-failed",
    retryable: true,
    message: "GitOps push appears to have succeeded, but the post-push fetch/recheck failed. Refresh mirror-stage cache before deciding another flush is needed.",
    payload: Object.keys(payload).length > 0 ? payload : null,
    refSources: nodeRuntimeGitMirrorRefSources(scoped, mirror),
    next: {
      sync: `bun scripts/cli.ts hwlab nodes git-mirror sync --node ${scoped.node} --lane ${scoped.lane} --confirm --wait`,
      status: `bun scripts/cli.ts hwlab nodes git-mirror status --node ${scoped.node} --lane ${scoped.lane}`,
    },
  };
}

export function sshTcpPoolDiagnosticsFromCommand(spec: HwlabRuntimeLaneSpec, result: CommandResult): Record<string, unknown> | null {
  if (isCommandSuccess(result)) return null;
  const failureKind = classifySshTcpPoolFailure(`${result.stderr}\n${result.stdout}`);
  if (failureKind === null) return null;
  return {
    classification: "ssh-tcp-pool-transient",
    failureKind,
    providerId: spec.nodeId,
    route: spec.nodeKubeRoute,
    message: "SSH tcp data pool failed while running the controlled command; treat this as transport/data-pool transient until provider labels and retry say otherwise.",
    next: {
      poolStatus: `bun scripts/cli.ts debug ssh-pool ${spec.nodeId}`,
      retrySmoke: `trans ${spec.nodeId} argv true`,
      retryApply: `bun scripts/cli.ts hwlab nodes control-plane apply --node ${spec.nodeId} --lane ${spec.lane} --confirm`,
      retryTriggerCurrent: `bun scripts/cli.ts hwlab nodes control-plane trigger-current --node ${spec.nodeId} --lane ${spec.lane} --confirm`,
    },
  };
}

export function nodeRuntimeUnsupportedAction(scoped: ReturnType<typeof parseNodeScopedDelegatedOptions>): Record<string, unknown> {
  return {
    ok: false,
    command: `hwlab nodes control-plane ${scoped.action} --node ${scoped.node} --lane ${scoped.lane}`,
    node: scoped.node,
    lane: scoped.lane,
    mutation: false,
    degradedReason: "unsupported-node-scoped-runtime-action",
    message: "node-scoped runtime currently supports plan/status/apply/refresh/sync/trigger-current/cleanup-runs/runtime-image/runtime-migration",
    expected: nodeRuntimeExpected(scoped.spec),
  };
}

export function transPath(): string {
  return join(repoRoot, "scripts", "trans");
}

export function runNodeHostScript(spec: HwlabRuntimeLaneSpec, script: string, timeoutSeconds: number, input = ""): CommandResult {
  return runCommand([transPath(), spec.nodeRoute, "sh", "--", script], repoRoot, { input, timeoutMs: timeoutSeconds * 1000 });
}

export function runNodeHostScriptAsync(spec: HwlabRuntimeLaneSpec, script: string, timeoutSeconds: number, label: string): CommandResult {
  const token = nodeRuntimeRenderToken();
  const stateDir = `/tmp/unidesk-hwlab-node-runtime/${label}-${token}`;
  const statusPath = `${stateDir}/status.json`;
  const stdoutPath = `${stateDir}/stdout.log`;
  const stderrPath = `${stateDir}/stderr.log`;
  const runnerPath = `${stateDir}/run.sh`;
  const startScript = [
    "set -eu",
    `state_dir=${shellQuote(stateDir)}`,
    `status_path=${shellQuote(statusPath)}`,
    `stdout_path=${shellQuote(stdoutPath)}`,
    `stderr_path=${shellQuote(stderrPath)}`,
    `runner_path=${shellQuote(runnerPath)}`,
    "rm -rf \"$state_dir\"",
    "mkdir -p \"$state_dir\"",
    "cat > \"$runner_path\" <<'UNIDESK_REMOTE_RUNNER'",
    script,
    "UNIDESK_REMOTE_RUNNER",
    "chmod 700 \"$runner_path\"",
    "python3 - \"$status_path\" <<'PY'",
    "import datetime, json, sys",
    "json.dump({'state':'running','ok':None,'pid':None,'startedAt':datetime.datetime.utcnow().isoformat()+'Z','finishedAt':None,'exitCode':None}, open(sys.argv[1], 'w'), indent=2)",
    "PY",
    "(",
    "  set +e",
    "  sh \"$runner_path\" >\"$stdout_path\" 2>\"$stderr_path\"",
    "  code=$?",
    "  python3 - \"$status_path\" \"$stdout_path\" \"$stderr_path\" \"$code\" <<'PY'",
    "import datetime, json, pathlib, sys",
    "status_path, stdout_path, stderr_path, code = sys.argv[1:5]",
    "def tail(path):",
    "    try:",
    "        text = pathlib.Path(path).read_text(errors='replace')",
    "    except FileNotFoundError:",
    "        return ''",
    "    return text[-12000:]",
    "payload = {'state':'finished','ok':code == '0','pid':None,'startedAt':None,'finishedAt':datetime.datetime.utcnow().isoformat()+'Z','exitCode':int(code),'stdout':tail(stdout_path),'stderr':tail(stderr_path)}",
    "try:",
    "    current = json.load(open(status_path))",
    "    payload['startedAt'] = current.get('startedAt')",
    "except Exception:",
    "    pass",
    "json.dump(payload, open(status_path, 'w'), indent=2)",
    "PY",
    ") >/dev/null 2>&1 &",
    "pid=$!",
    "python3 - \"$status_path\" \"$pid\" <<'PY'",
    "import json, sys",
    "path, pid = sys.argv[1:3]",
    "payload = json.load(open(path))",
    "payload['pid'] = int(pid)",
    "json.dump(payload, open(path, 'w'), indent=2)",
    "PY",
    "printf '{\"ok\":true,\"state\":\"started\",\"pid\":%s,\"statusPath\":\"%s\"}\\n' \"$pid\" \"$status_path\"",
  ].join("\n");
  const start = runNodeHostScript(spec, startScript, 55);
  if (!isCommandSuccess(start)) return start;
  const deadline = Date.now() + Math.max(1, timeoutSeconds) * 1000;
  let lastStatus: CommandResult | null = null;
  let payload: Record<string, unknown> = {};
  while (Date.now() <= deadline) {
    runCommand(["sleep", "2"], repoRoot, { timeoutMs: 3_000 });
    const status = runNodeHostScript(spec, `cat ${shellQuote(statusPath)} 2>/dev/null || printf '{"state":"missing","ok":false,"exitCode":127,"stdout":"","stderr":"missing async status"}\\n'`, 55);
    lastStatus = status;
    payload = parseJsonObject(status.stdout.trim());
    if (payload.state === "finished" || payload.ok === true || payload.ok === false) {
      return commandResultFromAsync(spec, payload, statusPath, false);
    }
  }
  const kill = runNodeHostScript(spec, [
    "set +e",
    `status_path=${shellQuote(statusPath)}`,
    "pid=$(python3 - \"$status_path\" <<'PY' 2>/dev/null",
    "import json, sys",
    "try: print(json.load(open(sys.argv[1])).get('pid') or '')",
    "except Exception: print('')",
    "PY",
    ")",
    "if [ -n \"$pid\" ]; then kill \"$pid\" 2>/dev/null || true; fi",
    "cat \"$status_path\" 2>/dev/null || true",
  ].join("\n"), 55);
  return {
    command: [transPath(), spec.nodeRoute, "sh", "--", "<remote async script>"],
    cwd: repoRoot,
    exitCode: 124,
    stdout: typeof payload.stdout === "string" ? payload.stdout : "",
    stderr: [
      `remote async command timed out after ${timeoutSeconds}s; statusPath=${statusPath}`,
      typeof payload.stderr === "string" ? payload.stderr : "",
      lastStatus?.stderr.trim() ?? "",
      kill.stderr.trim(),
    ].filter(Boolean).join("\n"),
    signal: null,
    timedOut: true,
  };
}