pikasTech-unidesk/scripts/src/platform-infra-wechat-archive.ts

import { randomUUID } from "node:crypto";
import { Buffer } from "node:buffer";
import { existsSync, readFileSync } from "node:fs";
import pathPosix from "node:path/posix";
import type { UniDeskConfig } from "./config";
import { rootPath } from "./config";
import { startJob } from "./jobs";
import {
  asRecord,
  assertProxyOk,
  booleanField,
  compactUnknown,
  containerPathToHostPath,
  fetchJsonWithTimeout,
  findBaiduFileByRemotePath,
  hostRootPath,
  microserviceProxy,
  normalizeRemotePath,
  numberField,
  optionalStringField,
  parseEnvFile,
  parseOpsApplyOptions,
  parseOpsCommonOptions,
  readYamlRecord,
  redactSensitiveUnknown,
  recordField,
  repoRelative,
  resolveRepoPath,
  sanitizePathSegment,
  sha256File,
  stringField,
  syncN8nWorkflow,
  waitForBaiduTransfer,
  type OpsApplyOptions,
  type OpsCommonOptions,
} from "./platform-infra-ops-library";
import { prepareLangBotSecretMaterial, readLangBotPostgresConninfo, readLangBotRuntimeConfig, readLangBotSecretMaterial } from "./platform-infra-langbot";
import { capture, compactCapture, fingerprintValues, parseJsonOutput, shQuote } from "./platform-infra-public-service";

const configFile = rootPath("config", "platform-infra", "wechat-archive.yaml");
const configLabel = "config/platform-infra/wechat-archive.yaml";

interface WechatArchiveConfig {
  version: number;
  kind: "platform-infra-wechat-archive";
  metadata: { id: string; owner: string; relatedIssues: number[] };
  target: { id: string; route: string; namespace: string };
  langbot: {
    configRef: string;
    publicBaseUrl: string;
    expectedAdapter: string;
    callbackPath: string;
    readOnlyRecord: { enabled: boolean; mode: string; webhookName: string; webhookDescription: string; discardPipelineUuid: string };
    pipeline: { name: string; description: string; runner: string; outputKey: string; timeoutSeconds: number };
    notes: string[];
  };
  n8n: {
    configRef: string;
    publicBaseUrl: string;
    workflow: { name: string; id: string; webhookPath: string; active: boolean; timezone: string };
  };
  archiveCallback: {
    publicUrl: string;
    secretRoot: string;
    tokenSourceRef: string;
    tokenKey: string;
    timeoutMs: number;
  };
  personalWechatIngress: {
    enabled: boolean;
    mode: string;
    target: { id: string; windowsRoute: string; hostRoute: string; kubeRoute: string };
    pcWechat: {
      isolation: string;
      requiredVersion: string;
      installerAsset: string;
      installMode: string;
      installRoot: string;
      dataRoot: string;
      autoUpdatePolicy: string;
    };
    wcfHost: {
      releaseTag: string;
      requiredWechatVersion: string;
      releaseUrl: string;
      sdkRoot: string;
      stateRoot: string;
      commandPort: number;
      messagePort: number;
      bindHost: string;
      supervisor: string;
      rdpPolicy: string;
      firewall: { publicExposure: boolean; allowedClientRefs: string[] };
    };
    collector: {
      runtime: string;
      namespace: string;
      createNamespace: boolean;
      workload: { kind: string; name: string; serviceAccountName: string; replicas: number; containerName: string };
      storage: { kind: string; name: string; mountPath: string; size: string; create: boolean };
      image: {
        repository: string;
        tag: string;
        pullPolicy: string;
        context: string;
        dockerfile: string;
        wcferryVersion: string;
        pip: { indexUrl: string; trustedHost: string; timeoutSeconds: number; retries: number };
      };
      wcfHost: string;
      commandPort: number;
      messagePort: number;
      stateRoot: string;
      queueMode: string;
      outboxMode: string;
      archiveCallbackRef: string;
      secretName: string;
      readOnly: { enabled: boolean; sendCapability: boolean; allowedMethods: string[]; forbiddenMethods: string[] };
    };
    poc: { accountPolicy: string; observationWindowHours: number; requiredMessageTypes: string[] };
  };
  baiduNetdisk: {
    serviceId: string;
    proxyMode: string;
    configRef: string;
    staging: { hostRoot: string; containerRoot: string; outboxDir: string; pullDir: string };
    archive: {
      remoteRoot: string;
      timezone: string;
      textExtension: string;
      imageExtension: string;
      pathTemplate: string;
      maxInlineBytes: number;
    };
  };
  messageTypes: Record<string, { enabled: boolean; contentField: string; mediaMode: string; filenameField?: string; mimeTypeField?: string }>;
  validate: {
    timeoutMs: number;
    pollIntervalMs: number;
    fixtures: {
      text: { conversationId: string; senderId: string; text: string };
      image: { conversationId: string; senderId: string; filename: string; mimeType: string; dataBase64: string };
    };
  };
}

interface PullOptions extends OpsCommonOptions {
  remotePath?: string;
  fsId?: string;
  localPath?: string;
}

export function wechatArchiveHelp(): Record<string, unknown> {
  return {
    command: "platform-infra wechat-archive plan|apply|status|validate|pull|wcf-host-*|collector-*",
    output: "json",
    usage: [
      "bun scripts/cli.ts platform-infra wechat-archive plan [--target G14]",
      "bun scripts/cli.ts platform-infra wechat-archive apply [--target G14] --dry-run",
      "bun scripts/cli.ts platform-infra wechat-archive apply [--target G14] --confirm",
      "bun scripts/cli.ts platform-infra wechat-archive status [--target G14] [--full|--raw]",
      "bun scripts/cli.ts platform-infra wechat-archive validate [--target G14] [--full|--raw]",
      "bun scripts/cli.ts platform-infra wechat-archive pull --remote-path /UniDesk/WeChatArchive/... [--target G14] [--full|--raw]",
      "bun scripts/cli.ts platform-infra wechat-archive pull --fs-id <baiduFsId> [--local-path wechat-archive/pulls/file] [--target G14]",
      "bun scripts/cli.ts platform-infra wechat-archive wcf-host-prepare --confirm",
      "bun scripts/cli.ts platform-infra wechat-archive wcf-host-status [--full|--raw]",
      "bun scripts/cli.ts platform-infra wechat-archive wcf-host-start --confirm",
      "bun scripts/cli.ts platform-infra wechat-archive collector-image-build --confirm",
      "bun scripts/cli.ts platform-infra wechat-archive collector-image-status [--full|--raw]",
      "bun scripts/cli.ts platform-infra wechat-archive collector-apply --dry-run",
      "bun scripts/cli.ts platform-infra wechat-archive collector-apply --confirm",
      "bun scripts/cli.ts platform-infra wechat-archive collector-status [--full|--raw]",
    ],
    configTruth: configLabel,
    boundary: {
      langbot: "Receives WeChat/OpenClaw events and forwards archive events to the YAML-declared n8n webhook.",
      n8n: "Stores a YAML-rendered workflow that normalizes WeChat payloads.",
      personalWechatIngress: "Optionally declares the D601 Windows PC WeChat + WeChatFerry host and D601 k3s read-only collector boundary.",
      baiduNetdisk: "Uploads and downloads through the private backend-core microservice proxy; Baidu credentials remain in the baidu-netdisk service.",
      database: "LangBot and n8n continue to use the single PK01/Pika01 external PostgreSQL instance with separate databases/roles.",
    },
  };
}

export async function runWechatArchiveCommand(config: UniDeskConfig, args: string[]): Promise<Record<string, unknown>> {
  const [action = "plan"] = args;
  if (action === "plan") return plan(parseOpsCommonOptions(args.slice(1)));
  if (action === "apply") return await apply(config, parseOpsApplyOptions(args.slice(1)));
  if (action === "status") return await status(parseOpsCommonOptions(args.slice(1)));
  if (action === "validate") return await validate(parseOpsCommonOptions(args.slice(1)));
  if (action === "pull") return await pull(parsePullOptions(args.slice(1)));
  if (action === "wcf-host-prepare") return await wcfHostPrepare(config, parseOpsApplyOptions(args.slice(1)));
  if (action === "wcf-host-start") return await wcfHostStart(config, parseOpsApplyOptions(args.slice(1)));
  if (action === "wcf-host-status") return await wcfHostStatus(config, parseOpsCommonOptions(args.slice(1)));
  if (action === "collector-plan") return collectorPlan(parseOpsCommonOptions(args.slice(1)));
  if (action === "collector-image-build") return await collectorImageBuild(config, parseOpsApplyOptions(args.slice(1)));
  if (action === "collector-image-status") return await collectorImageStatus(config, parseOpsCommonOptions(args.slice(1)));
  if (action === "collector-apply") return await collectorApply(config, parseOpsApplyOptions(args.slice(1)));
  if (action === "collector-status") return await collectorStatus(config, parseOpsCommonOptions(args.slice(1)));
  return { ok: false, error: "unsupported-platform-infra-wechat-archive-command", args, help: wechatArchiveHelp() };
}

function parsePullOptions(args: string[]): PullOptions {
  const parsed = parseOpsCommonOptions(args, { stringOptions: ["--remote-path", "--fs-id", "--local-path"] });
  return {
    targetId: parsed.targetId,
    full: parsed.full,
    raw: parsed.raw,
    remotePath: typeof parsed.remotePath === "string" ? parsed.remotePath : undefined,
    fsId: typeof parsed.fsId === "string" ? parsed.fsId : undefined,
    localPath: typeof parsed.localPath === "string" ? parsed.localPath : undefined,
  };
}

function plan(options: OpsCommonOptions): Record<string, unknown> {
  const archive = readConfig();
  assertTarget(archive, options.targetId);
  const workflow = renderN8nWorkflow(archive, null);
  const policy = policyChecks(archive);
  return {
    ok: policy.every((check) => check.ok),
    action: "platform-infra-wechat-archive-plan",
    mutation: false,
    config: configSummary(archive),
    workflow: {
      id: archive.n8n.workflow.id,
      name: archive.n8n.workflow.name,
      webhookUrl: webhookUrl(archive),
      nodes: Array.isArray(workflow.nodes) ? workflow.nodes.length : 0,
      active: archive.n8n.workflow.active,
    },
    policy,
    next: {
      dryRun: `bun scripts/cli.ts platform-infra wechat-archive apply --target ${archive.target.id} --dry-run`,
      apply: `bun scripts/cli.ts platform-infra wechat-archive apply --target ${archive.target.id} --confirm`,
      validate: `bun scripts/cli.ts platform-infra wechat-archive validate --target ${archive.target.id} --full`,
      pull: `bun scripts/cli.ts platform-infra wechat-archive pull --remote-path ${archive.baiduNetdisk.archive.remoteRoot}/... --target ${archive.target.id}`,
      personalWechatIngress: {
        prepareWcfHost: "bun scripts/cli.ts platform-infra wechat-archive wcf-host-prepare --confirm",
        startWcfHost: "bun scripts/cli.ts platform-infra wechat-archive wcf-host-start --confirm",
        buildCollectorImage: "bun scripts/cli.ts platform-infra wechat-archive collector-image-build --confirm",
        applyCollector: "bun scripts/cli.ts platform-infra wechat-archive collector-apply --confirm",
        status: "bun scripts/cli.ts platform-infra wechat-archive collector-status --full",
      },
    },
  };
}

async function apply(config: UniDeskConfig, options: OpsApplyOptions): Promise<Record<string, unknown>> {
  const archive = readConfig();
  assertTarget(archive, options.targetId);
  const policy = policyChecks(archive);
  if (!policy.every((check) => check.ok)) return { ok: false, action: "platform-infra-wechat-archive-apply", mode: "policy-blocked", policy };
  if (options.confirm && !options.wait) {
    const job = startJob(
      "platform_infra_wechat_archive_apply_g14",
      ["bun", "scripts/cli.ts", "platform-infra", "wechat-archive", "apply", "--target", archive.target.id, "--confirm", "--wait"],
      "Sync the YAML-rendered n8n workflow for WeChat -> Baidu Netdisk archive",
    );
    return {
      ok: true,
      action: "platform-infra-wechat-archive-apply",
      mode: "async-job",
      mutation: true,
      job,
      statusCommand: `bun scripts/cli.ts job status ${job.id} --tail-bytes 12000`,
    };
  }
  const callbackToken = options.dryRun ? null : readArchiveCallbackToken(archive);
  const workflowJson = renderN8nWorkflow(archive, callbackToken?.value ?? null);
  const sync = await syncN8nWorkflow(config, {
    targetRoute: archive.target.route,
    namespace: archive.target.namespace,
    deploymentName: "n8n",
    workflowJson,
    workflowId: archive.n8n.workflow.id,
    workflowName: archive.n8n.workflow.name,
    webhookPath: archive.n8n.workflow.webhookPath,
    active: archive.n8n.workflow.active,
    dryRun: options.dryRun,
  });
  const langbotBinding = options.dryRun ? langBotDryRun(archive) : await syncLangBotPipeline(archive);
  return {
    ok: sync.ok && langbotBinding.ok === true,
    action: "platform-infra-wechat-archive-apply",
    mode: options.dryRun ? "dry-run" : "confirmed",
    mutation: !options.dryRun,
    config: configSummary(archive),
    policy,
    n8nWorkflow: sync,
    archiveCallbackAuth: callbackToken === null ? { mode: "dry-run", valuesPrinted: false } : {
      sourceRef: callbackToken.sourceRef,
      keyName: callbackToken.keyName,
      fingerprint: callbackToken.fingerprint,
      valuesPrinted: false,
    },
    langbotBinding,
    next: {
      status: `bun scripts/cli.ts platform-infra wechat-archive status --target ${archive.target.id}`,
      validate: `bun scripts/cli.ts platform-infra wechat-archive validate --target ${archive.target.id} --full`,
    },
  };
}

async function status(options: OpsCommonOptions): Promise<Record<string, unknown>> {
  const archive = readConfig();
  assertTarget(archive, options.targetId);
  const health = microserviceProxy(archive.baiduNetdisk.serviceId, "/health", { timeoutMs: 30_000 });
  const auth = microserviceProxy(archive.baiduNetdisk.serviceId, "/api/auth/status", { timeoutMs: 30_000 });
  const transfers = microserviceProxy(archive.baiduNetdisk.serviceId, "/api/transfers?limit=10", { timeoutMs: 30_000 });
  const langbot = await inspectLangBotBinding(archive);
  return {
    ok: health.ok && auth.ok && langbot.ok === true,
    action: "platform-infra-wechat-archive-status",
    config: configSummary(archive),
    langbot,
    n8n: {
      webhookUrl: webhookUrl(archive),
      workflowId: archive.n8n.workflow.id,
      activeDesired: archive.n8n.workflow.active,
    },
    baiduNetdisk: {
      health: compactBaiduHealth(health),
      auth: compactBaiduAuth(auth),
      recentTransfers: compactBaiduTransfers(transfers),
    },
    valuesPrinted: false,
    ...(options.raw ? { raw: redactSensitiveUnknown({ health, auth, transfers }) } : {}),
  };
}

