From 37f48d4655b20bdc324cf24457881a488b46bb9a Mon Sep 17 00:00:00 2001 From: Codex Date: Sat, 13 Jun 2026 06:22:04 +0000 Subject: [PATCH] feat: enforce read-only wechat archive ingestion --- config/platform-infra/wechat-archive.yaml | 6 + docs/reference/platform-infra.md | 1 + scripts/src/platform-infra-langbot.ts | 9 + scripts/src/platform-infra-wechat-archive.ts | 256 ++++++++++++++++--- src/components/frontend/src/index.ts | 15 +- 5 files changed, 250 insertions(+), 37 deletions(-) diff --git a/config/platform-infra/wechat-archive.yaml b/config/platform-infra/wechat-archive.yaml index da767ea0..9dffd921 100644 --- a/config/platform-infra/wechat-archive.yaml +++ b/config/platform-infra/wechat-archive.yaml @@ -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. diff --git a/docs/reference/platform-infra.md b/docs/reference/platform-infra.md index 96e962eb..48659f1d 100644 --- a/docs/reference/platform-infra.md +++ b/docs/reference/platform-infra.md @@ -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. diff --git a/scripts/src/platform-infra-langbot.ts b/scripts/src/platform-infra-langbot.ts index 294574ba..b6ca15b4 100644 --- a/scripts/src/platform-infra-langbot.ts +++ b/scripts/src/platform-infra-langbot.ts @@ -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); diff --git a/scripts/src/platform-infra-wechat-archive.ts b/scripts/src/platform-infra-wechat-archive.ts index 70428de6..2f5d5bc2 100644 --- a/scripts/src/platform-infra-wechat-archive.ts +++ b/scripts/src/platform-infra-wechat-archive.ts @@ -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 { 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): 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): WechatArchiveConfig["langbot"]["pipeline"] { const timeoutSeconds = numberField(raw, "timeoutSeconds", `${configLabel}.langbot.pipeline`); return { @@ -522,7 +551,7 @@ function langBotDryRun(config: WechatArchiveConfig): Record { }, 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 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 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, archive: Wechat return next; } +async function syncLangBotReadOnlyBot(config: WechatArchiveConfig, baseUrl: string, apiKey: string): Promise> { + 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> { + 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; + return Object.entries(expected).every(([key, value]) => record[key] === value); + })); +} + +function syncLangBotWebhook(config: WechatArchiveConfig, url: string): Record { + 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 { + 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 | 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> { 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_") + .replace(/(postgres(?:ql)?:\/\/)[^@\s"']+@/giu, "$1@") + .replace(/(Bearer\s+)[A-Za-z0-9._~+/=-]+/giu, "$1") + .replace(/(["']?(?:token|password|secret|api[_-]?key|apikey|jwt[_-]?secret|database[_-]?url)["']?\s*[:=]\s*["']?)[^"',\s}]+(["']?)/giu, "$1$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, 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, diff --git a/src/components/frontend/src/index.ts b/src/components/frontend/src/index.ts index f469670a..538a0654 100644 --- a/src/components/frontend/src/index.ts +++ b/src/components/frontend/src/index.ts @@ -825,16 +825,24 @@ function archiveContentPayload(body: Record): Record 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 { 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,