fix(workflow): preserve UTF-8 in command outputs

docs/guide/workflow.md

@@ -24,6 +24,8 @@ steps:
 
 `schemaVersion` must be `v1alpha1`. `name` must match the file name. `description` is required. Each step has a unique `id` and exactly one action: `command`, `prompt`, `prompt_file`, or `workflow`. A step can omit the action when it has `end`, `approve`, or `choose`. `agent` is optional on prompt steps. `output` uses the existing shorthand: `string`, `number`, `boolean`, a literal union, an object, or an array with one item type. TeamAI validates successful command stdout or a prompt tool's `output` against it.
 
+Local command steps read stdout and stderr as UTF-8. Non-ASCII JSON string values are preserved in stored outputs and when later steps read them.
+
 A prompt step can set `model`: an alias from `models/aliases.yaml` or a model name, without spaces or templates. `defaults: { model: fast }` at the top of a file sets one for its prompt steps. A prompt step runs on the first of these that is set:
 
 1. the step's `model`;

docs/guide/zh-CN/workflow.md

@@ -24,6 +24,8 @@ steps:
 
 `schemaVersion` 必须是 `v1alpha1`。`name` 必须和文件名一致。`description` 必填。每个步骤有唯一的 `id`，并且只能使用 `command`、`prompt`、`prompt_file`、`workflow` 中的一种动作。带有 `end`、`approve` 或 `choose` 的步骤可以省略动作。提示词步骤可选填 `agent`。`output` 沿用已有简写：`string`、`number`、`boolean`、字面量联合、对象，或只含一种元素类型的数组。TeamAI 会用它校验成功命令的 stdout 或提示词工具返回的 `output`。
 
+本地命令步骤以 UTF-8 读取 stdout 和 stderr。JSON 字符串中的非 ASCII 文本会原样保存到输出中，后续步骤读取时也保持原文。
+
 提示词步骤可以设置 `model`：`models/aliases.yaml` 中的别名或模型名，不能含空格或模板。文件顶层的 `defaults: { model: fast }` 为其中的提示词步骤设置模型。提示词步骤使用以下第一个已设置的模型：
 
 1. 步骤的 `model`；

src/__tests__/e2e/workflow-command-utf8.test.ts

