Merge Meyer batch analysis integration into test

apps/pac-service/src/integrations/meyer/README.md

@@ -51,6 +51,16 @@
 
 从 `apps/pac-service` 执行。凭据只从 `MEYER_APP_KEY`、`MEYER_APP_SECRET` 环境变量读取；必须具备 SaaS 已同步诊所及患者所属应用的读取权限。`baseUrl` 使用 HTTPS，localhost HTTP 仅供本地模拟服务测试。
 
+### 当前接口契约（美亚 v0.4）
+
+- 影像列表使用 `query-by-clinic`，每页默认 200 张，上限 200；每轮处理条数仍受共享扫描额度限制。
+- 新扫描和待分析补查均使用 `symptom-batch`，每批最多 200 个 `imageIds`。不再向美亚发送旧的单张病症查询，也不发送患者或诊所参数。PAC 仍按诊所合批，先核对影像身份并保存影像，再查分析。
+- 批量结果按 `imageId` 关联，不依赖数组位置。缺项、空分析和单项格式异常分别留底、按原规则补查；重复或额外 ID 拒绝整批，不能错挂到其他患者。
+- 已保存的旧分页断点继续使用原 `pageSize` 完成固定窗口，下个窗口才采用新配置。不要删除断点或修改旧页大小。
+- 授权按 `accessExpiresAt` 缓存，提前 30 秒在下次调用时更新；优先轮换一次性 refresh token，失败或网络结果不确定时丢弃旧 refresh token，再用应用凭据签发。并发调用及同一客户端的不同时间窗口共享一次换取操作。
+- HTTP 401 或封套 `code=401/401001` 仅触发一次重新认证与业务重试。到期时长以美亚返回值为准，不写死为两小时或七天；令牌不写入文件或日志。
+- 不依赖 Java SDK，不新增影像事件推送；继续由现有单一定时器轮询。
+
 复制 `config.example.json` 到仓库外，填入允许采集的 clinicIds 和时间范围；namespace 是稳定的应用来源名，不要随运行改变。状态目录按环境和应用隔离，包含患者数据，应受权限及备份策略保护。
 
 ```sh
@@ -69,7 +79,7 @@ pnpm meyer-imaging -- --config=/absolute/meyer.json --state-dir=/absolute/privat
 
 ## 本期取舍与上线前核实
 
-- 已用测试患者完成真实只读联调：PDF 三项身份参数可用，分析可落库，文件 ID 可换取 OSS 签名地址。详见 QA 记录。
+- 历史联调记录中的单张三项参数属于旧契约；当前按上节 v0.4 批量接口执行。文件 ID 继续按既有接口换取 OSS 签名地址，文件地址接口的上限仍为 100，不能与影像列表/分析的 200 混用。
 - 无更新游标、完整快照分页和删除标记，无法承诺完整生命周期同步；分页数量变化、重复/缺页会报错，重跑固定窗口。只在空响应和缺字段时延后，不能把它们当作阴性结论。
 - 宿主若以后推送同一批美亚影像，必须先统一来源标识和接入责任，避免与本路线重复。首期应只开明确授权的试点诊所。
 - 测试环境曾运行固定窗口联调；这不等于已启用持续任务。启用前应核实实际进程、启动配置、持久状态及扫描水位。开发回归使用本地 HTTP 与独立 PostgreSQL 数据库。
@@ -166,7 +176,7 @@ pnpm meyer-imaging -- --config=/absolute/meyer.json --state-dir=/absolute/privat
 | requestIntervalMs | 500 | 串行请求间隔至少 500ms |
 | maxListPagesPerTick | 10 | 新影像扫描每轮最多查询 10 页（含续跑核对）；待处理项逐项也受此上限约束 |
 | errorBackoffMaxMs | 3600000 | 连续异常最长等待 1 小时 |
-| pageSize | 100 | 每次列表响应最多 100 张；不是每轮处理限额 |
+| pageSize | 200 | 每次列表响应最多 200 张；不是每轮处理限额，旧固定窗口断点保留原大小 |
 
 达到分析次数上限后，人工确认需要补查时，可使用同一配置、同一持久状态目录，把固定窗口限定为对应影像入库时刻，再执行默认单次命令（`--apply`，不带 `--tick/--watch`）。这次显式查询可越过等待时间和次数上限；若仍为空或异常，继续保留人工状态。它不绕过影像类型、身份或落库校验，也不重算已完成分析。
 

apps/pac-service/src/integrations/meyer/client.ts

@@ -3,14 +3,19 @@ import { z } from 'zod';
 import type { MeyerConfig, MeyerImage, MeyerScope, MeyerSource } from './contracts';
 
 const PageSchema = z.object({ total: z.number().int().nonnegative(), page: z.number().int().positive(),
-  size: z.number().int().min(1).max(100), images: z.array(z.unknown()).max(100) });
-const TokenSchema = z.object({ accessToken: z.string().min(1), accessExpiresAt: z.string().datetime({ offset: true }) });
+  size: z.number().int().min(1).max(200), images: z.array(z.unknown()).max(200) });
+const TokenSchema = z.object({ accessToken: z.string().min(1), accessExpiresAt: z.string().datetime({ offset: true }),
+  refreshToken: z.string().min(1).nullish(), refreshExpiresAt: z.string().datetime({ offset: true }).nullish() });
+interface TokenSession {
+  token?: { value: string; expires: number; refresh?: string; refreshExpires?: number };
+  refreshing?: Promise<string>;
+}
 
-/** 只实现三种只读操作；不接入 PAC 主服务，不调用美亚写接口。 */
+/** 只读影像、分析与文件查询；同一应用的时间窗口共享授权，不调用美亚写接口。 */
 export class MeyerClient implements MeyerSource {
   constructor(private readonly config: MeyerConfig,
     private readonly credentials: { appKey: string; secret: string },
-    private readonly session: { token?: { value: string; expires: number } } = {}) {
+    private readonly session: TokenSession = {}) {
     if (!credentials.appKey.trim() || !credentials.secret.trim()) throw new Error('Meyer credentials required');
   }
   forWindow(config: MeyerConfig): MeyerClient {
@@ -48,20 +53,45 @@ export class MeyerClient implements MeyerSource {
   }
   private async getToken(): Promise<string> {
     if (this.session.token && this.session.token.expires > Date.now() + 30000) return this.session.token.value;
-    const result = await this.request('/iam/external/oauth/tokens/pub', {
-      appKey: this.credentials.appKey, secret: this.credentials.secret, clientType: 'EXTERNAL_APP',
-    });
+    if (!this.session.refreshing) {
+      this.session.refreshing = this.refreshOrIssue().finally(() => { this.session.refreshing = undefined; });
+    }
+    return this.session.refreshing;
+  }
+  private acceptToken(result: { code: number; data?: unknown }): string {
     if (result.code !== 200) throw new Error(`Meyer authentication code=${result.code}`);
     const token = TokenSchema.safeParse(result.data);
     if (!token.success || Date.parse(token.data.accessExpiresAt) <= Date.now()) throw new Error('Meyer invalid token response');
-    this.session.token = { value: token.data.accessToken, expires: Date.parse(token.data.accessExpiresAt) };
+    this.session.token = { value: token.data.accessToken, expires: Date.parse(token.data.accessExpiresAt),
+      refresh: token.data.refreshToken ?? undefined,
+      refreshExpires: token.data.refreshExpiresAt ? Date.parse(token.data.refreshExpiresAt) : undefined };
     return this.session.token.value;
   }
+  private async refreshOrIssue(): Promise<string> {
+    const previous = this.session.token;
+    // 一次性 refreshToken 先移出缓存；网络结果不确定时也不能再次提交旧值。
+    this.session.token = undefined;
+    if (previous?.refresh && (previous.refreshExpires === undefined || previous.refreshExpires > Date.now())) {
+      try {
+        return this.acceptToken(await this.request('/iam/external/oauth/refreshes/pub', { refreshToken: previous.refresh }));
+      } catch { /* 不重试旧刷新令牌，回退到应用凭据签发新会话。 */ }
+    }
+    const result = await this.request('/iam/external/oauth/tokens/pub', {
+      appKey: this.credentials.appKey, secret: this.credentials.secret, clientType: 'EXTERNAL_APP',
+    });
+    return this.acceptToken(result);
+  }
   private async query(path: string, body: unknown): Promise<unknown> {
-    let result = await this.request(path, body, await this.getToken());
-    if (result.code === 401) {
-      this.session.token = undefined;
-      result = await this.request(path, body, await this.getToken());
+    let used = await this.getToken();
+    let result = await this.request(path, body, used);
+    if (result.code === 401 || result.code === 401001) {
+      // 并发返回的旧 401 不能清掉另一请求刚更新的令牌。
+      if (this.session.token?.value === used) this.session.token.expires = 0;
+      used = await this.getToken();
+      result = await this.request(path, body, used);
+      if ((result.code === 401 || result.code === 401001) && this.session.token?.value === used) {
+        this.session.token.expires = 0;
+      }
     }
     if (result.code !== 200) throw new Error(`Meyer business failure code=${result.code}: ${path}`);
     return result.data;
@@ -87,10 +117,26 @@ export class MeyerClient implements MeyerSource {
     }
     return result.urls;
   }
-  symptom(image: MeyerImage): Promise<unknown> {
-    // PDF 比旧 SDK 多 patientId，按文档携带完整归属；上线前需实测服务端版本。
-    return this.query('/imaging/external/images/symptom', {
-      imageClinicId: image.clinicId, patientId: image.patientId, imageId: image.imageId,
-    });
+  async symptoms(images: readonly MeyerImage[]): Promise<ReadonlyMap<string, unknown>> {
+    if (images.some(image => !this.config.scopes.some(s => s.clinicId === image.clinicId))) {
+      throw new Error('analysis clinic outside configured scope');
+    }
+    const ids = z.array(z.string().trim().min(1).max(256)).min(1).max(200).parse(images.map(image => image.imageId));
+    const imageIds = [...new Set(ids)];
+    const raw = await this.query('/imaging/external/images/symptom-batch', { imageIds });
+    const rows = z.array(z.object({ imageId: z.string().min(1).max(256) }).passthrough()).max(200).safeParse(raw);
+    if (!rows.success) throw new Error('Meyer invalid batch symptom response');
+    const result = new Map<string, unknown>();
+    for (const row of rows.data) {
+      if (!imageIds.includes(row.imageId) || result.has(row.imageId)) throw new Error('Meyer ambiguous batch symptom image');
+      const { imageId, ...symptom } = row;
+      result.set(imageId, symptom);
+    }
+    return result;
+  }
+  async symptom(image: MeyerImage): Promise<unknown> {
+    const result = await this.symptoms([image]);
+    if (!result.has(image.imageId)) throw new Error('Meyer batch symptom omitted image');
+    return result.get(image.imageId);
   }
 }

apps/pac-service/src/integrations/meyer/collector.ts

@@ -26,6 +26,181 @@ export async function collectMeyer(config: MeyerConfig, source: MeyerSource, sto
   const pending = new MeyerStore(join(store.directory, 'pending'));
   const result: CollectResult = { pages: 0, seen: 0, collected: 0, applied: 0,
     alreadyApplied: 0, deferred: 0, failed: 0, imagesApplied: 0, analysisFailed: 0, errors: [] };
+  type AnalysisRequest = { image: MeyerImage; begin: () => Promise<void> };
+  async function* collectImage(raw: unknown, scope: MeyerScope, seen: Set<string>, page: number): AsyncGenerator<AnalysisRequest, void, unknown> {
+    result.seen++;
+    let key = `page-${page}-row-${result.seen}`;
+    let retry: PendingImage | undefined;
+    let completed = false;
+    let imageReady = false;
+    let requestedAnalysis = false;
+    try {
+      const parsed = ImageSchema.safeParse(raw);
+      if (!parsed.success) throw new Error('invalid image contract');
+      const image = parsed.data;
+      key = imageKey(scope, image.imageId);
+      if (seen.has(image.imageId)) throw new Error('duplicate image across pages; rerun fixed window');
+      seen.add(image.imageId);
+      if (image.clinicId !== scope.clinicId) throw new Error('image clinic mismatch');
+      if (image.imageCreateTime && (sourceTime(image.imageCreateTime) < sourceTime(config.timeStart) ||
+        sourceTime(image.imageCreateTime) > sourceTime(config.timeEnd))) throw new Error('image outside requested window');
+      const previousRetry = await pending.read<PendingImage>(key);
+      // 超限条目只按键查看，不扫描人工目录；重叠窗口和手工固定窗口也不能绕过恢复入口。
+      if (await identityReviewStore(store).read(key)) { result.deferred++; return; }
+      if (!identityRetryInFlight && previousRetry && await retireIdentity(store, key, previousRetry, now)) { result.deferred++; return; }
+      if (respectRetryAt && previousRetry && isIdentityPending(previousRetry) &&
+        Date.parse(previousRetry.nextRetryAt) > now) { result.deferred++; return; }
+      retry = { scope, image, attempts: (previousRetry?.attempts ?? 0) + 1, nextRetryAt: '',
+        identityAttempts: previousRetry?.identityAttempts ?? 0,
+        ...(previousRetry?.identityRecovery ? { identityRecovery: previousRetry.identityRecovery } : {}),
+        analysisAttempts: previousRetry?.analysisAttempts ?? 0, autoRetry: true, reason: 'image_or_identity' };
+      const saved = await store.read<SavedBundle>(`bundle-${key}`);
+      const metadata = await store.read<SavedBundle>(`image-${key}`);
+      for (const retained of [saved, metadata]) {
+        if (!retained) continue;
+        if (hash(retained.bundle.scope) !== hash(scope) || retained.bundle.namespace !== config.namespace ||
+          retained.bundle.raw.image.imageId !== image.imageId ||
+          ![1, 2].includes(retained.bundle.version) ||
+          (retained === metadata && (retained.bundle.version !== 2 || retained.bundle.phase !== 'image')) ||
+          (retained === saved && retained.bundle.version === 2 && retained.bundle.phase !== 'analysis')) {
+          throw new Error('saved bundle scope mismatch');
+        }
+        if (retained.bundle.image.patientExternalId !== image.patientId) {
+          throw new MeyerIdentityError('image ownership changed; reconciliation is not supported');
+        }
+      }
+      if (saved?.applied) { completed = true; result.alreadyApplied++; return; }
+      // v1 重试仍使用原完整快照。新采集必须先成功保存影像，再请求分析。
+      if (saved?.bundle.version !== 1) {
+        const imageBundle = metadata?.bundle ?? mapMeyerImage(config, scope, image, new Date().toISOString());
+        if (!imageBundle) {
+          if (!image.patientId) retry.identityIssue = 'missing patientId';
+          await store.write(`deferred-${key}`, { reason: 'unbound patient or missing image creation time', image });
+          result.deferred++; return;
+        }
+        if (!metadata) await store.write(`image-${key}`, { bundle: imageBundle, applied: false } satisfies SavedBundle);
+        if (sink && !metadata?.applied) {
+          await sink.apply(imageBundle);
+          await store.write(`image-${key}`, { bundle: imageBundle, applied: true } satisfies SavedBundle);
+          result.imagesApplied++;
+        }
+        imageReady = true;
+      }
+      if (!saved) {
+        const capability = analysisCapability(image);
+        if (capability !== 'supported') {
+          await store.write(`analysis-status-${key}`, { checkedAt: new Date(now).toISOString(),
+            imageId: image.imageId, capability, reason: capability === 'unsupported'
+              ? 'image type not supported for analysis' : 'image type unknown; manual verification required' });
+          if (capability === 'unsupported') completed = !!sink;
+          else { retry.autoRetry = false; retry.reason = 'analysis_type_unknown'; }
+          result.deferred++; return;
+        }
+        if (previousRetry?.reason === 'analysis_type_unknown') await store.remove(`analysis-status-${key}`);
+        if (respectRetryAt && retry.analysisAttempts! >= config.maxAnalysisAttempts) {
+          retry.autoRetry = false; retry.reason = 'analysis_exhausted';
+          result.deferred++; return;
+        }
+      }
+      if (respectRetryAt && previousRetry && Date.parse(previousRetry.nextRetryAt) > now) {
+        // 不因扫描重遇同一行延长原重试时间。
+        retry = undefined; result.deferred++; return;
+      }
+      let bundle = saved?.bundle;
+      if (!bundle) {
+        const observedAt = new Date().toISOString();
+        if (!image.patientId) {
+          await store.write(`deferred-${key}`, { reason: 'unbound patient', observedAt, image });
+          result.deferred++; return;
+        }
+        retry.reason = 'analysis_pending';
+        // 先完成本页影像入库，再合并已通过身份校验的分析请求。
+        const rawSymptom = yield { image, begin: async () => {
+          retry!.analysisAttempts!++;
+          retry!.nextRetryAt = nextAnalysisRetry(config, image, now);
+          // 批量中的每张影像仍独立持久化次数，不能因重启重新取得额度。
+          await pending.write(key, retry!);
+          requestedAnalysis = true;
+        } };
+        // 错误/空/无时间都留底，但不发布阴性结论，也不宣称分析完成。
+        await store.write(`observed-${key}`, { observedAt, image, symptom: rawSymptom ?? null });
+        if (rawSymptom === undefined) throw new Error('Meyer batch symptom omitted image');
+        const symptom = SymptomSchema.safeParse(rawSymptom);
+        if (!symptom.success) throw new Error('invalid symptom contract');
+        bundle = mapMeyerAnalysis(config, scope, image, symptom.data, codes, observedAt) ?? undefined;
+        if (!bundle) {
+          await store.write(`deferred-${key}`, { observedAt,
+            reason: 'analysis not yet available' });
+          result.deferred++; return;
+        }
+        // 写入前固定完整快照；失败重试使用同一快照，不混入后来变化的远端结果。
+        await store.write(`bundle-${key}`, { bundle, applied: false } satisfies SavedBundle);
+      }
+      result.collected++;
+      retry.reason = 'analysis_apply';
+      if (sink) {
+        await sink.apply(bundle);
+        await store.write(`bundle-${key}`, { bundle, applied: true } satisfies SavedBundle);
+        result.applied++;
+      }
+      completed = !!sink;
+    } catch (error) {
+      if (error instanceof MeyerIdentityError && retry) retry.identityIssue = error.message;
+      if (requestedAnalysis && retry?.reason !== 'analysis_apply') retry!.reason = 'analysis_error';
+      if (imageReady) result.analysisFailed++; else result.failed++;
+      result.errors.push({ key, reason: error instanceof Error ? error.message : 'collection failed' });
+    } finally {
+      if (retry) {
+        if (completed) await pending.remove(key);
+        else {
+          if (retry.reason === 'analysis_pending' || retry.reason === 'analysis_error' || retry.reason === 'analysis_exhausted') {
+            retry.nextRetryAt = nextAnalysisRetry(config, retry.image, now);
+            if (retry.analysisAttempts! >= config.maxAnalysisAttempts) {
+              retry.autoRetry = false; retry.reason = 'analysis_exhausted';
+            }
+          } else retry.nextRetryAt = new Date(now + Math.min(3600000, 60000 * 2 ** Math.min(6, retry.attempts - 1))).toISOString();
+          await pending.write(key, retry);
+          await retireIdentity(store, key, retry, now);
+        }
+      }
+    }
+  }
+  async function collectPage(images: unknown[], scope: MeyerScope, seen: Set<string>, page: number) {
+    const jobs: Array<{ iterator: AsyncGenerator<AnalysisRequest, void, unknown>; request: AnalysisRequest }> = [];
+    try {
+      for (const raw of images) {
+        const iterator = collectImage(raw, scope, seen, page);
+        const prepared = await iterator.next();
+        if (!prepared.done) jobs.push({ iterator, request: prepared.value });
+      }
+      if (source.symptoms) {
+        if (!jobs.length) return;
+        if (source.canRequestAnalysis?.() === false) { result.deferred += jobs.length; return; }
+        const ready: typeof jobs = [];
+        for (const job of jobs) {
+          try { await job.request.begin(); ready.push(job); }
+          catch (error) { await job.iterator.throw(error); }
+        }
+        if (!ready.length) return;
+        let responses: ReadonlyMap<string, unknown>;
+        try { responses = await source.symptoms(ready.map(job => job.request.image)); }
+        catch (error) { for (const job of ready) await job.iterator.throw(error); return; }
+        for (const job of ready) await job.iterator.next(responses.get(job.request.image.imageId));
+      } else {
+        // 仅兼容历史本地来源；实际 MeyerClient 一律调用 symptom-batch。
+        for (const job of jobs) {
+          if (source.canRequestAnalysis?.() === false) { result.deferred++; continue; }
+          try {
+            await job.request.begin();
+            await job.iterator.next(await source.symptom(job.request.image));
+          } catch (error) { await job.iterator.throw(error); }
+        }
+      }
+    } finally {
+      // 预算用尽或持久化中断时也执行每张影像的待处理落盘。
+      for (const job of jobs) await job.iterator.return();
+    }
+  }
   for (const scope of config.scopes) {
     const seen = new Set<string>();
     let total: number | undefined;
@@ -41,145 +216,7 @@ export async function collectMeyer(config: MeyerConfig, source: MeyerSource, sto
         total = batch.total;
         const expected = Math.min(batch.size, Math.max(0, total - (page - 1) * batch.size));
         if (batch.images.length !== expected) throw new Error('incomplete page; no successful checkpoint');
-        for (const raw of batch.images) {
-          result.seen++;
-          let key = `page-${page}-row-${result.seen}`;
-          let retry: PendingImage | undefined;
-          let completed = false;
-          let imageReady = false;
-          let requestedAnalysis = false;
-          try {
-            const parsed = ImageSchema.safeParse(raw);
-            if (!parsed.success) throw new Error('invalid image contract');
-            const image = parsed.data;
-            key = imageKey(scope, image.imageId);
-            if (seen.has(image.imageId)) throw new Error('duplicate image across pages; rerun fixed window');
-            seen.add(image.imageId);
-            if (image.clinicId !== scope.clinicId) throw new Error('image clinic mismatch');
-            if (image.imageCreateTime && (sourceTime(image.imageCreateTime) < sourceTime(config.timeStart) ||
-              sourceTime(image.imageCreateTime) > sourceTime(config.timeEnd))) throw new Error('image outside requested window');
-            const previousRetry = await pending.read<PendingImage>(key);
-            // 超限条目只按键查看，不扫描人工目录；重叠窗口和手工固定窗口也不能绕过恢复入口。
-            if (await identityReviewStore(store).read(key)) { result.deferred++; continue; }
-            if (!identityRetryInFlight && previousRetry && await retireIdentity(store, key, previousRetry, now)) { result.deferred++; continue; }
-            if (respectRetryAt && previousRetry && isIdentityPending(previousRetry) &&
-              Date.parse(previousRetry.nextRetryAt) > now) { result.deferred++; continue; }
-            retry = { scope, image, attempts: (previousRetry?.attempts ?? 0) + 1, nextRetryAt: '',
-              identityAttempts: previousRetry?.identityAttempts ?? 0,
-              ...(previousRetry?.identityRecovery ? { identityRecovery: previousRetry.identityRecovery } : {}),
-              analysisAttempts: previousRetry?.analysisAttempts ?? 0, autoRetry: true, reason: 'image_or_identity' };
-            const saved = await store.read<SavedBundle>(`bundle-${key}`);
-            const metadata = await store.read<SavedBundle>(`image-${key}`);
-            for (const retained of [saved, metadata]) {
-              if (!retained) continue;
-              if (hash(retained.bundle.scope) !== hash(scope) || retained.bundle.namespace !== config.namespace ||
-                retained.bundle.raw.image.imageId !== image.imageId ||
-                ![1, 2].includes(retained.bundle.version) ||
-                (retained === metadata && (retained.bundle.version !== 2 || retained.bundle.phase !== 'image')) ||
-                (retained === saved && retained.bundle.version === 2 && retained.bundle.phase !== 'analysis')) {
-                throw new Error('saved bundle scope mismatch');
-              }
-              if (retained.bundle.image.patientExternalId !== image.patientId) {
-                throw new MeyerIdentityError('image ownership changed; reconciliation is not supported');
-              }
-            }
-            if (saved?.applied) { completed = true; result.alreadyApplied++; continue; }
-            // v1 重试仍使用原完整快照。新采集必须先成功保存影像，再请求分析。
-            if (saved?.bundle.version !== 1) {
-              const imageBundle = metadata?.bundle ?? mapMeyerImage(config, scope, image, new Date().toISOString());
-              if (!imageBundle) {
-                if (!image.patientId) retry.identityIssue = 'missing patientId';
-                await store.write(`deferred-${key}`, { reason: 'unbound patient or missing image creation time', image });
-                result.deferred++; continue;
-              }
-              if (!metadata) await store.write(`image-${key}`, { bundle: imageBundle, applied: false } satisfies SavedBundle);
-              if (sink && !metadata?.applied) {
-                await sink.apply(imageBundle);
-                await store.write(`image-${key}`, { bundle: imageBundle, applied: true } satisfies SavedBundle);
-                result.imagesApplied++;
-              }
-              imageReady = true;
-            }
-            if (!saved) {
-              const capability = analysisCapability(image);
-              if (capability !== 'supported') {
-                await store.write(`analysis-status-${key}`, { checkedAt: new Date(now).toISOString(),
-                  imageId: image.imageId, capability, reason: capability === 'unsupported'
-                    ? 'image type not supported for analysis' : 'image type unknown; manual verification required' });
-                if (capability === 'unsupported') completed = !!sink;
-                else { retry.autoRetry = false; retry.reason = 'analysis_type_unknown'; }
-                result.deferred++; continue;
-              }
-              if (previousRetry?.reason === 'analysis_type_unknown') await store.remove(`analysis-status-${key}`);
-              if (respectRetryAt && retry.analysisAttempts! >= config.maxAnalysisAttempts) {
-                retry.autoRetry = false; retry.reason = 'analysis_exhausted';
-                result.deferred++; continue;
-              }
-            }
-            if (respectRetryAt && previousRetry && Date.parse(previousRetry.nextRetryAt) > now) {
-              // 不因扫描重遇同一行延长原重试时间。
-              retry = undefined; result.deferred++; continue;
-            }
-            let bundle = saved?.bundle;
-            if (!bundle) {
-              const observedAt = new Date().toISOString();
-              if (!image.patientId) {
-                await store.write(`deferred-${key}`, { reason: 'unbound patient', observedAt, image });
-                result.deferred++; continue;
-              }
-              if (source.canRequestAnalysis?.() === false) {
-                retry.reason = 'analysis_pending';
-                result.deferred++; continue;
-              }
-              retry.analysisAttempts!++;
-              retry.reason = 'analysis_pending';
-              retry.nextRetryAt = nextAnalysisRetry(config, image, now);
-              // 请求前落盘，进程中断、异常或后续扫描不能重置查询次数。
-              await pending.write(key, retry);
-              requestedAnalysis = true;
-              const rawSymptom = await source.symptom(image);
-              // 错误/空/无时间都留底，但不发布阴性结论，也不宣称分析完成。
-              await store.write(`observed-${key}`, { observedAt, image, symptom: rawSymptom });
-              const symptom = SymptomSchema.safeParse(rawSymptom);
-              if (!symptom.success) throw new Error('invalid symptom contract');
-              bundle = mapMeyerAnalysis(config, scope, image, symptom.data, codes, observedAt) ?? undefined;
-              if (!bundle) {
-                await store.write(`deferred-${key}`, { observedAt,
-                  reason: 'analysis not yet available' });
-                result.deferred++; continue;
-              }
-              // 写入前固定完整快照；失败重试使用同一快照，不混入后来变化的远端结果。
-              await store.write(`bundle-${key}`, { bundle, applied: false } satisfies SavedBundle);
-            }
-            result.collected++;
-            retry.reason = 'analysis_apply';
-            if (sink) {
-              await sink.apply(bundle);
-              await store.write(`bundle-${key}`, { bundle, applied: true } satisfies SavedBundle);
-              result.applied++;
-            }
-            completed = !!sink;
-          } catch (error) {
-            if (error instanceof MeyerIdentityError && retry) retry.identityIssue = error.message;
-            if (requestedAnalysis && retry?.reason !== 'analysis_apply') retry!.reason = 'analysis_error';
-            if (imageReady) result.analysisFailed++; else result.failed++;
-            result.errors.push({ key, reason: error instanceof Error ? error.message : 'collection failed' });
-          } finally {
-            if (retry) {
-              if (completed) await pending.remove(key);
-              else {
-                if (retry.reason === 'analysis_pending' || retry.reason === 'analysis_error' || retry.reason === 'analysis_exhausted') {
-                  retry.nextRetryAt = nextAnalysisRetry(config, retry.image, now);
-                  if (retry.analysisAttempts! >= config.maxAnalysisAttempts) {
-                    retry.autoRetry = false; retry.reason = 'analysis_exhausted';
-                  }
-                } else retry.nextRetryAt = new Date(now + Math.min(3600000, 60000 * 2 ** Math.min(6, retry.attempts - 1))).toISOString();
-                await pending.write(key, retry);
-                await retireIdentity(store, key, retry, now);
-              }
-            }
-          }
-        }
+        await collectPage(batch.images, scope, seen, page);
         if (page * batch.size >= total) break;
       }
     } catch (error) {

apps/pac-service/src/integrations/meyer/config.example.json

@@ -5,7 +5,7 @@
   "timezone": "Asia/Shanghai",
   "timeStart": "2026-10-01 00:00:00",
   "timeEnd": "2026-10-16 23:59:59",
-  "pageSize": 100,
+  "pageSize": 200,
   "maxPages": 1000,
   "timeoutMs": 10000,
   "requestIntervalMs": 500,

apps/pac-service/src/integrations/meyer/contracts.ts

@@ -32,7 +32,7 @@ const ConfigFields = z.object({
   scopes: z.array(ScopeSchema).min(1).max(100),
   timeStart: sourceDate,
   timeEnd: sourceDate,
-  pageSize: z.number().int().min(1).max(100).default(100),
+  pageSize: z.number().int().min(1).max(200).default(200),
   maxPages: z.number().int().min(1).max(10000).default(1000),
   timeoutMs: z.number().int().min(100).max(60000).default(10000),
   roundTimeoutMs: z.number().int().min(1000).max(86400000).default(600000),
@@ -98,8 +98,12 @@ export interface MeyerSink { apply(bundle: MeyerBundle): Promise<void> }
 export interface MeyerSource {
   list(scope: MeyerScope, page: number): Promise<{ total: number; page: number; size: number; images: unknown[] }>;
   symptom(image: MeyerImage): Promise<unknown>;
+  /** 批量按影像 ID 关联；缺项不等于阴性结果。适配旧本地来源时可只实现 symptom。 */
+  symptoms?(images: readonly MeyerImage[]): Promise<ReadonlyMap<string, unknown>>;
   /** 调度预算不足时延后分析，不能虚增实际分析请求次数。 */
   canRequestAnalysis?(): boolean;
+  /** 合批前重读身份时，为已核实影像预留一次分析请求，避免列表耗尽整轮额度。 */
+  remainingRequests?(): number;
 }
 export function hash(value: unknown): string {
   const stable = (v: unknown): unknown => Array.isArray(v) ? v.map(stable) :

apps/pac-service/src/integrations/meyer/scan.ts

@@ -26,7 +26,7 @@ export async function scanWindow(config: MeyerConfig, sourceFor: (c: MeyerConfig
   }
   if (progress.version !== 1 || !/^[a-f0-9-]{36}$/.test(progress.generation ?? '') || progress.start !== start || progress.end !== end ||
     !Number.isInteger(progress.scopeIndex) || progress.scopeIndex < 0 || progress.scopeIndex > config.scopes.length ||
-    !Number.isInteger(progress.pageSize) || progress.pageSize < 1 || progress.pageSize > 100 ||
+    !Number.isInteger(progress.pageSize) || progress.pageSize < 1 || progress.pageSize > 200 ||
     !Number.isInteger(progress.row) || progress.row < 0 || !Number.isInteger(progress.page) || progress.page < 1) {
     throw new Error('invalid scan checkpoint; do not discard it');
   }
@@ -88,23 +88,30 @@ export async function scanWindow(config: MeyerConfig, sourceFor: (c: MeyerConfig
       }
       for (const raw of page.images.slice(0, progress.row)) seen.add(ids(raw));
       while (progress.row < page.images.length && result.examined < config.maxNewImagesPerTick) {
-        const image = page.images[progress.row]; result.examined++;
-        if (seen.has(ids(image))) {
-          progress.generation = randomUUID(); progress.page = 1; progress.row = 0;
-          progress.pageData = undefined; progress.total = undefined; progress.firstHash = undefined;
-          await store.write('scan-progress', progress);
-          throw new Error('duplicate image across scan pages; restart fixed window');
+        const count = source.symptoms ? Math.min(page.images.length - progress.row,
+          config.maxNewImagesPerTick - result.examined) : 1;
+        const images = page.images.slice(progress.row, progress.row + count);
+        result.examined += images.length;
+        for (const image of images) {
+          if (seen.has(ids(image))) {
+            progress.generation = randomUUID(); progress.page = 1; progress.row = 0;
+            progress.pageData = undefined; progress.total = undefined; progress.firstHash = undefined;
+            await store.write('scan-progress', progress);
+            throw new Error('duplicate image across scan pages; restart fixed window');
+          }
+          seen.add(ids(image));
         }
-        seen.add(ids(image));
         const one = await collectMeyer({ ...window, scopes: [scope] }, {
-          list: async () => ({ total: 1, page: 1, size: window.pageSize, images: [image] }),
+          list: async () => ({ total: images.length, page: 1, size: window.pageSize, images }),
           canRequestAnalysis: () => source.canRequestAnalysis?.() ?? true,
           symptom: im => source.symptom(im),
+          ...(source.symptoms ? { symptoms: (items: Parameters<NonNullable<MeyerSource['symptoms']>>[0]) => source.symptoms!(items) } : {}),
         }, evidenceStore, codes, sink, true, now);
         for (const key of ['seen', 'collected', 'applied', 'alreadyApplied', 'deferred', 'failed', 'imagesApplied', 'analysisFailed'] as const) result[key] += one[key];
         result.errors.push(...one.errors);
         if (one.failed) return result; // 不跳过未知失败；重试从本行恢复。
-        progress.row++; await store.write('scan-progress', progress);
+        // 整组都已落库或留下独立待处理记录后才推进；中断时安全重放这组。
+        progress.row += images.length; await store.write('scan-progress', progress);
       }
       if (progress.row === page.images.length) {
         await ledger.write(`page-${progress.page}`, page.images.map(ids));

apps/pac-service/src/integrations/meyer/scheduler.ts

@@ -3,7 +3,7 @@ import { rm } from 'node:fs/promises';
 import { join } from 'node:path';
 import { collectMeyer, type PendingImage } from './collector';
 import { MeyerStore, type SavedBundle } from './store';
-import { imageKey, sourceTime, type MeyerConfig, type MeyerSource, type MeyerSink, hash, type MeyerScope } from './contracts';
+import { imageKey, sourceTime, ImageSchema, type MeyerConfig, type MeyerSource, type MeyerSink, hash, type MeyerScope } from './contracts';
 import { analysisCapability } from './retry-policy';
 import { clinicStateStore, prepareClinicStates } from './clinic-state';
 import { MeyerBudgetExhausted, MeyerTickBudget } from './tick-budget';
@@ -46,16 +46,30 @@ async function runClinicTick(config: MeyerConfig, sourceFor: (c: MeyerConfig) =>
   }
   due.sort((a, b) => a.value.nextRetryAt.localeCompare(b.value.nextRetryAt));
   let retryFailed = 0, retryAttempted = 0;
+  const verified: Array<{ key: string; value: PendingImage; config: MeyerConfig; image: unknown }> = [];
+  const retryFailure = async (key: string, value: PendingImage, error: unknown) => {
+    retryFailed++;
+    // 移出队列的人工记录不能复活，落盘后的请求次数也不能回退。
+    if (await identityReviewStore(store).read(key)) { retiredIdentity++; return; }
+    const retained = await pending.read<PendingImage>(key) ?? value;
+    const next = { ...retained, attempts: retained.attempts + 1,
+      ...(error instanceof MeyerIdentityError ? { identityIssue: error.message,
+        identityAttempts: retained.identityAttempts ?? 1 } : {}),
+      nextRetryAt: new Date(now + 3600000).toISOString() };
+    await pending.write(key, next);
+    if (await retireIdentity(store, key, next, now)) retiredIdentity++;
+  };
   for (const entry of due.slice(0, config.maxRetriesPerTick)) {
     const { key } = entry;
     let value = entry.value;
     if (!available()) break;
-    retryAttempted++;
     if (hash(value.scope) !== hash(config.scopes[0])) throw new Error('pending scope mismatch');
-    if (!value.image.imageCreateTime) { retryFailed++; continue; }
+    if (!value.image.imageCreateTime) { retryAttempted++; retryFailed++; continue; }
     const t = Date.parse(sourceTime(value.image.imageCreateTime));
     const retryConfig = { ...config, scopes: [value.scope], timeStart: chinaTime(t - 1000), timeEnd: chinaTime(t + 1000) };
     const client = sourceFor(retryConfig);
+    if (verified.length && client.symptoms && (client.remainingRequests?.() ?? Infinity) < 2) break;
+    retryAttempted++;
     // 每次重试重新读取影像身份，避免拿旧 patientId 给后来改绑的影像写事实。
     try {
       if (isIdentityPending(value)) {
@@ -67,6 +81,9 @@ async function runClinicTick(config: MeyerConfig, sourceFor: (c: MeyerConfig) =>
       let total: number | undefined;
       for (let page = 1; ; page++) {
         if (page > Math.min(config.maxPages, config.maxListPagesPerTick)) throw new Error('pending page limit');
+        if (verified.length && client.symptoms && (client.remainingRequests?.() ?? Infinity) < 2) {
+          throw new MeyerBudgetExhausted('Meyer request reserved for verified analysis batch');
+        }
         const batch = await client.list(value.scope, page);
         if (batch.page !== page || batch.size !== config.pageSize || (total !== undefined && total !== batch.total) ||
           batch.images.length !== Math.min(batch.size, Math.max(0, batch.total - (page - 1) * batch.size))) throw new Error('pending page changed');
@@ -80,6 +97,15 @@ async function runClinicTick(config: MeyerConfig, sourceFor: (c: MeyerConfig) =>
       if (value.image.patientId && (found as { patientId?: string }).patientId !== value.image.patientId) {
         throw new MeyerIdentityError('pending image ownership changed');
       }
+      if (client.symptoms) {
+        // 每张影像都先从源端重读身份，不能为合批放宽原来的归属和时间校验。
+        const image = ImageSchema.parse(found);
+        if (image.clinicId !== value.scope.clinicId) throw new Error('pending image clinic mismatch');
+        if (image.imageCreateTime && (sourceTime(image.imageCreateTime) < sourceTime(retryConfig.timeStart) ||
+          sourceTime(image.imageCreateTime) > sourceTime(retryConfig.timeEnd))) throw new Error('pending image outside requested window');
+        verified.push({ key, value, config: retryConfig, image });
+        continue;
+      }
       const result = await collectMeyer(retryConfig, {
         list: async () => ({ total: 1, page: 1, size: config.pageSize, images: [found] }),
         canRequestAnalysis: () => client.canRequestAnalysis?.() ?? true,
@@ -89,17 +115,28 @@ async function runClinicTick(config: MeyerConfig, sourceFor: (c: MeyerConfig) =>
       if (await identityReviewStore(store).read(key)) retiredIdentity++;
     } catch (error) {
       if (error instanceof MeyerBudgetExhausted) break;
-      retryFailed++;
-      // 条目可能已移出队列，随后写摘要或 sync 失败；不能用旧值把人工记录重新激活。
-      if (await identityReviewStore(store).read(key)) { retiredIdentity++; continue; }
-      // collector 可能已持久化请求次数后才在写运行摘要时失败；不能回写旧次数。
-      const retained = await pending.read<PendingImage>(key) ?? value;
-      const next = { ...retained, attempts: retained.attempts + 1,
-        ...(error instanceof MeyerIdentityError ? { identityIssue: error.message,
-          identityAttempts: retained.identityAttempts ?? 1 } : {}),
-        nextRetryAt: new Date(now + 3600000).toISOString() };
-      await pending.write(key, next);
-      if (await retireIdentity(store, key, next, now)) retiredIdentity++;
+      await retryFailure(key, value, error);
+    }
+  }
+  // 只合并同一诊所、本轮已重新核实的条目；整轮重试名额与业务请求预算仍独立生效。
+  for (let offset = 0; offset < verified.length; offset += config.pageSize) {
+    const items = verified.slice(offset, offset + config.pageSize);
+    const retryConfig = { ...config,
+      timeStart: items.map(item => item.config.timeStart).sort()[0]!,
+      timeEnd: items.map(item => item.config.timeEnd).sort().at(-1)! };
+    const client = sourceFor(retryConfig);
+    try {
+      const result = await collectMeyer(retryConfig, {
+        list: async () => ({ total: items.length, page: 1, size: config.pageSize, images: items.map(item => item.image) }),
+        canRequestAnalysis: () => client.canRequestAnalysis?.() ?? true,
+        symptom: image => client.symptom(image),
+        symptoms: images => client.symptoms!(images),
+      }, store, codes, sink, true, now, true);
+      retryFailed += result.failed + result.analysisFailed;
+      for (const item of items) if (await identityReviewStore(store).read(item.key)) retiredIdentity++;
+    } catch (error) {
+      if (error instanceof MeyerBudgetExhausted) break;
+      for (const item of items) await retryFailure(item.key, item.value, error);
     }
   }
   let scan: Awaited<ReturnType<typeof scanWindow>> | undefined;

apps/pac-service/src/integrations/meyer/tick-budget.ts

@@ -11,6 +11,7 @@ export class MeyerTickBudget {
   source(factory: (config: MeyerConfig) => MeyerSource, allowance: number) {
     let used = 0;
     const available = () => used < allowance && this.available();
+    const remaining = () => Date.now() >= this.deadline ? 0 : Math.max(0, Math.min(allowance - used, this.limit - this.requests));
     const spend = () => {
       if (!available()) throw new MeyerBudgetExhausted('Meyer tick budget exhausted');
       used++; this.requests++;
@@ -20,7 +21,11 @@ export class MeyerTickBudget {
       return {
         list: async (scope, page) => { spend(); return source.list(scope, page); },
         canRequestAnalysis: available,
+        remainingRequests: remaining,
         symptom: async image => { spend(); return source.symptom(image); },
+        ...(source.symptoms ? { symptoms: async (images: Parameters<NonNullable<MeyerSource['symptoms']>>[0]) => {
+          spend(); return source.symptoms!(images);
+        } } : {}),
       };
     } };
   }

apps/pac-service/tests/meyer-batch.spec.ts

+import { createServer, type Server } from 'node:http';
+import { mkdtemp, rm } from 'node:fs/promises';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import { MeyerClient } from '../src/integrations/meyer/client';
+import { ConfigSchema, imageKey, type MeyerBundle, type MeyerConfig, type MeyerImage, type MeyerSource } from '../src/integrations/meyer/contracts';
+import { collectMeyer, type PendingImage } from '../src/integrations/meyer/collector';
+import { runMeyerTick } from '../src/integrations/meyer/scheduler';
+import { clinicStateStore } from '../src/integrations/meyer/clinic-state';
+import { MeyerStore } from '../src/integrations/meyer/store';
+import { codes, meyerConfig, meyerImage, meyerSymptom } from './meyer-imaging-fixtures';
+
+const empty = { teeth: [], diagnosis: '', description: '' };
+const rows = (count: number) => Array.from({ length: count }, (_, n) => ({ ...meyerImage(), imageId: `image-${n}` }));
+
+describe('Meyer 200-row and batch HTTP contract', () => {
+  let server: Server, config: MeyerConfig;
+  let handler: (path: string, body: any, bearer?: string) => Promise<{ code: number; data?: unknown }>;
+  let requests: Array<{ path: string; body: any; bearer?: string }>;
+  beforeEach(async () => {
+    requests = [];
+    server = createServer(async (req, res) => {
+      const chunks: Buffer[] = []; for await (const chunk of req) chunks.push(Buffer.from(chunk));
+      const call = { path: req.url!, body: JSON.parse(Buffer.concat(chunks).toString()), bearer: req.headers.authorization };
+      requests.push(call);
+      try { res.setHeader('Content-Type', 'application/json'); res.end(JSON.stringify(await handler(call.path, call.body, call.bearer))); }
+      catch { res.destroy(); }
+    });
+    await new Promise<void>(r => server.listen(0, '127.0.0.1', r));
+    config = { ...meyerConfig(), pageSize: 200, baseUrl: `http://127.0.0.1:${(server.address() as { port: number }).port}/api/v1` };
+  });
+  afterEach(async () => { server.closeAllConnections(); await new Promise<void>(r => server.close(() => r())); });
+  const token = (value = 'new', refresh = 'refresh-new') => ({ code: 200, data: { accessToken: value,
+    accessExpiresAt: new Date(Date.now() + 7200000).toISOString(), refreshToken: refresh,
+    refreshExpiresAt: new Date(Date.now() + 7 * 86400000).toISOString() } });
+  const page = { code: 200, data: { total: 0, page: 1, size: 200, images: [] } };
+  const client = () => new MeyerClient(config, { appKey: 'local-key', secret: 'local-secret' });
+
+  test('200 images and ID-only batch payload, with response association by ID and omitted items', async () => {
+    const images = rows(200);
+    handler = async path => path.endsWith('/tokens/pub') ? token() : path.endsWith('/query-by-clinic')
+      ? { code: 200, data: { total: 200, page: 1, size: 200, images } }
+      : { code: 200, data: images.slice(0, 199).reverse().map(im => ({ imageId: im.imageId, ...meyerSymptom() })) };
+    const c = client();
+    expect((await c.list(config.scopes[0]!, 1)).images).toHaveLength(200);
+    const result = await c.symptoms(images);
+    expect(result.size).toBe(199); expect(result.has('image-199')).toBe(false);
+    expect(result.get('image-0')).toEqual(meyerSymptom());
+    expect(requests.at(-1)!.body).toEqual({ imageIds: images.map(im => im.imageId) });
+    expect(requests.filter(r => r.path.endsWith('/tokens/pub'))).toHaveLength(1);
+    expect(requests.some(r => r.path.endsWith('/symptom'))).toBe(false);
+    await expect(c.symptoms(rows(201))).rejects.toThrow();
+    await expect(c.symptoms([{ ...images[0]!, clinicId: 'foreign' }])).rejects.toThrow('outside');
+    expect(requests).toHaveLength(3);
+  });
+
+  test.each(['duplicate', 'foreign'])('rejects %s image IDs instead of attaching findings to a requested patient', async mode => {
+    handler = async path => path.endsWith('/tokens/pub') ? token() : { code: 200,
+      data: mode === 'duplicate' ? [{ imageId: 'image-1', ...empty }, { imageId: 'image-1', ...empty }]
+        : [{ imageId: 'not-requested', ...empty }] };
+    await expect(client().symptoms([meyerImage()])).rejects.toThrow('ambiguous');
+  });
+
+  test('concurrent initial calls and window clients share one authentication request', async () => {
+    handler = async path => { if (path.endsWith('/tokens/pub')) { await new Promise(r => setTimeout(r, 10)); return token(); } return page; };
+    const c = client();
+    await Promise.all([c, c.forWindow(config), c.forWindow(config)].map(x => x.list(config.scopes[0]!, 1)));
+    expect(requests.filter(r => r.path.endsWith('/tokens/pub'))).toHaveLength(1);
+  });
+
+  test('concurrent HTTP-200 / 401001 responses refresh once, rotate the refresh token, and retry once', async () => {
+    const session = { token: { value: 'old', expires: Date.now() + 3600000, refresh: 'refresh-old', refreshExpires: Date.now() + 86400000 } };
+    handler = async (path, _, bearer) => path.endsWith('/refreshes/pub') ? token()
+      : bearer === 'Bearer old' ? { code: 401001 } : page;
+    const c = new MeyerClient(config, { appKey: 'key', secret: 'secret' }, session);
+    await Promise.all([c, c.forWindow(config), c.forWindow(config)].map(x => x.list(config.scopes[0]!, 1)));
+    const refresh = requests.filter(r => r.path.endsWith('/refreshes/pub'));
+    expect(refresh).toHaveLength(1); expect(refresh[0]!.body).toEqual({ refreshToken: 'refresh-old' });
+    expect(requests.filter(r => r.path.endsWith('/tokens/pub'))).toHaveLength(0);
+    expect(session.token.refresh).toBe('refresh-new');
+  });
+
+  test('near-expiry refresh network uncertainty discards the one-time token, including when fallback issue fails', async () => {
+    const session = { token: { value: 'old', expires: Date.now() + 20000, refresh: 'refresh-old', refreshExpires: Date.now() + 86400000 } };
+    let issues = 0;
+    handler = async path => {
+      if (path.endsWith('/refreshes/pub')) throw Error('socket reset after refresh may have succeeded');
+      if (path.endsWith('/tokens/pub')) return ++issues === 1 ? { code: 503001 } : token();
+      return page;
+    };
+    const c = new MeyerClient(config, { appKey: 'key', secret: 'secret' }, session);
+    await expect(c.list(config.scopes[0]!, 1)).rejects.toThrow('503001');
+    await expect(c.list(config.scopes[0]!, 1)).resolves.toMatchObject({ total: 0 });
+    expect(requests.filter(r => r.path.endsWith('/refreshes/pub'))).toHaveLength(1);
+    expect(issues).toBe(2);
+  });
+
+  test('repeated authentication rejection has a strict single retry limit', async () => {
+    handler = async path => path.endsWith('/tokens/pub') || path.endsWith('/refreshes/pub') ? token() : { code: 401001 };
+    await expect(client().list(config.scopes[0]!, 1)).rejects.toThrow('code=401001');
+    expect(requests.filter(r => r.path.endsWith('/query-by-clinic'))).toHaveLength(2);
+  });
+});
+
+describe('Meyer batch collection preserves per-image persistence and retry boundaries', () => {
+  let dir: string, store: MeyerStore;
+  const now = Date.parse('2026-10-10T03:00:00Z');
+  beforeEach(async () => { dir = await mkdtemp(join(tmpdir(), 'meyer-batch-')); store = new MeyerStore(dir); });
+  afterEach(async () => { await rm(dir, { recursive: true, force: true }); });
+
+  test('all images persist before one batch; missing, empty, and malformed results remain independent pending entries', async () => {
+    const c = { ...meyerConfig(), pageSize: 200 }, images = rows(200), written: MeyerBundle[] = [];
+    const pending = new MeyerStore(join(dir, 'pending'));
+    const batch = jest.fn(async (items: readonly MeyerImage[]) => {
+      expect(written.filter(b => b.phase === 'image')).toHaveLength(200);
+      for (const image of items) expect(await pending.read(imageKey(c.scopes[0]!, image.imageId))).toMatchObject({ analysisAttempts: 1 });
+      return new Map(items.slice(0, 199).map(im => [im.imageId,
+        im.imageId === 'image-197' ? empty : im.imageId === 'image-198' ? { teeth: null } : meyerSymptom()]));
+    });
+    const source: MeyerSource = { list: async () => ({ total: images.length, page: 1, size: 200, images }),
+      symptom: async () => { throw Error('single calls forbidden'); }, symptoms: batch };
+    const result = await collectMeyer(c, source, store, codes, { apply: async b => { written.push(b); } }, true, now);
+    expect(batch).toHaveBeenCalledTimes(1); expect(result.imagesApplied).toBe(200);
+    expect(result.applied).toBe(197); expect(result.analysisFailed).toBe(2); expect(result.failed).toBe(0);
+    expect(await pending.keys()).toHaveLength(3);
+    for (const image of images.slice(197)) {
+      expect(await pending.read(imageKey(c.scopes[0]!, image.imageId))).toMatchObject({ analysisAttempts: 1, autoRetry: true });
+      expect(await store.read(`bundle-${imageKey(c.scopes[0]!, image.imageId)}`)).toBeNull();
+    }
+  }, 60000);
+
+  test('global request quota counts a batch once rather than once per image', async () => {
+    const c = { ...meyerConfig(), pageSize: 200, maxSourceRequestsPerTick: 2, maxNewImagesPerTick: 200,
+      timeStart: '2026-10-10 09:00:00' }, images = rows(3);
+    const batch = jest.fn(async (items: readonly MeyerImage[]) => new Map(items.map(im => [im.imageId, empty])));
+    const source: MeyerSource = { list: async () => ({ total: 3, page: 1, size: 200, images }),
+      symptom: async () => { throw Error('single calls forbidden'); }, symptoms: batch };
+    const result = await runMeyerTick(c, () => source, store, codes, { apply: async () => {} }, now);
+    expect(result.sourceRequests).toBe(2); expect(batch).toHaveBeenCalledTimes(1);
+    const pending = new MeyerStore(join(dir, 'pending'));
+    for (const image of images) expect(await pending.read(imageKey(c.scopes[0]!, image.imageId))).toMatchObject({ analysisAttempts: 1 });
+  });
+
+  test('no analysis budget retains all images without consuming attempts or requesting a batch', async () => {
+    const c = meyerConfig(), images = rows(3), batch = jest.fn(), written: MeyerBundle[] = [];
+    const result = await collectMeyer(c, { list: async () => ({ total: 3, page: 1, size: c.pageSize, images }),
+      canRequestAnalysis: () => false, symptom: batch, symptoms: batch }, store, codes,
+    { apply: async b => { written.push(b); } }, true, now);
+    expect(result).toMatchObject({ imagesApplied: 3, failed: 0, analysisFailed: 0, deferred: 3 });
+    expect(batch).not.toHaveBeenCalled(); expect(written).toHaveLength(3);
+    const pending = new MeyerStore(join(dir, 'pending'));
+    for (const image of images) expect(await pending.read(imageKey(c.scopes[0]!, image.imageId))).toMatchObject({ analysisAttempts: 0, autoRetry: true });
+  });
+
+  test('a failed image write retains the batch checkpoint; recovery cannot duplicate completed analysis', async () => {
+    const c = { ...meyerConfig(), timeStart: '2026-10-10 09:00:00' }, images = rows(3), analyzed: string[] = [];
+    let fail = true;
+    const source: MeyerSource = { list: async () => ({ total: 3, page: 1, size: c.pageSize, images }),
+      symptom: async () => { throw Error('single calls forbidden'); },
+      symptoms: async items => { analyzed.push(...items.map(im => im.imageId)); return new Map(items.map(im => [im.imageId, meyerSymptom()])); } };
+    const sink = { apply: async (bundle: MeyerBundle) => {
+      if (bundle.phase === 'image' && bundle.raw.image.imageId === 'image-1' && fail) throw Error('database unavailable');
+    } };
+    const first = await runMeyerTick(c, () => source, store, codes, sink, now);
+    expect(first.scan.failed).toBe(1);
+    expect(await clinicStateStore(store, c.scopes[0]!).read('scan-progress')).toMatchObject({ row: 0 });
+    fail = false;
+    const recovered = await runMeyerTick(c, () => source, store, codes, sink, now + c.pollIntervalMs);
+    expect(recovered.scan.complete).toBe(true); expect(recovered.scan.failed).toBe(0);
+    expect(analyzed.sort()).toEqual(images.map(im => im.imageId).sort());
+  });
+
+  test('due retries re-read ownership and combine only eligible images into one batch', async () => {
+    const c = { ...meyerConfig(), pageSize: 200, timeStart: '2026-10-10 09:00:00' }, images = rows(4);
+    const pending = new MeyerStore(join(dir, 'pending'));
+    for (const image of images) await pending.write(imageKey(c.scopes[0]!, image.imageId), { scope: c.scopes[0], image,
+      attempts: 1, analysisAttempts: 1, reason: 'analysis_pending', nextRetryAt: '2000-01-01T00:00:00Z' });
+    await clinicStateStore(store, c.scopes[0]!).write('cursor', { through: '2026-10-10 11:00:00' });
+    const batch = jest.fn(async (items: readonly MeyerImage[]) => new Map(items.map(im => [im.imageId, meyerSymptom()])));
+    const source: MeyerSource = { list: async () => ({ total: 4, page: 1, size: 200,
+      images: images.map((im, n) => n === 3 ? { ...im, patientId: 'changed' } : im) }),
+      symptom: async () => { throw Error('single calls forbidden'); }, symptoms: batch };
+    const result = await runMeyerTick(c, () => source, store, codes, { apply: async () => {} }, now);
+    expect(result.retryAttempted).toBe(4); expect(result.retryFailed).toBe(1); expect(batch).toHaveBeenCalledTimes(1);
+    expect(batch.mock.calls[0]![0].map(im => im.imageId).sort()).toEqual(['image-0', 'image-1', 'image-2']);
+    expect(await pending.keys()).toHaveLength(1);
+  });
+
+  test('a two-request quota reserves analysis capacity and cannot starve all retries with repeated identity lists', async () => {
+    const c = { ...meyerConfig(), maxSourceRequestsPerTick: 2, timeStart: '2026-10-10 09:00:00' }, images = rows(3);
+    const pending = new MeyerStore(join(dir, 'pending'));
+    for (const image of images) await pending.write(imageKey(c.scopes[0]!, image.imageId), { scope: c.scopes[0], image,
+      attempts: 1, analysisAttempts: 0, reason: 'analysis_pending', nextRetryAt: '2000-01-01T00:00:00Z' });
+    await clinicStateStore(store, c.scopes[0]!).write('cursor', { through: '2026-10-10 11:00:00' });
+    const analyzed: string[] = [];
+    const source: MeyerSource = { list: async () => ({ total: 3, page: 1, size: c.pageSize, images }),
+      symptom: async () => { throw Error('single calls forbidden'); },
+      symptoms: async items => { analyzed.push(...items.map(im => im.imageId)); return new Map(items.map(im => [im.imageId, empty])); } };
+    for (let n = 0; n < 3; n++) {
+      const result = await runMeyerTick(c, () => source, store, codes, { apply: async () => {} }, now + n * 1000);
+      expect(result.sourceRequests).toBe(2); expect(result.retryAttempted).toBe(1);
+    }
+    expect(analyzed.sort()).toEqual(images.map(im => im.imageId).sort());
+    for (const image of images) expect(await pending.read(imageKey(c.scopes[0]!, image.imageId))).toMatchObject({ analysisAttempts: 1 });
+  });
+
+  test('switching from 100 to 200 rows keeps a live checkpoint and never loses or repeats analysis', async () => {
+    const c = { ...meyerConfig(), timeStart: '2026-10-10 09:00:00', maxNewImagesPerTick: 50 }, images = rows(201);
+    const analyzed: string[] = [], sizes: number[] = [];
+    const factory = (window: MeyerConfig): MeyerSource => ({
+      list: async (_, page) => { sizes.push(window.pageSize); return { total: images.length, page, size: window.pageSize,
+        images: images.slice((page - 1) * window.pageSize, page * window.pageSize) }; },
+      symptom: async () => { throw Error('single calls forbidden'); },
+      symptoms: async items => new Map(items.map(im => { analyzed.push(im.imageId); return [im.imageId, meyerSymptom()]; })),
+    });
+    await runMeyerTick(c, factory, store, codes, { apply: async () => {} }, now);
+    expect(await clinicStateStore(store, c.scopes[0]!).read('scan-progress')).toMatchObject({ pageSize: 100, row: 50 });
+    const result = await runMeyerTick({ ...c, pageSize: 200, maxNewImagesPerTick: 200 }, factory, store, codes, { apply: async () => {} }, now);
+    expect(result.scan.complete).toBe(true); expect(new Set(sizes)).toEqual(new Set([100]));
+    expect(analyzed).toHaveLength(201); expect(new Set(analyzed).size).toBe(201);
+    expect(ConfigSchema.parse({ ...c, pageSize: undefined }).pageSize).toBe(200);
+  }, 60000);
+});

