DeepSeek provider + 判据 ③ 机器路径跑通(零人工兜底)

Provider 侧:
  · 新增 `provider-profile`(纯数据):各家的默认端点 / 结构化出参姿势 / 是否回推理
    集中一处,adapter 只按画像行事 —— 新增 provider 只改画像,不写一族 if-else。
    DeepSeek:`https://api.deepseek.com`,支持 json_object,v4 系列是推理模型。
  · 建实例时 URL 可留空 → 取该 provider 的默认端点(openai-compatible 本就是
    「随便哪个兼容端点」,没有默认,必须自填)。
  · 推理正文归 `reasoning` 单列:那是**过程**不是结论,不进出参、不参与 schema 校验、
    不入录制(录的是语义入参与结论)。
  · 两档实例 deepseek-v4-flash / deepseek-v4-pro 已建;`quote-extractor` 是**用途名**,
    指向 flash 档 —— 换档只改这一处绑定,活动一行不动。密钥只从环境读、落库即密封,
    治理台只回「是否已绑定」(实测:库里是密文,视图无明文字段,三个实例探活全通)。

沿路修掉五处真问题:

① **同一任务被并发推进**:推进器与派发器各推各的,用的不是同一把锁 ——
   实测同一步跑两遍、同一张卡发两张、同一行草稿写出两条互相打架的版本。
   两者改用同一把 advisory lock(键的推导提取共用)。
② **LLM 步出参从不校验**:声明了 outputSchema,解析不出来就静默当 `{}` 写下去,
   毛病要到下游缺字段时才炸、还炸在别处。改为「出参必过 schema」(AM 12.3),
   不过即按声明的失败走向,并把模型原文记进留痕,好让人看见它到底回了什么。
③ **出参路径不认点号**:入参能走 `a.b.0.c`,出参只认扁平键 ——
   模型按 schema 回嵌套结构(最常见的情形)就写出一条空记录。两边共用一份解析。
④ **归一步拿不到原文**:通道载荷只带指针不带正文、start→A1 边也没把主题正文传下去,
   模型只拿到一个单号,于是回空 records。通道载荷带原文,边映射一路带到 A1。
⑤ **出参形状没讲给模型听**:声明了 schema 却只在提示里写「只输出 JSON」,
   模型只能猜键名(实测回 `{"items":[]}` 而声明要 `{"records":[...]}`)。
   改由网关按 outputSchema 统一附上说明 —— 这是系统的事,不该每个活动作者各写一遍。

另修:连接池没有 error 处理器,空闲连接被服务端掐断会让整个进程退出(一次运维操作
就能把 API 带崩)。

结果:一封 RFQ 入站 → 模型从正文抽出 pn/brand/eau/目标价/交期 →
D1 判定跳过补全 → 查历史价 → DMN 三分类 Type A → 验收放行卡,**全程零人工兜底**。

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

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

apps/api/src/app/container.ts