@@ -0,0 +1,88 @@
+import { execFileSync, spawnSync } from 'node:child_process';
+import fs from 'node:fs';
+import os from 'node:os';
+import path from 'node:path';
+import { fileURLToPath } from 'node:url';
+import { afterEach, expect, it } from 'vitest';
+import YAML from 'yaml';
+import { resolveProjectDataHome } from '../../config.js';
+import type { LocalConfig } from '../../types.js';
+import { syncWorkflowDefinitions } from '../../workflow/cache.js';
+import type { RunState } from '../../workflow/types.js';
+
+const cli = fileURLToPath(new URL('../../../dist/index.js', import.meta.url));
+const text = '检查完成 café 😀';
+let sandbox: string;
+afterEach(() => { if (sandbox) fs.rmSync(sandbox, { recursive: true, force: true }); });
+
+it.each([0, 1])('preserves split command UTF-8 through the real CLI (exit %i)', async (exitCode) => {
+  sandbox = fs.realpathSync.native(fs.mkdtempSync(path.join(os.tmpdir(), 'teamai-command-utf8-cli-')));
+  const repo = path.join(sandbox, 'app');
+  const team = path.join(sandbox, 'team');
+  fs.mkdirSync(repo);
+  fs.mkdirSync(path.join(team, 'workflows'), { recursive: true });
+  const git = (...args: string[]) => execFileSync('git', args, { cwd: repo, windowsHide: true, stdio: 'pipe' });
+  git('init', '-q', '-b', 'main');
+  git('remote', 'add', 'origin', 'http://127.0.0.1:9/team/workflow.git');
+  fs.writeFileSync(path.join(repo, 'emit.cjs'), `
+const fs = require('node:fs');
+const bytes = Buffer.from(${JSON.stringify(exitCode === 0 ? JSON.stringify({ summary: text }) : text)});
+const split = bytes.indexOf(Buffer.from('检')) + 1;
+fs.writeSync(${exitCode === 0 ? 1 : 2}, bytes.subarray(0, split));
+setTimeout(() => {
+  fs.writeSync(${exitCode === 0 ? 1 : 2}, bytes.subarray(split));
+  process.exitCode = ${exitCode};
+}, 150);
+`);
+  fs.writeFileSync(path.join(repo, 'consume.cjs'), 'process.stdout.write(JSON.stringify({ seen: process.env.SUMMARY }));\n');
+  git('add', '.');
+  git('-c', 'user.name=Test', '-c', 'user.email=test@teamai.test', 'commit', '-qm', 'fixture');
+
+  const dataHome = await resolveProjectDataHome(repo);
+  const config: LocalConfig = {
+    repo: { localPath: team, remote: 'http://127.0.0.1:9/team/workflow.git', kind: 'git' },
+    username: 'tester', scope: 'project', projectRoot: repo, dataHome, updatePolicy: 'skip', additionalRoles: [],
+  };
+  fs.mkdirSync(dataHome, { recursive: true });
+  fs.writeFileSync(path.join(dataHome, 'config.yaml'), YAML.stringify(config));
+  fs.writeFileSync(path.join(team, 'teamai.yaml'), YAML.stringify({
+    team: 'command-unicode', repo: config.repo.remote, provider: 'git', usageReport: false,
+  }));
+  fs.writeFileSync(path.join(team, 'workflows', 'unicode.yaml'), `schemaVersion: v1alpha1
+name: unicode
+description: Pass command output to the next step.
+steps:
+  - id: emit
+    command: node emit.cjs
+    output: { summary: string }
+  - id: consume
+    needs: [emit]
+    command: node consume.cjs
+    env:
+      SUMMARY: "{{ steps.emit.outputs.summary }}"
+    output: { seen: string }
+`);
+  await syncWorkflowDefinitions(config);
+  const result = spawnSync(process.execPath, [cli, 'workflow', 'start', 'unicode', '--driver', 'local'], {
+    cwd: repo, env: { ...process.env, FORCE_COLOR: '0' }, encoding: 'utf8', windowsHide: true, timeout: 30_000,
+  });
+  expect(result.error).toBeUndefined();
+  expect(result.status, result.stderr).toBe(exitCode);
+  const envelope = JSON.parse(result.stdout.trim().split('\n').filter((line) => line.startsWith('{')).at(-1)!);
+  expect(envelope, result.stderr).toMatchObject({ data: { done: { outcome: exitCode === 0 ? 'succeeded' : 'failed' } } });
+  const runId = envelope.data.done.run as string;
+  const runsDir = path.join(dataHome, 'workflows', 'runs');
+  const stored = JSON.parse(fs.readFileSync(path.join(runsDir, `${runId}.json`), 'utf8')) as RunState;
+  if (exitCode === 0) {
+    expect(stored.steps).toMatchObject([
+      { status: 'succeeded', outputs: { summary: text } },
+      { status: 'succeeded', outputs: { seen: text } },
+    ]);
+    const attempts = path.join(runsDir, runId, 'attempts');
+    const results = fs.readdirSync(attempts).map((attempt) => JSON.parse(fs.readFileSync(path.join(attempts, attempt, 'result.json'), 'utf8')));
+    expect(results).toContainEqual({ output: { summary: text } });
+    expect(results).toContainEqual({ output: { seen: text } });
+  } else {
+    expect(stored.steps[0]).toMatchObject({ status: 'failed', message: `command exited with status 1: ${text}` });
+  }
+}, 60_000);

src/__tests__/workflow-command-utf8.test.ts

