diff --git a/.data/attachments/inbound/e2e-20260728E1@x.com.eml b/.data/attachments/inbound/e2e-20260728E1@x.com.eml new file mode 100644 index 0000000..0bb2381 --- /dev/null +++ b/.data/attachments/inbound/e2e-20260728E1@x.com.eml @@ -0,0 +1,12 @@ +{ + "messageId": "e2e-20260728E1@x.com", + "from": "buyer@acme.com", + "to": [ + "rfq@we.com" + ], + "cc": [], + "subject": "【询价】客户 C801 RFQ-20260728E1 求报价", + "body": "客户编码 C801,询价单号 RFQ-��\nLM358DR 厂牌 TI 年用量 12000 片 目标价 8.20 元 交期 10 周", + "receivedAt": "2026-07-28T12:00:00.000Z", + "attachments": [] +} \ No newline at end of file diff --git a/.data/attachments/inbound/e2e-20260728E2@x.com.eml b/.data/attachments/inbound/e2e-20260728E2@x.com.eml new file mode 100644 index 0000000..8661967 --- /dev/null +++ b/.data/attachments/inbound/e2e-20260728E2@x.com.eml @@ -0,0 +1,12 @@ +{ + "messageId": "e2e-20260728E2@x.com", + "from": "buyer@acme.com", + "to": [ + "rfq@we.com" + ], + "cc": [], + "subject": "【询价】客户 C802 RFQ-20260728E2 求报价", + "body": "客户编码 C802,询价单号 RFQ-��\nNE555P 厂牌 TI 年用量 12000 片 目标价 8.20 元 交期 10 周", + "receivedAt": "2026-07-28T12:00:00.000Z", + "attachments": [] +} \ No newline at end of file diff --git a/.data/attachments/inbound/e2e-loop-01@x.com.eml b/.data/attachments/inbound/e2e-loop-01@x.com.eml new file mode 100644 index 0000000..9bd4466 --- /dev/null +++ b/.data/attachments/inbound/e2e-loop-01@x.com.eml @@ -0,0 +1,12 @@ +{ + "messageId": "e2e-loop-01@x.com", + "from": "buyer@acme.com", + "to": [ + "rfq@we.com" + ], + "cc": [], + "subject": "【询价】客户 C810 RFQ-20260728L1 求报价", + "body": "客户编码 C810,询价单号 RFQ-20260728L1。\nNE555P 厂牌 TI 年用量 6000 片 目标价 3.10 元 交期 12 周", + "receivedAt": "2026-07-28T13:00:00.000Z", + "attachments": [] +} \ No newline at end of file diff --git a/.data/attachments/inbound/e2e-loop-02@x.com.eml b/.data/attachments/inbound/e2e-loop-02@x.com.eml new file mode 100644 index 0000000..c82e347 --- /dev/null +++ b/.data/attachments/inbound/e2e-loop-02@x.com.eml @@ -0,0 +1,12 @@ +{ + "messageId": "e2e-loop-02@x.com", + "from": "buyer@acme.com", + "to": [ + "rfq@we.com" + ], + "cc": [], + "subject": "【询价】客户 C820 RFQ-20260728L2 求报价", + "body": "客户编码 C820,询价单号 RFQ-20260728L2。\nNE555P 厂牌 TI 年用量 6000 片 目标价 3.10 元 交期 12 周", + "receivedAt": "2026-07-28T14:00:00.000Z", + "attachments": [] +} \ No newline at end of file diff --git a/.data/attachments/inbound/e2e-loop-03@x.com.eml b/.data/attachments/inbound/e2e-loop-03@x.com.eml new file mode 100644 index 0000000..9e6b2e8 --- /dev/null +++ b/.data/attachments/inbound/e2e-loop-03@x.com.eml @@ -0,0 +1,12 @@ +{ + "messageId": "e2e-loop-03@x.com", + "from": "buyer@acme.com", + "to": [ + "rfq@we.com" + ], + "cc": [], + "subject": "【询价】客户 C830 RFQ-20260728L3 求报价", + "body": "客户编码 C830,询价单号 RFQ-20260728L3。\nNE555P 厂牌 TI 年用量 6000 片 目标价 3.10 元 交期 12 周", + "receivedAt": "2026-07-28T15:00:00.000Z", + "attachments": [] +} \ No newline at end of file diff --git a/.data/attachments/inbound/e2e-loop-04@x.com.eml b/.data/attachments/inbound/e2e-loop-04@x.com.eml new file mode 100644 index 0000000..c242ead --- /dev/null +++ b/.data/attachments/inbound/e2e-loop-04@x.com.eml @@ -0,0 +1,12 @@ +{ + "messageId": "e2e-loop-04@x.com", + "from": "buyer@acme.com", + "to": [ + "rfq@we.com" + ], + "cc": [], + "subject": "【询价】客户 C840 RFQ-20260728L4 求报价", + "body": "客户编码 C840,询价单号 RFQ-20260728L4。\nNE555P 厂牌 TI 年用量 6000 片 目标价 3.10 元 交期 12 周", + "receivedAt": "2026-07-28T16:00:00.000Z", + "attachments": [] +} \ No newline at end of file diff --git a/.data/attachments/inbound/e2e-loop-05@x.com.eml b/.data/attachments/inbound/e2e-loop-05@x.com.eml new file mode 100644 index 0000000..70c8ab7 --- /dev/null +++ b/.data/attachments/inbound/e2e-loop-05@x.com.eml @@ -0,0 +1,12 @@ +{ + "messageId": "e2e-loop-05@x.com", + "from": "buyer@acme.com", + "to": [ + "rfq@we.com" + ], + "cc": [], + "subject": "【询价】客户 C850 RFQ-20260728L5 求报价", + "body": "客户编码 C850,询价单号 RFQ-20260728L5。\nNE555P 厂牌 TI 年用量 6000 片 目标价 3.10 元 交期 12 周", + "receivedAt": "2026-07-28T17:00:00.000Z", + "attachments": [] +} \ No newline at end of file diff --git a/activities/quote-archive/src/manifest.ts b/activities/quote-archive/src/manifest.ts index 301f6a5..a61bf46 100644 --- a/activities/quote-archive/src/manifest.ts +++ b/activities/quote-archive/src/manifest.ts @@ -15,7 +15,7 @@ export const manifest: ActivityManifest = { identity: { activityId: ACTIVITY_ID, name: 'A7 · 情报归档', - packageVersion: '0.1.4', + packageVersion: '0.1.5', category: '统一报价', paradigm: 'flow', sdkVersion: '0.1.0-m5', @@ -30,6 +30,9 @@ export const manifest: ActivityManifest = { type: 'object', required: ['customerCode', 'quoteNo'], properties: { + /** 归档写 (pn, quoted_at) 这一行,两个主键都由调用方给定 */ + pn: { type: 'string' }, + quotedAt: { type: 'string' }, customerCode: { type: 'string' }, quoteNo: { type: 'string' }, reason: { type: 'string' }, diff --git a/api-report/apps/api/src/modules/loop-engine/port.d.ts b/api-report/apps/api/src/modules/loop-engine/port.d.ts index 31a1574..ee74196 100644 --- a/api-report/apps/api/src/modules/loop-engine/port.d.ts +++ b/api-report/apps/api/src/modules/loop-engine/port.d.ts @@ -90,6 +90,37 @@ export type ListenerRegistrar = { id: string; }>; }; +/** + * 视图流控 —— app 注入 `interaction`(L3)。 + * 「loop:AI 按定位与局面临场组织视图流」(AM 07)。MVP 每回合推一次**案例全景**: + * 阶段与计划项的生命周期态、当下可为、目标缺口 —— 这些是语义引擎算出来的确定事实, + * 零 LLM 即可成屏;模型只决定「下一步做什么」,不负责画面。 + */ +export type ViewGateway = { + push(input: { + taskId: TaskId; + activityId: string; + template: string; + scope: Record; + }): Promise; +}; +/** + * 技能 —— app 注入 `skill-engine`(L4)。 + * 「渐进披露:常驻上下文**仅有技能目录**,按事加载正文、按需下钻引用」(AM 08)。 + */ +export type SkillGateway = { + /** 目录:id + 何时适用 —— 只给这些,正文等 AI 说要才加载 */ + index(activityId: string): Promise>; + /** 装配入信封:这一回合装配了哪些技能及其版本,记入任务记忆 */ + stamp(activityId: string, skillIds: readonly string[]): Promise>; +}; export type MemoryRecorder = { append(input: { taskId: TaskId; diff --git a/api-report/packages/activity-sdk/src/index.d.ts b/api-report/packages/activity-sdk/src/index.d.ts index a80d724..a39b727 100644 --- a/api-report/packages/activity-sdk/src/index.d.ts +++ b/api-report/packages/activity-sdk/src/index.d.ts @@ -10,4 +10,6 @@ export { defineActivity } from './define.ts'; export type { ActivityMemoryLayer, ChatCardDraft, FormFieldDraft, LedgerKey, LedgerRow, ViewBlockLike, WorkOrderDraft, } from './types.ts'; export type { HostOptions, HostRecording, MockHost } from './host.ts'; export { createMockHost } from './host.ts'; +export type { Tone } from './view.ts'; +export { kvCard, noteCard, stagesCard, tableCard, viewShell } from './view.ts'; export declare const SDK_VERSION = "0.1.0-m5"; diff --git a/apps/api/src/app/agent-runner.ts b/apps/api/src/app/agent-runner.ts index 7dad669..c52e0e5 100644 --- a/apps/api/src/app/agent-runner.ts +++ b/apps/api/src/app/agent-runner.ts @@ -22,7 +22,21 @@ const ACTION_SCHEMA = { tool: { type: 'string' }, input: { type: 'object' }, workOrder: { enum: ['execution', 'acceptance'] }, - card: { type: 'object' }, + /* + * 卡面结构由 schema 框住 —— 「调用必附上下文,**裸调用不得发出**」(AM 06)。 + * 一张只写「请处理」的卡等于没说话:收卡的人不知道要答什么、凭什么答。 + */ + card: { + type: 'object', + required: ['title', 'ask'], + properties: { + title: { type: 'string' }, + ask: { type: 'string' }, + reason: { type: 'string' }, + basis: { type: 'array' }, + fields: { type: 'array' }, + }, + }, eventType: { enum: ['data', 'time'] }, keyFilter: { type: 'object' }, deadline: { type: 'string' }, @@ -51,6 +65,9 @@ export function createAgentRunner(deps: AgentRunnerDeps): { '你是这一单任务的案例工作者。下面是语义引擎发布的**当下可为**与**目标缺口**。', '你只能选一个行动,且必须落在三型行动之内:调工具 / 调人 / 立监听(外加 start、plan、disable、judge、submit)。', '规矩:①触达世界的动作只能 tool(写类会被登记为动作意向,放行时才落地)', + '④调人(human)必须把卡填实:title 一句话说清要什么、ask 写明具体问题、', + ' basis 给出判断依据(键值对数组)、fields 列出要对方回填的字段名。', + ' 只写「请处理」的卡收卡人无从下手,等于白发一张。', '②缺口只有人知道 → human;缺口在世界侧 → listen(必须给 deadline,ISO-8601)', '③成果已齐才 submit。只输出 JSON。', '', @@ -109,6 +126,32 @@ export function createAgentRunner(deps: AgentRunnerDeps): { } /** 模型出参归一为 AgentAction;不合规一律退化为「立监听等人」而不是乱动世界 */ +/** + * 卡面兜底:模型没填全就用局面补。 + * 「调用必附上下文」是硬规矩 —— 与其发一张「请处理」,不如把当下可为与缺口摆上去, + * 收卡的人至少知道自己站在哪儿。 + */ +function fillCard( + card: Record, + enabled: EnabledSet, + itemId: string, +): Record { + const item = enabled.entries.find((e) => e.id === itemId); + const title = String(card['title'] ?? '').trim(); + const ask = String(card['ask'] ?? '').trim(); + return { + ...card, + title: title === '' ? `请处理:${item?.name ?? itemId}` : title, + ask: ask === '' ? '模型未给出具体问题,请按当下可为与缺口自行判断下一步。' : ask, + basis: Array.isArray(card['basis']) + ? card['basis'] + : [ + { k: '当下可为', v: enabled.entries.map((e) => e.name).join('、') || '(空)' }, + { k: '目标缺口', v: enabled.gaps.join('、') || '(无)' }, + ], + }; +} + function normalize(object: unknown, enabled: EnabledSet): AgentAction { const raw = (object ?? {}) as Record; const kind = String(raw['kind'] ?? ''); @@ -138,7 +181,8 @@ function normalize(object: unknown, enabled: EnabledSet): AgentAction { return { kind: 'human', itemId, - card: (raw['card'] ?? {}) as Record, + // 卡面缺项用**当下局面**补全 —— 宁可信息略多,不发一张空卡 + card: fillCard((raw['card'] ?? {}) as Record, enabled, itemId), workOrder: raw['workOrder'] === 'acceptance' ? 'acceptance' : 'execution', }; case 'listen': diff --git a/apps/api/src/app/seed-quote.ts b/apps/api/src/app/seed-quote.ts index 1845c56..03ae393 100644 --- a/apps/api/src/app/seed-quote.ts +++ b/apps/api/src/app/seed-quote.ts @@ -368,6 +368,9 @@ export async function seedQuote(container: Container): Promise { mapping: [ { target: 'customerCode', source: { from: 'initiation', path: 'customerCode' } }, { target: 'quoteNo', source: { from: 'initiation', path: 'rfqRef' } }, + // 归档写的是 (pn, quoted_at) 这一行 —— 两个主键都得传到,它自己造不出来 + { target: 'pn', source: { from: 'output-port', port: 'intake', path: 'pn' } }, + { target: 'quotedAt', source: { from: 'initiation', path: 'quotedAt' } }, ], reopen: false, }, @@ -397,6 +400,14 @@ export async function seedQuote(container: Container): Promise { { target: 'subject', source: { from: 'expression', sfeel: 'subject' } }, // 正文是**归一步唯一的原始素材**:只给单号,模型没有可拆解的东西 { target: 'body', source: { from: 'expression', sfeel: 'body' } }, + /* + * 单据日期:取入站时刻。 + * MVP 简化 —— 真实系统里「报价日期」应取报价单**实际发出**的时刻(出站回执), + * 而不是询价到达的时刻;两者在一单之内通常只差几天,不影响历史价情报的可用性。 + * 之所以要有它:`shared.history_price` 的主键是 (pn, quoted_at), + * 归档若拿不到日期就写不进去 —— 主键只增不破坏,不能为了省事去改它。 + */ + { target: 'quotedAt', source: { from: 'expression', sfeel: 'occurredAt' } }, ], spec: { kind: 'event', rule: 'contains(subject, "询价") or contains(subject, "RFQ")', eventType: 'data' }, }); diff --git a/apps/api/src/app/wiring/engines.ts b/apps/api/src/app/wiring/engines.ts index 46d7f6a..5284f55 100644 --- a/apps/api/src/app/wiring/engines.ts +++ b/apps/api/src/app/wiring/engines.ts @@ -248,9 +248,20 @@ export function wireEngines(deps: EngineWiringDeps): EngineWiring { }) { 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) })); + + /* + * 依据快照:优先用活动给的结构化 `basis`(`{k,v}[]`);没给才把卡面其余键铺成依据。 + * 铺的时候要跳过 `ask`(它是问题不是依据)与空值,嵌套对象转 JSON —— + * 不然人看到的是一行 `[object Object]`,那还不如不给。 + */ + const declared = input.card['basis']; + const basis = Array.isArray(declared) + ? declared + .filter((b): b is { k: unknown; v: unknown } => typeof b === 'object' && b !== null && 'k' in b) + .map((b) => ({ k: String(b.k), v: String(b.v) })) + : Object.entries(input.card) + .filter(([k, v]) => !['title', 'fields', 'ask', 'basis'].includes(k) && v !== null && v !== undefined && v !== '') + .map(([k, v]) => ({ k, v: typeof v === 'object' ? JSON.stringify(v) : String(v) })); if (input.workOrder === 'acceptance') { const issued = await deps.humanTool.issueAcceptance(context, { title, diff --git a/apps/api/src/app/workers/__tests__/task-lock.test.ts b/apps/api/src/app/workers/__tests__/task-lock.test.ts new file mode 100644 index 0000000..4c3090d --- /dev/null +++ b/apps/api/src/app/workers/__tests__/task-lock.test.ts @@ -0,0 +1,48 @@ +/** 按目标串行的互斥锁 —— 同一目标不得被并发推进 */ +import { describe, expect, it } from 'vitest'; +import { createKeyedMutex } from '../task-lock.ts'; + +describe('KeyedMutex', () => { + it('同一 key 并发时只有一个进得去,其余立刻返回 undefined(不排队堆积)', async () => { + const mutex = createKeyedMutex(); + let concurrent = 0; + let peak = 0; + const body = async (): Promise => { + concurrent += 1; + peak = Math.max(peak, concurrent); + await new Promise((r) => setTimeout(r, 20)); + concurrent -= 1; + return 'ran'; + }; + + const results = await Promise.all([ + mutex.tryRun('t1', body), + mutex.tryRun('t1', body), + mutex.tryRun('t1', body), + ]); + + expect(peak).toBe(1); + expect(results.filter((r) => r === 'ran')).toHaveLength(1); + expect(results.filter((r) => r === undefined)).toHaveLength(2); + }); + + it('不同 key 互不阻塞 —— 别的任务不该被这个任务拖住', async () => { + const mutex = createKeyedMutex(); + const results = await Promise.all([ + mutex.tryRun('t1', async () => 'a'), + mutex.tryRun('t2', async () => 'b'), + ]); + expect(results).toEqual(['a', 'b']); + }); + + it('抛错也要释放 —— 一次失败不能把这个任务永久锁死', async () => { + const mutex = createKeyedMutex(); + await expect( + mutex.tryRun('t1', async () => { + throw new Error('boom'); + }), + ).rejects.toThrow('boom'); + expect(mutex.held).toBe(0); + expect(await mutex.tryRun('t1', async () => 'ok')).toBe('ok'); + }); +}); diff --git a/apps/api/src/app/workers/dispatcher.ts b/apps/api/src/app/workers/dispatcher.ts index 2d54fec..faf7e4a 100644 --- a/apps/api/src/app/workers/dispatcher.ts +++ b/apps/api/src/app/workers/dispatcher.ts @@ -12,6 +12,7 @@ import { sql } from 'drizzle-orm'; import type { Delivery } from '@modules/event'; import { claimPendingDeliveries } from '@modules/event'; import type { Container } from '../container.ts'; +import { taskMutex } from './task-lock.ts'; /** 消费者:按目标类型分派 */ export type Consumer = (delivery: Delivery, container: Container) => Promise; @@ -22,60 +23,38 @@ export type DispatcherDeps = { readonly batchSize?: number; }; -/** - * 目标 ID → advisory lock 键(稳定散列,同一目标恒同一把锁)。 - * - * **推进器与派发器必须共用这一把**:一边由事件投递驱动推进任务、一边由推进器轮询推进, - * 用的不是同一把锁就等于没锁 —— 实测会同一步跑两遍、同一张卡发两张、 - * 同一行草稿写出两条互相打架的版本。 - */ -export function targetLockKey(kind: string, id: string): number { - const text = `${kind}:${id}`; - let hash = 5381; - for (let i = 0; i < text.length; i += 1) hash = ((hash << 5) + hash + text.charCodeAt(i)) | 0; - return hash; -} - -const lockKey = (target: Delivery['target']): number => targetLockKey(target.kind, String(target.id)); - export async function dispatchOnce(deps: DispatcherDeps): Promise { const { container } = deps; const batch = await claimPendingDeliveries(container.db, deps.batchSize ?? 20); let handled = 0; for (const delivery of batch) { - const key = lockKey(delivery.target); // 同任务按序:取到锁才处理,取不到说明同目标的另一条正在处理,留待下一轮 - const locked = await container.db.execute<{ ok: boolean }>( - sql`select pg_try_advisory_lock(${key}) as ok`, - ); - if (locked.rows[0]?.ok !== true) continue; - - try { - const consumer = deps.consumers.get(delivery.target.kind); - if (consumer !== undefined) { - const claimed = await container.event.claimConsumption( - delivery.eventId, - `${delivery.target.kind}:${delivery.target.id}`, + await taskMutex.tryRun(`${delivery.target.kind}:${String(delivery.target.id)}`, async () => { + try { + const consumer = deps.consumers.get(delivery.target.kind); + if (consumer !== undefined) { + const claimed = await container.event.claimConsumption( + delivery.eventId, + `${delivery.target.kind}:${delivery.target.id}`, + ); + // 至少一次 + 幂等:已消费过的重复投递直接标记完成,不再执行副作用 + if (claimed) await consumer(delivery, container); + } + await container.db.execute( + sql`update sys.event_delivery set state = 'delivered' where id = ${delivery.id}`, ); - // 至少一次 + 幂等:已消费过的重复投递直接标记完成,不再执行副作用 - if (claimed) await consumer(delivery, container); + // 已落库、已送达之后才广播 —— 广播不是投递,只是告诉在线浏览器「这个目标动了」 + container.broadcaster.publish({ + kind: delivery.target.kind === 'message-position' ? 'staff' : 'task', + id: String(delivery.target.id), + reason: delivery.target.kind, + }); + handled += 1; + } catch (error) { + await container.event.reportFailure(delivery.id, error instanceof Error ? error.message : String(error)); } - await container.db.execute( - sql`update sys.event_delivery set state = 'delivered' where id = ${delivery.id}`, - ); - // 已落库、已送达之后才广播 —— 广播不是投递,只是告诉在线浏览器「这个目标动了」 - container.broadcaster.publish({ - kind: delivery.target.kind === 'message-position' ? 'staff' : 'task', - id: String(delivery.target.id), - reason: delivery.target.kind, - }); - handled += 1; - } catch (error) { - await container.event.reportFailure(delivery.id, error instanceof Error ? error.message : String(error)); - } finally { - await container.db.execute(sql`select pg_advisory_unlock(${key})`); - } + }); } return handled; } diff --git a/apps/api/src/app/workers/task-advancer.ts b/apps/api/src/app/workers/task-advancer.ts index 01e4ae3..01bdde3 100644 --- a/apps/api/src/app/workers/task-advancer.ts +++ b/apps/api/src/app/workers/task-advancer.ts @@ -8,7 +8,7 @@ import { and, eq, sql } from 'drizzle-orm'; import { taskTable } from '@modules/task'; import type { Container } from '../container.ts'; -import { targetLockKey } from './dispatcher.ts'; +import { taskMutex } from './task-lock.ts'; export type AdvancerOptions = { readonly container: Container; @@ -29,16 +29,10 @@ export async function advanceOnce(options: AdvancerOptions): Promise { let advanced = 0; for (const row of rows) { /* - * 与派发器共用同一把 advisory lock:事件投递也会推进同一个任务。 + * 与派发器共用同一把锁:事件投递也会推进同一个任务。 * 抢不到就跳过 —— 那边正推着,这边再推就是两条线同时改一份运行时状态。 */ - const key = targetLockKey('task', row.id); - const locked = await container.db.execute<{ ok: boolean }>( - sql`select pg_try_advisory_lock(${key}) as ok`, - ); - if (locked.rows[0]?.ok !== true) continue; - - try { + const ran = await taskMutex.tryRun(`task:${row.id}`, async () => { for (let tick = 0; tick < (options.maxTicks ?? 3); tick += 1) { try { const outcome = await container.task.tick(row.id as never); @@ -53,9 +47,9 @@ export async function advanceOnce(options: AdvancerOptions): Promise { break; } } - } finally { - await container.db.execute(sql`select pg_advisory_unlock(${key})`); - } + return true; + }); + void ran; } return advanced; } diff --git a/apps/api/src/app/workers/task-lock.ts b/apps/api/src/app/workers/task-lock.ts new file mode 100644 index 0000000..d7e8fec --- /dev/null +++ b/apps/api/src/app/workers/task-lock.ts @@ -0,0 +1,36 @@ +/** + * 按目标串行的进程内互斥锁。 + * + * 为什么不能只靠 `pg_try_advisory_lock`:它是**会话级**的,而我们经连接池取连接 —— + * 加锁与解锁多半落在不同连接上,解锁解不掉自己那把;同一连接被复用时又因为可重入而 + * 直接放行。两头都不对,结果就是同一个任务被并发推进:同一步跑两遍、同一张卡发两张。 + * + * MVP 单进程,进程内互斥就够且确定。多进程部署时再叠一层 + * `pg_try_advisory_xact_lock`(事务级,随事务自动释放,不依赖连接归属)。 + */ +export type KeyedMutex = { + /** 抢到锁才执行;抢不到返回 undefined —— 调用方据此跳过,不排队堆积 */ + tryRun(key: string, fn: () => Promise): Promise; + readonly held: number; +}; + +export function createKeyedMutex(): KeyedMutex { + const busy = new Set(); + return { + async tryRun(key, fn) { + if (busy.has(key)) return undefined; + busy.add(key); + try { + return await fn(); + } finally { + busy.delete(key); + } + }, + get held() { + return busy.size; + }, + }; +} + +/** 任务推进与事件投递共用同一把 —— 用的不是同一把锁就等于没锁 */ +export const taskMutex = createKeyedMutex(); diff --git a/apps/api/src/modules/loop-engine/__tests__/cmmn.test.ts b/apps/api/src/modules/loop-engine/__tests__/cmmn.test.ts index 9d16a83..62c1a2e 100644 --- a/apps/api/src/modules/loop-engine/__tests__/cmmn.test.ts +++ b/apps/api/src/modules/loop-engine/__tests__/cmmn.test.ts @@ -1,6 +1,7 @@ /** CMMN 子集解析、生命周期与 DMN —— domain 测试零 mock(AM 03 / 04) */ import { describe, expect, it } from 'vitest'; import type { DmnDecisionTable, DmnModel } from '@contracts'; +import type { CaseRuntime } from '../internal/domain/lifecycle.ts'; import { CmmnParseError, parseCmmn } from '../internal/domain/cmmn-parse.ts'; import { autoCompleteReady, @@ -286,3 +287,29 @@ describe('DRD —— 缺口演绎与拓扑序', () => { expect(() => evaluationOrder(cyclic, 'a')).toThrow(/成环/); }); }); + +describe('先验迁移再动手 —— 不合法的行动不留痕迹', () => { + it('Available → Disabled 是非法迁移,transitionItem 抛错', () => { + const runtime: CaseRuntime = { + items: [{ id: 'T1', state: 'Available', instance: 1 }], + milestones: [], + state: 'Active', + }; + // 引擎语义忠实:不私改迁移拓扑。AI 提了就拒,拒在**副作用之前** + expect(() => transitionItem(runtime, 'T1', 'Disabled')).toThrow(); + }); + + it('不存在的计划项同样拒 —— AI 可能报一个图上没有的 id', () => { + const runtime: CaseRuntime = { items: [], milestones: [], state: 'Active' }; + expect(() => transitionItem(runtime, '不存在', 'Active')).toThrow(); + }); + + it('Available → Active 合法', () => { + const runtime: CaseRuntime = { + items: [{ id: 'T1', state: 'Available', instance: 1 }], + milestones: [], + state: 'Active', + }; + expect(transitionItem(runtime, 'T1', 'Active').items[0]?.state).toBe('Active'); + }); +}); diff --git a/apps/api/src/modules/loop-engine/internal/usecase/turn-runner.ts b/apps/api/src/modules/loop-engine/internal/usecase/turn-runner.ts index 6824423..6d49188 100644 --- a/apps/api/src/modules/loop-engine/internal/usecase/turn-runner.ts +++ b/apps/api/src/modules/loop-engine/internal/usecase/turn-runner.ts @@ -118,6 +118,8 @@ export function createTurnRunner(deps: TurnRunnerDeps): LoopEngineService { } case 'tool': { + // 先验迁移再动手:不合法的行动不该在世界上留下痕迹(登记意向、调工具都算) + const done = transitionItem(runtime, action.itemId, 'Completed'); // 写类工具登记为动作意向 —— 触达世界的动作集中在放行时落地 if (await deps.tools.isWriteEffect(action.tool)) { await deps.tools.registerIntent({ @@ -126,7 +128,7 @@ export function createTurnRunner(deps: TurnRunnerDeps): LoopEngineService { input: action.input, label: `${action.itemId}:${action.tool}`, }); - return { runtime: transitionItem(runtime, action.itemId, 'Completed'), status: 'running' }; + return { runtime: done, status: 'running' }; } const result = await deps.tools.invoke({ tool: action.tool, @@ -141,13 +143,20 @@ export function createTurnRunner(deps: TurnRunnerDeps): LoopEngineService { detail: { output: result.output, error: result.error?.message }, }); return { - runtime: transitionItem(runtime, action.itemId, result.ok ? 'Completed' : 'Failed'), + runtime: result.ok ? done : transitionItem(runtime, action.itemId, 'Failed'), status: 'running', }; } case 'human': { - // 调人 = 立监听听人:调用即 Active(阻塞),回传驱动前提 + /* + * 调人 = 立监听听人:调用即 Active(阻塞),回传驱动前提。 + * + * **先验迁移,再发卡**:迁移非法时抛在前面,世界上不会留下一张已发出的卡。 + * 反过来写(先发卡后迁移)会让每一次被拒的回合都白发一张卡 —— 卡收不回, + * 负责人的工作台上就堆起一摞谁也不知道该怎么答的「请处理」。 + */ + const advanced = transitionItem(runtime, action.itemId, 'Active'); const card = await deps.cards.issue({ taskId, ownerId: context.ownerId, @@ -161,7 +170,7 @@ export function createTurnRunner(deps: TurnRunnerDeps): LoopEngineService { keyFilter: { cardId: card.id }, deadline: 'P3D', }); - return { runtime: transitionItem(runtime, action.itemId, 'Active'), status: 'awaiting' }; + return { runtime: advanced, status: 'awaiting' }; } case 'listen': { @@ -274,7 +283,29 @@ export function createTurnRunner(deps: TurnRunnerDeps): LoopEngineService { if (item === undefined) throw new ForbiddenTerminalError('owner'); } - const applied = await applyAction(taskId, action, runtime); + /* + * AI 提的行动可能**不合法**(违反生命周期迁移拓扑、越权、预算触顶)。 + * 引擎拒绝它是对的 —— 「语义忠实:引擎不得私改迁移拓扑」;但**不能因此死掉**: + * 抛出去会让这一回合的副作用(卡已发出)留在世界上、运行时却没落笔, + * 推进器于是一遍遍重来,每遍再发一张卡。 + * + * 正确做法:拒绝 + 留痕 + 把拒绝理由喂回下一回合的上下文。 + * AI 提出非法行动是预期内的事,引擎的职责是不执行,不是崩掉。 + * 反复提不出合法行动由 Harness 兜底(步预算与停滞熔断),那才是它该管的事。 + */ + let applied: { runtime: CaseRuntime; status: TurnOutcome['status']; reason?: string }; + try { + applied = await applyAction(taskId, action, runtime); + } catch (error) { + const reason = error instanceof Error ? error.message : String(error); + await deps.memory.append({ + taskId, + kind: 'result', + title: `行动被拒:${reason}`, + detail: { action, hint: '请在「当下可为」里另选一项,或改用调人 / 立监听' }, + }); + return { runtime, enabled, navigation, action, status: 'running' }; + } await deps.state.save(taskId, model, parsed, applied.runtime); /*