apps/pac-service/tests/meyer-imaging-fixtures.ts

@@ -5,7 +5,7 @@ import { loadDiagnosisCodes, mapMeyer } from '../src/integrations/meyer/mapper';
 export const meyerConfig = (): MeyerConfig => ConfigSchema.parse({ baseUrl: 'https://example.com/api/v1',
   namespace: 'test', host: 'friday', timezone: 'Asia/Shanghai', scopes: [
     { clinicId: 'clinic-a', tenantId: 'friday-market', sourceUnit: 'brand-a' }],
-  timeStart: '2026-10-01 00:00:00', timeEnd: '2026-10-16 23:59:59', requestIntervalMs: 0 });
+  timeStart: '2026-10-01 00:00:00', timeEnd: '2026-10-16 23:59:59', pageSize: 100, requestIntervalMs: 0 });
 export const meyerImage = () => ImageSchema.parse({ imageId: 'image-1', clinicId: 'clinic-a', patientId: 'patient-1',
   imageCreateTime: '2026-10-10 10:00:00', studyTime: '2026-10-09 09:00:00', modality: 'pano',
   imageUrl: 'https://example.com/temporary?signature=abc', imageOssId: 'oss-1' });

apps/pac-service/tests/meyer-imaging-postgres.spec.ts

@@ -12,6 +12,7 @@ import { MeyerCliModule } from '../src/integrations/meyer/meyer-cli.module';
 import { MeyerPacSink } from '../src/integrations/meyer/pac-sink';
 import { ParserPipeline } from '../src/modules/sync/pipeline/parser-pipeline.service';
 import { collectMeyer } from '../src/integrations/meyer/collector';