async function validate(options: OpsCommonOptions): Promise<Record<string, unknown>> {
  const archive = readConfig();
  assertTarget(archive, options.targetId);
  const observedAt = new Date();
  const textPayload = fixturePayload(archive, "text", observedAt);
  const imagePayload = fixturePayload(archive, "image", observedAt);
  const n8nText = await callWorkflow(archive, textPayload);
  const n8nImage = await callWorkflow(archive, imagePayload);
  const langbot = await inspectLangBotBinding(archive);
  const textArchive = archiveFromWorkflowResponse(textPayload, n8nText);
  const imageArchive = archiveFromWorkflowResponse(imagePayload, n8nImage);
  const textPull = await pullArchiveIfReady(archive, textArchive);
  const imagePull = await pullArchiveIfReady(archive, imageArchive);
  const ok = langbot.ok === true
    && n8nText.ok === true
    && n8nImage.ok === true
    && textArchive.ok === true
    && textArchive.skipPipeline === true
    && imageArchive.ok === true
    && imageArchive.skipPipeline === true
    && textPull.ok === true
    && imagePull.ok === true;
  return {
    ok,
    action: "platform-infra-wechat-archive-validate",
    config: configSummary(archive),
    langbot,
    fixtures: {
      text: validationSummary(textPayload, n8nText, textArchive, textPull, options.full),
      image: validationSummary(imagePayload, n8nImage, imageArchive, imagePull, options.full),
    },
    valuesPrinted: false,
    next: {
      pullText: `bun scripts/cli.ts platform-infra wechat-archive pull --remote-path ${textArchive.remotePath} --target ${archive.target.id}`,
      pullImage: `bun scripts/cli.ts platform-infra wechat-archive pull --remote-path ${imageArchive.remotePath} --target ${archive.target.id}`,
    },
  };
}

async function pull(options: PullOptions): Promise<Record<string, unknown>> {
  const archive = readConfig();
  assertTarget(archive, options.targetId);
  if (!options.remotePath && !options.fsId) throw new Error("pull requires --remote-path or --fs-id");
  let fsId = options.fsId;
  let remotePath = options.remotePath;
  let listed: Record<string, unknown> | null = null;
  if (!fsId) {
    remotePath = normalizeRemotePath(String(remotePath));
    listed = findBaiduFileByRemotePath(archive.baiduNetdisk.serviceId, remotePath);
    if (listed === null) throw new Error(`Baidu file not found: ${remotePath}`);
    fsId = String(listed.fsId || listed.fs_id || "");
    if (!fsId) throw new Error(`Baidu file has no fsId in listing: ${remotePath}`);
  }
  const localRel = options.localPath ?? pullLocalRel(archive, remotePath ?? `fsid-${fsId}`);
  const downloaded = await pullByFsId(archive, fsId, localRel);
  return {
    ok: downloaded.ok,
    action: "platform-infra-wechat-archive-pull",
    remotePath: remotePath ?? downloaded.remotePath,
    fsId,
    listed: redactBaiduFileEntry(listed),
    download: options.full ? downloaded : compactDownload(downloaded),
    valuesPrinted: false,
    ...(options.raw ? { raw: downloaded } : {}),
  };
}

function collectorPlan(options: OpsCommonOptions): Record<string, unknown> {
  const archive = readConfig();
  assertTarget(archive, options.targetId);
  const manifest = renderCollectorManifest(archive);
  const policy = collectorPolicyChecks(archive, manifest);
  return {
    ok: policy.every((check) => check.ok),
    action: "platform-infra-wechat-archive-collector-plan",
    mutation: false,
    config: personalWechatIngressSummary(archive.personalWechatIngress),
    manifest: {
      namespace: archive.personalWechatIngress.collector.namespace,
      image: collectorImageRef(archive.personalWechatIngress),
      bytes: Buffer.byteLength(manifest, "utf8"),
      sha256: fingerprintValues({ manifest }, ["manifest"]),
    },
    policy,
    next: {
      prepareWcfHost: "bun scripts/cli.ts platform-infra wechat-archive wcf-host-prepare --confirm",
      startWcfHost: "bun scripts/cli.ts platform-infra wechat-archive wcf-host-start --confirm",
      buildImage: "bun scripts/cli.ts platform-infra wechat-archive collector-image-build --confirm",
      dryRun: "bun scripts/cli.ts platform-infra wechat-archive collector-apply --dry-run",
      apply: "bun scripts/cli.ts platform-infra wechat-archive collector-apply --confirm",
      status: "bun scripts/cli.ts platform-infra wechat-archive collector-status --full",
    },
  };
}

async function wcfHostPrepare(config: UniDeskConfig, options: OpsApplyOptions): Promise<Record<string, unknown>> {
  const archive = readConfig();
  assertTarget(archive, options.targetId);
  if (options.dryRun) {
    return {
      ok: true,
      action: "platform-infra-wechat-archive-wcf-host-prepare",
      mode: "dry-run",
      mutation: false,
      target: archive.personalWechatIngress.target,
      scripts: windowsScriptSummary(),
    };
  }
  const result = await capture(config, archive.personalWechatIngress.target.windowsRoute, ["ps"], windowsScriptSyncAndStartPrepareScript(archive));
  const parsed = parseJsonOutput(result.stdout);
  return {
    ok: result.exitCode === 0 && boolField(parsed, "ok", false),
    action: "platform-infra-wechat-archive-wcf-host-prepare",
    mode: "confirmed",
    mutation: true,
    target: archive.personalWechatIngress.target,
    summary: parsed,
    remote: compactCapture(result, { full: options.full || result.exitCode !== 0 }),
    ...(options.raw ? { raw: result } : {}),
    next: {
      status: "bun scripts/cli.ts platform-infra wechat-archive wcf-host-status --full",
      start: "bun scripts/cli.ts platform-infra wechat-archive wcf-host-start --confirm",
    },
  };
}

async function wcfHostStart(config: UniDeskConfig, options: OpsApplyOptions): Promise<Record<string, unknown>> {
  const archive = readConfig();
  assertTarget(archive, options.targetId);
  if (options.dryRun) {
    return {
      ok: true,
      action: "platform-infra-wechat-archive-wcf-host-start",
      mode: "dry-run",
      mutation: false,
      target: archive.personalWechatIngress.target,
    };
  }
  const result = await capture(config, archive.personalWechatIngress.target.windowsRoute, ["ps"], windowsRunManagedScript(archive, "start.ps1"));
  const parsed = parseJsonOutput(result.stdout);
  return {
    ok: result.exitCode === 0 && boolField(parsed, "ok", false),
    action: "platform-infra-wechat-archive-wcf-host-start",
    mode: "confirmed",
    mutation: true,
    target: archive.personalWechatIngress.target,
    summary: parsed,
    remote: compactCapture(result, { full: options.full || result.exitCode !== 0 }),
    ...(options.raw ? { raw: result } : {}),
    next: {
      status: "bun scripts/cli.ts platform-infra wechat-archive wcf-host-status --full",
      collectorApply: "bun scripts/cli.ts platform-infra wechat-archive collector-apply --confirm",
    },
  };
}

async function wcfHostStatus(config: UniDeskConfig, options: OpsCommonOptions): Promise<Record<string, unknown>> {
  const archive = readConfig();
  assertTarget(archive, options.targetId);
  const result = await capture(config, archive.personalWechatIngress.target.windowsRoute, ["ps"], windowsRunManagedScript(archive, "status.ps1"));
  const parsed = parseJsonOutput(result.stdout);
  const status = parsed === null ? null : compactWcfHostStatus(parsed, options.full);
  return {
    ok: result.exitCode === 0 && parsed !== null && parsed.ok === true,
    action: "platform-infra-wechat-archive-wcf-host-status",
    target: archive.personalWechatIngress.target,
    summary: status,
    remote: compactCapture(result, { full: options.full || result.exitCode !== 0 }),
    ...(options.raw ? { raw: result } : {}),
  };
}

async function collectorImageBuild(config: UniDeskConfig, options: OpsApplyOptions): Promise<Record<string, unknown>> {
  const archive = readConfig();
  assertTarget(archive, options.targetId);
  if (options.confirm && !options.wait) {
    const job = startJob(
      "platform_infra_wechat_archive_collector_image_build_d601",
      ["bun", "scripts/cli.ts", "platform-infra", "wechat-archive", "collector-image-build", "--confirm", "--wait"],
      "Build and push the D601 k3s personal WeChat collector image through the controlled UniDesk CLI",
    );
    return {
      ok: true,
      action: "platform-infra-wechat-archive-collector-image-build",
      mode: "async-job",
      mutation: true,
      job,
      statusCommand: `bun scripts/cli.ts job status ${job.id} --tail-bytes 12000`,
    };
  }
  if (options.dryRun) {
    return {
      ok: true,
      action: "platform-infra-wechat-archive-collector-image-build",
      mode: "dry-run",
      mutation: false,
      image: collectorImageRef(archive.personalWechatIngress),
      source: collectorSourceSummary(archive),
    };
  }
  const result = await capture(config, archive.personalWechatIngress.target.hostRoute, ["sh"], collectorImageBuildScript(archive));
  const parsed = parseJsonOutput(result.stdout);
  return {
    ok: result.exitCode === 0 && boolField(parsed, "ok", false),
    action: "platform-infra-wechat-archive-collector-image-build",
    mode: "confirmed",
    mutation: true,
    image: collectorImageRef(archive.personalWechatIngress),
    summary: parsed,
    remote: compactCapture(result, { full: options.full || result.exitCode !== 0 }),
    ...(options.raw ? { raw: result } : {}),
  };
}

async function collectorImageStatus(config: UniDeskConfig, options: OpsCommonOptions): Promise<Record<string, unknown>> {
  const archive = readConfig();
  assertTarget(archive, options.targetId);
  const result = await capture(config, archive.personalWechatIngress.target.hostRoute, ["sh"], collectorImageStatusScript(archive));
  const parsed = parseJsonOutput(result.stdout);
  return {
    ok: result.exitCode === 0 && parsed !== null && parsed.ok === true,
    action: "platform-infra-wechat-archive-collector-image-status",
    image: collectorImageRef(archive.personalWechatIngress),
    summary: parsed,
    remote: compactCapture(result, { full: options.full || result.exitCode !== 0 }),
    ...(options.raw ? { raw: result } : {}),
  };
}

async function collectorApply(config: UniDeskConfig, options: OpsApplyOptions): Promise<Record<string, unknown>> {
  const archive = readConfig();
  assertTarget(archive, options.targetId);
  const manifest = renderCollectorManifest(archive);
  const policy = collectorPolicyChecks(archive, manifest);
  if (!policy.every((check) => check.ok)) return { ok: false, action: "platform-infra-wechat-archive-collector-apply", mode: "policy-blocked", policy };
  if (options.confirm && !options.wait) {
    const job = startJob(
      "platform_infra_wechat_archive_collector_apply_d601",
      ["bun", "scripts/cli.ts", "platform-infra", "wechat-archive", "collector-apply", "--confirm", "--wait"],
      "Apply D601 k3s personal WeChat collector objects through the controlled UniDesk CLI",
    );
    return {
      ok: true,
      action: "platform-infra-wechat-archive-collector-apply",
      mode: "async-job",
      mutation: true,
      job,
      statusCommand: `bun scripts/cli.ts job status ${job.id} --tail-bytes 12000`,
    };
  }
  if (options.dryRun) {
    const result = await capture(config, archive.personalWechatIngress.target.kubeRoute, ["sh"], collectorApplyScript(archive, manifest, null, true));
    const parsed = parseJsonOutput(result.stdout);
    return {
      ok: result.exitCode === 0 && boolField(parsed, "ok", false),
      action: "platform-infra-wechat-archive-collector-apply",
      mode: "dry-run",
      mutation: false,
      policy,
      target: archive.personalWechatIngress.target,
      summary: parsed,
      remote: compactCapture(result, { full: options.full || result.exitCode !== 0 }),
      ...(options.raw ? { raw: result } : {}),
    };
  }
  const token = readArchiveCallbackToken(archive);
  const result = await capture(config, archive.personalWechatIngress.target.kubeRoute, ["sh"], collectorApplyScript(archive, manifest, token.value, false));
  const parsed = parseJsonOutput(result.stdout);
  return {
    ok: result.exitCode === 0 && boolField(parsed, "ok", false),
    action: "platform-infra-wechat-archive-collector-apply",
    mode: "confirmed",
    mutation: true,
    policy,
    target: archive.personalWechatIngress.target,
    secret: {
      sourceRef: token.sourceRef,
      keyName: token.keyName,
      targetSecret: archive.personalWechatIngress.collector.secretName,
      fingerprint: token.fingerprint,
      valuesPrinted: false,
    },
    summary: parsed,
    remote: compactCapture(result, { full: options.full || result.exitCode !== 0 }),
    ...(options.raw ? { raw: result } : {}),
    next: {
      status: "bun scripts/cli.ts platform-infra wechat-archive collector-status --full",
      wcfHostStatus: "bun scripts/cli.ts platform-infra wechat-archive wcf-host-status --full",
    },
  };
}

async function collectorStatus(config: UniDeskConfig, options: OpsCommonOptions): Promise<Record<string, unknown>> {
  const archive = readConfig();
  assertTarget(archive, options.targetId);
  const result = await capture(config, archive.personalWechatIngress.target.kubeRoute, ["sh"], collectorStatusScript(archive, options.full));
  const parsed = parseJsonOutput(result.stdout);
  return {
    ok: result.exitCode === 0 && parsed !== null && parsed.ok === true,
    action: "platform-infra-wechat-archive-collector-status",
    target: archive.personalWechatIngress.target,
    summary: parsed,
    remote: compactCapture(result, { full: options.full || result.exitCode !== 0 }),
    ...(options.raw ? { raw: result } : {}),
  };
}

