Skip to content

P
pac
Commit 7cdccafe by luoqi
Options

Merge bounded Meyer analysis retries into test

parents 01b9ae78 7fd8e892
Pipeline #3837 failed in 0 seconds
Showing with 226 additions and 19 deletions
apps/pac-service/src/cli/meyer-imaging.cli.ts
......@@ -98,7 +98,8 @@ export async function runMeyerCli(argv: string[]): Promise<number> {
}
if (args.tick) {
const status = await runMeyerTick(config, c => client.forWindow(c), store, codes, sink!);
console.log(JSON.stringify({ mode: 'tick', pending: status.pending, retryAttempted: status.retryAttempted,
console.log(JSON.stringify({ mode: 'tick', pending: status.pending, autoPending: status.autoPending,
manualReview: status.manualReview, retiredUnsupported: status.retiredUnsupported, retryAttempted: status.retryAttempted,
retryFailed: status.retryFailed, scanned: status.scan?.examined ?? 0,
scanComplete: status.scan?.complete ?? false, scanFailed: status.scan?.failed ?? 0,
imagesApplied: status.scan?.imagesApplied ?? 0, analysisFailed: status.scan?.analysisFailed ?? 0 }));
......
apps/pac-service/src/integrations/meyer/README.md
......@@ -44,7 +44,7 @@
美亚没有提供分析完成状态、分析时间和版本。本期不伪造 `completed/analyzedAt`，不绕过既有 image_analysis 的完整快照要求；将已观察到的牙位结论转换为现有 `image_study`、`diagnosis`，使用原有解析器和 FactWriter。**这是检查时刻的影像发现，不是医生诊断或美亚分析完成时刻。** 原文、观察时间与 `timeBasis=studyTime` 保存在适配器文件和事务 rawPayload 中。
已绑定患者且有入库时间即可保存影像，不以分析是否存在为前提。无分析时持久化待处理，之后重试补充到同一影像；分析接口异常也不影响已入库影像或后续影像扫描。没有入库时间或未绑定患者仍留底待处理，不伪造时间或患者归属。缺检查时间时可以保存影像内分析，但不生成带推测时间的诊断。综合诊断/描述不拆分为新诊断。临时 URL 不写入长期影像地址；原片沿用宿主患者级入口。本期不新增 UI；若宿主使用单张影像模板，须先确认其能识别此来源的影像标识。
已绑定患者且有入库时间即可保存影像，不以分析是否存在为前提。无分析时按下述策略补查，取得结果后补充到同一影像；分析接口异常也不影响已入库影像或后续影像扫描。没有入库时间或未绑定患者仍留底待处理，不伪造时间或患者归属。缺检查时间时可以保存影像内分析，但不生成带推测时间的诊断。综合诊断/描述不拆分为新诊断。临时 URL 不写入长期影像地址；原片沿用宿主患者级入口。本期不新增 UI；若宿主使用单张影像模板，须先确认其能识别此来源的影像标识。
## 运行
......@@ -90,11 +90,13 @@ pnpm meyer-imaging -- --config=/absolute/meyer.json --state-dir=/absolute/privat
- 新影像先经过原有标准解析/事实写入流程入库。没有分析、分析请求失败或分析写入失败时保留 pending/，允许影像扫描水位前进；分页、身份校验或影像落库失败时不推进。
- `imagesApplied` 记录本轮新写入影像阶段数，`applied` 记录完成分析阶段数；`failed` 是阻止影像扫描的失败，`analysisFailed` 单独记录分析请求/写入异常，两类均通过 CLI 失败退出码反映。空分析计入 deferred，不制造分析完成或阴性结论。
- 每轮先处理最早到期的最多 10 个（可配置）待处理项，按 1 分钟起指数退避、最多 1 小时；不会因重试次数多而自动丢弃。
- 每轮先处理最早到期的最多 10 个（可配置）待处理项。全景片从影像入库时刻起，前 15 分钟按正常轮次补查，之后每小时最多查一次。累计实际调用分析接口 8 次仍未取得结果，停止自动查询，转人工处理；记录和已入库影像保留。旧队列的混合重试次数不计入分析次数。
- 当前只支持全景片分析（aiType=3，兼容明确的全景片名称/类型）。其他类型只保存影像，不进入分析补查队列；类型不明时保留人工核实。分类记录写入 analysis-status，不能解释为分析阴性。升级时本地清理已落库且没有待写分析快照的旧非分析类型，不额外请求美亚、不占重试名额。
- 次数在请求前持久化，进程重启或重叠扫描不会重置。分析请求异常也计次数；已取得分析但写入失败的，使用原快照继续恢复写入，不受查询上限限制。身份校验、患者未绑定和影像写入失败另行退避，不消耗分析次数。
- 重试通过原影像入库时间前后 1 秒的窄窗重新定位 imageId，再获取分析。不会扫描整个诊所历史。身份改绑、影像不可见均保留待处理并报错，不自动撤销或迁移事实。
- 因接口没有按 imageId 直接查询元数据的能力，缺少入库时间的条目不能自动定位，须人工处理；这类情况会有 retryFailed。
- scheduler-status.json 提供扫描结果、水位、待处理数、重试失败数。watch 遇网络、美亚或数据库异常会记录失败计数并退避，默认从 15 分钟增长到最多 1 小时；依赖恢复后继续，不丢弃失败数据。配置错误、损坏状态或归属冲突需要修复后才能恢复。SIGINT/SIGTERM 会等当前一轮收尾再停止。
- 已成功分析的影像不再重新查询分析；只有影像阶段完成时仍会重试。这是首次接入及晚到分析重试，不是历史对账；删除、撤回和成功后的修改仍不覆盖。
- scheduler-status.json 提供扫描结果、水位、总待处理数 pending、自动补查数 autoPending、人工处理数 manualReview、当轮移出的不适用条数 retiredUnsupported 和重试失败数。watch 遇网络、美亚或数据库异常会记录失败计数并退避，默认从 15 分钟增长到最多 1 小时；依赖恢复后继续，不丢弃失败数据。配置错误、损坏状态或归属冲突需要修复后才能恢复。SIGINT/SIGTERM 会等当前一轮收尾再停止。
- 已成功分析的影像不再重新查询分析；只有影像阶段完成且符合补查策略时才自动重试。这是首次接入及晚到分析重试，不是历史对账；删除、撤回和成功后的修改仍不覆盖。
- 状态目录必须持久化。先前仅有 deferred-* 的旧目录没有完整队列，升级后先按原窗口重跑一次再启动 tick/watch。不要用曾做远期手工采集的目录直接推断历史已覆盖。
## 原图地址按需刷新
......@@ -117,11 +119,16 @@ pnpm meyer-imaging -- --config=/absolute/meyer.json --state-dir=/absolute/privat
| pollIntervalMs | 900000 | 一轮结束后等 15 分钟 |
| maxNewImagesPerTick | 100 | 所有配置诊所合计最多处理 100 条扫描记录；含重复/待处理项 |
| maxRetriesPerTick | 10 | 每轮最多重试 10 张待处理影像 |
| analysisSlowAfterMs | 900000 | 影像入库 15 分钟后转低频补查 |
| analysisSlowRetryMs | 3600000 | 低频补查间隔 1 小时，且不短于正常轮次间隔 |
| maxAnalysisAttempts | 8 | 每张影像最多自动查询分析 8 次，含首次查询和请求异常 |
| requestIntervalMs | 500 | 串行请求间隔至少 500ms |
| maxListPagesPerTick | 10 | 新影像扫描每轮最多查询 10 页（含续跑核对）；待处理项逐项也受此上限约束 |
| errorBackoffMaxMs | 3600000 | 连续异常最长等待 1 小时 |
| pageSize | 100 | 每次列表响应最多 100 张；不是每轮处理限额 |
达到分析次数上限后，人工确认需要补查时，可使用同一配置、同一持久状态目录，把固定窗口限定为对应影像入库时刻，再执行默认单次命令（`--apply`，不带 `--tick/--watch`）。这次显式查询可越过等待时间和次数上限；若仍为空或异常，继续保留人工状态。它不绕过影像类型、身份或落库校验，也不重算已完成分析。
`--tick/--watch` 使用上述每轮限额；默认固定窗口单次命令仍用于明确范围的手动采集，不受每轮限额控制。修改频率后重启 watch 生效；扫描限额会在每轮重新读取配置。所有调用仍由主进程串行执行，没有并发轮次。
scan-progress 保存固定窗口、诊所、页码、页内位置和当前页原文；scan-ledger 保存已完成页的影像 ID。达到限额不推进 cursor，下一轮从断点继续。写文件先 sync 再原子替换，并同步目录；记录事实成功后才移动行断点。若数据库成功而进程在记断点前退出，下次幂等重试，不重复生成事实。
......@@ -140,7 +147,7 @@ scan-progress 保存固定窗口、诊所、页码、页内位置和当前页原
### 存储分工
- **独立状态目录**：identity、数据库目标指纹、cursor、scan-progress、scan-ledger、pending、image、bundle、observed、scheduler-status、access 文件；本期不新增数据库采集表。状态目录须放持久磁盘、限制权限、备份。只支持单机单写者，不用于多机器共享抢任务。
- **独立状态目录**：identity、数据库目标指纹、cursor、scan-progress、scan-ledger、pending、analysis-status、image、bundle、observed、scheduler-status、access 文件；本期不新增数据库采集表。状态目录须放持久磁盘、限制权限、备份。只支持单机单写者，不用于多机器共享抢任务。
- **PAC 共用表**：标准化事件写 patient_transactions，标准影像/诊断写 patient_facts，按 Friday 宿主、租户、品牌与患者隔离；patients 只查找，不创建或移动。原始美亚依据还保留在事务 rawPayload。
- **原片**：仍在美亚/OSS，本地只留文件 ID 和业务证据，不复制文件。
......
apps/pac-service/src/integrations/meyer/collector.ts
......@@ -3,9 +3,13 @@ import { ImageSchema, SymptomSchema, hash, imageKey, sourceTime,
type MeyerConfig, type MeyerSink, type MeyerSource, type MeyerImage, type MeyerScope } from './contracts';
import { mapMeyerImage, mapMeyerAnalysis } from './mapper';
import { MeyerStore, type SavedBundle } from './store';
import { analysisCapability, nextAnalysisRetry } from './retry-policy';
export interface PendingImage {
scope: MeyerScope; image: MeyerImage; attempts: number; nextRetryAt: string;
/** 只计实际分析请求；旧 attempts 同时包含身份/落库失败，不能拿来作为分析次数。 */
analysisAttempts?: number; autoRetry?: boolean;
reason?: 'analysis_pending' | 'analysis_error' | 'analysis_exhausted' | 'analysis_type_unknown' | 'image_or_identity' | 'analysis_apply';
}
export interface CollectResult {
pages: number; seen: number; collected: number; applied: number; alreadyApplied: number;
......@@ -14,7 +18,7 @@ export interface CollectResult {
/** 固定时间窗单次运行；不推进主流程游标，不删除、不改绑、不复查已完成影像。 */
export async function collectMeyer(config: MeyerConfig, source: MeyerSource, store: MeyerStore,
codes: Record<string, string>, sink?: MeyerSink, respectRetryAt = false): Promise<CollectResult> {
codes: Record<string, string>, sink?: MeyerSink, respectRetryAt = false, now = Date.now()): Promise<CollectResult> {
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: [] };
......@@ -39,6 +43,7 @@ export async function collectMeyer(config: MeyerConfig, source: MeyerSource, sto
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');
......@@ -50,7 +55,8 @@ export async function collectMeyer(config: MeyerConfig, source: MeyerSource, sto
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);
retry = { scope, image, attempts: (previousRetry?.attempts ?? 0) + 1, nextRetryAt: '' };
retry = { scope, image, attempts: (previousRetry?.attempts ?? 0) + 1, nextRetryAt: '',
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]) {
......@@ -81,11 +87,27 @@ export async function collectMeyer(config: MeyerConfig, source: MeyerSource, sto
result.imagesApplied++;
}
imageReady = true;
if (respectRetryAt && previousRetry && Date.parse(previousRetry.nextRetryAt) > Date.now()) {
}
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();
......@@ -93,6 +115,12 @@ export async function collectMeyer(config: MeyerConfig, source: MeyerSource, sto
await store.write(`deferred-${key}`, { reason: 'unbound patient', observedAt, image });
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 });
......@@ -108,6 +136,7 @@ export async function collectMeyer(config: MeyerConfig, source: MeyerSource, sto
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);
......@@ -115,13 +144,19 @@ export async function collectMeyer(config: MeyerConfig, source: MeyerSource, sto
}
completed = !!sink;
} catch (error) {
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 {
retry.nextRetryAt = new Date(Date.now() + Math.min(3600000, 60000 * 2 ** Math.min(6, retry.attempts - 1))).toISOString();
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);
}
}
......
apps/pac-service/src/integrations/meyer/contracts.ts
......@@ -39,6 +39,9 @@ const ConfigFields = z.object({
pollIntervalMs: z.number().int().min(1000).max(86400000).default(900000),
maxNewImagesPerTick: z.number().int().min(1).max(10000).default(100),
maxRetriesPerTick: z.number().int().min(0).max(1000).default(10),
analysisSlowAfterMs: z.number().int().min(1000).max(86400000).default(900000),
analysisSlowRetryMs: z.number().int().min(1000).max(86400000).default(3600000),
maxAnalysisAttempts: z.number().int().min(1).max(1000).default(8),
maxListPagesPerTick: z.number().int().min(2).max(1000).default(10),
errorBackoffMaxMs: z.number().int().min(1000).max(86400000).default(3600000),
requestIntervalMs: z.number().int().min(0).max(60000).default(500),
......
apps/pac-service/src/integrations/meyer/retry-policy.ts 0 → 100644
import { sourceTime, type MeyerConfig, type MeyerImage } from './contracts';
/** 美亚当前仅开放全景片分析；类型不明不能等同于无病变。 */
export function analysisCapability(image: MeyerImage): 'supported' | 'unsupported' | 'unknown' {
if (image.imageKind && image.imageKind !== 'IMAGE') return 'unsupported';
if (image.aiType !== undefined && image.aiType !== null && image.aiType !== '') {
return String(image.aiType) === '3' ? 'supported' : 'unsupported';
}
if (image.aiTypeName === '全景片' || ['pano', 'OPG', 'PANORAMIC'].includes(image.modality ?? '')) return 'supported';
if (image.aiTypeName || ['CT', 'CBCT', 'MOUTH_MODEL', 'PDF', 'OFFICE'].includes(image.modality ?? '')) return 'unsupported';
// IMAGE_2D 可能包含全景片，不能仅凭文件形态判定具体检查类型。
return 'unknown';
}
export function nextAnalysisRetry(config: MeyerConfig, image: MeyerImage, now: number): string {
const created = image.imageCreateTime ? Date.parse(sourceTime(image.imageCreateTime)) : now;
const interval = now - created >= config.analysisSlowAfterMs
? Math.max(config.pollIntervalMs, config.analysisSlowRetryMs) : config.pollIntervalMs;
return new Date(now + interval).toISOString();
}
apps/pac-service/src/integrations/meyer/scan.ts
......@@ -17,7 +17,7 @@ function checkPage(batch: Page, page: number, size: number) {
}
/** 已读取页先落盘；逐行成功/已持久化待处理后才移动页内位置。 */
export async function scanWindow(config: MeyerConfig, sourceFor: (c: MeyerConfig) => MeyerSource,
store: MeyerStore, codes: Record<string, string>, sink: MeyerSink, start: string, end: string) {
store: MeyerStore, codes: Record<string, string>, sink: MeyerSink, start: string, end: string, now = Date.now()) {
let progress = await store.read<ScanProgress>('scan-progress');
if (!progress) {
progress = { version: 1, generation: randomUUID(), start, end, scopeIndex: 0, page: 1, row: 0, pageSize: config.pageSize };
......@@ -98,7 +98,7 @@ export async function scanWindow(config: MeyerConfig, sourceFor: (c: MeyerConfig
const one = await collectMeyer({ ...window, scopes: [scope] }, {
list: async () => ({ total: 1, page: 1, size: window.pageSize, images: [image] }),
symptom: im => source.symptom(im),
}, store, codes, sink, true);
}, store, 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; // 不跳过未知失败；重试从本行恢复。
......
apps/pac-service/src/integrations/meyer/scheduler.ts
......@@ -2,8 +2,9 @@ import { scanWindow, type ScanProgress } from './scan';
import { rm } from 'node:fs/promises';
import { join } from 'node:path';
import { collectMeyer, type PendingImage } from './collector';
import { MeyerStore } from './store';
import { sourceTime, type MeyerConfig, type MeyerSource, type MeyerSink } from './contracts';
import { MeyerStore, type SavedBundle } from './store';
import { imageKey, sourceTime, type MeyerConfig, type MeyerSource, type MeyerSink } from './contracts';
import { analysisCapability } from './retry-policy';
export const chinaTime = (ms: number) => new Date(ms + 8 * 3600000).toISOString().slice(0, 19).replace('T', ' ');
/** 一轮独立任务：先限量重试，再扫描一个窗口；只推进本适配器水位。 */
......@@ -17,9 +18,27 @@ export async function runMeyerTick(config: MeyerConfig, sourceFor: (c: MeyerConf
const end = Math.min(now - 60000, through + 3600000);
const pending = new MeyerStore(join(store.directory, 'pending'));
const due: Array<{ key: string; value: PendingImage }> = [];
let retiredUnsupported = 0;
for (const key of await pending.keys()) {
const value = await pending.read<PendingImage>(key);
if (value && Date.parse(value.nextRetryAt) <= now) due.push({ key, value });
if (!value) continue;
if (!config.scopes.some(s => JSON.stringify(s) === JSON.stringify(value.scope))) throw new Error('pending scope mismatch');
const saved = await store.read<SavedBundle>(`bundle-${key}`);
// 旧队列只清理已成功落库、且没有待写分析快照的非分析类型。
// 不再访问美亚，也不消耗每轮重试名额；归属和命名空间仍由 collector 校验。
if (key === imageKey(value.scope, value.image.imageId) && value.image.imageCreateTime &&
analysisCapability(value.image) !== 'supported' &&
(await store.read<SavedBundle>(`image-${key}`))?.applied && !saved) {
const result = await collectMeyer({ ...config, scopes: [value.scope],
timeStart: value.image.imageCreateTime, timeEnd: value.image.imageCreateTime }, {
list: async () => ({ total: 1, page: 1, size: config.pageSize, images: [value.image] }),
symptom: async () => { throw new Error('unsupported analysis must not be requested'); },
}, store, codes, sink, true, now);
if (result.failed || result.analysisFailed) throw new Error('pending classification failed; evidence retained');
if (analysisCapability(value.image) === 'unsupported') retiredUnsupported++;
continue;
}
if ((value.autoRetry !== false || saved) && Date.parse(value.nextRetryAt) <= now) due.push({ key, value });
}
due.sort((a, b) => a.value.nextRetryAt.localeCompare(b.value.nextRetryAt));
let retryFailed = 0;
......@@ -51,7 +70,7 @@ export async function runMeyerTick(config: MeyerConfig, sourceFor: (c: MeyerConf
const result = await collectMeyer(retryConfig, {
list: async () => ({ total: 1, page: 1, size: config.pageSize, images: [found] }),
symptom: image => client.symptom(image),
}, store, codes, sink);
}, store, codes, sink, true, now);
retryFailed += result.failed + result.analysisFailed;
} catch {
retryFailed++;
......@@ -70,15 +89,19 @@ export async function runMeyerTick(config: MeyerConfig, sourceFor: (c: MeyerConf
if (progress || end > through) {
const start = progress?.start ?? chinaTime(Math.max(initial, through - 60000));
const finish = progress?.end ?? chinaTime(end);
scan = await scanWindow(config, sourceFor, store, codes, sink, start, finish);
scan = await scanWindow(config, sourceFor, store, codes, sink, start, finish, now);
if (scan.complete && !scan.failed) {
await store.write('cursor', { through: finish });
await store.remove('scan-progress');
await rm(join(store.directory, 'scan-ledger'), { recursive: true, force: true });
}
}
let manualReview = 0;
const keys = await pending.keys();
for (const key of keys) if ((await pending.read<PendingImage>(key))?.autoRetry === false) manualReview++;
const status = { checkedAt: new Date(now).toISOString(), retryAttempted: Math.min(config.maxRetriesPerTick, due.length), retryFailed,
pending: (await pending.keys()).length, scan, cursor: await store.read('cursor') };
pending: keys.length, manualReview, autoPending: keys.length - manualReview, retiredUnsupported,
scan, cursor: await store.read('cursor') };
await store.write('scheduler-status', status);
return status;
}
apps/pac-service/tests/meyer-analysis-retry.spec.ts 0 → 100644
import { mkdtemp, rm } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { collectMeyer, type PendingImage } from '../src/integrations/meyer/collector';
import { imageKey, type MeyerSource } from '../src/integrations/meyer/contracts';
import { analysisCapability } from '../src/integrations/meyer/retry-policy';
import { runMeyerTick, chinaTime } from '../src/integrations/meyer/scheduler';
import { MeyerStore } from '../src/integrations/meyer/store';
import { meyerConfig, meyerImage, meyerSymptom, codes } from './meyer-imaging-fixtures';
describe('Meyer analysis retry policy', () => {
let dir: string;
beforeEach(async () => { dir = await mkdtemp(join(tmpdir(), 'meyer-analysis-retry-')); });
afterEach(async () => { await rm(dir, { recursive: true, force: true }); });
const created = Date.parse('2026-10-10T02:00:00Z');
const empty = { teeth: [], diagnosis: '', description: '' };
function setup(image = meyerImage()) {
const config = meyerConfig(), store = new MeyerStore(dir), pending = new MeyerStore(join(dir, 'pending'));
const key = imageKey(config.scopes[0]!, image.imageId);
const source: MeyerSource = { list: async () => ({ total: 1, page: 1, size: 100, images: [image] }),
symptom: jest.fn(async () => empty) };
const sink = { apply: jest.fn(async (_bundle: unknown) => {}) };
const run = (now: number, automatic = true) => collectMeyer(config, source, store, codes, sink, automatic, now);
return { config, store, pending, key, source, sink, run };
}
test('first 15 minutes use normal cadence; older empty results wait an hour, not the study date', async () => {
const s = setup();
await s.run(created + 5 * 60000);
expect(await s.pending.read(s.key)).toMatchObject({ analysisAttempts: 1,
nextRetryAt: new Date(created + 20 * 60000).toISOString(), autoRetry: true });
await s.run(created + 20 * 60000);
expect(await s.pending.read(s.key)).toMatchObject({ analysisAttempts: 2,
nextRetryAt: new Date(created + 80 * 60000).toISOString() });
await s.run(created + 35 * 60000);
expect(s.source.symptom).toHaveBeenCalledTimes(2);
expect(await s.pending.read(s.key)).toMatchObject({ analysisAttempts: 2,
nextRetryAt: new Date(created + 80 * 60000).toISOString() });
});
test('eight actual queries survive restarts; automatic scans cannot bypass exhaustion; manual retry can resolve', async () => {
const s = setup();
for (let n = 0; n < 8; n++) await s.run(created + (n + 1) * 3600000);
expect(await s.pending.read(s.key)).toMatchObject({ analysisAttempts: 8, autoRetry: false, reason: 'analysis_exhausted' });
await collectMeyer(s.config, s.source, new MeyerStore(dir), codes, s.sink, true, created + 24 * 3600000);
expect(s.source.symptom).toHaveBeenCalledTimes(8);
expect(await s.store.read(`bundle-${s.key}`)).toBeNull();
s.source.symptom = jest.fn(async () => meyerSymptom());
await s.run(created + 25 * 3600000, false);
expect(s.source.symptom).toHaveBeenCalledTimes(1);
expect(await s.pending.read(s.key)).toBeNull();
expect(await s.store.read(`bundle-${s.key}`)).toMatchObject({ applied: true });
});
test('request count is durable before HTTP; errors use the same frequency and cap', async () => {
const s = setup(); s.config.maxAnalysisAttempts = 1;
s.source.symptom = jest.fn(async () => {
expect(await s.pending.read(s.key)).toMatchObject({ analysisAttempts: 1 });
throw Error('upstream timeout');
});
expect((await s.run(created + 3600000)).analysisFailed).toBe(1);
expect(await s.pending.read(s.key)).toMatchObject({ autoRetry: false, reason: 'analysis_exhausted',
nextRetryAt: new Date(created + 2 * 3600000).toISOString() });
await s.run(created + 3 * 3600000);
expect(s.source.symptom).toHaveBeenCalledTimes(1);
});
test('saved analysis keeps retrying database writes beyond query cap, without refetching', async () => {
const s = setup(); s.config.maxAnalysisAttempts = 1;
s.source.symptom = jest.fn(async () => meyerSymptom());
let fail = true;
s.sink.apply = jest.fn(async (bundle: any) => { if (bundle.phase === 'analysis' && fail) throw Error('db outage'); });
await s.run(created + 3600000);
expect(await s.pending.read(s.key)).toMatchObject({ analysisAttempts: 1, autoRetry: true, reason: 'analysis_apply' });
// 即使旧状态仍带停止查询标记，已有分析快照也必须恢复写入。
await s.pending.write(s.key, { ...await s.pending.read<PendingImage>(s.key), autoRetry: false });
await s.store.write('cursor', { through: chinaTime(created + 2 * 3600000) });
fail = false;
const status = await runMeyerTick(s.config, () => s.source, s.store, codes, s.sink, created + 2 * 3600000);
expect(status.retryAttempted).toBe(1);
expect(s.source.symptom).toHaveBeenCalledTimes(1);
expect(await s.pending.read(s.key)).toBeNull();
});
test.each([4, 6, 23, 32])('type %s saves image without polling analysis or fabricating a negative result', async aiType => {
const s = setup({ ...meyerImage(), aiType });
await s.run(created + 3600000);
expect(s.sink.apply).toHaveBeenCalledTimes(1);
expect(s.source.symptom).not.toHaveBeenCalled();
expect(await s.store.read(`image-${s.key}`)).toMatchObject({ applied: true });
expect(await s.store.read(`bundle-${s.key}`)).toBeNull();
expect(await s.pending.read(s.key)).toBeNull();
expect(await s.store.read(`analysis-status-${s.key}`)).toMatchObject({ capability: 'unsupported' });
});
test('unknown type retains manual review; missing patient never consumes analysis attempts', async () => {
const s = setup({ ...meyerImage(), modality: null });
await s.run(created + 3600000);
expect(await s.pending.read(s.key)).toMatchObject({ autoRetry: false, reason: 'analysis_type_unknown', analysisAttempts: 0 });
const unbound = setup({ ...meyerImage(), imageId: 'unbound', patientId: null });
await unbound.run(created + 3600000);
expect(await unbound.pending.read(unbound.key)).toMatchObject({ reason: 'image_or_identity', analysisAttempts: 0 });
expect(unbound.source.symptom).not.toHaveBeenCalled();
expect(analysisCapability({ ...meyerImage(), imageKind: 'DOCUMENT' })).toBe('unsupported');
expect(analysisCapability({ ...meyerImage(), aiType: '3', modality: 'IMAGE_2D' })).toBe('supported');
expect(analysisCapability({ ...meyerImage(), modality: 'IMAGE_2D' })).toBe('unknown');
});
test('legacy non-analysis pending is retired locally, does not consume network budget; exhausted items stay visible', async () => {
const s = setup({ ...meyerImage(), aiType: 32 });
await s.run(created + 3600000);
await s.pending.write(s.key, { scope: s.config.scopes[0], image: { ...meyerImage(), aiType: 32 },
attempts: 50, nextRetryAt: new Date(created).toISOString() });
const manual: PendingImage = { scope: s.config.scopes[0]!, image: { ...meyerImage(), imageId: 'exhausted' },
attempts: 8, analysisAttempts: 8, autoRetry: false, reason: 'analysis_exhausted', nextRetryAt: new Date(created).toISOString() };
await s.pending.write(imageKey(manual.scope, manual.image.imageId), manual);
await s.store.write('cursor', { through: chinaTime(created + 2 * 3600000) });
const factory = jest.fn(() => s.source);
const status = await runMeyerTick(s.config, factory, s.store, codes, s.sink, created + 2 * 3600000);
expect(status).toMatchObject({ pending: 1, manualReview: 1, autoPending: 0, retiredUnsupported: 1, retryAttempted: 0 });
expect(factory).not.toHaveBeenCalled();
expect(s.source.symptom).not.toHaveBeenCalled();
expect(s.sink.apply).toHaveBeenCalledTimes(1);
});
});
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment