feat: enforce read-only wechat archive ingestion

2026-06-13 06:22:04 +00:00
parent edb040d31a
commit 37f48d4655
5 changed files with 250 additions and 37 deletions
@@ -17,6 +17,12 @@ langbot:
  publicBaseUrl: https://langbot.pikapython.com
  expectedAdapter: openclaw-weixin
  callbackPath: /callback/command
+  readOnlyRecord:
+    enabled: true
+    mode: webhook-skip-pipeline
+    webhookName: UniDesk WeChat Archive Read Only
+    webhookDescription: Archive LangBot/OpenClaw inbound messages and return skip_pipeline=true so personal WeChat never receives automated replies.
+    discardPipelineUuid: __discard__
  pipeline:
    name: UniDesk WeChat Baidu Archive
    description: Forward WeChat/OpenClaw messages to the UniDesk n8n archive workflow.
@@ -65,6 +65,7 @@
 - The archive callback token is controlled by `archiveCallback.secretRoot`, `archiveCallback.tokenSourceRef`, and `archiveCallback.tokenKey` in YAML plus `config/secrets-distribution.yaml`. `secrets sync` may create the local source when YAML explicitly allows it; n8n receives the token only through controlled workflow rendering. Do not recover this token from the n8n database, frontend runtime, Baidu runtime, pod env, or logs.
 - For the current n8n runtime, production webhook reachability uses the registered path shape `workflowId/nodeName/webhookPath`; workflow node names used in generated webhooks should be ASCII path-safe, and `webhookPath` in YAML should remain one relative path segment.
 - Generated n8n workflows should use n8n-native HTTP Request nodes for outbound service callbacks. Code nodes may normalize payloads, but must not assume sandbox globals such as `fetch` exist in the runtime.
+- Personal WeChat ingestion must be read-only. The durable shape is a YAML-declared LangBot inbound webhook that mirrors messages to the archive workflow and returns `skip_pipeline=true`; the OpenClaw/LangBot bot must also have discard routing as fallback so webhook failure does not produce an automated reply. Do not connect personal WeChat through a normal reply pipeline, do not enable send-message surfaces for this purpose, and do not treat a successful archive upload as permission to reply.
 - If LangBot or n8n public HTTPS fails while in-cluster service and FRP local-port probes are healthy, restore the PK01 Caddy managed blocks through `platform-infra langbot apply --confirm --wait` or `platform-infra n8n apply --confirm --wait`. Do not manually edit Caddy as the durable fix.
 - The archive uses the same single PK01/Pika01 PostgreSQL instance indirectly through the existing LangBot and n8n databases. Adding this workflow must not create another PostgreSQL instance, in-cluster PostgreSQL StatefulSet, or ad hoc database namespace.
 - `platform-infra-wechat-archive` and future similar public workflow CLIs should reuse the common platform-infra operations library for YAML parsing, target selection, workflow sync, private microservice proxy calls, transfer polling, staging path mapping, redaction and bounded output. Service-specific modules should keep only their business mapping and workflow payload rendering.
@@ -980,6 +980,15 @@ function hasAllowAllNetworkPolicy(yaml: string, namespaceName: string): boolean
    && /^\s*podSelector:\s*\{\}\s*$/mu.test(document));
 }

+export function prepareLangBotSecretMaterial(): SecretMaterial {
+  return prepareSecretMaterial(readLangBotConfig());
+}
+
+export function readLangBotPostgresConninfo(): string {
+  const langbot = readLangBotConfig();
+  return postgresConninfo(langbot, prepareSecretMaterial(langbot));
+}
+
 function prepareSecretMaterial(langbot: LangBotConfig): SecretMaterial {
  const root = secretRoot(langbot);
  const dbSourcePath = join(root, langbot.runtime.database.sourceRef);
@@ -32,7 +32,7 @@ import {
  type OpsApplyOptions,
  type OpsCommonOptions,
 } from "./platform-infra-ops-library";
-import { readLangBotRuntimeConfig, readLangBotSecretMaterial } from "./platform-infra-langbot";
+import { prepareLangBotSecretMaterial, readLangBotPostgresConninfo, readLangBotRuntimeConfig, readLangBotSecretMaterial } from "./platform-infra-langbot";
 import { fingerprintValues } from "./platform-infra-public-service";

 const configFile = rootPath("config", "platform-infra", "wechat-archive.yaml");
@@ -48,6 +48,7 @@ interface WechatArchiveConfig {
    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[];
  };
@@ -262,7 +263,15 @@ async function validate(options: OpsCommonOptions): Promise<Record<string, unkno
  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 && imageArchive.ok === true && textPull.ok === true && imagePull.ok === true;
+  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",
@@ -356,6 +365,7 @@ function readConfig(): WechatArchiveConfig {
      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) : [],
    },
