fix: sync PPI scheduling and harden preflight

This commit is contained in:
wtx
2026-08-22 05:09:52 +08:00
parent 052b0c1d5d
commit 311a49110f
6 changed files with 465 additions and 109 deletions
+17 -2
View File
@@ -14188,10 +14188,25 @@ compare: [[flesh, peel]]`;
};
};
const fetchWithTransientRetry = async (input, init = {}) => {
const retryDelays = [500, 1500];
for (let attempt = 0; ; attempt += 1) {
try {
return await fetch(input, init);
} catch (err) {
const message = String(err?.message || err || '');
const transient = err instanceof TypeError
|| /NetworkError|Failed to fetch|Load failed/i.test(message);
if (!transient || attempt >= retryDelays.length) throw err;
await new Promise(resolve => setTimeout(resolve, retryDelays[attempt]));
}
}
};
const pollReferenceBundleDownloadTask = async (taskId, options = {}) => {
const intervalMs = Number(options.intervalMs || 3000);
while (true) {
const response = await fetch(`/api/task/${encodeURIComponent(taskId)}`);
const response = await fetchWithTransientRetry(`/api/task/${encodeURIComponent(taskId)}`);
const data = await response.json().catch(() => ({}));
if (!response.ok) {
throw new Error(referenceBundleErrorMessage(data, `Reference species data status request failed (HTTP ${response.status})`));
@@ -14533,7 +14548,7 @@ compare: [[flesh, peel]]`;
if (requestedStages.length === 1 && requestedStages[0] === 'ppi') {
canonicalizePpiOwnerBiologicalScope(payload);
}
const response = await fetch('/api/ppi/species-resolution', {
const response = await fetchWithTransientRetry('/api/ppi/species-resolution', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify(payload)
+151 -106
View File
@@ -5033,7 +5033,7 @@
</div>
<!-- 阶段特定参数 -->
<div v-if="!['ppi_kaks_reference_data', 'ppi_kaks_download_fallback'].includes(confirmParams.stage)" class="param-group">
<div v-if="!['ppi_kaks_reference_data', 'ppi_kaks_download_fallback', 'ppi_string_reference_confirmation', 'ppi_kaks_reference_status'].includes(confirmParams.stage)" class="param-group">
<h5>{{ t('基本参数') }}</h5>
<div class="param-row">
<div class="param-field param-field--wide">
@@ -6575,12 +6575,21 @@ RNA-seq KEGG/fgsea 会优先使用物种级 KEGG gene set;没有物种级缓
</div>
<div v-else-if="gwasConfirmationBlockedFor(confirmParams)" class="param-dialog-footer__status" style="font-size: 14px;">⏸ 等待选择有效 VCF</div>
<div v-else-if="confirmParams.stage === 'ppi_kaks_download_fallback'" class="param-dialog-footer__status" style="font-size: 14px;">⏸ 下载失败兜底需要手动确认</div>
<div v-else-if="confirmParams.stage === 'ppi_string_reference_confirmation'" class="param-dialog-footer__status" style="font-size: 14px;">⏸ Manual reference confirmation required</div>
<div v-else-if="confirmParams.stage === 'ppi_kaks_reference_status'" class="param-dialog-footer__status" style="font-size: 14px;">Reference status acknowledged</div>
<div v-else class="param-dialog-footer__status" style="font-size: 14px;">⏸ 倒计时已暂停,请确认参数后手动执行</div>
<div class="param-dialog-footer__actions">
<template v-if="isDirectReplotDialog">
<el-button @click="cancelStageConfirmDialog">{{ t('取消') }}</el-button>
<el-button type="primary" :loading="isExecuting" @click="confirmStageExecute">{{ t('确认重绘') }}</el-button>
</template>
<template v-else-if="confirmParams.stage === 'ppi_string_reference_confirmation'">
<el-button @click="skipStageExecute">Cancel PPI</el-button>
<el-button type="primary" @click="confirmStageExecute">Confirm top candidate</el-button>
</template>
<template v-else-if="confirmParams.stage === 'ppi_kaks_reference_status'">
<el-button type="primary" @click="confirmStageExecute">OK</el-button>
</template>
<template v-else>
<el-button type="danger" @click="stopBatchExecute">⏹️ 停止全部</el-button>
<el-button v-if="stageConfirmCountdown > 0" @click="clearStageConfirmTimer">{{ t('暂停倒计时') }}</el-button>
@@ -13907,32 +13916,16 @@ compare: [[flesh, peel]]`;
? resolution.candidates
: (Array.isArray(resolution.species_resolution?.candidates) ? resolution.species_resolution.candidates : []);
if (!candidates.length) return null;
const candidateLines = candidates.map((candidate, index) => {
const name = candidate.selected_string_name || candidate.species_name || '-';
const taxon = candidate.selected_string_taxon_id || candidate.taxon_id || '-';
const lca = candidate.lca_name || candidate.lca_taxon_name || candidate.lca_taxon_id || '-';
const distance = candidate.topology_distance ?? candidate.distance_edges ?? candidate.distance ?? '-';
const version = candidate.string_version || resolution.string_version || resolution.species_resolution?.string_version || '-';
return `${index + 1}. ${name} / STRING TaxID ${taxon} / LCA ${lca} / 拓扑边数 ${distance} / STRING ${version}`;
});
const selected = candidates[0];
const selectedTaxon = String(selected.selected_string_taxon_id || selected.taxon_id || '').trim();
if (!selectedTaxon) return null;
try {
await ElMessageBox.confirm(
[
t('string-close-species-line1'),
'',
...candidateLines,
'',
t('string-close-species-line2')
].join('\n'),
t('string-close-species-title'),
{ type: 'warning', confirmButtonText: t('string-confirm-first-candidate'), cancelButtonText: t('string-cancel-ppi') }
);
} catch (_err) {
return null;
}
const accepted = await showStageConfirmDialog({
id: 'ppi_string_reference_confirmation',
name: 'Confirm the STRING reference species for PPI',
dependencies: [],
params: { resolution, candidates }
}, 1, 1);
if (!accepted) return null;
return {
taxonId: selectedTaxon,
storageKey: `string_taxon_${selectedTaxon}`,
@@ -14051,17 +14044,18 @@ compare: [[flesh, peel]]`;
const clearPpiKaksReferencePause = (requestedStages = []) => {
const pauseStageIds = ppiKaksPauseStageIds(requestedStages);
const current = batchExecuteParams.value || {};
if (current.ppiKaksDependencyPause) {
const next = { ...current };
delete next.ppiKaksDependencyPause;
batchExecuteParams.value = next;
}
pauseStageIds.forEach(stageId => {
const nextPauses = { ...(current.ppiKaksDependencyPauses || {}) };
requestedStages.forEach(ownerStageId => delete nextPauses[ownerStageId]);
const next = { ...current, ppiKaksDependencyPauses: nextPauses };
delete next.ppiKaksDependencyPause;
batchExecuteParams.value = next;
[...pauseStageIds, 'ortholog'].forEach(stageId => {
clearStagePause(stageId);
if (stageExecutionStatus.value[stageId] === 'paused') {
stageExecutionStatus.value[stageId] = 'pending';
}
});
applyStoredPpiKaksPauseStatuses();
};
const applyArabidopsisFallbackForPpiKaks = (requestedStages = [], reason = '') => {
@@ -14280,10 +14274,25 @@ compare: [[flesh, peel]]`;
};
};
const fetchWithTransientRetry = async (input, init = {}) => {
const retryDelays = [500, 1500];
for (let attempt = 0; ; attempt += 1) {
try {
return await fetch(input, init);
} catch (err) {
const message = String(err?.message || err || '');
const transient = err instanceof TypeError
|| /NetworkError|Failed to fetch|Load failed/i.test(message);
if (!transient || attempt >= retryDelays.length) throw err;
await new Promise(resolve => setTimeout(resolve, retryDelays[attempt]));
}
}
};
const pollReferenceBundleDownloadTask = async (taskId, options = {}) => {
const intervalMs = Number(options.intervalMs || 3000);
while (true) {
const response = await fetch(`/api/task/${encodeURIComponent(taskId)}`);
const response = await fetchWithTransientRetry(`/api/task/${encodeURIComponent(taskId)}`);
const data = await response.json().catch(() => ({}));
if (!response.ok) {
throw new Error(referenceBundleErrorMessage(data, `Reference species data status request failed (HTTP ${response.status})`));
@@ -14303,11 +14312,32 @@ compare: [[flesh, peel]]`;
}
};
const ppiKaksPauseStageIds = (requestedStages = []) => {
const ids = new Set((requestedStages || []).filter(Boolean));
const plannedIds = new Set((batchExecuteStages.value || []).map(stage => getStageId(stage)).filter(Boolean));
if (plannedIds.has('ortholog')) ids.add('ortholog');
return Array.from(ids);
const ppiKaksOwnedOrthologOwners = (orthologStage = {}, batchParams = {}) => {
const stageParams = orthologStage && typeof orthologStage === 'object' ? (orthologStage.params || {}) : {};
const intentStages = Array.isArray(batchParams.intentStages) ? batchParams.intentStages : [];
if (intentStages.includes('ortholog')) return [];
const dependencyOwner = String(stageParams.dependencyOwner || '').trim();
if (dependencyOwner === 'ppi' || dependencyOwner === 'kaks') return [dependencyOwner];
const autoInsertedFor = Array.isArray(stageParams.autoInsertedFor)
? stageParams.autoInsertedFor.filter(owner => owner === 'ppi' || owner === 'kaks')
: [];
if (autoInsertedFor.length) return Array.from(new Set(autoInsertedFor));
if (intentStages.includes('kaks')) return ['kaks'];
return [];
};
const ppiKaksPauseStageIds = (requestedStages = []) => (
Array.from(new Set((requestedStages || []).filter(Boolean)))
);
const ppiKaksDependencyPauses = () => {
const current = batchExecuteParams.value || {};
if (current.ppiKaksDependencyPauses && typeof current.ppiKaksDependencyPauses === 'object') {
return current.ppiKaksDependencyPauses;
}
const legacyPause = current.ppiKaksDependencyPause;
const legacyOwner = Array.isArray(legacyPause?.requestedStages) ? legacyPause.requestedStages[0] : '';
return legacyOwner ? { [legacyOwner]: legacyPause } : {};
};
const storePpiKaksReferencePause = (requestedStages = [], resolution = {}, message = '') => {
@@ -14320,21 +14350,37 @@ compare: [[flesh, peel]]`;
requestedStages,
pauseStageIds
};
batchExecuteParams.value = {
...(batchExecuteParams.value || {}),
ppiKaksDependencyPause: pauseInfo
};
const nextPauses = { ...ppiKaksDependencyPauses() };
requestedStages.forEach(ownerStageId => {
if (ownerStageId) nextPauses[ownerStageId] = pauseInfo;
});
const next = { ...(batchExecuteParams.value || {}), ppiKaksDependencyPauses: nextPauses };
delete next.ppiKaksDependencyPause;
batchExecuteParams.value = next;
return pauseInfo;
};
const applyStoredPpiKaksPauseStatuses = () => {
const pause = batchExecuteParams.value?.ppiKaksDependencyPause;
const pauseStageIds = Array.isArray(pause?.pauseStageIds) ? pause.pauseStageIds : [];
pauseStageIds.forEach(stageId => {
if (!stageId) return;
stageExecutionStatus.value[stageId] = 'paused';
recordStagePause(stageId, pause);
const pauses = ppiKaksDependencyPauses();
Object.values(pauses).forEach(pause => {
const pauseStageIds = Array.isArray(pause?.pauseStageIds) ? pause.pauseStageIds : [];
pauseStageIds.forEach(stageId => {
if (!stageId) return;
stageExecutionStatus.value[stageId] = 'paused';
recordStagePause(stageId, pause);
});
});
const orthologStage = (batchExecuteStages.value || []).find(stage => getStageId(stage) === 'ortholog');
const orthologOwners = ppiKaksOwnedOrthologOwners(orthologStage, batchExecuteParams.value || {});
const plannedOwners = new Set((batchExecuteStages.value || [])
.map(stage => getStageId(stage))
.filter(stageId => stageId === 'ppi' || stageId === 'kaks'));
const activeOrthologOwners = orthologOwners.filter(owner => plannedOwners.has(owner));
if (activeOrthologOwners.length && activeOrthologOwners.every(owner => Boolean(pauses[owner]))) {
const pause = pauses[activeOrthologOwners[0]];
stageExecutionStatus.value.ortholog = 'paused';
recordStagePause('ortholog', pause);
}
};
const mergePpiReferenceSpeciesIntoBatch = (speciesCode, requestedStages = [], policySource = 'ppi_kaks_reference_bundle_download', replaceExisting = false) => {
@@ -14588,7 +14634,7 @@ compare: [[flesh, peel]]`;
if (requestedStages.length === 1 && requestedStages[0] === 'ppi') {
canonicalizePpiOwnerBiologicalScope(payload);
}
const response = await fetch('/api/ppi/species-resolution', {
const response = await fetchWithTransientRetry('/api/ppi/species-resolution', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify(payload)
@@ -14603,21 +14649,23 @@ compare: [[flesh, peel]]`;
return { allowed: true, resolution, message: '' };
}
if (options.allowNeedsOrtholog === true && resolution.status === 'needs_ortholog') {
if (!ppiNeedsNearReferenceDecision(resolution)) {
if (!ppiNeedsNearReferenceDecision(resolution, requestedStages)) {
return { allowed: true, resolution, message: '' };
}
}
const message = formatPpiSpeciesResolutionBlockMessage(resolution);
if (
options.suppressNeedsReferenceDataAlert === true
&& (resolution.status === 'needs_reference_data' || ppiNeedsNearReferenceDecision(resolution))
&& (resolution.status === 'needs_reference_data' || ppiNeedsNearReferenceDecision(resolution, requestedStages))
) {
return { allowed: false, resolution, message };
}
await ElMessageBox.alert(message, 'PPI/KaKs 需要先处理参考物种/Ortholog', {
type: resolution.status === 'invalid_mapping' ? 'error' : 'warning',
confirmButtonText: '知道了'
});
await showStageConfirmDialog({
id: 'ppi_kaks_reference_status',
name: 'PPI/KaKs Reference Status',
dependencies: [],
params: { resolution, requestedStages }
}, 1, 1);
addAgentMessage('assistant', message);
return { allowed: false, resolution, message };
};
@@ -14851,62 +14899,23 @@ compare: [[flesh, peel]]`;
};
const ensureOrthologStageForPpiKaks = (stages, requestedStages, requestBody, resolution = {}) => {
const stageIds = stages.map(stage => getStageId(stage)).filter(Boolean);
const recoveredReferenceApplied = applyResolvedPpiReferenceToBatch(resolution, requestedStages, requestBody);
if (recoveredReferenceApplied) {
addAgentMessage('assistant', t('✅ PPI/KaKs 前置数据检查通过:检测到已下载近源参考 ') + resolution.reference_species + t(',将用该参考先生成 Ortholog mapping。'));
addAgentMessage('assistant', `✅ PPI/KaKs 前置数据检查通过:检测到已下载近源参考 ${resolution.reference_species},将用该参考先生成 Ortholog mapping。`);
}
const mappingLabel = ppiOrthologMappingLabel(resolution);
if (mappingLabel && resolution?.status !== 'needs_ortholog') {
addAgentMessage('assistant', t('✅ PPI/KaKs 前置数据检查通过:检测到已有 Ortholog 映射(') + mappingLabel + t('),将复用该映射执行 ') + requestedStages.join('、') + '。');
requestedStages.forEach(owner => rememberPpiKaksInternalOrthologProducer(owner, null));
addAgentMessage('assistant', `✅ PPI/KaKs 前置数据检查通过:检测到已有 Ortholog 映射(${mappingLabel}),将复用该映射执行 ${requestedStages.join('、')}。`);
return true;
}
attachOrthologDependencyToOwners(requestedStages);
if (stageIds.includes('ortholog')) {
if (requestedStages.includes('ppi')) {
const ownerState = {
ppiReferenceTaxonId: requestBody.ppiReferenceTaxonId || '',
ppiReferenceStorageKey: requestBody.ppiReferenceStorageKey || '',
ppiReferenceSelectionMode: requestBody.ppiReferenceSelectionMode || '',
ppiReferenceUserConfirmed: requestBody.ppiReferenceUserConfirmed === true
};
batchExecuteStages.value = (batchExecuteStages.value || stages).map(stage => (
getStageId(stage) === 'ortholog'
? buildPpiOwnedOrthologStage(requestBody, ownerState, requestedStages)
: stage
));
}
addAgentMessage('assistant', t('✅ PPI/KaKs 前置数据检查通过:当前批次已有 Ortholog,将先生成同源映射再继续下游分析。'));
return true;
}
const insertAt = Math.max(0, stages.findIndex(stage => ppiKaksDependencyStages.has(getStageId(stage))));
const orthologStage = requestedStages.includes('ppi')
? buildPpiOwnedOrthologStage(requestBody, requestBody, requestedStages)
: {
id: 'ortholog',
name: 'Ortholog',
dependencies: ['identify'],
params: {
...requestBody,
autoInsertedFor: requestedStages,
requestedStages,
referencePolicySource: 'ppi_kaks_dependency_preflight'
}
};
const nextStages = [...(batchExecuteStages.value || stages)];
nextStages.splice(insertAt < 0 ? 0 : insertAt, 0, orthologStage);
batchExecuteStages.value = nextStages;
stageExecutionStatus.value.ortholog = 'pending';
if (Array.isArray(batchExecuteParams.value?.completed_stages)) {
batchExecuteParams.value = {
...(batchExecuteParams.value || {}),
completed_stages: batchExecuteParams.value.completed_stages.filter(stageId => stageId !== 'ortholog')
};
}
if (!requestedStages.includes('ppi')) return true;
const orthologStage = buildPpiOwnedOrthologStage(requestBody, requestBody, ['ppi']);
rememberPpiKaksInternalOrthologProducer('ppi', orthologStage);
const reason = resolution?.issue_code === 'stale_ortholog_reference_conflict'
? '已有 Ortholog 映射与当前 PPI/KaKs 参考物种不一致'
: '缺少可用同源映射';
addAgentMessage('assistant', t('✅ PPI/KaKs 前置数据检查通过:') + reason + t(',已自动插入 Ortholog,将先生成 mapping 再执行 ') + requestedStages.join('、') + '。');
addAgentMessage('assistant', `✅ PPI 前置数据检查通过:${reason};执行 PPI 时将先在内部生成专属同源映射。`);
return true;
};
@@ -15679,11 +15688,7 @@ compare: [[flesh, peel]]`;
await loadInterspeciesSpeciesOptions();
}
normalizeBatchExecutionSpeciesForStages();
const dependencyPreflightOk = await ensureOrthologProducerForPpiKaks();
if (!dependencyPreflightOk) {
addAgentMessage('assistant', t('⏸️ PPI/KaKs 前置数据检查未通过,已暂停批量执行;请补齐参考注释包或确认参考物种后重新执行。'));
return;
}
await ensureOrthologProducerForPpiKaks();
isBatchExecuting.value = true;
shouldStopBatchExecute.value = false;
@@ -22701,9 +22706,27 @@ compare: [[flesh, peel]]`;
throw createStructuredExecuteError(data);
};
const isPpiKaksBackgroundPause = (stageId, stageParams = {}, err = {}, intentStages = []) => {
const resolution = err.ppiSpeciesResolution || err.responseData?.ppi_species_resolution || null;
const status = String(err.status || resolution?.status || '').trim();
const isPreflightBlock = Boolean(resolution) || [
'needs_reference_data', 'confirmation_required', 'needs_ortholog',
'unresolved', 'ppi_kaks_dependency_resolution_blocked'
].includes(status);
if (!isPreflightBlock) return false;
if (stageId === 'ppi' || stageId === 'kaks') return true;
if (stageId !== 'ortholog' || intentStages.includes('ortholog')) return false;
const dependencyOwner = String(stageParams.dependencyOwner || '').trim();
const autoInsertedFor = Array.isArray(stageParams.autoInsertedFor) ? stageParams.autoInsertedFor : [];
return dependencyOwner === 'ppi'
|| dependencyOwner === 'kaks'
|| autoInsertedFor.some(owner => owner === 'ppi' || owner === 'kaks');
};
const startBackgroundBatchExecution = async (stages, plan, planData = {}) => {
const batchId = `bg_${Date.now()}`;
const stageIds = stages.map(getStageId).filter(Boolean);
const intentStages = Array.isArray(planData.intent?.stages) ? planData.intent.stages : [];
const stageMap = {};
stages.forEach(stage => {
stageMap[getStageId(stage)] = stage;
@@ -22734,6 +22757,7 @@ compare: [[flesh, peel]]`;
let completed = 0;
let failed = 0;
let paused = 0;
while (true) {
const pending = stageIds.filter(stageId => localStatus[stageId] === 'pending');
@@ -22749,6 +22773,21 @@ compare: [[flesh, peel]]`;
const executable = limitGwasModelHeavyWave(executableCandidates, localStatus);
if (executable.length === 0) {
const pausedDeps = pending.filter(stageId => {
let deps = stageDependencies[stageId] || [];
if (stageId === 'rnaseq_fgsea' && !stageIds.includes('rnaseq_kegg_enrich')) {
deps = deps.filter(dep => dep !== 'rnaseq_kegg_enrich');
}
return deps.some(dep => localStatus[dep] === 'paused');
});
if (pausedDeps.length > 0) {
pausedDeps.forEach(stageId => {
localStatus[stageId] = 'paused';
paused++;
});
addAgentMessage('assistant', t('⏸️ 后台批次中以下阶段等待上游暂停项处理: ') + pausedDeps.map(getStageLabel).join(', '));
continue;
}
const blocked = pending.filter(stageId => {
let deps = stageDependencies[stageId] || [];
if (stageId === 'rnaseq_fgsea' && !stageIds.includes('rnaseq_kegg_enrich')) {
@@ -22777,15 +22816,21 @@ compare: [[flesh, peel]]`;
localStatus[stageId] = 'running';
const stage = stageMap[stageId] || stageId;
const stageName = getStageName(stage);
const stageParams = (plan.stages_detail || []).find(s => s.id === stageId)?.params || plan.params || {};
addAgentMessage('assistant', t('⚙️ 后台执行阶段: ') + stageName + '...');
try {
const stageParams = (plan.stages_detail || []).find(s => s.id === stageId)?.params || plan.params || {};
await executeStageDirect(stageId, stageParams);
localStatus[stageId] = 'completed';
completed++;
addAgentMessage('assistant', t('✅ 后台阶段 ') + stageName + t(' 完成'));
} catch (err) {
if (isPpiKaksBackgroundPause(stageId, stageParams, err, intentStages)) {
localStatus[stageId] = 'paused';
paused++;
addAgentMessage('assistant', t('⏸️ 后台阶段 ') + stageName + t(' 前置数据未就绪,已暂停;其它阶段继续执行。'));
return;
}
localStatus[stageId] = 'failed';
failed++;
addAgentMessage('assistant', await showFailureDiagnosis(stageId, err, { fallbackMessage: err.message }));
@@ -22793,7 +22838,7 @@ compare: [[flesh, peel]]`;
}));
}
addAgentMessage('assistant', t('📦 后台批次完成:成功 ') + completed + '/' + stageIds.length + t(',失败 ') + failed + '/' + stageIds.length);
addAgentMessage('assistant', t('📦 后台批次完成:成功 ') + completed + '/' + stageIds.length + t(',暂停 ') + paused + '/' + stageIds.length + t(',失败 ') + failed + '/' + stageIds.length);
await refreshResults();
};
+19
View File
@@ -4312,6 +4312,25 @@ def initialize_agent_conversation_store() -> None:
agent_conversation_store = PersistentConversationStore(repository)
@app.on_event("startup")
def warm_string_species_snapshot_cache() -> None:
"""Warm the fixed STRING snapshot before the first PPI preflight request."""
started_at = time.perf_counter()
try:
from apps.web import pipeline_backend
snapshot = pipeline_backend.load_string_snapshot(
pipeline_backend.STRING_SPECIES_SNAPSHOT_ROOT
)
elapsed = time.perf_counter() - started_at
print(
"[PPI] STRING snapshot cache warmed "
f"in {elapsed:.3f}s: {snapshot.snapshot_id}"
)
except Exception as exc:
print(f"[WARN] STRING snapshot cache warm failed: {exc}")
class AgentConversationCreateRequest(BaseModel):
user_id: str = "default"
title: str = ""