ChaiHao/BPOS-MVP
Archived
1
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

M5/M6 修复串:判据 ③ 全链路跑通 —— 六处「声明与引擎对不上」

Browse Source 
判据 ③ 现在从入站到 released 一路走通:步序推进 → LLM 步(密钥是占位符,
按声明降级为人兜底)→ 台账草稿层 → D1 条件跳步 → S4 工具 → S5 DMN 三分类
(LM358DR → Type A · 直接快答)→ 验收放行 → 草稿合并正式版 2 行 → 输出口
intake 发射 → 终态 released。

沿路修掉的六处,每一处都是「声明是这么写的,引擎不这么读」:

1. 人兜底卡只问一个笼统「结果」框 → 改问**这一步声明的出参字段**。
   出参契约属于步、不属于执行者:同一步谁做完,落进作用域的形状都得一样。
2. 人事件的载荷是回传协议壳 `{mode,workOrderId,values}`,
   引擎把整个壳当出参写 → 取 `values`。
3. 引擎作用域是 `{'域.对象': 行数组}`,而所有活动的步声明写的都是裸字段名 →
   任务视图三层铺开:起单入参在底、各对象最新行的字段铺顶层、地址式整表保留。
4. 起单入参从不落库、也进不了作用域 → task 存 `input_data`,断点续跑后仍在。
5. 步出参无法声明**写哪一行** → 出参加 `key`(主键列 → 作用域路径);
   六个活动包的每处出参补上键。台账几乎都有主键,不声明键就只能写无主键行。
6. `tick` 不处理 `finished`:谱走完仍按 running 一拍拍推,把预算烧到 exhausted,
   把「没人裁」误报成「跑爆了」→ 放行通过则落地收敛后 released;未放行则挂起等人。
   `flow.advance` 据末步验收裁决报 released —— **完成只出自放行通过**。

另两处装配缺口:
  · `release.land()` 实现完整却无人调用 —— 放行后从不落地。经 LandingGateway 接上。
  · `process.onTaskTerminal()` 实现完整却无人调用 —— 任务终态报不到流程层,
    流就停在那个节点上,既不流转也不走失败走向。task 在 L6 不得反向依赖 L7,
    这一跳由 app 缝合(延后绑定,同 container 里 TriggerMatcher 的手法)。

还有一处版本化缺陷:资产播种恒用 v1,`shouldSeedAsset` 一比就跳过 ——
只有**第一个**包版本的资产此生有效,之后装什么都跑旧的。改为播下一版;
进化管线放行出的更高版仍不被盖过;同版重复安装不再播种。

