新流程包端到端:从零建一条流并跑通 —— 编排层可独立开工的验收证据

八例 unified-quote 证明的是「既有这条流能跑」;这一例证明另一件事: **拿编排层去开一条新流,不碰活动一行代码**。走的正是设计者那条路 —— 建包(草稿,不生效)→ 链接已装载的活动 → 激活(双判据闸)→ 起单 → 走完。六例(真库): ① 建包存下来是未生效草稿;未生效版起单被拒(一次只能激活一个版本) ② 激活闸判据一:图编排不闭合 → NotClosedError ③ 激活闸判据二:活动未装载(unknown-activity)→ NotClosedError ④ 两面都齐即可激活,一条流恰一个生效版本 ⑤ 起单跑通:受理 → 审批(运营时分配审批人)→ 归档 → 闭环,末态 completed ⑥ 新流与老流引用同一批活动各跑各的 —— 建新流没动到活动这条新流(quick-archive:受理→审批→归档)只用了编排层的能力: 节点两种、条件出边(审批点的 verdict 出边)、边映射五源、激活双判据。 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01KeDBx2PHLavbURXB8LJJ7m
2026-07-28 23:51:04 +08:00
parent f30f24e0ba
commit a06e5df90a
4 changed files with 326 additions and 3 deletions
@@ -0,0 +1,278 @@
+/**
+ * **从零建一个新流程包并跑通** —— 编排层与活动解耦的验收证据(真库)。
+ *
+ * 八例 `unified-quote` 端到端证明的是「这一条既有的流能跑」;这里证明的是另一件事:
+ * **拿编排层去开一条新流,不碰活动一行代码**。走的正是设计者的那条路 ——
+ *   建包(草稿存下,不生效)→ 链接已装载的活动 → 激活(双判据闸)→ 起单 → 走完。
+ *
+ * 这就是「为后续 MVP 开发一个流程包做准备」到底准备好了没有的判据。
+ * 库的选择与八例同规矩:**绝不回落到开发库**;跳过必须可见。
+ */
+import { afterAll, beforeAll, describe, expect, it } from 'vitest';
+import { Pool } from 'pg';
+import { drizzle } from 'drizzle-orm/node-postgres';
+import { migrate } from 'drizzle-orm/node-postgres/migrator';
+import { existsSync } from 'node:fs';
+import { dirname, join, resolve } from 'node:path';
+import { fileURLToPath } from 'node:url';
+import type { ProcessPackage, TaskId, WorkOrderId } from '@contracts';
+import { processPackageSchema } from '@contracts';
+import { NotClosedError } from '@modules/process';
+import { loadConfig } from '../config.ts';
+import { createContainer } from '../container.ts';
+import type { Container } from '../container.ts';
+import { seedMvp } from '../seed-mvp.ts';
+import { advanceOnce } from '../workers/task-advancer.ts';
+import { dispatchOnce } from '../workers/dispatcher.ts';
+import { createConsumers } from '../workers/consumers.ts';
+
+const IT_DB = 'jcai_it_newpkg';
+
+async function provisionDatabase(): Promise<{ url: string; stop(): Promise<void> } | undefined> {
+  const explicit = process.env['JCAI_TEST_DATABASE_URL'];
+  if (explicit !== undefined) return { url: explicit, stop: async () => undefined };
+
+  const base = process.env['DATABASE_URL'] ?? 'postgres://jcai:jcai@localhost:54329/jcai';
+  try {
+    const admin = new Pool({ connectionString: base, max: 1, connectionTimeoutMillis: 1500 });
+    const exists = await admin.query(`select 1 from pg_database where datname = $1`, [IT_DB]);
+    if (exists.rowCount === 0) await admin.query(`create database ${IT_DB}`);
+    await admin.end();
+    return { url: base.replace(/\/[^/]*$/, `/${IT_DB}`), stop: async () => undefined };
+  } catch {
+    /* 开发实例连不上,试 Testcontainers */
+  }
+  try {
+    const { PostgreSqlContainer } = await import('@testcontainers/postgresql');
+    const started = await new PostgreSqlContainer('postgres:16-alpine').start();
+    return { url: started.getConnectionUri(), stop: () => started.stop().then(() => undefined) };
+  } catch (error) {
+    process.stderr.write(
+      `⚠️  新建流程包 E2E 跳过:连不上开发实例也起不了 Testcontainers(${error instanceof Error ? error.message : String(error)})\n`,
+    );
+    return undefined;
+  }
+}
+
+function migrationsFolder(): string {
+  let dir = dirname(fileURLToPath(import.meta.url));
+  for (let i = 0; i < 8; i += 1) {
+    const candidate = join(dir, 'db', 'migrations');
+    if (existsSync(candidate)) return candidate;
+    const parent = resolve(dir, '..');
+    if (parent === dir) break;
+    dir = parent;
+  }
+  throw new Error('找不到 db/migrations');
+}
+
+const database = await provisionDatabase();
+
+let pool: Pool | undefined;
+let container: Container;
+let chaihao: string;
+
+beforeAll(async () => {
+  if (database === undefined) return;
+  pool = new Pool({ connectionString: database.url, max: 4 });
+  for (const schema of ['sys', 'ledger', 'mastra', 'queue']) {
+    await pool.query(`drop schema if exists "${schema}" cascade`);
+    await pool.query(`create schema "${schema}"`);
+  }
+  await migrate(drizzle(pool), { migrationsFolder: migrationsFolder(), migrationsSchema: 'sys' });
+  container = await createContainer(loadConfig({ ...process.env, DATABASE_URL: database.url }));
+  await container.org.ensureSeeded();
+  // 种子只为**装载活动**(六个槽位)—— 新流程包引用它们,不改它们一行
+  await seedMvp(container);
+  const { staff } = await container.org.listStaff();
+  chaihao = staff.find((s) => s.account === 'Chaihao')!.id;
+}, 180_000);
+
+afterAll(async () => {
+  if (database === undefined) return;
+  await container?.close();
+  await pool?.end();
+  await database.stop();
+});
+
+const consumers = createConsumers();
+
+async function pump(): Promise<void> {
+  for (let i = 0; i < 60; i += 1) {
+    const delivered = await dispatchOnce({ container, consumers });
+    const advanced = await advanceOnce({ container });
+    if (delivered === 0 && advanced === 0) return;
+  }
+  throw new Error('泵 60 轮未静止 —— 流转在空转');
+}
+
+async function currentTask(workflowId: string): Promise<TaskId> {
+  const rows = await pool!.query<{ id: string }>(
+    `select id from sys.task where workflow_id = $1 and terminal is null order by started_at desc limit 1`,
+    [workflowId],
+  );
+  expect(rows.rowCount, `流 ${workflowId} 应有在途任务`).toBe(1);
+  return rows.rows[0]!.id as TaskId;
+}
+
+async function acceptCurrent(workflowId: string, values: Record<string, string> = {}): Promise<void> {
+  const taskId = await currentTask(workflowId);
+  const open = await container.humanTool.openWorkOrders(taskId);
+  const order = open.find((o) => o.mode === 'acceptance');
+  expect(order, '应有待答验收单').toBeDefined();
+  await container.humanTool.replyWorkOrder(
+    { mode: 'acceptance', workOrderId: order!.id as WorkOrderId, verdict: 'pass', values, selectedPorts: {} },
+    chaihao as never,
+  );
+  await pump();
+}
+
+async function seq(workflowId: string): Promise<string[]> {
+  const instances = await container.process.instancesOf(workflowId);
+  return instances.map((i) => `${i.nodeId}#${i.instance}=${i.terminal ?? '…'}`);
+}
+
+/**
+ * 一条**新流**:受理 → 审批 → 归档。
+ * 引用的是已装载的活动(`rfq-intake` / `intel-archive`)—— 编排只链接,不碰实现。
+ */
+const NEW_FLOW = (over: Partial<ProcessPackage> = {}): ProcessPackage =>
+  processPackageSchema.parse({
+    id: 'quick-archive',
+    name: '快速归档',
+    version: 1,
+    state: 'active',
+    callInterface: {
+      input: {
+        type: 'object',
+        required: ['customerCode', 'rfqRef'],
+        properties: { customerCode: { type: 'string' }, rfqRef: { type: 'string' } },
+      },
+      output: { type: 'object' },
+    },
+    nodes: [
+      { id: 'start', kind: 'start', label: '起', layout: { row: 0, col: 0 } },
+      { id: 'A1', kind: 'activity', label: '受理', activityId: 'rfq-intake', failure: { kind: 'terminate' } },
+      { id: 'A2', kind: 'approval', label: '归档审批', reject: { kind: 'terminate' } },
+      { id: 'A3', kind: 'activity', label: '归档', activityId: 'intel-archive', failure: { kind: 'terminate' } },
+      { id: 'end', kind: 'end', label: '闭环' },
+    ],
+    edges: [
+      {
+        id: 'E1',
+        from: 'start',
+        to: 'A1',
+        mapping: [
+          { target: 'customerCode', source: { from: 'initiation', path: 'customerCode' } },
+          { target: 'rfqRef', source: { from: 'initiation', path: 'rfqRef' } },
+        ],
+        reopen: false,
+      },
+      { id: 'E2', from: 'A1', to: 'A2', mapping: [], reopen: false },
+      // 审批点的去向就是它自己的条件出边:通过才归档
+      {
+        id: 'E3',
+        from: 'A2',
+        to: 'A3',
+        condition: 'verdict = "pass"',
+        mapping: [
+          { target: 'outcome', source: { from: 'constant', value: 'archived' } },
+          { target: 'quote_no', source: { from: 'initiation', path: 'rfqRef' } },
+          { target: 'customer_code', source: { from: 'initiation', path: 'customerCode' } },
+          { target: 'quotedAt', source: { from: 'constant', value: '2026-07-28' } },
+        ],
+        reopen: false,
+      },
+      { id: 'E4', from: 'A3', to: 'end', mapping: [], reopen: false },
+    ],
+    variables: [],
+    ...over,
+  });
+
+describe.skipIf(database === undefined)('从零建一条新流并跑通 —— 编排层可独立开工', () => {
+  it('① 建包:存下来是**未生效**的草稿,不影响任何在途', async () => {
+    const saved = await container.process.savePackage(NEW_FLOW());
+    expect(saved.version).toBe(1);
+    expect(saved.state).toBe('retired');
+    // 未生效版不得起单 —— 「一次只能激活一个版本」对新起单是硬约束
+    await expect(
+      container.process.start({
+        packageId: 'quick-archive',
+        subject: '未激活试起',
+        source: { kind: 'registration', ref: 'it', by: chaihao },
+        data: { customerCode: 'C1', rfqRef: 'R1' },
+      }),
+    ).rejects.toThrow();
+  });
+
+  it('② 激活闸:图编排不闭合的包激活被拒(判据一)', async () => {
+    // 把 A3 的入边摘掉:A3 成孤岛、A2 的出边指向消失 —— 不闭合
+    const broken = NEW_FLOW({ id: 'broken-flow' as never, edges: NEW_FLOW().edges.filter((e) => e.id !== 'E3') });
+    const saved = await container.process.savePackage(broken);
+    await expect(container.process.setPackageState('broken-flow', saved.version, 'active')).rejects.toBeInstanceOf(
+      NotClosedError,
+    );
+  });
+
+  it('③ 激活闸:活动未装载的包激活被拒(判据二)', async () => {
+    const unlinked = NEW_FLOW({
+      id: 'unlinked-flow' as never,
+      nodes: NEW_FLOW().nodes.map((n) =>
+        n.id === 'A1' && n.kind === 'activity' ? { ...n, activityId: '未安装:随手画的' as never } : n,
+      ),
+    });
+    const saved = await container.process.savePackage(unlinked);
+    const report = await container.process.validate({ ...unlinked, version: saved.version });
+    expect(report.violations.some((v) => v.kind === 'unknown-activity')).toBe(true);
+    await expect(container.process.setPackageState('unlinked-flow', saved.version, 'active')).rejects.toBeInstanceOf(
+      NotClosedError,
+    );
+  });
+
+  it('④ 两面都齐即可激活 —— 一条流恰一个生效版本', async () => {
+    const packages = await container.process.setPackageState('quick-archive', 1, 'active');
+    const mine = packages.filter((p) => p.id === 'quick-archive');
+    expect(mine.filter((p) => p.state === 'active')).toHaveLength(1);
+    expect(mine.find((p) => p.version === 1)?.state).toBe('active');
+  });
+
+  it('⑤ 起单跑通:受理 → 审批 → 归档 → 闭环,末态 completed', async () => {
+    const record = await container.process.start({
+      packageId: 'quick-archive',
+      subject: '新流首单',
+      source: { kind: 'registration', ref: 'it', by: chaihao },
+      data: { customerCode: 'C-NEW-1', rfqRef: 'RFQ-NEW-1' },
+    });
+    await pump();
+
+    // A1 受理:占位活动的收口验收
+    await acceptCurrent(record.id, { needAlternative: 'false', needSourcing: 'false' });
+
+    // A2 审批:审批人由运营时分配(员工组/全员轮值),裁决通过
+    await container.process.onApprovalVerdict(record.id, 'A2', 'pass', { by: chaihao });
+    await pump();
+
+    // A3 归档:再一张收口验收
+    await acceptCurrent(record.id);
+
+    expect(await seq(record.id)).toEqual([
+      'A1#1=released',
+      'A2#1=released',
+      'A3#1=released',
+      'end#1=released',
+    ]);
+    expect((await container.process.getWorkflow(record.id)).state).toBe('completed');
+  });
+
+  it('⑥ 全程没碰活动一行:两条流引用同一批活动,各跑各的', async () => {
+    // 既有的 unified-quote 照旧可起单 —— 新流的建立没有动到它引用的活动
+    const other = await container.process.start({
+      packageId: 'unified-quote',
+      subject: '老流照跑',
+      source: { kind: 'registration', ref: 'it', by: chaihao },
+      data: { customerCode: 'C-OLD', rfqRef: 'RFQ-OLD', subject: 's', body: 'b' },
+    });
+    await pump();
+    expect((await seq(other.id)).some((s) => s.startsWith('A1#1'))).toBe(true);
+  });
+});