forked from ChaiHao/BPOS-MVP
M5(2/3): process 用例与流转 —— 流程引擎的确定性调度
- routing domain:选边(XOR 恰取一条 + 默认分支兜底;AND 全走)、边映射装配(四类来源 + S-FEEL 转换)、AND 汇合回齐(容忍分支不阻汇合)、审批点驳回走向声明式 - process-service:起单 → 调度节点 → 负责人四策略解析(结果与依据留痕)→ 活动起单 / 审批点卡工位 / 终点收束按来源回执;任务终态到达继续流转或走失败走向四选一; 审批驳回按声明回指定节点重开新实例;状态每次变动后重算聚合投影 - adapter:流程包(id × version)、工作流实例、节点实例三表持久 —— 崩溃后可续跑 ADR 待补项登记为 debt M5-1:§6.1 的 Mastra workflow 编译器作为可替换 adapter 后补, 当前由自持确定性状态机承担,语义等价、usecase 不动 325 个测试通过,门禁五道全绿 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01KeDBx2PHLavbURXB8LJJ7m
This commit is contained in:
JCAI Main Agent
co-authored by
Claude Opus 5
Claude Opus 5
parent
291fc39179
commit
a8f6eef610
@@ -0,0 +1,370 @@
|
||||
/**
|
||||
* process 用例 —— 流程引擎:确定性流转,**零 LLM、零临场**(编码规范 §5.3;语义出处 OS 03 / 04)。
|
||||
*
|
||||
* 起单 · 调度节点 · 审批点卡工位 · 网关分合 · 每次流转记入事件。
|
||||
*/
|
||||
import type { ProcessNode, ProcessPackage, StaffId, TerminalState, WorkflowState } from '@contracts';
|
||||
import type { ProcessService, StartWorkflowInput } from '../../contract.ts';
|
||||
import { NotPublishableError, PositionVacantError } from '../../contract.ts';
|
||||
import type {
|
||||
ActivityDirectory,
|
||||
ApprovalGateway,
|
||||
Clock,
|
||||
FlowEventPublisher,
|
||||
IdGenerator,
|
||||
OrgDirectory,
|
||||
ProcessStore,
|
||||
ReceiptNotifier,
|
||||
Sfeel,
|
||||
TaskGateway,
|
||||
WorkflowRecord,
|
||||
} from '../../port.ts';
|
||||
import { checkClosure } from '../domain/closure.ts';
|
||||
import { aggregateWorkflowState, isDigested, nextInstanceNumber, resolveOwner } from '../domain/owner-resolution.ts';
|
||||
import type { NodeInstance } from '../domain/owner-resolution.ts';
|
||||
import { assemble, dispatchOf, joinReady, rejectRouteOf, selectEdges, startNodeOf } from '../domain/routing.ts';
|
||||
|
||||
export type ProcessServiceDeps = {
|
||||
readonly store: ProcessStore;
|
||||
readonly activities: ActivityDirectory;
|
||||
readonly org: OrgDirectory;
|
||||
readonly tasks: TaskGateway;
|
||||
readonly approvals: ApprovalGateway;
|
||||
readonly sfeel: Sfeel;
|
||||
readonly events: FlowEventPublisher;
|
||||
readonly receipts: ReceiptNotifier;
|
||||
readonly clock: Clock;
|
||||
readonly ids: IdGenerator;
|
||||
};
|
||||
|
||||
export function createProcessService(deps: ProcessServiceDeps): ProcessService {
|
||||
async function requirePackage(id: string, version?: number): Promise<ProcessPackage> {
|
||||
const pkg = await deps.store.getPackage(id, version);
|
||||
if (pkg === undefined) throw new Error(`流程包不存在:${id}`);
|
||||
return pkg;
|
||||
}
|
||||
|
||||
async function requireWorkflow(id: string): Promise<WorkflowRecord> {
|
||||
const record = await deps.store.getWorkflow(id);
|
||||
if (record === undefined) throw new Error(`工作流不存在:${id}`);
|
||||
return record;
|
||||
}
|
||||
|
||||
function scopeOf(record: WorkflowRecord) {
|
||||
return { ports: record.ports, vars: record.vars, initiation: record.initiation };
|
||||
}
|
||||
|
||||
/** 起一个节点:活动 → 起单;审批点 → 卡工位;终点 → 收束 */
|
||||
async function enter(
|
||||
record: WorkflowRecord,
|
||||
pkg: ProcessPackage,
|
||||
node: ProcessNode,
|
||||
payload: Record<string, unknown>,
|
||||
): Promise<void> {
|
||||
const dispatch = dispatchOf(node, payload);
|
||||
|
||||
if (dispatch === undefined) {
|
||||
// 网关本身不是责任单元:直接继续流转
|
||||
await advanceFrom(record, pkg, node.id, payload);
|
||||
return;
|
||||
}
|
||||
|
||||
const instances = await deps.store.listInstances(record.id);
|
||||
const instance = nextInstanceNumber(instances, node.id);
|
||||
|
||||
if (dispatch.kind === 'finish') {
|
||||
await deps.store.insertInstance(record.id, {
|
||||
nodeId: node.id,
|
||||
instance,
|
||||
running: false,
|
||||
waiting: false,
|
||||
activated: true,
|
||||
terminal: 'released',
|
||||
});
|
||||
const finished = await deps.store.updateWorkflow(record.id, {
|
||||
state: 'completed',
|
||||
finishedAt: deps.clock.now().toISOString(),
|
||||
ports: { ...record.ports, __output: payload },
|
||||
});
|
||||
await deps.events.publish({ workflowId: record.id, kind: 'workflow-completed', detail: payload });
|
||||
// 凡单必有来源 —— 登记人收消息回执
|
||||
if (record.source.by !== undefined) {
|
||||
await deps.receipts.notify({
|
||||
to: record.source.by as StaffId,
|
||||
workflowId: record.id,
|
||||
subject: record.subject,
|
||||
state: finished.state,
|
||||
});
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
// 负责人 / 审批人:编排声明岗位,起单时确定性解析;结果与依据留痕
|
||||
const spec = dispatch.kind === 'start-task' ? dispatch.node.owner : dispatch.node.approver;
|
||||
const candidates = await deps.org.candidates(spec.position);
|
||||
const resolution = resolveOwner(spec, { candidates });
|
||||
if (resolution.kind === 'vacant') throw new PositionVacantError(spec.position);
|
||||
|
||||
await deps.events.publish({
|
||||
workflowId: record.id,
|
||||
kind: 'owner-resolved',
|
||||
detail: { nodeId: node.id, position: spec.position, staffId: resolution.staffId, by: resolution.by, rationale: resolution.rationale },
|
||||
});
|
||||
|
||||
if (dispatch.kind === 'start-task') {
|
||||
const task = await deps.tasks.start({
|
||||
activityId: dispatch.node.activityId,
|
||||
ownerId: resolution.staffId as StaffId,
|
||||
title: dispatch.node.label,
|
||||
workflowId: record.id,
|
||||
nodeId: node.id,
|
||||
data: payload,
|
||||
...(dispatch.node.message === undefined ? {} : { message: dispatch.node.message }),
|
||||
});
|
||||
await deps.store.insertInstance(record.id, {
|
||||
nodeId: node.id,
|
||||
instance,
|
||||
running: true,
|
||||
waiting: false,
|
||||
activated: true,
|
||||
taskId: task.id,
|
||||
});
|
||||
await deps.events.publish({ workflowId: record.id, kind: 'task-started', detail: { nodeId: node.id, taskId: task.id } });
|
||||
return;
|
||||
}
|
||||
|
||||
const approval = await deps.approvals.raise({
|
||||
workflowId: record.id,
|
||||
nodeId: node.id,
|
||||
approverId: resolution.staffId as StaffId,
|
||||
subject: dispatch.node.label,
|
||||
basis: Object.entries(payload).map(([k, v]) => ({ k, v: String(v) })),
|
||||
});
|
||||
await deps.store.insertInstance(record.id, {
|
||||
nodeId: node.id,
|
||||
instance,
|
||||
running: false,
|
||||
waiting: true,
|
||||
activated: true,
|
||||
approvalId: approval.id,
|
||||
});
|
||||
await deps.events.publish({ workflowId: record.id, kind: 'approval-raised', detail: { nodeId: node.id, approvalId: approval.id } });
|
||||
}
|
||||
|
||||
/** 从某节点继续流转:选边 → 装配 → 进下一节点(AND 汇合要回齐) */
|
||||
async function advanceFrom(
|
||||
record: WorkflowRecord,
|
||||
pkg: ProcessPackage,
|
||||
fromNodeId: string,
|
||||
payload: Record<string, unknown>,
|
||||
): Promise<void> {
|
||||
const scope = scopeOf(record);
|
||||
const flat = { ...scope.initiation, ...scope.vars, ...payload };
|
||||
const edges = selectEdges(pkg, fromNodeId, flat, deps.sfeel.condition);
|
||||
|
||||
for (const edge of edges) {
|
||||
const target = pkg.nodes.find((n) => n.id === edge.to);
|
||||
if (target === undefined) continue;
|
||||
|
||||
if (target.kind === 'gateway' && target.mode === 'join') {
|
||||
const instances = await deps.store.listInstances(record.id);
|
||||
const converged = new Set(instances.filter((i) => i.terminal === 'released').map((i) => i.nodeId));
|
||||
const tolerated = new Set(instances.filter((i) => i.failureDigested === true).map((i) => i.nodeId));
|
||||
// 未回齐就等着 —— 别的分支到了会再来一次
|
||||
if (!joinReady(pkg, target.id, converged, tolerated)) continue;
|
||||
}
|
||||
|
||||
const assembled = assemble(edge, { ...scope, ports: { ...scope.ports, [fromNodeId]: payload } }, deps.sfeel.expression);
|
||||
await enter(record, pkg, target, assembled);
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
async savePackage(pkg) {
|
||||
await deps.store.savePackage(pkg);
|
||||
return pkg;
|
||||
},
|
||||
|
||||
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(),
|
||||
positions: await deps.org.positions(),
|
||||
});
|
||||
},
|
||||
|
||||
/** 不闭合不得发布 */
|
||||
async publish(id, version) {
|
||||
const pkg = await requirePackage(id, version);
|
||||
const report = await this.validate(pkg);
|
||||
if (!report.publishable) throw new NotPublishableError(report);
|
||||
await deps.store.publishPackage(id, version);
|
||||
return report;
|
||||
},
|
||||
|
||||
/** 起单:经调用接口调起流程包,生成一条工作流 */
|
||||
async start(input: StartWorkflowInput) {
|
||||
const pkg = await requirePackage(input.packageId, input.packageVersion);
|
||||
if (pkg.state !== 'published') throw new Error(`流程包未发布,不得起单:${pkg.id}@${pkg.version}`);
|
||||
|
||||
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(),
|
||||
};
|
||||
await deps.store.insertWorkflow(record);
|
||||
await deps.events.publish({ workflowId: record.id, kind: 'workflow-started', detail: { source: input.source } });
|
||||
|
||||
const start = startNodeOf(pkg);
|
||||
if (start !== undefined) await advanceFrom(record, pkg, start.id, input.data);
|
||||
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 onTaskTerminal(taskId, terminal: TerminalState, ports: Record<string, Record<string, unknown>>) {
|
||||
const located = await deps.store.findInstanceByTask(taskId);
|
||||
if (located === undefined) return undefined;
|
||||
|
||||
const record = await requireWorkflow(located.workflowId);
|
||||
const pkg = await requirePackage(record.packageId, record.packageVersion);
|
||||
const node = pkg.nodes.find((n) => n.id === located.nodeId);
|
||||
|
||||
await deps.store.updateInstance(record.id, located.nodeId, located.instance, {
|
||||
running: false,
|
||||
waiting: false,
|
||||
terminal,
|
||||
});
|
||||
await deps.events.publish({ workflowId: record.id, kind: 'task-terminal', detail: { taskId, terminal } });
|
||||
|
||||
const merged = await deps.store.updateWorkflow(record.id, { ports: { ...record.ports, ...ports } });
|
||||
|
||||
if (terminal === 'released') {
|
||||
// 落地收敛后才发射输出口 —— 下游只见落地完成的任务
|
||||
const payload = Object.values(ports)[0] ?? {};
|
||||
await advanceFrom(merged, pkg, located.nodeId, payload);
|
||||
} else if (node?.kind === 'activity') {
|
||||
// 非完成终态触发**声明的失败走向**;流程层不自动重试
|
||||
await this.applyFailure(record.id, located.nodeId, located.instance, node, terminal);
|
||||
}
|
||||
|
||||
return this.refreshState(record.id);
|
||||
},
|
||||
|
||||
/** 失败走向四选一:终止流(默认)/ 改道 / 升审批 / 容忍 */
|
||||
async applyFailure(workflowId, nodeId, instance, node, terminal) {
|
||||
const record = await requireWorkflow(workflowId);
|
||||
const pkg = await requirePackage(record.packageId, record.packageVersion);
|
||||
const disposition = node.failure;
|
||||
const digested = isDigested(disposition.kind);
|
||||
|
||||
await deps.store.updateInstance(workflowId, nodeId, instance, { failureDigested: digested });
|
||||
await deps.events.publish({
|
||||
workflowId,
|
||||
kind: 'failure-disposition',
|
||||
detail: { nodeId, terminal, disposition: disposition.kind },
|
||||
});
|
||||
|
||||
if (disposition.kind === 'reroute') {
|
||||
const edge = pkg.edges.find((e) => e.id === disposition.edgeId);
|
||||
const target = edge === undefined ? undefined : pkg.nodes.find((n) => n.id === edge.to);
|
||||
if (edge !== undefined && target !== undefined) {
|
||||
// 失败载荷 = 终态、原因码与说明
|
||||
const payload = assemble(edge, scopeOf(record), deps.sfeel.expression);
|
||||
await enter(record, pkg, target, { ...payload, __terminal: terminal });
|
||||
}
|
||||
}
|
||||
|
||||
if (disposition.kind === 'escalate') {
|
||||
const target = pkg.nodes.find((n) => n.id === disposition.approvalNodeId);
|
||||
if (target !== undefined) await enter(record, pkg, target, { __terminal: terminal, __from: nodeId });
|
||||
}
|
||||
|
||||
return this.refreshState(workflowId);
|
||||
},
|
||||
|
||||
/** 审批裁决:通过继续流转;驳回按**声明的走向**回指定节点重开新实例 */
|
||||
async onApprovalVerdict(workflowId, nodeId, verdict, payload) {
|
||||
const record = await requireWorkflow(workflowId);
|
||||
const pkg = await requirePackage(record.packageId, record.packageVersion);
|
||||
const node = pkg.nodes.find((n) => n.id === nodeId);
|
||||
if (node?.kind !== 'approval') throw new Error(`节点不是审批点:${nodeId}`);
|
||||
|
||||
const instances = await deps.store.listInstances(workflowId);
|
||||
const current = instances.filter((i) => i.nodeId === nodeId).at(-1);
|
||||
if (current !== undefined) {
|
||||
await deps.store.updateInstance(workflowId, nodeId, current.instance, {
|
||||
waiting: false,
|
||||
terminal: 'released',
|
||||
});
|
||||
}
|
||||
await deps.events.publish({ workflowId, kind: 'approval-verdict', detail: { nodeId, verdict } });
|
||||
|
||||
if (verdict === 'pass') {
|
||||
await advanceFrom(record, pkg, nodeId, payload);
|
||||
} else {
|
||||
const route = rejectRouteOf(node, scopeOf(record), deps.sfeel.expression);
|
||||
if (route.kind === 'terminate') {
|
||||
await deps.store.updateWorkflow(workflowId, { state: 'terminated', finishedAt: deps.clock.now().toISOString() });
|
||||
} else {
|
||||
// 回指定节点**重开新实例**(同节点、新任务 ID、留痕关联),不复活旧任务
|
||||
const target = pkg.nodes.find((n) => n.id === route.nodeId);
|
||||
if (target !== undefined) await enter(record, pkg, target, { ...route.payload, ...payload });
|
||||
}
|
||||
}
|
||||
return this.refreshState(workflowId);
|
||||
},
|
||||
|
||||
/** 工作流状态是聚合投影 —— 每次变动后重算 */
|
||||
async refreshState(workflowId): Promise<WorkflowState> {
|
||||
const record = await requireWorkflow(workflowId);
|
||||
if (record.state === 'terminated') return 'terminated';
|
||||
const instances = await deps.store.listInstances(workflowId);
|
||||
const state = aggregateWorkflowState(instances as NodeInstance[], false);
|
||||
if (state !== record.state) {
|
||||
await deps.store.updateWorkflow(workflowId, {
|
||||
state,
|
||||
...(state === 'completed' ? { finishedAt: deps.clock.now().toISOString() } : {}),
|
||||
});
|
||||
}
|
||||
return state;
|
||||
},
|
||||
|
||||
async terminate(workflowId, reason) {
|
||||
await deps.store.updateWorkflow(workflowId, {
|
||||
state: 'terminated',
|
||||
finishedAt: deps.clock.now().toISOString(),
|
||||
});
|
||||
// 工作流终止级联:在途任务终止、待答卡与征答消息作废、审批点撤卡
|
||||
await deps.events.publish({ workflowId, kind: 'workflow-terminated', detail: { reason } });
|
||||
return 'terminated';
|
||||
},
|
||||
};
|
||||
}
|
||||
Reference in New Issue
Block a user