/** * 流程引擎的确定性流转 —— 纯逻辑,零 IO、**零 LLM**(编码规范 §5.3;语义出处 OS 04)。 * * 「流程引擎 —— 确定性流转:接收调用接口的起单、按结构调度节点、把审批点卡到人的工位、 * 执行网关的分合、把每次流转记入事件。**零 LLM、零临场**。」 */ import type { Edge, ProcessNode, ProcessPackage } from '@contracts'; export type PredicateEvaluator = (expression: string, scope: Readonly>) => boolean; /** 一次流转要做的事 */ export type Dispatch = /** 起一单任务 */ | { readonly kind: 'start-task'; readonly node: Extract; readonly payload: Record } /** 把审批点卡到人的工位 */ | { readonly kind: 'raise-approval'; readonly node: Extract; readonly payload: Record } /** 流程包显式终点:按输出声明交付、按来源回执 */ | { readonly kind: 'finish'; readonly node: Extract; readonly payload: Record }; /** 边映射装配:上游输出口载荷 / 流程变量 / 起单载荷 / 常量 → 下游 data 字段 */ export type AssembleScope = { readonly ports: Readonly>>; readonly vars: Readonly>; readonly initiation: Readonly>; }; export function assemble( edge: Edge, scope: AssembleScope, evaluateExpression: (sfeel: string, scope: Readonly>) => unknown, ): Record { const out: Record = {}; const flat: Record = { ...scope.initiation, ...scope.vars, ...scope.ports }; for (const mapping of edge.mapping) { switch (mapping.source.from) { case 'output-port': out[mapping.target] = pick(scope.ports[mapping.source.port], mapping.source.path); break; case 'workflow-var': out[mapping.target] = scope.vars[mapping.source.name]; break; case 'initiation': out[mapping.target] = pick(scope.initiation, mapping.source.path); break; case 'constant': out[mapping.target] = mapping.source.value; break; case 'expression': out[mapping.target] = evaluateExpression(mapping.source.sfeel, flat); break; } } return out; } function pick(source: Readonly> | undefined, path: string): unknown { if (source === undefined) return undefined; return path.split('.').reduce((acc, seg) => { if (acc === null || acc === undefined || typeof acc !== 'object') return undefined; return (acc as Record)[seg]; }, source); } /** * 从某节点出发,算出接下来该走的边。 * XOR:条件求值,**恰取一条**;AND:全部出边同时走(分叉)。 * 条件声明式、静态可检,**不得内嵌自由逻辑**。 */ export function selectEdges( pkg: ProcessPackage, fromNodeId: string, scope: Readonly>, evaluate: PredicateEvaluator, ): Edge[] { const node = pkg.nodes.find((n) => n.id === fromNodeId); const outgoing = pkg.edges.filter((e) => e.from === fromNodeId); if (node?.kind === 'gateway' && node.gateway === 'exclusive' && node.mode === 'fork') { // 判定依据入事件记录;无条件的边作默认分支兜底 const hit = outgoing.find((e) => e.condition !== undefined && evaluate(e.condition, scope)); const fallback = outgoing.find((e) => e.condition === undefined); const chosen = hit ?? fallback; return chosen === undefined ? [] : [chosen]; } // 并行网关分叉、活动节点、审批点:全部出边 return outgoing; } /** * AND 汇合是否回齐:全部**责任分支**的来源都已收敛。 * * 网关本身不是责任单元、永远不会有实例,所以来源是网关的入边(分叉直连汇合的**直通道**) * 不参与等待 —— 分叉一发生它就算过了。若把它也算进去,汇合会永远等一个不可能到来的 * 「网关收敛」,整条流卡死在这里。 */ export function joinReady( pkg: ProcessPackage, joinNodeId: string, converged: ReadonlySet, tolerated: ReadonlySet, ): boolean { const node = pkg.nodes.find((n) => n.id === joinNodeId); /* * **异或汇合不等。** * * 上面那句「回齐」是按并行汇合写的:AND 分叉出去的每一支都要跑,所以汇合处要等齐。 * 异或汇合是另一回事 —— 两条分支互斥,**结构上只会来一条**,另一条永远不激活。 * 拿「全部入边都收敛」去要求它,就是在等一个不可能到来的分支: * 实测 G2J(寻源汇合)在「不需要寻源」这条路上永远 false,流无声停在 G2J, * 而且**连死路事件都没有** —— 边是选中了的,只是在汇合判定里被 continue 跳过, * 从外面看就是 A1 办结后什么都不再发生。 */ if (node?.kind === 'gateway' && node.gateway === 'exclusive') return true; const isGateway = (nodeId: string): boolean => pkg.nodes.find((n) => n.id === nodeId)?.kind === 'gateway'; const inbound = pkg.edges.filter((e) => e.to === joinNodeId && !isGateway(e.from)); // 容忍分支以「已消化」计,不阻汇合 return inbound.every((e) => converged.has(e.from) || tolerated.has(e.from)); } /** 节点该被怎么调度 */ export function dispatchOf(node: ProcessNode, payload: Record): Dispatch | undefined { if (node.kind === 'activity') return { kind: 'start-task', node, payload }; if (node.kind === 'approval') return { kind: 'raise-approval', node, payload }; if (node.kind === 'end') return { kind: 'finish', node, payload }; return undefined; } /** 审批点的驳回走向:声明式,**不由临场决定** */ export type RejectRoute = | { readonly kind: 'reopen'; readonly nodeId: string; readonly payload: Record } | { readonly kind: 'terminate' }; export function rejectRouteOf( node: Extract, scope: AssembleScope, evaluateExpression: (sfeel: string, scope: Readonly>) => unknown, ): RejectRoute { if (node.reject.kind === 'terminate') return { kind: 'terminate' }; return { kind: 'reopen', nodeId: node.reject.nodeId, payload: assemble( { id: `${node.id}-reject`, from: node.id, to: node.reject.nodeId, mapping: node.reject.mapping, reopen: true }, scope, evaluateExpression, ), }; } /** 起点:找出流程包的入口节点 */ export function startNodeOf(pkg: ProcessPackage): ProcessNode | undefined { return pkg.nodes.find((n) => n.kind === 'start'); } /** * 推进次序 —— 按**边**做拓扑序,不看图纸坐标。 * * 「链」要回答的是「这一单接下来往哪走」,那是边的事实。曾经拿 `layout.col` 排, * 因为 v1 的包是一行从左画到右,列位恰好等于推进次序 —— 但 MVP 图是**回形走线** * (第 0 行左→右、第 1 行右→左、第 2 行折回最左),A7 的列位跟 A1 并列,链就读成了 * A1 → A7 → A2 → A6 → A3(实测)。图纸声明为「不影响语义」,那它也不配决定次序。 * * 回边(`reopen`)排除在外:它是**重新入场**,不是往前推进。留着它图上就有环, * 拓扑序无从下手(G4 → A4 会让 A4 等自己的下游)。 * * 真并行的分支(G2 分出的 A2 / A3)在图上本无先后,链却必须挑一个顺序: * 按**节点编号**定,于是同样的包在谁手上都读作同一条链。 * 进不了拓扑序的节点(不可达、或排除回边后仍有环)一律按编号缀在末尾 —— 少排序,不丢节点。 */ export function advanceOrder(pkg: ProcessPackage): readonly ProcessNode[] { const byId = new Map(pkg.nodes.map((n) => [n.id, n])); const byNodeId = (a: string, b: string) => (a < b ? -1 : a > b ? 1 : 0); const forward = pkg.edges.filter((e) => e.reopen !== true && byId.has(e.from) && byId.has(e.to)); const indegree = new Map(pkg.nodes.map((n) => [n.id, 0])); const next = new Map(); for (const e of forward) { indegree.set(e.to, (indegree.get(e.to) ?? 0) + 1); next.set(e.from, [...(next.get(e.from) ?? []), e.to]); } // Kahn:每轮从「无未满足前驱」的节点里取编号最小的,顺序才与书写无关 const ready = pkg.nodes.filter((n) => indegree.get(n.id) === 0).map((n) => n.id); const ordered: ProcessNode[] = []; const placed = new Set(); while (ready.length > 0) { ready.sort(byNodeId); const id = ready.shift()!; if (placed.has(id)) continue; placed.add(id); const node = byId.get(id); if (node !== undefined) ordered.push(node); for (const to of next.get(id) ?? []) { const left = (indegree.get(to) ?? 0) - 1; indegree.set(to, left); if (left === 0) ready.push(to); } } const stranded = pkg.nodes.filter((n) => !placed.has(n.id)).sort((a, b) => byNodeId(a.id, b.id)); return [...ordered, ...stranded]; }