Archived
xcsandClaude Fable 5
7b85a79323
human-tool / ledger / loop-engine / release / task / interaction 六模块与 app 层(desk 路由、engines、consumers、 data-order-landing、case-stamp、reset-runs、seed-activities、容器与 server 装配),web 的 desk-task (ViewRegion / chat-bubbles / review-face)与 app.css、ui-kit 图标;契约 card / manifest; 迁移 0034_task_seq 与迁移日志;api-report 快照;CLAUDE.md 快照、debt、活动开发规范、HTML 报告规范同步。 逐条口径见 CLAUDE.md 快照(2026-08-27 下午至晚)。另一会话做的,一直留在工作树没入库;与同批另两条提交互相依赖。 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_0167qWjeyBwvfdGovd7FWbCu
383 lines
16 KiB
TypeScript
383 lines
16 KiB
TypeScript
/**
|
|
* ledger 用例 —— 编排 domain,依赖 port.ts,不 import 任何 adapter(编码规范 §5.3)。
|
|
*
|
|
* 写三分的不变量在此落实(OS 09):
|
|
* systemWrite → 正式版 + **即时**数据事件
|
|
* draftWrite → 草稿层,**零**事件
|
|
* releaseWrite → 冲突检查 → 原子合并 → **此刻**抛事件
|
|
*/
|
|
import type { EnvelopeDraft, LedgerPage, LedgerQuery, ObjectDeclaration, TaskId } from '@contracts';
|
|
import { eventIdSchema } from '@contracts';
|
|
import type {
|
|
DraftWrite,
|
|
LedgerService,
|
|
ObjectAddress,
|
|
ReleaseScope,
|
|
ReleaseWriteResult,
|
|
SchemaApplyResult,
|
|
SystemWrite,
|
|
} from '../../contract.ts';
|
|
import { DraftConflictError, ObjectNotMaterializedError } from '../../contract.ts';
|
|
import type {
|
|
Clock,
|
|
DraftStore,
|
|
EventPublisher,
|
|
IdGenerator,
|
|
LedgerStore,
|
|
ObjectRef,
|
|
SimulationProbe,
|
|
} from '../../port.ts';
|
|
import { mergeDeclarations, planIsNonDestructive, planSchemaChanges } from '../domain/schema-merge.ts';
|
|
import { detectConflicts, keyOf, overlay, pickKey } from '../domain/task-view.ts';
|
|
|
|
export type LedgerServiceDeps = {
|
|
readonly store: LedgerStore;
|
|
readonly drafts: DraftStore;
|
|
readonly events: EventPublisher;
|
|
readonly clock: Clock;
|
|
readonly ids: IdGenerator;
|
|
/** 试跑判定 —— 放行写据此把主数据挡在正式表外(见 port.ts `SimulationProbe`) */
|
|
readonly simulation: SimulationProbe;
|
|
};
|
|
|
|
function groupDeclarations(declarations: readonly ObjectDeclaration[]): Map<string, ObjectDeclaration[]> {
|
|
const groups = new Map<string, ObjectDeclaration[]>();
|
|
for (const d of declarations) {
|
|
const key = `${d.domain}.${d.object}`;
|
|
const bucket = groups.get(key);
|
|
if (bucket === undefined) groups.set(key, [d]);
|
|
else bucket.push(d);
|
|
}
|
|
return groups;
|
|
}
|
|
|
|
/** 范围过滤:不给范围 = 全收 */
|
|
function inScope(only: readonly ObjectAddress[] | undefined): (draft: ObjectRef) => boolean {
|
|
if (only === undefined) return () => true;
|
|
const wanted = new Set(only.map((a) => `${a.domain}.${a.object}`));
|
|
return (draft) => wanted.has(`${draft.domain}.${draft.object}`);
|
|
}
|
|
|
|
/** 落地后的基线:只取这一笔碰过的那几列,值取表里那一份 */
|
|
function baselineOf(stored: Record<string, unknown>, patch: Record<string, unknown>): Record<string, unknown> {
|
|
const out: Record<string, unknown> = {};
|
|
for (const field of Object.keys(patch)) out[field] = stored[field];
|
|
return out;
|
|
}
|
|
|
|
export function createLedgerService(deps: LedgerServiceDeps): LedgerService {
|
|
function dataEvent(
|
|
ref: ObjectRef,
|
|
key: Record<string, string | number>,
|
|
changed: readonly string[],
|
|
operation: 'insert' | 'update' | 'delete',
|
|
taskId: TaskId | undefined,
|
|
cause: { kind: 'ledger-write'; ref: string },
|
|
): EnvelopeDraft {
|
|
return {
|
|
id: eventIdSchema.parse(deps.ids.next()),
|
|
type: 'data',
|
|
correlation: {
|
|
// 未绑任务的系统写不指定消费目标 —— 三级消费由 `event` 判定(OS 6.3)
|
|
...(taskId === undefined ? {} : { target: { kind: 'task' as const, id: taskId } }),
|
|
cause,
|
|
keys: Object.fromEntries(Object.entries(key).map(([k, v]) => [k, String(v)])),
|
|
},
|
|
payload: {
|
|
data: {
|
|
domain: ref.domain,
|
|
object: ref.object,
|
|
primaryKey: key,
|
|
changed: [...changed],
|
|
operation,
|
|
},
|
|
},
|
|
occurredAt: deps.clock.now().toISOString(),
|
|
source: { kind: 'engine', detail: 'ledger' },
|
|
};
|
|
}
|
|
|
|
async function requireSchema(address: ObjectAddress) {
|
|
const described = await deps.store.describe(address);
|
|
if (described === undefined) throw new ObjectNotMaterializedError(`${address.domain}.${address.object}`);
|
|
return described;
|
|
}
|
|
|
|
return {
|
|
async applyDeclarations(declarations) {
|
|
const results: SchemaApplyResult[] = [];
|
|
for (const [, group] of groupDeclarations(declarations)) {
|
|
const first = group[0]!;
|
|
const ref: ObjectRef = { domain: first.domain, object: first.object };
|
|
const merge = mergeDeclarations(group);
|
|
if (merge.merged === undefined || merge.conflicts.length > 0) {
|
|
results.push({ merged: merge.merged, notices: merge.notices, conflicts: merge.conflicts, applied: false });
|
|
continue;
|
|
}
|
|
const current = await deps.store.describe(ref);
|
|
const plan = planSchemaChanges(merge.merged, current);
|
|
const conflicts = [...merge.conflicts, ...plan.conflicts];
|
|
if (conflicts.length > 0 || !planIsNonDestructive(plan)) {
|
|
results.push({ merged: merge.merged, notices: [...merge.notices, ...plan.notices], conflicts, applied: false });
|
|
continue;
|
|
}
|
|
if (plan.changes.length > 0) await deps.store.applyChanges(ref, plan.changes);
|
|
/*
|
|
* 性质每次都记 —— 它随声明走,不随建表走。
|
|
* 声明合并已经定了口径(平台的 master 优先于活动的缺省 record),这里照写。
|
|
*/
|
|
await deps.store.setObjectMeta(ref, merge.merged.kind, merge.merged.caseKey);
|
|
results.push({
|
|
merged: merge.merged,
|
|
notices: [...merge.notices, ...plan.notices],
|
|
conflicts: [],
|
|
applied: true,
|
|
});
|
|
}
|
|
return results;
|
|
},
|
|
|
|
async describe(address) {
|
|
return deps.store.describe(address);
|
|
},
|
|
|
|
async listObjects() {
|
|
return deps.store.listObjects();
|
|
},
|
|
|
|
/** ① 系统写:世界已发生的事实,免门,即时抛事件 */
|
|
async systemWrite(write: SystemWrite) {
|
|
await requireSchema(write.address);
|
|
await deps.store.upsert(write.address, write.key, write.values);
|
|
await deps.events.publish(
|
|
dataEvent(
|
|
write.address,
|
|
write.key,
|
|
Object.keys(write.values),
|
|
'update',
|
|
write.taskId,
|
|
{ kind: 'ledger-write', ref: `${write.cause.kind}:${write.cause.ref}` },
|
|
),
|
|
);
|
|
},
|
|
|
|
/** ② 草稿写:读己之写,正式版不可见,**不产生数据事件** */
|
|
async draftWrite(write: DraftWrite) {
|
|
const schema = await requireSchema(write.address);
|
|
const existing = await deps.store.selectByKeys(write.address, [write.key]);
|
|
const baseline: Record<string, unknown> = {};
|
|
const row = existing[0];
|
|
if (row !== undefined) {
|
|
for (const field of Object.keys(write.patch)) baseline[field] = row[field];
|
|
}
|
|
await deps.drafts.put(write.taskId, write.address, {
|
|
key: pickKey({ ...write.key }, schema.primaryKey.length > 0 ? schema.primaryKey : Object.keys(write.key)),
|
|
patch: write.patch,
|
|
tombstone: write.tombstone ?? false,
|
|
baseline,
|
|
});
|
|
},
|
|
|
|
/**
|
|
* 撤掉本单草稿层的一行 —— **只撤自己写过的**。
|
|
*
|
|
* 回 `false` 表示草稿层里根本没有这一行:那多半是主键写法对不上
|
|
* (`DAG…` 与 `DAG…-R0` 是两行),如实说出来比静默成功有用 ——
|
|
* 静默成功会让模型以为撤掉了,下一拍判据照旧不过,它又去撤一次。
|
|
*/
|
|
async draftDrop(drop) {
|
|
const mine = await deps.drafts.listForObject(drop.taskId, drop.address);
|
|
const schema = await requireSchema(drop.address);
|
|
const keys = schema.primaryKey.length > 0 ? schema.primaryKey : Object.keys(drop.key);
|
|
const wanted = pickKey({ ...drop.key }, keys);
|
|
const hit = mine.some((d) => keys.every((k) => String((d.key as Record<string, unknown>)[k] ?? '') === String(wanted[k] ?? '')));
|
|
if (!hit) return false;
|
|
await deps.drafts.drop(drop.taskId, drop.address, wanted);
|
|
return true;
|
|
},
|
|
|
|
/**
|
|
* ③ 放行写:冲突检查 → 原子合并 → 此刻抛事件。
|
|
*
|
|
* `scope` 给了就是**数据工单确认那一道门**(2026-08-27 用户定:「数据工单里填的数据,
|
|
* 确认这个数据工单后就应该入库」):只并那几张表,并过的行留在草稿层打「已落地」记号 ——
|
|
* 它仍是本单写下的行(`mine:` / `_rows` / 任务视图照旧认它,不然 P-08 那张件当场变空、
|
|
* S1 的哨兵 `parts_profile_rows >= created` 永远开不了),只是整单放行时不再写它。
|
|
*/
|
|
async releaseWrite(taskId, scope?: ReleaseScope): Promise<ReleaseWriteResult> {
|
|
const conflicts = await this.checkConflicts(taskId, scope?.only);
|
|
if (conflicts.length > 0) throw new DraftConflictError(conflicts);
|
|
|
|
const drafts = (await deps.drafts.list(taskId)).filter(inScope(scope?.only));
|
|
const nothing: ReleaseWriteResult = { mergedRows: 0, emittedEvents: 0, landedEarlierRows: 0, simulatedRows: 0, merged: [] };
|
|
if (drafts.length === 0) return nothing;
|
|
const scoped = scope !== undefined;
|
|
const causeRef = scoped ? scope.cause : `release:${taskId}`;
|
|
/*
|
|
* 事件指不指回本单:整单放行指回(它正要收尾,与从前一样);数据工单那一道**不指回** ——
|
|
* 本单此刻正等着自己的人事件(卡的回传)叫醒,再喂它七八条 data 事件只会在日志里结巴,
|
|
* 而它对这些行的「读己之写」从没断过。别单的监听照旧按正式版事件认(跨任务只认过门的事实)。
|
|
*/
|
|
const target = scoped ? undefined : taskId;
|
|
|
|
/*
|
|
* **试跑单不落主数据** —— 挡在这一处,是因为这里是草稿层通往世界的唯一一道口子。
|
|
*
|
|
* 试跑照旧是真跑:过装载两审、过预算、过放行门,单据照写。挡的只有主数据 ——
|
|
* 它按业务实体分行,试跑往 `customer.profile` 建一位客户,那位客户就永久留在库里,
|
|
* 「新客户来函」那条样例第二次跑就成了老客户,这条路从此试不出来。
|
|
* 单据不挡:它按单标识分行,一次试跑写自己那一行,清运行数据就干净了。
|
|
*
|
|
* 被挡下的行仍然进 `merged`(出口载荷取的就是它)—— 拦的是**落到世界**,
|
|
* 不是拦这一单自己看得见:活动缓存里它一直在,这一单该怎么跑还怎么跑。
|
|
*/
|
|
const simulated = await deps.simulation.isSimulated(taskId);
|
|
const masters = new Set<string>();
|
|
if (simulated) {
|
|
for (const ref of await deps.store.masterRefs()) masters.add(`${ref.domain}.${ref.object}`);
|
|
}
|
|
const heldBack = (draft: { domain: string; object: string }): boolean =>
|
|
simulated && masters.has(`${draft.domain}.${draft.object}`);
|
|
let simulatedRows = 0;
|
|
let landedEarlierRows = 0;
|
|
|
|
const pending: EnvelopeDraft[] = [];
|
|
/*
|
|
* 合并后**读回来**的行 —— 「落地即事实」,事实是表里那一份,不是写进去之前的草稿。
|
|
*
|
|
* 从前这里直接回 `draft.patch`,于是出口载荷可以跟台账不一致:草稿里是字符串
|
|
* `"true"`(人的回传值天生是字符串),列是 boolean,PG 存的时候转了,可发出去的口里
|
|
* 还是字符串 —— 下游活动的入参校验当场拒收「/needSourcing must be boolean」,
|
|
* 节点起不来、报 blocked,而台账里明明是对的(实测)。
|
|
* 读回来之后,出口载荷与台账逐字一致,类型也由列定义说了算。
|
|
*/
|
|
const landed: Array<{ domain: string; object: string; key: Record<string, unknown>; values: Record<string, unknown> }> = [];
|
|
|
|
await deps.store.transaction(async (store, draftStore) => {
|
|
for (const draft of drafts) {
|
|
const ref: ObjectRef = { domain: draft.domain, object: draft.object };
|
|
if (draft.landed === true) {
|
|
/*
|
|
* 早已在数据工单确认那一道门上落过地:这次不写、不抛事件、不计入合并数;
|
|
* 只把表里那一份读回来当事实(出口载荷要它)。表里读不到(别单后来删了)就不摆。
|
|
*/
|
|
landedEarlierRows += 1;
|
|
if (draft.tombstone) continue;
|
|
const current = (await store.selectByKeys(ref, [draft.key]))[0];
|
|
if (current !== undefined) landed.push({ domain: draft.domain, object: draft.object, key: draft.key, values: current });
|
|
continue;
|
|
}
|
|
if (heldBack(draft)) {
|
|
simulatedRows += 1;
|
|
if (draft.tombstone) continue; // 试跑不删主数据,连一行也不许少
|
|
/*
|
|
* 值取「表里现有的 + 这一单改的」——没真写进去,读不回 PG 归型后的那一份,
|
|
* 所以这里的值是草稿的原样(人的回传天生是字符串)。如实记在这:
|
|
* 主数据不进出口载荷,下游拿不到它,归不归型都影响不到谁。
|
|
*/
|
|
const current = (await store.selectByKeys(ref, [draft.key]))[0];
|
|
landed.push({
|
|
domain: draft.domain,
|
|
object: draft.object,
|
|
key: draft.key,
|
|
values: { ...(current ?? {}), ...draft.patch },
|
|
});
|
|
continue;
|
|
}
|
|
if (draft.tombstone) {
|
|
await store.deleteRow(ref, draft.key);
|
|
pending.push(dataEvent(ref, draft.key, [], 'delete', target, { kind: 'ledger-write', ref: causeRef }));
|
|
if (scoped) await draftStore.markLanded(taskId, ref, draft.key, {});
|
|
continue;
|
|
}
|
|
const existing = await store.selectByKeys(ref, [draft.key]);
|
|
await store.upsert(ref, draft.key, draft.patch);
|
|
const stored = (await store.selectByKeys(ref, [draft.key]))[0];
|
|
landed.push({
|
|
domain: draft.domain,
|
|
object: draft.object,
|
|
key: draft.key,
|
|
// 读不回来只能退回草稿(理论上不该发生:刚 upsert 过);宁可退也不丢这一行
|
|
values: stored ?? draft.patch,
|
|
});
|
|
pending.push(
|
|
dataEvent(
|
|
ref,
|
|
draft.key,
|
|
Object.keys(draft.patch),
|
|
existing.length === 0 ? 'insert' : 'update',
|
|
target,
|
|
{ kind: 'ledger-write', ref: causeRef },
|
|
),
|
|
);
|
|
/*
|
|
* 数据工单那一道门:条目留着、打记号,基线换成落地后表里的值(PG 归型后的那一份)——
|
|
* 之后再往这一行写一笔,冲突检查比的才是「我改之前表里是什么」。
|
|
*/
|
|
if (scoped) await draftStore.markLanded(taskId, ref, draft.key, baselineOf(stored ?? draft.patch, draft.patch));
|
|
}
|
|
if (!scoped) await draftStore.clear(taskId);
|
|
});
|
|
|
|
for (const envelope of pending) await deps.events.publish(envelope);
|
|
return {
|
|
mergedRows: drafts.length - simulatedRows - landedEarlierRows,
|
|
emittedEvents: pending.length,
|
|
landedEarlierRows,
|
|
simulatedRows,
|
|
merged: landed,
|
|
};
|
|
},
|
|
|
|
async query(query: LedgerQuery): Promise<LedgerPage> {
|
|
const { rows, total } = await deps.store.select(query);
|
|
return { rows, total };
|
|
},
|
|
|
|
async taskView(taskId, query): Promise<LedgerPage> {
|
|
const ref: ObjectRef = { domain: query.address.domain, object: query.address.object };
|
|
const schema = await requireSchema(ref);
|
|
const [official, drafts] = await Promise.all([
|
|
deps.store.select(query),
|
|
deps.drafts.listForObject(taskId, ref),
|
|
]);
|
|
const rows = overlay(official.rows, drafts, schema.primaryKey);
|
|
return { rows, total: rows.length };
|
|
},
|
|
|
|
async taskDrafts(taskId) {
|
|
const drafts = await deps.drafts.list(taskId);
|
|
return drafts.map((d) => ({ domain: d.domain, object: d.object, key: d.key, patch: d.patch, landed: d.landed === true }));
|
|
},
|
|
|
|
async checkConflicts(taskId, only) {
|
|
const drafts = (await deps.drafts.list(taskId)).filter(inScope(only));
|
|
if (drafts.length === 0) return [];
|
|
const byObject = new Map<string, Array<(typeof drafts)[number]>>();
|
|
for (const d of drafts) {
|
|
const k = `${d.domain}.${d.object}`;
|
|
const bucket = byObject.get(k);
|
|
if (bucket === undefined) byObject.set(k, [d]);
|
|
else bucket.push(d);
|
|
}
|
|
const findings = [];
|
|
for (const [, group] of byObject) {
|
|
const first = group[0]!;
|
|
const ref: ObjectRef = { domain: first.domain, object: first.object };
|
|
const schema = await requireSchema(ref);
|
|
const rows = await deps.store.selectByKeys(
|
|
ref,
|
|
group.map((g) => g.key),
|
|
);
|
|
const officialByKey = new Map(rows.map((r) => [keyOf(pickKey(r, schema.primaryKey)), r]));
|
|
findings.push(...detectConflicts(group, officialByKey));
|
|
}
|
|
return findings;
|
|
},
|
|
|
|
async discardDrafts(taskId) {
|
|
await deps.drafts.clear(taskId);
|
|
},
|
|
};
|
|
}
|