function readConfig(): WechatArchiveConfig {
  const root = readYamlRecord(configFile, "platform-infra-wechat-archive");
  const metadata = recordField(root, "metadata", configLabel);
  const target = recordField(root, "target", configLabel);
  const langbot = recordField(root, "langbot", configLabel);
  const n8n = recordField(root, "n8n", configLabel);
  const n8nWorkflow = recordField(n8n, "workflow", `${configLabel}.n8n`);
  const archiveCallback = recordField(root, "archiveCallback", configLabel);
  const personalWechatIngress = parsePersonalWechatIngress(recordField(root, "personalWechatIngress", configLabel));
  const baiduNetdisk = recordField(root, "baiduNetdisk", configLabel);
  const staging = recordField(baiduNetdisk, "staging", `${configLabel}.baiduNetdisk`);
  const archive = recordField(baiduNetdisk, "archive", `${configLabel}.baiduNetdisk`);
  const messageTypesRaw = recordField(root, "messageTypes", configLabel);
  const validateRaw = recordField(root, "validate", configLabel);
  const fixtures = recordField(validateRaw, "fixtures", `${configLabel}.validate`);
  const textFixture = recordField(fixtures, "text", `${configLabel}.validate.fixtures`);
  const imageFixture = recordField(fixtures, "image", `${configLabel}.validate.fixtures`);
  const relatedIssuesRaw = metadata.relatedIssues;
  if (!Array.isArray(relatedIssuesRaw)) throw new Error(`${configLabel}.metadata.relatedIssues must be an array`);
  const relatedIssues = relatedIssuesRaw.map((item) => Number(item));
  const messageTypes: WechatArchiveConfig["messageTypes"] = {};
  for (const [key, value] of Object.entries(messageTypesRaw)) {
    const record = asRecord(value, `${configLabel}.messageTypes.${key}`);
    messageTypes[key] = {
      enabled: booleanField(record, "enabled", `${configLabel}.messageTypes.${key}`),
      contentField: stringField(record, "contentField", `${configLabel}.messageTypes.${key}`),
      mediaMode: stringField(record, "mediaMode", `${configLabel}.messageTypes.${key}`),
      filenameField: optionalStringField(record, "filenameField", `${configLabel}.messageTypes.${key}`),
      mimeTypeField: optionalStringField(record, "mimeTypeField", `${configLabel}.messageTypes.${key}`),
    };
  }
  const config: WechatArchiveConfig = {
    version: Number(root.version),
    kind: "platform-infra-wechat-archive",
    metadata: {
      id: stringField(metadata, "id", configLabel),
      owner: stringField(metadata, "owner", configLabel),
      relatedIssues,
    },
    target: {
      id: stringField(target, "id", configLabel),
      route: stringField(target, "route", configLabel),
      namespace: stringField(target, "namespace", configLabel),
    },
    langbot: {
      configRef: stringField(langbot, "configRef", configLabel),
      publicBaseUrl: stringField(langbot, "publicBaseUrl", configLabel),
      expectedAdapter: stringField(langbot, "expectedAdapter", configLabel),
      callbackPath: stringField(langbot, "callbackPath", configLabel),
      readOnlyRecord: parseReadOnlyRecord(recordField(langbot, "readOnlyRecord", `${configLabel}.langbot`)),
      pipeline: parseLangBotPipeline(recordField(langbot, "pipeline", `${configLabel}.langbot`)),
      notes: Array.isArray(langbot.notes) ? langbot.notes.map(String) : [],
    },
    n8n: {
      configRef: stringField(n8n, "configRef", configLabel),
      publicBaseUrl: stringField(n8n, "publicBaseUrl", configLabel),
      workflow: {
        name: stringField(n8nWorkflow, "name", `${configLabel}.n8n.workflow`),
        id: stringField(n8nWorkflow, "id", `${configLabel}.n8n.workflow`),
        webhookPath: stringField(n8nWorkflow, "webhookPath", `${configLabel}.n8n.workflow`),
        active: booleanField(n8nWorkflow, "active", `${configLabel}.n8n.workflow`),
        timezone: stringField(n8nWorkflow, "timezone", `${configLabel}.n8n.workflow`),
      },
    },
    archiveCallback: {
      publicUrl: stringField(archiveCallback, "publicUrl", `${configLabel}.archiveCallback`),
      secretRoot: stringField(archiveCallback, "secretRoot", `${configLabel}.archiveCallback`),
      tokenSourceRef: stringField(archiveCallback, "tokenSourceRef", `${configLabel}.archiveCallback`),
      tokenKey: stringField(archiveCallback, "tokenKey", `${configLabel}.archiveCallback`),
      timeoutMs: numberField(archiveCallback, "timeoutMs", `${configLabel}.archiveCallback`),
    },
    personalWechatIngress,
    baiduNetdisk: {
      serviceId: stringField(baiduNetdisk, "serviceId", `${configLabel}.baiduNetdisk`),
      proxyMode: stringField(baiduNetdisk, "proxyMode", `${configLabel}.baiduNetdisk`),
      configRef: stringField(baiduNetdisk, "configRef", `${configLabel}.baiduNetdisk`),
      staging: {
        hostRoot: stringField(staging, "hostRoot", `${configLabel}.baiduNetdisk.staging`),
        containerRoot: stringField(staging, "containerRoot", `${configLabel}.baiduNetdisk.staging`),
        outboxDir: stringField(staging, "outboxDir", `${configLabel}.baiduNetdisk.staging`),
        pullDir: stringField(staging, "pullDir", `${configLabel}.baiduNetdisk.staging`),
      },
      archive: {
        remoteRoot: normalizeRemotePath(stringField(archive, "remoteRoot", `${configLabel}.baiduNetdisk.archive`)),
        timezone: stringField(archive, "timezone", `${configLabel}.baiduNetdisk.archive`),
        textExtension: stringField(archive, "textExtension", `${configLabel}.baiduNetdisk.archive`),
        imageExtension: stringField(archive, "imageExtension", `${configLabel}.baiduNetdisk.archive`),
        pathTemplate: stringField(archive, "pathTemplate", `${configLabel}.baiduNetdisk.archive`),
        maxInlineBytes: numberField(archive, "maxInlineBytes", `${configLabel}.baiduNetdisk.archive`),
      },
    },
    messageTypes,
    validate: {
      timeoutMs: numberField(validateRaw, "timeoutMs", `${configLabel}.validate`),
      pollIntervalMs: numberField(validateRaw, "pollIntervalMs", `${configLabel}.validate`),
      fixtures: {
        text: {
          conversationId: stringField(textFixture, "conversationId", `${configLabel}.validate.fixtures.text`),
          senderId: stringField(textFixture, "senderId", `${configLabel}.validate.fixtures.text`),
          text: stringField(textFixture, "text", `${configLabel}.validate.fixtures.text`),
        },
        image: {
          conversationId: stringField(imageFixture, "conversationId", `${configLabel}.validate.fixtures.image`),
          senderId: stringField(imageFixture, "senderId", `${configLabel}.validate.fixtures.image`),
          filename: stringField(imageFixture, "filename", `${configLabel}.validate.fixtures.image`),
          mimeType: stringField(imageFixture, "mimeType", `${configLabel}.validate.fixtures.image`),
          dataBase64: stringField(imageFixture, "dataBase64", `${configLabel}.validate.fixtures.image`),
        },
      },
    },
  };
  validateConfig(config);
  return config;
}

const collectorFieldManager = "unidesk-platform-infra-wechat-archive";
const windowsOpsRoot = "C:\\UniDesk\\personal-wechat\\ops";

function collectorImageRef(config: WechatArchiveConfig["personalWechatIngress"]): string {
  return `${config.collector.image.repository}:${config.collector.image.tag}`;
}

function boolField(value: Record<string, unknown> | null, key: string, fallback: boolean): boolean {
  if (value === null) return fallback;
  const raw = value[key];
  return typeof raw === "boolean" ? raw : fallback;
}

function collectorSourceSummary(archive: WechatArchiveConfig): Record<string, unknown> {
  const ingress = archive.personalWechatIngress;
  const dockerfile = resolveRepoPath(ingress.collector.image.dockerfile);
  const collectorPy = resolveRepoPath(pathPosix.join(ingress.collector.image.context, "collector.py"));
  return {
    dockerfile: ingress.collector.image.dockerfile,
    collectorPy: repoRelative(collectorPy),
    dockerfileSha256: sha256File(dockerfile),
    collectorPySha256: sha256File(collectorPy),
    wcferryVersion: ingress.collector.image.wcferryVersion,
  };
}

function windowsScriptSummary(): Record<string, unknown> {
  const files = ["prepare.ps1", "start.ps1", "status.ps1", "wcf_host.py"].map((name) => {
    const filePath = rootPath("scripts", "windows", "personal-wechat", name);
    return { name, path: repoRelative(filePath), sha256: sha256File(filePath), bytes: readFileSync(filePath).byteLength };
  });
  return { targetRoot: windowsOpsRoot, files };
}

function b64(value: string): string {
  return Buffer.from(value, "utf8").toString("base64");
}

function fileB64(repoPath: string): string {
  return readFileSync(resolveRepoPath(repoPath)).toString("base64");
}

function managedWindowsFiles(): Array<{ name: string; contentB64: string }> {
  const base = rootPath("scripts", "windows", "personal-wechat");
  return ["prepare.ps1", "start.ps1", "status.ps1", "wcf_host.py"].map((name) => ({
    name,
    contentB64: readFileSync(pathPosix.join(base, name)).toString("base64"),
  }));
}

function windowsScriptSyncBlock(): string {
  const writes = managedWindowsFiles().map((file) => {
    const target = `${windowsOpsRoot}\\${file.name}`;
    return `
$target = ${psSingleQuote(target)}
$content = [Text.Encoding]::UTF8.GetString([Convert]::FromBase64String(${psSingleQuote(file.contentB64)}))
Set-Content -LiteralPath $target -Encoding UTF8 -Value $content
`;
  }).join("\n");
  return `
$ErrorActionPreference = "Stop"
$OpsRoot = ${psSingleQuote(windowsOpsRoot)}
New-Item -ItemType Directory -Force $OpsRoot | Out-Null
${writes}
`;
}

function windowsScriptSyncAndStartPrepareScript(_archive: WechatArchiveConfig): string {
  const prepare = `${windowsOpsRoot}\\prepare.ps1`;
  const stateRoot = "C:\\UniDesk\\personal-wechat\\wcf-state";
  const stdout = `${stateRoot}\\prepare.stdout.log`;
  const stderr = `${stateRoot}\\prepare.stderr.log`;
  const resultFile = `${stateRoot}\\prepare-result.json`;
  return `${windowsScriptSyncBlock()}
$StateRoot = ${psSingleQuote(stateRoot)}
New-Item -ItemType Directory -Force $StateRoot | Out-Null
if (Test-Path ${psSingleQuote(resultFile)}) { Remove-Item -Force ${psSingleQuote(resultFile)} }
$Wrapper = Join-Path $StateRoot "prepare-runner.cmd"
@"
@echo off
powershell.exe -NoProfile -ExecutionPolicy Bypass -File "${prepare}" > "${stdout}" 2> "${stderr}"
"@ | Set-Content -Encoding ASCII $Wrapper
$cmd = '/c start "" /min "' + $Wrapper + '"'
$proc = Start-Process -FilePath "cmd.exe" -ArgumentList $cmd -WindowStyle Hidden -PassThru
[pscustomobject]@{
  ok = $true
  started = $true
  launcherPid = $proc.Id
  opsRoot = $OpsRoot
  stdout = ${psSingleQuote(stdout)}
  stderr = ${psSingleQuote(stderr)}
  result = ${psSingleQuote(resultFile)}
  statusCommand = "bun scripts/cli.ts platform-infra wechat-archive wcf-host-status --full"
} | ConvertTo-Json -Depth 6
`;
}

function windowsRunManagedScript(_archive: WechatArchiveConfig, scriptName: "start.ps1" | "status.ps1"): string {
  const script = `${windowsOpsRoot}\\${scriptName}`;
  return `${windowsScriptSyncBlock()}
& ${psSingleQuote(script)}
if ($LASTEXITCODE -ne $null -and $LASTEXITCODE -ne 0) { exit $LASTEXITCODE }
`;
}

function psSingleQuote(value: string): string {
  return `'${value.replaceAll("'", "''")}'`;
}

function compactWcfHostStatus(parsed: Record<string, unknown>, full: boolean): Record<string, unknown> {
  const status = typeof parsed.status === "object" && parsed.status !== null && !Array.isArray(parsed.status)
    ? parsed.status as Record<string, unknown>
    : null;
  const prepare = typeof parsed.prepare === "object" && parsed.prepare !== null && !Array.isArray(parsed.prepare)
    ? parsed.prepare as Record<string, unknown>
    : null;
  return {
    ok: parsed.ok,
    prepared: parsed.prepared,
    running: parsed.running,
    pid: parsed.pid,
    ports: parsed.ports,
    login: status === null ? null : {
      isLogin: status.isLogin,
      qrcodePresent: status.qrcodePresent,
      qrcode: full ? status.qrcode : undefined,
      ts: status.ts,
    },
    prepare: prepare === null ? null : {
      ok: prepare.ok,
      wechatExe: prepare.wechatExe,
      installAttempted: prepare.installAttempted,
      installExitCode: prepare.installExitCode,
      wcferryVersion: prepare.wcferryVersion,
      next: prepare.next,
    },
    paths: parsed.paths,
    logs: full ? parsed.logs : undefined,
    valuesPrinted: false,
  };
}

function renderCollectorManifest(archive: WechatArchiveConfig): string {
  const ingress = archive.personalWechatIngress;
  const collector = ingress.collector;
  const labels = [
    "app.kubernetes.io/name: personal-wechat-collector",
    "app.kubernetes.io/part-of: platform-infra",
    "app.kubernetes.io/managed-by: unidesk",
    "unidesk.ai/component: personal-wechat-collector",
  ].join("\n    ");
  return `apiVersion: v1
kind: ServiceAccount
metadata:
  name: ${collector.workload.serviceAccountName}
  namespace: ${collector.namespace}
  labels:
    ${labels}
---
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
  name: ${collector.storage.name}
  namespace: ${collector.namespace}
  labels:
    ${labels}
spec:
  accessModes:
    - ReadWriteOnce
  resources:
    requests:
      storage: ${collector.storage.size}
---
apiVersion: v1
kind: ConfigMap
metadata:
  name: personal-wechat-collector-config
  namespace: ${collector.namespace}
  labels:
    ${labels}
data:
  WCF_HOST: "${collector.wcfHost}"
  WCF_COMMAND_PORT: "${collector.commandPort}"
  STATE_ROOT: "${collector.stateRoot}"
  ARCHIVE_CALLBACK_URL: "${archive.archiveCallback.publicUrl}"
  ARCHIVE_REMOTE_ROOT: "${archive.baiduNetdisk.archive.remoteRoot}"
  ARCHIVE_TIMEZONE: "${archive.baiduNetdisk.archive.timezone}"
  POLL_IDLE_SECONDS: "1"
  STATUS_INTERVAL_SECONDS: "15"
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: ${collector.workload.name}
  namespace: ${collector.namespace}
  labels:
    ${labels}
spec:
  replicas: ${collector.workload.replicas}
  selector:
    matchLabels:
      app.kubernetes.io/name: personal-wechat-collector
      unidesk.ai/component: personal-wechat-collector
  template:
    metadata:
      labels:
        app.kubernetes.io/name: personal-wechat-collector
        app.kubernetes.io/part-of: platform-infra
        unidesk.ai/component: personal-wechat-collector
      annotations:
        unidesk.ai/wcf-host: "${collector.wcfHost}:${collector.commandPort}"
        unidesk.ai/read-only: "${collector.readOnly.enabled}"
        unidesk.ai/send-capability: "${collector.readOnly.sendCapability}"
        unidesk.ai/archive-callback-ref: "${collector.archiveCallbackRef}"
    spec:
      serviceAccountName: ${collector.workload.serviceAccountName}
      containers:
        - name: ${collector.workload.containerName}
          image: ${collectorImageRef(ingress)}
          imagePullPolicy: ${collector.image.pullPolicy}
          envFrom:
            - configMapRef:
                name: personal-wechat-collector-config
          env:
            - name: ARCHIVE_CALLBACK_TOKEN
              valueFrom:
                secretKeyRef:
                  name: ${collector.secretName}
                  key: ${archive.archiveCallback.tokenKey}
          volumeMounts:
            - name: state
              mountPath: ${collector.storage.mountPath}
      volumes:
        - name: state
          persistentVolumeClaim:
            claimName: ${collector.storage.name}
`;
}

