判据 ④⑤ 解锁:G2 并行网关起出 A3 / A2 —— 输出口发射的是落地事实

G2 之后一直不分流,根因三条,逐条修:

① 输出口载荷是空的。`outputPortsOf` 原先从**作用域**里捞字段拼载荷,
   而放行那一刻草稿层已合并并清空,捞到的只剩起单入参 —— 下游节点必然缺字段。
   改为:载荷取自**落地收敛的事实**,由 `releaseWrite` 回报合并了哪些行,
   经 release → task → app 一路带上来。这也正是「落地收敛,输出口才发射」的字面意思:
   发射的就该是已经落地的那些事实。
   `LandedFacts` 给两种读法:铺平(标量取值)与逐行(数组取值),
   输出口声明里 `type: 'array'` 的属性填逐行,标量属性填铺平值(驼峰/下划线两种写法都认)。

② 一条分支起不来,把整条终态上报连坐掀翻 —— 于是**上游任务的终态也报不上去**,
   流停在原地,一行日志都没有。改为:该分支记 `node-start-failed` 事件 + 落一个
   blocked 实例,其余分支照走,在治理台「需处置」里现形。
   (这条是先修的:正因为它,前面两条才一直查不出来。)

③ 边映射引用了不存在的路径:`G2→A3` 的 pn 取 `intake.records`(那是想象中的 LLM 出参形状)、
   `G2J→A4` 的 records 取 `initiation.records`(起单时根本还没有这些行)。
   按输出口的真实声明改正。

结果:A1 released → G2 并行分叉 → **A3 渠道寻源落到林佩萱、A2 替代选型落到王梓名**,
两个 loop 活动同时在跑,岗位解析正确。

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:
2026-07-27 04:52:53 +08:00
co-authored by Claude Opus 5
parent 1721bcf6e3
commit 9de33a5416
22 changed files with 223 additions and 35 deletions
@@ -170,7 +170,7 @@ describe('ledger · 真实 PG', () => {
expect(official.rows[0]).toMatchObject({ grade: 'C' });
// 草稿已清空:再放行是空操作
expect(await ledger.releaseWrite(T1)).toEqual({ mergedRows: 0, emittedEvents: 0 });
expect(await ledger.releaseWrite(T1)).toEqual({ mergedRows: 0, emittedEvents: 0, merged: [] });
}, 60_000);
itDb()('数据冲突检查:他单已改同一字段 → 放行写打回,不做部分合并', async () => {
+11
View File
@@ -69,6 +69,17 @@ export type DraftWrite = {
export type ReleaseWriteResult = {
readonly mergedRows: number;
readonly emittedEvents: number;
/**
* 合并进正式版的那些行(键 + 值)。
* 「落地收敛,输出口才发射」——发射的载荷就该是**这些已落地的事实**,
* 而不是从作用域里再猜一遍:放行后草稿层已清空,再猜就什么也猜不到。
*/
readonly merged: ReadonlyArray<{
readonly domain: string;
readonly object: string;
readonly key: Record<string, unknown>;
readonly values: Record<string, unknown>;
}>;
};
/**
@@ -154,7 +154,7 @@ export function createLedgerService(deps: LedgerServiceDeps): LedgerService {
if (conflicts.length > 0) throw new DraftConflictError(conflicts);
const drafts = await deps.drafts.list(taskId);
if (drafts.length === 0) return { mergedRows: 0, emittedEvents: 0 };
if (drafts.length === 0) return { mergedRows: 0, emittedEvents: 0, merged: [] };
const pending: EnvelopeDraft[] = [];
@@ -185,7 +185,13 @@ export function createLedgerService(deps: LedgerServiceDeps): LedgerService {
});
for (const envelope of pending) await deps.events.publish(envelope);
return { mergedRows: drafts.length, emittedEvents: pending.length };
return {
mergedRows: drafts.length,
emittedEvents: pending.length,
merged: drafts
.filter((d) => !d.tombstone)
.map((d) => ({ domain: d.domain, object: d.object, key: d.key, values: d.patch })),
};
},
async query(query: LedgerQuery): Promise<LedgerPage> {
@@ -177,7 +177,30 @@ export function createProcessService(deps: ProcessServiceDeps): ProcessService {
}
const assembled = assemble(edge, { ...scope, ports: { ...scope.ports, [fromNodeId]: payload } }, deps.sfeel.expression);
await enter(record, pkg, target, assembled);
try {
await enter(record, pkg, target, assembled);
} 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',
});
}
}
}
+7
View File
@@ -70,6 +70,13 @@ export type LandOutcome = {
/** 落地收敛后才发射的输出口 */
readonly emittedPorts: readonly string[];
readonly ledgerRowsMerged: number;
/** 已落地的事实 —— 输出口发射的载荷取自这里 */
readonly merged: ReadonlyArray<{
readonly domain: string;
readonly object: string;
readonly key: Record<string, unknown>;
readonly values: Record<string, unknown>;
}>;
/** 落地受阻时发出的落地失败卡 */
readonly failureWorkOrderId?: string;
};
@@ -189,11 +189,17 @@ export function createReleaseService(deps: ReleaseServiceDeps): ReleaseService {
verdict,
emittedPorts: [],
ledgerRowsMerged: merged.mergedRows,
merged: merged.merged,
failureWorkOrderId: card.id,
};
}
// 落地收敛,输出口才发射 —— 下游只见落地完成的任务
// 落地收敛,输出口才发射 —— 下游只见落地完成的任务。
// 载荷取自**已落地的事实**:放行后草稿层已清空,再回作用域里捞就什么也捞不到。
const landedFacts = merged.merged.reduce<Record<string, unknown>>(
(acc, row) => ({ ...acc, ...row.key, ...row.values }),
{},
);
const ports = await deps.tasks.outputPortsOf(input.taskId);
const emitted = emissionPortNames(
{
@@ -206,7 +212,7 @@ export function createReleaseService(deps: ReleaseServiceDeps): ReleaseService {
input.selectedPorts,
);
for (const port of emitted) {
await deps.ports.emit({ taskId: input.taskId, port, payload: {} });
await deps.ports.emit({ taskId: input.taskId, port, payload: landedFacts });
}
await deps.memory.append({
taskId: input.taskId,
@@ -214,7 +220,7 @@ export function createReleaseService(deps: ReleaseServiceDeps): ReleaseService {
title: `落地收敛,发射输出口:${emitted.join('、') || '(无)'}`,
});
return { verdict, emittedPorts: emitted, ledgerRowsMerged: merged.mergedRows };
return { verdict, emittedPorts: emitted, ledgerRowsMerged: merged.mergedRows, merged: merged.merged };
},
};
}
+11 -1
View File
@@ -16,7 +16,17 @@ export type IntentStore = {
/** 台账 —— app 注入 `ledger`(L1);放行写与冲突检查都在它那 */
export type LedgerGateway = {
checkConflicts(taskId: TaskId): Promise<Array<{ field: string; key: Record<string, string | number> }>>;
releaseWrite(taskId: TaskId): Promise<{ mergedRows: number; emittedEvents: number }>;
releaseWrite(taskId: TaskId): Promise<{
mergedRows: number;
emittedEvents: number;
/** 合并进正式版的行 —— 输出口发射的载荷取自这里 */
merged: ReadonlyArray<{
domain: string;
object: string;
key: Record<string, unknown>;
values: Record<string, unknown>;
}>;
}>;
};
/** 工具 —— app 注入 `toolbox`(L2);预检的②③④项与落地执行都经它 */
+8 -1
View File
@@ -41,6 +41,12 @@ export class TaskAlreadyTerminalError extends Error {
}
/** 一单任务 —— 活动的运行时实例;**恰有一名负责人** */
/** 落地事实的两种读法:铺平(标量取值)与逐行(数组取值)—— 输出口载荷由此装配 */
export type LandedFacts = {
readonly flat: Record<string, unknown>;
readonly rows: ReadonlyArray<{ key: Record<string, unknown>; values: Record<string, unknown> }>;
};
export type Task = {
readonly id: TaskId;
readonly activityId: string;
@@ -100,7 +106,8 @@ export type TaskService = {
reassign(taskId: TaskId, to: StaffId, by: StaffId): Promise<Task>;
/** 终态裁定 —— 完成由框架裁定,终态由系统命名 */
finish(taskId: TaskId, cause: TerminalCause): Promise<Task>;
/** `facts` = 落地收敛的事实(仅放行路径有);作为输出口载荷上报流程层 */
finish(taskId: TaskId, cause: TerminalCause, facts?: LandedFacts): Promise<Task>;
/** 断点续跑的三份持久状态是否齐备 */
resumeReadiness(taskId: TaskId): Promise<ResumeReadiness>;
};
+1
View File
@@ -3,6 +3,7 @@
*/
export type {
BudgetBreach,
LandedFacts,
BudgetCaps,
Consumption,
ControlOutcome,
@@ -179,7 +179,7 @@ export function createTaskService(deps: TaskServiceDeps): TaskService {
kind: 'engine-transition',
title: `落地收敛:合并 ${landed.ledgerRowsMerged} 行,发射输出口 ${landed.emittedPorts.join('、') || '—'}`,
});
const finished = await this.finish(taskId, { kind: 'released' });
const finished = await this.finish(taskId, { kind: 'released' }, landed.facts);
return { task: finished, verdict, progressed: true };
}
@@ -247,7 +247,7 @@ export function createTaskService(deps: TaskServiceDeps): TaskService {
},
/** 终态由系统命名;封卷触发蒸馏钩子 */
async finish(taskId, cause) {
async finish(taskId, cause, facts) {
const task = await require(taskId);
if (task.terminal !== undefined) return task;
@@ -261,7 +261,7 @@ export function createTaskService(deps: TaskServiceDeps): TaskService {
});
await deps.memory.append({ taskId, kind: 'terminal', title: `终态:${terminal}`, detail: { reason } });
await deps.events.publishTerminal(taskId, terminal, reason);
await deps.events.publishTerminal(taskId, terminal, reason, facts ?? { flat: {}, rows: [] });
await deps.interaction.setPhase(taskId, 'terminal');
// 终态封卷 → 归档 → 蒸馏钩子产出活动记忆候选条目
+5 -2
View File
@@ -3,7 +3,7 @@
*/
import type { BudgetCaps, Paradigm, StaffId, TaskId, TerminalState } from '@contracts';
import type { Consumption } from './internal/domain/harness.ts';
import type { Task } from './contract.ts';
import type { LandedFacts, Task } from './contract.ts';
export type Clock = { now(): Date };
export type IdGenerator = { next(): string };
@@ -61,6 +61,8 @@ export type LandingGateway = {
converged: boolean;
emittedPorts: readonly string[];
ledgerRowsMerged: number;
/** 已落地的事实 —— 输出口载荷,随终态报给流程层 */
facts: LandedFacts;
}>;
};
@@ -92,7 +94,8 @@ export type PersistenceProbe = {
/** 控制事件与终态事件 —— app 注入 `event`(L1) */
export type TaskEventPublisher = {
publishControl(taskId: TaskId, semantic: string, by: StaffId, reason?: string): Promise<void>;
publishTerminal(taskId: TaskId, terminal: TerminalState, reason: string): Promise<void>;
/** `facts` = 落地收敛的事实,作为输出口载荷随终态报给流程层;非放行终态为空 */
publishTerminal(taskId: TaskId, terminal: TerminalState, reason: string, facts: LandedFacts): Promise<void>;
};
/** 改派通知 —— app 注入 `message`(L2):改派留痕并**通知双方** */