@@ -0,0 +1,103 @@
+import { execFileSync } from 'node:child_process';
+import fs from 'node:fs';
+import os from 'node:os';
+import path from 'node:path';
+import { afterEach, describe, expect, it, vi } from 'vitest';
+import type { LocalConfig } from '../types.js';
+import { syncWorkflowDefinitions } from '../workflow/cache.js';
+import { startRun } from '../workflow/engine.js';
+import { runLocalDriver } from '../workflow/local-driver.js';
+import { readRun } from '../workflow/store.js';
+
+const output = vi.hoisted(() => ({ stdout: Buffer.alloc(0), stderr: Buffer.alloc(0), code: 0 }));
+vi.mock('node:child_process', async (importOriginal) => {
+  const actual = await importOriginal<typeof import('node:child_process')>();
+  const { EventEmitter } = await import('node:events');
+  const { PassThrough } = await import('node:stream');
+  return {
+    ...actual,
+    spawn: vi.fn((...args: Parameters<typeof actual.spawn>) => {
+      if (args[0] !== 'unicode-fixture') return actual.spawn(...args);
+      const child = Object.assign(new EventEmitter(), {
+        stdout: new PassThrough(), stderr: new PassThrough(), kill: vi.fn(),
+      });
+      queueMicrotask(() => {
+        // Force byte boundaries and interleave streams without relying on OS pipe timing.
+        for (let i = 0; i < Math.max(output.stdout.length, output.stderr.length); i++) {
+          if (i < output.stdout.length) child.stdout.write(output.stdout.subarray(i, i + 1));
+          if (i < output.stderr.length) child.stderr.write(output.stderr.subarray(i, i + 1));
+        }
+        child.stdout.end();
+        child.stderr.end();
+        child.emit('close', output.code);
+      });
+      return child;
+    }),
+  };
+});
+
+let root: string;
+afterEach(() => { if (root) fs.rmSync(root, { recursive: true, force: true }); });
+
+async function run() {
+  root = fs.realpathSync.native(fs.mkdtempSync(path.join(os.tmpdir(), 'teamai-command-utf8-')));
+  const workspace = path.join(root, 'workspace');
+  const team = path.join(root, 'team');
+  fs.mkdirSync(workspace);
+  fs.mkdirSync(path.join(team, 'workflows'), { recursive: true });
+  execFileSync('git', ['init', '-q'], { cwd: workspace });
+  fs.writeFileSync(path.join(workspace, 'README.md'), '# fixture\n');
+  execFileSync('git', ['add', '.'], { cwd: workspace });
+  execFileSync('git', ['-c', 'user.name=Test', '-c', 'user.email=test@teamai.test', 'commit', '-qm', 'fixture'], { cwd: workspace });
+  fs.writeFileSync(path.join(team, 'workflows', 'unicode.yaml'), `schemaVersion: v1alpha1
+name: unicode
+description: Preserve command output.
+steps:
+  - id: emit
+    command: unicode-fixture
+`);
+  const config: LocalConfig = {
+    repo: { localPath: team, remote: 'local', kind: 'git' }, username: 'tester',
+    scope: 'project', projectRoot: workspace, dataHome: path.join(root, 'data'), additionalRoles: [],
+  };
+  await syncWorkflowDefinitions(config);
+  const started = await startRun(config, 'unicode');
+  if ('error' in started) throw new Error(started.error.message);
+  const runId = String(started.data.run);
+  const done = await runLocalDriver(config, runId, workspace);
+  const state = await readRun(config, runId);
+  if (typeof state === 'string') throw new Error(`Workflow state is ${state}`);
+  const attempts = path.join(config.dataHome!, 'workflows', 'runs', runId, 'attempts');
+  const attempt = fs.readdirSync(attempts)[0]!;
+  const result = JSON.parse(fs.readFileSync(path.join(attempts, attempt, 'result.json'), 'utf8'));
+  return { done, state, result };
+}
+
+describe('local workflow command UTF-8', () => {
+  it.each(['检查完成', 'café', '😀', 'plain ASCII'])('preserves %s in state and the attempt result', async (summary) => {
+    output.stdout = Buffer.from(JSON.stringify({ summary }));
+    output.stderr = Buffer.from('诊断 café 😀');
+    output.code = 0;
+    const { done, state, result } = await run();
+    expect(done).toMatchObject({ data: { done: { outcome: 'succeeded' } } });
+    expect(state?.steps[0]).toMatchObject({ status: 'succeeded', outputs: { summary } });
+    expect(result).toEqual({ output: { summary } });
+  });
+
+  it('preserves split stderr in the stored command failure', async () => {
+    output.stdout = Buffer.alloc(0);
+    output.stderr = Buffer.from('执行失败 café 😀');
+    output.code = 1;
+    const { done, state } = await run();
+    expect(done).toMatchObject({ data: { done: { outcome: 'failed' } } });
+    expect(state?.steps[0]).toMatchObject({ message: 'command exited with status 1: 执行失败 café 😀' });
+  });
+
+  it('flushes an incomplete trailing UTF-8 sequence on stderr at EOF', async () => {
+    output.stdout = Buffer.alloc(0);
+    output.stderr = Buffer.concat([Buffer.from('failure: '), Buffer.from([0xe4, 0xbd])]);
+    output.code = 1;
+    const { state } = await run();
+    expect(state?.steps[0]).toMatchObject({ message: 'command exited with status 1: failure: �' });
+  });
+});

src/workflow/local-driver.ts

@@ -82,8 +82,10 @@ function startCommand(command: string, cwd: string, env: Record<string, string>,
   const exit = new Promise<CommandResult>((resolve) => {
     let stdout = '';
     let stderr = '';
-    child.stdout?.on('data', (chunk: Buffer) => { stdout += chunk.toString(); });
-    child.stderr?.on('data', (chunk: Buffer) => { stderr += chunk.toString(); });
+    child.stdout?.setEncoding('utf8');
+    child.stderr?.setEncoding('utf8');
+    child.stdout?.on('data', (chunk: string) => { stdout += chunk; });
+    child.stderr?.on('data', (chunk: string) => { stderr += chunk; });
     child.on('error', (error) => resolve({ exitCode: 1, stdout, stderr: error.message }));
     child.on('close', (code) => resolve({ exitCode: code ?? 1, stdout, stderr }));
   });