调优:并发推进的真根因 + loop 的三处行为修正

E2E 跑 loop 时 A3 一口气发了八张一模一样的「请处理」卡。逐层查下去,四处:

① **并发推进的真根因**:`pg_try_advisory_lock` 是**会话级**的,而我们经连接池取连接 ——
   加锁与解锁多半落在不同连接上,解锁解不掉自己那把;同一连接被复用时又因可重入而直接放行。
   两头都不对,于是同一任务被并发推进(现象:同一步跑两遍、同一张卡发两张、
   同一行草稿写出两条互相打架的版本)。前一轮「共用同一把锁」的修法方向对、器械错。
   改用**进程内按目标互斥**(MVP 单进程,确定且够用),推进器与派发器共用一把;
   多进程部署时再叠事务级 `pg_try_advisory_xact_lock`(随事务释放,不依赖连接归属)。
   实测:同一单从 8 张卡降到 1 张,2 回合后落到等待。

② **非法行动会把回合打死**:AI 提了不合法的生命周期迁移(Available → Disabled),
   `applyAction` 抛出、运行时没落笔、推进器一遍遍重来。引擎拒绝它是对的
   ——「语义忠实:引擎不得私改迁移拓扑」——但不该因此崩掉。
   改为:拒绝 + 留痕 + 把理由喂回下一回合的上下文。AI 提出非法行动是预期内的事。

③ **副作用抢在校验之前**:卡在迁移合法性判定**之前**就发出去了,于是每个被拒的回合
   都白发一张卡 —— 卡收不回,负责人工作台上堆起一摞没人知道怎么答的「请处理」。
   human / tool 两个分支改为**先验迁移再动手**。

④ **裸调用**:模型只回一个「请处理」。按「调用必附上下文,裸调用不得发出」(AM 06)
   把卡面结构写进出参 schema(title / ask / basis / fields 必填),提示里讲明要求,
   归一时用当下局面兜底;卡片装配也修了 —— 结构化 basis 优先,嵌套对象转 JSON,
   不再出现一行 `[object Object]`。
   实测卡面:「请确认 NE555P 历史报价并判定采用策略」+ 三个回填字段。

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

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KeDBx2PHLavbURXB8LJJ7m

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<string> => {
+      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');
+  });
+});

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<void>;
@@ -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<number> {
   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;
 }

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<number> {
   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<number> {
           break;
         }
       }
-    } finally {
-      await container.db.execute(sql`select pg_advisory_unlock(${key})`);
-    }
+      return true;
+    });
+    void ran;
   }
   return advanced;
 }

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<T>(key: string, fn: () => Promise<T>): Promise<T | undefined>;
+  readonly held: number;
+};
+
+export function createKeyedMutex(): KeyedMutex {
+  const busy = new Set<string>();
+  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();