gates: tsc / eslint / depcruise(317 模块)/ api-report / 393 测试 全绿

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KeDBx2PHLavbURXB8LJJ7m
This commit is contained in:
ChaiHao
2026-07-27 04:31:23 +08:00
co-authored by Claude Opus 5
parent 3b9587563a
commit e72182003a
35 changed files with 3901 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/api/src/modules/activity-registry/__tests__/versioning.test.ts
+29 -1
View File
@@ -9,7 +9,7 @@ import {
mayTakeEffect,
shouldSeedAsset,
} from '../internal/domain/versioning.ts';
import type { AssetVersion, PackageVersion, ReplayCase } from '../internal/domain/versioning.ts';
import type { AssetKind, AssetVersion, PackageVersion, ReplayCase } from '../internal/domain/versioning.ts';
import { scanArtifact } from '../internal/domain/static-audit.ts';
const pkg: PackageVersion = {
@@ -150,3 +150,31 @@ describe('scanArtifact —— 活动包零环境能力(ADR-004)', () => {
expect(scanArtifact('import fs from "node:fs";').capabilities).toContain('node:fs');
});
});
describe('包新版携带的资产:播下一版,不恒播 v1', () => {
const asset = (kind: AssetKind, version: number): AssetVersion => ({
activityId: 'a1',
kind,
version,
effective: version === 1,
source: 'package-seed',
createdAt: '2026-07-27T00:00:00.000Z',
});
it('库里已有 v1 时,新包的资产应播 v2 —— 否则只有首版资产此生有效', () => {
const existing = [asset('business-model', 1)];
expect(shouldSeedAsset(existing, 'business-model', 2)).toBe(true);
// 恒播 v1 就会被这条挡掉,新包的业务模型永远进不来
expect(shouldSeedAsset(existing, 'business-model', 1)).toBe(false);
});
it('进化管线放行出的更高版不被播种盖过', () => {
const existing = [asset('business-model', 1), asset('business-model', 5)];
expect(shouldSeedAsset(existing, 'business-model', 2)).toBe(false);
expect(shouldSeedAsset(existing, 'business-model', 6)).toBe(true);
});
it('库里没有该类资产时从 v1 起播', () => {
expect(shouldSeedAsset([], 'dmn', 1)).toBe(true);
});
});
apps/api/src/modules/activity-registry/internal/adapter/registry-store.ts
+5 -2
View File
@@ -12,7 +12,7 @@ import { activityAssetTable, activityPackageTable, assetIncrementTable } from '.
export function createRegistryStore(db: Db): RegistryStore {
return {
async savePackage(manifest: ActivityManifest, artifact: string) {
await db
const inserted = await db
.insert(activityPackageTable)
.values({
activityId: manifest.identity.activityId,
@@ -25,7 +25,10 @@ export function createRegistryStore(db: Db): RegistryStore {
state: 'active',
enabled: true,
})
.onConflictDoNothing();
.onConflictDoNothing()
.returning({ version: activityPackageTable.version });
// 回报是否**真的新增了这一版**:重复安装同版不该再播一次资产
return inserted.length > 0;
},
async listPackages(activityId) {
apps/api/src/modules/activity-registry/internal/usecase/registry-service.ts
+25 -13
View File
@@ -89,22 +89,34 @@ export function createRegistryService(deps: RegistryServiceDeps): ActivityRegist
if (!verdict.accepted) throw new InstallRejectedError(verdict);
// 安装恒为新增、版本并存;在途任务按其启动时版本跑完
await deps.store.savePackage(manifest, artifact);
const isNewPackageVersion = await deps.store.savePackage(manifest, artifact);
// 资产初始版随包播种 —— **不覆盖已有更高的资产版**
/*
* 资产初始版随包播种。
*
* 播的是**下一个**资产版,不是恒定的 v1:包版本与资产版本是两条序列,
* 但新包携带的初始资产同样过了安装两审,理应可用 —— 若恒播 v1,
* 就只有第一个包版本的资产此生有效,之后再装什么都跑旧的。
* 「不覆盖已有更高的资产版」仍然成立:进化管线放行出来的版本号更高,播种不会盖过它。
* 同版重复安装不再播种;在途任务按启动时版本跑完,不受影响。
*/
const existing = await deps.store.listAssets(manifest.identity.activityId);
const now = deps.clock.now().toISOString();
for (const kind of SEED_KINDS) {
if (!shouldSeedAsset(existing, kind, 1)) continue;
const asset: AssetVersion = {
activityId: manifest.identity.activityId,
kind,
version: 1,
effective: true,
source: 'package-seed',
createdAt: now,
};
await deps.store.saveAsset(asset, serializeAsset(manifest, kind));
if (isNewPackageVersion) {
for (const kind of SEED_KINDS) {
const highest = existing.filter((a) => a.kind === kind).reduce((max, a) => Math.max(max, a.version), 0);
const seedVersion = highest + 1;
if (!shouldSeedAsset(existing, kind, seedVersion)) continue;
const asset: AssetVersion = {
activityId: manifest.identity.activityId,
kind,
version: seedVersion,
effective: true,
source: 'package-seed',
createdAt: now,
};
await deps.store.saveAsset(asset, serializeAsset(manifest, kind));
}
}
return verdict;
apps/api/src/modules/activity-registry/port.ts
+2 -1
View File
@@ -8,7 +8,8 @@ export type Clock = { now(): Date };
export type IdGenerator = { next(): string };
export type RegistryStore = {
savePackage(manifest: ActivityManifest, artifact: string): Promise<void>;
/** 返回是否新增了这一包版本(同版重复安装返回 false) */
savePackage(manifest: ActivityManifest, artifact: string): Promise<boolean>;
listPackages(activityId?: string): Promise<PackageVersion[]>;
manifest(activityId: string, packageVersion?: string): Promise<ActivityManifest | undefined>;
artifact(activityId: string, packageVersion: string): Promise<string | undefined>;
apps/api/src/modules/flow-engine/__tests__/step-sequence.test.ts
+105
View File
@@ -0,0 +1,105 @@
/** flow 步序:声明校验 / 失败走向 / 兜底出参 —— domain 测试零 mock(AM 12.3) */
import { describe, expect, it } from 'vitest';
import { flowStepSchema } from '@contracts';
import type { FlowStep } from '@contracts';
import {
checkDeclaration,
fallbackFields,
failureAction,
initialPosition,
lastStepIsAcceptance,
markStep,
parallelGroupComplete,
} from '../internal/domain/step-sequence.ts';
const llmStep = (over: Record<string, unknown> = {}): FlowStep =>
flowStepSchema.parse({
kind: 'llm',
name: 'S3',
label: 'S3 逐条解析归一',
input: { rfqRef: 'rfqRef' },
output: { domain: 'quote_intake', object: 'record', fields: { pn: 'pn', brand: 'brand', eau: 'eau' } },
model: 'quote-extractor',
template: '解析 {{rfqRef}}',
outputSchema: { type: 'object' },
failure: { retry: { times: 2, backoff: 'PT30S' }, fallback: { kind: 'human', workOrder: 'execution' } },
...over,
});
describe('人兜底:出参契约属于步,不属于执行者', () => {
it('兜底卡问的字段 = 这一步声明的出参 —— 人填完能落进作用域', () => {
// 若只给一个笼统的「结果」框,下游按 output.fields 取值会取空,在下一步崩
expect(fallbackFields(llmStep())).toEqual(['pn', 'brand', 'eau']);
});
it('无出参声明的步不凭空造字段', () => {
const step = flowStepSchema.parse({
kind: 'llm',
name: 'S9',
label: '只说不写的一步',
input: {},
model: 'm',
template: 't',
outputSchema: { type: 'object' },
failure: { retry: { times: 0, backoff: 'PT1S' }, fallback: { kind: 'blocked' } },
});
expect(fallbackFields(step)).toEqual([]);
});
});
describe('失败走向声明给定,不临场发明', () => {
it('重试次数未用尽 → retry', () => {
expect(failureAction(llmStep(), 1).kind).toBe('retry');
});
it('重试用尽 → 按声明的 fallback 转人', () => {
const action = failureAction(llmStep(), 2);
expect(action.kind).toBe('human');
});
it('声明 blocked 的步不转人 —— 走向是声明的,不是猜的', () => {
const step = llmStep({ failure: { retry: { times: 0, backoff: 'PT1S' }, fallback: { kind: 'blocked' } } });
expect(failureAction(step, 0).kind).toBe('blocked');
});
});
describe('步序位置', () => {
it('初始位置每步待跑', () => {
const position = initialPosition([llmStep(), llmStep({ name: 'S4' })]);
expect(position.index).toBe(0);
expect(position.steps.map((s) => s.state)).toEqual(['pending', 'pending']);
});
it('markStep 只改那一步,其余原样', () => {
const position = initialPosition([llmStep(), llmStep({ name: 'S4' })]);
const next = markStep(position, 1, { state: 'done' });
expect(next.steps.map((s) => s.state)).toEqual(['pending', 'done']);
});
it('并行组要全完成才算完成', () => {
const position = markStep(initialPosition([llmStep(), llmStep({ name: 'S4' })]), 0, { state: 'done' });
expect(parallelGroupComplete(position, ['S3', 'S4'])).toBe(false);
expect(parallelGroupComplete(markStep(position, 1, { state: 'done' }), ['S3', 'S4'])).toBe(true);
});
});
describe('声明校验', () => {
it('步名重复即拒', () => {
expect(checkDeclaration([llmStep(), llmStep()]).length).toBeGreaterThan(0);
});
it('末步是验收工单可辨', () => {
const acceptance: FlowStep = flowStepSchema.parse({
kind: 'human',
name: 'S6',
label: '验收',
input: {},
workOrder: 'acceptance',
card: {},
deadline: 'P1D',
failure: { retry: { times: 0, backoff: 'PT1S' }, fallback: { kind: 'blocked' } },
});
expect(lastStepIsAcceptance([llmStep(), acceptance])).toBe(true);
expect(lastStepIsAcceptance([llmStep()])).toBe(false);
});
});
apps/api/src/modules/flow-engine/internal/domain/step-sequence.ts
+24
View File
@@ -17,6 +17,8 @@ export type StepRuntime = {
readonly attempt: number;
readonly output?: Record<string, unknown>;
readonly error?: string;
/** 验收工单的放行裁决:pass / 驳回类别 —— 末步据此定终局 */
readonly verdict?: string;
};
export type FlowPosition = {
@@ -121,3 +123,25 @@ export function parallelGroupComplete(position: FlowPosition, members: readonly
return runtime?.state === 'done' || runtime?.state === 'skipped';
});
}
/**
* 人兜底时要回填的字段 —— 取自这一步**声明的出参**。
*
* 出参契约属于**步**,不属于执行者:同一步无论由 LLM、工具还是人做完,
* 落进作用域的形状必须一致。若兜底卡只给一个笼统的「结果」文本框,
* 人填了也落不进去 —— 下游按 `output.fields` 取值,取不到就在下一步崩。
*
* 返回键名数组,与 human 步的 `card.fields` 同形:卡的渲染只有一套。
*/
export function fallbackFields(step: FlowStep): readonly string[] {
return step.output === undefined ? [] : Object.values(step.output.fields);
}
/**
* 末步的放行裁决是否「通过」—— **完成只出自放行通过**(OS 05)。
* 驳回(重做 / 迭代)不是完成:谱虽走完,任务不得进 released。
*/
export function lastVerdictPassed(position: FlowPosition): boolean {
const last = position.steps[position.steps.length - 1];
return last?.verdict === 'pass';
}
apps/api/src/modules/flow-engine/internal/usecase/flow-runner.ts
+35 -6
View File
@@ -19,6 +19,9 @@ import type {
import type { FlowPosition } from '../domain/step-sequence.ts';
import {
checkDeclaration,
fallbackFields,
lastStepIsAcceptance,
lastVerdictPassed,
failureAction,
initialPosition,
markStep,
@@ -52,15 +55,29 @@ export function createFlowRunner(deps: FlowRunnerDeps): FlowEngineService {
for (const [field, path] of Object.entries(step.output.fields)) {
patch[field] = output[path] ?? output[field];
}
// 写哪一行:主键取自出参,取不到再回作用域(起单入参往往就带着主键)
const scope = await deps.data.view(taskId);
const key: Record<string, string | number> = {};
for (const [column, path] of Object.entries(step.output.key)) {
const value = output[path] ?? scope[path];
// 主键值只能是标量;取不到就不填,让台账的非空约束在合并时如实报出来
if (typeof value === 'string' || typeof value === 'number') key[column] = value;
}
await deps.data.draftWrite({
taskId,
domain: step.output.domain,
object: step.output.object,
key: {},
key,
patch,
});
}
/** 回传协议 → 出参:取 `values`;非回传形状的人事件按原样当出参 */
function replyValuesOf(payload: Record<string, unknown>): Record<string, unknown> {
const values = payload['values'];
return typeof values === 'object' && values !== null ? (values as Record<string, unknown>) : payload;
}
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)) {
@@ -188,7 +205,13 @@ export function createFlowRunner(deps: FlowRunnerDeps): FlowEngineService {
ownerId: context.ownerId,
humanCallCap: context.humanCallCap,
workOrder: action.workOrder,
card: { title: `步「${step.label}」重试仍败,转人兜底`, reason: error },
card: {
title: `步「${step.label}」重试仍败,转人兜底`,
reason: error,
// 兜底的人要交的东西,和这一步**本来该产出的东西**同形:
// 出参契约是步的契约,不因换谁来做而变(统一执行契约)。
fields: fallbackFields(step),
},
});
await deps.listeners.register({
taskId,
@@ -231,8 +254,10 @@ export function createFlowRunner(deps: FlowRunnerDeps): FlowEngineService {
switch (action.kind) {
case 'finished': {
// 末步是验收工单时,终态由其裁决给出;此处只报「谱走完」
return { position, status: 'finished', released: false };
// 末步是验收工单时,终态由其**裁决**给出 —— 完成只出自放行通过。
// 裁决记在末步的 runtime 上(resume 时落笔),谱走完在这里读出来。
const released = lastStepIsAcceptance(model.steps) && lastVerdictPassed(position);
return { position, status: 'finished', released };
}
case 'skip': {
await deps.memory.append({
@@ -268,10 +293,14 @@ export function createFlowRunner(deps: FlowRunnerDeps): FlowEngineService {
title: `事件到达,步 ${index + 1} 完成条件满足(${event.kind})`,
detail: { ref: event.ref },
});
// 放行裁决落笔在这一步的运行时上:谱走完时据此定终局
const verdict = event.kind === 'human' ? String(event.payload['verdict'] ?? '') : '';
if (step !== undefined && event.kind === 'human') {
await writeOutput(taskId, step, event.payload);
// 人事件的载荷是**回传协议**的形状(`{mode, workOrderId, values}`),
// 出参在 `values` 里;直接把整个载荷当出参写,落进去的会是协议壳而不是答案。
await writeOutput(taskId, step, replyValuesOf(event.payload));
}
const next = markStep(position, index, { state: 'done' });
const next = markStep(position, index, verdict === '' ? { state: 'done' } : { state: 'done', verdict });
await deps.state.save(taskId, model, next);
return { position: next, status: 'running' };
},
apps/api/src/modules/task/contract.ts
+2
View File
@@ -55,6 +55,8 @@ export type Task = {
readonly reason?: string;
/** 装载信封:包版本 + 各资产生效版 —— 行为可归因 */
readonly envelope: Readonly<Record<string, unknown>>;
/** 起单入参:契约的 data —— 引擎求值的作用域底料 */
readonly data: Readonly<Record<string, unknown>>;
readonly caps: BudgetCaps;
readonly used: Consumption;
readonly startedAt: string;
apps/api/src/modules/task/internal/adapter/task-schema.ts
+2
View File
@@ -21,6 +21,8 @@ export const taskTable = sys.table(
reason: text('reason'),
/** 装载信封:包版本 + 各资产生效版 —— 行为可归因 */
envelope: jsonb('envelope').notNull(),
/** 起单入参 —— 步序声明按字段名读它;不存下来,断点续跑后作用域就空了 */
inputData: jsonb('input_data').notNull().default({}),
caps: jsonb('caps').notNull(),
usedSteps: integer('used_steps').notNull().default(0),
usedTokens: integer('used_tokens').notNull().default(0),
apps/api/src/modules/task/internal/adapter/task-store.ts
+2
View File
@@ -24,6 +24,7 @@ function toTask(row: Row): Task {
...(row.terminal === null ? {} : { terminal: row.terminal as TerminalState }),
...(row.reason === null ? {} : { reason: row.reason }),
envelope: row.envelope as Record<string, unknown>,
data: row.inputData as Record<string, unknown>,
caps: row.caps as BudgetCaps,
used: {
steps: row.usedSteps,
@@ -49,6 +50,7 @@ export function createTaskStore(db: Db): TaskStore {
title: task.title,
runtime: task.runtime,
envelope: task.envelope,
inputData: task.data,
caps: task.caps,
startedAt: new Date(task.startedAt),
});
apps/api/src/modules/task/internal/usecase/task-service.ts
+36
View File
@@ -12,6 +12,7 @@ import type {
Clock,
IdGenerator,
InteractionPhaseSetter,
LandingGateway,
ListenerProbe,
MemoryRecorder,
ParadigmEngine,
@@ -31,6 +32,8 @@ export type TaskServiceDeps = {
/** 范式引擎按 paradigm 分派 —— 对上层不可区分 */
readonly engines: Readonly<Record<'loop' | 'flow', ParadigmEngine>>;
readonly memory: MemoryRecorder;
/** 落地收敛 —— 放行通过的那一刻才碰事实 */
readonly landing: LandingGateway;
readonly listeners: ListenerProbe;
readonly persistence: PersistenceProbe;
readonly events: TaskEventPublisher;
@@ -80,6 +83,7 @@ export function createTaskService(deps: TaskServiceDeps): TaskService {
ownerId: input.ownerId,
title: input.title,
runtime: 'running',
data: input.data,
envelope: loaded.envelope,
caps: loaded.caps,
used: ZERO_CONSUMPTION,
@@ -147,6 +151,38 @@ export function createTaskService(deps: TaskServiceDeps): TaskService {
const finished = await this.finish(taskId, { kind: 'blocked', reason: outcome.reason ?? '无法推进' });
return { task: finished, verdict, progressed: true };
}
/*
* 谱走完了。**完成只出自放行通过**(OS 05):
* · 放行通过 → 落地收敛(草稿合并 + 动作执行 → 收敛才发射输出口)→ released;
* · 未放行(还没裁 / 驳回)→ 不是完成,也**不再空推** —— 挂起等人,
* 否则推进器会一拍一拍把预算烧到 exhausted,把「没人裁」误报成「跑爆了」。
*/
if (outcome.status === 'finished') {
if (outcome.released !== true) {
const waiting = await deps.store.update(taskId, { runtime: 'awaiting-human' });
await deps.interaction.setPhase(taskId, 'awaiting-event');
return { task: waiting, verdict, progressed: false };
}
const landed = await deps.landing.land({
taskId,
ownerId: task.ownerId,
humanCallCap: task.caps.humanCalls,
});
if (!landed.converged) {
// 落地未收敛:输出口不得发射,任务受阻待人(落地失败卡已发)
const blocked = await this.finish(taskId, { kind: 'blocked', reason: '落地未收敛,输出口未发射' });
return { task: blocked, verdict, progressed: true };
}
await deps.memory.append({
taskId,
kind: 'engine-transition',
title: `落地收敛:合并 ${landed.ledgerRowsMerged} 行,发射输出口 ${landed.emittedPorts.join('、') || '—'}`,
});
const finished = await this.finish(taskId, { kind: 'released' });
return { task: finished, verdict, progressed: true };
}
return { task: updated, verdict, progressed: true };
},
apps/api/src/modules/task/port.ts
+21 -2
View File
@@ -38,11 +38,30 @@ export type SchemaValidator = {
/** 范式引擎 —— loop / flow 在此汇流;对上层不可区分(§9 双层边界) */
export type ParadigmEngine = {
load(taskId: TaskId, model: unknown): Promise<void>;
advance(taskId: TaskId): Promise<{ status: 'running' | 'awaiting' | 'submitted' | 'finished' | 'blocked'; reason?: string }>;
advance(taskId: TaskId): Promise<EngineOutcome>;
onEvent(
taskId: TaskId,
event: { kind: 'data' | 'time' | 'human' | 'control'; ref: string; payload: Record<string, unknown> },
): Promise<{ status: 'running' | 'awaiting' | 'submitted' | 'finished' | 'blocked'; reason?: string }>;
): Promise<EngineOutcome>;
};
export type EngineOutcome = {
readonly status: 'running' | 'awaiting' | 'submitted' | 'finished' | 'blocked';
readonly reason?: string;
/** 谱走完时:放行是否通过 —— **完成只出自放行通过**,引擎不自行宣布完成 */
readonly released?: boolean;
};
/**
* 落地收敛 —— app 注入 `release`(L4)。
* 放行通过的那一刻才落地:草稿合并入正式版、动作意向执行、收敛后才发射输出口。
*/
export type LandingGateway = {
land(input: { taskId: TaskId; ownerId: StaffId; humanCallCap: number }): Promise<{
converged: boolean;
emittedPorts: readonly string[];
ledgerRowsMerged: number;
}>;
};
export type MemoryRecorder = {