ChaiHaoandClaude Opus 5
22aa44c775
把 `unified-quote` 从起单一路手工答到底(这是逐个开发活动之前该做的验证),
一路撞出五处,按撞到的顺序:
① 七个活动派出七张一模一样的「请处理」
槽位骨架给的是 `card: {}`,工单标题/字段全落回默认。收卡的人不知道自己走到
哪一步、这一步要干什么,整条流没法手工走通。→ 卡面自述:标题就是步骤标签,
依据带「第几步 / 共几步」,并如实说「这一步的自动化还没开发,先由你来办」。
② 走到第 3 步就 `exhausted` 终态、工作流转 abnormal
六个槽位共用 `humanCalls: 3`,而 A1 声明了 6 个人类步 —— 活动**结构上永远走不完**。
→ `budgetFor(n)` 按各槽位自己的步数给。步/项清单提成 `STEPS` 常量,
预算与骨架共用同一个数组,数不上不会漂移。治理台申报值也同源(否则台上摆假数)。
③ 六步答完,任务卡在「等人,却没有任何东西等人答」
`flow-runner` 写死了 `released = lastStepIsAcceptance && lastVerdictPassed` ——
末步不是验收工单,活动就永远不 release。→ 声明的六步来自 MVP 作业指导书,
不篡改语义,在其后**追加**一步「收口验收」。
④ 只走了 A1(一个节点实例)却报 `completed` ← **最危险的一处**
状态聚合把「没有在跑的了」当成完成。界面上是一单已完成的报价,业务上一步没走。
→ 「完成」凭**到过终点节点**;收敛而未到终点 = 卡住 = `abnormal`,进驾驶舱异常卡。
`aggregateWorkflowState` 收第三个入参(终点 id 集),不传退回旧口径。
⑤ 异或网关一条边都选不中时,流无声停住
④ 让它至少报异常了,但只说「有问题」不说「哪儿」。→ 发 `route-dead-end` 事件:
哪个节点、各出边的条件、判据里当时有哪些键。只记键名不记值 —— 诊断够用,
载荷不进事件明文。实测这条痕直接指出「G1 要 needAlternative,而键里没有」。
顺带:改槽位内容必须升包版(0.1.0 → 0.4.0)。「安装恒为新增、版本并存」是硬不变量,
同版重装是 no-op、manifest 一字不动 —— 我第一次改完重播种,屏幕报「已装入」而库里还是旧的。
`installSlots` 现在先问库、如实报「已装」还是「已在库,未改动 —— 要换内容请升包版」。
**走查后的诚实状态**:A1 走得完、活动 released;流**过不了 G1**,因为出口的必填去向旗
产不出来。这不是缺陷 —— 出口载荷由台账落地事实推导,槽位没实现台账写入就没有事实,
凭空捏一个就是在系统内部 mock。于是 A1 的开发完成判据也具体了:必须让
`needAlternative` / `needSourcing` 成为落地事实,否则 G1 永远判不了。债务簿已记。
另记一条闸的盲点:api:check 只快照 index/contract/port.d.ts,而 index 是 re-export 名单 ——
给纯函数加形参它看不见(本次实测)。改签名要手动同步 CONTRACT.md,别指望闸提醒。
typecheck ✓ lint ✓ depcruise 374 模块 ✓ api:check ✓ 490 测试 ✓(均以退出码验)
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KeDBx2PHLavbURXB8LJJ7m
476 lines
20 KiB
TypeScript
476 lines
20 KiB
TypeScript
/**
|
|
* process 用例 —— 流程引擎:确定性流转,**零 LLM、零临场**(编码规范 §5.3;语义出处 OS 03 / 04)。
|
|
*
|
|
* 起单 · 调度节点 · 审批点卡工位 · 网关分合 · 每次流转记入事件。
|
|
*/
|
|
import type { ProcessNode, ProcessPackage, StaffId, TerminalState, WorkflowState } from '@contracts';
|
|
import type { ProcessService, StartWorkflowInput } from '../../contract.ts';
|
|
import { CallInterfaceViolationError, NotPublishableError, PositionVacantError } from '../../contract.ts';
|
|
import type {
|
|
ActivityDirectory,
|
|
ApprovalGateway,
|
|
Clock,
|
|
FlowEventPublisher,
|
|
IdGenerator,
|
|
OrgDirectory,
|
|
ProcessStore,
|
|
ReceiptNotifier,
|
|
SchemaValidator,
|
|
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 schemas: SchemaValidator;
|
|
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 latestPublished(id: string): Promise<ProcessPackage> {
|
|
const all = await deps.store.listPackages();
|
|
const hit = all
|
|
.filter((pkg) => pkg.id === id && pkg.state === 'published')
|
|
.sort((a, b) => b.version - a.version)[0];
|
|
if (hit === undefined) throw new Error(`流程包无已发布版本,不得起单:${id}`);
|
|
return hit;
|
|
}
|
|
|
|
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);
|
|
/*
|
|
* 绑定先于策略:工作组说了这个岗位是谁,策略就无从发问(ADR-011)。
|
|
* 在岗与否在这里核 —— 绑定是**跨岗位**指派,拿该岗位的候选集核不了它。
|
|
*/
|
|
const boundTo = record.workGroup?.[spec.position];
|
|
const bound = boundTo !== undefined && (await deps.org.isActive(boundTo)) ? boundTo : undefined;
|
|
const resolution = resolveOwner(spec, { candidates, ...(bound === undefined ? {} : { bound }) });
|
|
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);
|
|
|
|
/*
|
|
* 一条边都没选中,而这个节点确实有出边 —— 流到此为止,再没有谁会来推它。
|
|
* 这是**死路**,不是终点。异或网关最容易碰上:发布闸要求它的每条出边都带条件,
|
|
* 所以没有无条件兜底边可用,判据一缺就全不成立(实测:A1 的出口少了两枚去向旗,
|
|
* G1 一条分支都选不中,流无声停在 A1)。
|
|
*
|
|
* 状态聚合已经会把它报成 `abnormal`(没到终点就不算完成),但那只说「有问题」、
|
|
* 不说「哪儿有问题」。这里留一条可诊断的痕:哪个节点、出边各自的条件、判据里到底有哪些键。
|
|
* 只记键名不记值 —— 诊断够用,载荷不进事件明文。
|
|
*/
|
|
if (edges.length === 0) {
|
|
const outgoing = pkg.edges.filter((e) => e.from === fromNodeId);
|
|
if (outgoing.length > 0) {
|
|
await deps.events.publish({
|
|
workflowId: record.id,
|
|
kind: 'route-dead-end',
|
|
detail: {
|
|
nodeId: fromNodeId,
|
|
edges: outgoing.map((e) => ({ id: e.id, to: e.to, condition: e.condition ?? '(无条件)' })),
|
|
availableKeys: Object.keys(flat).sort(),
|
|
},
|
|
});
|
|
}
|
|
}
|
|
|
|
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);
|
|
/*
|
|
* 网关不是责任单元,**按到达的东西路由**:它没有入参契约要满足,
|
|
* 上游的输出口载荷要原样穿过去,判据才读得到(如 G4 读 outcome)。
|
|
* 若只给边映射装配出来的那份,无映射的边就把判据依据整个丢掉,
|
|
* 异或网关会一条分支都选不中,整条流悄无声息地停住。
|
|
* 活动与审批点仍只收**装配后的**载荷 —— 它们有契约,不该被上游多余字段污染。
|
|
*/
|
|
const arriving = target.kind === 'gateway' ? { ...payload, ...assembled } : assembled;
|
|
try {
|
|
await enter(record, pkg, target, arriving);
|
|
} catch (error) {
|
|
/*
|
|
* 一条分支起不来,不该把整条终态上报连坐掀翻 —— 那会让**上游任务的终态也报不上去**,
|
|
* 流停在原地且一行日志都没有。这里把它记成工作流事件并落一个未激活的实例:
|
|
* 其余分支照走,这一支在治理台「需处置」里现形,等人处置。
|
|
*/
|
|
const detail = error instanceof Error ? error.message : String(error);
|
|
await deps.events.publish({
|
|
workflowId: record.id,
|
|
kind: 'node-start-failed',
|
|
detail: { nodeId: target.id, from: fromNodeId, error: detail },
|
|
});
|
|
const instances = await deps.store.listInstances(record.id);
|
|
await deps.store.insertInstance(record.id, {
|
|
nodeId: target.id,
|
|
instance: nextInstanceNumber(instances, target.id),
|
|
running: false,
|
|
waiting: true,
|
|
activated: true,
|
|
terminal: 'blocked',
|
|
});
|
|
}
|
|
}
|
|
}
|
|
|
|
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 =
|
|
input.packageVersion === undefined
|
|
? await latestPublished(input.packageId)
|
|
: await requirePackage(input.packageId, input.packageVersion);
|
|
if (pkg.state !== 'published') throw new Error(`流程包未发布,不得起单:${pkg.id}@${pkg.version}`);
|
|
|
|
// 先校验后建单:不合调用接口契约的输入**一条工作流都不留**。
|
|
// (校验若放在建单之后,失败就会留下一条永远推不动的在途单,污染在途口径。)
|
|
const check = deps.schemas.validate(pkg.callInterface.input, input.data);
|
|
if (!check.ok) throw new CallInterfaceViolationError(pkg.id, check.errors);
|
|
|
|
// 工作组按**来源人**装载(ADR-011):无来源人、或此人没建组 → 不装载,全走四策略
|
|
const binding =
|
|
input.source.by === undefined ? undefined : await deps.org.workGroupBinding(input.source.by);
|
|
|
|
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 } }),
|
|
};
|
|
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: Record<string, unknown> = Object.assign({}, ...Object.values(ports));
|
|
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 });
|
|
}
|
|
|
|
if (disposition.kind === 'tolerate') {
|
|
/*
|
|
* 容忍 = 这条支线的失败被吸收,**流继续往下走**。
|
|
* 只把实例标成「已消化」而不往下推,汇合就永远等不到它:
|
|
* `joinReady` 认 tolerated,可没人再去问一次它够不够齐。
|
|
* 载荷为空 —— 容忍分支本就没有产出,下游的必填映射不得依赖它(闭合校验已拦)。
|
|
*/
|
|
await advanceFrom(record, pkg, 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 pkg = await deps.store.getPackage(record.packageId, record.packageVersion);
|
|
const endNodeIds = (pkg?.nodes ?? []).filter((n) => n.kind === 'end').map((n) => n.id);
|
|
const state = aggregateWorkflowState(instances as NodeInstance[], false, endNodeIds);
|
|
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';
|
|
},
|
|
};
|
|
}
|