CLI 知识流水线：退出码、JSON 合同与最终状态

CLI 可以把导入、摘要、查询和检查串进调度任务。可靠流水线需要保存每一步的退出码、响应形状和最终状态，而不是让日志里出现“queued”就继续导出报告。以下命令对照固定提交，示例验收只用合成响应，不启动真实服务或读取个人历史。

先固定连接与非交互输出

server/src/cli/index.ts 提供全局 --base-url、--token、--json、--no-interactive。server/src/cli/connection.ts 按显式参数、环境、保存连接、隐式本地默认选择目标；自动启动只属于不可达的隐式本地默认。用户配置的 loopback 还要符合 cloudMode 检查。计划任务应明确目标和凭据来源，避免工作目录或保存连接变化后写错实例。远端用 HTTPS；当前客户端的传输检查不能代替部署层保障。

这些是现有命令形式，只有最后一个示例会请求摘要入队；在已审查的连接环境执行前先确认模式和数据范围：

crystal --json status
crystal --json conversations --status imported --limit 20
crystal --json notes list --tag typescript --limit 20
crystal --json search "transaction rollback" --limit 5
crystal --json tags
crystal --json summarize --all

JSON 形状和命令能力分开核对

命令	当前 JSON 结果	流水线不能假设的事
status	对象，统计在 stats	没有 queue；缺字段不能默认为队列空
conversations / notes list	items 与 total 等分页字段	第一页不是全部结果
search / tags	数组	search 不是 results 包装对象
summarize –all	queued、total 和 queue	入队不等于摘要、嵌入均成功
config get / set	配置结果；set 接受 llm 或 embedding 的点键	没有 sources.paths 配置命令

server/src/cli/commands/status.ts 和 server/src/cli/commands/search.ts 是逐项依据。serve 的形式是 crystal serve --daemon --port 3721、crystal serve stop、crystal serve status，不是 start 子命令。当前 serve status 的错误显示路径没有显式设非零退出，config test 也可能把连接失败打印出来但正常结束；调度器应读业务字段/预期响应，不能把这两条的进程成功单独当就绪。

server/src/cli/commands/import.ts 的本地路径会调用扫描流，TTY 会进入面板；本地分支没有把 –source 传下去。远端导入才使用 –source 过滤，并在 JSON 或非交互情况下要求 –yes。–yes 表示同意传输源对话，不是摘要开关。不要把某来源过滤的意图写成实际全来源扫描的验收结果。

server/src/cli/commands/summarize.ts 的单 ID JSON 分支当前调用 summarize 两次；不能示范它来承诺一次请求。定向自动化可按 server/src/routes/notes.ts 的 batch-ids、status-ids API 记录明确 ID 集合；鉴权、HTTP 状态、success、data 和 schema 都要检查。status-ids 中 unknown/error 应失败，imported/summarizing 继续等待，summarized/filtered 为不同终态。filtered 不代表生成了笔记；摘要终态也不保证 embedding_status=done。

命令的其余合同分别见 notes、tags、config、serve 与 HTTP 客户端。固定提交中的显示和请求边界应在部署前复验，不能用新版本的帮助页替它作保证。

一个会拒绝假成功的离线流水线

用 Node 24 将程序保存为 workflow.mjs，运行 node workflow.mjs。它真正生成子进程退出码、stdout 和 stderr，但子进程只是合成命令响应，不执行 crystal。按 ID 检查和轮询上限也是教学夹具，没有 HTTP、SSE、模型或生产数据库。

import assert from 'node:assert/strict';
import { spawnSync } from 'node:child_process';

