M2: 基座能力 —— memory / toolbox / channel / message

- memory 任务记忆四性质(只追加·信封盖戳·治理先行·归侧)、活动记忆四层与条目三件套、任务日志两级投影(结构化零 LLM / 叙述性小结不入事实库)、度量八指标只读投影 - toolbox 注册两审(静态审 + 试运行审)、效应等级决定运行时待遇、契约演进只增不破坏、白名单与纯工具私有作用域、ToolInvoker(两缝之二,live/record/replay) - channel 通道实例配置、入站归一(解析 → 系统写 → 数据事件)、出站发送与关联记录登记 - message 通知 / 跨任务征答(就地作答 · 驳回重答只增不改)/ 死信 / 无主兜底 app:message 接上 event 的 MessageFallback;派发器按目标 ID 取 advisory lock 串行消费 + 消费幂等表;/api/ingest 模拟邮件通道(mock 只在外界) 已验证:无主 RFQ 邮件入站 → 归属解析 no-match → 事件落库 → 兜底落 OP 岗位的消息要素门禁五道全绿;150 个单元测试 + 1 个 PG 集成测试 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01KeDBx2PHLavbURXB8LJJ7m
2026-07-27 01:16:52 +08:00
parent 14d4fa046a
commit f0444f1417
73 changed files with 7488 additions and 31 deletions
@@ -0,0 +1,87 @@
+/**
+ * 投递派发器 —— ADR-008 #2。
+ *
+ * 按 `state='pending' and next_attempt_at <= now()` 取投递记录,
+ * 对**目标 ID 取 advisory lock** 串行执行 —— 这正是「同任务按序」的实现(§7.1)。
+ * 消费幂等由 `sys.event_consumption(event_id, consumer)` 唯一约束承担。
+ *
+ * 消费者(任务 / 触发器)随 M4 / M5 落地;在此之前派发器把投递标记为已送达并留痕,
+ * 保证投递链路本身可测、不积压(docs/debt.md · M1-3)。
+ */
+import { sql } from 'drizzle-orm';
+import type { Delivery } from '@modules/event';
+import { claimPendingDeliveries } from '@modules/event';
+import type { Container } from '../container.ts';
+
+/** 消费者:按目标类型分派 */
+export type Consumer = (delivery: Delivery, container: Container) => Promise<void>;
+
+export type DispatcherDeps = {
+  readonly container: Container;
+  readonly consumers: ReadonlyMap<Delivery['target']['kind'], Consumer>;
+  readonly batchSize?: number;
+};
+
+/** 目标 ID → advisory lock 键(稳定散列,同一目标恒同一把锁) */
+function lockKey(target: Delivery['target']): number {
+  const text = `${target.kind}:${target.id}`;
+  let hash = 5381;
+  for (let i = 0; i < text.length; i += 1) hash = ((hash << 5) + hash + text.charCodeAt(i)) | 0;
+  return hash;
+}
+
+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}`,
+        );
+        // 至少一次 + 幂等:已消费过的重复投递直接标记完成,不再执行副作用
+        if (claimed) await consumer(delivery, container);
+      }
+      await container.db.execute(
+        sql`update sys.event_delivery set state = 'delivered' where id = ${delivery.id}`,
+      );
+      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;
+}
+
+export function startDispatcher(deps: DispatcherDeps, intervalMs: number): { stop(): void } {
+  let stopped = false;
+  const tick = async (): Promise<void> => {
+    if (stopped) return;
+    try {
+      await dispatchOnce(deps);
+    } catch (error) {
+      process.stderr.write(`[dispatcher] ${error instanceof Error ? error.message : String(error)}\n`);
+    }
+  };
+  const timer = setInterval(() => void tick(), intervalMs);
+  timer.unref();
+  return {
+    stop() {
+      stopped = true;
+      clearInterval(timer);
+    },
+  };
+}