From a06e5df90a6347402ba56270448d18c7018d06ad Mon Sep 17 00:00:00 2001 From: 18628386721 <18628386721@163.com> Date: Tue, 28 Jul 2026 23:51:04 +0800 Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E6=B5=81=E7=A8=8B=E5=8C=85=E7=AB=AF?= =?UTF-8?q?=E5=88=B0=E7=AB=AF:=E4=BB=8E=E9=9B=B6=E5=BB=BA=E4=B8=80?= =?UTF-8?q?=E6=9D=A1=E6=B5=81=E5=B9=B6=E8=B7=91=E9=80=9A=20=E2=80=94?= =?UTF-8?q?=E2=80=94=20=E7=BC=96=E6=8E=92=E5=B1=82=E5=8F=AF=E7=8B=AC?= =?UTF-8?q?=E7=AB=8B=E5=BC=80=E5=B7=A5=E7=9A=84=E9=AA=8C=E6=94=B6=E8=AF=81?= =?UTF-8?q?=E6=8D=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 八例 unified-quote 证明的是「既有这条流能跑」;这一例证明另一件事: **拿编排层去开一条新流,不碰活动一行代码**。走的正是设计者那条路 —— 建包(草稿,不生效)→ 链接已装载的活动 → 激活(双判据闸)→ 起单 → 走完。 六例(真库): ① 建包存下来是未生效草稿;未生效版起单被拒(一次只能激活一个版本) ② 激活闸判据一:图编排不闭合 → NotClosedError ③ 激活闸判据二:活动未装载(unknown-activity)→ NotClosedError ④ 两面都齐即可激活,一条流恰一个生效版本 ⑤ 起单跑通:受理 → 审批(运营时分配审批人)→ 归档 → 闭环,末态 completed ⑥ 新流与老流引用同一批活动各跑各的 —— 建新流没动到活动 这条新流(quick-archive:受理→审批→归档)只用了编排层的能力: 节点两种、条件出边(审批点的 verdict 出边)、边映射五源、激活双判据。 Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01KeDBx2PHLavbURXB8LJJ7m --- .../app/__tests__/new-package-flow.it.test.ts | 278 ++++++++++++++++++ spike/mastra-reopen/package.json | 11 +- .../dowhile-shape-mismatch-runtime.ts | 37 +++ spike/mastra-reopen/tsconfig.json | 3 +- 4 files changed, 326 insertions(+), 3 deletions(-) create mode 100644 apps/api/src/app/__tests__/new-package-flow.it.test.ts create mode 100644 spike/mastra-reopen/src/negative/dowhile-shape-mismatch-runtime.ts diff --git a/apps/api/src/app/__tests__/new-package-flow.it.test.ts b/apps/api/src/app/__tests__/new-package-flow.it.test.ts new file mode 100644 index 0000000..3cd626c --- /dev/null +++ b/apps/api/src/app/__tests__/new-package-flow.it.test.ts @@ -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 } | 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 { + 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 { + 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 = {}): Promise { + 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 { + 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 => + 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); + }); +}); diff --git a/spike/mastra-reopen/package.json b/spike/mastra-reopen/package.json index 2caa9ea..44d57f6 100644 --- a/spike/mastra-reopen/package.json +++ b/spike/mastra-reopen/package.json @@ -3,11 +3,18 @@ "private": true, "version": "0.0.0", "type": "module", - "description": "隔离验证原型:回边 reopen 在 Mastra workflow 里怎么表达。不属于主 workspace。", + "description": "隔离验证原型:回边 reopen 在 Mastra workflow 里怎么表达。不属于主 workspace,用 pnpm install --ignore-workspace 装。", "scripts": { "route-a": "tsx src/route-a-dowhile.ts", "route-b": "tsx src/route-b-external.ts", - "probe": "tsx src/probe.ts" + "probe:smoke": "tsx src/probe-00-smoke.ts", + "probe:a-snapshot": "SPIKE_NO_AUTORUN=1 tsx src/probe-01-route-a-snapshot.ts", + "probe:a-pitfalls": "SPIKE_NO_AUTORUN=1 tsx src/probe-02-route-a-pitfalls.ts", + "probe:a-loop-detail": "SPIKE_NO_AUTORUN=1 tsx src/probe-03-loop-snapshot-detail.ts", + "probe:b-pitfalls": "SPIKE_NO_AUTORUN=1 tsx src/probe-04-route-b-pitfalls.ts", + "probe:misresume": "SPIKE_NO_AUTORUN=1 tsx src/probe-05-misresume-blast-radius.ts", + "neg:runtime": "tsx src/negative/dowhile-shape-mismatch-runtime.ts", + "typecheck": "tsc --noEmit" }, "dependencies": { "@mastra/core": "latest", diff --git a/spike/mastra-reopen/src/negative/dowhile-shape-mismatch-runtime.ts b/spike/mastra-reopen/src/negative/dowhile-shape-mismatch-runtime.ts new file mode 100644 index 0000000..d9916d4 --- /dev/null +++ b/spike/mastra-reopen/src/negative/dowhile-shape-mismatch-runtime.ts @@ -0,0 +1,37 @@ +/** 反例(运行时):循环体输出形状 ≠ 输入形状,第 2 轮会发生什么 */ +import { Mastra } from '@mastra/core'; +import { InMemoryStore } from '@mastra/core/storage'; +import { createStep, createWorkflow } from '@mastra/core/workflows'; +import { z } from 'zod'; + +const In = z.object({ round: z.number(), reopen: z.boolean() }); +const Out = z.object({ somethingElse: z.string() }); +let n = 0; + +const body = createStep({ + id: 'body', + inputSchema: In, + outputSchema: Out, + execute: async ({ inputData }) => { + n += 1; + console.log(` body 第 ${n} 次执行,拿到的 inputData =`, JSON.stringify(inputData)); + return { somethingElse: 'x' }; + }, +}); +const start = createStep({ + id: 'start', + inputSchema: z.object({}), + outputSchema: In, + execute: async () => ({ round: 1, reopen: true }), +}); + +const wf = createWorkflow({ id: 'neg', inputSchema: z.object({}), outputSchema: Out }) + .then(start) + .dowhile(body, async () => n < 2) + .commit(); + +const m = new Mastra({ workflows: { wf }, storage: new InMemoryStore(), logger: false }); +const run = await m.getWorkflow('wf').createRun(); +const res = await run.start({ inputData: {} }); +console.log('status =', res.status); +console.log('error =', JSON.stringify((res as { error?: { message?: string } }).error?.message ?? null)); diff --git a/spike/mastra-reopen/tsconfig.json b/spike/mastra-reopen/tsconfig.json index c41db61..ffe0666 100644 --- a/spike/mastra-reopen/tsconfig.json +++ b/spike/mastra-reopen/tsconfig.json @@ -9,5 +9,6 @@ "noEmit": true, "types": ["node"] }, - "include": ["src/**/*.ts"] + "include": ["src/**/*.ts"], + "exclude": ["src/negative"] }