+import { MeyerClient } from '../src/integrations/meyer/client';
 import { MeyerStore } from '../src/integrations/meyer/store';
 import { mapMeyerImage, mapMeyerAnalysis } from '../src/integrations/meyer/mapper';
 import { imageKey } from '../src/integrations/meyer/contracts';
@@ -70,6 +71,38 @@ describePg('Meyer | isolated PostgreSQL + actual minimal CLI module', () => {
       clinicId: 'clinic-a', action: 'visit_registered', subjectType: 'visit', subjectId: 'host-visit',
       rawPayload: {}, payloadHash: 'test', occurredAt: new Date() } });
   }
+  test('batch HTTP → canonical facts keeps images when individual results are empty or omitted, and replays safely', async () => {
+    const directory = await mkdtemp(join(tmpdir(), 'meyer-batch-pg-'));
+    const state = new MeyerStore(directory), images = [0, 1, 2].map(n => ({ ...meyerImage(), imageId: `batch-${n}` }));
+    let batches = 0;
+    const server = createServer(async (req, res) => {
+      for await (const _chunk of req) { /* 不记录凭据或患者原文。 */ }
+      const data = req.url?.endsWith('/tokens/pub') ? { accessToken: 'local', accessExpiresAt: new Date(Date.now() + 3600000).toISOString() }
+        : req.url?.endsWith('/query-by-clinic') ? { total: 3, page: 1, size: 200, images }
+        : (batches++, [{ imageId: 'batch-1', teeth: [], diagnosis: '', description: '' },
+          { imageId: 'batch-0', ...meyerSymptom() }]);
+      res.setHeader('Content-Type', 'application/json'); res.end(JSON.stringify({ code: 200, data }));
+    });
+    try {
+      await new Promise<void>(r => server.listen(0, '127.0.0.1', r));
+      const c = { ...meyerConfig(), pageSize: 200, baseUrl: `http://127.0.0.1:${(server.address() as { port: number }).port}/api/v1` };
+      const client = new MeyerClient(c, { appKey: 'local', secret: 'local' });
+      const first = await collectMeyer(c, client, state, codes, sink, true);
+      expect(first).toMatchObject({ imagesApplied: 3, applied: 1, analysisFailed: 1, failed: 0 });
+      const facts = await prisma.patientFact.findMany({ where: { hostId, patientId, status: 'active' }, orderBy: { subjectId: 'asc' } });
+      expect(facts.filter(f => f.type === FactType.IMAGE_RECORD)).toHaveLength(3);
+      expect(facts.filter(f => f.type === FactType.DIAGNOSIS_RECORD)).toHaveLength(2);
+      const pending = new MeyerStore(join(directory, 'pending'));
+      expect(await pending.keys()).toHaveLength(2);
+      await collectMeyer(c, client, state, codes, sink, true);
+      expect(await prisma.patientFact.findMany({ where: { hostId, patientId, status: 'active' }, orderBy: { subjectId: 'asc' } })).toEqual(facts);
+      expect(batches).toBe(1);
+      expect(await prisma.followupPlan.count({ where: { hostId } })).toBe(0);
+    } finally {
+      server.closeAllConnections(); await new Promise<void>(r => server.close(() => r()));
+      await rm(directory, { recursive: true, force: true });
+    }
+  });
   test('identity recovery CLI verifies live binding and full PAC identity, then resumes the normal tick', async () => {
     await hostRecord();
     const directory = await mkdtemp(join(tmpdir(), 'meyer-identity-pg-'));
@@ -79,7 +112,7 @@ describePg('Meyer | isolated PostgreSQL + actual minimal CLI module', () => {
       for await (const _chunk of req) { /* 不记录凭据或请求原文。 */ }
       res.setHeader('content-type', 'application/json');
       const data = req.url?.includes('/tokens/pub') ? { accessToken: 'local-token', accessExpiresAt: new Date(Date.now() + 3600000).toISOString() }
-        : req.url?.endsWith('/query-by-clinic') ? { total: 1, page: 1, size: 100, images: [image] } : meyerSymptom();
+        : req.url?.endsWith('/query-by-clinic') ? { total: 1, page: 1, size: 100, images: [image] } : [{ imageId: 'image-1', ...meyerSymptom() }];
       res.end(JSON.stringify({ code: 200, data }));
     });
     const previousKey = process.env.MEYER_APP_KEY, previousSecret = process.env.MEYER_APP_SECRET;
@@ -350,7 +383,7 @@ describePg('Meyer | isolated PostgreSQL + actual minimal CLI module', () => {
         accessToken: 'test-token', accessExpiresAt: new Date(Date.now() + 600000).toISOString(),
       } : req.url!.endsWith('/query-by-clinic') ? {
         total: 1, page: 1, size: 100, images: [meyerImage()],
-      } : meyerSymptom();
+      } : [{ imageId: 'image-1', ...meyerSymptom() }];
       res.writeHead(200, { 'Content-Type': 'application/json' }); res.end(JSON.stringify({ code: 200, data }));
     });
     const previousKey = process.env.MEYER_APP_KEY; const previousSecret = process.env.MEYER_APP_SECRET;

