M4: 范式引擎与任务运行时
skill-engine(L4):载方法不载权力(白名单外工具即拒)、结晶片段经 ToolInvoker 留痕、 装配入信封、技能目录渐进披露 flow-engine(L5):四种步型、确定性谓词跳步、重试与声明给定的兜底、 异步步即监听(时限必配)、写类工具登记动作意向、步序位置持久 loop-engine(L5):自研 CMMN 子集解析器(表外构件直接报错,子集封顶可机器证明)、 两副标准生命周期与迁移拓扑(引擎不得私改)、Enabled ≠ Active、 语义性前提标待裁量评估(引擎不代裁量)、DMN 确定性求值 + 命中行留痕、 DRD 缺口演绎与拓扑序、双向驱动与交集导航(断点即人环)、三型行动、 终局唯放行与 Harness task(L6):终态五名裁定(released 唯一成功)、Harness 预算环外强制与停滞熔断 (监听中是健康态)、控制三语义(叫停不是终态,记最重档纠正)、改派留痕通知双方、 断点续跑三份持久状态、封卷触发蒸馏钩子 301 个测试通过,门禁五道全绿 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
2850e6d4ef
commit
3c974d2860
@@ -0,0 +1,14 @@
|
||||
/**
|
||||
* flow-engine 的 sys schema 表(编码规范 §7 / ADR-002)。
|
||||
* 运行时实例状态 —— 断点续跑的三份持久状态之一(OS 09)。
|
||||
*/
|
||||
import { jsonb, text, timestamp } from 'drizzle-orm/pg-core';
|
||||
import { sys } from '@shared/db.ts';
|
||||
|
||||
export const flowStateTable = sys.table('flow_state', {
|
||||
taskId: text('task_id').primaryKey(),
|
||||
model: jsonb('model').notNull(),
|
||||
/** 步序位置 */
|
||||
position: jsonb('position').notNull(),
|
||||
updatedAt: timestamp('updated_at', { withTimezone: true }).notNull().defaultNow(),
|
||||
});
|
||||
@@ -0,0 +1,30 @@
|
||||
/**
|
||||
* flow-engine 存储 adapter —— Drizzle 只准出现在 adapter(编码规范 §5.3)。
|
||||
*/
|
||||
import { eq } from 'drizzle-orm';
|
||||
import type { FlowModel, TaskId } from '@contracts';
|
||||
import { flowModelSchema } from '@contracts';
|
||||
import type { Db } from '@shared/db.ts';
|
||||
import type { FlowStateStore } from '../../port.ts';
|
||||
import type { FlowPosition } from '../domain/step-sequence.ts';
|
||||
import { flowStateTable } from './flow-state-schema.ts';
|
||||
|
||||
export function createFlowStateStore(db: Db): FlowStateStore {
|
||||
return {
|
||||
async save(taskId: TaskId, model: FlowModel, position: FlowPosition) {
|
||||
await db
|
||||
.insert(flowStateTable)
|
||||
.values({ taskId, model, position, updatedAt: new Date() })
|
||||
.onConflictDoUpdate({
|
||||
target: flowStateTable.taskId,
|
||||
set: { model, position, updatedAt: new Date() },
|
||||
});
|
||||
},
|
||||
async load(taskId: TaskId) {
|
||||
const rows = await db.select().from(flowStateTable).where(eq(flowStateTable.taskId, taskId)).limit(1);
|
||||
const row = rows[0];
|
||||
if (row === undefined) return undefined;
|
||||
return { model: flowModelSchema.parse(row.model), position: row.position as FlowPosition };
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,123 @@
|
||||
/**
|
||||
* flow 步序推进 —— 纯逻辑,零 IO(编码规范 §5.3;语义出处 AM 05)。
|
||||
*
|
||||
* 「flow 是活动内的**确定性范式**:业务模型 = 步序声明 —— 一条谱;
|
||||
* 引擎照谱执行,**零临场调度**,调度权归系统。
|
||||
* 差异只有一处:下一步由**声明给定**,不由 AI 决断。
|
||||
* LLM 在 flow 里可以出现,但只以『被调用的工具』出现,不是调度者。」
|
||||
*/
|
||||
import type { FlowStep } from '@contracts';
|
||||
|
||||
export const STEP_STATES = ['pending', 'running', 'awaiting', 'done', 'skipped', 'failed'] as const;
|
||||
export type StepState = (typeof STEP_STATES)[number];
|
||||
|
||||
export type StepRuntime = {
|
||||
readonly name: string;
|
||||
readonly state: StepState;
|
||||
readonly attempt: number;
|
||||
readonly output?: Record<string, unknown>;
|
||||
readonly error?: string;
|
||||
};
|
||||
|
||||
export type FlowPosition = {
|
||||
/** 步序位置 —— 三份持久状态之一(OS 09 断点续跑) */
|
||||
readonly index: number;
|
||||
readonly steps: readonly StepRuntime[];
|
||||
};
|
||||
|
||||
export function initialPosition(steps: readonly FlowStep[]): FlowPosition {
|
||||
return { index: 0, steps: steps.map((s) => ({ name: s.name, state: 'pending', attempt: 0 })) };
|
||||
}
|
||||
|
||||
/** 条件跳步:条件必须是任务数据上的**确定性谓词**,由引擎求值,零 LLM(AM 5.2) */
|
||||
export type PredicateEvaluator = (expression: string, scope: Readonly<Record<string, unknown>>) => boolean;
|
||||
|
||||
export type NextAction =
|
||||
| { readonly kind: 'run'; readonly step: FlowStep; readonly index: number }
|
||||
| { readonly kind: 'skip'; readonly step: FlowStep; readonly index: number; readonly reason: string }
|
||||
| { readonly kind: 'await'; readonly step: FlowStep; readonly index: number }
|
||||
| { readonly kind: 'finished' };
|
||||
|
||||
/**
|
||||
* 下一步由声明给定。异步步(人类工具步 / 等事件的步)进入 `await` —— 完成条件 = 监听。
|
||||
*/
|
||||
export function nextAction(
|
||||
declaration: readonly FlowStep[],
|
||||
position: FlowPosition,
|
||||
scope: Readonly<Record<string, unknown>>,
|
||||
evaluate: PredicateEvaluator,
|
||||
): NextAction {
|
||||
for (let i = position.index; i < declaration.length; i += 1) {
|
||||
const step = declaration[i]!;
|
||||
const runtime = position.steps[i];
|
||||
if (runtime?.state === 'done' || runtime?.state === 'skipped') continue;
|
||||
if (runtime?.state === 'awaiting') return { kind: 'await', step, index: i };
|
||||
|
||||
if (step.when !== undefined && !evaluate(step.when, scope)) {
|
||||
return { kind: 'skip', step, index: i, reason: `条件不成立:${step.when}` };
|
||||
}
|
||||
return { kind: 'run', step, index: i };
|
||||
}
|
||||
return { kind: 'finished' };
|
||||
}
|
||||
|
||||
export function markStep(position: FlowPosition, index: number, patch: Partial<StepRuntime>): FlowPosition {
|
||||
const steps = position.steps.map((s, i) => (i === index ? { ...s, ...patch } : s));
|
||||
const advanced = patch.state === 'done' || patch.state === 'skipped';
|
||||
return { index: advanced ? index + 1 : index, steps };
|
||||
}
|
||||
|
||||
/* ── 失败与缺口:声明给定,不临场(AM 5.3) ────────────────────────── */
|
||||
|
||||
export type FailureAction =
|
||||
| { readonly kind: 'retry'; readonly attempt: number; readonly backoff: string }
|
||||
/** 转人兜底:改发人类工具卡,卡内容随声明 */
|
||||
| { readonly kind: 'human'; readonly workOrder: 'execution' | 'acceptance' }
|
||||
/** 受阻待人:挂起,携阻因 */
|
||||
| { readonly kind: 'blocked' };
|
||||
|
||||
/**
|
||||
* 「每步声明重试次数与间隔;重试仍败,按声明走下一步。
|
||||
* **flow 内不得出现『由 AI 决定怎么办』的失败处理** —— 需要临场,即说明该活动不该是 flow。」
|
||||
*/
|
||||
export function failureAction(step: FlowStep, attempt: number): FailureAction {
|
||||
if (attempt < step.failure.retry.times) {
|
||||
return { kind: 'retry', attempt: attempt + 1, backoff: step.failure.retry.backoff };
|
||||
}
|
||||
return step.failure.fallback.kind === 'human'
|
||||
? { kind: 'human', workOrder: step.failure.fallback.workOrder }
|
||||
: { kind: 'blocked' };
|
||||
}
|
||||
|
||||
/* ── 静态检查:谱必须全程可静态检查(AM 12.3) ────────────────────── */
|
||||
|
||||
export type FlowIssue = { readonly step: string; readonly message: string };
|
||||
|
||||
export function checkDeclaration(steps: readonly FlowStep[]): FlowIssue[] {
|
||||
const out: FlowIssue[] = [];
|
||||
const names = new Set<string>();
|
||||
for (const step of steps) {
|
||||
if (names.has(step.name)) out.push({ step: step.name, message: '步名重复' });
|
||||
names.add(step.name);
|
||||
// 步序声明完整:步型 / 入参映射 / 出参写入 / 完成条件 / 失败策略齐备
|
||||
if (step.failure.retry.times < 0) out.push({ step: step.name, message: '重试次数不得为负' });
|
||||
if (step.kind === 'human' && step.deadline.trim() === '') {
|
||||
out.push({ step: step.name, message: '人类工具步必配时限 —— 无死等' });
|
||||
}
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
/** 终态映射(AM 5.3):放行完成 = 末步验收工单「放行通过」 */
|
||||
export function lastStepIsAcceptance(steps: readonly FlowStep[]): boolean {
|
||||
const last = steps.at(-1);
|
||||
return last !== undefined && last.kind === 'human' && last.workOrder === 'acceptance';
|
||||
}
|
||||
|
||||
/** 并行组:组内步无依赖、**全完成再继续** */
|
||||
export function parallelGroupComplete(position: FlowPosition, members: readonly string[]): boolean {
|
||||
return members.every((name) => {
|
||||
const runtime = position.steps.find((s) => s.name === name);
|
||||
return runtime?.state === 'done' || runtime?.state === 'skipped';
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,279 @@
|
||||
/**
|
||||
* flow 运行器 —— 照谱执行,零临场调度(编码规范 §5.3;语义出处 AM 05)。
|
||||
*/
|
||||
import type { FlowStep, TaskId } from '@contracts';
|
||||
import type { AdvanceOutcome, FlowEngineService } from '../../contract.ts';
|
||||
import { FlowDeclarationInvalidError } from '../../contract.ts';
|
||||
import type {
|
||||
CardGateway,
|
||||
Clock,
|
||||
FlowStateStore,
|
||||
ListenerRegistrar,
|
||||
LlmStepGateway,
|
||||
MemoryRecorder,
|
||||
PredicateEvaluator,
|
||||
TaskContext,
|
||||
TaskDataGateway,
|
||||
ToolStepInvoker,
|
||||
} from '../../port.ts';
|
||||
import type { FlowPosition } from '../domain/step-sequence.ts';
|
||||
import {
|
||||
checkDeclaration,
|
||||
failureAction,
|
||||
initialPosition,
|
||||
markStep,
|
||||
nextAction,
|
||||
} from '../domain/step-sequence.ts';
|
||||
|
||||
export type FlowRunnerDeps = {
|
||||
readonly state: FlowStateStore;
|
||||
readonly evaluate: PredicateEvaluator;
|
||||
readonly data: TaskDataGateway;
|
||||
readonly tools: ToolStepInvoker;
|
||||
readonly llm: LlmStepGateway;
|
||||
readonly cards: CardGateway;
|
||||
readonly listeners: ListenerRegistrar;
|
||||
readonly memory: MemoryRecorder;
|
||||
readonly tasks: TaskContext;
|
||||
readonly clock: Clock;
|
||||
};
|
||||
|
||||
export function createFlowRunner(deps: FlowRunnerDeps): FlowEngineService {
|
||||
async function loaded(taskId: TaskId) {
|
||||
const found = await deps.state.load(taskId);
|
||||
if (found === undefined) throw new Error(`flow 未装载:${taskId}`);
|
||||
return found;
|
||||
}
|
||||
|
||||
/** 出参写入落**草稿层**(草稿态);放行前不落事实 */
|
||||
async function writeOutput(taskId: TaskId, step: FlowStep, output: Record<string, unknown>): Promise<void> {
|
||||
if (step.output === undefined) return;
|
||||
const patch: Record<string, unknown> = {};
|
||||
for (const [field, path] of Object.entries(step.output.fields)) {
|
||||
patch[field] = output[path] ?? output[field];
|
||||
}
|
||||
await deps.data.draftWrite({
|
||||
taskId,
|
||||
domain: step.output.domain,
|
||||
object: step.output.object,
|
||||
key: {},
|
||||
patch,
|
||||
});
|
||||
}
|
||||
|
||||
function inputOf(step: FlowStep, scope: Record<string, unknown>): Record<string, unknown> {
|
||||
const out: Record<string, unknown> = {};
|
||||
for (const [key, path] of Object.entries(step.input)) {
|
||||
out[key] = path.split('.').reduce<unknown>((acc, seg) => {
|
||||
if (acc === null || acc === undefined || typeof acc !== 'object') return undefined;
|
||||
return (acc as Record<string, unknown>)[seg];
|
||||
}, scope);
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
async function runOne(
|
||||
taskId: TaskId,
|
||||
step: FlowStep,
|
||||
index: number,
|
||||
position: FlowPosition,
|
||||
): Promise<AdvanceOutcome> {
|
||||
const context = await deps.tasks.of(taskId);
|
||||
const scope = await deps.data.view(taskId);
|
||||
const payload = inputOf(step, scope);
|
||||
const attempt = (position.steps[index]?.attempt ?? 0) + 1;
|
||||
|
||||
await deps.memory.append({
|
||||
taskId,
|
||||
kind: 'action',
|
||||
title: `步 ${index + 1} · ${step.label}(${step.kind})`,
|
||||
detail: { step: step.name, attempt, input: payload },
|
||||
});
|
||||
|
||||
switch (step.kind) {
|
||||
case 'tool':
|
||||
case 'pure-tool': {
|
||||
// 写类工具在 flow 里同样登记为动作意向 —— 触达世界的动作集中在放行时落地
|
||||
if (step.kind === 'tool' && (await deps.tools.isWriteEffect(step.tool))) {
|
||||
await deps.tools.registerIntent({ taskId, tool: step.tool, input: payload, label: step.label });
|
||||
const next = markStep(position, index, { state: 'done', attempt });
|
||||
await deps.state.save(taskId, (await loaded(taskId)).model, next);
|
||||
return { position: next, status: 'running' };
|
||||
}
|
||||
const result = await deps.tools.invoke({
|
||||
tool: step.tool,
|
||||
input: payload,
|
||||
activityId: context.activityId,
|
||||
taskId,
|
||||
stepNo: index,
|
||||
});
|
||||
if (!result.ok) return handleFailure(taskId, step, index, position, attempt, result.error?.message ?? '工具失败');
|
||||
if (result.output !== undefined) await writeOutput(taskId, step, result.output);
|
||||
const next = markStep(position, index, { state: 'done', attempt, ...(result.output === undefined ? {} : { output: result.output }) });
|
||||
await deps.state.save(taskId, (await loaded(taskId)).model, next);
|
||||
return { position: next, status: 'running' };
|
||||
}
|
||||
|
||||
case 'llm': {
|
||||
// LLM 只以「被调用的工具」出现,不是调度者;模板与出参 schema 随声明版本冻结
|
||||
const prompt = step.template.replace(/\{\{\s*([\w.]+)\s*\}\}/g, (_, path: string) => String(payload[path] ?? ''));
|
||||
const result = await deps.llm.generate({
|
||||
model: step.model,
|
||||
prompt,
|
||||
outputSchema: step.outputSchema,
|
||||
taskId,
|
||||
stepNo: index,
|
||||
});
|
||||
if (!result.ok) return handleFailure(taskId, step, index, position, attempt, result.error ?? 'LLM 步失败');
|
||||
const output = (result.object ?? {}) as Record<string, unknown>;
|
||||
await writeOutput(taskId, step, output);
|
||||
const next = markStep(position, index, { state: 'done', attempt, output });
|
||||
await deps.state.save(taskId, (await loaded(taskId)).model, next);
|
||||
return { position: next, status: 'running' };
|
||||
}
|
||||
|
||||
case 'human': {
|
||||
// 异步步:完成条件 = 人事件;时限必配
|
||||
const card = await deps.cards.issue({
|
||||
taskId,
|
||||
ownerId: context.ownerId,
|
||||
humanCallCap: context.humanCallCap,
|
||||
workOrder: step.workOrder,
|
||||
card: step.card,
|
||||
});
|
||||
await deps.listeners.register({
|
||||
taskId,
|
||||
eventType: 'human',
|
||||
keyFilter: { cardId: card.id },
|
||||
deadline: step.deadline,
|
||||
});
|
||||
const next = markStep(position, index, { state: 'awaiting', attempt });
|
||||
await deps.state.save(taskId, (await loaded(taskId)).model, next);
|
||||
return {
|
||||
position: next,
|
||||
status: 'awaiting',
|
||||
waitingFor: { kind: 'human', ref: card.id, deadline: step.deadline },
|
||||
};
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** 失败路径**声明给定**,不临场发明 */
|
||||
async function handleFailure(
|
||||
taskId: TaskId,
|
||||
step: FlowStep,
|
||||
index: number,
|
||||
position: FlowPosition,
|
||||
attempt: number,
|
||||
error: string,
|
||||
): Promise<AdvanceOutcome> {
|
||||
const action = failureAction(step, attempt);
|
||||
await deps.memory.append({
|
||||
taskId,
|
||||
kind: 'result',
|
||||
title: `步 ${index + 1} 失败:${error} → ${action.kind}`,
|
||||
detail: { step: step.name, attempt, action },
|
||||
});
|
||||
|
||||
if (action.kind === 'retry') {
|
||||
const next = markStep(position, index, { state: 'pending', attempt, error });
|
||||
await deps.state.save(taskId, (await loaded(taskId)).model, next);
|
||||
return { position: next, status: 'running' };
|
||||
}
|
||||
|
||||
if (action.kind === 'human') {
|
||||
const context = await deps.tasks.of(taskId);
|
||||
const card = await deps.cards.issue({
|
||||
taskId,
|
||||
ownerId: context.ownerId,
|
||||
humanCallCap: context.humanCallCap,
|
||||
workOrder: action.workOrder,
|
||||
card: { title: `步「${step.label}」重试仍败,转人兜底`, reason: error },
|
||||
});
|
||||
await deps.listeners.register({
|
||||
taskId,
|
||||
eventType: 'human',
|
||||
keyFilter: { cardId: card.id },
|
||||
deadline: 'P3D',
|
||||
});
|
||||
const next = markStep(position, index, { state: 'awaiting', attempt, error });
|
||||
await deps.state.save(taskId, (await loaded(taskId)).model, next);
|
||||
return { position: next, status: 'awaiting', waitingFor: { kind: 'human', ref: card.id, deadline: 'P3D' } };
|
||||
}
|
||||
|
||||
const next = markStep(position, index, { state: 'failed', attempt, error });
|
||||
await deps.state.save(taskId, (await loaded(taskId)).model, next);
|
||||
return { position: next, status: 'blocked', blockedReason: error };
|
||||
}
|
||||
|
||||
return {
|
||||
async load(taskId, model) {
|
||||
const issues = checkDeclaration(model.steps);
|
||||
if (issues.length > 0) throw new FlowDeclarationInvalidError(issues);
|
||||
const position = initialPosition(model.steps);
|
||||
await deps.state.save(taskId, model, position);
|
||||
await deps.memory.append({
|
||||
taskId,
|
||||
kind: 'engine-transition',
|
||||
title: `装载步序声明:${model.steps.length} 步`,
|
||||
});
|
||||
return position;
|
||||
},
|
||||
|
||||
async position(taskId) {
|
||||
return (await deps.state.load(taskId))?.position;
|
||||
},
|
||||
|
||||
async advance(taskId) {
|
||||
const { model, position } = await loaded(taskId);
|
||||
const scope = await deps.data.view(taskId);
|
||||
const action = nextAction(model.steps, position, scope, deps.evaluate);
|
||||
|
||||
switch (action.kind) {
|
||||
case 'finished': {
|
||||
// 末步是验收工单时,终态由其裁决给出;此处只报「谱走完」
|
||||
return { position, status: 'finished', released: false };
|
||||
}
|
||||
case 'skip': {
|
||||
await deps.memory.append({
|
||||
taskId,
|
||||
kind: 'engine-transition',
|
||||
title: `跳步:${action.step.label} —— ${action.reason}`,
|
||||
});
|
||||
const next = markStep(position, action.index, { state: 'skipped' });
|
||||
await deps.state.save(taskId, model, next);
|
||||
return { position: next, status: 'running' };
|
||||
}
|
||||
case 'await':
|
||||
return {
|
||||
position,
|
||||
status: 'awaiting',
|
||||
waitingFor: { kind: 'event', ref: action.step.name, deadline: 'P3D' },
|
||||
};
|
||||
case 'run':
|
||||
return runOne(taskId, action.step, action.index, position);
|
||||
}
|
||||
},
|
||||
|
||||
/** 事件即步的完成条件 —— 到达即续跑,挂起期间事件不丢 */
|
||||
async resume(taskId, event) {
|
||||
const { model, position } = await loaded(taskId);
|
||||
const index = position.steps.findIndex((s) => s.state === 'awaiting');
|
||||
if (index < 0) return { position, status: 'running' };
|
||||
|
||||
const step = model.steps[index];
|
||||
await deps.memory.append({
|
||||
taskId,
|
||||
kind: 'engine-transition',
|
||||
title: `事件到达,步 ${index + 1} 完成条件满足(${event.kind})`,
|
||||
detail: { ref: event.ref },
|
||||
});
|
||||
if (step !== undefined && event.kind === 'human') {
|
||||
await writeOutput(taskId, step, event.payload);
|
||||
}
|
||||
const next = markStep(position, index, { state: 'done' });
|
||||
await deps.state.save(taskId, model, next);
|
||||
return { position: next, status: 'running' };
|
||||
},
|
||||
};
|
||||
}
|
||||
Reference in New Issue
Block a user