@@ -429,6 +439,8 @@ function validateConfig(config: WechatArchiveConfig): void {
  }
  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("/")) {
@@ -453,6 +465,12 @@ function configSummary(config: WechatArchiveConfig): Record<string, unknown> {
      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: {
@@ -487,6 +505,7 @@ function policyChecks(config: WechatArchiveConfig): Array<Record<string, unknown
    { 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: "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." },
@@ -494,6 +513,16 @@ function policyChecks(config: WechatArchiveConfig): Array<Record<string, unknown
  ];
 }

+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 {
@@ -522,7 +551,7 @@ function langBotDryRun(config: WechatArchiveConfig): Record<string, unknown> {
    },
    bot: {
      expectedAdapter: config.langbot.expectedAdapter,
-      desiredUsePipelineName: config.langbot.pipeline.name,
+      desiredRouting: "discard person/group messages after read-only webhook mirroring",
    },
    valuesPrinted: false,
  };
@@ -559,19 +588,10 @@ async function syncLangBotPipeline(config: WechatArchiveConfig): Promise<Record<
    description: config.langbot.pipeline.description,
    config: nextConfig,
  });
-  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 alreadyBound = String(botRecord.use_pipeline_uuid || "") === pipelineUuid;
-  if (!alreadyBound) {
-    await langBotRequest(baseUrl, apiKey, "PUT", `/api/v1/platform/bots/${encodeURIComponent(botUuid)}`, { use_pipeline_uuid: pipelineUuid });
-  }
+  const webhookBinding = syncLangBotWebhook(config, webhookUrl(config));
+  const botBinding = await syncLangBotReadOnlyBot(config, baseUrl, apiKey);
  return {
-    ok: true,
+    ok: webhookBinding.ok === true && botBinding.ok === true,
    mode: "confirmed",
    pipeline: {
      uuid: pipelineUuid,
@@ -580,14 +600,8 @@ async function syncLangBotPipeline(config: WechatArchiveConfig): Promise<Record<
      webhookUrl: webhookUrl(config),
      created,
    },
-    bot: {
-      uuid: botUuid,
-      adapter: config.langbot.expectedAdapter,
-      bound: true,
-      wasAlreadyBound: alreadyBound,
-      usePipelineUuid: pipelineUuid,
-      usePipelineName: config.langbot.pipeline.name,
-    },
+    readOnlyWebhook: webhookBinding,
+    bot: botBinding,
    auth: {
      sourceRef: secret.sourceRef,
      keyName: secret.keyName,
@@ -618,11 +632,14 @@ async function inspectLangBotBinding(config: WechatArchiveConfig): Promise<Recor
    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 botPipelineUuid = String(botRecord?.use_pipeline_uuid || "");
+    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)
-      && botPipelineUuid === pipelineUuid;
+      && hasDiscardRouting
+      && readOnlyWebhook.ok === true;
    return {
      ok,
      pipeline: {
@@ -636,9 +653,10 @@ async function inspectLangBotBinding(config: WechatArchiveConfig): Promise<Recor
        exists: botRecord !== null,
        uuid: String(botRecord?.uuid || "") || null,
        adapter: config.langbot.expectedAdapter,
-        usePipelineUuid: botPipelineUuid || null,
-        bound: Boolean(pipelineUuid) && botPipelineUuid === pipelineUuid,
+        usePipelineUuid: String(botRecord?.use_pipeline_uuid || "") || null,
+        readOnlyDiscardRouting: hasDiscardRouting,
      },
+      readOnlyWebhook,
      auth: {
        sourceRef: secret.sourceRef,
        keyName: secret.keyName,
@@ -670,6 +688,137 @@ function langBotPipelineConfig(current: Record<string, unknown>, archive: Wechat
  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);
@@ -687,13 +836,21 @@ async function langBotRequest(baseUrl: string, apiKey: string, method: string, a
    } catch {
      parsed = text;
    }
-    if (!response.ok) throw new Error(`LangBot ${method} ${apiPath} failed with HTTP ${response.status}: ${JSON.stringify(compactUnknown(parsed)).slice(0, 800)}`);
+    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) {
@@ -721,6 +878,27 @@ function renderN8nWorkflow(config: WechatArchiveConfig, callbackToken: string |
  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);
@@ -756,18 +934,25 @@ function normalizeRemotePath(path) {
  const collapsed = String(path).replace(/\\/+/g, '/');
  return collapsed.startsWith('/') ? collapsed : '/' + collapsed;
 }
-const messageType = String(pick('messageType', 'type', 'msgType') || 'text').toLowerCase();
-const receivedAt = String(pick('receivedAt', 'timestamp') || new Date().toISOString());
+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') || String(Date.now()), 'message');
-const conversationId = sanitizePathSegment(pick('conversationId', 'chatId', 'roomId', 'fromUser', 'conversation_id', 'session_id') || 'unknown-conversation', 'conversation');
-const senderId = sanitizePathSegment(pick('senderId', 'userId', 'fromUserName', 'fromUser', 'user_id') || 'unknown-sender', 'sender');
+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') || '');
+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}'),
@@ -939,6 +1124,8 @@ function archiveFromWorkflowResponse(originalPayload: Record<string, unknown>, n
  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,
@@ -1012,6 +1199,7 @@ function validationSummary(
    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,
@@ -825,16 +825,24 @@ function archiveContentPayload(body: Record<string, unknown>): Record<string, un
  const message = recordFrom(body.message);
  const archive = recordFrom(body.archive);
  const media = recordFrom(message.media);
+  const langbotData = recordFrom(body.data);
+  const langbotMessage = recordFrom(langbotData.message);
+  const langbotComponents = Array.isArray(langbotMessage.root) ? langbotMessage.root : [];
+  const langbotPlain = langbotComponents
+    .filter((item) => recordFrom(item).type === "Plain")
+    .map((item) => stringFrom(recordFrom(item).text))
+    .join("");
+  const langbotImage = recordFrom(langbotComponents.find((item) => recordFrom(item).type === "Image"));
  const messageType = stringFrom(message.messageType || body.messageType || "text").toLowerCase();
  const remotePath = stringFrom(archive.remotePath);
  if (!isValidArchiveRemotePath(remotePath)) throw new Error("archive.remotePath is outside the WeChat archive root");
  const filename = stringFrom(archive.filename || remotePath.split("/").filter(Boolean).pop() || "wechat-archive.txt");
  if (messageType === "image") {
-    const dataBase64 = stringFrom(media.dataBase64 || archive.dataBase64 || body.dataBase64);
+    const dataBase64 = stringFrom(media.dataBase64 || archive.dataBase64 || body.dataBase64 || langbotImage.base64);
    if (!dataBase64) throw new Error("image archive payload is missing media.dataBase64");
    return { dataBase64, filename, remotePath, maxBytes: 10 * 1024 * 1024 };
  }
-  const text = stringFrom(message.text || body.text || body.message || body.content);
+  const text = stringFrom(message.text || body.text || body.message || body.content || langbotPlain);
  const payloadHash = new Bun.CryptoHasher("sha256").update(JSON.stringify(body)).digest("hex");
  const content = `${text}\n\n---\nsource=unidesk-wechat-archive\npayloadSha256=${payloadHash}\n`;
  return { content, filename, remotePath, maxBytes: 10 * 1024 * 1024 };
@@ -912,7 +920,8 @@ async function wechatArchiveWebhook(req: Request): Promise<Response> {
    logger(ok ? "info" : "warn", "wechat_archive_upload_finished", { ok, jobId, remotePath, fsIdPresent: fsId.length > 0 });
    return jsonResponse({
      ok,
-      response: ok ? `已归档到百度网盘：${remotePath}` : "归档失败",
+      skip_pipeline: ok,
+      readOnly: true,
      archive: {
        remotePath,
        fsId,