Skip to content

P
pac
Commit 1ebcfd66 by luoqi
Options

fix(meyer): ingest images before late analysis

parent 355a7550
Pipeline #3834 failed in 0 seconds
Showing with 250 additions and 42 deletions
apps/pac-service/src/cli/meyer-imaging.cli.ts
......@@ -74,7 +74,7 @@ export async function runMeyerCli(argv: string[]): Promise<number> {
const matches: Array<{ key: string; saved: SavedBundle }> = [];
for (const scope of config.scopes) {
const key = imageKey(scope, args.imageFiles);
const saved = await store.read<SavedBundle>(`bundle-${key}`);
const saved = await store.read<SavedBundle>(`bundle-${key}`) ?? await store.read<SavedBundle>(`image-${key}`);
if (saved) matches.push({ key, saved });
}
if (matches.length !== 1) throw new Error('requires exactly one collected image in configured scope');
......@@ -99,14 +99,16 @@ 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,
retryFailed: status.retryFailed, scanned: status.scan?.examined ?? 0, scanComplete: status.scan?.complete ?? false, scanFailed: status.scan?.failed ?? 0 }));
return status.retryFailed || status.scan?.failed ? 1 : status.pending ? 2 : 0;
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 }));
return status.retryFailed || status.scan?.failed || status.scan?.analysisFailed ? 1 : status.pending ? 2 : 0;
}
const result = await collectMeyer(config, client, store, codes, sink);
// 控制台只输出数量；原文、患者标识和逐条错误留在私有状态目录。
const { errors: _errors, ...counts } = result;
console.log(JSON.stringify({ mode: args.apply ? 'apply' : 'preview', ...counts }));
return result.failed ? 1 : result.deferred ? 2 : 0;
return result.failed || result.analysisFailed ? 1 : result.deferred ? 2 : 0;
} finally {
try { await app?.close(); } finally { await release(); }
}
......
apps/pac-service/src/integrations/meyer/collector.ts
import { join } from 'node:path';
import { ImageSchema, SymptomSchema, hash, imageKey, sourceTime,
type MeyerConfig, type MeyerSink, type MeyerSource, type MeyerImage, type MeyerScope } from './contracts';
import { mapMeyer } from './mapper';
import { mapMeyerImage, mapMeyerAnalysis } from './mapper';
import { MeyerStore, type SavedBundle } from './store';
export interface PendingImage {
......@@ -9,7 +9,7 @@ export interface PendingImage {
}
export interface CollectResult {
pages: number; seen: number; collected: number; applied: number; alreadyApplied: number;
deferred: number; failed: number; errors: Array<{ key: string; reason: string }>;
deferred: number; failed: number; imagesApplied: number; analysisFailed: number; errors: Array<{ key: string; reason: string }>;
}
/** 固定时间窗单次运行；不推进主流程游标，不删除、不改绑、不复查已完成影像。 */
......@@ -17,7 +17,7 @@ export async function collectMeyer(config: MeyerConfig, source: MeyerSource, sto
codes: Record<string, string>, sink?: MeyerSink, respectRetryAt = false): 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, errors: [] };
alreadyApplied: 0, deferred: 0, failed: 0, imagesApplied: 0, analysisFailed: 0, errors: [] };
for (const scope of config.scopes) {
const seen = new Set<string>();
let total: number | undefined;
......@@ -38,6 +38,7 @@ export async function collectMeyer(config: MeyerConfig, source: MeyerSource, sto
let key = `page-${page}-row-${result.seen}`;
let retry: PendingImage | undefined;
let completed = false;
let imageReady = false;
try {
const parsed = ImageSchema.safeParse(raw);
if (!parsed.success) throw new Error('invalid image contract');
......@@ -49,19 +50,42 @@ 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);
if (respectRetryAt && previousRetry && Date.parse(previousRetry.nextRetryAt) > Date.now()) {
result.deferred++; continue;
}
retry = { scope, image, attempts: (previousRetry?.attempts ?? 0) + 1, nextRetryAt: '' };
const saved = await store.read<SavedBundle>(`bundle-${key}`);
if (saved && (hash(saved.bundle.scope) !== hash(scope) || saved.bundle.namespace !== config.namespace ||
saved.bundle.raw.image.imageId !== image.imageId || saved.bundle.version !== 1)) {
throw new Error('saved bundle scope mismatch');
}
if (saved && saved.bundle.image.patientExternalId !== image.patientId) {
throw new Error('image ownership changed; reconciliation is not supported');
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 Error('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) {
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 (respectRetryAt && previousRetry && Date.parse(previousRetry.nextRetryAt) > Date.now()) {
// 不因扫描重遇同一行延长原重试时间。
retry = undefined; result.deferred++; continue;
}
}
let bundle = saved?.bundle;
if (!bundle) {
const observedAt = new Date().toISOString();
......@@ -74,10 +98,10 @@ export async function collectMeyer(config: MeyerConfig, source: MeyerSource, sto
await store.write(`observed-${key}`, { observedAt, image, symptom: rawSymptom });
const symptom = SymptomSchema.safeParse(rawSymptom);
if (!symptom.success) throw new Error('invalid symptom contract');
bundle = mapMeyer(config, scope, image, symptom.data, codes, observedAt) ?? undefined;
bundle = mapMeyerAnalysis(config, scope, image, symptom.data, codes, observedAt) ?? undefined;
if (!bundle) {
await store.write(`deferred-${key}`, { observedAt,
reason: 'no tooth findings or missing image/study time' });
reason: 'analysis not yet available' });
result.deferred++; continue;
}
// 写入前固定完整快照；失败重试使用同一快照，不混入后来变化的远端结果。
......@@ -91,7 +115,7 @@ export async function collectMeyer(config: MeyerConfig, source: MeyerSource, sto
}
completed = !!sink;
} catch (error) {
result.failed++;
if (imageReady) result.analysisFailed++; else result.failed++;
result.errors.push({ key, reason: error instanceof Error ? error.message : 'collection failed' });
} finally {
if (retry) {
......
apps/pac-service/src/integrations/meyer/contracts.ts
......@@ -77,15 +77,17 @@ export const SymptomSchema = z.object({
export type MeyerSymptom = z.infer<typeof SymptomSchema>;
export interface MeyerBundle {
version: 1;
version: 1 | 2;
/** v1 是历史完整快照；v2 分影像和分析两阶段，不改写旧留底。 */
phase?: 'image' | 'analysis';
namespace: string;
scope: MeyerScope;
observedAt: string;
// 分析状态/分析时刻未提供。诊断时间是影像检查时刻，绝不以采集时刻代替。
timeBasis: 'studyTime';
timeBasis: 'studyTime' | 'imageCreateTime';
image: ImageStudyCanonical;
diagnoses: DiagnosisCanonical[];
raw: { image: MeyerImage; symptom: MeyerSymptom };
raw: { image: MeyerImage; symptom: MeyerSymptom | null };
}
export interface MeyerSink { apply(bundle: MeyerBundle): Promise<void> }
export interface MeyerSource {
......
apps/pac-service/src/integrations/meyer/mapper.ts
......@@ -52,3 +52,39 @@ export function mapMeyer(config: MeyerConfig, scope: MeyerScope, image: MeyerIma
} }),
diagnoses, raw: { image, symptom } };
}
/** 影像元数据独立接入；没有分析或检查时间也不阻止保存影像。 */
export function mapMeyerImage(config: MeyerConfig, scope: MeyerScope, image: MeyerImage,
observedAt: string): MeyerBundle | null {
if (image.clinicId !== scope.clinicId) throw new Error('Meyer clinic differs from configured scope');
if (!image.patientId || !image.imageCreateTime) return null;
const uploadedAt = sourceTime(image.imageCreateTime);
if (uploadedAt > observedAt) throw new Error('Meyer image has future time');
return { version: 2, phase: 'image', namespace: config.namespace, scope, observedAt,
timeBasis: 'imageCreateTime', diagnoses: [], raw: { image, symptom: null },
image: ImageStudyCanonicalSchema.parse({
externalId: `meyer:${config.namespace}:${encodeURIComponent(image.imageId)}`,
patientExternalId: image.patientId, clinicId: scope.clinicId, uploadedAt,
modality: ['pa', 'bw', 'pano', 'cbct', 'intraoral_photo', 'scan', 'other'].includes(image.modality ?? '')
? image.modality : 'other', fileType: image.imageExt ?? null,
}) };
}
/** 晚到分析补到同一张影像；缺检查时间时只保存分析，不制造诊断时间。 */
export function mapMeyerAnalysis(config: MeyerConfig, scope: MeyerScope, image: MeyerImage,
symptom: MeyerSymptom, codes: Record<string, string>, observedAt: string): MeyerBundle | null {
const metadata = mapMeyerImage(config, scope, image, observedAt);
if (!metadata || (!symptom.teeth.length && !symptom.diagnosis.trim() && !symptom.description.trim())) return null;
// 有检查时刻和牙位时沿用原来的诊断映射，保持 v1 已入库事务哈希兼容。
const legacy = mapMeyer(config, scope, image, symptom, codes, observedAt);
if (legacy) return { ...legacy, version: 2, phase: 'analysis' };
if (image.studyTime && sourceTime(image.studyTime) > observedAt) throw new Error('Meyer image has future time');
const findings = [...new Map(symptom.teeth.map(t => [hash([t.position, t.content]), t])).entries()]
.sort(([a], [b]) => a.localeCompare(b));
return { ...metadata, phase: 'analysis', timeBasis: image.studyTime ? 'studyTime' : 'imageCreateTime',
image: ImageStudyCanonicalSchema.parse({ ...metadata.image, analysis: {
source: 'meyer', summary: symptom.diagnosis || null, description: symptom.description || null,
analyzedAt: null, findings: findings.map(([, t]) => ({ name: t.content,
toothPosition: t.position, code: resolveMeyerDiagnosisCode(t.content, codes) })),
} }), raw: { image, symptom } };
}
apps/pac-service/src/integrations/meyer/pac-sink.ts
......@@ -17,7 +17,11 @@ export class MeyerPacSink implements MeyerSink {
const { tenantId, sourceUnit, clinicId } = bundle.scope;
const image = ImageStudyCanonicalSchema.parse(bundle.image);
const diagnoses = bundle.diagnoses.map(d => DiagnosisCanonicalSchema.parse(d));
if (!diagnoses.length || diagnoses.some(d => d.patientExternalId !== image.patientExternalId ||
if (![1, 2].includes(bundle.version) || (bundle.version === 1 && !diagnoses.length) ||
(bundle.version === 2 && (!['image', 'analysis'].includes(bundle.phase ?? '') ||
(bundle.phase === 'image' && (image.analysis !== undefined || diagnoses.length > 0)) ||
(bundle.phase === 'analysis' && !image.analysis))) ||
diagnoses.some(d => d.patientExternalId !== image.patientExternalId ||
d.clinicId !== clinicId || d.sourceImageExternalId !== image.externalId) || image.clinicId !== clinicId) {
throw new Error('canonical bundle ownership mismatch');
}
......@@ -32,6 +36,16 @@ export class MeyerPacSink implements MeyerSink {
...diagnoses.map(row => ({ resource: 'diagnosis', row, action: Action.DIAGNOSIS_RECORDED,
subjectType: SubjectType.DIAGNOSIS, timeField: 'occurredAt' }))];
for (const item of rows) {
// 元数据重试不能清掉已补充的分析；患者归属始终按完整身份校验。
if (bundle.version === 2 && bundle.phase === 'image') {
const current = await this.prisma.patientFact.findFirst({ where: {
hostId: host.id, tenantId, sourceUnit, subjectId: `image_record:${image.externalId}`,
}, orderBy: { version: 'desc' } });
if (current && (current.patientId !== patient.id || current.clinicId !== clinicId)) {
throw new Error('existing supplemental snapshot differs; image ownership changed');
}
if ((current?.content as { analysis?: unknown } | undefined)?.analysis) continue;
}
const canonicalHash = hash(item.resource === 'image_study' ? { image, diagnoses } : item.row);
const rawRow = { ...item.row, _meyer: { namespace: bundle.namespace,
timeBasis: bundle.timeBasis, observedAt: bundle.observedAt, sourceUnit, canonicalHash,
......@@ -39,7 +53,8 @@ export class MeyerPacSink implements MeyerSink {
const data = this.synthesizer.synthesize({ rawRow, canonicalRow: item.row,
emits: { action: item.action, subjectType: item.subjectType, occurredAtField: item.timeField, contentVersion: false },
resource: item.resource, hostId: host.id, tenantId, sourceUnit,
patientIdResolver: () => patient.id, sourceContext: `supplement:meyer:${bundle.namespace}` });
patientIdResolver: () => patient.id, sourceContext: `supplement:meyer:${bundle.namespace}` +
(bundle.version === 2 && bundle.phase === 'image' ? ':image' : '') });
if (!data) throw new Error('canonical transaction rejected');
// 固定业务身份；重复提交只恢复未完成事实，不把不同内容静默当作重试。
await this.prisma.patientTransaction.createMany({ data: [data], skipDuplicates: true });
......
apps/pac-service/src/integrations/meyer/scan.ts
......@@ -33,7 +33,7 @@ export async function scanWindow(config: MeyerConfig, sourceFor: (c: MeyerConfig
const window = { ...config, timeStart: start, timeEnd: end, pageSize: progress.pageSize };
const source = sourceFor(window);
const result: CollectResult & { complete: boolean; examined: number } = { pages: 0, seen: 0, collected: 0,
applied: 0, alreadyApplied: 0, deferred: 0, failed: 0, errors: [], complete: false, examined: 0 };
applied: 0, alreadyApplied: 0, deferred: 0, failed: 0, imagesApplied: 0, analysisFailed: 0, errors: [], complete: false, examined: 0 };
const limit = config.maxListPagesPerTick;
let validatedScope = -1;
const fetchPage = async (scope: MeyerScope, page: number) => {
......@@ -99,7 +99,7 @@ export async function scanWindow(config: MeyerConfig, sourceFor: (c: MeyerConfig
list: async () => ({ total: 1, page: 1, size: window.pageSize, images: [image] }),
symptom: im => source.symptom(im),
}, store, codes, sink, true);
for (const key of ['seen', 'collected', 'applied', 'alreadyApplied', 'deferred', 'failed'] as const) result[key] += one[key];
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);
......
apps/pac-service/src/integrations/meyer/scheduler.ts
......@@ -52,7 +52,7 @@ export async function runMeyerTick(config: MeyerConfig, sourceFor: (c: MeyerConf
list: async () => ({ total: 1, page: 1, size: config.pageSize, images: [found] }),
symptom: image => client.symptom(image),
}, store, codes, sink);
retryFailed += result.failed;
retryFailed += result.failed + result.analysisFailed;
} catch {
retryFailed++;
await pending.write(key, { ...value, attempts: value.attempts + 1,
......
apps/pac-service/tests/meyer-imaging-postgres.spec.ts
......@@ -11,6 +11,11 @@ import { PrismaService } from '../src/prisma/prisma.service';
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 { MeyerStore } from '../src/integrations/meyer/store';
import { mapMeyerImage, mapMeyerAnalysis } from '../src/integrations/meyer/mapper';
import { imageKey } from '../src/integrations/meyer/contracts';
import { codes } from './meyer-imaging-fixtures';
import { meyerBundle } from './meyer-imaging-fixtures';
import { meyerConfig, meyerImage, meyerSymptom } from './meyer-imaging-fixtures';
import { runMeyerCli } from '../src/cli/meyer-imaging.cli';
......@@ -111,6 +116,64 @@ describePg('Meyer | isolated PostgreSQL + actual minimal CLI module', () => {
await expect(sink.apply(meyerBundle())).rejects.toThrow('snapshot differs');
expect(await prisma.patientFact.count({ where: { hostId } })).toBe(3);
});
test('image-only fact survives analysis outage; late analysis enriches same subject and retries never downgrade', async () => {
const directory = await mkdtemp(join(tmpdir(), 'meyer-image-first-pg-'));
const config = meyerConfig(), image = meyerImage(), scope = config.scopes[0]!;
const store = new MeyerStore(directory);
let ready = false;
const source = { list: async () => ({ total: 1, page: 1, size: 100, images: [image] }),
symptom: async () => { if (!ready) throw Error('analysis outage'); return meyerSymptom(); } };
try {
const first = await collectMeyer(config, source, store, codes, sink);
expect(first).toMatchObject({ imagesApplied: 1, analysisFailed: 1, failed: 0 });
const imageOnly = await prisma.patientFact.findFirstOrThrow({ where: { hostId, patientId } });
expect(imageOnly).toMatchObject({ type: FactType.IMAGE_RECORD, version: 1, status: 'active' });
expect(imageOnly.content).not.toHaveProperty('analysis');
expect(await prisma.patientTransaction.count({ where: { hostId } })).toBe(1);
ready = true;
expect((await collectMeyer(config, source, store, codes, sink)).applied).toBe(1);
const versions = await prisma.patientFact.findMany({ where: { hostId, patientId,
type: FactType.IMAGE_RECORD }, orderBy: { version: 'asc' } });
expect(versions.map(f => [f.subjectId, f.version, f.status])).toEqual([
[imageOnly.subjectId, 1, 'superseded'], [imageOnly.subjectId, 2, 'active']]);
expect(versions[1]!.content).toMatchObject({ analysis: { source: 'meyer', findings: expect.any(Array) } });
const diagnoses = await prisma.patientFact.findMany({ where: { hostId, patientId, type: FactType.DIAGNOSIS_RECORD } });
expect(diagnoses).toHaveLength(2);
expect(diagnoses.every(f => (f.content as any).source_image_external_id === meyerBundle().image.externalId)).toBe(true);
// 即使重放旧影像元数据，已完成分析也不能被清空。
await sink.apply(mapMeyerImage(config, scope, image, '2026-10-17T00:00:00.000Z')!);
await collectMeyer(config, source, store, codes, sink);
expect(await prisma.patientFact.count({ where: { hostId } })).toBe(4);
expect(await prisma.patientTransaction.count({ where: { hostId } })).toBe(4);
const current = await prisma.patientFact.findFirstOrThrow({ where: { hostId, patientId,
subjectId: imageOnly.subjectId, status: 'active' } });
expect(current.version).toBe(2); expect(current.content).toHaveProperty('analysis');
const key = imageKey(scope, image.imageId);
expect(await store.read(`image-${key}`)).toMatchObject({ applied: true });
expect(await new MeyerStore(join(directory, 'pending')).keys()).toEqual([]);
} finally { await rm(directory, { recursive: true, force: true }); }
});
test('analysis without study time is kept on image without invented dated diagnosis', async () => {
const c = meyerConfig(), image = { ...meyerImage(), studyTime: null }, observedAt = '2026-10-17T00:00:00.000Z';
const metadata = mapMeyerImage(c, c.scopes[0]!, image, observedAt)!;
const analysis = mapMeyerAnalysis(c, c.scopes[0]!, image, meyerSymptom(), codes, observedAt)!;
await sink.apply(metadata); await sink.apply(analysis); await sink.apply(analysis);
expect(await prisma.patientFact.count({ where: { hostId, type: FactType.DIAGNOSIS_RECORD } })).toBe(0);
const current = await prisma.patientFact.findFirstOrThrow({ where: { hostId, status: 'active' } });
expect(current.content).toMatchObject({ analysis: { findings: expect.any(Array), analyzedAt: null } });
expect(current.occurredAt?.toISOString()).toBe(metadata.image.uploadedAt);
});
test('v2 rescan of existing legacy analysis is idempotent; changed snapshots still fail closed', async () => {
const c = meyerConfig(), observedAt = '2026-10-17T00:00:00.000Z';
await sink.apply(meyerBundle());
await sink.apply(mapMeyerImage(c, c.scopes[0]!, meyerImage(), observedAt)!);
await sink.apply(mapMeyerAnalysis(c, c.scopes[0]!, meyerImage(), meyerSymptom(), codes, observedAt)!);
expect(await prisma.patientFact.count({ where: { hostId } })).toBe(3);
expect(await prisma.patientTransaction.count({ where: { hostId } })).toBe(3);
const changed = mapMeyerAnalysis(c, c.scopes[0]!, meyerImage(), { ...meyerSymptom(), diagnosis: 'changed' }, codes, observedAt)!;
await expect(sink.apply(changed)).rejects.toThrow('snapshot differs');
expect(await prisma.patientFact.count({ where: { hostId } })).toBe(3);
});
test('same external patient/image IDs stay isolated across Friday brands; other host untouched', async () => {
const a = meyerBundle(); const b = meyerBundle(); b.scope.sourceUnit = 'brand-b';
await sink.apply(a); await sink.apply(b);
......@@ -180,8 +243,8 @@ describePg('Meyer | isolated PostgreSQL + actual minimal CLI module', () => {
const args = [`--config=${join(directory, 'config.json')}`, `--state-dir=${join(directory, 'state')}`, '--apply'];
expect(await runMeyerCli([...args, '--tick'])).toBe(0);
expect(await runMeyerCli([...args, '--tick'])).toBe(0);
expect(await prisma.patientFact.count({ where: { hostId, patientId } })).toBe(3);
expect(await prisma.patientTransaction.count({ where: { hostId } })).toBe(4);
expect(await prisma.patientFact.count({ where: { hostId, patientId } })).toBe(4);
expect(await prisma.patientTransaction.count({ where: { hostId } })).toBe(5);
await prisma.host.update({ where: { id: hostId }, data: {
appSecretHash: hashSecret('local-meyer-open'), openFactsReadEnabled: true,
clinicNames: { 'clinic-a': '本地影像验收诊所' },
......@@ -236,7 +299,7 @@ describePg('Meyer | isolated PostgreSQL + actual minimal CLI module', () => {
await hostRecord(otherPatientId);
await expect(runMeyerCli([...args, '--tick'])).rejects.toThrow('conflicting');
expect(await prisma.patientFact.count({ where: { hostId, patientId } })).toBe(3);
expect(await prisma.patientFact.count({ where: { hostId, patientId } })).toBe(4);
} finally {
await open?.close();
output.mockRestore();
......
apps/pac-service/tests/meyer-imaging.spec.ts
......@@ -7,7 +7,7 @@ import { ConfigSchema, ImageSchema, SymptomSchema, imageKey, sourceTime,
type MeyerBundle, type MeyerConfig, type MeyerSource } from '../src/integrations/meyer/contracts';
import { MeyerClient } from '../src/integrations/meyer/client';
import { collectMeyer } from '../src/integrations/meyer/collector';
import { loadDiagnosisCodes, mapMeyer } from '../src/integrations/meyer/mapper';
import { loadDiagnosisCodes, mapMeyer, mapMeyerImage, mapMeyerAnalysis } from '../src/integrations/meyer/mapper';
import { MeyerStore, type SavedBundle } from '../src/integrations/meyer/store';
import { parseMeyerArgs, runMeyerCli } from '../src/cli/meyer-imaging.cli';
import { ImageParser } from '../src/modules/sync/pipeline/parsers/image.parser';
......@@ -44,7 +44,7 @@ describe('Meyer canonical boundary', () => {
const facts = [...new DiagnosisParser().parse(ctx), ...new ImageParser().parse({ ...ctx, canonicalRow: bundle.image })];
for (const fact of facts) expect(() => validateFactContent(fact.type, fact.subjectId, fact.content)).not.toThrow();
expect(facts[0]!.content).toMatchObject({ code: 'K02', code_source: 'image_ai', tooth_position: '16' });
expect(bundle.raw.symptom.diagnosis).toBe('原始综合诊断');
expect(bundle.raw.symptom!.diagnosis).toBe('原始综合诊断');
expect(bundle.image.analysis).toEqual({ source: 'meyer', summary: '原始综合诊断', description: '描述',
analyzedAt: null, findings: bundle.diagnoses.map(d => ({ name: d.name, toothPosition: d.toothPosition, code: d.code })) });
expect(facts.find(f => f.type === FactType.IMAGE_RECORD)!.content).toMatchObject({ analysis: bundle.image.analysis });
......@@ -146,23 +146,22 @@ describe('Meyer collector durable retry and isolation', () => {
expect((await collectMeyer(c, api, store, codes, sink)).applied).toBe(1);
expect(sink.apply).toHaveBeenCalledWith(retained!.bundle);
expect((await collectMeyer(c, api, store, codes, sink)).alreadyApplied).toBe(1);
expect(sink.apply).toHaveBeenCalledTimes(1);
expect(sink.apply).toHaveBeenCalledTimes(2);
expect(api.symptom).toHaveBeenCalledTimes(1);
const file = join(directory, `bundle-${imageKey(c.scopes[0]!, 'image-1')}.json`);
expect((await stat(file)).mode & 0o777).toBe(0o600);
});
test('retrying an older retained bundle does not rewrite its canonical snapshot', async () => {
const c = meyerConfig(); const api = source();
await collectMeyer(c, api, store, codes);
const key = `bundle-${imageKey(c.scopes[0]!, 'image-1')}`;
const saved = (await store.read<SavedBundle>(key))!;
const saved: SavedBundle = { bundle: meyerBundle(), applied: false };
delete saved.bundle.image.analysis;
await store.write(key, saved);
const sink = { apply: jest.fn(async (_bundle: MeyerBundle) => {}) };
expect((await collectMeyer(c, api, store, codes, sink)).applied).toBe(1);
expect(sink.apply).toHaveBeenCalledWith(saved.bundle);
expect(sink.apply.mock.calls[0]![0].image).not.toHaveProperty('analysis');
expect(api.symptom).toHaveBeenCalledTimes(1);
expect(api.symptom).not.toHaveBeenCalled();
});
test('empty results remain eligible on rerun, never claim a zero-finding completed analysis', async () => {
const api = source(); api.symptom = jest.fn(async () => ({ teeth: [], diagnosis: '', description: '' }));
......@@ -170,6 +169,61 @@ describe('Meyer collector durable retry and isolation', () => {
api.symptom = jest.fn(async () => meyerSymptom());
expect((await collectMeyer(meyerConfig(), api, store, codes)).collected).toBe(1);
});
test('image is written before querying unavailable analysis; late analysis links the same image', async () => {
const c = meyerConfig(), api = source();
const written: MeyerBundle[] = [];
const sink = { apply: jest.fn(async (b: MeyerBundle) => { written.push(b); }) };
api.symptom = jest.fn(async () => {
expect(written).toHaveLength(1);
expect(written[0]!.phase).toBe('image');
throw Error('analysis unavailable');
});
const first = await collectMeyer(c, api, store, codes, sink);
expect(first).toMatchObject({ imagesApplied: 1, failed: 0, analysisFailed: 1 });
expect(written[0]!.image).not.toHaveProperty('analysis');
expect(written[0]!.diagnoses).toEqual([]);
const pending = new MeyerStore(join(directory, 'pending'));
expect(await pending.keys()).toHaveLength(1);
api.symptom = jest.fn(async () => meyerSymptom());
const next = await collectMeyer(c, api, store, codes, sink);
expect(next).toMatchObject({ imagesApplied: 0, applied: 1, failed: 0, analysisFailed: 0 });
expect(written).toHaveLength(2);
expect(written[1]!.image.externalId).toBe(written[0]!.image.externalId);
expect(written[1]!.diagnoses.every(d => d.sourceImageExternalId === written[0]!.image.externalId)).toBe(true);
expect(await pending.keys()).toEqual([]);
expect((await collectMeyer(c, api, store, codes, sink)).alreadyApplied).toBe(1);
expect(written).toHaveLength(2);
});
test('blank analysis keeps a persisted image; missing study time and summary-only analysis do not invent diagnoses', async () => {
const c = meyerConfig(), im = { ...meyerImage(), studyTime: null };
const metadata = mapMeyerImage(c, c.scopes[0]!, im, '2026-10-17T00:00:00.000Z')!;
expect(metadata.image).not.toHaveProperty('analysis'); expect(metadata.diagnoses).toEqual([]);
const analysis = mapMeyerAnalysis(c, c.scopes[0]!, im, meyerSymptom(), codes, metadata.observedAt)!;
expect(analysis.image.analysis!.findings).toHaveLength(2);
expect(analysis.diagnoses).toEqual([]); expect(analysis.image.analysis!.analyzedAt).toBeNull();
const summary = mapMeyerAnalysis(c, c.scopes[0]!, meyerImage(), {
teeth: [], diagnosis: '源综合结论', description: '' }, codes, metadata.observedAt)!;
expect(summary.image.analysis!.summary).toBe('源综合结论'); expect(summary.diagnoses).toEqual([]);
const api = source(); api.symptom = jest.fn(async () => ({ teeth: [], diagnosis: '', description: '' }));
const sink = { apply: jest.fn(async (_b: MeyerBundle) => {}) };
const result = await collectMeyer(c, api, store, codes, sink);
expect(result).toMatchObject({ imagesApplied: 1, deferred: 1, applied: 0, failed: 0 });
expect(sink.apply).toHaveBeenCalledTimes(1);
const key = imageKey(c.scopes[0]!, 'image-1');
expect(await store.read(`image-${key}`)).toMatchObject({ applied: true });
expect(await store.read(`bundle-${key}`)).toBeNull();
});
test('analysis write failure retries its immutable snapshot without reapplying metadata', async () => {
const c = meyerConfig(), api = source();
const sink = { apply: jest.fn(async (b: MeyerBundle) => { if (b.phase === 'analysis') throw Error('write failed'); }) };
expect(await collectMeyer(c, api, store, codes, sink)).toMatchObject({ imagesApplied: 1, analysisFailed: 1, failed: 0 });
api.symptom = jest.fn(async () => { throw Error('must use retained analysis'); });
const recovered = { apply: jest.fn(async (_b: MeyerBundle) => {}) };
expect((await collectMeyer(c, api, store, codes, recovered)).applied).toBe(1);
expect(recovered.apply).toHaveBeenCalledTimes(1);
expect(recovered.apply.mock.calls[0]![0].phase).toBe('analysis');
expect(api.symptom).not.toHaveBeenCalled();
});
test('wrong clinic and changed owner do not produce writes', async () => {
const api = source(); const c = meyerConfig();
await collectMeyer(c, api, store, codes);
......
apps/pac-service/tests/meyer-resume.spec.ts
......@@ -21,7 +21,7 @@ describe('Meyer conservative limits and recovery', () => {
});
test('row checkpoint survives restart; cursor waits for all pages; renewed URLs do not reset progress', async () => {
const rows = images(), c = config(); const applied: string[] = [];
const sink = { apply: async (b: MeyerBundle) => { applied.push(b.raw.image.imageId); } };
const sink = { apply: async (b: MeyerBundle) => { if (b.phase === 'analysis') applied.push(b.raw.image.imageId); } };
const tick = () => runMeyerTick(c, () => source(rows), new MeyerStore(dir), codes, sink, now);
const first = await tick(); expect(first.scan?.examined).toBe(2); expect(first.cursor).toBeNull();
expect(await new MeyerStore(dir).read('scan-progress')).toMatchObject({ page: 1, row: 2 });
......@@ -33,7 +33,7 @@ describe('Meyer conservative limits and recovery', () => {
});
test('offset shift restarts fixed window and does not skip inserted image', async () => {
const rows = images(), c = config(), written: string[] = [];
const sink = { apply: async (b: MeyerBundle) => { written.push(b.raw.image.imageId); } };
const sink = { apply: async (b: MeyerBundle) => { if (b.phase === 'analysis') written.push(b.raw.image.imageId); } };
await runMeyerTick(c, () => source(rows), new MeyerStore(dir), codes, sink, now);
rows.unshift({ ...meyerImage(), imageId: 'inserted' });
for (let n = 0; n < 4; n++) {
......@@ -53,7 +53,7 @@ describe('Meyer conservative limits and recovery', () => {
expect(first.scan?.failed).toBe(1); expect(first.cursor).toBeNull();
down = false;
const next = await runMeyerTick(c, factory, new MeyerStore(dir), codes, sink, now);
expect(next.scan?.complete).toBe(true); expect(sink.apply).toHaveBeenCalledTimes(5);
expect(next.scan?.complete).toBe(true); expect(sink.apply).toHaveBeenCalledTimes(10);
});
test('duplicate caused by offset drift blocks completion and can recover by re-scan', async () => {
const c = { ...config(), maxNewImagesPerTick: 10 }, rows = images(); let drift = true;
......@@ -66,7 +66,7 @@ describe('Meyer conservative limits and recovery', () => {
expect(first.scan?.failed).toBe(1); expect(first.cursor).toBeNull();
drift = false;
const next = await runMeyerTick(c, factory, new MeyerStore(dir), codes, sink, now);
expect(next.scan?.complete).toBe(true); expect(sink.apply).toHaveBeenCalledTimes(5);
expect(next.scan?.complete).toBe(true); expect(sink.apply).toHaveBeenCalledTimes(10);
});
test('watch backs off for thrown outages and error exits, recovers, then stops cleanly', async () => {
const c = { ...config(), pollIntervalMs: 1000, errorBackoffMaxMs: 4000 };
......
apps/pac-service/tests/meyer-scheduler.spec.ts
......@@ -23,13 +23,25 @@ describe('independent Meyer scheduling', () => {
symptom: async () => ready ? meyerSymptom() : { teeth: [], diagnosis: '', description: '' },
});
const first = await runMeyerTick(c, factory, store, codes, sink, now);
expect(first.pending).toBe(1); expect(first.scan?.failed).toBe(0); expect(sink.apply).not.toHaveBeenCalled();
expect(first.pending).toBe(1); expect(first.scan?.failed).toBe(0); expect(sink.apply).toHaveBeenCalledTimes(1);
expect(first.scan?.imagesApplied).toBe(1);
expect(await store.read('cursor')).toEqual({ through: '2026-10-10 10:00:00' });
const key = imageKey(c.scopes[0]!, im.imageId);
const item = await pending.read<any>(key); await pending.write(key, { ...item, nextRetryAt: '2000-01-01T00:00:00Z' });
ready = true;
const next = await runMeyerTick(c, factory, store, codes, sink, now);
expect(next.retryAttempted).toBe(1); expect(next.pending).toBe(0); expect(sink.apply).toHaveBeenCalledTimes(1);
expect(next.retryAttempted).toBe(1); expect(next.pending).toBe(0); expect(sink.apply).toHaveBeenCalledTimes(2);
});
test('analysis outage does not block the image scan cursor or later images', async () => {
const c = config(), store = new MeyerStore(dir), written: string[] = [];
const rows = [meyerImage(), { ...meyerImage(), imageId: 'second' }];
const first = await runMeyerTick(c, () => ({
list: async () => ({ total: 2, page: 1, size: 100, images: rows }),
symptom: async () => { throw Error('analysis service down'); },
}), store, codes, { apply: async b => { written.push(b.image.externalId); } }, now);
expect(written).toHaveLength(2);
expect(first.scan).toMatchObject({ complete: true, imagesApplied: 2, failed: 0, analysisFailed: 2 });
expect(first.pending).toBe(2); expect(first.cursor).toEqual({ through: '2026-10-10 10:00:00' });
});
test('failed page never advances cursor; minute overlap is explicit', async () => {
const store = new MeyerStore(dir); const windows: string[] = [];
......
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