apps/pac-service/tests/meyer-imaging.spec.ts

@@ -273,7 +273,7 @@ describe('Meyer actual HTTP contract', () => {
   });
   afterEach(async () => { server.closeAllConnections(); await new Promise<void>(r => server.close(() => r())); });
   const auth = () => ({ code: 200, data: { accessToken: 'test-token', accessExpiresAt: new Date(Date.now() + 600000).toISOString() } });
-  test('authentication, complete patient ownership request, and HTTP-200 business errors', async () => {
+  test('authentication, image ID batch request, and HTTP-200 business errors', async () => {
     const calls: string[] = [];
     handler = (url, body, token) => {
       calls.push(url);
@@ -286,7 +286,8 @@ describe('Meyer actual HTTP contract', () => {
         expect(body).toMatchObject({ clinicId: 'clinic-a', page: 1, size: 100, timeStart: config.timeStart });
         return { body: { code: 200, data: { total: 1, page: 1, size: 100, images: [meyerImage()] } } };
       }
-      expect(body).toEqual({ imageClinicId: 'clinic-a', patientId: 'patient-1', imageId: 'image-1' });
+      expect(url.endsWith('/symptom-batch')).toBe(true);
+      expect(body).toEqual({ imageIds: ['image-1'] });
       return { body: { code: 500, msg: 'private patient info' } };
     };
     const client = new MeyerClient(config, { appKey: 'key', secret: 'secret' });
@@ -333,7 +334,7 @@ describe('Meyer actual HTTP contract', () => {
     const stdout = jest.spyOn(console, 'log').mockImplementation(() => {});
     try {
       handler = url => ({ body: url.endsWith('/tokens/pub') ? auth() : { code: 200,
-        data: url.endsWith('/query-by-clinic') ? { total: 1, page: 1, size: 100, images: [meyerImage()] } : meyerSymptom() } });
+        data: url.endsWith('/query-by-clinic') ? { total: 1, page: 1, size: 100, images: [meyerImage()] } : [{ imageId: 'image-1', ...meyerSymptom() }] } });
       const { writeFile } = await import('node:fs/promises');
       await writeFile(join(dir, 'config.json'), JSON.stringify((({ scopes, ...rest }) => ({ ...rest, clinicIds: scopes.map(s => s.clinicId) }))(config)));
       expect(await runMeyerCli([`--config=${join(dir, 'config.json')}`, `--state-dir=${join(dir, 'state')}`])).toBe(0);

docs/qa/2026-10-21-meyer-batch-contract.md

+# 美亚 v0.4：200 条分页、批量分析与授权更新
+
+## 范围
+
+- 分支：`codex/meyer-clinic-isolation`，基线 `ff1c039`；本轮修改未提交、未合并、未部署 PAC。
+- 依据：2026-10-21 提供的《PACS_瑞尔接口对接文档.md》v0.4 与 Java SDK；文件指纹见 `output/meyer-batch-20261021/vendor-inputs.json`。
+- PAC 保持直接 HTTP 对接，不引入 JAR。按本轮确认不接入影像事件推送，沿用单一定时器。
+
+## 改动
+
+- `query-by-clinic` 的配置默认值及响应校验上限改为 200；现存 100 条分页断点完成后才切换新大小，不删除游标或留底。
+- 新扫描和到期补查使用 `symptom-batch`，只传 `imageIds`，每批最多 200。同一诊所先完成影像身份校验和影像保存，再合并分析请求。
+- 结果按影像 ID 对应。省略项、空分析、单项格式错误分别留待补查；额外或重复 ID 拒绝整批。分析次数与补查时间仍逐张持久化，已应用快照不重复拉取。
+- 保留原诊所隔离、整轮额度、分析降频/上限和身份异常最多 8 次补查。身份重读阶段为已核实影像预留一次分析请求，避免小额度只反复查列表。
+- access token 按返回的到期时间提前 30 秒更新。共享同一客户端的并发换取；优先刷新，丢弃结果不确定的一次性 refresh token 后回退应用凭据签发。HTTP 401、业务码 401/401001 最多触发一次重认证与重试。
+- 原标准字段、临床映射、时区、FactWriter 和画像/召回判断未调整。文件 URL 接口上限仍为 100，不随影像列表和分析扩到 200。
+
+## 美亚测试接口实测
+
+2026-10-21 14:54（北京），使用当前测试配置访问 `https://re-pacs.myyun.com/api/v1`，仅查询及签发/刷新测试会话，没有写入 PAC 或修改美亚业务数据。
+
+| 项目 | 结果 |
+|---|---|
+| 列表请求每页 200 | 返回 `size=200`；本次诊所和时间范围共 58 张 |
+| 批量 58 张已有影像及 1 个不存在 ID | 返回 58 项，不存在 ID 被省略 |
+| 非空分析 | 本次返回 2 项；不据空视图认定阴性或分析已完成 |
+| 抽取 3 张全景片与单张接口比较 | 结果一致 |
+| 临近过期刷新 | refresh token 正常轮换；新 access token 剩余约 7200 秒 |
+
+脱敏结果：`output/meyer-batch-20261021/vendor-validation.json`。实测只代表当前测试配置，不推断生产是否共用同一美亚应用。
+
+## 本地验证
+
+- 新增批量/授权用例 14 项，覆盖 200 张合批、缺项、重复/额外 ID、空和异常结果、并发授权、401001、刷新网络异常、旧分页断点、写入恢复及极小请求额度。
+- 原有影像摄入、诊所隔离、身份恢复、补查上限和文件访问回归通过；诊断映射 40 项通过。
+- 独立本地 PostgreSQL 16 项通过：批量 HTTP → 归一 → 真实事实写入；影像先保存、晚到分析更新同一影像、空/缺项保留补查、重复执行事实不变。临时测试库已清理。
+- 后端源码及测试类型检查、构建、差异格式检查通过。
+
+测试日志与构建日志在 `output/meyer-batch-20261021/`。未修改测试/生产 PAC 的采集配置、定时器、业务数据或部署版本。