function collectorPolicyChecks(archive: WechatArchiveConfig, manifest: string): Array<Record<string, unknown>> {
  const collector = archive.personalWechatIngress.collector;
  return [
    { name: "no-namespace-create", ok: !/^\s*kind:\s*Namespace\s*$/mu.test(manifest) && collector.createNamespace === false, detail: "collector apply must reuse the existing platform-infra namespace and must not render/create Namespace." },
    { name: "target-d601-k3s", ok: archive.personalWechatIngress.target.kubeRoute === "D601:k3s" && collector.runtime === "d601-k3s", detail: "collector runtime is D601 native k3s." },
    { name: "namespace-platform-infra", ok: collector.namespace === "platform-infra", detail: "collector deploys into the existing platform-infra namespace." },
    { name: "no-public-exposure", ok: !/^\s*kind:\s*(Ingress|Service)\s*$/mu.test(manifest) && archive.personalWechatIngress.wcfHost.firewall.publicExposure === false, detail: "collector has no public ingress, node port, load balancer or service." },
    { name: "no-host-network", ok: !/^\s*hostNetwork:\s*true\s*$/mu.test(manifest), detail: "collector must not use hostNetwork." },
    { name: "read-only-methods", ok: collector.readOnly.enabled === true && collector.readOnly.sendCapability === false, detail: "collector only invokes read-only WCF methods in source." },
    { name: "callback-token-source-ref", ok: Boolean(archive.archiveCallback.tokenSourceRef && archive.archiveCallback.tokenKey && collector.secretName), detail: "callback token is synced from YAML-declared sourceRef into a runtime Secret." },
  ];
}

function collectorImageBuildScript(archive: WechatArchiveConfig): string {
  const ingress = archive.personalWechatIngress;
  const dockerfileB64 = fileB64(ingress.collector.image.dockerfile);
  const collectorB64 = fileB64(pathPosix.join(ingress.collector.image.context, "collector.py"));
  const image = collectorImageRef(ingress);
  const jobRoot = "/tmp/unidesk-personal-wechat-collector-build";
  return `
set -u
job_root=${shQuote(jobRoot)}
job_dir="$job_root/work"
status_file="$job_root/status.json"
runner="$job_root/build-runner.sh"
mkdir -p "$job_root"
if [ -f "$job_root/pid" ] && kill -0 "$(cat "$job_root/pid")" 2>/dev/null; then
  python3 - "$status_file" "$(cat "$job_root/pid")" <<'PY'
import json, sys
payload = {"ok": True, "started": False, "alreadyRunning": True, "pid": sys.argv[2], "statusFile": sys.argv[1]}
print(json.dumps(payload, ensure_ascii=False, indent=2))
PY
  exit 0
fi
rm -rf "$job_dir"
mkdir -p "$job_dir"
printf '%s' '${dockerfileB64}' | base64 -d > "$job_dir/Dockerfile"
printf '%s' '${collectorB64}' | base64 -d > "$job_dir/collector.py"
build_log="$job_dir/docker-build.log"
push_log="$job_dir/docker-push.log"
cat >"$runner" <<'SH'
#!/usr/bin/env bash
set +e
job_dir=${shQuote(`${jobRoot}/work`)}
status_file=${shQuote(`${jobRoot}/status.json`)}
build_log="$job_dir/docker-build.log"
push_log="$job_dir/docker-push.log"
python3 - "$status_file" "$build_log" "$push_log" <<'PY'
import json, sys, time
payload = {"ok": False, "status": "running", "startedAt": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "image": "${image}", "logs": {"build": sys.argv[2], "push": sys.argv[3]}}
open(sys.argv[1], "w", encoding="utf-8").write(json.dumps(payload, ensure_ascii=False, indent=2))
PY
docker build \\
  --build-arg WCFERRY_VERSION=${shQuote(ingress.collector.image.wcferryVersion)} \\
  --build-arg PIP_INDEX_URL=${shQuote(ingress.collector.image.pip.indexUrl)} \\
  --build-arg PIP_TRUSTED_HOST=${shQuote(ingress.collector.image.pip.trustedHost)} \\
  --build-arg PIP_TIMEOUT_SECONDS=${shQuote(String(ingress.collector.image.pip.timeoutSeconds))} \\
  --build-arg PIP_RETRIES=${shQuote(String(ingress.collector.image.pip.retries))} \\
  -t ${shQuote(image)} "$job_dir" >"$build_log" 2>&1
build_rc=$?
if [ "$build_rc" -eq 0 ]; then
  docker push ${shQuote(image)} >"$push_log" 2>&1
  push_rc=$?
else
  : >"$push_log"
  push_rc=1
fi
python3 - "$status_file" "$build_rc" "$push_rc" "$build_log" "$push_log" <<'PY'
import json, sys
status_file, build_rc, push_rc = sys.argv[1], int(sys.argv[2]), int(sys.argv[3])
def text(path, limit=8000):
    try:
        return open(path, encoding="utf-8", errors="replace").read()[-limit:]
    except FileNotFoundError:
        return ""
payload = {
    "ok": build_rc == 0 and push_rc == 0,
    "status": "succeeded" if build_rc == 0 and push_rc == 0 else "failed",
    "image": "${image}",
    "wcferryVersion": "${ingress.collector.image.wcferryVersion}",
    "steps": {
        "build": {"exitCode": build_rc, "log": sys.argv[4], "logTail": text(sys.argv[4])},
        "push": {"exitCode": push_rc, "log": sys.argv[5], "logTail": text(sys.argv[5])},
    },
}
open(status_file, "w", encoding="utf-8").write(json.dumps(payload, ensure_ascii=False, indent=2) + "\\n")
sys.exit(0 if payload["ok"] else 1)
PY
SH
chmod +x "$runner"
nohup "$runner" >"$job_root/nohup.out" 2>"$job_root/nohup.err" &
pid=$!
printf '%s' "$pid" >"$job_root/pid"
python3 - "$status_file" "$pid" "$job_root" <<'PY'
import json, sys
payload = {"ok": True, "started": True, "pid": sys.argv[2], "statusFile": sys.argv[1], "jobRoot": sys.argv[3], "statusCommand": "bun scripts/cli.ts platform-infra wechat-archive collector-image-status --full"}
print(json.dumps(payload, ensure_ascii=False, indent=2))
PY
`;
}

function collectorImageStatusScript(archive: WechatArchiveConfig): string {
  const image = collectorImageRef(archive.personalWechatIngress);
  return `
set -u
tmp="$(mktemp -d)"
trap 'rm -rf "$tmp"' EXIT
job_root=/tmp/unidesk-personal-wechat-collector-build
status_file="$job_root/status.json"
pid_file="$job_root/pid"
running=0
if [ -f "$pid_file" ] && kill -0 "$(cat "$pid_file")" 2>/dev/null; then
  running=1
fi
docker image inspect ${shQuote(image)} >"$tmp/local.json" 2>"$tmp/local.err"
local_rc=$?
docker manifest inspect ${shQuote(image)} >"$tmp/manifest.json" 2>"$tmp/manifest.err"
manifest_rc=$?
python3 - "$local_rc" "$manifest_rc" "$running" "$status_file" "$tmp/local.json" "$tmp/local.err" "$tmp/manifest.json" "$tmp/manifest.err" <<'PY'
import json, sys
local_rc, manifest_rc, running = int(sys.argv[1]), int(sys.argv[2]), sys.argv[3] == "1"
def load(path):
    try:
        return json.load(open(path, encoding="utf-8"))
    except Exception:
        return None
def text(path):
    try:
        return open(path, encoding="utf-8", errors="replace").read()[-2000:]
    except FileNotFoundError:
        return ""
build_status = load(sys.argv[4])
local = load(sys.argv[5])
payload = {
    "ok": True,
    "image": "${image}",
    "build": build_status,
    "buildRunning": running,
    "localPresent": local_rc == 0,
    "registryManifestPresent": manifest_rc == 0,
    "local": {
        "id": local[0].get("Id") if isinstance(local, list) and local else None,
        "created": local[0].get("Created") if isinstance(local, list) and local else None,
        "repoDigests": local[0].get("RepoDigests") if isinstance(local, list) and local else [],
    },
    "errors": {
        "local": text(sys.argv[6]) if local_rc != 0 else "",
        "manifest": text(sys.argv[8]) if manifest_rc != 0 else "",
    },
}
print(json.dumps(payload, ensure_ascii=False, indent=2))
PY
`;
}

function collectorApplyScript(archive: WechatArchiveConfig, manifest: string, token: string | null, dryRun: boolean): string {
  const ingress = archive.personalWechatIngress;
  const collector = ingress.collector;
  const manifestB64 = b64(manifest);
  const tokenB64 = token === null ? "" : b64(token);
  const mode = dryRun ? "dry-run" : "confirmed";
  return `
set -u
tmp="$(mktemp -d)"
trap 'rm -rf "$tmp"' EXIT
manifest="$tmp/personal-wechat-collector.yaml"
printf '%s' '${manifestB64}' | base64 -d > "$manifest"
kubectl get namespace ${collector.namespace} >"$tmp/ns.out" 2>"$tmp/ns.err"
ns_rc=$?
secret_rc=0
apply_rc=1
if [ "$ns_rc" -eq 0 ]; then
  if [ ${dryRun ? "1" : "0"} -eq 1 ]; then
    kubectl apply --dry-run=client -f "$manifest" >"$tmp/apply.out" 2>"$tmp/apply.err"
    apply_rc=$?
  else
    printf '%s' '${tokenB64}' | base64 -d > "$tmp/archive-token"
    kubectl -n ${collector.namespace} create secret generic ${collector.secretName} \\
      --from-file=${archive.archiveCallback.tokenKey}="$tmp/archive-token" \\
      --dry-run=client -o yaml | kubectl apply --server-side --force-conflicts --field-manager=${collectorFieldManager} -f - >"$tmp/secret.out" 2>"$tmp/secret.err"
    secret_rc=$?
    if [ "$secret_rc" -eq 0 ]; then
      kubectl apply --server-side --force-conflicts --field-manager=${collectorFieldManager} -f "$manifest" >"$tmp/apply.out" 2>"$tmp/apply.err"
      apply_rc=$?
    else
      : >"$tmp/apply.out"
      printf '%s\\n' 'skipped because secret sync failed' >"$tmp/apply.err"
      apply_rc=1
    fi
  fi
else
  : >"$tmp/secret.out"; : >"$tmp/secret.err"; : >"$tmp/apply.out"
  printf '%s\\n' 'skipped because declared namespace does not exist' >"$tmp/apply.err"
fi
python3 - "$ns_rc" "$secret_rc" "$apply_rc" "$tmp/ns.out" "$tmp/ns.err" "$tmp/secret.out" "$tmp/secret.err" "$tmp/apply.out" "$tmp/apply.err" <<'PY'
import json, sys
ns_rc, secret_rc, apply_rc = int(sys.argv[1]), int(sys.argv[2]), int(sys.argv[3])
def text(path, limit=8000):
    try:
        return open(path, encoding="utf-8", errors="replace").read()[-limit:]
    except FileNotFoundError:
        return ""
payload = {
    "ok": ns_rc == 0 and secret_rc == 0 and apply_rc == 0,
    "mode": "${mode}",
    "namespace": "${collector.namespace}",
    "image": "${collectorImageRef(ingress)}",
    "secret": {"name": "${collector.secretName}", "key": "${archive.archiveCallback.tokenKey}", "applied": ${dryRun ? "False" : "True"}, "valuesPrinted": False},
    "steps": {
        "namespace": {"exitCode": ns_rc, "stdout": text(sys.argv[4]), "stderr": text(sys.argv[5])},
        "secret": {"exitCode": secret_rc, "stdout": text(sys.argv[6]), "stderr": text(sys.argv[7])},
        "apply": {"exitCode": apply_rc, "stdout": text(sys.argv[8]), "stderr": text(sys.argv[9])},
    },
}
print(json.dumps(payload, ensure_ascii=False, indent=2))
sys.exit(0 if payload["ok"] else 1)
PY
`;
}

function collectorStatusScript(archive: WechatArchiveConfig, full: boolean): string {
  const collector = archive.personalWechatIngress.collector;
  return `
set -u
tmp="$(mktemp -d)"
trap 'rm -rf "$tmp"' EXIT
capture() {
  name="$1"
  shift
  "$@" -o json >"$tmp/$name.json" 2>"$tmp/$name.err"
  printf '%s' "$?" >"$tmp/$name.rc"
}
capture namespace kubectl get namespace ${collector.namespace}
capture deploy kubectl -n ${collector.namespace} get deploy ${collector.workload.name}
capture pods kubectl -n ${collector.namespace} get pods -l app.kubernetes.io/name=personal-wechat-collector
capture pvc kubectl -n ${collector.namespace} get pvc ${collector.storage.name}
capture secret kubectl -n ${collector.namespace} get secret ${collector.secretName}
capture configmap kubectl -n ${collector.namespace} get configmap personal-wechat-collector-config
capture events kubectl -n ${collector.namespace} get events --sort-by=.lastTimestamp
pod="$(kubectl -n ${collector.namespace} get pods -l app.kubernetes.io/name=personal-wechat-collector -o jsonpath='{.items[0].metadata.name}' 2>/dev/null || true)"
if [ -n "$pod" ]; then
  kubectl -n ${collector.namespace} logs "$pod" --tail ${full ? "160" : "40"} >"$tmp/logs.txt" 2>"$tmp/logs.err"
  logs_rc=$?
else
  : >"$tmp/logs.txt"
  printf '%s\\n' 'collector pod not found' >"$tmp/logs.err"
  logs_rc=1
fi
python3 - "$tmp" "$logs_rc" "$pod" <<'PY'
import json, os, sys
tmp, logs_rc, pod = sys.argv[1], int(sys.argv[2]), sys.argv[3]
def load(name):
    try:
        return json.load(open(os.path.join(tmp, f"{name}.json"), encoding="utf-8"))
    except Exception:
        return None
def rc(name):
    try:
        return int(open(os.path.join(tmp, f"{name}.rc"), encoding="utf-8").read() or "1")
    except FileNotFoundError:
        return 1
def text(path, limit=8000):
    try:
        return open(path, encoding="utf-8", errors="replace").read()[-limit:]
    except FileNotFoundError:
        return ""
deploy = load("deploy") or {}
status = deploy.get("status", {}) if isinstance(deploy, dict) else {}
spec = deploy.get("spec", {}) if isinstance(deploy, dict) else {}
pods = load("pods") or {"items": []}
events = load("events") or {"items": []}
pod_items = pods.get("items", []) if isinstance(pods, dict) else []
def pod_summary(item):
    pod_status = item.get("status", {})
    containers = pod_status.get("containerStatuses") or []
    return {
        "name": item.get("metadata", {}).get("name"),
        "phase": pod_status.get("phase"),
        "podIP": pod_status.get("podIP"),
        "nodeName": item.get("spec", {}).get("nodeName"),
        "containers": [
            {
                "name": c.get("name"),
                "ready": c.get("ready"),
                "restartCount": c.get("restartCount"),
                "state": c.get("state"),
                "lastState": c.get("lastState"),
            }
            for c in containers
        ],
    }
def event_summary(event):
    involved = event.get("involvedObject", {})
    if involved.get("name") and "personal-wechat-collector" not in involved.get("name", ""):
        return None
    return {
        "type": event.get("type"),
        "reason": event.get("reason"),
        "message": str(event.get("message") or "")[-600:],
        "count": event.get("count"),
        "lastTimestamp": event.get("lastTimestamp") or event.get("eventTime"),
        "involvedObject": {"kind": involved.get("kind"), "name": involved.get("name")},
    }
related_events = [item for item in (event_summary(e) for e in events.get("items", [])[-30:]) if item]
desired_replicas = spec.get("replicas", 1)
paused = rc("deploy") == 0 and desired_replicas == 0
ready = rc("deploy") == 0 and (paused or status.get("readyReplicas", 0) >= desired_replicas)
payload = {
    "ok": rc("namespace") == 0 and rc("deploy") == 0,
    "namespace": "${collector.namespace}",
    "deployment": {
        "name": "${collector.workload.name}",
        "replicas": spec.get("replicas"),
        "readyReplicas": status.get("readyReplicas", 0),
        "updatedReplicas": status.get("updatedReplicas", 0),
        "availableReplicas": status.get("availableReplicas", 0),
        "ready": ready,
        "paused": paused,
    },
    "pvc": {"present": rc("pvc") == 0, "name": "${collector.storage.name}"},
    "secret": {"present": rc("secret") == 0, "name": "${collector.secretName}", "valuesPrinted": False},
    "configMap": {"present": rc("configmap") == 0, "name": "personal-wechat-collector-config"},
    "pods": [pod_summary(item) for item in pod_items],
    "logs": {
        "pod": pod,
        "exitCode": 0 if paused else logs_rc,
        "tail": text(os.path.join(tmp, "logs.txt")),
        "stderrTail": "collector paused by YAML replicas=0\\n" if paused and not pod else text(os.path.join(tmp, "logs.err"), 2000),
    },
    "events": related_events[-12:],
    "valuesPrinted": False,
}
print(json.dumps(payload, ensure_ascii=False, indent=2))
PY
`;
}

