/** * 范式引擎与任务运行时的装配 —— app 是唯一组装层(编码规范 §5.2)。 * * 这里把 M4/M5 的模块用**端口**接起来:跨层协作一律经端口,模块之间不互相认识。 */ import type { Db } from '@shared/db.ts'; import type { EventService } from '@modules/event'; import type { LedgerService } from '@modules/ledger'; import type { MemoryService } from '@modules/memory'; import type { ToolboxService, ToolInvoker } from '@modules/toolbox'; import type { HumanToolService } from '@modules/human-tool'; import type { InteractionService } from '@modules/interaction'; import type { MessageService } from '@modules/message'; import type { OrgService } from '@modules/org'; import type { LlmService, ModelGateway } from '@modules/llm'; import type { ActivityRegistryService } from '@modules/activity-registry'; import type { ReleaseService } from '@modules/release'; import { createFlowRunner, createFlowStateStore } from '@modules/flow-engine'; import { createLoopStateStore, createTurnRunner } from '@modules/loop-engine'; import { createSkillEngineService } from '@modules/skill-engine'; import { createTaskService, createTaskStore } from '@modules/task'; import type { TaskService } from '@modules/task'; import { createProcessService, createProcessStore } from '@modules/process'; import type { ProcessService } from '@modules/process'; import { createTriggerService, createTriggerStore } from '@modules/trigger'; import type { TriggerService } from '@modules/trigger'; import { evaluate as evalSfeel, evaluateCondition, evaluateUnaryTest } from '@sfeel'; import type { StaffId, TaskId } from '@contracts'; import { createAgentRunner } from '../agent-runner.ts'; import type { Clock, IdGenerator } from '../runtime.ts'; export type EngineWiringDeps = { readonly db: Db; readonly clock: Clock; readonly ids: IdGenerator; readonly event: EventService; readonly ledger: LedgerService; readonly memory: MemoryService; readonly toolbox: ToolboxService; readonly toolInvoker: ToolInvoker; readonly humanTool: HumanToolService; readonly interaction: InteractionService; readonly message: MessageService; readonly org: OrgService; readonly llm: LlmService; readonly modelGateway: ModelGateway; readonly registry: ActivityRegistryService; readonly release: ReleaseService; readonly schemas: { validate(schema: Record, value: unknown): { ok: boolean; errors: string[] } }; }; export type EngineWiring = { readonly task: TaskService; readonly process: ProcessService; readonly trigger: TriggerService; }; function writeLog(line: string): void { globalThis.process.stderr.write(`${line}\n`); } /** S-FEEL 三件套 —— 求值异常按「不成立」处置,不让流转停摆 */ const sfeel = { condition: (expression: string, scope: Readonly>): boolean => { try { return evaluateCondition(expression, scope); } catch { return false; } }, expression: (expression: string, scope: Readonly>): unknown => { try { return evalSfeel(expression, scope); } catch { return undefined; } }, unaryTest: (cell: string, value: unknown, scope: Readonly>): boolean => { try { return evaluateUnaryTest(cell, value, scope); } catch { return false; } }, }; export function wireEngines(deps: EngineWiringDeps): EngineWiring { const { db, clock, ids } = deps; /* ── 任务上下文:活动 ID / 负责人 / 调人预算 —— 引擎共用 ─────────── */ const taskStore = createTaskStore(db); const taskContext = { async of(taskId: TaskId) { const task = await taskStore.get(taskId); if (task === undefined) throw new Error(`任务不存在:${taskId}`); return { activityId: task.activityId, ownerId: task.ownerId, humanCallCap: task.caps.humanCalls }; }, }; /* ── 任务数据:引擎读**任务视图** = 正式版 + 本任务草稿层 ───────── */ const taskData = { async view(taskId: TaskId) { // 引擎对任务视图求值,不依赖事件(AM 4.2) const objects = await deps.ledger.listObjects(); const scope: Record = {}; for (const address of objects) { const page = await deps.ledger.taskView(taskId, { address, where: {}, orderBy: [], limit: 200, offset: 0, }); scope[`${address.domain}.${address.object}`] = page.rows; } return scope; }, async draftWrite(input: { taskId: TaskId; domain: string; object: string; key: Record; patch: Record; }) { // 一切过程写入落草稿层 —— 放行前不落事实 await deps.ledger.draftWrite({ taskId: input.taskId, address: { domain: input.domain, object: input.object }, key: input.key, patch: input.patch, }); }, }; const memoryRecorder = { append: async (input: { taskId: TaskId; kind: string; title: string; detail?: Record }) => { await deps.memory.append({ taskId: input.taskId, kind: input.kind as never, title: input.title, ...(input.detail === undefined ? {} : { detail: input.detail }), stamp: { contractVersion: '1', businessModelVersion: '1', skillVersions: [] }, }); }, }; const listeners = { async register(input: { taskId: TaskId; eventType: 'data' | 'time' | 'human'; keyFilter: Record; deadline: string }) { const listener = await deps.event.registerListener(input.taskId, { origin: 'dynamic', eventType: input.eventType, keyFilter: input.keyFilter, semantic: 'complete-step', deadline: input.deadline, escalation: ['nudge', 'escalate'], }); return { id: listener.id }; }, }; const cards = { async issue(input: { taskId: TaskId; ownerId: StaffId; humanCallCap: number; workOrder: 'execution' | 'acceptance'; card: Record; }) { const context = { taskId: input.taskId, ownerId: input.ownerId, humanCallCap: input.humanCallCap }; const title = String(input.card['title'] ?? '请处理'); const basis = Object.entries(input.card) .filter(([k]) => k !== 'title' && k !== 'fields') .map(([k, v]) => ({ k, v: String(v) })); if (input.workOrder === 'acceptance') { const issued = await deps.humanTool.issueAcceptance(context, { title, basis: basis.length > 0 ? basis : [{ k: '来源', v: '步序声明' }], deliverable: String(input.card['deliverable'] ?? 'deliverable') as never, criteria: [{ label: '成果齐备', level: 'L5', pass: false }], fields: [], }); return { id: issued.order.id }; } const fields = Array.isArray(input.card['fields']) ? (input.card['fields'] as string[]).map((key) => ({ key, label: key, kind: 'text' as const, required: true })) : [{ key: 'result', label: '结果', kind: 'textarea' as const, required: true }]; const issued = await deps.humanTool.issueExecution(context, { title, basis: basis.length > 0 ? basis : [{ k: '来源', v: '步序声明' }], ask: String(input.card['reason'] ?? title), fields, }); return { id: issued.order.id }; }, }; const toolGateway = { invoke: async (input: { tool: string; input: Record; activityId: string; taskId: TaskId; stepNo?: number }) => { const result = await deps.toolInvoker.invoke({ tool: input.tool, input: input.input, activityId: input.activityId, taskId: input.taskId, ...(input.stepNo === undefined ? {} : { stepNo: input.stepNo }), }); return { ok: result.ok, ...(result.output === undefined ? {} : { output: result.output }), ...(result.error === undefined ? {} : { error: { message: result.error.message, retryable: result.error.retryable } }), }; }, isWriteEffect: async (tool: string) => (await deps.toolbox.get(tool))?.effect === 'write', registerIntent: async (input: { taskId: TaskId; tool: string; input: Record; label: string }) => { // 写类动作登记为动作意向:入参冻结,放行后由系统执行落地 await deps.release.registerIntent(input.taskId, { id: ids.next(), tool: input.tool, input: input.input, idempotencyKey: `${input.taskId}:${input.tool}:${ids.next()}`, label: input.label, }); }, }; /* ── skill-engine ─────────────────────────────────────────────── */ const skills = createSkillEngineService({ catalog: { skillsOf: async (activityId) => (await deps.registry.manifestOf(activityId)).assets.skills, allowlistOf: (activityId) => deps.toolbox.allowlistOf(activityId), }, tools: { invoke: (input) => toolGateway.invoke({ ...input, ...(input.stepNo === undefined ? {} : { stepNo: input.stepNo }) }), }, memory: memoryRecorder as never, clock, }); void skills; /* ── flow-engine ──────────────────────────────────────────────── */ const flow = createFlowRunner({ state: createFlowStateStore(db), evaluate: sfeel.condition, data: taskData, tools: { ...toolGateway, invoke: (i) => toolGateway.invoke(i) }, llm: { async generate(input) { // LLM 在 flow 里只以「被调用的工具」出现,不是调度者 try { const result = await deps.modelGateway.generate({ model: input.model, prompt: input.prompt, outputSchema: input.outputSchema, taskId: input.taskId, stepNo: input.stepNo, label: 'flow-step', }); return { ok: result.object !== undefined, ...(result.object === undefined ? { error: '出参不合 schema' } : { object: result.object }) }; } catch (error) { return { ok: false, error: error instanceof Error ? error.message : String(error) }; } }, }, cards, listeners, memory: memoryRecorder as never, tasks: taskContext, clock, }); /* ── loop-engine ──────────────────────────────────────────────── */ const loop = createTurnRunner({ state: createLoopStateStore(db), sfeel, data: taskData, agent: createAgentRunner({ gateway: deps.modelGateway, modelOf: async (taskId) => { const { activityId } = await taskContext.of(taskId); const manifest = await deps.registry.manifestOf(activityId); return manifest.declarations.models[0] ?? 'default'; }, }), tools: toolGateway, cards, listeners, memory: { append: memoryRecorder.append as never, loadActivityMemory: async (activityId: string, limit: number) => (await deps.memory.loadActivityMemory(activityId, { limit })).map((m) => ({ layer: m.layer, statement: m.statement })), }, tasks: taskContext, clock, }); /* ── task(L6):范式在此汇流,对上层不可区分 ──────────────────── */ const task = createTaskService({ store: taskStore, activities: { async load(activityId) { const manifest = await deps.registry.manifestOf(activityId); const envelope = await deps.registry.loadEnvelope(activityId); return { paradigm: manifest.identity.paradigm, caps: manifest.contract.resources.budget, envelope: { ...envelope }, inputSchema: manifest.contract.input.data, businessModel: manifest.assets.businessModel, stallAfter: manifest.declarations.harness.stallAfter, }; }, }, schemas: deps.schemas, // 范式在此汇流:两个引擎按统一的 ParadigmEngine 契约暴露给 task —— 对上层不可区分(§9) engines: { loop: { load: (taskId, model) => loop.load(taskId, model as never).then(() => undefined), advance: async (taskId) => { const outcome = await loop.runTurn(taskId); return { status: outcome.status, ...(outcome.blockedReason === undefined ? {} : { reason: outcome.blockedReason }) }; }, onEvent: async (taskId, event) => { const outcome = await loop.onEvent(taskId, event); return { status: outcome.status, ...(outcome.blockedReason === undefined ? {} : { reason: outcome.blockedReason }) }; }, }, flow: { load: (taskId, model) => flow.load(taskId, model as never).then(() => undefined), advance: async (taskId) => { const outcome = await flow.advance(taskId); return { status: outcome.status, ...(outcome.blockedReason === undefined ? {} : { reason: outcome.blockedReason }) }; }, onEvent: async (taskId, event) => { const outcome = await flow.resume(taskId, { kind: event.kind === 'control' ? 'time' : event.kind, ref: event.ref, payload: event.payload, }); return { status: outcome.status, ...(outcome.blockedReason === undefined ? {} : { reason: outcome.blockedReason }) }; }, }, }, memory: { append: memoryRecorder.append as never, seal: (taskId) => deps.memory.seal(taskId), distill: (input) => deps.memory.distill(input), hasHistory: async (taskId) => (await deps.memory.history(taskId)).length > 0, }, listeners: { hasArmedListener: async (taskId) => (await deps.event.listListeners(taskId)).some((l) => l.state === 'armed'), }, persistence: { hasDrafts: async (taskId) => (await deps.ledger.checkConflicts(taskId)) !== undefined, hasRuntimeState: async (taskId) => (await flow.position(taskId)) !== undefined || (await loop.runtime(taskId)) !== undefined, }, events: { publishControl: async (taskId, semantic, by, reason) => { await deps.event.publish({ type: 'control', correlation: { target: { kind: 'task', id: taskId }, cause: { kind: 'governance', ref: by }, keys: {} }, payload: { data: { semantic, issuedBy: by, reason } }, occurredAt: clock.now().toISOString(), source: { kind: 'governance' }, }); }, publishTerminal: async (taskId, terminal, reason) => { await deps.event.publish({ type: 'data', correlation: { target: { kind: 'task', id: taskId }, cause: { kind: 'engine', ref: 'terminal' }, keys: { terminal } }, payload: { data: { terminal, reason } }, occurredAt: clock.now().toISOString(), source: { kind: 'engine', detail: 'task' }, }); }, }, notifier: { notify: async (input) => { // 改派留痕并**通知双方** for (const to of [input.from, input.to]) { await deps.message.notify({ staffId: to, title: `任务改派:${input.title}`, body: `任务 ${input.taskId} 的负责人由 ${input.from} 改为 ${input.to}(操作人 ${input.by})`, }); } }, }, interaction: { setPhase: async (taskId, phase) => void (await deps.interaction.setPhase(taskId, phase)) }, clock, ids, }); /* ── process(L7) ─────────────────────────────────────────────── */ const process = createProcessService({ store: createProcessStore(db), activities: { async faces() { const cards = await deps.registry.list(); const map = new Map> extends never ? never : never>(); void map; const faces = new Map(); for (const card of cards) { const manifest = await deps.registry.manifestOf(card.activityId); const input = manifest.contract.input.data as { required?: string[]; properties?: Record }; const outputs: Record> = {}; for (const port of manifest.contract.output.mandatory) { const payload = port.payload as { properties?: Record }; outputs[port.name] = Object.fromEntries( Object.entries(payload.properties ?? {}).map(([k, v]) => [k, v.type ?? 'string']), ); } faces.set(card.activityId, { activityId: card.activityId, requiredInputs: input.required ?? [], inputTypes: Object.fromEntries( Object.entries(input.properties ?? {}).map(([k, v]) => [k, v.type ?? 'string']), ), outputs, }); } return faces; }, }, org: { positions: async () => new Set((await deps.org.listStaff()).positions.map((p) => p.code)), candidates: async (position) => { const staff = await deps.org.staffOfPosition(position); const withLoad = await Promise.all( staff.map(async (s) => ({ staffId: s.id, name: s.name, openTasks: (await task.openTasksOf(s.id)).length, })), ); return withLoad; }, }, tasks: { start: async (input) => task.start({ activityId: input.activityId, ownerId: input.ownerId, title: input.title, workflowId: input.workflowId, nodeId: input.nodeId, data: input.data, ...(input.message === undefined ? {} : { message: input.message }), }), }, approvals: { async raise(input) { // 审批点:卡到审批人工位 —— 与活动级的放行不同层 const issued = await deps.humanTool.issueExecution( { taskId: input.workflowId as never, ownerId: input.approverId, humanCallCap: 99 }, { title: `审批:${input.subject}`, basis: input.basis.length > 0 ? [...input.basis] : [{ k: '节点', v: input.nodeId }], ask: '请裁决:通过 / 驳回(驳回必附意见)', fields: [ { key: 'verdict', label: '裁决', kind: 'radio', required: true, options: [ { value: 'pass', label: '通过' }, { value: 'reject', label: '驳回', negative: true }, ], }, { key: 'comment', label: '依据 / 意见', kind: 'textarea' }, ], }, ); return { id: issued.order.id }; }, }, sfeel, events: { publish: async (input) => { await deps.event.publish({ type: 'data', correlation: { target: { kind: 'workflow', id: input.workflowId as never }, cause: { kind: 'engine', ref: input.kind }, keys: {}, }, payload: { data: input.detail }, occurredAt: clock.now().toISOString(), source: { kind: 'engine', detail: 'process' }, }); }, }, receipts: { notify: async (input) => { await deps.message.notify({ staffId: input.to, title: `工作流回执:${input.subject}`, body: `工作流 ${input.workflowId} 状态:${input.state}`, }); }, }, clock, ids, }); /* ── trigger(L7) ────────────────────────────────────────────── */ const trigger = createTriggerService({ store: createTriggerStore(db), process: { start: async (input) => process.start(input), requiredInputs: async (packageId) => { const pkg = await process.getPackage(packageId); const input = pkg?.callInterface.input as { required?: string[] } | undefined; return input?.required ?? []; }, }, sfeel, deadLetters: { // 起单失败入死信 —— 落日志 + 送治理台的消息要素,不静默丢弃 report: async (input) => { writeLog(`[trigger] 起单失败入死信:${input.triggerId} —— ${input.reason}`); const { staff } = await deps.org.listStaff(); for (const person of staff) { const governs = await deps.org.authorize(person.id, { kind: 'governance', function: 'trigger' }, 'govern'); if (!governs) continue; await deps.message.notify({ staffId: person.id, title: `触发器起单失败:${input.triggerId}`, body: input.reason, }); } }, }, ledger: { scan: async (input) => { const page = await deps.ledger.query({ address: { domain: input.domain, object: input.object }, where: {}, orderBy: [], limit: 200, offset: 0, }); return page.rows.flatMap((row) => { const date = row[input.dateField]; if (typeof date !== 'string') return []; return [{ key: { id: String(row['id'] ?? '') }, date, row }]; }); }, }, clock, ids, }); return { task, process, trigger }; }