/** * process 用例 —— 流程引擎:确定性流转,**零 LLM、零临场**(编码规范 §5.3;语义出处 OS 03 / 04)。 * * 起单 · 调度节点 · 审批点卡工位 · 条件出边选路 · 每次流转记入事件。 */ import type { ProcessPackage } from '@contracts'; import type { ProcessService, StartWorkflowInput } from '../../contract.ts'; import { ActiveVersionError, NotClosedError, PinnedVersionError, StaleVersionError, VersionTakenError, } from '../../contract.ts'; import type { ActivityDirectory, ApprovalGateway, Clock, FlowEventPublisher, FlowRuntime, IdGenerator, OrgDirectory, ProcessStore, ReceiptNotifier, SchemaValidator, Sfeel, TaskGateway, WorkflowRecord, } from '../../port.ts'; import { checkClosure } from '../domain/closure.ts'; import { createSelfHostedRuntime } from '../runtime/self-hosted-runtime.ts'; export type ProcessServiceDeps = { readonly store: ProcessStore; readonly activities: ActivityDirectory; readonly org: OrgDirectory; readonly tasks: TaskGateway; readonly approvals: ApprovalGateway; readonly sfeel: Sfeel; readonly schemas: SchemaValidator; readonly events: FlowEventPublisher; readonly receipts: ReceiptNotifier; readonly clock: Clock; readonly ids: IdGenerator; /** * 流程运行时(可替换切面,M5-1):缺省 = 自持状态机。 * 换 Mastra 编译产物时从这里注入 —— usecase 与全部路由/接线不动。 */ readonly runtime?: FlowRuntime; }; export function createProcessService(deps: ProcessServiceDeps): ProcessService { async function requirePackage(id: string, version?: number): Promise { const pkg = await deps.store.getPackage(id, version); if (pkg === undefined) throw new Error(`流程包不存在:${id}`); return pkg; } /** * 起单取版:取这条流的**生效版本** —— 一条流恰一个(或零个)生效版本。 * * 「生效」是运行时开关,不是内容属性:保存新版即生效(旧生效版随之停用), * 也可以手动把生效切回旧版本、或全部停用让整条流下线。 * 在途工作流按启动时钉住的版本号读定义,启停对它们零影响 —— 这正是 * 「切换版本不影响旧实例」的机制保证:实例上钉的是版本号,不是「当前生效」指针。 */ async function activeVersion(id: string): Promise { const all = await deps.store.listPackages(); const hit = all.find((pkg) => pkg.id === id && pkg.state === 'active'); if (hit === undefined) throw new Error(`流程包无生效版本,不得起单:${id} —— 去流程包里激活一个版本`); return hit; } /** 下一个版本号 —— 只在显式新建 / 新流程首版时用(编辑即其身,写回不长号) */ async function nextVersion(id: string): Promise { const all = await deps.store.listPackages(); return all.filter((pkg) => pkg.id === id).reduce((max, pkg) => Math.max(max, pkg.version), 0) + 1; } async function requireWorkflow(id: string): Promise { const record = await deps.store.getWorkflow(id); if (record === undefined) throw new Error(`工作流不存在:${id}`); return record; } const runtime = deps.runtime ?? createSelfHostedRuntime(deps); return { /** * 保存 —— **严格检查 + 即刻生效**。没有发布这道门,所以闸落在这里。 * * 两件事一起做,顺序不能换: * ① 先校验:不闭合直接抛,**一个字节都不落库**。存进来的就是能起单的。 * ② 再定版:恒取「当前最大版本 + 1」,不采信入参的 version。 * * 为什么不能就地覆盖:工作流按 (包id, 版本) 去读定义。覆盖 v1 就等于改掉 * 所有在途 v1 任务正在读的编排 —— 它们会中途换了规则却毫不知情, * 而「旧任务走完旧版本」正是要防这个。恒新增之后,旧版本是不可变的历史, * 在途任务读得到的永远是它启动那天的那一份。 */ /** * 保存**不检查**:画图必然经过不完整的中间态(节点没接完边、活动还没安装), * 拿闭合校验拦保存,画到一半就存不了(实测 412 卡死编辑)。 * 存下来的是**未生效**的新版本 —— 完整性闸在激活那一步。 */ async savePackage(pkg, opts) { const existing = opts?.asNew === true ? undefined : await deps.store.getPackage(pkg.id, pkg.version); if (existing === undefined) { /* * 流程已存在而这个版本号查无此版 = 陈旧编辑器(实测:停在已清理旧版上的 * 标签页发来保存,把老内容静默复活成了新版本)—— 拒掉,让人刷新后再改。 * 只有整条流都不存在(新流程首版)或显式 asNew 才走插入。 */ if (opts?.asNew !== true) { const newest = await deps.store.getPackage(pkg.id); if (newest !== undefined) throw new StaleVersionError(pkg.id, pkg.version); } // 显式新建 / 新流程首版:插入下一个版本号,不生效 —— 新版本只出自这条路 const saved: ProcessPackage = { ...pkg, version: await nextVersion(pkg.id), state: 'retired' }; await deps.store.savePackage(saved); return saved; } /* * **编辑即其身**(产品决策 2026-07-28 第四轮):保存写回当前版本,号不变, * 不再每存一次长一版。两道底线: * ① 在途钉版不可变 —— 工作流按 (包, 版本) 钉定义,有未完结的就不许改字 * ② 生效版必须始终可运行 —— 写回生效版要过闭合闸(未生效版画图不检查) */ const pinned = await deps.store.countPinnedInFlight(pkg.id, pkg.version); if (pinned > 0) throw new PinnedVersionError(pkg.id, pkg.version, pinned); const saved: ProcessPackage = { ...pkg, version: existing.version, state: existing.state }; if (existing.state === 'active') { const report = await this.validate(saved); if (!report.closed) throw new NotClosedError(report); } await deps.store.updatePackageDefinition(pkg.id, pkg.version, saved); return saved; }, async setPackageState(id, version, state) { const all = await deps.store.listPackages(); const target = all.find((p) => p.id === id && p.version === version); if (target === undefined) throw new Error(`流程包版本不存在:${id}@${version}`); if (state === 'active') { // **激活才查完整性**:不闭合不得激活 —— 画图不拦,上线拦 const report = await this.validate(target); if (!report.closed) throw new NotClosedError(report); // 激活即独占:同流其余生效版一并停用 —— 「一次只能激活一个版本」 for (const sib of all) { if (sib.id === id && sib.version !== version && sib.state === 'active') { await deps.store.setPackageState(id, sib.version, 'retired'); } } } await deps.store.setPackageState(id, version, state); const after = await deps.store.listPackages(); return after.filter((p) => p.id === id).sort((a, b) => b.version - a.version); }, async deleteVersion(id, version) { const all = await deps.store.listPackages(); const target = all.find((p) => p.id === id && p.version === version); if (target === undefined) throw new Error(`流程包版本不存在:${id}@${version}`); // 生效版不可删:先停用 —— 挡的是「手滑把正在起单的定义删了」 if (target.state === 'active') throw new ActiveVersionError(id, version); /* * 在途钉版不可删:那些工作流按启动时钉住的这一版跑完,定义没了就再也推不动 —— * 与「在途钉版不可改」同一条理由、同一个错。 */ const pinned = await deps.store.countPinnedInFlight(id, version); if (pinned > 0) throw new PinnedVersionError(id, version, pinned); await deps.store.deletePackage(id, version); const after = await deps.store.listPackages(); return after.filter((p) => p.id === id).sort((a, b) => b.version - a.version); }, async deleteFlow(id) { const all = await deps.store.listPackages(); const mine = all.filter((p) => p.id === id); if (mine.length === 0) throw new Error(`流程不存在:${id}`); /* * 生效版这道闸**不设**:删整条流是明示动作,先停用再删只是多一步仪式。 * 在途单那道闸必须设 —— 定义没了那些单就再也推不动,与删单版同一条理由。 */ for (const v of mine) { const pinned = await deps.store.countPinnedInFlight(id, v.version); if (pinned > 0) throw new PinnedVersionError(id, v.version, pinned); } for (const v of mine) await deps.store.deletePackage(id, v.version); return deps.store.listPackages(); }, async editVersion(id, version, patch) { const one = await deps.store.getPackage(id, version); if (one === undefined) throw new StaleVersionError(id, version); const to = patch.version; if (to !== undefined && to !== version) { const all = await deps.store.listPackages(); if (all.some((p) => p.id === id && p.version === to)) throw new VersionTakenError(id, to); /* * 在途钉版不许改号:那些单正按这个号找定义,搬号等于在它们脚下抽地板。 * 已完结的单不拦 —— 它们的 `packageVersion` 由 store 跟着一起改指,留痕仍指同一份定义。 */ const pinned = await deps.store.countPinnedInFlight(id, version); if (pinned > 0) throw new PinnedVersionError(id, version, pinned); await deps.store.renumberVersion(id, version, to); } /* * 名不设闸:**在途钉版也允许改** —— 名字是标签不参与流转, * 那几道闸守的是能跑的东西(定义与号),与名字无关。 */ if (patch.name !== undefined && patch.name !== one.name) { await deps.store.renameVersion(id, to ?? version, patch.name); } const after = await deps.store.listPackages(); return after.filter((p) => p.id === id).sort((a, b) => b.version - a.version); }, async listPackages() { return deps.store.listPackages(); }, async getPackage(id, version) { return deps.store.getPackage(id, version); }, /** 画图即校验:发布前跑闭合校验 */ async validate(pkg) { return checkClosure({ pkg, contracts: await deps.activities.faces(), }); }, /** 不闭合不得发布 */ /** 起单:经调用接口调起流程包,生成一条工作流 */ async start(input: StartWorkflowInput) { const pkg = input.packageVersion === undefined ? await activeVersion(input.packageId) : await requirePackage(input.packageId, input.packageVersion); // 指名道姓要某一版也得是生效版 —— 「一次只能激活一个版本」对新起单是硬约束 if (pkg.state !== 'active') throw new Error(`该版本未生效,不得起单:${pkg.id}@${pkg.version}`); /* * 宽进(2026-07-30,用户定):调用接口的输入**可以不定或为空** —— 门口不设缺项闸。 * 收敛点在活动:进活动后受理归一,信息不全由任务侧派补全单收齐。 * (从前这里「先校验后建单」,缺一个字段整条流都起不来 —— 现实是客户来信 * 本来就常缺客户码,拦在门外等于要求外界替系统把信息备齐。) */ // 工作组按**来源人**装载(ADR-011):无来源人、或此人没建组 → 不装载,全走四策略 const binding = input.source.by === undefined ? undefined : await deps.org.workGroupBinding(input.source.by); /* * 负责人合并(2026-07-30):运营中心 · 工作流里配的「节点负责人」垫在显式指派之下。 * 显式指派赢 —— 测试台点名是这一单的意图;运营配置是常态,轮值是兜底。 */ const opsOwners = await deps.store.nodeOwners(input.packageId); const assignments: Record = { ...opsOwners, ...input.assignments }; const record: WorkflowRecord = { id: deps.ids.next(), packageId: pkg.id, packageVersion: pkg.version, subject: input.subject, state: 'running', source: input.source, initiation: input.data, ports: {}, vars: {}, startedAt: deps.clock.now().toISOString(), ...(binding === undefined ? {} : { workGroup: { ...binding } }), // 节点指派三层:显式指派(测试台定向)> 运营配置的节点负责人 > 轮值 ...(Object.keys(assignments).length === 0 ? {} : { assignments }), }; await deps.store.insertWorkflow(record); await deps.events.publish({ workflowId: record.id, kind: 'workflow-started', detail: { source: input.source } }); await runtime.begin(record, pkg); return record; }, async getWorkflow(id) { return requireWorkflow(id); }, async listWorkflows(limit = 50) { return deps.store.listWorkflows(limit); }, async instancesOf(workflowId) { return deps.store.listInstances(workflowId); }, async nodeOwners(packageId) { return deps.store.nodeOwners(packageId); }, async setNodeOwner(packageId, nodeId, staffId) { await deps.store.setNodeOwner(packageId, nodeId, staffId); }, /* * 执行面五方法 —— 全部委托 FlowRuntime(M5-1 切面): * 语义、留痕与副作用编排住在运行时实现里,薄壳只转发。 */ async onTaskTerminal(taskId, terminal, ports) { return runtime.onTaskTerminal(taskId, terminal, ports); }, async applyFailure(workflowId, nodeId, instance, node, terminal) { return runtime.applyFailure(workflowId, nodeId, instance, node, terminal); }, async onApprovalVerdict(workflowId, nodeId, verdict, payload) { return runtime.onApprovalVerdict(workflowId, nodeId, verdict, payload); }, async refreshState(workflowId) { return runtime.refreshState(workflowId); }, async terminate(workflowId, reason) { return runtime.terminate(workflowId, reason); }, }; }