function validateConfig(config: WechatArchiveConfig): void {
  if (config.version !== 1) throw new Error(`${configLabel}.version must be 1`);
  if (config.target.id !== "G14" || config.target.route !== "G14:k3s") throw new Error(`${configLabel}.target currently supports only G14:k3s`);
  for (const ref of [config.langbot.configRef, config.n8n.configRef]) {
    const path = resolveRepoPath(ref);
    if (!existsSync(path)) throw new Error(`${ref} does not exist`);
  }
  if (!config.n8n.publicBaseUrl.startsWith("https://")) throw new Error(`${configLabel}.n8n.publicBaseUrl must be https`);
  if (!config.langbot.publicBaseUrl.startsWith("https://")) throw new Error(`${configLabel}.langbot.publicBaseUrl must be https`);
  if (config.langbot.readOnlyRecord.enabled && config.langbot.readOnlyRecord.mode !== "webhook-skip-pipeline") throw new Error(`${configLabel}.langbot.readOnlyRecord.mode must be webhook-skip-pipeline`);
  if (config.langbot.readOnlyRecord.enabled && config.langbot.readOnlyRecord.discardPipelineUuid !== "__discard__") throw new Error(`${configLabel}.langbot.readOnlyRecord.discardPipelineUuid must be __discard__ for read-only personal WeChat recording`);
  if (!/^https?:\/\//u.test(config.archiveCallback.publicUrl)) throw new Error(`${configLabel}.archiveCallback.publicUrl must be http or https`);
  if (!/^[A-Za-z_][A-Za-z0-9_]*$/u.test(config.archiveCallback.tokenKey)) throw new Error(`${configLabel}.archiveCallback.tokenKey must be an env key`);
  if (!config.n8n.workflow.webhookPath || config.n8n.workflow.webhookPath.startsWith("/") || config.n8n.workflow.webhookPath.includes("/")) {
    throw new Error(`${configLabel}.n8n.workflow.webhookPath must be one relative path segment`);
  }
  if (!config.baiduNetdisk.archive.pathTemplate.includes("{{remoteRoot}}")) throw new Error(`${configLabel}.baiduNetdisk.archive.pathTemplate must include {{remoteRoot}}`);
  if (config.baiduNetdisk.proxyMode !== "backend-core-microservice-proxy") throw new Error(`${configLabel}.baiduNetdisk.proxyMode must be backend-core-microservice-proxy`);
  validatePersonalWechatIngress(config.personalWechatIngress);
}

function assertTarget(config: WechatArchiveConfig, targetId: string): void {
  if (targetId !== config.target.id) throw new Error(`wechat archive target ${targetId} is not declared in ${configLabel}`);
}

function configSummary(config: WechatArchiveConfig): Record<string, unknown> {
  return {
    path: configLabel,
    id: config.metadata.id,
    relatedIssues: config.metadata.relatedIssues,
    target: config.target,
    langbot: {
      configRef: config.langbot.configRef,
      publicBaseUrl: config.langbot.publicBaseUrl,
      expectedAdapter: config.langbot.expectedAdapter,
      callbackPath: config.langbot.callbackPath,
      readOnlyRecord: {
        enabled: config.langbot.readOnlyRecord.enabled,
        mode: config.langbot.readOnlyRecord.mode,
        webhookName: config.langbot.readOnlyRecord.webhookName,
        discardPipelineUuid: config.langbot.readOnlyRecord.discardPipelineUuid,
      },
      pipeline: config.langbot.pipeline,
    },
    n8n: {
      configRef: config.n8n.configRef,
      publicBaseUrl: config.n8n.publicBaseUrl,
      workflow: config.n8n.workflow,
      webhookUrl: webhookUrl(config),
    },
    archiveCallback: {
      publicUrl: config.archiveCallback.publicUrl,
      secretRoot: config.archiveCallback.secretRoot,
      tokenSourceRef: config.archiveCallback.tokenSourceRef,
      tokenKey: config.archiveCallback.tokenKey,
      timeoutMs: config.archiveCallback.timeoutMs,
      valuesPrinted: false,
    },
    personalWechatIngress: personalWechatIngressSummary(config.personalWechatIngress),
    baiduNetdisk: {
      serviceId: config.baiduNetdisk.serviceId,
      proxyMode: config.baiduNetdisk.proxyMode,
      configRef: config.baiduNetdisk.configRef,
      remoteRoot: config.baiduNetdisk.archive.remoteRoot,
      staging: config.baiduNetdisk.staging,
    },
    valuesPrinted: false,
  };
}

function policyChecks(config: WechatArchiveConfig): Array<Record<string, unknown>> {
  return [
    { name: "yaml-source", ok: repoRelative(configFile) === configLabel, detail: `${configLabel} is the archive workflow source of truth.` },
    { name: "langbot-yaml-ref", ok: config.langbot.configRef === "config/platform-infra/langbot.yaml", detail: "LangBot base service remains declared in YAML." },
    { name: "n8n-yaml-ref", ok: config.n8n.configRef === "config/platform-infra/n8n.yaml", detail: "n8n base service remains declared in YAML." },
    { name: "baidu-private-proxy", ok: config.baiduNetdisk.proxyMode === "backend-core-microservice-proxy", detail: "Baidu Netdisk access stays behind backend-core microservice proxy." },
    { name: "callback-token-source-ref", ok: Boolean(config.archiveCallback.tokenSourceRef && config.archiveCallback.tokenKey), detail: "n8n archive callback token is read from a YAML-declared local sourceRef and never reverse-engineered from runtime." },
    { name: "personal-wechat-read-only", ok: config.langbot.readOnlyRecord.enabled && config.langbot.readOnlyRecord.discardPipelineUuid === "__discard__", detail: "LangBot inbound webhook must return skip_pipeline=true and bot fallback must discard so personal WeChat receives no automated replies." },
    { name: "personal-wechat-ingress-yaml", ok: config.personalWechatIngress.enabled === true, detail: "D601 PC WeChat + WeChatFerry host and k3s collector boundary is declared in YAML." },
    { name: "wcf-version-pin", ok: config.personalWechatIngress.pcWechat.requiredVersion === config.personalWechatIngress.wcfHost.requiredWechatVersion, detail: "The isolated PC WeChat version and WCF host requirement must match in YAML." },
    { name: "wcf-no-public-exposure", ok: config.personalWechatIngress.wcfHost.firewall.publicExposure === false, detail: "WeChatFerry raw RPC must not be publicly exposed." },
    { name: "collector-existing-namespace", ok: config.personalWechatIngress.collector.runtime === "d601-k3s" && config.personalWechatIngress.collector.namespace === "platform-infra" && config.personalWechatIngress.collector.createNamespace === false, detail: "D601 collector reuses the existing platform-infra namespace; namespace creation is disabled." },
    { name: "collector-read-only", ok: config.personalWechatIngress.collector.readOnly.enabled === true && config.personalWechatIngress.collector.readOnly.sendCapability === false, detail: "k3s collector is a read-only WCF RPC client and must not export send/friend/group/database surfaces." },
    { name: "collector-port-alignment", ok: config.personalWechatIngress.collector.commandPort === config.personalWechatIngress.wcfHost.commandPort && config.personalWechatIngress.collector.messagePort === config.personalWechatIngress.wcfHost.messagePort, detail: "Collector WCF ports must match the Windows WCF host ports declared in YAML." },
    { name: "test-account-first", ok: config.personalWechatIngress.poc.accountPolicy === "test-account-first", detail: "Initial PoC must use a test or low-risk WeChat account before any production account promotion." },
    { name: "single-pk01-postgres", ok: true, detail: "LangBot and n8n use dedicated databases inside the single PK01/Pika01 external PostgreSQL instance; this workflow adds no PostgreSQL instance." },
    { name: "text-enabled", ok: config.messageTypes.text?.enabled === true, detail: "Text messages are archive-enabled." },
    { name: "image-enabled", ok: config.messageTypes.image?.enabled === true, detail: "Image messages are archive-enabled." },
    { name: "no-secret-output", ok: true, detail: "Baidu, LangBot and n8n secrets are referenced by existing services and are not printed by this CLI." },
  ];
}

function parsePersonalWechatIngress(raw: Record<string, unknown>): WechatArchiveConfig["personalWechatIngress"] {
  const target = recordField(raw, "target", `${configLabel}.personalWechatIngress`);
  const pcWechat = recordField(raw, "pcWechat", `${configLabel}.personalWechatIngress`);
  const wcfHost = recordField(raw, "wcfHost", `${configLabel}.personalWechatIngress`);
  const firewall = recordField(wcfHost, "firewall", `${configLabel}.personalWechatIngress.wcfHost`);
  const collector = recordField(raw, "collector", `${configLabel}.personalWechatIngress`);
  const workload = recordField(collector, "workload", `${configLabel}.personalWechatIngress.collector`);
  const storage = recordField(collector, "storage", `${configLabel}.personalWechatIngress.collector`);
  const image = recordField(collector, "image", `${configLabel}.personalWechatIngress.collector`);
  const pip = recordField(image, "pip", `${configLabel}.personalWechatIngress.collector.image`);
  const readOnly = recordField(collector, "readOnly", `${configLabel}.personalWechatIngress.collector`);
  const poc = recordField(raw, "poc", `${configLabel}.personalWechatIngress`);
  return {
    enabled: booleanField(raw, "enabled", `${configLabel}.personalWechatIngress`),
    mode: stringField(raw, "mode", `${configLabel}.personalWechatIngress`),
    target: {
      id: stringField(target, "id", `${configLabel}.personalWechatIngress.target`),
      windowsRoute: stringField(target, "windowsRoute", `${configLabel}.personalWechatIngress.target`),
      hostRoute: stringField(target, "hostRoute", `${configLabel}.personalWechatIngress.target`),
      kubeRoute: stringField(target, "kubeRoute", `${configLabel}.personalWechatIngress.target`),
    },
    pcWechat: {
      isolation: stringField(pcWechat, "isolation", `${configLabel}.personalWechatIngress.pcWechat`),
      requiredVersion: stringField(pcWechat, "requiredVersion", `${configLabel}.personalWechatIngress.pcWechat`),
      installerAsset: stringField(pcWechat, "installerAsset", `${configLabel}.personalWechatIngress.pcWechat`),
      installMode: stringField(pcWechat, "installMode", `${configLabel}.personalWechatIngress.pcWechat`),
      installRoot: stringField(pcWechat, "installRoot", `${configLabel}.personalWechatIngress.pcWechat`),
      dataRoot: stringField(pcWechat, "dataRoot", `${configLabel}.personalWechatIngress.pcWechat`),
      autoUpdatePolicy: stringField(pcWechat, "autoUpdatePolicy", `${configLabel}.personalWechatIngress.pcWechat`),
    },
    wcfHost: {
      releaseTag: stringField(wcfHost, "releaseTag", `${configLabel}.personalWechatIngress.wcfHost`),
      requiredWechatVersion: stringField(wcfHost, "requiredWechatVersion", `${configLabel}.personalWechatIngress.wcfHost`),
      releaseUrl: stringField(wcfHost, "releaseUrl", `${configLabel}.personalWechatIngress.wcfHost`),
      sdkRoot: stringField(wcfHost, "sdkRoot", `${configLabel}.personalWechatIngress.wcfHost`),
      stateRoot: stringField(wcfHost, "stateRoot", `${configLabel}.personalWechatIngress.wcfHost`),
      commandPort: numberField(wcfHost, "commandPort", `${configLabel}.personalWechatIngress.wcfHost`),
      messagePort: numberField(wcfHost, "messagePort", `${configLabel}.personalWechatIngress.wcfHost`),
      bindHost: stringField(wcfHost, "bindHost", `${configLabel}.personalWechatIngress.wcfHost`),
      supervisor: stringField(wcfHost, "supervisor", `${configLabel}.personalWechatIngress.wcfHost`),
      rdpPolicy: stringField(wcfHost, "rdpPolicy", `${configLabel}.personalWechatIngress.wcfHost`),
      firewall: {
        publicExposure: booleanField(firewall, "publicExposure", `${configLabel}.personalWechatIngress.wcfHost.firewall`),
        allowedClientRefs: stringArrayField(firewall, "allowedClientRefs", `${configLabel}.personalWechatIngress.wcfHost.firewall`),
      },
    },
    collector: {
      runtime: stringField(collector, "runtime", `${configLabel}.personalWechatIngress.collector`),
      namespace: stringField(collector, "namespace", `${configLabel}.personalWechatIngress.collector`),
      createNamespace: booleanField(collector, "createNamespace", `${configLabel}.personalWechatIngress.collector`),
      workload: {
        kind: stringField(workload, "kind", `${configLabel}.personalWechatIngress.collector.workload`),
        name: stringField(workload, "name", `${configLabel}.personalWechatIngress.collector.workload`),
        serviceAccountName: stringField(workload, "serviceAccountName", `${configLabel}.personalWechatIngress.collector.workload`),
        replicas: numberField(workload, "replicas", `${configLabel}.personalWechatIngress.collector.workload`),
        containerName: stringField(workload, "containerName", `${configLabel}.personalWechatIngress.collector.workload`),
      },
      storage: {
        kind: stringField(storage, "kind", `${configLabel}.personalWechatIngress.collector.storage`),
        name: stringField(storage, "name", `${configLabel}.personalWechatIngress.collector.storage`),
        mountPath: stringField(storage, "mountPath", `${configLabel}.personalWechatIngress.collector.storage`),
        size: stringField(storage, "size", `${configLabel}.personalWechatIngress.collector.storage`),
        create: booleanField(storage, "create", `${configLabel}.personalWechatIngress.collector.storage`),
      },
      image: {
        repository: stringField(image, "repository", `${configLabel}.personalWechatIngress.collector.image`),
        tag: stringField(image, "tag", `${configLabel}.personalWechatIngress.collector.image`),
        pullPolicy: stringField(image, "pullPolicy", `${configLabel}.personalWechatIngress.collector.image`),
        context: stringField(image, "context", `${configLabel}.personalWechatIngress.collector.image`),
        dockerfile: stringField(image, "dockerfile", `${configLabel}.personalWechatIngress.collector.image`),
        wcferryVersion: stringField(image, "wcferryVersion", `${configLabel}.personalWechatIngress.collector.image`),
        pip: {
          indexUrl: stringField(pip, "indexUrl", `${configLabel}.personalWechatIngress.collector.image.pip`),
          trustedHost: stringField(pip, "trustedHost", `${configLabel}.personalWechatIngress.collector.image.pip`),
          timeoutSeconds: numberField(pip, "timeoutSeconds", `${configLabel}.personalWechatIngress.collector.image.pip`),
          retries: numberField(pip, "retries", `${configLabel}.personalWechatIngress.collector.image.pip`),
        },
      },
      wcfHost: stringField(collector, "wcfHost", `${configLabel}.personalWechatIngress.collector`),
      commandPort: numberField(collector, "commandPort", `${configLabel}.personalWechatIngress.collector`),
      messagePort: numberField(collector, "messagePort", `${configLabel}.personalWechatIngress.collector`),
      stateRoot: stringField(collector, "stateRoot", `${configLabel}.personalWechatIngress.collector`),
      queueMode: stringField(collector, "queueMode", `${configLabel}.personalWechatIngress.collector`),
      outboxMode: stringField(collector, "outboxMode", `${configLabel}.personalWechatIngress.collector`),
      archiveCallbackRef: stringField(collector, "archiveCallbackRef", `${configLabel}.personalWechatIngress.collector`),
      secretName: stringField(collector, "secretName", `${configLabel}.personalWechatIngress.collector`),
      readOnly: {
        enabled: booleanField(readOnly, "enabled", `${configLabel}.personalWechatIngress.collector.readOnly`),
        sendCapability: booleanField(readOnly, "sendCapability", `${configLabel}.personalWechatIngress.collector.readOnly`),
        allowedMethods: stringArrayField(readOnly, "allowedMethods", `${configLabel}.personalWechatIngress.collector.readOnly`),
        forbiddenMethods: stringArrayField(readOnly, "forbiddenMethods", `${configLabel}.personalWechatIngress.collector.readOnly`),
      },
    },
    poc: {
      accountPolicy: stringField(poc, "accountPolicy", `${configLabel}.personalWechatIngress.poc`),
      observationWindowHours: numberField(poc, "observationWindowHours", `${configLabel}.personalWechatIngress.poc`),
      requiredMessageTypes: stringArrayField(poc, "requiredMessageTypes", `${configLabel}.personalWechatIngress.poc`),
    },
  };
}

function validatePersonalWechatIngress(config: WechatArchiveConfig["personalWechatIngress"]): void {
  if (!config.enabled) return;
  if (config.mode !== "d601-windows-wcf-host-with-k3s-collector") {
    throw new Error(`${configLabel}.personalWechatIngress.mode must be d601-windows-wcf-host-with-k3s-collector`);
  }
  if (config.target.id !== "D601" || config.target.kubeRoute !== "D601:k3s") {
    throw new Error(`${configLabel}.personalWechatIngress.target must route the collector through D601:k3s`);
  }
  if (config.collector.runtime !== "d601-k3s") {
    throw new Error(`${configLabel}.personalWechatIngress.collector.runtime must be d601-k3s`);
  }
  if (config.collector.namespace !== "platform-infra" || config.collector.createNamespace !== false) {
    throw new Error(`${configLabel}.personalWechatIngress.collector must reuse platform-infra with createNamespace=false`);
  }
  if (config.collector.workload.kind !== "Deployment") {
    throw new Error(`${configLabel}.personalWechatIngress.collector.workload.kind must be Deployment`);
  }
  if (config.collector.storage.kind !== "PersistentVolumeClaim") {
    throw new Error(`${configLabel}.personalWechatIngress.collector.storage.kind must be PersistentVolumeClaim`);
  }
  for (const repoPath of [config.collector.image.context, config.collector.image.dockerfile]) {
    if (!existsSync(resolveRepoPath(repoPath))) throw new Error(`${configLabel}.personalWechatIngress.collector.image path does not exist: ${repoPath}`);
  }
  if (!["Always", "IfNotPresent", "Never"].includes(config.collector.image.pullPolicy)) {
    throw new Error(`${configLabel}.personalWechatIngress.collector.image.pullPolicy must be Always, IfNotPresent, or Never`);
  }
  if (config.pcWechat.requiredVersion !== config.wcfHost.requiredWechatVersion) {
    throw new Error(`${configLabel}.personalWechatIngress pcWechat.requiredVersion and wcfHost.requiredWechatVersion must match`);
  }
  if (config.collector.commandPort !== config.wcfHost.commandPort || config.collector.messagePort !== config.wcfHost.messagePort) {
    throw new Error(`${configLabel}.personalWechatIngress collector ports must match wcfHost ports`);
  }
  if (config.wcfHost.firewall.publicExposure !== false) {
    throw new Error(`${configLabel}.personalWechatIngress.wcfHost.firewall.publicExposure must be false`);
  }
  if (config.collector.readOnly.enabled !== true || config.collector.readOnly.sendCapability !== false) {
    throw new Error(`${configLabel}.personalWechatIngress.collector.readOnly must enable read-only mode and set sendCapability=false`);
  }
  const allowed = new Set(config.collector.readOnly.allowedMethods);
  const overlap = config.collector.readOnly.forbiddenMethods.filter((method) => allowed.has(method));
  if (overlap.length > 0) throw new Error(`${configLabel}.personalWechatIngress.collector.readOnly method allowlist overlaps forbiddenMethods: ${overlap.join(",")}`);
  if (config.collector.archiveCallbackRef !== "archiveCallback") {
    throw new Error(`${configLabel}.personalWechatIngress.collector.archiveCallbackRef must point at archiveCallback`);
  }
  if (config.collector.wcfHost === "host.docker.internal") {
    throw new Error(`${configLabel}.personalWechatIngress.collector.wcfHost must be reachable from D601 k3s, not Docker host.docker.internal`);
  }
}

function personalWechatIngressSummary(config: WechatArchiveConfig["personalWechatIngress"]): Record<string, unknown> {
  return {
    enabled: config.enabled,
    mode: config.mode,
    target: config.target,
    pcWechat: {
      isolation: config.pcWechat.isolation,
      requiredVersion: config.pcWechat.requiredVersion,
      installerAsset: config.pcWechat.installerAsset,
      installMode: config.pcWechat.installMode,
      installRoot: config.pcWechat.installRoot,
      dataRoot: config.pcWechat.dataRoot,
      autoUpdatePolicy: config.pcWechat.autoUpdatePolicy,
    },
    wcfHost: {
      releaseTag: config.wcfHost.releaseTag,
      requiredWechatVersion: config.wcfHost.requiredWechatVersion,
      releaseUrl: config.wcfHost.releaseUrl,
      sdkRoot: config.wcfHost.sdkRoot,
      stateRoot: config.wcfHost.stateRoot,
      commandPort: config.wcfHost.commandPort,
      messagePort: config.wcfHost.messagePort,
      bindHost: config.wcfHost.bindHost,
      supervisor: config.wcfHost.supervisor,
      rdpPolicy: config.wcfHost.rdpPolicy,
      firewall: config.wcfHost.firewall,
    },
    collector: {
      runtime: config.collector.runtime,
      namespace: config.collector.namespace,
      createNamespace: config.collector.createNamespace,
      workload: config.collector.workload,
      storage: config.collector.storage,
      image: {
        reference: collectorImageRef(config),
        repository: config.collector.image.repository,
        tag: config.collector.image.tag,
        pullPolicy: config.collector.image.pullPolicy,
        context: config.collector.image.context,
        dockerfile: config.collector.image.dockerfile,
        wcferryVersion: config.collector.image.wcferryVersion,
        pip: {
          indexUrl: config.collector.image.pip.indexUrl,
          trustedHost: config.collector.image.pip.trustedHost,
          timeoutSeconds: config.collector.image.pip.timeoutSeconds,
          retries: config.collector.image.pip.retries,
        },
      },
      wcfHost: config.collector.wcfHost,
      commandPort: config.collector.commandPort,
      messagePort: config.collector.messagePort,
      stateRoot: config.collector.stateRoot,
      queueMode: config.collector.queueMode,
      outboxMode: config.collector.outboxMode,
      archiveCallbackRef: config.collector.archiveCallbackRef,
      secretName: config.collector.secretName,
      readOnly: {
        enabled: config.collector.readOnly.enabled,
        sendCapability: config.collector.readOnly.sendCapability,
        allowedMethods: config.collector.readOnly.allowedMethods,
        forbiddenMethods: config.collector.readOnly.forbiddenMethods,
      },
    },
    poc: config.poc,
    valuesPrinted: false,
  };
}

function stringArrayField(record: Record<string, unknown>, key: string, label: string): string[] {
  const value = record[key];
  if (!Array.isArray(value) || value.some((item) => typeof item !== "string" || item.trim().length === 0)) {
    throw new Error(`${label}.${key} must be a string array`);
  }
  return value.map((item) => String(item));
}

function parseReadOnlyRecord(raw: Record<string, unknown>): WechatArchiveConfig["langbot"]["readOnlyRecord"] {
  return {
    enabled: booleanField(raw, "enabled", `${configLabel}.langbot.readOnlyRecord`),
    mode: stringField(raw, "mode", `${configLabel}.langbot.readOnlyRecord`),
    webhookName: stringField(raw, "webhookName", `${configLabel}.langbot.readOnlyRecord`),
    webhookDescription: stringField(raw, "webhookDescription", `${configLabel}.langbot.readOnlyRecord`),
    discardPipelineUuid: stringField(raw, "discardPipelineUuid", `${configLabel}.langbot.readOnlyRecord`),
  };
}

function parseLangBotPipeline(raw: Record<string, unknown>): WechatArchiveConfig["langbot"]["pipeline"] {
  const timeoutSeconds = numberField(raw, "timeoutSeconds", `${configLabel}.langbot.pipeline`);
  return {
    name: stringField(raw, "name", `${configLabel}.langbot.pipeline`),
    description: stringField(raw, "description", `${configLabel}.langbot.pipeline`),
    runner: stringField(raw, "runner", `${configLabel}.langbot.pipeline`),
    outputKey: stringField(raw, "outputKey", `${configLabel}.langbot.pipeline`),
    timeoutSeconds,
  };
}

function webhookUrl(config: WechatArchiveConfig): string {
  return `${config.n8n.publicBaseUrl.replace(/\/+$/u, "")}/webhook/${registeredWebhookPath(config)}`;
}

function langBotDryRun(config: WechatArchiveConfig): Record<string, unknown> {
  return {
    ok: true,
    mode: "dry-run",
    pipeline: {
      name: config.langbot.pipeline.name,
      runner: config.langbot.pipeline.runner,
      webhookUrl: webhookUrl(config),
      outputKey: config.langbot.pipeline.outputKey,
      timeoutSeconds: config.langbot.pipeline.timeoutSeconds,
    },
    bot: {
      expectedAdapter: config.langbot.expectedAdapter,
      desiredRouting: "discard person/group messages after read-only webhook mirroring",
    },
    valuesPrinted: false,
  };
}

async function syncLangBotPipeline(config: WechatArchiveConfig): Promise<Record<string, unknown>> {
  const runtime = readLangBotRuntimeConfig();
  const secret = readLangBotSecretMaterial();
  const apiKey = String(secret.apiKey || "");
  const baseUrl = String(runtime.publicBaseUrl || config.langbot.publicBaseUrl);
  const pipelinesResp = await langBotRequest(baseUrl, apiKey, "GET", "/api/v1/pipelines");
  const pipelines = arrayFromPath(pipelinesResp.body, ["data", "pipelines"]);
  let pipeline = pipelines.find((item) => String(asRecord(item, "pipeline").name || "") === config.langbot.pipeline.name);
  let created = false;
  if (pipeline === undefined) {
    const createdResp = await langBotRequest(baseUrl, apiKey, "POST", "/api/v1/pipelines", {
      name: config.langbot.pipeline.name,
      description: config.langbot.pipeline.description,
      emoji: "archive",
    });
    const uuid = String(nestedValue(createdResp.body, ["data", "uuid"]) || "");
    if (!uuid) throw new Error("LangBot create pipeline did not return uuid");
    const getResp = await langBotRequest(baseUrl, apiKey, "GET", `/api/v1/pipelines/${encodeURIComponent(uuid)}`);
    pipeline = asRecord(nestedValue(getResp.body, ["data", "pipeline"]), "created pipeline");
    created = true;
  }
  const pipelineRecord = asRecord(pipeline, "pipeline");
  const pipelineUuid = String(pipelineRecord.uuid || "");
  if (!pipelineUuid) throw new Error("LangBot pipeline record is missing uuid");
  const configRecord = asRecord(pipelineRecord.config, "pipeline.config");
  const nextConfig = langBotPipelineConfig(configRecord, config);
  await langBotRequest(baseUrl, apiKey, "PUT", `/api/v1/pipelines/${encodeURIComponent(pipelineUuid)}`, {
    name: config.langbot.pipeline.name,
    description: config.langbot.pipeline.description,
    config: nextConfig,
  });
  const webhookBinding = syncLangBotWebhook(config, webhookUrl(config));
  const botBinding = await syncLangBotReadOnlyBot(config, baseUrl, apiKey);
  return {
    ok: webhookBinding.ok === true && botBinding.ok === true,
    mode: "confirmed",
    pipeline: {
      uuid: pipelineUuid,
      name: config.langbot.pipeline.name,
      runner: config.langbot.pipeline.runner,
      webhookUrl: webhookUrl(config),
      created,
    },
    readOnlyWebhook: webhookBinding,
    bot: botBinding,
    auth: {
      sourceRef: secret.sourceRef,
      keyName: secret.keyName,
      apiKeyFingerprint: secret.apiKeyFingerprint,
      valuesPrinted: false,
    },
    valuesPrinted: false,
  };
}

async function inspectLangBotBinding(config: WechatArchiveConfig): Promise<Record<string, unknown>> {
  try {
    const runtime = readLangBotRuntimeConfig();
    const secret = readLangBotSecretMaterial();
    const apiKey = String(secret.apiKey || "");
    const baseUrl = String(runtime.publicBaseUrl || config.langbot.publicBaseUrl);
    const pipelinesResp = await langBotRequest(baseUrl, apiKey, "GET", "/api/v1/pipelines");
    const pipelines = arrayFromPath(pipelinesResp.body, ["data", "pipelines"]);
    const pipeline = pipelines.find((item) => String(asRecord(item, "pipeline").name || "") === config.langbot.pipeline.name);
    const pipelineRecord = pipeline === undefined ? null : asRecord(pipeline, "pipeline");
    const pipelineUuid = String(pipelineRecord?.uuid || "");
    const pipelineConfig = typeof pipelineRecord?.config === "object" && pipelineRecord.config !== null && !Array.isArray(pipelineRecord.config)
      ? pipelineRecord.config as Record<string, unknown>
      : {};
    const runner = String(nestedValue(pipelineConfig, ["ai", "runner", "runner"]) || "");
    const webhook = String(nestedValue(pipelineConfig, ["ai", "n8n-service-api", "webhook-url"]) || "");
    const botsResp = await langBotRequest(baseUrl, apiKey, "GET", "/api/v1/platform/bots");
    const bots = arrayFromPath(botsResp.body, ["data", "bots"]);
    const bot = bots.find((item) => String(asRecord(item, "bot").adapter || "") === config.langbot.expectedAdapter);
    const botRecord = bot === undefined ? null : asRecord(bot, "bot");
    const routingRules = Array.isArray(botRecord?.pipeline_routing_rules) ? botRecord.pipeline_routing_rules as unknown[] : [];
    const hasDiscardRouting = hasReadOnlyDiscardRouting(routingRules, config);
    const readOnlyWebhook = inspectLangBotWebhook(config, webhookUrl(config));
    const ok = Boolean(pipelineUuid)
      && runner === config.langbot.pipeline.runner
      && webhook === webhookUrl(config)
      && hasDiscardRouting
      && readOnlyWebhook.ok === true;
    return {
      ok,
      pipeline: {
        exists: Boolean(pipelineUuid),
        uuid: pipelineUuid || null,
        name: config.langbot.pipeline.name,
        runner,
        webhookMatches: webhook === webhookUrl(config),
      },
      bot: {
        exists: botRecord !== null,
        uuid: String(botRecord?.uuid || "") || null,
        adapter: config.langbot.expectedAdapter,
        usePipelineUuid: String(botRecord?.use_pipeline_uuid || "") || null,
        readOnlyDiscardRouting: hasDiscardRouting,
      },
      readOnlyWebhook,
      auth: {
        sourceRef: secret.sourceRef,
        keyName: secret.keyName,
        apiKeyFingerprint: secret.apiKeyFingerprint,
        valuesPrinted: false,
      },
      valuesPrinted: false,
    };
  } catch (error) {
    return {
      ok: false,
      error: error instanceof Error ? error.message : String(error),
      valuesPrinted: false,
    };
  }
}

function langBotPipelineConfig(current: Record<string, unknown>, archive: WechatArchiveConfig): Record<string, unknown> {
  const next = JSON.parse(JSON.stringify(current)) as Record<string, unknown>;
  const ai = asMutableRecord(next, "ai");
  const runner = asMutableRecord(ai, "runner");
  runner.runner = archive.langbot.pipeline.runner;
  runner["expire-time"] = 0;
  const n8n = asMutableRecord(ai, "n8n-service-api");
  n8n["webhook-url"] = webhookUrl(archive);
  n8n["auth-type"] = "none";
  n8n.timeout = archive.langbot.pipeline.timeoutSeconds;
  n8n["output-key"] = archive.langbot.pipeline.outputKey;
  return next;
}

async function syncLangBotReadOnlyBot(config: WechatArchiveConfig, baseUrl: string, apiKey: string): Promise<Record<string, unknown>> {
  const botsResp = await langBotRequest(baseUrl, apiKey, "GET", "/api/v1/platform/bots");
  const bots = arrayFromPath(botsResp.body, ["data", "bots"]);
  const bot = bots.find((item) => String(asRecord(item, "bot").adapter || "") === config.langbot.expectedAdapter);
  if (bot === undefined) throw new Error(`LangBot bot adapter ${config.langbot.expectedAdapter} not found`);
  const botRecord = asRecord(bot, "bot");
  const botUuid = String(botRecord.uuid || "");
  if (!botUuid) throw new Error("LangBot bot record is missing uuid");
  const currentRules = Array.isArray(botRecord.pipeline_routing_rules) ? botRecord.pipeline_routing_rules : [];
  const nextRules = readOnlyDiscardRoutingRules(config);
  const alreadyDiscard = hasReadOnlyDiscardRouting(currentRules, config);
  if (!alreadyDiscard || JSON.stringify(currentRules) !== JSON.stringify(nextRules)) {
    await langBotRequest(baseUrl, apiKey, "PUT", `/api/v1/platform/bots/${encodeURIComponent(botUuid)}`, { pipeline_routing_rules: nextRules });
  }
  return {
    ok: true,
    uuid: botUuid,
    adapter: config.langbot.expectedAdapter,
    mode: config.langbot.readOnlyRecord.mode,
    usePipelineUuid: String(botRecord.use_pipeline_uuid || "") || null,
    routingRules: nextRules,
    readOnlyDiscardRouting: true,
    hadDiscardRouting: alreadyDiscard,
    valuesPrinted: false,
  };
}

function readOnlyDiscardRoutingRules(config: WechatArchiveConfig): Array<Record<string, string>> {
  const pipelineUuid = config.langbot.readOnlyRecord.discardPipelineUuid;
  return [
    { type: "launcher_type", operator: "eq", value: "person", pipeline_uuid: pipelineUuid },
    { type: "launcher_type", operator: "eq", value: "group", pipeline_uuid: pipelineUuid },
  ];
}

function hasReadOnlyDiscardRouting(rules: unknown[], config: WechatArchiveConfig): boolean {
  const desired = readOnlyDiscardRoutingRules(config);
  return desired.every((expected) => rules.some((rule) => {
    if (typeof rule !== "object" || rule === null || Array.isArray(rule)) return false;
    const record = rule as Record<string, unknown>;
    return Object.entries(expected).every(([key, value]) => record[key] === value);
  }));
}

function syncLangBotWebhook(config: WechatArchiveConfig, url: string): Record<string, unknown> {
  const secret = prepareLangBotSecretMaterial();
  const conn = readLangBotPostgresConninfo();
  const sql = [
    "WITH updated AS (",
    "  UPDATE webhooks",
    `  SET url = ${sqlLiteral(url)}, description = ${sqlLiteral(config.langbot.readOnlyRecord.webhookDescription)}, enabled = true, updated_at = now()`,
    `  WHERE name = ${sqlLiteral(config.langbot.readOnlyRecord.webhookName)}`,
    "  RETURNING id, name, enabled, url",
    "), inserted AS (",
    "  INSERT INTO webhooks (name, url, description, enabled)",
    `  SELECT ${sqlLiteral(config.langbot.readOnlyRecord.webhookName)}, ${sqlLiteral(url)}, ${sqlLiteral(config.langbot.readOnlyRecord.webhookDescription)}, true`,
    "  WHERE NOT EXISTS (SELECT 1 FROM updated)",
    "  RETURNING id, name, enabled, url",
    ")",
    "SELECT id, name, enabled, url FROM updated UNION ALL SELECT id, name, enabled, url FROM inserted LIMIT 1;",
  ].join("\n");
  const result = Bun.spawnSync(["psql", "-Atq", conn, "-c", sql], {
    stdout: "pipe",
    stderr: "pipe",
    env: { ...process.env, PGPASSWORD: secret.values.dbPassword },
  });
  const stdout = new TextDecoder().decode(result.stdout).trim();
  const stderr = new TextDecoder().decode(result.stderr).trim();
  return {
    ok: result.exitCode === 0,
    mode: "db-upsert",
    name: config.langbot.readOnlyRecord.webhookName,
    url,
    enabled: true,
    row: result.exitCode === 0 ? compactWebhookRow(stdout) : "",
    stderrTail: redactSensitiveUnknown(stderr).toString().slice(-1200),
    auth: {
      dbSourceRef: secret.dbSourceRef,
      appSourceRef: secret.appSourceRef,
      fingerprint: secret.fingerprint,
      valuesPrinted: false,
    },
    valuesPrinted: false,
  };
}

function inspectLangBotWebhook(config: WechatArchiveConfig, url: string): Record<string, unknown> {
  const secret = prepareLangBotSecretMaterial();
  const conn = readLangBotPostgresConninfo();
  const sql = [
    "SELECT id, name, enabled, url FROM webhooks",
    `WHERE url = ${sqlLiteral(url)} OR name = ${sqlLiteral(config.langbot.readOnlyRecord.webhookName)}`,
    "ORDER BY enabled DESC, updated_at DESC LIMIT 1;",
  ].join("\n");
  const result = Bun.spawnSync(["psql", "-Atq", conn, "-c", sql], {
    stdout: "pipe",
    stderr: "pipe",
    env: { ...process.env, PGPASSWORD: secret.values.dbPassword },
  });
  const stdout = new TextDecoder().decode(result.stdout).trim();
  const stderr = new TextDecoder().decode(result.stderr).trim();
  const row = parseWebhookRow(stdout);
  return {
    ok: result.exitCode === 0 && row !== null && row.enabled === true && row.url === url,
    exists: row !== null,
    name: row?.name ?? config.langbot.readOnlyRecord.webhookName,
    enabled: row?.enabled ?? false,
    urlMatches: row?.url === url,
    row: row === null ? null : { id: row.id, name: row.name, enabled: row.enabled },
    stderrTail: result.exitCode === 0 ? "" : redactSensitiveUnknown(stderr).toString().slice(-1200),
    valuesPrinted: false,
  };
}

function parseWebhookRow(value: string): { id: string; name: string; enabled: boolean; url: string } | null {
  if (!value) return null;
  const [id = "", name = "", enabled = "", url = ""] = value.split("|");
  if (!id) return null;
  return { id, name, enabled: enabled === "t" || enabled === "true", url };
}

function compactWebhookRow(value: string): Record<string, unknown> | null {
  const row = parseWebhookRow(value);
  if (row === null) return null;
  return { id: row.id, name: row.name, enabled: row.enabled };
}

function sqlLiteral(value: string): string {
  return `'${value.replace(/'/gu, "''")}'`;
}

async function langBotRequest(baseUrl: string, apiKey: string, method: string, apiPath: string, body?: unknown): Promise<Record<string, unknown>> {
  const controller = new AbortController();
  const timer = setTimeout(() => controller.abort(), 45_000);
  try {
    const response = await fetch(`${baseUrl.replace(/\/+$/u, "")}${apiPath}`, {
      method,
      headers: { "content-type": "application/json", "x-api-key": apiKey },
      body: body === undefined ? undefined : JSON.stringify(body),
      signal: controller.signal,
    });
    const text = await response.text();
    let parsed: unknown = text;
    try {
      parsed = text.length > 0 ? JSON.parse(text) as unknown : null;
    } catch {
      parsed = text;
    }
    if (!response.ok) throw new Error(`LangBot ${method} ${apiPath} failed with HTTP ${response.status}: ${redactText(JSON.stringify(compactUnknown(parsed))).slice(0, 800)}`);
    return { ok: true, status: response.status, body: parsed };
  } finally {
    clearTimeout(timer);
  }
}

function redactText(text: string): string {
  return text
    .replace(/lbk_[A-Za-z0-9_-]+/gu, "lbk_<redacted>")
    .replace(/(postgres(?:ql)?:\/\/)[^@\s"']+@/giu, "$1<redacted>@")
    .replace(/(Bearer\s+)[A-Za-z0-9._~+/=-]+/giu, "$1<redacted>")
    .replace(/(["']?(?:token|password|secret|api[_-]?key|apikey|jwt[_-]?secret|database[_-]?url)["']?\s*[:=]\s*["']?)[^"',\s}]+(["']?)/giu, "$1<redacted>$2");
}

function nestedValue(value: unknown, path: string[]): unknown {
  let current = value;
  for (const part of path) {
    if (typeof current !== "object" || current === null || Array.isArray(current)) return undefined;
    current = (current as Record<string, unknown>)[part];
  }
  return current;
}

function arrayFromPath(value: unknown, path: string[]): unknown[] {
  const nested = nestedValue(value, path);
  return Array.isArray(nested) ? nested : [];
}

function asMutableRecord(parent: Record<string, unknown>, key: string): Record<string, unknown> {
  const current = parent[key];
  if (typeof current === "object" && current !== null && !Array.isArray(current)) return current as Record<string, unknown>;
  const next: Record<string, unknown> = {};
  parent[key] = next;
  return next;
}

function renderN8nWorkflow(config: WechatArchiveConfig, callbackToken: string | null): Record<string, unknown> {
  const tokenForWorkflow = callbackToken ?? "__UNIDESK_WECHAT_ARCHIVE_TOKEN_FROM_SOURCE_REF__";
  const code = `
const input = $input.first()?.json ?? {};
const body = input.body && typeof input.body === 'object' ? input.body : input;
const data = body.data && typeof body.data === 'object' ? body.data : {};
const langbotMessage = data.message && typeof data.message === 'object' ? data.message : {};
const langbotComponents = Array.isArray(langbotMessage.root)
  ? langbotMessage.root
  : Array.isArray(langbotMessage)
    ? langbotMessage
    : [];
function firstComponent(type) {
  return langbotComponents.find((item) => item && typeof item === 'object' && String(item.type || '').toLowerCase() === type.toLowerCase()) || {};
}
function firstText() {
  const plain = firstComponent('Plain');
  if (plain.text !== undefined && plain.text !== null) return String(plain.text);
  return langbotComponents
    .filter((item) => item && typeof item === 'object' && item.text !== undefined)
    .map((item) => String(item.text))
    .join('');
}
function firstImage() {
  return firstComponent('Image');
}
function pick(...keys) {
  for (const key of keys) {
    const value = key.split('.').reduce((acc, part) => acc && typeof acc === 'object' ? acc[part] : undefined, body);
    if (value !== undefined && value !== null && String(value).length > 0) return value;
  }
  return '';
}
function sanitizePathSegment(value, fallback = 'unknown') {
  const normalized = String(value || fallback)
    .normalize('NFKC')
    .replace(/[\\\\/\\0\\r\\n\\t]+/g, '-')
    .replace(/[^A-Za-z0-9._-]+/g, '-')
    .replace(/^-+|-+$/g, '')
    .slice(0, 120);
  return normalized.length > 0 ? normalized : fallback;
}
function dateInTimeZone(date, timeZone) {
  const parts = new Intl.DateTimeFormat('en-CA', {
    timeZone,
    year: 'numeric',
    month: '2-digit',
    day: '2-digit',
  }).formatToParts(date).reduce((acc, part) => {
    acc[part.type] = part.value;
    return acc;
  }, {});
  return [parts.year, parts.month, parts.day].join('-');
}
function renderTemplate(template, values) {
  return template.replace(/\\{\\{\\s*([A-Za-z0-9_]+)\\s*\\}\\}/g, (_match, key) => values[key] || '');
}
function normalizeRemotePath(path) {
  const collapsed = String(path).replace(/\\/+/g, '/');
  return collapsed.startsWith('/') ? collapsed : '/' + collapsed;
}
const imageComponent = firstImage();
const langbotText = firstText();
const langbotSender = data.sender && typeof data.sender === 'object' ? data.sender : {};
const langbotGroup = data.group && typeof data.group === 'object' ? data.group : {};
const messageType = String(pick('messageType', 'type', 'msgType') || (Object.keys(imageComponent).length > 0 ? 'image' : 'text')).toLowerCase();
const receivedAt = String(pick('receivedAt', 'timestamp', 'data.timestamp') || new Date().toISOString());
const receivedDate = Number.isNaN(new Date(receivedAt).getTime()) ? new Date() : new Date(receivedAt);
const messageId = sanitizePathSegment(pick('messageId', 'msgId', 'id', 'message_id', 'uuid') || String(Date.now()), 'message');
const conversationId = sanitizePathSegment(pick('conversationId', 'chatId', 'roomId', 'fromUser', 'conversation_id', 'session_id') || langbotGroup.id || langbotSender.id || 'unknown-conversation', 'conversation');
const senderId = sanitizePathSegment(pick('senderId', 'userId', 'fromUserName', 'fromUser', 'user_id') || langbotSender.id || 'unknown-sender', 'sender');
const media = body.media && typeof body.media === 'object' ? body.media : {};
if (imageComponent.base64 && !media.dataBase64) media.dataBase64 = imageComponent.base64;
if (imageComponent.url && !media.url) media.url = imageComponent.url;
if (!media.filename && messageType === 'image') media.filename = messageId + '.${config.baiduNetdisk.archive.imageExtension}';
const extension = messageType === 'image' ? '${config.baiduNetdisk.archive.imageExtension}' : '${config.baiduNetdisk.archive.textExtension}';
const filename = messageType === 'image'
  ? sanitizePathSegment(media.filename || messageId + '.' + extension, messageId + '.' + extension)
  : messageId + '.' + extension;
const text = String(pick('text', 'content', 'message', 'chatInput', 'query', 'query_text') || langbotText || '');
const remotePath = normalizeRemotePath(renderTemplate('${config.baiduNetdisk.archive.pathTemplate}', {
  remoteRoot: '${config.baiduNetdisk.archive.remoteRoot}',
  date: dateInTimeZone(receivedDate, '${config.baiduNetdisk.archive.timezone}'),
  timestamp: receivedDate.toISOString().replace(/[-:.TZ]/g, ''),
  conversationId,
  senderId,
  messageId,
  type: messageType,
  extension,
  filename,
}));
const normalized = {
  ok: true,
  source: 'n8n',
  workflow: '${config.n8n.workflow.id}',
  message: {
    messageId,
    conversationId,
    senderId,
    messageType,
    text,
    receivedAt,
    media,
  },
  archive: {
    remoteRoot: '${config.baiduNetdisk.archive.remoteRoot}',
    pathTemplate: '${config.baiduNetdisk.archive.pathTemplate}',
    remotePath,
    filename,
  },
};
return [{ json: normalized }];
`.trim();
  return {
    id: config.n8n.workflow.id,
    name: config.n8n.workflow.name,
    active: config.n8n.workflow.active,
    nodes: [
      {
        parameters: {
          httpMethod: "POST",
          path: config.n8n.workflow.webhookPath,
          responseMode: "lastNode",
          options: {},
        },
        id: "webhook",
        name: "wechat-archive-webhook",
        type: "n8n-nodes-base.webhook",
        typeVersion: 2,
        position: [0, 0],
      },
      {
        parameters: { jsCode: code },
        id: "normalize",
        name: "Normalize WeChat Message",
        type: "n8n-nodes-base.code",
        typeVersion: 2,
        position: [280, 0],
      },
      {
        parameters: {
          method: "POST",
          url: config.archiveCallback.publicUrl,
          sendHeaders: true,
          headerParameters: {
            parameters: [
              { name: "Authorization", value: `Bearer ${tokenForWorkflow}` },
              { name: "Content-Type", value: "application/json" },
            ],
          },
          sendBody: true,
          specifyBody: "json",
          jsonBody: "={{ JSON.stringify($json) }}",
          options: {
            timeout: config.archiveCallback.timeoutMs,
          },
        },
        id: "archive-callback",
        name: "Archive To Baidu Callback",
        type: "n8n-nodes-base.httpRequest",
        typeVersion: 4.2,
        position: [560, 0],
      },
    ],
    connections: {
      "wechat-archive-webhook": {
        main: [[{ node: "Normalize WeChat Message", type: "main", index: 0 }]],
      },
      "Normalize WeChat Message": {
        main: [[{ node: "Archive To Baidu Callback", type: "main", index: 0 }]],
      },
    },
    settings: {
      executionOrder: "v1",
      timezone: config.n8n.workflow.timezone,
    },
    staticData: null,
    tags: [],
  };
}

function registeredWebhookPath(config: WechatArchiveConfig): string {
  return [
    config.n8n.workflow.id,
    "wechat-archive-webhook",
    config.n8n.workflow.webhookPath.replace(/^\/+|\/+$/gu, ""),
  ].map((segment) => encodeURIComponent(segment)).join("/");
}

function fixturePayload(config: WechatArchiveConfig, type: "text" | "image", observedAt: Date): Record<string, unknown> {
  const suffix = randomUUID().slice(0, 8);
  if (type === "text") {
    const fixture = config.validate.fixtures.text;
    return {
      source: "unidesk-cli-validate",
      messageType: "text",
      messageId: `validate-text-${suffix}`,
      conversationId: fixture.conversationId,
      senderId: fixture.senderId,
      text: `${fixture.text} ${observedAt.toISOString()}`,
      receivedAt: observedAt.toISOString(),
    };
  }
  const fixture = config.validate.fixtures.image;
  return {
    source: "unidesk-cli-validate",
    messageType: "image",
    messageId: `validate-image-${suffix}`,
    conversationId: fixture.conversationId,
    senderId: fixture.senderId,
    receivedAt: observedAt.toISOString(),
    media: {
      filename: fixture.filename,
      mimeType: fixture.mimeType,
      dataBase64: fixture.dataBase64,
    },
  };
}

async function callWorkflow(config: WechatArchiveConfig, payload: Record<string, unknown>): Promise<Record<string, unknown>> {
  const response = await fetchJsonWithTimeout(webhookUrl(config), payload, Math.max(45_000, config.archiveCallback.timeoutMs + 30_000));
  const body = typeof response.body === "object" && response.body !== null && !Array.isArray(response.body)
    ? response.body as Record<string, unknown>
    : {};
  return {
    ok: response.ok && (body.ok === true || response.status >= 200 && response.status < 300),
    status: response.status,
    webhookUrl: webhookUrl(config),
    body,
  };
}

function archiveFromWorkflowResponse(originalPayload: Record<string, unknown>, n8nResponse: Record<string, unknown>): Record<string, unknown> {
  if (n8nResponse.ok !== true) {
    return {
      ok: false,
      error: "n8n-webhook-failed",
      n8n: {
        status: n8nResponse.status,
        body: compactUnknown(n8nResponse.body),
      },
    };
  }
  const body = asRecord(n8nResponse.body ?? {}, "n8n response body");
  const archive = asRecord(body.archive ?? {}, "n8n response archive");
  const normalized = asRecord(body.message ?? originalPayload, "normalized message");
  const type = String(normalized.messageType ?? originalPayload.messageType ?? "text").toLowerCase();
  const remotePath = String(archive.remotePath || "");
  const fsId = String(archive.fsId || archive.fs_id || "");
  return {
    ok: body.ok === true && Boolean(remotePath) && Boolean(fsId),
    skipPipeline: body.skip_pipeline === true,
    readOnly: body.readOnly === true,
    type,
    remotePath,
    fsId,
    messageId: normalized.messageId ?? originalPayload.messageId,
    conversationId: normalized.conversationId ?? originalPayload.conversationId,
    senderId: normalized.senderId ?? originalPayload.senderId,
    local: typeof archive.local === "object" && archive.local !== null && !Array.isArray(archive.local) ? archive.local : {},
    uploadJob: typeof archive.uploadJob === "object" && archive.uploadJob !== null && !Array.isArray(archive.uploadJob) ? archive.uploadJob : {},
    listed: typeof archive.listed === "object" && archive.listed !== null && !Array.isArray(archive.listed) ? redactBaiduFileEntry(archive.listed as Record<string, unknown>) : null,
  };
}

async function pullByFsId(config: WechatArchiveConfig, fsId: string, localRel: string): Promise<Record<string, unknown>> {
  if (!fsId) return { ok: false, error: "fsId is required" };
  const created = assertProxyOk(microserviceProxy(config.baiduNetdisk.serviceId, "/api/transfers/download-to-path", {
    method: "POST",
    body: { fsId, localPath: localRel },
    timeoutMs: 60_000,
  }), "baidu download create");
  const job = asRecord(created.job, "baidu download job");
  const jobId = String(job.id || "");
  const waited = await waitForBaiduTransfer(config.baiduNetdisk.serviceId, jobId, {
    timeoutMs: config.validate.timeoutMs,
    pollIntervalMs: config.validate.pollIntervalMs,
  });
  const waitedJob = asRecord(waited.job ?? {}, "baidu waited download job");
  const containerPath = String(waitedJob.localPath || "");
  const hostPath = containerPathToHostPath(config.baiduNetdisk.staging.containerRoot, config.baiduNetdisk.staging.hostRoot, containerPath)
    ?? hostRootPath(pathPosix.join(config.baiduNetdisk.staging.hostRoot, localRel));
  return {
    ok: waited.ok === true && existsSync(hostPath),
    fsId,
    remotePath: waitedJob.remotePath,
    localPath: containerPath,
    hostPath,
    sha256: existsSync(hostPath) ? sha256File(hostPath) : null,
    downloadJob: waitedJob,
  };
}

async function pullArchiveIfReady(config: WechatArchiveConfig, archive: Record<string, unknown>): Promise<Record<string, unknown>> {
  const fsId = String(archive.fsId || "");
  const remotePath = String(archive.remotePath || "");
  if (archive.ok !== true || !fsId || !remotePath) {
    return {
      ok: false,
      skipped: true,
      reason: "archive-upload-not-ready",
      fsId,
      remotePath,
    };
  }
  return await pullByFsId(config, fsId, pullLocalRel(config, remotePath));
}

function pullLocalRel(config: WechatArchiveConfig, remotePath: unknown): string {
  const name = sanitizePathSegment(pathPosix.basename(String(remotePath || `pull-${randomUUID()}`)), `pull-${randomUUID()}`);
  return pathPosix.join(config.baiduNetdisk.staging.pullDir, name);
}

function validationSummary(
  payload: Record<string, unknown>,
  n8n: Record<string, unknown>,
  archive: Record<string, unknown>,
  downloaded: Record<string, unknown>,
  full: boolean,
): Record<string, unknown> {
  return {
    ok: n8n.ok === true && archive.ok === true && downloaded.ok === true,
    messageId: payload.messageId,
    n8n: full ? n8n : { ok: n8n.ok, status: n8n.status, webhookUrl: n8n.webhookUrl },
    archive: full ? archive : {
      ok: archive.ok,
      skipPipeline: archive.skipPipeline,
      remotePath: archive.remotePath,
      fsId: archive.fsId,
      local: archive.local,
    },
    pull: full ? downloaded : compactDownload(downloaded),
  };
}

function compactDownload(downloaded: Record<string, unknown>): Record<string, unknown> {
  return {
    ok: downloaded.ok,
    fsId: downloaded.fsId,
    remotePath: downloaded.remotePath,
    hostPath: downloaded.hostPath,
    sha256: downloaded.sha256,
  };
}

function compactBaiduHealth(response: ReturnType<typeof microserviceProxy>): Record<string, unknown> {
  const body = typeof response.body === "object" && response.body !== null && !Array.isArray(response.body)
    ? response.body as Record<string, unknown>
    : {};
  const storage = typeof body.storage === "object" && body.storage !== null && !Array.isArray(body.storage) ? body.storage as Record<string, unknown> : {};
  const queue = typeof body.queue === "object" && body.queue !== null && !Array.isArray(body.queue) ? body.queue as Record<string, unknown> : {};
  const transfers = typeof queue.transfers === "object" && queue.transfers !== null && !Array.isArray(queue.transfers) ? queue.transfers as Record<string, unknown> : {};
  return {
    ok: response.ok,
    status: response.status,
    service: body.service,
    storage: {
      postgres: storage.postgres,
      stagingDir: storage.stagingDir,
    },
    queue: {
      transfers: {
        succeeded: transfers.succeeded,
        failed: transfers.failed,
      },
    },
  };
}

function compactBaiduAuth(response: ReturnType<typeof microserviceProxy>): Record<string, unknown> {
  const body = typeof response.body === "object" && response.body !== null && !Array.isArray(response.body)
    ? response.body as Record<string, unknown>
    : {};
  const auth = typeof body.auth === "object" && body.auth !== null && !Array.isArray(body.auth) ? body.auth as Record<string, unknown> : {};
  const account = typeof auth.account === "object" && auth.account !== null && !Array.isArray(auth.account) ? auth.account as Record<string, unknown> : {};
  return {
    ok: response.ok,
    status: response.status,
    configured: auth.configured,
    loggedIn: auth.loggedIn,
    account: {
      present: Object.keys(account).length > 0,
      rootPath: account.rootPath,
      quota: compactBaiduQuota(account.quota),
    },
    valuesPrinted: false,
  };
}

function compactBaiduQuota(value: unknown): Record<string, unknown> | null {
  if (typeof value !== "object" || value === null || Array.isArray(value)) return null;
  const quota = value as Record<string, unknown>;
  return {
    total: quota.total,
    used: quota.used,
    free: quota.free,
    usedPercent: quota.usedPercent,
  };
}

function compactBaiduTransfers(response: ReturnType<typeof microserviceProxy>): Record<string, unknown> {
  const body = typeof response.body === "object" && response.body !== null && !Array.isArray(response.body)
    ? response.body as Record<string, unknown>
    : {};
  const counts = typeof body.counts === "object" && body.counts !== null && !Array.isArray(body.counts) ? body.counts as Record<string, unknown> : {};
  const jobs = typeof body.jobs === "object" && body.jobs !== null && !Array.isArray(body.jobs) ? body.jobs as Record<string, unknown> : {};
  const items = Array.isArray(jobs.items)
    ? jobs.items
    : Array.isArray(jobs.arrayPreview)
      ? jobs.arrayPreview
      : [];
  return {
    ok: response.ok,
    status: response.status,
    counts,
    jobs: {
      count: typeof jobs.count === "number" ? jobs.count : items.length,
      items: items.slice(0, 10).map(compactBaiduTransferJob),
    },
  };
}

function compactBaiduTransferJob(value: unknown): Record<string, unknown> {
  const job = typeof value === "object" && value !== null && !Array.isArray(value) ? value as Record<string, unknown> : {};
  const result = typeof job.result === "object" && job.result !== null && !Array.isArray(job.result) ? job.result as Record<string, unknown> : {};
  const baidu = typeof result.baidu === "object" && result.baidu !== null && !Array.isArray(result.baidu) ? result.baidu as Record<string, unknown> : {};
  return {
    id: job.id,
    direction: job.direction,
    status: job.status,
    createdAt: job.createdAt,
    updatedAt: job.updatedAt,
    remotePath: job.remotePath ?? result.remotePath,
    fsId: job.fsId || baidu.fs_id || result.fsId,
    sizeBytes: job.sizeBytes ?? result.sizeBytes ?? baidu.size,
    progressPercent: job.progressPercent,
    error: job.error || undefined,
  };
}

function redactBaiduFileEntry(value: Record<string, unknown> | null): Record<string, unknown> | null {
  if (value === null) return null;
  const copy: Record<string, unknown> = { ...value };
  if (copy.thumbs !== undefined) copy.thumbs = "<redacted-signed-thumbnail-urls>";
  if (copy.dlink !== undefined) copy.dlink = "<redacted-download-link>";
  return copy;
}

function readArchiveCallbackToken(config: WechatArchiveConfig): { sourceRef: string; keyName: string; value: string; fingerprint: string } {
  const sourcePath = pathPosix.join(config.archiveCallback.secretRoot, config.archiveCallback.tokenSourceRef);
  if (!existsSync(sourcePath)) {
    throw new Error(`${config.archiveCallback.tokenSourceRef} is missing; run bun scripts/cli.ts secrets sync --config config/secrets-distribution.yaml --scope platform-infra --confirm first`);
  }
  const values = parseEnvFile(readFileSync(sourcePath, "utf8"));
  const value = values[config.archiveCallback.tokenKey] ?? "";
  if (value.length === 0) throw new Error(`${config.archiveCallback.tokenSourceRef} is missing required key ${config.archiveCallback.tokenKey}`);
  return {
    sourceRef: config.archiveCallback.tokenSourceRef,
    keyName: config.archiveCallback.tokenKey,
    value,
    fingerprint: fingerprintValues({ [config.archiveCallback.tokenKey]: value }, [config.archiveCallback.tokenKey]),
  };
}