@@ -150,6 +150,14 @@ export const firstPartyToolHandlers = new Map<string, ToolHandler>(FIRST_PARTY_T
 
 export async function createContainer(config: AppConfig): Promise<Container> {
   const pool = new Pool({ connectionString: config.databaseUrl, max: 10 });
+  /*
+   * 空闲连接被服务端掐断(重启、`pg_terminate_backend`、网络抖动)会以 'error' 事件抛到池上;
+   * **没有处理器就是未捕获异常,整个进程随之退出** —— 一次运维操作能把 API 带崩。
+   * 池会自行剔除坏连接并在下次取用时重连,这里只负责让它别掀桌子。
+   */
+  pool.on('error', (error: Error) => {
+    process.stderr.write(`[pg] 空闲连接异常(已剔除,下次取用自动重连):${error.message}\n`);
+  });
   const db = drizzle(pool);
   const ids = createIdGenerator();
   const secrets = createSecretBox(loadSecretKey());

apps/api/src/app/seed-quote.ts

@@ -102,17 +102,37 @@ export async function seedQuote(container: Container): Promise<void> {
     process.stdout.write(line + '\n');
   }
 
-  /* ① LLM 实例:四元绑定(活动只见自定义模型名) */
+  /*
+   * ① LLM 实例:四元绑定(活动只见自定义模型名)。
+   *
+   * 密钥**只从环境读**,不入源码、不入包、不入留痕(OS 7.3 凭证隔离 / ADR-009);
+   * 落库即密封,治理台也只回「是否已绑定」。URL 留空 → 取 provider 画像的默认端点。
+   */
+  const deepseekKey = process.env['JCAI_DEEPSEEK_API_KEY'] ?? process.env['JCAI_LLM_API_KEY'] ?? 'sk-placeholder';
+
+  // 两档 DeepSeek:flash 快而省,pro 强而慢 —— 活动按自己的场景声明引用哪一档
+  for (const [name, model, systemUse] of [
+    ['deepseek-v4-flash', 'deepseek-v4-flash', false],
+    ['deepseek-v4-pro', 'deepseek-v4-pro', false],
+  ] as const) {
+    await container.llm.upsert({ name, provider: 'deepseek', baseUrl: '', model, apiKey: deepseekKey, systemUse, enabled: true });
+  }
+
+  /*
+   * 活动声明里引用的逻辑名是 `quote-extractor` —— 它是**用途名**,不是型号名。
+   * 把它指到 flash 档:受理归一是抽取活,快档足够;要换档只改这一处绑定,活动一行不动。
+   * 系统用途(叙述性小结 / 汇报 / 蒸馏)也走它。
+   */
   await container.llm.upsert({
     name: 'quote-extractor',
-    provider: 'openai-compatible',
-    baseUrl: process.env['JCAI_LLM_BASE_URL'] ?? 'https://api.deepseek.com/v1',
-    model: process.env['JCAI_LLM_MODEL'] ?? 'deepseek-chat',
-    apiKey: process.env['JCAI_LLM_API_KEY'] ?? 'sk-placeholder',
+    provider: 'deepseek',
+    baseUrl: process.env['JCAI_LLM_BASE_URL'] ?? '',
+    model: process.env['JCAI_LLM_MODEL'] ?? 'deepseek-v4-flash',
+    apiKey: deepseekKey,
     systemUse: true,
     enabled: true,
   });
-  log('① LLM 实例已绑定:quote-extractor');
+  log('① LLM 实例已绑定:deepseek-v4-flash · deepseek-v4-pro · quote-extractor(→ flash,系统用途)');
 
   /* ② 工具注册两审 + 白名单 */
   const tools = [
@@ -248,6 +268,9 @@ export async function seedQuote(container: Container): Promise<void> {
         mapping: [
           { target: 'customerCode', source: { from: 'initiation', path: 'customerCode' } },
           { target: 'rfqRef', source: { from: 'initiation', path: 'rfqRef' } },
+          // 主题与正文要一路带到 A1:归一步拆的就是它们
+          { target: 'subject', source: { from: 'initiation', path: 'subject' } },
+          { target: 'body', source: { from: 'initiation', path: 'body' } },
         ],
         reopen: false,
       },
@@ -372,6 +395,8 @@ export async function seedQuote(container: Container): Promise<void> {
       { target: 'customerCode', source: { from: 'expression', sfeel: 'keys.customerCode' } },
       { target: 'rfqRef', source: { from: 'expression', sfeel: 'keys.rfqRef' } },
       { target: 'subject', source: { from: 'expression', sfeel: 'subject' } },
+      // 正文是**归一步唯一的原始素材**:只给单号,模型没有可拆解的东西
+      { target: 'body', source: { from: 'expression', sfeel: 'body' } },
     ],
     spec: { kind: 'event', rule: 'contains(subject, "询价") or contains(subject, "RFQ")', eventType: 'data' },
   });

apps/api/src/app/wiring/engines.ts

@@ -284,6 +284,7 @@ export function wireEngines(deps: EngineWiringDeps): EngineWiring {
     evaluate: sfeel.condition,
     data: taskData,
     tools: { ...toolGateway, invoke: (i) => toolGateway.invoke(i) },
+    schemas: deps.schemas,
     llm: {
       async generate(input) {
         // LLM 在 flow 里只以「被调用的工具」出现,不是调度者
@@ -296,7 +297,12 @@ export function wireEngines(deps: EngineWiringDeps): EngineWiring {
             stepNo: input.stepNo,
             label: 'flow-step',
           });
-          return { ok: result.object !== undefined, ...(result.object === undefined ? { error: '出参不合 schema' } : { object: result.object }) };
+          // 解析与校验归引擎(它才知道这一步声明了什么);这里只如实上交结果与原文
+          return {
+            ok: true,
+            ...(result.object === undefined ? {} : { object: result.object }),
+            text: result.text,
+          };
         } catch (error) {
           return { ok: false, error: error instanceof Error ? error.message : String(error) };
         }

apps/api/src/app/workers/dispatcher.ts

@@ -22,14 +22,22 @@ export type DispatcherDeps = {
   readonly batchSize?: number;
 };
 
-/** 目标 ID → advisory lock 键(稳定散列,同一目标恒同一把锁) */
-function lockKey(target: Delivery['target']): number {
-  const text = `${target.kind}:${target.id}`;
+/**
+ * 目标 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);

apps/api/src/app/workers/task-advancer.ts

@@ -8,6 +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';
 
 export type AdvancerOptions = {
   readonly container: Container;
@@ -27,19 +28,33 @@ export async function advanceOnce(options: AdvancerOptions): Promise<number> {
 
   let advanced = 0;
   for (const row of rows) {
-    for (let tick = 0; tick < (options.maxTicks ?? 3); tick += 1) {
-      try {
-        const outcome = await container.task.tick(row.id as never);
-        advanced += 1;
-        // 进入等待或终结就停手 —— 挂起在等的任务由事件唤醒,不空转
-        if (outcome.task.runtime !== 'running' || outcome.task.terminal !== undefined) break;
-        if (!outcome.progressed) break;
-      } catch (error) {
-        process.stderr.write(
-          `[task-advancer] ${row.id}:${error instanceof Error ? error.message : String(error)}\n`,
-        );
-        break;
+    /*
+     * 与派发器共用同一把 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 {
+      for (let tick = 0; tick < (options.maxTicks ?? 3); tick += 1) {
+        try {
+          const outcome = await container.task.tick(row.id as never);
+          advanced += 1;
+          // 进入等待或终结就停手 —— 挂起在等的任务由事件唤醒,不空转
+          if (outcome.task.runtime !== 'running' || outcome.task.terminal !== undefined) break;
+          if (!outcome.progressed) break;
+        } catch (error) {
+          process.stderr.write(
+            `[task-advancer] ${row.id}:${error instanceof Error ? error.message : String(error)}\n`,
+          );
+          break;
+        }
       }
+    } finally {
+      await container.db.execute(sql`select pg_advisory_unlock(${key})`);
     }
   }
   return advanced;