From 74ae56506d8d859912e2cf27dc3bf2f7acf32f91 Mon Sep 17 00:00:00 2001 From: l1shen <648952316@qq.com> Date: Fri, 31 Jul 2026 18:02:05 +0800 Subject: [PATCH 1/6] feat: add local Open Flow command delegation --- docs/commands.md | 38 ++++ docs/commands.zh-CN.md | 34 ++++ .../__snapshots__/run-cli.test.ts.snap | 6 + src/application/bootstrap/run-cli.ts | 54 +++++- src/application/commands/catalog.ts | 2 + .../__snapshots__/index.cli.test.ts.snap | 2 + src/application/commands/flow.cli.test.ts | 175 ++++++++++++++++++ src/application/commands/flow.ts | 131 +++++++++++++ .../commands/telemetry-decisions.test.ts | 4 + src/i18n/catalog.ts | 22 +++ 10 files changed, 460 insertions(+), 8 deletions(-) create mode 100644 src/application/commands/flow.cli.test.ts create mode 100644 src/application/commands/flow.ts diff --git a/docs/commands.md b/docs/commands.md index 06dd1ca..16652c9 100644 --- a/docs/commands.md +++ b/docs/commands.md @@ -23,6 +23,12 @@ use. Truthy values are `1`, `true`, `yes`, or `on` (case-insensitive). `XDG_CONFIG_HOME`. - `OO_DATA_DIR`: Override the data directory that holds the local cache, uploads, and download-session state. Defaults to `/data`. +- `OO_OPEN_FLOW_COMMAND_DIR`: During local Open Flow integration testing, point + `oo flow` at an expanded Open Flow command artifact directory. The directory + must contain `entry.js`; the standard repository build writes it to + `packages/open-flow/dist/command/open-flow-command`. This variable is + currently required because remote artifact distribution is not connected + yet. - `OO_LOG_DIR`: Override the debug-log directory. Takes precedence over every platform default. - `OO_API_KEY`: Run execution commands with this API key without an interactive @@ -80,6 +86,38 @@ use. Truthy values are `1`, `true`, `yes`, or `on` (case-insensitive). - `OO_NO_SELF_UPDATE`: A truthy value disables `oo update`, `oo install`, and `oo check-update` and forces self-update PATH modification off. +## Open Flow + +### `oo flow [args...]` + +Run the Open Flow CLI from the command artifact directory configured by +`OO_OPEN_FLOW_COMMAND_DIR`. + +- Every argument after `flow` is passed to Open Flow unchanged. The main `oo` + CLI does not parse, reorder, or log these arguments. +- `oo flow --help` and `oo flow --version` are therefore Open Flow commands. + Use `oo help flow` to show the host-side command description without loading + Open Flow. +- Main `oo` global options such as `--lang` and `--debug` must appear before + `flow`. Options after `flow` belong to Open Flow. +- Open Flow uses the current process's working directory, standard streams, + environment, and signals. Its exit code becomes the `oo` exit code. +- The loaded artifact must target the same Bun version embedded in `oo`. +- Local authoring, validation, development, and execution commands are + available. Open Flow Cloud commands remain unavailable until the OOMOL host + request transport is connected. + +Local repository example: + +```bash +cd /path/to/open-flow +bun run --filter @oomol-lab/open-flow build + +cd /path/to/oo-cli +OO_OPEN_FLOW_COMMAND_DIR=/path/to/open-flow/packages/open-flow/dist/command/open-flow-command \ + bun run index.ts flow --help +``` + ## JSON Output Commands that document `--format=json` and `--json` share the following diff --git a/docs/commands.zh-CN.md b/docs/commands.zh-CN.md index 84ad9f9..cc9dc82 100644 --- a/docs/commands.zh-CN.md +++ b/docs/commands.zh-CN.md @@ -21,6 +21,11 @@ CLI 读取以下环境变量以支持内置和自动化场景。真值为 `1`、 子目录）。优先级高于 `XDG_CONFIG_HOME`。 - `OO_DATA_DIR`：覆盖数据目录，其中包含本地缓存、上传和下载会话状态。默认值为 `<配置根目录>/data`。 +- `OO_OPEN_FLOW_COMMAND_DIR`：本地联调 Open Flow 时，让 `oo flow` + 使用指定的已展开命令产物目录。该目录必须包含 `entry.js`；Open Flow + 仓库的标准构建会将它写到 + `packages/open-flow/dist/command/open-flow-command`。远端产物分发尚未接入， + 因此当前必须设置此变量。 - `OO_LOG_DIR`：覆盖 debug 日志目录。优先级高于所有平台默认值。 - `OO_API_KEY`：使用该 API key 执行命令，无需交互式登录。设置后 CLI 会构造一个 内存账号，不读取、不要求、也不写入 `auth.toml`，且优先级高于任何已保存的账号。 @@ -65,6 +70,35 @@ CLI 读取以下环境变量以支持内置和自动化场景。真值为 `1`、 - `OO_NO_SELF_UPDATE`：设为真值会禁用 `oo update`、`oo install` 和 `oo check-update`，并强制关闭 self-update 的 PATH 改写。 +## Open Flow + +### `oo flow [args...]` + +从 `OO_OPEN_FLOW_COMMAND_DIR` 配置的命令产物目录运行 Open Flow CLI。 + +- `flow` 后的全部参数都会原样传给 Open Flow；主 `oo` CLI + 不解析、不重排，也不把这些参数写入日志。 +- 因此 `oo flow --help` 和 `oo flow --version` 都属于 Open Flow 命令。 + 如需在不加载 Open Flow 的情况下查看宿主侧命令说明，请使用 `oo help flow`。 +- `--lang`、`--debug` 等 `oo` 全局选项必须放在 `flow` 前面； + `flow` 后的选项归 Open Flow 所有。 +- Open Flow 使用当前进程的工作目录、标准输入输出、环境变量和信号； + 它的退出码会直接成为 `oo` 的退出码。 +- 被加载的产物必须与 `oo` 内嵌的 Bun 版本一致。 +- 当前可以使用本地 authoring、校验、开发和执行命令。OOMOL + 宿主请求传输接入前，Open Flow Cloud 子命令暂不可用。 + +本地仓库联调示例： + +```bash +cd /path/to/open-flow +bun run --filter @oomol-lab/open-flow build + +cd /path/to/oo-cli +OO_OPEN_FLOW_COMMAND_DIR=/path/to/open-flow/packages/open-flow/dist/command/open-flow-command \ + bun run index.ts flow --help +``` + ## JSON 输出 文档中带有 `--format=json` 和 `--json` 的命令遵循以下约定： diff --git a/src/application/bootstrap/__snapshots__/run-cli.test.ts.snap b/src/application/bootstrap/__snapshots__/run-cli.test.ts.snap index c242a43..63b2a35 100644 --- a/src/application/bootstrap/__snapshots__/run-cli.test.ts.snap +++ b/src/application/bootstrap/__snapshots__/run-cli.test.ts.snap @@ -45,6 +45,7 @@ Commands: check-update [options] Check for CLI updates connector Manage connector actions file Manage temporary file transfers + flow [args...] Run Open Flow info [options] Show CLI environment info llm Manage LLM client config login [options] Log in with an OOMOL account (alias for auth @@ -87,6 +88,7 @@ Commands: check-update [options] Check for CLI updates connector Manage connector actions file Manage temporary file transfers + flow [args...] Run Open Flow info [options] Show CLI environment info llm Manage LLM client config login [options] Log in with an OOMOL account (alias for auth @@ -132,6 +134,7 @@ Commands: check-update [options] Check for CLI updates connector Manage connector actions file Manage temporary file transfers + flow [args...] Run Open Flow info [options] Show CLI environment info llm Manage LLM client config login [options] Log in with an OOMOL account (alias for auth @@ -174,6 +177,7 @@ exports[`runCli bootstrap renders help in English and Chinese 1`] = ` check-update [options] 检查 CLI 更新 connector 管理 connector action file 管理临时文件传输 + flow [args...] 运行 Open Flow info [options] 显示 CLI 环境信息 llm 管理 LLM client 配置 login [options] 登录 OOMOL 账号（auth login 的别名） @@ -211,6 +215,7 @@ Commands: check-update [options] Check for CLI updates connector Manage connector actions file Manage temporary file transfers + flow [args...] Run Open Flow info [options] Show CLI environment info llm Manage LLM client config login [options] Log in with an OOMOL account (alias for auth @@ -254,6 +259,7 @@ Commands: check-update [options] Check for CLI updates connector Manage connector actions file Manage temporary file transfers + flow [args...] Run Open Flow info [options] Show CLI environment info llm Manage LLM client config login [options] Log in with an OOMOL account (alias for auth diff --git a/src/application/bootstrap/run-cli.ts b/src/application/bootstrap/run-cli.ts index e0cb170..2985b50 100644 --- a/src/application/bootstrap/run-cli.ts +++ b/src/application/bootstrap/run-cli.ts @@ -36,6 +36,10 @@ import { import { createTranslator } from "../../i18n/translator.ts"; import { migrateLegacyDefaultTeam } from "../auth/default-team.ts"; import { createCliCatalog } from "../commands/catalog.ts"; +import { + resolveOpenFlowInvocation, + runOpenFlowCommand, +} from "../commands/flow.ts"; import { synchronizeManagedSkillsForAvailableHosts } from "../commands/skills/auto-sync.ts"; import { APP_NAME } from "../config/app-config.ts"; import { @@ -155,8 +159,12 @@ export async function executeCli(invocation: CliInvocation): Promise { const startTimeMs = Date.now(); const sessionId = Bun.randomUUIDv7(); const telemetryRecorder = createTelemetryInvocationRecorder(); - const debugPathEnabled = hasCliDebugFlag(invocation.argv); - const rawCliLanguage = detectCliLanguageFlag(invocation.argv); + const openFlowInvocation = resolveOpenFlowInvocation(invocation.argv); + const ooArgv = openFlowInvocation === undefined + ? invocation.argv + : invocation.argv.slice(0, openFlowInvocation.commandIndex); + const debugPathEnabled = hasCliDebugFlag(ooArgv); + const rawCliLanguage = detectCliLanguageFlag(ooArgv); const parsedCliLanguage = parseExplicitLocale(rawCliLanguage); const storePaths = resolveStorePaths({ appName: APP_NAME, @@ -326,12 +334,31 @@ export async function executeCli(invocation: CliInvocation): Promise { await synchronizeManagedSkillsForAvailableHosts(context); } - exitCode = await adapter.run({ - argv: invocation.argv, - catalog, - context, - observer: telemetryRecorder.observer, - }); + if (openFlowInvocation !== undefined) { + telemetryRecorder.observer.onCommandResolved?.({ + argCount: 0, + commandPath: ["flow"], + excludeFromTelemetry: false, + flagsCount: 0, + outputFormat: "text", + }); + exitCode = await runOpenFlowCommand(openFlowInvocation.args, context); + + if (exitCode === 0) { + telemetryRecorder.observer.onCommandCompleted?.({ exitCode }); + } + else { + telemetryRecorder.observer.onCommandFailed?.({ exitCode }); + } + } + else { + exitCode = await adapter.run({ + argv: invocation.argv, + catalog, + context, + observer: telemetryRecorder.observer, + }); + } } catch (error) { if (error instanceof CliUserError) { @@ -676,6 +703,17 @@ function getSystemLocale(): string | undefined { } function redactSensitiveCliArguments(argv: readonly string[]): string[] { + const openFlowInvocation = resolveOpenFlowInvocation(argv); + + if (openFlowInvocation !== undefined) { + return [ + ...argv.slice(0, openFlowInvocation.commandIndex + 1), + ...(openFlowInvocation.args.length === 0 + ? [] + : [redactedCliArgumentValue]), + ]; + } + const positionalRule = sensitiveCliPositionalRules.find(rule => rule.commandPath.every((word, index) => argv[index] === word), ); diff --git a/src/application/commands/catalog.ts b/src/application/commands/catalog.ts index 44d8fd8..c166a6a 100644 --- a/src/application/commands/catalog.ts +++ b/src/application/commands/catalog.ts @@ -7,6 +7,7 @@ import { completionCommand } from "./completion.ts"; import { configCommand } from "./config/index.ts"; import { connectorCommand } from "./connector/index.ts"; import { fileCommand } from "./file/index.ts"; +import { flowCommand } from "./flow.ts"; import { infoCommand } from "./info.ts"; import { installCommand } from "./install.ts"; import { llmCommand } from "./llm/index.ts"; @@ -48,6 +49,7 @@ export function createCliCatalog(): CliCatalog { checkUpdateCommand, connectorCommand, fileCommand, + flowCommand, infoCommand, installCommand, llmCommand, diff --git a/src/application/commands/config/__snapshots__/index.cli.test.ts.snap b/src/application/commands/config/__snapshots__/index.cli.test.ts.snap index fac8f15..680fda9 100644 --- a/src/application/commands/config/__snapshots__/index.cli.test.ts.snap +++ b/src/application/commands/config/__snapshots__/index.cli.test.ts.snap @@ -31,6 +31,7 @@ Commands: check-update [options] Check for CLI updates connector Manage connector actions file Manage temporary file transfers + flow [args...] Run Open Flow info [options] Show CLI environment info llm Manage LLM client config login [options] Log in with an OOMOL account (alias for auth @@ -69,6 +70,7 @@ Commands: check-update [options] 检查 CLI 更新 connector 管理 connector action file 管理临时文件传输 + flow [args...] 运行 Open Flow info [options] 显示 CLI 环境信息 llm 管理 LLM client 配置 login [options] 登录 OOMOL 账号（auth login 的别名） diff --git a/src/application/commands/flow.cli.test.ts b/src/application/commands/flow.cli.test.ts new file mode 100644 index 0000000..2cfa032 --- /dev/null +++ b/src/application/commands/flow.cli.test.ts @@ -0,0 +1,175 @@ +import { mkdir, rm, writeFile } from "node:fs/promises"; +import { join } from "node:path"; + +import { describe, expect, test } from "bun:test"; +import { + createCliSandbox, + createTemporaryDirectory, + readLatestLogContent, +} from "../../../__tests__/helpers.ts"; +import { resolveOpenFlowInvocation } from "./flow.ts"; + +describe("flow CLI", () => { + test("recognizes flow after oo global options without consuming delegated options", () => { + expect(resolveOpenFlowInvocation([ + "--debug", + "--lang=zh", + "flow", + "dev", + "--lang", + "en", + ])).toEqual({ + args: ["dev", "--lang", "en"], + commandIndex: 2, + }); + expect(resolveOpenFlowInvocation(["help", "flow"])).toBeUndefined(); + }); + + test("delegates all flow arguments and returns the Open Flow exit code", async () => { + const sandbox = await createCliSandbox(); + const commandDirectory = await createTemporaryDirectory("oo-open-flow-command"); + const captureKey = `open-flow-args-${Bun.randomUUIDv7()}`; + + try { + await writeCommandEntry(commandDirectory, [ + `Reflect.set(globalThis, ${JSON.stringify(captureKey)}, [...args]);`, + "return 7;", + ]); + sandbox.env.OO_OPEN_FLOW_COMMAND_DIR = commandDirectory; + + const result = await sandbox.run([ + "--lang", + "en", + "flow", + "run", + "project.oo.yaml", + "--connector-token", + "secret-token", + ]); + + expect(result).toEqual({ + exitCode: 7, + stderr: "", + stdout: "", + }); + expect(Reflect.get(globalThis, captureKey)).toEqual([ + "run", + "project.oo.yaml", + "--connector-token", + "secret-token", + ]); + } + finally { + Reflect.deleteProperty(globalThis, captureKey); + await Promise.all([ + sandbox.cleanup(), + rm(commandDirectory, { force: true, recursive: true }), + ]); + } + }); + + test("keeps every delegated argument out of the debug log", async () => { + const sandbox = await createCliSandbox(); + const commandDirectory = await createTemporaryDirectory("oo-open-flow-command"); + + try { + await writeCommandEntry(commandDirectory, ["return 0;"]); + sandbox.env.OO_OPEN_FLOW_COMMAND_DIR = commandDirectory; + + const result = await sandbox.run([ + "flow", + "run", + "private/project.oo.yaml", + "--connector-token", + "secret-token", + ]); + const logContent = await readLatestLogContent(sandbox); + + expect(result.exitCode).toBe(0); + expect(logContent).toContain("\"argv\":[\"flow\",\"\"]"); + expect(logContent).not.toContain("private/project.oo.yaml"); + expect(logContent).not.toContain("secret-token"); + } + finally { + await Promise.all([ + sandbox.cleanup(), + rm(commandDirectory, { force: true, recursive: true }), + ]); + } + }); + + test("explains how to configure the local command directory", async () => { + const sandbox = await createCliSandbox(); + + try { + const result = await sandbox.run(["flow", "--help"]); + + expect(result.exitCode).toBe(1); + expect(result.stdout).toBe(""); + expect(result.stderr).toContain("OO_OPEN_FLOW_COMMAND_DIR"); + } + finally { + await sandbox.cleanup(); + } + }); + + test("rejects a command artifact built for another Bun version", async () => { + const sandbox = await createCliSandbox(); + const commandDirectory = await createTemporaryDirectory("oo-open-flow-command"); + + try { + await writeCommandEntry(commandDirectory, ["return 0;"], "0.0.0"); + sandbox.env.OO_OPEN_FLOW_COMMAND_DIR = commandDirectory; + + const result = await sandbox.run(["flow", "--version"]); + + expect(result.exitCode).toBe(1); + expect(result.stdout).toBe(""); + expect(result.stderr).toContain( + `Open Flow requires Bun 0.0.0, but oo is running Bun ${Bun.version}.`, + ); + } + finally { + await Promise.all([ + sandbox.cleanup(), + rm(commandDirectory, { force: true, recursive: true }), + ]); + } + }); + + test("lists flow in root help and provides host-side help", async () => { + const sandbox = await createCliSandbox(); + + try { + const rootHelp = await sandbox.run(["--help"]); + const flowHelp = await sandbox.run(["help", "flow"]); + + expect(rootHelp.exitCode).toBe(0); + expect(rootHelp.stdout).toContain("flow"); + expect(flowHelp.exitCode).toBe(0); + expect(flowHelp.stdout).toContain("Arguments passed to Open Flow"); + } + finally { + await sandbox.cleanup(); + } + }); +}); + +async function writeCommandEntry( + commandDirectory: string, + body: readonly string[], + requiredBunVersion: string = Bun.version, +): Promise { + await mkdir(commandDirectory, { recursive: true }); + await writeFile( + join(commandDirectory, "entry.js"), + [ + "export const commandArtifactVersion = 1;", + `export const requiredBunVersion = ${JSON.stringify(requiredBunVersion)};`, + "export async function runOpenFlowCommand(args) {", + ...body, + "}", + "", + ].join("\n"), + ); +} diff --git a/src/application/commands/flow.ts b/src/application/commands/flow.ts new file mode 100644 index 0000000..ae1e9db --- /dev/null +++ b/src/application/commands/flow.ts @@ -0,0 +1,131 @@ +import type { + CliCommandDefinition, + CliExecutionContext, +} from "../contracts/cli.ts"; + +import { join, resolve } from "node:path"; +import { pathToFileURL } from "node:url"; +import { CliUserError } from "../contracts/cli.ts"; + +const commandDirectoryEnvName = "OO_OPEN_FLOW_COMMAND_DIR"; +const commandArtifactVersion = 1; + +interface CommandModule { + readonly commandArtifactVersion: unknown; + readonly requiredBunVersion: unknown; + readonly runOpenFlowCommand: unknown; +} + +interface OpenFlowInvocation { + readonly args: readonly string[]; + readonly commandIndex: number; +} + +export const flowCommand = { + name: "flow", + summaryKey: "commands.flow.summary", + descriptionKey: "commands.flow.description", + arguments: [ + { + name: "args", + descriptionKey: "arguments.flowArgs", + required: false, + variadic: true, + }, + ], +} satisfies CliCommandDefinition; + +export function resolveOpenFlowInvocation( + argv: readonly string[], +): OpenFlowInvocation | undefined { + let commandIndex = 0; + + while (commandIndex < argv.length) { + const argument = argv[commandIndex]; + + if (argument === "--debug" || argument?.startsWith("--lang=")) { + commandIndex += 1; + continue; + } + + if (argument === "--lang") { + commandIndex += 2; + continue; + } + + break; + } + + if (argv[commandIndex] !== "flow") { + return undefined; + } + + return { + args: argv.slice(commandIndex + 1), + commandIndex, + }; +} + +export async function runOpenFlowCommand( + args: readonly string[], + context: Pick, +): Promise { + const configuredDirectory = context.env[commandDirectoryEnvName]?.trim(); + + if (!configuredDirectory) { + throw new CliUserError("errors.flow.commandDirectoryRequired", 1); + } + + const entryPath = join(resolve(context.cwd, configuredDirectory), "entry.js"); + let loaded: unknown; + + try { + loaded = await import(pathToFileURL(entryPath).href); + } + catch (error) { + throw new CliUserError("errors.flow.commandEntryLoadFailed", 1, { + message: error instanceof Error ? error.message : String(error), + path: entryPath, + }); + } + + if (loaded === null || typeof loaded !== "object") { + throw new CliUserError("errors.flow.commandEntryInvalid", 1, { + path: entryPath, + }); + } + + const commandModule = loaded as CommandModule; + + if ( + commandModule.commandArtifactVersion !== commandArtifactVersion + || typeof commandModule.requiredBunVersion !== "string" + || typeof commandModule.runOpenFlowCommand !== "function" + ) { + throw new CliUserError("errors.flow.commandEntryInvalid", 1, { + path: entryPath, + }); + } + + if (commandModule.requiredBunVersion !== Bun.version) { + throw new CliUserError("errors.flow.bunVersionMismatch", 1, { + actual: Bun.version, + required: commandModule.requiredBunVersion, + }); + } + + const exitCode = await commandModule.runOpenFlowCommand(args); + + if ( + typeof exitCode !== "number" + || !Number.isInteger(exitCode) + || exitCode < 0 + || exitCode > 255 + ) { + throw new CliUserError("errors.flow.commandEntryInvalid", 1, { + path: entryPath, + }); + } + + return exitCode; +} diff --git a/src/application/commands/telemetry-decisions.test.ts b/src/application/commands/telemetry-decisions.test.ts index 93b68b8..3e38546 100644 --- a/src/application/commands/telemetry-decisions.test.ts +++ b/src/application/commands/telemetry-decisions.test.ts @@ -240,6 +240,10 @@ const commandTelemetryDecisions = { properties: ["bytes_total_bucket", "rejected_too_large"], reason: "Records upload size bucket and rejection state without path or filename.", }, + "flow": { + kind: "generic", + reason: "Generic command telemetry records only the delegated flow command and its exit code; Open Flow arguments, flags, paths, project identities, and tokens are not inspected.", + }, "info": { kind: "generic", reason: "Generic command telemetry is enough; environment paths and agent presence are not recorded.", diff --git a/src/i18n/catalog.ts b/src/i18n/catalog.ts index 526fd76..d3eabc1 100644 --- a/src/i18n/catalog.ts +++ b/src/i18n/catalog.ts @@ -134,6 +134,18 @@ export const enMessages = { "commands.file.upload.description": "Upload a file and store the signed download URL locally.", "commands.file.upload.summary": "Upload a file", + "commands.flow.description": + "Run the Open Flow CLI with all following arguments passed through unchanged.", + "commands.flow.summary": "Run Open Flow", + "arguments.flowArgs": "Arguments passed to Open Flow", + "errors.flow.commandDirectoryRequired": + "Open Flow is not available locally. Set OO_OPEN_FLOW_COMMAND_DIR to the built Open Flow command directory.", + "errors.flow.commandEntryInvalid": + "The Open Flow command entry at {path} is invalid.", + "errors.flow.commandEntryLoadFailed": + "Failed to load the Open Flow command entry at {path}: {message}", + "errors.flow.bunVersionMismatch": + "Open Flow requires Bun {required}, but oo is running Bun {actual}.", "commands.info.description": "Print CLI environment details, persisted store paths, and detected skill agents.", "commands.info.summary": "Show CLI environment info", @@ -1521,6 +1533,16 @@ export const zhMessages = { "commands.file.summary": "管理临时文件传输", "commands.file.upload.description": "上传文件，并在本地保存带签名的下载地址。", "commands.file.upload.summary": "上传文件", + "commands.flow.description": "运行 Open Flow CLI，并将后续参数原样传递给它。", + "commands.flow.summary": "运行 Open Flow", + "arguments.flowArgs": "传递给 Open Flow 的参数", + "errors.flow.commandDirectoryRequired": + "本地 Open Flow 尚不可用。请将 OO_OPEN_FLOW_COMMAND_DIR 设置为构建后的 Open Flow 命令目录。", + "errors.flow.commandEntryInvalid": "{path} 中的 Open Flow 命令入口无效。", + "errors.flow.commandEntryLoadFailed": + "无法加载 {path} 中的 Open Flow 命令入口：{message}", + "errors.flow.bunVersionMismatch": + "Open Flow 需要 Bun {required}，但 oo 当前运行的是 Bun {actual}。", "commands.info.description": "打印 CLI 运行环境信息、本地存储路径以及检测到的 skill 代理。", "commands.info.summary": "显示 CLI 环境信息", From 0d05f6773b5464957ffab88eeb52bc39ccfca3b4 Mon Sep 17 00:00:00 2001 From: l1shen <648952316@qq.com> Date: Thu, 6 Aug 2026 11:31:44 +0800 Subject: [PATCH 2/6] feat: distribute the Open Flow command release --- docs/commands.md | 45 +- docs/commands.zh-CN.md | 33 +- .../static-completion-renderer.test.ts | 10 + .../__snapshots__/run-cli.test.ts.snap | 6 - src/application/bootstrap/run-cli.ts | 2 +- src/application/commands/catalog.ts | 9 +- .../__snapshots__/index.cli.test.ts.snap | 2 - src/application/commands/connector/shared.ts | 34 +- src/application/commands/file/download.ts | 2 +- .../file/download/file-system.test.ts | 2 +- .../commands/file/download/file-system.ts | 2 +- .../commands/flow-artifact.test.ts | 357 +++++++ src/application/commands/flow-artifact.ts | 874 ++++++++++++++++++ src/application/commands/flow-release.ts | 23 + src/application/commands/flow.cli.test.ts | 189 +++- src/application/commands/flow.ts | 174 +++- .../download-progress.test.ts} | 23 +- .../download-progress.ts} | 36 +- src/application/commands/team/identity.ts | 21 + .../commands/telemetry-decisions.test.ts | 2 +- src/i18n/catalog.ts | 12 +- 21 files changed, 1772 insertions(+), 86 deletions(-) create mode 100644 src/application/commands/flow-artifact.test.ts create mode 100644 src/application/commands/flow-artifact.ts create mode 100644 src/application/commands/flow-release.ts rename src/application/commands/{file/download/progress.test.ts => shared/download-progress.test.ts} (71%) rename src/application/commands/{file/download/progress.ts => shared/download-progress.ts} (76%) diff --git a/docs/commands.md b/docs/commands.md index 16652c9..fdc13ec 100644 --- a/docs/commands.md +++ b/docs/commands.md @@ -26,9 +26,9 @@ use. Truthy values are `1`, `true`, `yes`, or `on` (case-insensitive). - `OO_OPEN_FLOW_COMMAND_DIR`: During local Open Flow integration testing, point `oo flow` at an expanded Open Flow command artifact directory. The directory must contain `entry.js`; the standard repository build writes it to - `packages/open-flow/dist/command/open-flow-command`. This variable is - currently required because remote artifact distribution is not connected - yet. + `packages/open-flow/dist/command/open-flow-command`. When this variable is + unset, `oo flow` uses the Open Flow release bundled with the current `oo` + release. - `OO_LOG_DIR`: Override the debug-log directory. Takes precedence over every platform default. - `OO_API_KEY`: Run execution commands with this API key without an interactive @@ -90,22 +90,40 @@ use. Truthy values are `1`, `true`, `yes`, or `on` (case-insensitive). ### `oo flow [args...]` -Run the Open Flow CLI from the command artifact directory configured by -`OO_OPEN_FLOW_COMMAND_DIR`. +Run the Open Flow CLI release pinned by the current `oo` release. The first +invocation downloads and verifies its immutable command archive. Later +invocations reuse the verified local cache without checking for updates or +requiring network access. - Every argument after `flow` is passed to Open Flow unchanged. The main `oo` CLI does not parse, reorder, or log these arguments. +- The effective `oo --lang` locale is passed to Open Flow as `en` or + `zh-CN`. Open Flow owns and versions its translated command text. - `oo flow --help` and `oo flow --version` are therefore Open Flow commands. Use `oo help flow` to show the host-side command description without loading Open Flow. +- Root help and generated shell completions list `flow` only when + `OO_ENDPOINT=oomol.dev`. Other endpoints hide it without disabling direct + `oo flow` or `oo help flow` invocations. - Main `oo` global options such as `--lang` and `--debug` must appear before `flow`. Options after `flow` belong to Open Flow. - Open Flow uses the current process's working directory, standard streams, environment, and signals. Its exit code becomes the `oo` exit code. -- The loaded artifact must target the same Bun version embedded in `oo`. -- Local authoring, validation, development, and execution commands are - available. Open Flow Cloud commands remain unavailable until the OOMOL host - request transport is connected. +- The downloaded archive is accepted only when its length, SHA-256 digest, + internal manifest, complete file set, and Bun version match the release + pinned by `oo`. +- On a cache miss, interactive terminals show in-place byte progress on stderr; + non-interactive streams receive one start and one completion line. Cache hits + remain silent. +- `OO_OPEN_FLOW_COMMAND_DIR` bypasses download and cache resolution for local + repository integration testing. +- Open Flow Cloud commands use the current `oo` credential and effective Team. + The Cloud gateway is derived from the current endpoint as + `https://open-flow.`; for example, `OO_ENDPOINT=oomol.dev` uses + `https://open-flow.oomol.dev`. +- The `oo` credential and Team identity are attached only to `/v1/` requests + for that gateway. Deployment package uploads use the temporary upload URL + returned by Cloud without the `oo` credential. Local repository example: @@ -118,6 +136,15 @@ OO_OPEN_FLOW_COMMAND_DIR=/path/to/open-flow/packages/open-flow/dist/command/open bun run index.ts flow --help ``` +Test the online dev Cloud with a locally built Open Flow command (requires a +dev login and an effective Team): + +```bash +OO_ENDPOINT=oomol.dev \ +OO_OPEN_FLOW_COMMAND_DIR=/path/to/open-flow/packages/open-flow/dist/command/open-flow-command \ + bun run index.ts flow project list +``` + ## JSON Output Commands that document `--format=json` and `--json` share the following diff --git a/docs/commands.zh-CN.md b/docs/commands.zh-CN.md index cc9dc82..79cceb6 100644 --- a/docs/commands.zh-CN.md +++ b/docs/commands.zh-CN.md @@ -24,8 +24,8 @@ CLI 读取以下环境变量以支持内置和自动化场景。真值为 `1`、 - `OO_OPEN_FLOW_COMMAND_DIR`：本地联调 Open Flow 时，让 `oo flow` 使用指定的已展开命令产物目录。该目录必须包含 `entry.js`；Open Flow 仓库的标准构建会将它写到 - `packages/open-flow/dist/command/open-flow-command`。远端产物分发尚未接入， - 因此当前必须设置此变量。 + `packages/open-flow/dist/command/open-flow-command`。未设置时，`oo flow` + 使用当前 `oo` 版本固定的 Open Flow release。 - `OO_LOG_DIR`：覆盖 debug 日志目录。优先级高于所有平台默认值。 - `OO_API_KEY`：使用该 API key 执行命令，无需交互式登录。设置后 CLI 会构造一个 内存账号，不读取、不要求、也不写入 `auth.toml`，且优先级高于任何已保存的账号。 @@ -74,19 +74,32 @@ CLI 读取以下环境变量以支持内置和自动化场景。真值为 `1`、 ### `oo flow [args...]` -从 `OO_OPEN_FLOW_COMMAND_DIR` 配置的命令产物目录运行 Open Flow CLI。 +运行当前 `oo` 版本固定的 Open Flow CLI release。首次调用会下载并验证对应的 +不可变命令归档；之后直接离线复用已验证的本地缓存，不会在每次启动时检查更新。 - `flow` 后的全部参数都会原样传给 Open Flow；主 `oo` CLI 不解析、不重排，也不把这些参数写入日志。 +- 当前生效的 `oo --lang` locale 会以 `en` 或 `zh-CN` 传给 Open Flow； + Open Flow 自己拥有并随版本发布对应的命令翻译文案。 - 因此 `oo flow --help` 和 `oo flow --version` 都属于 Open Flow 命令。 如需在不加载 Open Flow 的情况下查看宿主侧命令说明，请使用 `oo help flow`。 +- 只有设置 `OO_ENDPOINT=oomol.dev` 时，根帮助和生成的 shell 补全才会列出 + `flow`。其他 endpoint 只会隐藏该命令，不会禁用直接调用 `oo flow` 或 + `oo help flow`。 - `--lang`、`--debug` 等 `oo` 全局选项必须放在 `flow` 前面； `flow` 后的选项归 Open Flow 所有。 - Open Flow 使用当前进程的工作目录、标准输入输出、环境变量和信号； 它的退出码会直接成为 `oo` 的退出码。 -- 被加载的产物必须与 `oo` 内嵌的 Bun 版本一致。 -- 当前可以使用本地 authoring、校验、开发和执行命令。OOMOL - 宿主请求传输接入前，Open Flow Cloud 子命令暂不可用。 +- 只有归档长度、SHA-256、内部 manifest、完整文件集合和 Bun 版本都与 `oo` + 固定的 release 一致时，下载内容才会被接受。 +- 缓存未命中时，交互式终端会在 stderr 原地刷新字节进度；非交互式输出会分别打印 + 一行开始和完成信息。命中缓存时保持静默。 +- `OO_OPEN_FLOW_COMMAND_DIR` 仅用于本地仓库联调；设置后会跳过下载和缓存解析。 +- Open Flow Cloud 子命令使用当前 `oo` 登录凭证和生效的 Team。Cloud gateway + 由当前 endpoint 派生为 `https://open-flow.`；例如 + `OO_ENDPOINT=oomol.dev` 使用 `https://open-flow.oomol.dev`。 +- `oo` 登录凭证和 Team identity 只附加到上述 gateway 的 `/v1/` 请求。部署包上传使用 + Cloud 返回的临时上传地址，不携带 `oo` 登录凭证。 本地仓库联调示例： @@ -99,6 +112,14 @@ OO_OPEN_FLOW_COMMAND_DIR=/path/to/open-flow/packages/open-flow/dist/command/open bun run index.ts flow --help ``` +使用本地构建的 Open Flow 命令测试线上 dev Cloud（需要已登录 dev 账号及已选择 Team）： + +```bash +OO_ENDPOINT=oomol.dev \ +OO_OPEN_FLOW_COMMAND_DIR=/path/to/open-flow/packages/open-flow/dist/command/open-flow-command \ + bun run index.ts flow project list +``` + ## JSON 输出 文档中带有 `--format=json` 和 `--json` 的命令遵循以下约定： diff --git a/src/adapters/completion/static-completion-renderer.test.ts b/src/adapters/completion/static-completion-renderer.test.ts index b24d40c..e038ffe 100644 --- a/src/adapters/completion/static-completion-renderer.test.ts +++ b/src/adapters/completion/static-completion-renderer.test.ts @@ -58,4 +58,14 @@ describe("StaticCompletionRenderer", () => { ); expect(output).toContain("en zh"); }); + + test("shows flow completion only for the online dev endpoint", () => { + const renderer = new StaticCompletionRenderer(createTranslator("en")); + const hiddenOutput = renderer.render("fish", createCliCatalog()); + const devOutput = renderer.render("fish", createCliCatalog("oomol.dev")); + const flowCompletion = `complete -c ${APP_NAME} -n '__fish_use_subcommand' -a 'flow'`; + + expect(hiddenOutput).not.toContain(flowCompletion); + expect(devOutput).toContain(flowCompletion); + }); }); diff --git a/src/application/bootstrap/__snapshots__/run-cli.test.ts.snap b/src/application/bootstrap/__snapshots__/run-cli.test.ts.snap index 63b2a35..c242a43 100644 --- a/src/application/bootstrap/__snapshots__/run-cli.test.ts.snap +++ b/src/application/bootstrap/__snapshots__/run-cli.test.ts.snap @@ -45,7 +45,6 @@ Commands: check-update [options] Check for CLI updates connector Manage connector actions file Manage temporary file transfers - flow [args...] Run Open Flow info [options] Show CLI environment info llm Manage LLM client config login [options] Log in with an OOMOL account (alias for auth @@ -88,7 +87,6 @@ Commands: check-update [options] Check for CLI updates connector Manage connector actions file Manage temporary file transfers - flow [args...] Run Open Flow info [options] Show CLI environment info llm Manage LLM client config login [options] Log in with an OOMOL account (alias for auth @@ -134,7 +132,6 @@ Commands: check-update [options] Check for CLI updates connector Manage connector actions file Manage temporary file transfers - flow [args...] Run Open Flow info [options] Show CLI environment info llm Manage LLM client config login [options] Log in with an OOMOL account (alias for auth @@ -177,7 +174,6 @@ exports[`runCli bootstrap renders help in English and Chinese 1`] = ` check-update [options] 检查 CLI 更新 connector 管理 connector action file 管理临时文件传输 - flow [args...] 运行 Open Flow info [options] 显示 CLI 环境信息 llm 管理 LLM client 配置 login [options] 登录 OOMOL 账号（auth login 的别名） @@ -215,7 +211,6 @@ Commands: check-update [options] Check for CLI updates connector Manage connector actions file Manage temporary file transfers - flow [args...] Run Open Flow info [options] Show CLI environment info llm Manage LLM client config login [options] Log in with an OOMOL account (alias for auth @@ -259,7 +254,6 @@ Commands: check-update [options] Check for CLI updates connector Manage connector actions file Manage temporary file transfers - flow [args...] Run Open Flow info [options] Show CLI environment info llm Manage LLM client config login [options] Log in with an OOMOL account (alias for auth diff --git a/src/application/bootstrap/run-cli.ts b/src/application/bootstrap/run-cli.ts index 2985b50..cab3f10 100644 --- a/src/application/bootstrap/run-cli.ts +++ b/src/application/bootstrap/run-cli.ts @@ -275,7 +275,7 @@ export async function executeCli(invocation: CliInvocation): Promise { systemLocale: invocation.systemLocale, }), ); - const catalog = createCliCatalog(); + const catalog = createCliCatalog(invocation.env.OO_ENDPOINT?.trim()); const completionRenderer = new StaticCompletionRenderer(translator); const packageName = invocation.packageName ?? packageManifest.name; diff --git a/src/application/commands/catalog.ts b/src/application/commands/catalog.ts index c166a6a..b27cac5 100644 --- a/src/application/commands/catalog.ts +++ b/src/application/commands/catalog.ts @@ -39,7 +39,9 @@ const globalOptions = [ }, ] as const; -export function createCliCatalog(): CliCatalog { +const onlineDevEndpoint = "oomol.dev"; + +export function createCliCatalog(endpoint?: string): CliCatalog { return { name: APP_NAME, descriptionKey: "app.description", @@ -49,7 +51,10 @@ export function createCliCatalog(): CliCatalog { checkUpdateCommand, connectorCommand, fileCommand, - flowCommand, + { + ...flowCommand, + hidden: endpoint !== onlineDevEndpoint, + }, infoCommand, installCommand, llmCommand, diff --git a/src/application/commands/config/__snapshots__/index.cli.test.ts.snap b/src/application/commands/config/__snapshots__/index.cli.test.ts.snap index 680fda9..fac8f15 100644 --- a/src/application/commands/config/__snapshots__/index.cli.test.ts.snap +++ b/src/application/commands/config/__snapshots__/index.cli.test.ts.snap @@ -31,7 +31,6 @@ Commands: check-update [options] Check for CLI updates connector Manage connector actions file Manage temporary file transfers - flow [args...] Run Open Flow info [options] Show CLI environment info llm Manage LLM client config login [options] Log in with an OOMOL account (alias for auth @@ -70,7 +69,6 @@ Commands: check-update [options] 检查 CLI 更新 connector 管理 connector action file 管理临时文件传输 - flow [args...] 运行 Open Flow info [options] 显示 CLI 环境信息 llm 管理 LLM client 配置 login [options] 登录 OOMOL 账号（auth login 的别名） diff --git a/src/application/commands/connector/shared.ts b/src/application/commands/connector/shared.ts index d142b6a..678d27e 100644 --- a/src/application/commands/connector/shared.ts +++ b/src/application/commands/connector/shared.ts @@ -14,6 +14,7 @@ import { isNetworkRestrictedSandboxError, requestOo, } from "../shared/oo-request.ts"; +import { teamIdentityHeaders } from "../team/identity.ts"; export const connectorActionDefinitionSchema = z.object({ description: z.string().optional().default(""), @@ -204,7 +205,7 @@ export async function searchConnectorActions( // The action list itself is identity-independent, but each result's // `authenticated` flag reflects the effective identity's connected // apps, so the identity headers are forwarded like `apps`/`run`. - headers: connectorIdentityHeaders(options.identity), + headers: teamIdentityHeaders(options.identity), host: { baseUrl: options.target.baseUrl }, label: "Connector action search", logFields: { @@ -240,7 +241,7 @@ export async function listConnectorApps( authorization: options.target.authorization, context, errors: { scope: "connectorApps" }, - headers: connectorIdentityHeaders(options.identity), + headers: teamIdentityHeaders(options.identity), host: { baseUrl: options.target.baseUrl }, label: "Connector apps list", path: "/v1/apps", @@ -268,7 +269,7 @@ export async function listConnectorAppsByService( authorization: options.target.authorization, context, errors: { scope: "connectorApps" }, - headers: connectorIdentityHeaders(options.identity), + headers: teamIdentityHeaders(options.identity), host: { baseUrl: options.target.baseUrl }, label: "Connector apps list", logFields: { @@ -366,7 +367,7 @@ export async function runConnectorAction( errors: { scope: "connectorRun" }, headers: { ...connectorConnectionSelectorHeaders(options.connectionSelector), - ...connectorIdentityHeaders(options.identity), + ...teamIdentityHeaders(options.identity), }, host: { baseUrl: options.target.baseUrl }, jsonBody: { input: options.inputData }, @@ -407,7 +408,7 @@ export async function runConnectorProxy( authorization: options.target.authorization, context, errors: { scope: "connectorProxy" }, - headers: connectorIdentityHeaders(options.identity), + headers: teamIdentityHeaders(options.identity), host: { baseUrl: options.target.baseUrl }, jsonBody: options.proxyRequest, label: "Connector proxy", @@ -437,29 +438,6 @@ function connectorActionPath(serviceName: string, actionName: string): string { return `/v1/actions/${encodeURIComponent(serviceName)}.${encodeURIComponent(actionName)}`; } -// Builds the identity request headers (`x-oo-team-name` / `x-oo-team-id`) -// from whichever dimensions the identity carries. Returns an empty object for -// the personal identity so callers can spread it unconditionally. -function connectorIdentityHeaders( - identity: TeamIdentity | undefined, -): Record { - const headers: Record = {}; - - if (identity === undefined) { - return headers; - } - - if (identity.name !== null) { - headers["x-oo-team-name"] = identity.name; - } - - if (identity.id !== null) { - headers["x-oo-team-id"] = identity.id; - } - - return headers; -} - // Self-hosted servers are typically local processes, so a connection failure // usually means the server is not running — the sandbox hint the OOMOL paths // use would send users down the wrong path. diff --git a/src/application/commands/file/download.ts b/src/application/commands/file/download.ts index 893cb9a..18d2ae8 100644 --- a/src/application/commands/file/download.ts +++ b/src/application/commands/file/download.ts @@ -10,6 +10,7 @@ import { getConfiguredFileDownloadOutDir, } from "../../schemas/settings.ts"; import { bucketTelemetryBytes } from "../../telemetry/buckets.ts"; +import { createDownloadProgressReporter } from "../shared/download-progress.ts"; import { finalizeDownloadedFile, openTemporaryDownloadFile, @@ -22,7 +23,6 @@ import { parseFileDownloadUrl, } from "./download/input.ts"; import { resolveDownloadPlan } from "./download/plan.ts"; -import { createDownloadProgressReporter } from "./download/progress.ts"; import { createDownloadSessionKey, deleteDownloadSessionBestEffort, diff --git a/src/application/commands/file/download/file-system.test.ts b/src/application/commands/file/download/file-system.test.ts index 4d13672..bd96206 100644 --- a/src/application/commands/file/download/file-system.test.ts +++ b/src/application/commands/file/download/file-system.test.ts @@ -7,6 +7,7 @@ import { createTemporaryDirectory, createTextBuffer, } from "../../../../../__tests__/helpers.ts"; +import { createDownloadProgressReporter } from "../../shared/download-progress.ts"; import { createDownloadSessionRecordFixture, createDownloadSessionStoreSpy, @@ -20,7 +21,6 @@ import { resolveTemporaryDownloadFileName, writeDownloadToTemporaryFile, } from "./file-system.ts"; -import { createDownloadProgressReporter } from "./progress.ts"; describe("resolveTemporaryDownloadFileName", () => { test("skips reserved and existing temporary file names", async () => { diff --git a/src/application/commands/file/download/file-system.ts b/src/application/commands/file/download/file-system.ts index 1cdbbdf..71f7ea3 100644 --- a/src/application/commands/file/download/file-system.ts +++ b/src/application/commands/file/download/file-system.ts @@ -1,6 +1,6 @@ import type { FileHandle } from "node:fs/promises"; import type { CliExecutionContext } from "../../../contracts/cli.ts"; -import type { DownloadProgressReporter } from "./progress.ts"; +import type { DownloadProgressReporter } from "../../shared/download-progress.ts"; import type { ExistingDownloadSession, WriteDownloadPlan } from "./types.ts"; import { link, lstat, open, rm, unlink } from "node:fs/promises"; diff --git a/src/application/commands/flow-artifact.test.ts b/src/application/commands/flow-artifact.test.ts new file mode 100644 index 0000000..14d1c0b --- /dev/null +++ b/src/application/commands/flow-artifact.test.ts @@ -0,0 +1,357 @@ +import type { Fetcher } from "../contracts/cli.ts"; +import type { OpenFlowCommandRelease } from "./flow-release.ts"; + +import { createHash } from "node:crypto"; +import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { gzipSync } from "node:zlib"; +import { afterEach, describe, expect, test } from "bun:test"; +import { installOpenFlowCommandRelease } from "./flow-artifact.ts"; +import { openFlowCommandRelease } from "./flow-release.ts"; + +const temporaryDirectories = new Set(); +const textEncoder = new TextEncoder(); + +afterEach(async () => { + await Promise.all(Array.from(temporaryDirectories, path => + rm(path, { force: true, recursive: true }))); + temporaryDirectories.clear(); +}); + +describe("Open Flow command artifact", () => { + test("pins the published Open Flow command release", () => { + expect(openFlowCommandRelease).toEqual({ + archive: { + digest: "cc9b77573f04dbf1e936e2785fe210db2f64cda03267bfd1cd31a799ac6bbcac", + length: 5_006_561, + url: "https://static.oomol.com/release/apps/open-flow/command/open-flow-0.0.4-dev-cc9b77573f04dbf1e936e2785fe210db2f64cda03267bfd1cd31a799ac6bbcac.tar.gz", + }, + bunVersion: "1.3.14", + format: "open-flow-command-release", + openFlowVersion: "0.0.4-dev", + version: 1, + }); + }); + + test("reports one verified archive download and reuses its cache offline", async () => { + const fixture = createCommandReleaseFixture(); + const environment = await createTestEnvironment(); + const progress: number[] = []; + let requestCount = 0; + const fetcher = createArchiveFetcher(fixture.archive, () => { + requestCount += 1; + }); + const options = { + env: environment.env, + execPath: process.execPath, + fetcher, + onDownloadProgress: (downloadedBytes: number) => { + progress.push(downloadedBytes); + }, + }; + + const commandDirectory = await installOpenFlowCommandRelease( + fixture.release, + options, + ); + const cachedDirectory = await installOpenFlowCommandRelease( + fixture.release, + options, + ); + + expect(cachedDirectory).toBe(commandDirectory); + expect(requestCount).toBe(1); + expect(progress).toEqual([0, fixture.archive.byteLength]); + expect(await readFile(join(commandDirectory, "entry.js"), "utf8")) + .toBe(fixture.entrySource); + }); + + test("removes a corrupt cache entry and downloads the pinned archive again", async () => { + const fixture = createCommandReleaseFixture(); + const environment = await createTestEnvironment(); + let requestCount = 0; + const fetcher = createArchiveFetcher(fixture.archive, () => { + requestCount += 1; + }); + const options = { + env: environment.env, + execPath: process.execPath, + fetcher, + }; + const commandDirectory = await installOpenFlowCommandRelease( + fixture.release, + options, + ); + await writeFile(join(commandDirectory, "entry.js"), "corrupt\n"); + + const repairedDirectory = await installOpenFlowCommandRelease( + fixture.release, + options, + ); + + expect(repairedDirectory).toBe(commandDirectory); + expect(requestCount).toBe(2); + expect(await readFile(join(commandDirectory, "entry.js"), "utf8")) + .toBe(fixture.entrySource); + }); + + test("rejects archive bytes that do not match the pinned digest", async () => { + const fixture = createCommandReleaseFixture(); + const environment = await createTestEnvironment(); + const digest = "0".repeat(64); + const release = { + ...fixture.release, + archive: { + ...fixture.release.archive, + digest, + url: `https://static.example.test/open-flow-${fixture.release.openFlowVersion}-${digest}.tar.gz`, + }, + } satisfies OpenFlowCommandRelease; + + await expect(installOpenFlowCommandRelease(release, { + env: environment.env, + execPath: process.execPath, + fetcher: createArchiveFetcher(fixture.archive), + })).rejects.toThrow("digest does not match its release record"); + }); + + test("rejects links before writing any command artifact files", async () => { + const archive = encodeTarGzip([{ + body: new Uint8Array(), + mode: 0o644, + path: "open-flow-command/entry.js", + type: "2", + }]); + const release = createRelease(archive); + const environment = await createTestEnvironment(); + + await expect(installOpenFlowCommandRelease(release, { + env: environment.env, + execPath: process.execPath, + fetcher: createArchiveFetcher(archive), + })).rejects.toThrow("link, directory, device, metadata, or other non-file"); + }); + + test("serializes concurrent installation of the same digest", async () => { + const fixture = createCommandReleaseFixture(); + const environment = await createTestEnvironment(); + const requestStarted = Promise.withResolvers(); + const resumeRequest = Promise.withResolvers(); + let requestCount = 0; + const fetcher: Fetcher = async () => { + requestCount += 1; + requestStarted.resolve(); + await resumeRequest.promise; + return new Response(fixture.archive, { + headers: { + "content-length": String(fixture.archive.byteLength), + }, + }); + }; + const options = { + env: environment.env, + execPath: process.execPath, + fetcher, + }; + const first = installOpenFlowCommandRelease(fixture.release, options); + await requestStarted.promise; + const second = installOpenFlowCommandRelease(fixture.release, options); + resumeRequest.resolve(); + + const directories = await Promise.all([first, second]); + + expect(directories[0]).toBe(directories[1]); + expect(requestCount).toBe(1); + }); +}); + +function createCommandReleaseFixture(): { + archive: Uint8Array; + entrySource: string; + release: OpenFlowCommandRelease; +} { + const entrySource = [ + "export const commandArtifactVersion = 1;", + `export const requiredBunVersion = ${JSON.stringify(Bun.version)};`, + "export async function runOpenFlowCommand() { return 0; }", + "", + ].join("\n"); + const payload = [ + { body: textEncoder.encode("export {};\n"), path: "deployment-runtime.js" }, + { body: textEncoder.encode(entrySource), path: "entry.js" }, + { body: textEncoder.encode("# Open Flow\n"), path: "skills/SKILL.md" }, + { body: textEncoder.encode("\n"), path: "web/index.html" }, + ].toSorted((left, right) => left.path < right.path ? -1 : 1); + const files = payload.map(file => ({ + digest: sha256(file.body), + length: file.body.byteLength, + path: file.path, + })); + const manifest = `${JSON.stringify({ + bunVersion: Bun.version, + deploymentRuntime: "deployment-runtime.js", + entry: "entry.js", + files, + format: "open-flow-command-artifact", + openFlowVersion: "1.2.3-test", + skillsRoot: "skills", + version: 1, + webRoot: "web", + })}\n`; + const archive = encodeTarGzip([ + { + body: textEncoder.encode(manifest), + mode: 0o644, + path: "open-flow-command/command-artifact.json", + type: "0", + }, + ...payload.map(file => ({ + ...file, + mode: file.path === "entry.js" ? 0o755 : 0o644, + path: `open-flow-command/${file.path}`, + type: "0" as const, + })), + ]); + + return { + archive, + entrySource, + release: createRelease(archive), + }; +} + +function createRelease(archive: Uint8Array): OpenFlowCommandRelease { + const digest = sha256(archive); + const openFlowVersion = "1.2.3-test"; + + return { + archive: { + digest, + length: archive.byteLength, + url: `https://static.example.test/open-flow-${openFlowVersion}-${digest}.tar.gz`, + }, + bunVersion: Bun.version, + format: "open-flow-command-release", + openFlowVersion, + version: 1, + }; +} + +function createArchiveFetcher( + archive: Uint8Array, + onRequest: () => void = () => {}, +): Fetcher { + return () => { + onRequest(); + return Promise.resolve(new Response(archive, { + headers: { + "content-length": String(archive.byteLength), + }, + })); + }; +} + +async function createTestEnvironment(): Promise<{ + env: Record; +}> { + const root = await mkdtemp(join(tmpdir(), "oo-flow-artifact-")); + temporaryDirectories.add(root); + + return { + env: { + HOME: join(root, "home"), + LOCALAPPDATA: join(root, "local-app-data"), + USERPROFILE: join(root, "home"), + XDG_CACHE_HOME: join(root, "cache"), + }, + }; +} + +function encodeTarGzip(entries: readonly { + body: Uint8Array; + mode: number; + path: string; + type: "0" | "2"; +}[]): Uint8Array { + const chunks: Uint8Array[] = []; + + for (const entry of entries) { + const header = new Uint8Array(512); + writeTarText(header, 0, 100, entry.path); + writeTarOctal(header, 100, 8, entry.mode); + writeTarOctal(header, 108, 8, 0); + writeTarOctal(header, 116, 8, 0); + writeTarOctal(header, 124, 12, entry.body.byteLength); + writeTarOctal(header, 136, 12, 0); + header.fill(0x20, 148, 156); + header[156] = entry.type.charCodeAt(0); + header.set(textEncoder.encode("ustar\0"), 257); + header.set(textEncoder.encode("00"), 263); + const checksum = header.reduce((sum, byte) => sum + byte, 0); + writeTarChecksum(header, checksum); + chunks.push(header, entry.body); + + const padding = (512 - entry.body.byteLength % 512) % 512; + + if (padding > 0) { + chunks.push(new Uint8Array(padding)); + } + } + + chunks.push(new Uint8Array(1024)); + const tar = new Uint8Array(chunks.reduce((length, chunk) => length + chunk.byteLength, 0)); + let offset = 0; + + for (const chunk of chunks) { + tar.set(chunk, offset); + offset += chunk.byteLength; + } + + const archive = new Uint8Array(gzipSync(tar, { level: 9 })); + archive[3] = 0; + archive[4] = 0; + archive[5] = 0; + archive[6] = 0; + archive[7] = 0; + archive[8] = 2; + archive[9] = 255; + return archive; +} + +function writeTarText( + header: Uint8Array, + offset: number, + length: number, + value: string, +): void { + const bytes = textEncoder.encode(value); + + if (bytes.byteLength >= length) { + throw new TypeError(`Tar fixture path is too long: ${value}`); + } + + header.set(bytes, offset); +} + +function writeTarOctal( + header: Uint8Array, + offset: number, + length: number, + value: number, +): void { + const source = value.toString(8).padStart(length - 1, "0"); + header.set(textEncoder.encode(source), offset); + header[offset + length - 1] = 0; +} + +function writeTarChecksum(header: Uint8Array, value: number): void { + const source = value.toString(8).padStart(6, "0"); + header.set(textEncoder.encode(source), 148); + header[154] = 0; + header[155] = 0x20; +} + +function sha256(bytes: Uint8Array): string { + return createHash("sha256").update(bytes).digest("hex"); +} diff --git a/src/application/commands/flow-artifact.ts b/src/application/commands/flow-artifact.ts new file mode 100644 index 0000000..732b991 --- /dev/null +++ b/src/application/commands/flow-artifact.ts @@ -0,0 +1,874 @@ +import type { Fetcher } from "../contracts/cli.ts"; +import type { OpenFlowCommandRelease } from "./flow-release.ts"; + +import { createHash } from "node:crypto"; +import { mkdir, open, readdir, readFile, rename, rm, writeFile } from "node:fs/promises"; +import { basename, dirname, join } from "node:path"; +import process from "node:process"; +import { gunzipSync, gzipSync } from "node:zlib"; +import { z } from "zod"; +import { resolveHomeDirectory } from "../path/home-directory.ts"; +import { acquireDownloadTempLock } from "../shared/download-temp-lock.ts"; + +const commandArtifactFormat = "open-flow-command-artifact"; +const commandArtifactVersion = 1; +const commandArtifactManifestFile = "command-artifact.json"; +const commandArtifactEntryFile = "entry.js"; +const commandArtifactDeploymentRuntimeFile = "deployment-runtime.js"; +const commandArtifactWebRoot = "web"; +const commandArtifactSkillsRoot = "skills"; +const commandArchiveRoot = "open-flow-command"; +const commandArchiveMediaType = "application/vnd.open-flow.command-artifact+tar+gzip"; +const archivePrefix = `${commandArchiveRoot}/`; +const cacheFormatDirectory = "command-artifact-v1"; +const tarBlockSize = 512; +const tarEndMarkerSize = tarBlockSize * 2; +const gzipHeaderSize = 10; +const gzipFooterSize = 8; +const lockWaitTimeoutMs = 300_000; +const textDecoder = new TextDecoder("utf-8", { + fatal: true, + ignoreBOM: false, +}); +const textEncoder = new TextEncoder(); +const ustarMagic = textEncoder.encode("ustar\0"); +const ustarVersion = textEncoder.encode("00"); + +interface CommandArtifactFile { + readonly digest: string; + readonly length: number; + readonly path: string; +} + +interface CommandArchiveEntry { + readonly bytes: Uint8Array; + readonly mode: number; + readonly path: string; +} + +const manifestFileSchema = z.object({ + digest: z.string().refine(isSha256Digest, "Command artifact file digest must be a lowercase SHA-256 digest."), + length: z.number().int().nonnegative().max(Number.MAX_SAFE_INTEGER), + path: z.string().refine(isNormalizedArtifactPath, "Command artifact file paths must be normalized relative paths."), +}).strict(); + +const commandArtifactManifestSchema = z.object({ + bunVersion: z.string().min(1), + deploymentRuntime: z.literal(commandArtifactDeploymentRuntimeFile), + entry: z.literal(commandArtifactEntryFile), + files: z.array(manifestFileSchema), + format: z.literal(commandArtifactFormat), + openFlowVersion: z.string().min(1), + skillsRoot: z.literal(commandArtifactSkillsRoot), + version: z.literal(commandArtifactVersion), + webRoot: z.literal(commandArtifactWebRoot), +}).strict().superRefine((manifest, context) => { + let previousPath: string | undefined; + + for (const [index, file] of manifest.files.entries()) { + if (file.path === commandArtifactManifestFile) { + context.addIssue({ + code: "custom", + message: "The command artifact manifest cannot list itself.", + path: ["files", index, "path"], + }); + } + + if ( + previousPath !== undefined + && compareArtifactPaths(previousPath, file.path) >= 0 + ) { + context.addIssue({ + code: "custom", + message: "Command artifact files must have unique paths in Unicode code-point order.", + path: ["files", index, "path"], + }); + } + + previousPath = file.path; + } + + for (const path of [ + commandArtifactEntryFile, + commandArtifactDeploymentRuntimeFile, + ]) { + if (!manifest.files.some(file => file.path === path)) { + context.addIssue({ + code: "custom", + message: `Command artifact is missing required file ${JSON.stringify(path)}.`, + path: ["files"], + }); + } + } + + for (const root of [commandArtifactWebRoot, commandArtifactSkillsRoot]) { + if (!manifest.files.some(file => file.path.startsWith(`${root}/`))) { + context.addIssue({ + code: "custom", + message: `Command artifact is missing files under ${JSON.stringify(root)}.`, + path: ["files"], + }); + } + } +}); + +type CommandArtifactManifest = z.infer; + +export async function installOpenFlowCommandRelease( + release: OpenFlowCommandRelease, + options: { + env: Record; + execPath: string; + fetcher: Fetcher; + onDownloadProgress?: (downloadedBytes: number) => void; + }, +): Promise { + if (release.bunVersion !== Bun.version) { + invalid(`Open Flow requires Bun ${release.bunVersion}; received ${Bun.version}.`); + } + + const cacheRoot = resolveCommandCacheRoot(options.env); + const commandDirectory = join(cacheRoot, release.archive.digest); + + if (await validCommandArtifactDirectory(commandDirectory, release)) { + return commandDirectory; + } + + await mkdir(cacheRoot, { recursive: true }); + const lockFilePath = join(cacheRoot, ".locks", `${release.archive.digest}.lock`); + await mkdir(dirname(lockFilePath), { recursive: true }); + const sessionId = Bun.randomUUIDv7(); + const lock = await waitForArtifactLock({ + archiveName: basename(new URL(release.archive.url).pathname), + execPath: options.execPath, + lockFilePath, + sessionId, + }); + + try { + if (await validCommandArtifactDirectory(commandDirectory, release)) { + return commandDirectory; + } + + await rm(commandDirectory, { force: true, recursive: true }); + + const temporaryId = `${process.pid}-${Bun.randomUUIDv7()}`; + const archivePath = join(cacheRoot, `.archive-${temporaryId}.tar.gz`); + const extractionDirectory = join(cacheRoot, `.extract-${temporaryId}`); + + try { + await downloadCommandArchive( + release, + options.fetcher, + archivePath, + options.onDownloadProgress, + ); + const archive = await readFile(archivePath); + const decoded = decodeCommandArchive(archive, release); + await writeCommandArtifactDirectory(extractionDirectory, decoded); + await validateCommandArtifactDirectory(extractionDirectory, release); + await rename(extractionDirectory, commandDirectory); + } + finally { + await Promise.all([ + rm(archivePath, { force: true }), + rm(extractionDirectory, { force: true, recursive: true }), + ]); + } + + return commandDirectory; + } + finally { + await lock.close(); + } +} + +async function waitForArtifactLock(options: { + archiveName: string; + execPath: string; + lockFilePath: string; + sessionId: string; +}): Promise>, { status: "acquired" }>["handle"]> { + const deadline = Date.now() + lockWaitTimeoutMs; + + while (Date.now() < deadline) { + const result = await acquireDownloadTempLock({ + execPath: options.execPath, + lockFilePath: options.lockFilePath, + sessionId: options.sessionId, + tempFileName: options.archiveName, + }); + + if (result.status === "acquired") { + return result.handle; + } + + await Bun.sleep(100); + } + + invalid("Timed out waiting for another Open Flow command artifact installation."); +} + +async function downloadCommandArchive( + release: OpenFlowCommandRelease, + fetcher: Fetcher, + archivePath: string, + onProgress: ((downloadedBytes: number) => void) | undefined, +): Promise { + onProgress?.(0); + const response = await fetcher(release.archive.url, { + headers: { + accept: commandArchiveMediaType, + }, + }); + + if (!response.ok) { + invalid(`Open Flow command archive request failed with status ${response.status}.`); + } + + const declaredLength = response.headers.get("content-length"); + + if (declaredLength !== null) { + const parsedLength = Number(declaredLength); + + if ( + !Number.isSafeInteger(parsedLength) + || parsedLength < 0 + || parsedLength !== release.archive.length + ) { + invalid("Open Flow command archive response length does not match its release record."); + } + } + + if (response.body === null) { + invalid("Open Flow command archive response has no body."); + } + + const fileHandle = await open(archivePath, "wx"); + const digest = createHash("sha256"); + const reader = response.body.getReader(); + let length = 0; + + try { + while (true) { + const chunk = await reader.read(); + + if (chunk.done) { + break; + } + + if (length + chunk.value.byteLength > release.archive.length) { + invalid("Open Flow command archive is longer than its release record."); + } + + let written = 0; + + while (written < chunk.value.byteLength) { + const result = await fileHandle.write( + chunk.value, + written, + chunk.value.byteLength - written, + length + written, + ); + + if (result.bytesWritten === 0) { + invalid("Open Flow command archive download stopped before completion."); + } + + written += result.bytesWritten; + } + + digest.update(chunk.value); + length += chunk.value.byteLength; + onProgress?.(length); + } + + await fileHandle.sync(); + } + finally { + reader.releaseLock(); + await fileHandle.close(); + } + + if (length !== release.archive.length) { + invalid("Open Flow command archive length does not match its release record."); + } + + if (digest.digest("hex") !== release.archive.digest) { + invalid("Open Flow command archive digest does not match its release record."); + } +} + +function decodeCommandArchive( + archive: Uint8Array, + release: OpenFlowCommandRelease, +): readonly CommandArchiveEntry[] { + const tar = decodeCanonicalGzip(archive); + const files = decodeTar(tar); + const manifestFile = files.find(file => file.path === commandArtifactManifestFile); + + if (manifestFile === undefined) { + invalid(`Command archive is missing ${commandArtifactManifestFile}.`); + } + + const manifest = decodeCommandArtifactManifest(manifestFile.bytes); + validateManifestRelease(manifest, release); + const payloadFiles = files.filter(file => file.path !== commandArtifactManifestFile); + + if (payloadFiles.length !== manifest.files.length) { + invalid("Command archive file set does not match its manifest."); + } + + for (const [index, file] of payloadFiles.entries()) { + const expected = manifest.files[index]; + + if (expected === undefined || file.path !== expected.path) { + invalid("Command archive file set does not match its manifest."); + } + + validateFile(file.bytes, expected); + } + + return files; +} + +function decodeCanonicalGzip(archive: Uint8Array): Uint8Array { + if ( + archive.byteLength < gzipHeaderSize + gzipFooterSize + || archive[0] !== 0x1F + || archive[1] !== 0x8B + || archive[2] !== 8 + || archive[3] !== 0 + || archive[4] !== 0 + || archive[5] !== 0 + || archive[6] !== 0 + || archive[7] !== 0 + || archive[8] !== 2 + || archive[9] !== 255 + ) { + invalid("Command archive does not have the canonical gzip header."); + } + + let tar: Uint8Array; + + try { + tar = new Uint8Array(gunzipSync(archive)); + } + catch (error) { + throw new TypeError("Command archive gzip stream cannot be decoded.", { + cause: error, + }); + } + + if (!sameBytes(canonicalGzip(tar), archive)) { + invalid("Command archive gzip stream is truncated, has trailing data, or is not canonically encoded."); + } + + return tar; +} + +function canonicalGzip(tar: Uint8Array): Uint8Array { + const compressed = new Uint8Array(gzipSync(tar, { level: 9 })); + compressed[3] = 0; + compressed[4] = 0; + compressed[5] = 0; + compressed[6] = 0; + compressed[7] = 0; + compressed[8] = 2; + compressed[9] = 255; + return compressed; +} + +function decodeTar(tar: Uint8Array): readonly CommandArchiveEntry[] { + if (tar.byteLength < tarEndMarkerSize || tar.byteLength % tarBlockSize !== 0) { + invalid("Command archive is not a complete block-aligned tar stream."); + } + + const files: CommandArchiveEntry[] = []; + let offset = 0; + let previousPath: string | undefined; + + while (offset < tar.byteLength) { + if (zeroBytes(tar, offset, tarBlockSize)) { + if ( + offset + tarEndMarkerSize !== tar.byteLength + || !zeroBytes(tar, offset + tarBlockSize, tarBlockSize) + ) { + invalid("Command archive has an invalid tar end marker."); + } + + return files; + } + + validateTarChecksum(tar, offset); + + if ( + !sameBytes(tar.subarray(offset + 257, offset + 263), ustarMagic) + || !sameBytes(tar.subarray(offset + 263, offset + 265), ustarVersion) + ) { + invalid("Command archive contains a non-USTAR entry."); + } + + if (tar[offset + 156] !== 0x30) { + invalid("Command archive contains a link, directory, device, metadata, or other non-file tar entry."); + } + + const name = decodeTarText(tar, offset, 100, "name"); + const prefix = decodeTarText(tar, offset + 345, 155, "prefix"); + const archivePath = prefix === "" ? name : `${prefix}/${name}`; + + if (!archivePath.startsWith(archivePrefix)) { + invalid(`Command archive entry is outside ${archivePrefix}.`); + } + + const path = archivePath.slice(archivePrefix.length); + + if (!isNormalizedArtifactPath(path)) { + invalid(`Command archive contains an invalid file path: ${archivePath}`); + } + + if ( + previousPath !== undefined + && compareArtifactPaths(previousPath, path) >= 0 + ) { + invalid("Command archive paths are not sorted uniquely."); + } + + const mode = decodeTarOctal(tar, offset + 100, 8, "mode"); + + if ( + mode !== modeForArtifactPath(path) + || decodeTarOctal(tar, offset + 108, 8, "uid") !== 0 + || decodeTarOctal(tar, offset + 116, 8, "gid") !== 0 + || decodeTarOctal(tar, offset + 136, 12, "mtime") !== 0 + || decodeTarText(tar, offset + 157, 100, "linkname") !== "" + || decodeTarText(tar, offset + 265, 32, "uname") !== "" + || decodeTarText(tar, offset + 297, 32, "gname") !== "" + || decodeOptionalTarOctal(tar, offset + 329, 8, "devmajor") !== 0 + || decodeOptionalTarOctal(tar, offset + 337, 8, "devminor") !== 0 + || !zeroBytes(tar, offset + 500, 12) + ) { + invalid(`Command archive entry has invalid metadata: ${archivePath}`); + } + + const size = decodeTarOctal(tar, offset + 124, 12, "size"); + const bodyStart = offset + tarBlockSize; + const bodyEnd = bodyStart + size; + const nextOffset = bodyStart + Math.ceil(size / tarBlockSize) * tarBlockSize; + + if (bodyEnd > tar.byteLength - tarEndMarkerSize || nextOffset > tar.byteLength - tarEndMarkerSize) { + invalid(`Command archive contains a truncated file: ${archivePath}`); + } + + if (!zeroBytes(tar, bodyEnd, nextOffset - bodyEnd)) { + invalid(`Command archive file has non-zero tar padding: ${archivePath}`); + } + + files.push({ + bytes: tar.subarray(bodyStart, bodyEnd), + mode, + path, + }); + offset = nextOffset; + previousPath = path; + } + + invalid("Command archive is missing its tar end marker."); +} + +function validateTarChecksum(tar: Uint8Array, offset: number): void { + const expected = decodeTarOctal(tar, offset + 148, 8, "checksum"); + let actual = 0; + + for (let index = 0; index < tarBlockSize; index += 1) { + actual += index >= 148 && index < 156 + ? 0x20 + : tar[offset + index] ?? 0; + } + + if (actual !== expected) { + invalid("Command archive contains a tar header with an invalid checksum."); + } +} + +function decodeTarText( + tar: Uint8Array, + offset: number, + length: number, + field: string, +): string { + const bytes = tar.subarray(offset, offset + length); + let end = bytes.indexOf(0); + + if (end < 0) { + end = bytes.length; + } + else if (!zeroBytes(bytes, end, bytes.length - end)) { + invalid(`Command archive contains a non-canonical tar ${field} field.`); + } + + try { + return textDecoder.decode(bytes.subarray(0, end)); + } + catch (error) { + throw new TypeError(`Command archive tar ${field} is not valid UTF-8.`, { + cause: error, + }); + } +} + +function decodeTarOctal( + tar: Uint8Array, + offset: number, + length: number, + field: string, +): number { + const bytes = tar.subarray(offset, offset + length); + let digits = ""; + let terminated = false; + + for (const byte of bytes) { + if (byte === 0 || byte === 0x20) { + terminated = true; + continue; + } + + if (terminated || byte < 0x30 || byte > 0x37) { + invalid(`Command archive contains a non-canonical tar ${field} field.`); + } + + digits += String.fromCharCode(byte); + } + + if (digits === "") { + invalid(`Command archive contains an empty tar ${field} field.`); + } + + const value = Number.parseInt(digits, 8); + + if (!Number.isSafeInteger(value)) { + invalid(`Command archive contains an unsupported tar ${field} value.`); + } + + return value; +} + +function decodeOptionalTarOctal( + tar: Uint8Array, + offset: number, + length: number, + field: string, +): number { + return zeroBytes(tar, offset, length) + ? 0 + : decodeTarOctal(tar, offset, length, field); +} + +function decodeCommandArtifactManifest(bytes: Uint8Array): CommandArtifactManifest { + let source: string; + + try { + source = textDecoder.decode(bytes); + } + catch (error) { + throw new TypeError(`${commandArtifactManifestFile} is not valid UTF-8.`, { + cause: error, + }); + } + + let value: unknown; + + try { + value = JSON.parse(source); + } + catch (error) { + throw new TypeError(`${commandArtifactManifestFile} is not valid JSON.`, { + cause: error, + }); + } + + const result = commandArtifactManifestSchema.safeParse(value); + + if (!result.success) { + throw new TypeError(`${commandArtifactManifestFile} is invalid.`, { + cause: result.error, + }); + } + + if (`${stringifyCanonicalJson(result.data)}\n` !== source) { + invalid(`${commandArtifactManifestFile} is not canonical.`); + } + + return result.data; +} + +async function writeCommandArtifactDirectory( + directory: string, + files: readonly CommandArchiveEntry[], +): Promise { + await Promise.all(files.map(async (file) => { + const path = join(directory, ...file.path.split("/")); + await mkdir(dirname(path), { recursive: true }); + await writeFile(path, file.bytes, { mode: file.mode }); + })); +} + +async function validCommandArtifactDirectory( + directory: string, + release: OpenFlowCommandRelease, +): Promise { + try { + await validateCommandArtifactDirectory(directory, release); + return true; + } + catch { + return false; + } +} + +async function validateCommandArtifactDirectory( + directory: string, + release: OpenFlowCommandRelease, +): Promise { + const manifest = decodeCommandArtifactManifest( + await readFile(join(directory, commandArtifactManifestFile)), + ); + validateManifestRelease(manifest, release); + const actual = await collectArtifactDirectory(directory); + const expectedFiles = [ + commandArtifactManifestFile, + ...manifest.files.map(file => file.path), + ].toSorted(compareArtifactPaths); + const expectedDirectories = collectExpectedDirectories(expectedFiles); + + if ( + !sameStrings(actual.files, expectedFiles) + || !sameStrings(actual.directories, expectedDirectories) + ) { + invalid("Command artifact cache file set does not match its manifest."); + } + + await Promise.all(manifest.files.map(async (file) => { + const bytes = await readFile(join(directory, ...file.path.split("/"))); + validateFile(bytes, file); + })); +} + +async function collectArtifactDirectory( + directory: string, + prefix = "", +): Promise<{ directories: string[]; files: string[] }> { + const entries = await readdir(directory, { withFileTypes: true }); + const directories: string[] = []; + const files: string[] = []; + + for (const entry of entries) { + const path = prefix === "" ? entry.name : `${prefix}/${entry.name}`; + + if (!isNormalizedArtifactPath(path)) { + invalid(`Command artifact cache contains an invalid path: ${path}`); + } + + if (entry.isFile()) { + files.push(path); + continue; + } + + if (!entry.isDirectory()) { + invalid(`Command artifact cache contains a non-file entry: ${path}`); + } + + directories.push(path); + const nested = await collectArtifactDirectory( + join(directory, entry.name), + path, + ); + directories.push(...nested.directories); + files.push(...nested.files); + } + + return { + directories: directories.toSorted(compareArtifactPaths), + files: files.toSorted(compareArtifactPaths), + }; +} + +function collectExpectedDirectories(files: readonly string[]): string[] { + const directories = new Set(); + + for (const file of files) { + const parts = file.split("/"); + + for (let index = 1; index < parts.length; index += 1) { + directories.add(parts.slice(0, index).join("/")); + } + } + + return [...directories].toSorted(compareArtifactPaths); +} + +function validateManifestRelease( + manifest: CommandArtifactManifest, + release: OpenFlowCommandRelease, +): void { + if ( + manifest.openFlowVersion !== release.openFlowVersion + || manifest.bunVersion !== release.bunVersion + ) { + invalid("Command artifact manifest does not match its release record."); + } +} + +function validateFile(bytes: Uint8Array, expected: CommandArtifactFile): void { + if (bytes.byteLength !== expected.length) { + invalid(`Command artifact file length does not match its manifest: ${expected.path}`); + } + + if (sha256(bytes) !== expected.digest) { + invalid(`Command artifact file digest does not match its manifest: ${expected.path}`); + } +} + +function resolveCommandCacheRoot(env: Record): string { + const homeDirectory = resolveHomeDirectory(env); + let platformCacheRoot: string; + + switch (process.platform) { + case "darwin": + platformCacheRoot = join(homeDirectory, "Library", "Caches"); + break; + case "win32": + platformCacheRoot = env.LOCALAPPDATA + ?? join(homeDirectory, "AppData", "Local"); + break; + default: + platformCacheRoot = env.XDG_CACHE_HOME + ?? join(homeDirectory, ".cache"); + } + + return join(platformCacheRoot, "oo", "open-flow", cacheFormatDirectory); +} + +function isSha256Digest(value: string): boolean { + if (value.length !== 64) { + return false; + } + + for (const character of value) { + if ( + !(character >= "0" && character <= "9") + && !(character >= "a" && character <= "f") + ) { + return false; + } + } + + return true; +} + +function isNormalizedArtifactPath(path: string): boolean { + if ( + path === "" + || !path.isWellFormed() + || path.startsWith("/") + || path.includes("\\") + || path.includes("\0") + ) { + return false; + } + + const first = path[0]; + + if ( + first !== undefined + && path[1] === ":" + && path[2] === "/" + && ((first >= "A" && first <= "Z") || (first >= "a" && first <= "z")) + ) { + return false; + } + + return path.split("/").every(part => part !== "" && part !== "." && part !== ".."); +} + +function compareArtifactPaths(left: string, right: string): number { + let leftIndex = 0; + let rightIndex = 0; + + while (leftIndex < left.length && rightIndex < right.length) { + const leftCodePoint = left.codePointAt(leftIndex); + const rightCodePoint = right.codePointAt(rightIndex); + + if (leftCodePoint === undefined || rightCodePoint === undefined) { + break; + } + + if (leftCodePoint !== rightCodePoint) { + return leftCodePoint < rightCodePoint ? -1 : 1; + } + + leftIndex += leftCodePoint > 0xFFFF ? 2 : 1; + rightIndex += rightCodePoint > 0xFFFF ? 2 : 1; + } + + return leftIndex < left.length ? 1 : rightIndex < right.length ? -1 : 0; +} + +function stringifyCanonicalJson(value: unknown): string { + if (value === null || typeof value === "boolean" || typeof value === "string") { + return JSON.stringify(value); + } + + if (typeof value === "number") { + if (!Number.isFinite(value)) { + invalid("Command artifact manifests cannot contain non-finite numbers."); + } + + return JSON.stringify(Object.is(value, -0) ? 0 : value); + } + + if (Array.isArray(value)) { + return `[${value.map(stringifyCanonicalJson).join(",")}]`; + } + + if (typeof value === "object") { + return `{${Object.keys(value) + .toSorted(compareArtifactPaths) + .map(key => `${JSON.stringify(key)}:${stringifyCanonicalJson(Reflect.get(value, key))}`) + .join(",")}}`; + } + + invalid(`Command artifact manifests cannot contain ${typeof value} values.`); +} + +function modeForArtifactPath(path: string): number { + return path === commandArtifactEntryFile ? 0o755 : 0o644; +} + +function sha256(bytes: Uint8Array): string { + return createHash("sha256").update(bytes).digest("hex"); +} + +function sameBytes(left: Uint8Array, right: Uint8Array): boolean { + return left.byteLength === right.byteLength + && left.every((byte, index) => byte === right[index]); +} + +function sameStrings(left: readonly string[], right: readonly string[]): boolean { + return left.length === right.length + && left.every((value, index) => value === right[index]); +} + +function zeroBytes(bytes: Uint8Array, offset: number, length: number): boolean { + for (let index = offset; index < offset + length; index += 1) { + if (bytes[index] !== 0) { + return false; + } + } + + return true; +} + +function invalid(message: string): never { + throw new TypeError(message); +} diff --git a/src/application/commands/flow-release.ts b/src/application/commands/flow-release.ts new file mode 100644 index 0000000..ed4cd08 --- /dev/null +++ b/src/application/commands/flow-release.ts @@ -0,0 +1,23 @@ +export interface OpenFlowCommandRelease { + readonly archive: { + readonly digest: string; + readonly length: number; + readonly url: string; + }; + readonly bunVersion: string; + readonly format: "open-flow-command-release"; + readonly openFlowVersion: string; + readonly version: 1; +} + +export const openFlowCommandRelease = { + archive: { + digest: "cc9b77573f04dbf1e936e2785fe210db2f64cda03267bfd1cd31a799ac6bbcac", + length: 5_006_561, + url: "https://static.oomol.com/release/apps/open-flow/command/open-flow-0.0.4-dev-cc9b77573f04dbf1e936e2785fe210db2f64cda03267bfd1cd31a799ac6bbcac.tar.gz", + }, + bunVersion: "1.3.14", + format: "open-flow-command-release", + openFlowVersion: "0.0.4-dev", + version: 1, +} as const satisfies OpenFlowCommandRelease; diff --git a/src/application/commands/flow.cli.test.ts b/src/application/commands/flow.cli.test.ts index 2cfa032..824f120 100644 --- a/src/application/commands/flow.cli.test.ts +++ b/src/application/commands/flow.cli.test.ts @@ -6,8 +6,12 @@ import { createCliSandbox, createTemporaryDirectory, readLatestLogContent, + toRequest, + writeAuthFile, } from "../../../__tests__/helpers.ts"; +import { openFlowCommandRelease } from "./flow-release.ts"; import { resolveOpenFlowInvocation } from "./flow.ts"; +import { formatByteCount } from "./shared/download-progress.ts"; describe("flow CLI", () => { test("recognizes flow after oo global options without consuming delegated options", () => { @@ -29,17 +33,19 @@ describe("flow CLI", () => { const sandbox = await createCliSandbox(); const commandDirectory = await createTemporaryDirectory("oo-open-flow-command"); const captureKey = `open-flow-args-${Bun.randomUUIDv7()}`; + const languageKey = `open-flow-language-${Bun.randomUUIDv7()}`; try { await writeCommandEntry(commandDirectory, [ `Reflect.set(globalThis, ${JSON.stringify(captureKey)}, [...args]);`, + `Reflect.set(globalThis, ${JSON.stringify(languageKey)}, host.language);`, "return 7;", ]); sandbox.env.OO_OPEN_FLOW_COMMAND_DIR = commandDirectory; const result = await sandbox.run([ "--lang", - "en", + "zh", "flow", "run", "project.oo.yaml", @@ -58,6 +64,125 @@ describe("flow CLI", () => { "--connector-token", "secret-token", ]); + expect(Reflect.get(globalThis, languageKey)).toBe("zh-CN"); + } + finally { + Reflect.deleteProperty(globalThis, captureKey); + Reflect.deleteProperty(globalThis, languageKey); + await Promise.all([ + sandbox.cleanup(), + rm(commandDirectory, { force: true, recursive: true }), + ]); + } + }); + + test("connects Cloud requests to the endpoint gateway with the active credential and team", async () => { + const sandbox = await createCliSandbox(); + const commandDirectory = await createTemporaryDirectory("oo-open-flow-command"); + + try { + await writeAuthFile(sandbox, { + accounts: [ + { + id: "user-1", + name: "Alice", + apiKey: "dev-secret", + endpoint: "oomol.com", + team: "platform", + teamId: "team-1", + }, + ], + }); + await writeCommandEntry(commandDirectory, [ + "const cloudResponse = await host.cloudRequest('/v1/projects?limit=10', {", + " method: 'POST',", + " headers: { authorization: 'artifact-secret', 'content-type': 'application/json', 'x-oo-team-id': 'forged-team' },", + " body: JSON.stringify({ name: 'Example' }),", + "});", + "const uploadResponse = await host.cloudUpload('https://uploads.example.test/package', {", + " method: 'PUT',", + " headers: { 'x-upload-token': 'upload-secret' },", + " body: 'package-bytes',", + "});", + "return cloudResponse.status === 202 && uploadResponse.status === 201 ? 0 : 9;", + ]); + sandbox.env.OO_ENDPOINT = "oomol.dev"; + sandbox.env.OO_OPEN_FLOW_COMMAND_DIR = commandDirectory; + + const requests: Request[] = []; + const result = await sandbox.run(["flow", "project", "list"], { + fetcher: async (input, init) => { + const request = toRequest(input, init); + + requests.push(request); + + return new Response(null, { + status: request.url.startsWith("https://uploads.example.test/") + ? 201 + : 202, + }); + }, + }); + + expect(result.exitCode).toBe(0); + expect(requests).toHaveLength(2); + expect(requests[0]?.url).toBe( + "https://open-flow.oomol.dev/v1/projects?limit=10", + ); + expect(requests[0]?.method).toBe("POST"); + expect(requests[0]?.headers.get("authorization")).toBe("dev-secret"); + expect(requests[0]?.headers.get("x-oo-team-name")).toBe("platform"); + expect(requests[0]?.headers.get("x-oo-team-id")).toBe("team-1"); + expect(await requests[0]?.text()).toBe("{\"name\":\"Example\"}"); + expect(requests[1]?.url).toBe( + "https://uploads.example.test/package", + ); + expect(requests[1]?.headers.get("authorization")).toBeNull(); + expect(requests[1]?.headers.get("x-oo-team-name")).toBeNull(); + expect(requests[1]?.headers.get("x-oo-team-id")).toBeNull(); + expect(requests[1]?.headers.get("x-upload-token")).toBe( + "upload-secret", + ); + } + finally { + await Promise.all([ + sandbox.cleanup(), + rm(commandDirectory, { force: true, recursive: true }), + ]); + } + }); + + test("rejects Cloud control requests outside the configured gateway", async () => { + const sandbox = await createCliSandbox(); + const commandDirectory = await createTemporaryDirectory("oo-open-flow-command"); + const captureKey = `open-flow-cloud-error-${Bun.randomUUIDv7()}`; + + try { + await writeAuthFile(sandbox); + await writeCommandEntry(commandDirectory, [ + "try {", + " await host.cloudRequest('https://attacker.example/v1/projects', { method: 'GET' });", + " return 9;", + "} catch (error) {", + ` Reflect.set(globalThis, ${JSON.stringify(captureKey)}, error instanceof Error ? error.message : String(error));`, + " return 0;", + "}", + ]); + sandbox.env.OO_OPEN_FLOW_COMMAND_DIR = commandDirectory; + + let requestCount = 0; + const result = await sandbox.run(["flow", "project", "list"], { + fetcher: async () => { + requestCount += 1; + return new Response(null, { status: 200 }); + }, + }); + + expect(result.exitCode).toBe(0); + expect(requestCount).toBe(0); + expect(Reflect.get(globalThis, captureKey)).toBe( + "Open Flow Cloud requests must target the configured /v1/ gateway.", + ); } finally { Reflect.deleteProperty(globalThis, captureKey); @@ -98,15 +223,42 @@ describe("flow CLI", () => { } }); - test("explains how to configure the local command directory", async () => { + test("reports when the pinned Open Flow release cannot be downloaded", async () => { const sandbox = await createCliSandbox(); try { - const result = await sandbox.run(["flow", "--help"]); + const result = await sandbox.run(["flow", "--help"], { + fetcher: () => Promise.resolve(new Response(null, { status: 404 })), + }); expect(result.exitCode).toBe(1); expect(result.stdout).toBe(""); - expect(result.stderr).toContain("OO_OPEN_FLOW_COMMAND_DIR"); + expect(result.stderr).toContain( + `Downloading Open Flow ${openFlowCommandRelease.openFlowVersion}...`, + ); + expect(result.stderr).toContain( + `Open Flow ${openFlowCommandRelease.openFlowVersion} could not be downloaded or verified.`, + ); + } + finally { + await sandbox.cleanup(); + } + }); + + test("renders byte progress while downloading in an interactive terminal", async () => { + const sandbox = await createCliSandbox(); + + try { + const result = await sandbox.run(["flow", "--help"], { + fetcher: () => Promise.resolve(new Response(null, { status: 404 })), + stderr: { isTTY: true }, + }); + + expect(result.exitCode).toBe(1); + expect(result.stdout).toBe(""); + expect(result.stderr).toContain( + `Downloading Open Flow ${openFlowCommandRelease.openFlowVersion}: 0 B / ${formatByteCount(openFlowCommandRelease.archive.length)} (0%)`, + ); } finally { await sandbox.cleanup(); @@ -137,15 +289,34 @@ describe("flow CLI", () => { } }); - test("lists flow in root help and provides host-side help", async () => { + test("shows flow in root help only for the online dev endpoint", async () => { + const sandbox = await createCliSandbox(); + + try { + const defaultHelp = await sandbox.run(["--help"]); + sandbox.env.OO_ENDPOINT = "oomol.com"; + const productionHelp = await sandbox.run(["--help"]); + sandbox.env.OO_ENDPOINT = " oomol.dev "; + const devHelp = await sandbox.run(["--help"]); + + expect(defaultHelp.exitCode).toBe(0); + expect(defaultHelp.stdout).not.toContain(" flow [args...]"); + expect(productionHelp.exitCode).toBe(0); + expect(productionHelp.stdout).not.toContain(" flow [args...]"); + expect(devHelp.exitCode).toBe(0); + expect(devHelp.stdout).toContain(" flow [args...]"); + } + finally { + await sandbox.cleanup(); + } + }); + + test("provides host-side flow help while the command is hidden", async () => { const sandbox = await createCliSandbox(); try { - const rootHelp = await sandbox.run(["--help"]); const flowHelp = await sandbox.run(["help", "flow"]); - expect(rootHelp.exitCode).toBe(0); - expect(rootHelp.stdout).toContain("flow"); expect(flowHelp.exitCode).toBe(0); expect(flowHelp.stdout).toContain("Arguments passed to Open Flow"); } @@ -166,7 +337,7 @@ async function writeCommandEntry( [ "export const commandArtifactVersion = 1;", `export const requiredBunVersion = ${JSON.stringify(requiredBunVersion)};`, - "export async function runOpenFlowCommand(args) {", + "export async function runOpenFlowCommand(args, host) {", ...body, "}", "", diff --git a/src/application/commands/flow.ts b/src/application/commands/flow.ts index ae1e9db..7387dbb 100644 --- a/src/application/commands/flow.ts +++ b/src/application/commands/flow.ts @@ -5,7 +5,18 @@ import type { import { join, resolve } from "node:path"; import { pathToFileURL } from "node:url"; +import { resolveRequestLanguage } from "../../i18n/locale.ts"; +import { readDefaultTeam } from "../auth/default-team.ts"; +import { requireIdentity } from "../auth/identity.ts"; import { CliUserError } from "../contracts/cli.ts"; +import { installOpenFlowCommandRelease } from "./flow-artifact.ts"; +import { openFlowCommandRelease } from "./flow-release.ts"; +import { createDownloadProgressReporter } from "./shared/download-progress.ts"; +import { + requireValidTeamIdentity, + resolveTeamIdentity, + teamIdentityHeaders, +} from "./team/identity.ts"; const commandDirectoryEnvName = "OO_OPEN_FLOW_COMMAND_DIR"; const commandArtifactVersion = 1; @@ -21,6 +32,18 @@ interface OpenFlowInvocation { readonly commandIndex: number; } +interface OpenFlowCloudSession { + readonly authorization: string; + readonly origin: URL; + readonly teamHeaders: Record; +} + +interface OpenFlowCommandHost { + readonly cloudRequest: (path: string, init: RequestInit) => Promise; + readonly cloudUpload: CliExecutionContext["fetcher"]; + readonly language: "en" | "zh-CN"; +} + export const flowCommand = { name: "flow", summaryKey: "commands.flow.summary", @@ -68,15 +91,92 @@ export function resolveOpenFlowInvocation( export async function runOpenFlowCommand( args: readonly string[], - context: Pick, + context: Pick< + CliExecutionContext, + | "authStore" + | "connectorStore" + | "cwd" + | "env" + | "execPath" + | "fetcher" + | "logger" + | "settingsStore" + | "stderr" + | "translator" + >, ): Promise { const configuredDirectory = context.env[commandDirectoryEnvName]?.trim(); + let commandDirectory: string; + + if (configuredDirectory) { + commandDirectory = resolve(context.cwd, configuredDirectory); + } + else { + if (openFlowCommandRelease.bunVersion !== Bun.version) { + throw new CliUserError("errors.flow.bunVersionMismatch", 1, { + actual: Bun.version, + required: openFlowCommandRelease.bunVersion, + }); + } - if (!configuredDirectory) { - throw new CliUserError("errors.flow.commandDirectoryRequired", 1); + const progressReporter = createDownloadProgressReporter( + context.stderr, + openFlowCommandRelease.archive.length, + `Open Flow ${openFlowCommandRelease.openFlowVersion}`, + ); + let downloadStarted = false; + let downloadedBytes = 0; + + try { + commandDirectory = await installOpenFlowCommandRelease( + openFlowCommandRelease, + { + env: context.env, + execPath: context.execPath, + fetcher: context.fetcher, + onDownloadProgress(nextDownloadedBytes) { + if (!downloadStarted && progressReporter === undefined) { + context.stderr.write(`${context.translator.t( + "flow.download.start", + { version: openFlowCommandRelease.openFlowVersion }, + )}\n`); + } + + downloadStarted = true; + downloadedBytes = nextDownloadedBytes; + progressReporter?.render(downloadedBytes); + }, + }, + ); + + if (downloadStarted) { + if (progressReporter === undefined) { + context.stderr.write(`${context.translator.t( + "flow.download.complete", + { version: openFlowCommandRelease.openFlowVersion }, + )}\n`); + } + else { + progressReporter.complete(downloadedBytes); + } + } + } + catch (error) { + if (downloadStarted) { + progressReporter?.finish(downloadedBytes); + } + + context.logger.debug( + { err: error }, + "Open Flow command artifact preparation failed.", + ); + throw new CliUserError("errors.flow.commandArtifactUnavailable", 1, { + version: openFlowCommandRelease.openFlowVersion, + }); + } } - const entryPath = join(resolve(context.cwd, configuredDirectory), "entry.js"); + const entryPath = join(commandDirectory, "entry.js"); let loaded: unknown; try { @@ -114,7 +214,39 @@ export async function runOpenFlowCommand( }); } - const exitCode = await commandModule.runOpenFlowCommand(args); + let cloudSession: Promise | undefined; + const host: OpenFlowCommandHost = { + async cloudRequest(path, init) { + cloudSession ??= resolveOpenFlowCloudSession(context); + const session = await cloudSession; + const url = new URL(path, session.origin); + + if ( + url.origin !== session.origin.origin + || !url.pathname.startsWith("/v1/") + || url.username !== "" + || url.password !== "" + || url.hash !== "" + ) { + throw new TypeError( + "Open Flow Cloud requests must target the configured /v1/ gateway.", + ); + } + + const headers = new Headers(init.headers); + + headers.set("authorization", session.authorization); + + for (const [name, value] of Object.entries(session.teamHeaders)) { + headers.set(name, value); + } + + return await context.fetcher(url, { ...init, headers }); + }, + cloudUpload: context.fetcher, + language: resolveRequestLanguage(context.translator.locale), + }; + const exitCode = await commandModule.runOpenFlowCommand(args, host); if ( typeof exitCode !== "number" @@ -129,3 +261,35 @@ export async function runOpenFlowCommand( return exitCode; } + +async function resolveOpenFlowCloudSession( + context: Pick< + CliExecutionContext, + | "authStore" + | "connectorStore" + | "env" + | "fetcher" + | "logger" + | "settingsStore" + | "translator" + >, +): Promise { + const { account } = await requireIdentity(context); + const identity = requireValidTeamIdentity( + await resolveTeamIdentity( + { + account, + defaultTeam: await readDefaultTeam(context), + resolveAgainstBackend: true, + }, + context, + ), + context, + ); + + return { + authorization: account.apiKey, + origin: new URL(`https://open-flow.${account.endpoint}`), + teamHeaders: teamIdentityHeaders(identity), + }; +} diff --git a/src/application/commands/file/download/progress.test.ts b/src/application/commands/shared/download-progress.test.ts similarity index 71% rename from src/application/commands/file/download/progress.test.ts rename to src/application/commands/shared/download-progress.test.ts index 9c76804..079d9f4 100644 --- a/src/application/commands/file/download/progress.test.ts +++ b/src/application/commands/shared/download-progress.test.ts @@ -1,7 +1,7 @@ import { describe, expect, test } from "bun:test"; -import { createTextBuffer } from "../../../../../__tests__/helpers.ts"; -import { createDownloadProgressReporter, formatByteCount } from "./progress.ts"; +import { createTextBuffer } from "../../../../__tests__/helpers.ts"; +import { createDownloadProgressReporter, formatByteCount } from "./download-progress.ts"; describe("formatByteCount", () => { test("keeps byte-sized values in bytes", () => { @@ -44,4 +44,23 @@ describe("createDownloadProgressReporter", () => { + "\u001B[1A\r\u001B[2KDownloaded 4 B / 4 B (100%)\n", ); }); + + test("identifies a named download", () => { + const stderr = createTextBuffer({ + isTTY: true, + }); + const reporter = createDownloadProgressReporter( + stderr.writer, + 4, + "Open Flow 1.2.3", + ); + + reporter!.render(1); + reporter!.complete(4); + + expect(stderr.read()).toBe( + "Downloading Open Flow 1.2.3: 1 B / 4 B (25%)\n" + + "\u001B[1A\r\u001B[2KDownloaded Open Flow 1.2.3: 4 B / 4 B (100%)\n", + ); + }); }); diff --git a/src/application/commands/file/download/progress.ts b/src/application/commands/shared/download-progress.ts similarity index 76% rename from src/application/commands/file/download/progress.ts rename to src/application/commands/shared/download-progress.ts index 2cf0196..b3148e5 100644 --- a/src/application/commands/file/download/progress.ts +++ b/src/application/commands/shared/download-progress.ts @@ -1,19 +1,20 @@ -import type { CliExecutionContext } from "../../../contracts/cli.ts"; +import type { CliExecutionContext } from "../../contracts/cli.ts"; import { moveCursorUp, rewriteTerminalLine, -} from "../../../terminal-control.ts"; +} from "../../terminal-control.ts"; export function createDownloadProgressReporter( writer: CliExecutionContext["stderr"], totalBytes: number | undefined, + subject?: string, ): DownloadProgressReporter | undefined { if (writer.isTTY !== true) { return undefined; } - return new DownloadProgressReporter(writer, totalBytes); + return new DownloadProgressReporter(writer, totalBytes, subject); } export class DownloadProgressReporter { @@ -25,6 +26,7 @@ export class DownloadProgressReporter { constructor( private readonly writer: Pick, private readonly totalBytes: number | undefined, + private readonly subject?: string, ) {} render(downloadedBytes: number): void { @@ -44,7 +46,12 @@ export class DownloadProgressReporter { this.lastRenderedBytes = downloadedBytes; this.lastRenderedAt = now; this.writeProgressLine( - formatProgressStatusLine("Downloading", downloadedBytes, this.totalBytes), + formatProgressStatusLine( + "Downloading", + downloadedBytes, + this.totalBytes, + this.subject, + ), ); } @@ -52,7 +59,12 @@ export class DownloadProgressReporter { this.lastRenderedBytes = downloadedBytes; this.lastRenderedAt = Date.now(); this.writeProgressLine( - formatProgressStatusLine("Downloading", downloadedBytes, this.totalBytes), + formatProgressStatusLine( + "Downloading", + downloadedBytes, + this.totalBytes, + this.subject, + ), ); } @@ -60,7 +72,12 @@ export class DownloadProgressReporter { this.lastRenderedBytes = downloadedBytes; this.lastRenderedAt = Date.now(); this.writeProgressLine( - formatProgressStatusLine("Downloaded", downloadedBytes, this.totalBytes), + formatProgressStatusLine( + "Downloaded", + downloadedBytes, + this.totalBytes, + this.subject, + ), ); } @@ -85,9 +102,12 @@ function formatProgressStatusLine( status: "Downloaded" | "Downloading", downloadedBytes: number, totalBytes: number | undefined, + subject: string | undefined, ): string { + const prefix = subject === undefined ? status : `${status} ${subject}:`; + if (totalBytes === undefined) { - return `${status} ${formatByteCount(downloadedBytes)}`; + return `${prefix} ${formatByteCount(downloadedBytes)}`; } const percent = totalBytes === 0 @@ -95,7 +115,7 @@ function formatProgressStatusLine( : Math.max(0, Math.min(100, Math.round((downloadedBytes / totalBytes) * 100))); return [ - status, + prefix, formatByteCount(downloadedBytes), "/", formatByteCount(totalBytes), diff --git a/src/application/commands/team/identity.ts b/src/application/commands/team/identity.ts index 56af42e..58110ef 100644 --- a/src/application/commands/team/identity.ts +++ b/src/application/commands/team/identity.ts @@ -176,6 +176,27 @@ export function teamNameStatusForTelemetry( return identity?.status ?? "none"; } +// Builds the standard team identity headers for OOMOL service requests. +export function teamIdentityHeaders( + identity: TeamIdentity | undefined, +): Record { + const headers: Record = {}; + + if (identity === undefined) { + return headers; + } + + if (identity.name !== null) { + headers["x-oo-team-name"] = identity.name; + } + + if (identity.id !== null) { + headers["x-oo-team-id"] = identity.id; + } + + return headers; +} + // Renders the identity for humans: the name with its id in parentheses when // both are known, otherwise whichever one is. export function formatTeamIdentityValue( diff --git a/src/application/commands/telemetry-decisions.test.ts b/src/application/commands/telemetry-decisions.test.ts index 3e38546..60dac05 100644 --- a/src/application/commands/telemetry-decisions.test.ts +++ b/src/application/commands/telemetry-decisions.test.ts @@ -242,7 +242,7 @@ const commandTelemetryDecisions = { }, "flow": { kind: "generic", - reason: "Generic command telemetry records only the delegated flow command and its exit code; Open Flow arguments, flags, paths, project identities, and tokens are not inspected.", + reason: "Generic command telemetry records only the delegated flow command and its exit code; Open Flow arguments, flags, paths, project, account, and team identities, and tokens are not inspected.", }, "info": { kind: "generic", diff --git a/src/i18n/catalog.ts b/src/i18n/catalog.ts index d3eabc1..b8bdd92 100644 --- a/src/i18n/catalog.ts +++ b/src/i18n/catalog.ts @@ -138,8 +138,10 @@ export const enMessages = { "Run the Open Flow CLI with all following arguments passed through unchanged.", "commands.flow.summary": "Run Open Flow", "arguments.flowArgs": "Arguments passed to Open Flow", - "errors.flow.commandDirectoryRequired": - "Open Flow is not available locally. Set OO_OPEN_FLOW_COMMAND_DIR to the built Open Flow command directory.", + "flow.download.complete": "Downloaded and verified Open Flow {version}.", + "flow.download.start": "Downloading Open Flow {version}...", + "errors.flow.commandArtifactUnavailable": + "Open Flow {version} could not be downloaded or verified. Check the network connection and local cache, then try again.", "errors.flow.commandEntryInvalid": "The Open Flow command entry at {path} is invalid.", "errors.flow.commandEntryLoadFailed": @@ -1536,8 +1538,10 @@ export const zhMessages = { "commands.flow.description": "运行 Open Flow CLI，并将后续参数原样传递给它。", "commands.flow.summary": "运行 Open Flow", "arguments.flowArgs": "传递给 Open Flow 的参数", - "errors.flow.commandDirectoryRequired": - "本地 Open Flow 尚不可用。请将 OO_OPEN_FLOW_COMMAND_DIR 设置为构建后的 Open Flow 命令目录。", + "flow.download.complete": "已下载并验证 Open Flow {version}。", + "flow.download.start": "正在下载 Open Flow {version}…", + "errors.flow.commandArtifactUnavailable": + "无法下载或验证 Open Flow {version}。请检查网络连接和本地缓存后重试。", "errors.flow.commandEntryInvalid": "{path} 中的 Open Flow 命令入口无效。", "errors.flow.commandEntryLoadFailed": "无法加载 {path} 中的 Open Flow 命令入口：{message}", From 8e45c659a6126aaf4db7c10de418f0f32d2c4a82 Mon Sep 17 00:00:00 2001 From: l1shen <648952316@qq.com> Date: Thu, 6 Aug 2026 12:01:01 +0800 Subject: [PATCH 3/6] fix: address Open Flow review feedback --- .../commands/flow-artifact.test.ts | 73 +++++++++++++++++-- src/application/commands/flow-artifact.ts | 42 +++++------ .../commands/uninstall.cli.test.ts | 8 ++ src/application/self-update/uninstall.test.ts | 7 ++ src/application/self-update/uninstall.ts | 19 ++++- 5 files changed, 118 insertions(+), 31 deletions(-) diff --git a/src/application/commands/flow-artifact.test.ts b/src/application/commands/flow-artifact.test.ts index 14d1c0b..40ae54c 100644 --- a/src/application/commands/flow-artifact.test.ts +++ b/src/application/commands/flow-artifact.test.ts @@ -133,6 +133,62 @@ describe("Open Flow command artifact", () => { })).rejects.toThrow("link, directory, device, metadata, or other non-file"); }); + for (const path of [ + "open-flow-command/../escape.js", + "open-flow-command//escape.js", + ]) { + test(`rejects unsafe tar path ${JSON.stringify(path)}`, async () => { + const archive = encodeTarGzip([{ + body: new Uint8Array(), + mode: 0o644, + path, + type: "0", + }]); + const release = createRelease(archive); + const environment = await createTestEnvironment(); + + await expect(installOpenFlowCommandRelease(release, { + env: environment.env, + execPath: process.execPath, + fetcher: createArchiveFetcher(archive), + })).rejects.toThrow("invalid file path"); + }); + } + + test("guards archive downloads with a timeout signal", async () => { + const fixture = createCommandReleaseFixture(); + const environment = await createTestEnvironment(); + let signal: AbortSignal | null | undefined; + + await installOpenFlowCommandRelease(fixture.release, { + env: environment.env, + execPath: process.execPath, + fetcher: createArchiveFetcher(fixture.archive, (init) => { + signal = init?.signal; + }), + }); + + expect(signal).toBeInstanceOf(AbortSignal); + expect(signal?.aborted).toBe(false); + }); + + test("accepts a valid archive from a different gzip compressor level", async () => { + const fixture = createCommandReleaseFixture(0); + const environment = await createTestEnvironment(); + + const commandDirectory = await installOpenFlowCommandRelease( + fixture.release, + { + env: environment.env, + execPath: process.execPath, + fetcher: createArchiveFetcher(fixture.archive), + }, + ); + + expect(await readFile(join(commandDirectory, "entry.js"), "utf8")) + .toBe(fixture.entrySource); + }); + test("serializes concurrent installation of the same digest", async () => { const fixture = createCommandReleaseFixture(); const environment = await createTestEnvironment(); @@ -157,6 +213,9 @@ describe("Open Flow command artifact", () => { const first = installOpenFlowCommandRelease(fixture.release, options); await requestStarted.promise; const second = installOpenFlowCommandRelease(fixture.release, options); + await Bun.sleep(150); + + expect(requestCount).toBe(1); resumeRequest.resolve(); const directories = await Promise.all([first, second]); @@ -166,7 +225,7 @@ describe("Open Flow command artifact", () => { }); }); -function createCommandReleaseFixture(): { +function createCommandReleaseFixture(compressionLevel = 9): { archive: Uint8Array; entrySource: string; release: OpenFlowCommandRelease; @@ -212,7 +271,7 @@ function createCommandReleaseFixture(): { path: `open-flow-command/${file.path}`, type: "0" as const, })), - ]); + ], compressionLevel); return { archive, @@ -240,10 +299,10 @@ function createRelease(archive: Uint8Array): OpenFlowCommandRelease { function createArchiveFetcher( archive: Uint8Array, - onRequest: () => void = () => {}, + onRequest: (init: RequestInit | undefined) => void = () => {}, ): Fetcher { - return () => { - onRequest(); + return (_input, init) => { + onRequest(init); return Promise.resolve(new Response(archive, { headers: { "content-length": String(archive.byteLength), @@ -273,7 +332,7 @@ function encodeTarGzip(entries: readonly { mode: number; path: string; type: "0" | "2"; -}[]): Uint8Array { +}[], compressionLevel = 9): Uint8Array { const chunks: Uint8Array[] = []; for (const entry of entries) { @@ -308,7 +367,7 @@ function encodeTarGzip(entries: readonly { offset += chunk.byteLength; } - const archive = new Uint8Array(gzipSync(tar, { level: 9 })); + const archive = new Uint8Array(gzipSync(tar, { level: compressionLevel })); archive[3] = 0; archive[4] = 0; archive[5] = 0; diff --git a/src/application/commands/flow-artifact.ts b/src/application/commands/flow-artifact.ts index 732b991..f3e19d1 100644 --- a/src/application/commands/flow-artifact.ts +++ b/src/application/commands/flow-artifact.ts @@ -5,7 +5,7 @@ import { createHash } from "node:crypto"; import { mkdir, open, readdir, readFile, rename, rm, writeFile } from "node:fs/promises"; import { basename, dirname, join } from "node:path"; import process from "node:process"; -import { gunzipSync, gzipSync } from "node:zlib"; +import { gunzipSync } from "node:zlib"; import { z } from "zod"; import { resolveHomeDirectory } from "../path/home-directory.ts"; import { acquireDownloadTempLock } from "../shared/download-temp-lock.ts"; @@ -26,6 +26,7 @@ const tarEndMarkerSize = tarBlockSize * 2; const gzipHeaderSize = 10; const gzipFooterSize = 8; const lockWaitTimeoutMs = 300_000; +const commandArchiveDownloadTimeoutMs = 300_000; const textDecoder = new TextDecoder("utf-8", { fatal: true, ignoreBOM: false, @@ -127,7 +128,10 @@ export async function installOpenFlowCommandRelease( invalid(`Open Flow requires Bun ${release.bunVersion}; received ${Bun.version}.`); } - const cacheRoot = resolveCommandCacheRoot(options.env); + const cacheRoot = resolveOpenFlowCommandCacheRoot({ + env: options.env, + platform: process.platform, + }); const commandDirectory = join(cacheRoot, release.archive.digest); if (await validCommandArtifactDirectory(commandDirectory, release)) { @@ -220,6 +224,7 @@ async function downloadCommandArchive( headers: { accept: commandArchiveMediaType, }, + signal: AbortSignal.timeout(commandArchiveDownloadTimeoutMs), }); if (!response.ok) { @@ -360,25 +365,9 @@ function decodeCanonicalGzip(archive: Uint8Array): Uint8Array { }); } - if (!sameBytes(canonicalGzip(tar), archive)) { - invalid("Command archive gzip stream is truncated, has trailing data, or is not canonically encoded."); - } - return tar; } -function canonicalGzip(tar: Uint8Array): Uint8Array { - const compressed = new Uint8Array(gzipSync(tar, { level: 9 })); - compressed[3] = 0; - compressed[4] = 0; - compressed[5] = 0; - compressed[6] = 0; - compressed[7] = 0; - compressed[8] = 2; - compressed[9] = 255; - return compressed; -} - function decodeTar(tar: Uint8Array): readonly CommandArchiveEntry[] { if (tar.byteLength < tarEndMarkerSize || tar.byteLength % tarBlockSize !== 0) { invalid("Command archive is not a complete block-aligned tar stream."); @@ -729,20 +718,27 @@ function validateFile(bytes: Uint8Array, expected: CommandArtifactFile): void { } } -function resolveCommandCacheRoot(env: Record): string { - const homeDirectory = resolveHomeDirectory(env); +export function resolveOpenFlowCommandCacheRoot(options: { + env: Record; + homeDirectory?: string; + platform: NodeJS.Platform; +}): string { + const homeDirectory = resolveHomeDirectory( + options.env, + options.homeDirectory, + ); let platformCacheRoot: string; - switch (process.platform) { + switch (options.platform) { case "darwin": platformCacheRoot = join(homeDirectory, "Library", "Caches"); break; case "win32": - platformCacheRoot = env.LOCALAPPDATA + platformCacheRoot = options.env.LOCALAPPDATA ?? join(homeDirectory, "AppData", "Local"); break; default: - platformCacheRoot = env.XDG_CACHE_HOME + platformCacheRoot = options.env.XDG_CACHE_HOME ?? join(homeDirectory, ".cache"); } diff --git a/src/application/commands/uninstall.cli.test.ts b/src/application/commands/uninstall.cli.test.ts index af6bde7..1ba7d4a 100644 --- a/src/application/commands/uninstall.cli.test.ts +++ b/src/application/commands/uninstall.cli.test.ts @@ -9,6 +9,7 @@ import { resolveStorePaths } from "../../adapters/store/store-path.ts"; import { APP_NAME } from "../config/app-config.ts"; import { resolveSelfUpdatePaths } from "../self-update/paths.ts"; import { pathExists } from "../shared/fs-utils.ts"; +import { resolveOpenFlowCommandCacheRoot } from "./flow-artifact.ts"; import { resolveManagedSkillMetadataFilePath } from "./skills/managed-skill-paths.ts"; import { createBundledSkillMetadata, @@ -185,6 +186,10 @@ describe("oo uninstall", () => { try { await seedRuntime(sandbox); const store = storePaths(sandbox); + const openFlowCommandCacheRoot = resolveOpenFlowCommandCacheRoot({ + env: sandbox.env, + platform: process.platform, + }); await mkdir(store.rootDirectory, { recursive: true }); await writeFile(store.authFilePath, "id = \"\"\n"); @@ -192,6 +197,8 @@ describe("oo uninstall", () => { // A residual file that no explicit child item targets: it must still // be gone because --purge removes the whole config root. await writeFile(join(store.rootDirectory, "leftover.txt"), "x"); + await mkdir(openFlowCommandCacheRoot, { recursive: true }); + await writeFile(join(openFlowCommandCacheRoot, "entry.js"), "cached"); const registrySkill = await seedHostSkill({ sandbox, skillName: "demo", @@ -216,6 +223,7 @@ describe("oo uninstall", () => { // every platform. expect(await Bun.file(store.authFilePath).exists()).toBe(false); expect(await Bun.file(store.settingsFilePath).exists()).toBe(false); + expect(await pathExists(openFlowCommandCacheRoot)).toBe(false); // All registry skills removed under purge expect(await Bun.file(join(registrySkill, "SKILL.md")).exists()).toBe(false); // Local still retained even under purge diff --git a/src/application/self-update/uninstall.test.ts b/src/application/self-update/uninstall.test.ts index 65fd914..2fd024e 100644 --- a/src/application/self-update/uninstall.test.ts +++ b/src/application/self-update/uninstall.test.ts @@ -200,6 +200,13 @@ describe("buildSelfUninstallPlan", () => { expect(userData.length).toBeGreaterThanOrEqual(4); expect(paths(userData).some(path => path.endsWith("auth.toml"))).toBe(true); expect(paths(userData).some(path => path.endsWith("settings.toml"))).toBe(true); + expect(paths(userData)).toContain(join( + tempHome, + ".cache", + "oo", + "open-flow", + "command-artifact-v1", + )); // The config root itself is removed, and it must be the last user-data // item so it sweeps anything the explicit child items did not cover. diff --git a/src/application/self-update/uninstall.ts b/src/application/self-update/uninstall.ts index 813fd3e..fae2ae9 100644 --- a/src/application/self-update/uninstall.ts +++ b/src/application/self-update/uninstall.ts @@ -6,6 +6,7 @@ import type { SelfUpdatePaths } from "./paths.ts"; import { mkdir, rm } from "node:fs/promises"; import { dirname, join } from "node:path"; import { resolveStorePaths } from "../../adapters/store/store-path.ts"; +import { resolveOpenFlowCommandCacheRoot } from "../commands/flow-artifact.ts"; import { canonicalBundledSkillsDirectoryName, managedSkillsDirectoryName, @@ -147,7 +148,17 @@ export async function buildSelfUninstallPlan( }); if (options.purge) { - addUserDataItems({ deferred, immediate, isWindows, storePaths }); + addUserDataItems({ + deferred, + immediate, + isWindows, + openFlowCommandCacheRoot: resolveOpenFlowCommandCacheRoot({ + env: options.env, + homeDirectory: options.homeDirectory, + platform: options.platform, + }), + storePaths, + }); } return { @@ -300,6 +311,7 @@ function addUserDataItems(args: { deferred: UninstallPlanItem[]; immediate: UninstallPlanItem[]; isWindows: boolean; + openFlowCommandCacheRoot: string; storePaths: ReturnType; }): void { args.immediate.push( @@ -327,6 +339,11 @@ function addUserDataItems(args: { }); args.immediate.push( + { + category: "user-data", + label: "Open Flow command cache", + path: args.openFlowCommandCacheRoot, + }, { category: "user-data", label: "Telemetry", From 02a324f69f4cee88c56ce1553f9384ec0cf6e629 Mon Sep 17 00:00:00 2001 From: l1shen <648952316@qq.com> Date: Thu, 6 Aug 2026 13:03:10 +0800 Subject: [PATCH 4/6] test: exercise Open Flow download timeout --- .../commands/flow-artifact.test.ts | 50 +++++++++++++++---- 1 file changed, 39 insertions(+), 11 deletions(-) diff --git a/src/application/commands/flow-artifact.test.ts b/src/application/commands/flow-artifact.test.ts index 40ae54c..e7f11f7 100644 --- a/src/application/commands/flow-artifact.test.ts +++ b/src/application/commands/flow-artifact.test.ts @@ -2,18 +2,23 @@ import type { Fetcher } from "../contracts/cli.ts"; import type { OpenFlowCommandRelease } from "./flow-release.ts"; import { createHash } from "node:crypto"; -import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; +import { mkdtemp, readdir, readFile, rm, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { gzipSync } from "node:zlib"; -import { afterEach, describe, expect, test } from "bun:test"; -import { installOpenFlowCommandRelease } from "./flow-artifact.ts"; +import { afterEach, describe, expect, jest, test } from "bun:test"; +import { requireAbortSignal } from "../../../__tests__/helpers.ts"; +import { + installOpenFlowCommandRelease, + resolveOpenFlowCommandCacheRoot, +} from "./flow-artifact.ts"; import { openFlowCommandRelease } from "./flow-release.ts"; const temporaryDirectories = new Set(); const textEncoder = new TextEncoder(); afterEach(async () => { + jest.useRealTimers(); await Promise.all(Array.from(temporaryDirectories, path => rm(path, { force: true, recursive: true }))); temporaryDirectories.clear(); @@ -155,21 +160,44 @@ describe("Open Flow command artifact", () => { }); } - test("guards archive downloads with a timeout signal", async () => { + test("aborts timed out archive downloads and removes temporary files", async () => { + jest.useFakeTimers(); const fixture = createCommandReleaseFixture(); const environment = await createTestEnvironment(); - let signal: AbortSignal | null | undefined; + const requestStarted = Promise.withResolvers(); - await installOpenFlowCommandRelease(fixture.release, { + const installation = installOpenFlowCommandRelease(fixture.release, { env: environment.env, execPath: process.execPath, - fetcher: createArchiveFetcher(fixture.archive, (init) => { - signal = init?.signal; - }), + fetcher: (_input, init) => { + const signal = requireAbortSignal(init); + requestStarted.resolve(signal); + + return new Promise((_resolve, reject) => { + signal.addEventListener("abort", () => { + reject(signal.reason); + }, { once: true }); + }); + }, + }); + const signal = await requestStarted.promise; + + jest.advanceTimersByTime(300_000); + + await expect(installation).rejects.toBe(signal.reason); + expect(signal.aborted).toBe(true); + + const cacheRoot = resolveOpenFlowCommandCacheRoot({ + env: environment.env, + platform: process.platform, }); + const temporaryEntries = (await readdir(cacheRoot, { recursive: true })) + .filter(entry => + entry.startsWith(".archive-") + || entry.startsWith(".extract-") + || entry.endsWith(".lock")); - expect(signal).toBeInstanceOf(AbortSignal); - expect(signal?.aborted).toBe(false); + expect(temporaryEntries).toEqual([]); }); test("accepts a valid archive from a different gzip compressor level", async () => { From 67ce6e21fa459e643ceff61ee1bd62a1a83ac4b3 Mon Sep 17 00:00:00 2001 From: l1shen <648952316@qq.com> Date: Thu, 6 Aug 2026 13:13:19 +0800 Subject: [PATCH 5/6] fix: stabilize Open Flow download timeout --- src/application/commands/flow-artifact.ts | 133 ++++++++++++---------- 1 file changed, 73 insertions(+), 60 deletions(-) diff --git a/src/application/commands/flow-artifact.ts b/src/application/commands/flow-artifact.ts index f3e19d1..ecbe2f7 100644 --- a/src/application/commands/flow-artifact.ts +++ b/src/application/commands/flow-artifact.ts @@ -220,87 +220,100 @@ async function downloadCommandArchive( onProgress: ((downloadedBytes: number) => void) | undefined, ): Promise { onProgress?.(0); - const response = await fetcher(release.archive.url, { - headers: { - accept: commandArchiveMediaType, - }, - signal: AbortSignal.timeout(commandArchiveDownloadTimeoutMs), - }); + const abortController = new AbortController(); + const timeoutId = setTimeout(() => { + abortController.abort(new DOMException( + "The operation was aborted due to timeout", + "TimeoutError", + )); + }, commandArchiveDownloadTimeoutMs); - if (!response.ok) { - invalid(`Open Flow command archive request failed with status ${response.status}.`); - } + try { + const response = await fetcher(release.archive.url, { + headers: { + accept: commandArchiveMediaType, + }, + signal: abortController.signal, + }); - const declaredLength = response.headers.get("content-length"); + if (!response.ok) { + invalid(`Open Flow command archive request failed with status ${response.status}.`); + } - if (declaredLength !== null) { - const parsedLength = Number(declaredLength); + const declaredLength = response.headers.get("content-length"); - if ( - !Number.isSafeInteger(parsedLength) - || parsedLength < 0 - || parsedLength !== release.archive.length - ) { - invalid("Open Flow command archive response length does not match its release record."); + if (declaredLength !== null) { + const parsedLength = Number(declaredLength); + + if ( + !Number.isSafeInteger(parsedLength) + || parsedLength < 0 + || parsedLength !== release.archive.length + ) { + invalid("Open Flow command archive response length does not match its release record."); + } } - } - if (response.body === null) { - invalid("Open Flow command archive response has no body."); - } + if (response.body === null) { + invalid("Open Flow command archive response has no body."); + } - const fileHandle = await open(archivePath, "wx"); - const digest = createHash("sha256"); - const reader = response.body.getReader(); - let length = 0; + const fileHandle = await open(archivePath, "wx"); + const digest = createHash("sha256"); + const reader = response.body.getReader(); + let length = 0; - try { - while (true) { - const chunk = await reader.read(); + try { + while (true) { + const chunk = await reader.read(); - if (chunk.done) { - break; - } + if (chunk.done) { + break; + } - if (length + chunk.value.byteLength > release.archive.length) { - invalid("Open Flow command archive is longer than its release record."); - } + if (length + chunk.value.byteLength > release.archive.length) { + invalid("Open Flow command archive is longer than its release record."); + } + + let written = 0; - let written = 0; + while (written < chunk.value.byteLength) { + const result = await fileHandle.write( + chunk.value, + written, + chunk.value.byteLength - written, + length + written, + ); - while (written < chunk.value.byteLength) { - const result = await fileHandle.write( - chunk.value, - written, - chunk.value.byteLength - written, - length + written, - ); + if (result.bytesWritten === 0) { + invalid("Open Flow command archive download stopped before completion."); + } - if (result.bytesWritten === 0) { - invalid("Open Flow command archive download stopped before completion."); + written += result.bytesWritten; } - written += result.bytesWritten; + digest.update(chunk.value); + length += chunk.value.byteLength; + onProgress?.(length); } - digest.update(chunk.value); - length += chunk.value.byteLength; - onProgress?.(length); + await fileHandle.sync(); + } + finally { + reader.releaseLock(); + await fileHandle.close(); } - await fileHandle.sync(); - } - finally { - reader.releaseLock(); - await fileHandle.close(); - } + if (length !== release.archive.length) { + invalid("Open Flow command archive length does not match its release record."); + } - if (length !== release.archive.length) { - invalid("Open Flow command archive length does not match its release record."); + if (digest.digest("hex") !== release.archive.digest) { + invalid("Open Flow command archive digest does not match its release record."); + } } - - if (digest.digest("hex") !== release.archive.digest) { - invalid("Open Flow command archive digest does not match its release record."); + finally { + clearTimeout(timeoutId); } } From 586e17a9c784c9f1fe3e46ba426e7404ea6411a7 Mon Sep 17 00:00:00 2001 From: l1shen <648952316@qq.com> Date: Thu, 6 Aug 2026 13:25:46 +0800 Subject: [PATCH 6/6] test: isolate Open Flow timeout scheduling --- .../commands/flow-artifact.test.ts | 38 ++++++++++++++++--- src/application/commands/flow-artifact.ts | 21 +++++++++- 2 files changed, 52 insertions(+), 7 deletions(-) diff --git a/src/application/commands/flow-artifact.test.ts b/src/application/commands/flow-artifact.test.ts index e7f11f7..00c69c4 100644 --- a/src/application/commands/flow-artifact.test.ts +++ b/src/application/commands/flow-artifact.test.ts @@ -6,7 +6,7 @@ import { mkdtemp, readdir, readFile, rm, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { gzipSync } from "node:zlib"; -import { afterEach, describe, expect, jest, test } from "bun:test"; +import { afterEach, describe, expect, test } from "bun:test"; import { requireAbortSignal } from "../../../__tests__/helpers.ts"; import { installOpenFlowCommandRelease, @@ -18,7 +18,6 @@ const temporaryDirectories = new Set(); const textEncoder = new TextEncoder(); afterEach(async () => { - jest.useRealTimers(); await Promise.all(Array.from(temporaryDirectories, path => rm(path, { force: true, recursive: true }))); temporaryDirectories.clear(); @@ -161,10 +160,13 @@ describe("Open Flow command artifact", () => { } test("aborts timed out archive downloads and removes temporary files", async () => { - jest.useFakeTimers(); const fixture = createCommandReleaseFixture(); const environment = await createTestEnvironment(); const requestStarted = Promise.withResolvers(); + const timeoutScheduled = Promise.withResolvers<{ + advanceBy: (milliseconds: number) => void; + isCancelled: () => boolean; + }>(); const installation = installOpenFlowCommandRelease(fixture.release, { env: environment.env, @@ -179,13 +181,39 @@ describe("Open Flow command artifact", () => { }, { once: true }); }); }, + scheduleDownloadTimeout: (onTimeout, timeoutMs) => { + let elapsedMilliseconds = 0; + let isCancelled = false; + + timeoutScheduled.resolve({ + advanceBy: (milliseconds) => { + elapsedMilliseconds += milliseconds; + + if (!isCancelled && elapsedMilliseconds >= timeoutMs) { + onTimeout(); + } + }, + isCancelled: () => isCancelled, + }); + + return () => { + isCancelled = true; + }; + }, }); - const signal = await requestStarted.promise; + const [signal, fakeTimer] = await Promise.all([ + requestStarted.promise, + timeoutScheduled.promise, + ]); + + fakeTimer.advanceBy(299_999); + expect(signal.aborted).toBe(false); - jest.advanceTimersByTime(300_000); + fakeTimer.advanceBy(1); await expect(installation).rejects.toBe(signal.reason); expect(signal.aborted).toBe(true); + expect(fakeTimer.isCancelled()).toBe(true); const cacheRoot = resolveOpenFlowCommandCacheRoot({ env: environment.env, diff --git a/src/application/commands/flow-artifact.ts b/src/application/commands/flow-artifact.ts index ecbe2f7..04164ba 100644 --- a/src/application/commands/flow-artifact.ts +++ b/src/application/commands/flow-artifact.ts @@ -122,6 +122,10 @@ export async function installOpenFlowCommandRelease( execPath: string; fetcher: Fetcher; onDownloadProgress?: (downloadedBytes: number) => void; + scheduleDownloadTimeout?: ( + onTimeout: () => void, + timeoutMs: number, + ) => () => void; }, ): Promise { if (release.bunVersion !== Bun.version) { @@ -166,6 +170,7 @@ export async function installOpenFlowCommandRelease( options.fetcher, archivePath, options.onDownloadProgress, + options.scheduleDownloadTimeout ?? scheduleDownloadTimeout, ); const archive = await readFile(archivePath); const decoded = decodeCommandArchive(archive, release); @@ -218,10 +223,14 @@ async function downloadCommandArchive( fetcher: Fetcher, archivePath: string, onProgress: ((downloadedBytes: number) => void) | undefined, + scheduleTimeout: ( + onTimeout: () => void, + timeoutMs: number, + ) => () => void, ): Promise { onProgress?.(0); const abortController = new AbortController(); - const timeoutId = setTimeout(() => { + const cancelTimeout = scheduleTimeout(() => { abortController.abort(new DOMException( "The operation was aborted due to timeout", "TimeoutError", @@ -313,10 +322,18 @@ async function downloadCommandArchive( } } finally { - clearTimeout(timeoutId); + cancelTimeout(); } } +function scheduleDownloadTimeout( + onTimeout: () => void, + timeoutMs: number, +): () => void { + const timeoutId = setTimeout(onTimeout, timeoutMs); + return () => clearTimeout(timeoutId); +} + function decodeCommandArchive( archive: Uint8Array, release: OpenFlowCommandRelease,