// Synthetic executable responses, never a real crystal process.
function commandFixture(stdout, status = 0) {
  const code =
    'process.stderr.write("fixture diagnostic\\n");process.stdout.write(process.argv[1]);process.exit(Number(process.argv[2]));';
  const result = spawnSync(process.execPath, ['-e', code, stdout, String(status)], {
    encoding: 'utf8',
    timeout: 1000,
  });
  if (result.error || result.status !== 0) throw new Error('command failed');
  return JSON.parse(result.stdout);
}
function requireStatus(value) {
  assert.equal(typeof value.stats?.totalConversations, 'number');
  return value;
}
function requireImport(value) {
  assert.ok(Array.isArray(value.summarizationCandidateIds));
  assert.equal(value.errors, 0);
  return value.summarizationCandidateIds;
}
function inspectIds(value, expected) {
  assert.ok(Array.isArray(value.items));
  const byId = new Map(value.items.map((item) => [item.id, item.status]));
  assert.equal(byId.size, expected.length);
  let pending = false;
  for (const id of expected) {
    const status = byId.get(id);
    if (!status || ['error', 'unknown'].includes(status)) throw new Error(`failed ${id}`);
    if (['imported', 'summarizing'].includes(status)) pending = true;
    else if (!['summarized', 'filtered'].includes(status)) throw new Error('unknown state');
  }
  return pending ? 'pending' : 'finished';
}
async function waitForIds(read, ids, maxPolls = 3) {
  for (let i = 0; i < maxPolls; i++) {
    if (inspectIds(await read(), ids) === 'finished') return;
  }
  throw new Error('deadline exceeded');
}

const status = requireStatus(commandFixture(JSON.stringify({ stats: { totalConversations: 2 } })));
assert.equal(status.queue, undefined);
assert.throws(() => commandFixture('{}', 7), /command failed/);
assert.throws(() => commandFixture('not JSON'), SyntaxError);
assert.throws(() => requireStatus(commandFixture('{}')));
const ids = requireImport(
  commandFixture(JSON.stringify({ errors: 0, summarizationCandidateIds: ['a', 'b'] }))
);
assert.throws(() => requireImport({ errors: 1, summarizationCandidateIds: ids }));
const receipt = commandFixture('{"queued":2,"total":2}');
assert.equal(receipt.queued, 2); // Receipt alone never proves completion.
let polls = 0;
await waitForIds(
  async () => ({
    items: ids.map((id) => ({ id, status: ++polls <= 2 ? 'imported' : 'summarized' })),
  }),
  ids
);
assert.ok(polls > 2);
await assert.rejects(waitForIds(async () => ({ items: [] }), ids));
await assert.rejects(
  waitForIds(async () => ({ items: ids.map((id) => ({ id, status: 'error' })) }), ids)
);
await assert.rejects(
  waitForIds(async () => ({ items: ids.map((id) => ({ id, status: 'imported' })) }), ids),
  /deadline/
);
assert.equal(
  inspectIds(
    {
      items: [
        { id: 'a', status: 'filtered' },
        { id: 'b', status: 'summarized' },
      ],
    },
    ids
  ),
  'finished'
);
console.log('exit, JSON shape, per-ID terminal state, and deadlines pass on synthetic fixtures');

故障演练、调度和备份

删去非零退出保护，退出 7 的子进程会穿过解析，预期拒绝断言应失败。让空 items 直接返回 finished，缺失 ID 的预期拒绝断言应失败；恢复后再运行。stderr 与 stdout 分开保存，不用 2>&1 混进 JSON；命令超时、HTTP 错误、坏 JSON、缺字段、未知 ID、长期 pending 都要退出非零。轮询次数只是例子的确定性截止，真实任务用墙钟截止、单请求超时和退避。

server/src/queue/tracker.ts 的 tracker 完成空闲后 5 秒可清空；迟到的空队列不能证明刚才那批完成。保留目标 ID 并查持久化会话状态，再按任务目的核对笔记、嵌入和关系结果。操作系统调度另加单实例锁、固定 PATH/工作目录、日志保留和失败通知；这些是外围作业设计，没有在本文假装成为新 CLI 功能。

备份从 server/src/runtime/paths.ts 与 server/src/services/vector-index.ts 确认 chatcrystal.db、config.json、vectra-index 等实际位置，按部署把需要的鉴权文件纳入受控恢复计划。先停止所有写入者，或使用经过一致性验证的快照，再复制整个约定的数据集；逐份复制活动文件不构成跨文件事务。恢复到隔离实例，核对会话/笔记数量、目标 ID 和检索结果，不能只看压缩包存在。端口、连接和停止方式参见桌面服务生命周期，批量成本参见导入性能验收。