From 0282cfeeeb24f57dac14a17b2b40b1bbf756455b Mon Sep 17 00:00:00 2001 From: admin Date: Thu, 24 Sep 2026 16:48:52 +0800 Subject: [PATCH] =?UTF-8?q?feat(bioflow):=20=E9=98=B6=E6=AE=B5=E8=BF=9B?= =?UTF-8?q?=E5=BA=A6=E6=9D=A1+=E6=BB=9A=E5=8A=A8=E6=97=A5=E5=BF=97+?= =?UTF-8?q?=E7=82=B9=E5=87=BB=E5=AE=9A=E4=BD=8D=E5=85=A8=E9=93=BE;?= =?UTF-8?q?=E5=9B=BE=E7=89=87=E6=B3=A8=E5=85=A5=E4=B8=89=E4=BF=AE;QA?= =?UTF-8?q?=E5=90=AF=E5=8A=A8=E5=B0=B1=E7=BB=AA=E7=AD=89=E5=BE=85?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 生信进度:后端新增 stage_progress 事件帧+进度解析白名单(rnaseq_qc/sample_qc), kernel 日志流式落盘(20行/1s flush+fsync),logs 分页端点(offset/limit/eof), 事件 seq 跨重启扫描续增(2026-09-17 断流修复);前端 FlowProgressCard 确定性/ 不确定两态进度条+点击定位到生信任务 TAB,BioJobsPane DAG 行展开滚动日志区 (3s 轮询),log_tail 4000 截断,重跑 stage_started 清零进度 - 图片注入:figures 三源重构(标准DOI图库主路径/遗留体系兜底/PMC兜底), [[FIG:]] 宽松归一化识别 PMC/PMID/DOI 三种形态,引文兜底扩展至 [PMCID:]/[DOI:] 且标记未满配额继续注入,注入预算 90s/10篇(175 实测校准), PMC graphic 无 content-type 宽松正则,空结果不缓存(176 冷连接三连败修复) - QA 启动自愈:start.sh 启动前等 Qdrant readyz(30min)+Neo4j TCP(10min); backend_guard 标签校验改子集判定 - 样本配对:同 inode 去重优先真实文件,R1/R2 归一配对一对一认领 - gitignore 新增 *.bak-*/*.before-*/bioflow_service/bioflow.db 测试:bioflow_service pytest 358/358,前端 bioflow vitest 205/205 --- .gitignore | 5 + agent/qa/scripts/start.sh | 16 ++ agent/qa/src/rag/qa_v1_1_73/backend_guard.py | 2 +- bioflow_service/app/routers/biojobs.py | 40 +++ .../app/services/bioflow/events.py | 40 ++- .../app/services/bioflow/kernel.py | 86 +++++-- .../app/services/bioflow/manifests.py | 17 +- .../app/services/bioflow/runner.py | 165 +++++++++++- bioflow_service/tests/conftest.py | 1 + bioflow_service/tests/test_api.py | 52 ++++ bioflow_service/tests/test_events.py | 67 +++++ bioflow_service/tests/test_kernel.py | 89 +++++++ bioflow_service/tests/test_manifests.py | 61 +++++ bioflow_service/tests/test_runner.py | 203 +++++++++++++++ sdk/app/services/figures.py | 243 +++++++++++++++--- sdk/app/services/normalize.py | 76 ++++-- sdk/app/services/sse_relay.py | 84 ++++-- .../unit/bioflow/bioflow.store.test.js | 72 +++++- tests/frontend/unit/bioflow/panes.test.js | 20 +- ui/src/api/bioflow.js | 5 + ui/src/components/bench/BioJobsPane.vue | 216 +++++++++++++++- ui/src/components/chat/AssistantMessage.vue | 3 +- ui/src/components/chat/ChatPanel.vue | 8 + ui/src/components/chat/FlowProgressCard.vue | 110 +++++++- ui/src/components/chat/MessageList.vue | 3 +- ui/src/i18n/bioflow.js | 1 + ui/src/stores/bioflow.js | 116 ++++++++- ui/src/stores/chat.js | 47 +++- 28 files changed, 1720 insertions(+), 128 deletions(-) diff --git a/.gitignore b/.gitignore index e09e899..661b8f0 100644 --- a/.gitignore +++ b/.gitignore @@ -31,3 +31,8 @@ kb-libraries/aqua/data/ # 18 阶段拓展分析测试用例数据(2.6G,不入库;测试时从本机同步) tests/extension_cases_18/ + +# 开发过程备份文件与本地运行库 +*.bak-* +bioflow_service/bioflow.db +*.before-* diff --git a/agent/qa/scripts/start.sh b/agent/qa/scripts/start.sh index c666342..019b54b 100755 --- a/agent/qa/scripts/start.sh +++ b/agent/qa/scripts/start.sh @@ -28,6 +28,22 @@ if [[ -f "$AGENT_ROOT/env/active_library.env" ]]; then set -a; . "$AGENT_ROOT/env/active_library.env"; set +a fi + +# ---- 开机自愈:后端校验(Qdrant/Neo4j)失败即崩且零重试,启动前必须等依赖就绪(2026-09-23) ---- +QD_READYZ="${QDRANT_URL:-http://127.0.0.1:27433}/readyz" +QD_DEADLINE=$(( $(date +%s) + 1800 )) +until curl -sf --max-time 3 "$QD_READYZ" >/dev/null 2>&1; do + [ "$(date +%s)" -ge "$QD_DEADLINE" ] && { echo "[qa] Qdrant 30 分钟未就绪,放弃启动" >&2; exit 1; } + sleep 5 +done +N4_HP="${NEO4J_URI#bolt://}"; N4_H="${N4_HP%:*}"; N4_P="${N4_HP##*:}" +N4_DEADLINE=$(( $(date +%s) + 600 )) +until timeout 3 bash -c "echo > /dev/tcp/$N4_H/$N4_P" 2>/dev/null; do + [ "$(date +%s)" -ge "$N4_DEADLINE" ] && { echo "[qa] Neo4j 10 分钟未就绪,放弃启动" >&2; exit 1; } + sleep 5 +done +# ---- 就绪等待结束 ---- + # reranker 模型路径回退：env 未指定时用安装包内模型（install/models/bge-reranker-base） if [[ -z "${QA_V1_BGE_RERANKER_MODEL:-}" && -d "$AGENT_ROOT/../models/bge-reranker-base" ]]; then export QA_V1_BGE_RERANKER_MODEL="$AGENT_ROOT/../models/bge-reranker-base" diff --git a/agent/qa/src/rag/qa_v1_1_73/backend_guard.py b/agent/qa/src/rag/qa_v1_1_73/backend_guard.py index 4cc7579..7a9708b 100644 --- a/agent/qa/src/rag/qa_v1_1_73/backend_guard.py +++ b/agent/qa/src/rag/qa_v1_1_73/backend_guard.py @@ -131,7 +131,7 @@ def _validate_normalized_backends() -> Dict[str, Any]: finally: driver.close() - if labels != {"Entity", "Paper"}: + if not {"Entity", "Paper"}.issubset(labels): raise BackendIdentityError( f"normalized Neo4j labels mismatch: {sorted(labels)}" ) diff --git a/bioflow_service/app/routers/biojobs.py b/bioflow_service/app/routers/biojobs.py index 0bb8f6c..d5d1e4c 100644 --- a/bioflow_service/app/routers/biojobs.py +++ b/bioflow_service/app/routers/biojobs.py @@ -486,6 +486,46 @@ async def stage_manifest(job_id: str, stage: str) -> dict[str, Any]: }} +@router.get("/jobs/{job_id}/stages/{stage}/logs") +async def stage_logs( + job_id: str, + stage: str, + offset: int = Query(default=0, ge=0), + limit: int = Query(default=500, ge=1, le=2000), +) -> dict[str, Any]: + """阶段日志分页读取（offset=已读行数；执行中可读,前端轮询滚动 log）。 + + - kernel 流式 fsync 追加落盘,本端点按行切片返回增量; + - offset 超尾 → 空 lines + eof:true; + - 返回 total_bytes 供 eof 判定(文件总行数随执行增长)。 + """ + record = await job_or_404(job_id) + _stage_exists(record, stage) + log_path = record.logs_dir / f"{stage}.log" + if not log_path.exists(): + return {"ok": True, "data": { + "job_id": job_id, "stage": stage, + "offset": offset, "limit": limit, "lines": [], + "total_bytes": 0, "eof": True, "log_tail": "", + }} + try: + text = log_path.read_text(encoding="utf-8", errors="replace") + except OSError: + return _err("INTERNAL", "日志读取失败", 500) + lines = text.splitlines() + total = len(lines) + page = lines[offset:offset + limit] + eof = offset + len(page) >= total + return {"ok": True, "data": { + "job_id": job_id, "stage": stage, + "offset": offset, "limit": limit, + "lines": page, + "total_bytes": log_path.stat().st_size if log_path.exists() else 0, + "eof": eof, + "log_tail": "\n".join(lines[-get_settings().log_tail_lines:]) if total else "", + }} + + # --------------------------------------------------------------------------- # events SSE（FR-BIO-014 断线重连回放） # --------------------------------------------------------------------------- diff --git a/bioflow_service/app/services/bioflow/events.py b/bioflow_service/app/services/bioflow/events.py index fedf2af..31d8bfc 100644 --- a/bioflow_service/app/services/bioflow/events.py +++ b/bioflow_service/app/services/bioflow/events.py @@ -23,8 +23,9 @@ from typing import Any logger = logging.getLogger(__name__) EVENT_TYPES = ( - "job_created", "plan_ready", "stage_started", "stage_log", "stage_done", - "stage_failed", "stage_skipped", "stage_await_confirm", "stage_await_files", + "job_created", "plan_ready", "stage_started", "stage_log", "stage_progress", + "stage_done", "stage_failed", "stage_skipped", "stage_await_confirm", + "stage_await_files", "job_done", "job_paused", "job_archived", "job_recovered", # 用户自定义扩展分析事件帧（契约 §4；并行链路，不触碰现有 job_* 语义） "extension_created", "extension_build", "extension_trial", "extension_repair", @@ -76,9 +77,34 @@ class EventBus: self._subscribers: set[Subscriber] = set() self._sub_lock = asyncio.Lock() self._buffer: deque[dict[str, Any]] = deque(maxlen=MEMORY_BUFFER_LIMIT) - self._seq = 0 + # 跨重启 seq 单调递增（BUG-FIX 2026-09-17）：旧进程事件已持久化在 + # events/*.jsonl，重启后 _seq 归零会让新事件 seq 小于浏览器 last_seq + # 水位，前端回放被过滤、收不到 stage_await_files 等关键帧。 + # 启动时扫描历史日志取最大 seq 作为起始值。 + self._seq = self._scan_max_seq() self._file_lock = threading.Lock() + def _scan_max_seq(self) -> int: + """扫描 events_dir 下全部 jsonl，取历史最大 seq（无文件/坏行 → 0）。""" + max_seq = 0 + try: + for path in self.events_dir.glob("*.jsonl"): + try: + for line in path.read_text(encoding="utf-8").splitlines(): + if not line.strip(): + continue + try: + seq = int(json.loads(line).get("seq") or 0) + except (ValueError, TypeError): + continue + if seq > max_seq: + max_seq = seq + except OSError: + continue + except OSError: # pragma: no cover - 目录扫描失败(权限等)不影响启动 + return 0 + return max_seq + # ---- 内部 ---- def _events_file(self, job_id: str) -> Path: @@ -182,6 +208,14 @@ class EventBus: elif t == "stage_skipped": st = stage_states.setdefault(str(d.get("stage")), {"stage": str(d.get("stage"))}) st["status"] = "skipped" + elif t == "stage_progress": + st = stage_states.setdefault(str(d.get("stage")), {"stage": str(d.get("stage"))}) + pct = d.get("percent") + if isinstance(pct, (int, float)): + st["progress"] = max(float(st.get("progress") or 0.0), float(pct)) + msg = d.get("message") + if msg: + st["progressMessage"] = str(msg) elif t == "stage_await_confirm": st = stage_states.setdefault(str(d.get("stage")), {"stage": str(d.get("stage"))}) st["status"] = "waiting_confirm" diff --git a/bioflow_service/app/services/bioflow/kernel.py b/bioflow_service/app/services/bioflow/kernel.py index f7d5cc4..86d4cfa 100644 --- a/bioflow_service/app/services/bioflow/kernel.py +++ b/bioflow_service/app/services/bioflow/kernel.py @@ -19,14 +19,19 @@ import json import logging import os import shlex +import time from dataclasses import dataclass, field from pathlib import Path -from typing import Any +from typing import Any, Awaitable, Callable logger = logging.getLogger(__name__) CONTAINER_CMD_PREFIX = ("python", "-B") # 容器内命令形态（TC-BIO-RUN-037） +# 日志流式落盘/回调节流常数：每 N 行或每 S 秒 flush+fsync 并推送一次 +LOG_FLUSH_LINES = 20 +LOG_FLUSH_SECS = 1.0 + class ImageMapError(Exception): """镜像映射表缺失/非法（服务可用但所有 docker 执行前置拦截）。""" @@ -265,6 +270,19 @@ class Kernel: # ---- 执行 ---- + @staticmethod + async def _terminate_then_kill(proc: asyncio.subprocess.Process) -> None: + """terminate → 3s → kill（TC-BIO-KERN-009；与 extension_exec.run_stream 同构）。""" + try: + proc.terminate() + try: + await asyncio.wait_for(proc.wait(), timeout=3.0) + except asyncio.TimeoutError: + proc.kill() + await proc.wait() + except ProcessLookupError: # pragma: no cover - 进程已退出 + pass + async def run_stage( self, stage: str, @@ -277,13 +295,17 @@ class Kernel: memory: str | None = None, extra_mounts: list[str] | None = None, container_name: str | None = None, + on_log: Callable[[str], Awaitable[None]] | None = None, ) -> dict[str, Any]: """执行一个阶段 docker：返回 {exit_code, timed_out, log_tail, cmd}。 - extra_mounts：本次执行额外只读挂载的宿主绝对路径（登记路径,src==dst） - container_name：docker run --name(暂停时按名 kill 容器用) - - 超时：terminate → 3s → kill（TC-BIO-KERN-009） - - 日志全量写 log_path（任务 logs/），返回尾部 log_tail_lines 行 + - on_log：增量日志回调(每 LOG_FLUSH_LINES 行或 LOG_FLUSH_SECS 秒推送 + 一次 chunk,阶段运行期间实时可见；尾块兜底再推一次)。None 时不回调。 + - 超时：单行等待以 stage_timeout_sec 为限(与旧 communicate 语义一致, + 保守方案:长时间无输出不误杀)→ terminate → 3s → kill(TC-BIO-KERN-009) + - 日志流式追加写 log_path(任务 logs/,执行中可读),返回尾部 log_tail_lines 行 - 退出码非 0 / 超时 → 失败语义由 runner 判定（本函数只报事实） """ params = params or {} @@ -302,25 +324,53 @@ class Kernel: stderr=asyncio.subprocess.STDOUT, ) timed_out = False + text_parts: list[str] = [] + # 追加模式落盘：阶段执行中前端 logs 端点即可读；fsync 保证跨进程可见 + fh = log_path.open("a", encoding="utf-8") + flush_buf: list[str] = [] + last_flush = time.monotonic() try: - stdout, _ = await asyncio.wait_for( - proc.communicate(), timeout=self.cfg.stage_timeout_sec - ) - except asyncio.TimeoutError: - timed_out = True - try: - proc.terminate() + while True: try: - await asyncio.wait_for(proc.wait(), timeout=3.0) + line = await asyncio.wait_for( + proc.stdout.readline(), timeout=self.cfg.stage_timeout_sec + ) except asyncio.TimeoutError: - proc.kill() - await proc.wait() - except ProcessLookupError: - pass - stdout = b"" + timed_out = True + await self._terminate_then_kill(proc) + break + if not line: + break + decoded = line.decode("utf-8", errors="replace") + text_parts.append(decoded) + fh.write(decoded) + flush_buf.append(decoded) + now = time.monotonic() + if len(flush_buf) >= LOG_FLUSH_LINES or now - last_flush >= LOG_FLUSH_SECS: + fh.flush() + os.fsync(fh.fileno()) + if on_log is not None: + await on_log("".join(flush_buf)) + flush_buf = [] + last_flush = now + finally: + # 尾块兜底 + 进程收尾(正常 EOF 或超时 kill 都落此处) + if flush_buf: + try: + fh.flush() + os.fsync(fh.fileno()) + except OSError: # pragma: no cover - fd 异常不阻塞结果 + pass + if on_log is not None: + try: + await on_log("".join(flush_buf)) + except Exception: # pragma: no cover - 回调异常不吞结果 + logger.exception("bioflow kernel: on_log 尾块回调失败") + fh.close() + if proc.returncode is None: + await proc.wait() exit_code = proc.returncode - text = (stdout or b"").decode("utf-8", errors="replace") - log_path.write_text(text, encoding="utf-8") + text = "".join(text_parts) return { "exit_code": exit_code, "timed_out": timed_out, diff --git a/bioflow_service/app/services/bioflow/manifests.py b/bioflow_service/app/services/bioflow/manifests.py index cb529bc..7d36c94 100644 --- a/bioflow_service/app/services/bioflow/manifests.py +++ b/bioflow_service/app/services/bioflow/manifests.py @@ -60,10 +60,18 @@ class ManifestError(Exception): def locate_manifest(workspace: str | Path, stage: str, prefix: str | None = None) -> Path | None: """在 workspace final/ 下定位阶段 manifest（prefix + 阶段名双匹配）。 + 匹配优先级： + 1. prefix 匹配 + 阶段名中缀：final/{prefix}.{stage}.manifest.json + 2. 阶段名中缀：final/{任意}.{stage}.manifest.json + 3. 无中缀名兜底（BUG-FIX 2026-09-18）：final/{stage}.manifest.json + （rnaseq 系 CLI 空 prefix 时产出的文件名形态，如 rnaseq_qc.manifest.json； + 此前无此分支 → 退出码 0 也被判 no_manifest）。 + 不存在返回 None（runner 据此判 reason=no_manifest）。 - 注意：无兜底——其他阶段的 manifest 绝不误配给当前阶段 - （重启恢复按 manifest 判成败时曾因兜底把 identify 的 manifest 判给 - phylogeny，导致 running 阶段被误判 done 不重跑）。 + 注意：兜底只按「文件名精确等于 {stage}.manifest.json」匹配——其他阶段的 + manifest 绝不误配给当前阶段（重启恢复按 manifest 判成败时曾因任意 + *.manifest.json 兜底把 identify 的 manifest 判给 phylogeny，导致 running + 阶段被误判 done 不重跑）。 """ final_dir = Path(workspace) / "final" if not final_dir.is_dir(): @@ -78,6 +86,9 @@ def locate_manifest(workspace: str | Path, stage: str, prefix: str | None = None for cand in candidates: if f".{stage}." in cand.name: return cand + for cand in candidates: + if cand.name == f"{stage}.manifest.json": + return cand return None diff --git a/bioflow_service/app/services/bioflow/runner.py b/bioflow_service/app/services/bioflow/runner.py index bcfdaf4..dc3a5e4 100644 --- a/bioflow_service/app/services/bioflow/runner.py +++ b/bioflow_service/app/services/bioflow/runner.py @@ -19,6 +19,8 @@ from __future__ import annotations import asyncio import logging +import re +import time import uuid from dataclasses import dataclass, field from pathlib import Path @@ -200,8 +202,33 @@ def ensure_samples_yaml(job_input_dir: Path, source_roots: list[Path] | None = N yield from root.rglob("*") # ---- 2. 实况扫描:源目录(真实文件)+ 任务 input(跳过 symlink 重复) ---- + # 同文件去重:符号链接与其目标(samefile)只留一个 —— 优先保留真实文件(非链接), + # 避免链接的别名文件名(如 CRR809753_2.fq.gz → _r2.fq.gz)被当成第二个样本。 seq_files: list[Path] = [] archives: list[Path] = [] + seen_inodes: set[tuple[int, int]] = set() + + def _seq_dedup(p: Path, pool: list[Path]) -> bool: + try: + key = (p.stat().st_dev, p.stat().st_ino) + except OSError: + key = None + if key is None or key not in seen_inodes: + if key is not None: + seen_inodes.add(key) + pool.append(p) + return True + # 已存在同 inode 文件:若新候选是真实文件、旧的是链接,用真实文件替换 + for i, q in enumerate(pool): + if q.samefile(p): + if p.is_symlink() and not q.is_symlink(): + return False + if not p.is_symlink() and q.is_symlink(): + pool[i] = p + return True + return False + return False + for root in roots: for p in sorted(_walk(root)): if not p.is_file(): @@ -209,10 +236,9 @@ def ensure_samples_yaml(job_input_dir: Path, source_roots: list[Path] | None = N if p.is_symlink() and root == job_input_dir: continue # input 里是登记 symlink,源目录已有真实文件 if _is_seq(p): - seq_files.append(p) + _seq_dedup(p, seq_files) elif p.name.lower().endswith((".tar.gz", ".tar", ".zip", ".tgz")): - archives.append(p) - # 去重(源目录与 input 可能重叠) + _seq_dedup(p, archives) seq_files = sorted({str(p): p for p in seq_files}.values(), key=lambda p: str(p)) archives = sorted({str(p): p for p in archives}.values(), key=lambda p: str(p)) @@ -307,19 +333,47 @@ def ensure_samples_yaml(job_input_dir: Path, source_roots: list[Path] | None = N if p.name not in known: described.append({"name": p.stem, "run": "", "r1": str(p), "r2": "", "meta": {}}) + def _pair_key(name: str) -> str: + """归一配对比较键(小写):_f1/_r2/_1/_2 别名归一到同一样本。""" + return _re.sub( + r"[._-](f1|r1|1|forward|f2|r2|2|reverse)([._-]|$).*", "", name.lower() + ) + + def _pair_base(name: str) -> str: + """保留原大小写的样本名:剥离 R1/R2 语义后缀。""" + return _re.sub( + r"[._-](f1|r1|1|forward|f2|r2|2|reverse)([._-]|$).*", "", name, + flags=_re.IGNORECASE, + ) + # ---- 无描述文件:实况配对(不假设命名规律,按 R1/R2 语义) ---- if not described and seq_files: r1s = [p for p in seq_files if _r1_sem(p.name)] r2s = [p for p in seq_files if _r2_sem(p.name)] + used_r2: set[str] = set() # 一对一配对:已被认领的 r2 不再参与 for p in r1s: - base = _re.sub(r"[._-](f1|r1|1|forward)(\.|$).*", "", p.name) - mates = [q for q in r2s if q.name.startswith(base)] + base = _pair_key(p.name) + mate_cands = [ + q for q in r2s if _pair_key(q.name) == base and str(q) not in used_r2 + ] + mate: Path | None = None + if len(mate_cands) == 1: + mate = mate_cands[0] + elif len(mate_cands) > 1: + # 多个同名 stem 候选:优先真实 f2/r2/reverse 命名,纯数字 _2 别名靠后 + pref = [ + q for q in mate_cands + if _re.search(r"[._-](f2|r2|reverse)([._-]|$)", q.name.lower()) + ] + mate = pref[0] if pref else mate_cands[0] described.append({ - "name": base or p.stem, "run": "", - "r1": str(p), "r2": str(mates[0]) if mates else "", "meta": {}, + "name": _pair_base(p.name) or p.stem, "run": "", + "r1": str(p), "r2": str(mate) if mate else "", "meta": {}, }) + if mate is not None: + used_r2.add(str(mate)) for p in r2s: - if not any(d["r2"] == str(p) for d in described): + if str(p) not in used_r2: described.append({"name": p.stem, "run": "", "r1": "", "r2": str(p), "meta": {}}) for p in seq_files: if not _r1_sem(p.name) and not _r2_sem(p.name): @@ -522,6 +576,41 @@ async def llm_sample_fallback(job_input_dir: Path, source_roots: list[Path]) -> ExecutorFn = Callable[..., Awaitable[dict[str, Any]]] +# 增量日志事件节流：每 N 行或每 S 秒发布一次 stage_log(增量) +LOG_EMIT_LINES = 20 +LOG_EMIT_SECS = 0.5 +# 进度事件节流(秒)：stage_progress 最低发布间隔 +PROGRESS_THROTTLE_SECS = 2.0 +# 进度解析白名单：只有这些阶段的 CLI 有 [PROGRESS] 标记(其余阶段不编造数字) +PROGRESS_STAGES = ("rnaseq_qc", "rnaseq_sample_qc") + + +# 阶段进度解析：CLI 日志中的 [PROGRESS] 标记 → 进度增量信息。 +# 保守策略：只识别 rnaseq 系 CLI 的既有标记，非白名单阶段不编造数字。 +_PROGRESS_SAMPLE_RE = re.compile( + r"\[PROGRESS\]\s*Processing:\s*(.+?)\s*$" +) +_PROGRESS_COPY_RE = re.compile( + r"\[PROGRESS\]\s*Copying\s+results" +) + + +def parse_progress_marker(line: str) -> dict[str, Any] | None: + """解析单行日志中的 [PROGRESS] 标记（纯函数，可单测）。 + + - `[PROGRESS] Processing: X` → {"kind": "sample", "message": "X"} + - `[PROGRESS] Copying results ...` → {"kind": "copy", "percent": 95} + - 其余行 → None + """ + if "[PROGRESS]" not in line: + return None + m = _PROGRESS_SAMPLE_RE.search(line) + if m: + return {"kind": "sample", "message": m.group(1).strip()} + if _PROGRESS_COPY_RE.search(line): + return {"kind": "copy", "percent": 95} + return None + @dataclass class JobRecord: @@ -776,11 +865,64 @@ class Runner: }) attempt_no = int(record._stage_states.get(stage, {}).get("attempt") or 0) + 1 record._running_container = f"plantai-bio-{record.id}-{stage}-{attempt_no}" + + # 进度/增量日志状态(每次 attempt 重置) + emit_buf: list[str] = [] + emit_last = time.monotonic() + prog_state: dict[str, Any] = {"seen": 0, "total": 0, "last": 0.0} + track_progress = stage in PROGRESS_STAGES + total_re = re.compile(r"Processing\s+(\d+)\s+samples") + valid_re = re.compile(r"Valid\s+samples:\s*(\d+)") + + async def _on_log(chunk: str) -> None: + nonlocal emit_last + emit_buf.append(chunk) + now = time.monotonic() + if len(emit_buf) >= LOG_EMIT_LINES or now - emit_last >= LOG_EMIT_SECS: + self.events.publish("stage_log", { + "job_id": record.id, "stage": stage, "log": "".join(emit_buf), + }) + emit_buf.clear() + emit_last = now + # [PROGRESS] 解析(白名单内才跟踪) + if track_progress: + for line in chunk.splitlines(): + marker = parse_progress_marker(line) + if not marker: + continue + if marker["kind"] == "copy": + prog_state["copying"] = True + elif marker["kind"] == "sample": + prog_state["seen"] += 1 + prog_state["sample"] = marker["message"] + tm = total_re.search(chunk) + if tm: + prog_state["total"] = int(tm.group(1)) + vm = valid_re.search(chunk) + if vm: + prog_state["total"] = int(vm.group(1)) + if (prog_state["seen"] and prog_state["total"] + and now - prog_state["last"] >= PROGRESS_THROTTLE_SECS): + prog_state["last"] = now + percent = min(int(prog_state["seen"] * 100 / prog_state["total"]), 99) + self.events.publish("stage_progress", { + "job_id": record.id, "stage": stage, + "percent": percent, + "message": str(prog_state.get("sample") or ""), + "sample": str(prog_state.get("sample") or ""), + }) + result = await self.executor(stage, record.input_dir, record.workspace_dir, params, log_path=record.logs_dir / f"{stage}.log", extra_mounts=self._registered_mounts(record), - container_name=record._running_container) + container_name=record._running_container, + on_log=_on_log) record._running_container = None + # 尾块兜底(增量 stage_log 与日志文件对齐) + if emit_buf: + self.events.publish("stage_log", { + "job_id": record.id, "stage": stage, "log": "".join(emit_buf), + }) log_tail = str(result.get("log_tail") or "") self._set_stage(record, stage, log_tail=log_tail) await self._persist_stage(record, stage) @@ -850,6 +992,11 @@ class Runner: self._set_stage(record, stage, status=STAGE_DONE, manifest_path=manifest.path, finished_at=_now_iso()) await self._persist_stage(record, stage) + # 终态强制 100%(白名单阶段;非白名单阶段前端会因 done 状态显示完成) + if stage in PROGRESS_STAGES: + self.events.publish("stage_progress", { + "job_id": record.id, "stage": stage, "percent": 100, "message": "", + }) self.events.publish("stage_done", { "job_id": record.id, "stage": stage, "exit_code": exit_code, "manifest_ref": manifest.path, diff --git a/bioflow_service/tests/conftest.py b/bioflow_service/tests/conftest.py index 0aa65a8..59b32f1 100644 --- a/bioflow_service/tests/conftest.py +++ b/bioflow_service/tests/conftest.py @@ -57,6 +57,7 @@ STAGE_MAP_JSON = { "note": "跨环境: 主 rnaR; rnaseq_quant 矩阵子步骤拆 bioflow-rnaseq", }, "rnaseq_quant": {"image": "bioflow-rnaseq", "module": "cli.run_rnaseq_quant", "executable": True}, + "rnaseq_qc": {"image": "bioflow-rnaseq", "module": "cli.run_rnaseq_qc", "executable": True}, "synteny_plot_revise": {"image": "bioflow-gf", "module": None, "executable": True, "note": "入口脚本缺失"}, }, diff --git a/bioflow_service/tests/test_api.py b/bioflow_service/tests/test_api.py index a4dc69f..46f3ece 100644 --- a/bioflow_service/tests/test_api.py +++ b/bioflow_service/tests/test_api.py @@ -660,6 +660,58 @@ def test_manifest_unknown_stage(api): assert m.status_code == 404 +def test_stage_logs_pagination(api, api_context): + """TC-BIO-API-102：logs 分页端点(offset/limit 切片、eof、log_tail、超尾)。""" + api_context.executor.log = "line0\nline1\nline2\nline3\n" + resp = _create_job(api, auto_continue=True) + job_id = _job_id(resp) + + def done(): + return _job_stage_status(api, job_id) == "done" + _wait_until(done, timeout=5) + + r1 = api.get(f"/api/bioflow/jobs/{job_id}/stages/identify/logs?offset=0&limit=1") + assert r1.status_code == 200 + d1 = r1.json()["data"] + assert d1["lines"] == ["line0"] + assert d1["eof"] is False + # 增量衔接:offset=1&limit=2 + r2 = api.get(f"/api/bioflow/jobs/{job_id}/stages/identify/logs?offset=1&limit=2") + d2 = r2.json()["data"] + assert d2["lines"] == ["line1", "line2"] + assert d2["eof"] is False + # 到尾:offset=2&limit=500 → eof:true + r3 = api.get(f"/api/bioflow/jobs/{job_id}/stages/identify/logs?offset=2&limit=500") + d3 = r3.json()["data"] + assert d3["lines"] == ["line2", "line3"] + assert d3["eof"] is True + # 超尾:offset=1000 → 空 lines + eof:true + log_tail 仍在 + r4 = api.get(f"/api/bioflow/jobs/{job_id}/stages/identify/logs?offset=1000&limit=500") + d4 = r4.json()["data"] + assert d4["lines"] == [] + assert d4["eof"] is True + assert d4["log_tail"] + + +def test_stage_logs_no_file_and_404(api): + """TC-BIO-API-103：log 文件不存在 → 空 lines + eof;未知 job/阶段 404。""" + resp = _create_job(api) # 手动模式:identify 停在 waiting_confirm,从未执行 + job_id = _job_id(resp) + + def waiting(): + return _job_stage_status(api, job_id) == "waiting_confirm" + _wait_until(waiting, timeout=5) + + r = api.get(f"/api/bioflow/jobs/{job_id}/stages/identify/logs") + assert r.status_code == 200 + d = r.json()["data"] + assert d["lines"] == [] and d["eof"] is True and d["total_bytes"] == 0 + # 未知 job 404 + assert api.get("/api/bioflow/jobs/nopejob/stages/identify/logs").status_code == 404 + # 未知阶段 404 + assert api.get(f"/api/bioflow/jobs/{job_id}/stages/nope/logs").status_code == 404 + + def test_rerun_and_skip_on_failed(api, api_context): """TC-BIO-API-062/065：rerun/skip 失败阶段。""" api_context.executor.exit_code = 1 diff --git a/bioflow_service/tests/test_events.py b/bioflow_service/tests/test_events.py index 9c29f3f..f133c37 100644 --- a/bioflow_service/tests/test_events.py +++ b/bioflow_service/tests/test_events.py @@ -47,6 +47,24 @@ def test_event_types_complete(): "job_done", "job_paused", "job_archived", } assert expected <= set(EVENT_TYPES) + assert "stage_progress" in EVENT_TYPES + + +def test_snapshot_restores_progress(tmp_path): + """TC-BIO-SSE-025：stage_progress 进快照推导(断线重连恢复进度条)。""" + bus = EventBus(tmp_path / "events") + bus.publish("job_created", {"job_id": "j1", "stages": [ + {"stage": "rnaseq_qc", "status": "waiting"}, + ]}) + bus.publish("stage_started", {"job_id": "j1", "stage": "rnaseq_qc"}) + bus.publish("stage_progress", {"job_id": "j1", "stage": "rnaseq_qc", + "percent": 33, "message": "A"}) + bus.publish("stage_progress", {"job_id": "j1", "stage": "rnaseq_qc", + "percent": 66, "message": "B"}) + snap = bus.snapshot("j1") + st = {s["stage"]: s for s in snap["stages"]}["rnaseq_qc"] + assert st["progress"] == 66.0 + assert st["progressMessage"] == "B" def test_subscribe_broadcast(tmp_path): @@ -125,3 +143,52 @@ def test_memory_replay_after_seq(tmp_path): bus.publish("stage_done", {"job_id": "j1", "stage": "a"}) replay = bus.memory_replay(after_seq=e1["seq"]) assert [e["type"] for e in replay] == ["stage_done"] + + +def test_seq_persists_across_restart(tmp_path): + """BUG-FIX 2026-09-17 回归锁定:重启后新事件 seq 从历史最大继续递增, + 不允许低于已持久化事件的 seq(否则前端 last_seq 水位过滤吞掉关键帧)。""" + events_dir = tmp_path / "events" + bus1 = EventBus(events_dir) + e1 = bus1.publish("job_created", {"job_id": "j1", "stages": []}) + e2 = bus1.publish("stage_await_files", {"job_id": "j1", "stage": "a"}) + assert e2["seq"] == e1["seq"] + 1 + + # 模拟重启:同目录新建 EventBus + bus2 = EventBus(events_dir) + assert bus2._seq == e2["seq"] + e3 = bus2.publish("stage_started", {"job_id": "j1", "stage": "a"}) + assert e3["seq"] == e2["seq"] + 1 + + +def test_seq_scan_empty_and_bad_lines(tmp_path): + """空目录 → 0;含坏行的 jsonl 只跳过坏行不中断扫描。""" + bus = EventBus(tmp_path / "empty") + assert bus._seq == 0 + + events_dir = tmp_path / "bad" + events_dir.mkdir() + (events_dir / "j1.jsonl").write_text( + json.dumps({"type": "job_created", "seq": 42, "data": {"job_id": "j1"}}) + "\n" + + "not-a-json-line\n" + + json.dumps({"type": "job_done", "seq": 41, "data": {"job_id": "j1"}}) + "\n", + encoding="utf-8", + ) + bus2 = EventBus(events_dir) + assert bus2._seq == 42 + + +def test_seq_scan_multiple_files_max(tmp_path): + """多个 job jsonl 取全局最大 seq。""" + events_dir = tmp_path / "multi" + events_dir.mkdir() + (events_dir / "a.jsonl").write_text( + json.dumps({"type": "job_created", "seq": 7, "data": {"job_id": "a"}}) + "\n", + encoding="utf-8", + ) + (events_dir / "b.jsonl").write_text( + json.dumps({"type": "job_created", "seq": 17, "data": {"job_id": "b"}}) + "\n", + encoding="utf-8", + ) + bus = EventBus(events_dir) + assert bus._seq == 17 diff --git a/bioflow_service/tests/test_kernel.py b/bioflow_service/tests/test_kernel.py index ec352ac..0f0b263 100644 --- a/bioflow_service/tests/test_kernel.py +++ b/bioflow_service/tests/test_kernel.py @@ -244,3 +244,92 @@ def test_run_stage_mock(tmp_path, kernel): "log_tail": tail_lines(proc.stdout, 80), "cmd": ["echo", "hello", "world"]} assert result["exit_code"] == 0 assert "hello world" in result["log_tail"] + + +# --------------------------------------------------------------------------- +# 流式执行（2026-09-17：readline 循环 + on_log 回调 + 追加落盘 + 超时路径） +# --------------------------------------------------------------------------- + + +class _FakeProc: + """最小 fake 子进程：逐行出 stdout；stall_after 行后静默（模拟卡死）。""" + + def __init__(self, lines, stall_after=None): + self._lines = [l.encode() for l in lines] + self._i = 0 + self.stall_after = stall_after + self.returncode = 0 + self.stdout = self._Stdout(self) + + class _Stdout: + def __init__(self, parent): + self._p = parent + + async def readline(self): + if self._p.stall_after is not None and self._p._i >= self._p.stall_after: + await asyncio.sleep(60) + return b"" + if self._p._i < len(self._p._lines): + line = self._p._lines[self._p._i] + self._p._i += 1 + return line + return b"" + + async def wait(self): + return 0 + + def terminate(self): + self.returncode = -15 + + def kill(self): + self.returncode = -9 + + +def _patch_exec(monkeypatch, proc): + async def fake_exec(*args, **kwargs): + return proc + monkeypatch.setattr("asyncio.create_subprocess_exec", fake_exec) + + +def test_run_stage_streaming_chunks(tmp_path, kernel, monkeypatch): + """TC-BIO-PROG-K01：25 行输出 → 20+5 两个 on_log chunk；日志逐行追加落盘。""" + _patch_exec(monkeypatch, _FakeProc([f"L{i}\n" for i in range(25)])) + chunks: list[str] = [] + + async def on_log(text): + chunks.append(text) + + async def go(): + return await kernel.run_stage("identify", tmp_path / "in", tmp_path / "ws", + log_path=tmp_path / "s.log", on_log=on_log) + + result = asyncio.new_event_loop().run_until_complete(go()) + assert result["exit_code"] == 0 + assert not result["timed_out"] + assert len(chunks) == 2 + assert chunks[0].count("\n") == 20 + assert chunks[1].count("\n") == 5 + # 25 行 < log_tail_lines(80) → log_tail 保留全部 + assert result["log_tail"].splitlines() == [f"L{i}" for i in range(25)] + written = (tmp_path / "s.log").read_text(encoding="utf-8") + assert written.count("\n") == 25 + + +def test_run_stage_timeout_path(tmp_path, kernel, monkeypatch): + """TC-BIO-PROG-K02：行间隔超时 → timed_out=True,超时前日志已落盘。""" + kernel.cfg.stage_timeout_sec = 0.3 # 加速测试 + _patch_exec(monkeypatch, _FakeProc(["a\n", "b\n"], stall_after=2)) + chunks: list[str] = [] + + async def on_log(text): + chunks.append(text) + + async def go(): + return await kernel.run_stage("identify", tmp_path / "in", tmp_path / "ws", + log_path=tmp_path / "t.log", on_log=on_log) + + result = asyncio.new_event_loop().run_until_complete(go()) + assert result["timed_out"] is True + assert result["exit_code"] == -15 + assert "a\nb\n" in (tmp_path / "t.log").read_text(encoding="utf-8") + assert chunks # 超时前的增量已回调 diff --git a/bioflow_service/tests/test_manifests.py b/bioflow_service/tests/test_manifests.py index 24aec60..9de6545 100644 --- a/bioflow_service/tests/test_manifests.py +++ b/bioflow_service/tests/test_manifests.py @@ -86,6 +86,67 @@ def test_locate_manifest_strict_no_fallback(tmp_path): assert locate_manifest(tmp_path, "identify") == path +def test_locate_manifest_no_infix_name(tmp_path): + """BUG-FIX 2026-09-18：无中缀名兜底 final/{stage}.manifest.json + （rnaseq 系 CLI 空 prefix 形态,如 rnaseq_qc.manifest.json）。""" + final = tmp_path / "final" + final.mkdir(parents=True, exist_ok=True) + raw = { + "manifest_version": "1.0", "job_id": "unknown", "stage": "rnaseq_qc", + "status": "success", "prefix": "", + "artifacts": {"clean": {"full_path": str(final / "x.clean.fastq.gz")}}, + "parameters": {}, "metrics": {}, + } + path = final / "rnaseq_qc.manifest.json" + path.write_text(json.dumps(raw), encoding="utf-8") + + assert locate_manifest(tmp_path, "rnaseq_qc") == path + assert locate_manifest(tmp_path, "rnaseq_qc", prefix="") == path + m = parse_manifest(tmp_path, "rnaseq_qc") + assert m.status == "success" + assert m.stage == "rnaseq_qc" + + +def test_locate_manifest_no_cross_stage_misattribution(tmp_path): + """无中缀兜底不跨阶段误配：stage=rnaseq 不命中 rnaseq_qc.manifest.json; + stage=rnaseq_qc 也不命中他阶段带 prefix 的 manifest。""" + final = tmp_path / "final" + final.mkdir(parents=True, exist_ok=True) + _write_manifest(tmp_path, "rnaseq_qc") # Ft.SPL.rnaseq_qc.manifest.json + # 加一个裸名 rnaseq_qc.manifest.json 模拟 rnaseq 系实际产出 + raw = { + "manifest_version": "1.0", "job_id": "j1", "stage": "rnaseq_qc", + "status": "success", "prefix": "", + "artifacts": {}, "parameters": {}, "metrics": {}, + } + (final / "rnaseq_qc.manifest.json").write_text(json.dumps(raw), encoding="utf-8") + + # stage=rnaseq 绝不命中 rnaseq_qc.manifest.json（精确等名,无前缀混淆） + assert locate_manifest(tmp_path, "rnaseq") is None + # 中缀匹配优先于裸名兜底（有 prefix 中缀文件时先命中） + qc = locate_manifest(tmp_path, "rnaseq_qc") + assert qc is not None + assert qc.name == "Ft.SPL.rnaseq_qc.manifest.json" + + +def test_locate_manifest_bare_name_priority(tmp_path): + """只有裸名文件时兜底命中;同目录存在其他阶段 manifest 不干扰。""" + final = tmp_path / "final" + final.mkdir(parents=True, exist_ok=True) + _write_manifest(tmp_path, "identify") # Ft.SPL.identify.manifest.json + raw = { + "manifest_version": "1.0", "job_id": "j1", "stage": "rnaseq_qc", + "status": "success", "prefix": "", + "artifacts": {}, "parameters": {}, "metrics": {}, + } + bare = final / "rnaseq_qc.manifest.json" + bare.write_text(json.dumps(raw), encoding="utf-8") + + assert locate_manifest(tmp_path, "rnaseq_qc") == bare + # identify 照常中缀命中,不受裸名文件影响 + assert locate_manifest(tmp_path, "identify").name == "Ft.SPL.identify.manifest.json" + + def test_artifact_paths(): artifacts = { "a": {"full_path": "/x/a.id", "type": "table"}, diff --git a/bioflow_service/tests/test_runner.py b/bioflow_service/tests/test_runner.py index 17afaad..9682d1b 100644 --- a/bioflow_service/tests/test_runner.py +++ b/bioflow_service/tests/test_runner.py @@ -720,3 +720,206 @@ def test_event_seq_strictly_increasing(db, events, catalog, fake_executor): seqs = [e["seq"] for e in _collected(events, record.id)] assert seqs == sorted(seqs) assert len(set(seqs)) == len(seqs) + + +# --------------------------------------------------------------------------- +# ensure_samples_yaml 配对修复（2026-09-17 水稻 M34 真实缺陷回归） +# --------------------------------------------------------------------------- + +def _mk_fq(dirp, name): + p = dirp / name + p.write_text(">seq\nACGT\n") + return p + + +def test_samples_pairing_symlink_alias_dedup(tmp_path): + """TC-BIO-FIX-01：目录含 f1/r2 真实文件 + _1/_2 符号链接别名时， + 只产出 N 个配对样本，无 _r2.fq 孤儿条目，无 _2 别名抢配。""" + from app.services.bioflow.runner import ensure_samples_yaml + + src = tmp_path / "fastq" + src.mkdir() + for sid in ("CRR1", "CRR2", "CRR3"): + sub = src / sid + sub.mkdir() + real_r1 = _mk_fq(sub, f"{sid}_f1.fq.gz") + real_r2 = _mk_fq(sub, f"{sid}_r2.fq.gz") + # 别名符号链接（指向真实文件；模拟 M34 水稻目录实况） + (sub / f"{sid}_1.fq.gz").symlink_to(real_r1) + (sub / f"{sid}_2.fq.gz").symlink_to(real_r2) + + job_in = tmp_path / "job" / "input" + job_in.mkdir(parents=True) + mounts = ensure_samples_yaml(job_in, source_roots=[src]) + + import yaml + samples = yaml.safe_load((job_in / "samples.yaml").read_text())["samples"] + assert len(samples) == 3, f"期望 3 个配对样本,实际 {list(samples)}" + assert all(not n.endswith("_r2.fq") for n in samples) + for sid in ("CRR1", "CRR2", "CRR3"): + s = samples[sid] + assert s["r1"].endswith(f"{sid}_f1.fq.gz") + assert s["r2"].endswith(f"{sid}_r2.fq.gz") + assert mounts + + +def test_samples_pairing_real_r2_not_orphaned(tmp_path): + """TC-BIO-FIX-02：真实 _r2 与别名 _2 并存时，r2 必须取真实文件， + 且 _2 别名不产生孤儿样本条目。""" + from app.services.bioflow.runner import ensure_samples_yaml + + src = tmp_path / "fastq" + src.mkdir() + sub = src / "S1" + sub.mkdir() + real_r1 = _mk_fq(sub, "S1_f1.fq.gz") + real_r2 = _mk_fq(sub, "S1_r2.fq.gz") + (sub / "S1_1.fq.gz").symlink_to(real_r1) + (sub / "S1_2.fq.gz").symlink_to(real_r2) + + job_in = tmp_path / "job" / "input" + job_in.mkdir(parents=True) + ensure_samples_yaml(job_in, source_roots=[src]) + + import yaml + samples = yaml.safe_load((job_in / "samples.yaml").read_text())["samples"] + assert len(samples) == 1, f"期望 1 个配对样本,实际 {list(samples)}" + s = samples["S1"] + assert s["r1"].endswith("S1_f1.fq.gz") + assert s["r2"].endswith("S1_r2.fq.gz") + + +def test_samples_pairing_unpaired_r2_kept(tmp_path): + """TC-BIO-FIX-03：只有单端 r2（无 r1）时保留孤儿条目（原契约）。""" + from app.services.bioflow.runner import ensure_samples_yaml + + src = tmp_path / "fastq" + src.mkdir() + _mk_fq(src, "WX_alone_r2.fq.gz") + + job_in = tmp_path / "job" / "input" + job_in.mkdir(parents=True) + ensure_samples_yaml(job_in, source_roots=[src]) + + import yaml + samples = yaml.safe_load((job_in / "samples.yaml").read_text())["samples"] + assert len(samples) == 1 + s = next(iter(samples.values())) + assert not s.get("r1") + assert s["r2"].endswith("WX_alone_r2.fq.gz") + + +# --------------------------------------------------------------------------- +# 阶段进度条（2026-09-17：流式日志 + [PROGRESS] 标记解析 + stage_progress 事件） +# --------------------------------------------------------------------------- + + +def test_parse_progress_marker_table(): + """parse_progress_marker 表驱动:样本/拷贝标记识别,无关行 None。""" + from app.services.bioflow.runner import parse_progress_marker + + cases = [ + ("[PROGRESS] Processing: SRR1", {"kind": "sample", "message": "SRR1"}), + (" [PROGRESS] Processing: sample A ", {"kind": "sample", "message": "sample A"}), + ("[PROGRESS] Copying results to final directory...", + {"kind": "copy", "percent": 95}), + ("[PROGRESS] Copying results to final", {"kind": "copy", "percent": 95}), + ("[INFO] not progress", None), + ("Processing: SRR1", None), # 无标记前缀不识别 + ("", None), + ] + for line, expected in cases: + assert parse_progress_marker(line) == expected, f"line={line!r}" + + +class FakeStreamingExecutor: + """mock docker 流式执行器：记录 on_log 回调并在调用中分块推送进度标记。""" + + def __init__(self, chunks: list[str], exit_code: int = 0): + self.chunks = chunks + self.exit_code = exit_code + self.calls: list[dict] = [] + self.on_log = None + + async def __call__(self, stage, input_dir, workspace, params=None, + log_path=None, **kwargs): + self.calls.append({"stage": stage}) + self.on_log = kwargs.get("on_log") + for chunk in self.chunks: + if self.on_log is not None: + await self.on_log(chunk) + # 写成功 manifest(默认 FakeExecutor 语义:退出码 0 → success) + if self.exit_code == 0: + from conftest import FakeExecutor + FakeExecutor._write_manifest(self, workspace, stage, "success") + return {"exit_code": self.exit_code, "timed_out": False, + "log_tail": "".join(self.chunks), "cmd": []} + + +def test_stage_progress_events_rnaseq(db, events, catalog): + """TC-BIO-PROG-01：rnaseq_qc 白名单阶段收到 [PROGRESS] → 发布 + stage_progress(percent=seen/total,单调递增);成功后强制 100。""" + chunks = [ + "Input validation passed: 2 sample(s) ready\n", + "Processing 2 samples...\n", + "[PROGRESS] Processing: A\n", + "fastp done\n", + "[PROGRESS] Processing: B\n", + ] + executor = FakeStreamingExecutor(chunks) + + async def scenario(): + runner = make_runner(db, events, executor, catalog) + record = await runner.create_job( + "s1", "t", ["rnaseq_qc"], auto_continue=True, auto_start=False) + await runner.run_job(record) + return runner, record + + runner, record = _scenario(scenario) + assert runner.stage_status(record, "rnaseq_qc") == STAGE_DONE + prog = [e["data"] for e in _collected(events, record.id) + if e["type"] == "stage_progress"] + assert prog, "应发布 stage_progress 事件" + percents = [p["percent"] for p in prog] + assert percents[-1] == 100 # 终态强制 100 + assert percents[:-1] == sorted(percents[:-1]) # 中间值单调 + assert any(p["percent"] > 0 for p in prog) # 有实际进度(非只有 0/100) + assert prog[-1]["message"] == "" # 终态清空 message + + +def test_stage_progress_not_for_non_rnaseq(db, events, catalog): + """TC-BIO-PROG-02：非白名单阶段(identify)不发布 stage_progress。""" + chunks = ["Processing 5 samples...\n", "[PROGRESS] Processing: A\n"] + executor = FakeStreamingExecutor(chunks) + + async def scenario(): + runner = make_runner(db, events, executor, catalog) + record = await runner.create_job( + "s1", "t", ["identify"], auto_continue=True, auto_start=False) + await runner.run_job(record) + return runner, record + + runner, record = _scenario(scenario) + assert runner.stage_status(record, "identify") == STAGE_DONE + prog = [e for e in _collected(events, record.id) if e["type"] == "stage_progress"] + assert prog == [] + + +def test_incremental_stage_log_events(db, events, catalog): + """TC-BIO-PROG-03：on_log 分块 → 增量 stage_log 发布 + 终态 log_tail 并存。""" + executor = FakeStreamingExecutor(["chunk1\n", "chunk2\n"]) + + async def scenario(): + runner = make_runner(db, events, executor, catalog) + record = await runner.create_job( + "s1", "t", ["identify"], auto_continue=True, auto_start=False) + await runner.run_job(record) + return runner, record + + runner, record = _scenario(scenario) + log_events = [e["data"] for e in _collected(events, record.id) + if e["type"] == "stage_log"] + assert log_events + joined = "".join(e["log"] for e in log_events) + assert "chunk1" in joined and "chunk2" in joined + assert record._stage_states["identify"]["log_tail"] # 终态 log_tail 仍在 diff --git a/sdk/app/services/figures.py b/sdk/app/services/figures.py index 9a6e811..09310d5 100644 --- a/sdk/app/services/figures.py +++ b/sdk/app/services/figures.py @@ -1,10 +1,15 @@ """services/figures.py —— 论文图片解析（2026-08-28 图文并茂方案，BFF 侧）。 -双源策略： -1. 本地 MinerU 图片（优先）：methodology_upload_jobs 下 completed 任务的 - `*_content_list.json` 中 type=="image" 条目（img_path + caption + page_idx）。 - 经 paper_metadata_registry（pmcid → DOI 风格 paper_id）与 job 目录映射。 -2. PMC 外网兜底：Europe PMC fullTextXML 拿 figure 文件名 + caption； +三源策略（按优先级降级）： +1. 标准本地图库（默认主路径，2026-09-20 定为共通逻辑）：所有库的论文图片均按 + DOI 组织——/images/（软链/目录）指向 + /results//，content_list.json（或 _v2）里 + type=="image" 条目给出 img_path（相对 results 目录）+ image_caption + page_idx。 + 新库无需改代码，只需数据落盘符合此布局并配置 PLANTAI_FIGURES_ROOT。 + 懒加载 per-DOI：入口清单一次枚举，单个 DOI 首次命中才解析并缓存。 +2. 遗留本地图库（methodology_upload_jobs 体系，次级兜底）：job 目录 + csv + 反查 pmcid 的老布局，保留兼容，命中率低（csv 占位符导致历史 0 键）。 +3. PMC 外网兜底：Europe PMC fullTextXML 拿 figure 文件名 + caption； license 经 Europe PMC REST search；图片直链 pmc.ncbi.nlm.nih.gov。 （实测：PMC5488377 全链路 200；服务器可出外网。） @@ -12,6 +17,8 @@ """ from __future__ import annotations +import asyncio +import json as _json import logging import os import re @@ -24,8 +31,15 @@ import httpx logger = logging.getLogger(__name__) -# 本地 MinerU 图片根 / 元数据注册表路径（缺陷 #37/#38，2026-09-10 起 env 化： -# PLANTAI_FIGURES_MINERU_ROOT / PLANTAI_FIGURES_REGISTRY 覆盖，默认值保持 175 现路径） +# 标准本地图库根（默认主路径；env 可覆盖。旧键 PLANTAI_FIGURES_YZ_ROOT 保留兼容） +FIGURES_ROOT = Path( + os.environ.get( + "PLANTAI_FIGURES_ROOT", + os.environ.get("PLANTAI_FIGURES_YZ_ROOT", "/home/xzq/yz"), + ) +) +# 遗留本地图库（methodology_upload_jobs 体系，次级兜底）/ 元数据注册表 +# （缺陷 #37/#38，2026-09-10 起 env 化：PLANTAI_FIGURES_MINERU_ROOT / PLANTAI_FIGURES_REGISTRY） MINERU_ROOT = Path( os.environ.get( "PLANTAI_FIGURES_MINERU_ROOT", @@ -47,11 +61,15 @@ PMC_IMAGE_BASE = "https://pmc.ncbi.nlm.nih.gov/articles/instance/{num}/bin/{file PROXY_PATH = "/api/figures/proxy?pmcid={pmcid}&url={url}" _cache: dict[str, dict[str, Any]] = {} # 键(pmcid 或 doi 原串) -> {ts, figures, license} -_local_index: dict[str, list[dict]] | None = None # 双键: pmcid数字 与 doi 原串 -> 本地图片清单 +_local_entries: dict[str, Path] | None = None # DOI 下划线形态键 -> results 目录 +_local_figures_cache: dict[str, list[dict]] = {} +_registry_pmcid_doi: dict[str, str] | None = None +_legacy_index: dict[str, list[dict]] | None = None # 遗留双键: pmcid数字 与 doi 原串 -> 本地图片清单 EXTERNAL_TIMEOUT = 15.0 # 176 实测 EBI 2-8s 波动，5s 必超时（2026-09-12 S2 部署质检） CACHE_TTL = 3600.0 + def _pmcid_num(pmcid: str) -> str: """'PMC5488377' -> '5488377'（容错各种大小写）""" m = re.match(r"^PMC(\d+)$", pmcid.strip().upper()) @@ -77,15 +95,134 @@ def _doi_key(pmcid_param: str) -> str | None: # --------------------------------------------------------------------------- -# 本地索引（懒构建） +# 标准本地图库（默认主路径,懒加载） # --------------------------------------------------------------------------- -def _build_local_index() -> dict[str, list[dict]]: +def _doi_variants(key: str) -> list[str]: + """DOI 键两种形态(斜杠 / 与下划线 _),去重保序。""" + variants = [key, key.replace("/", "_"), key.replace("_", "/")] + return list(dict.fromkeys(variants)) + + +def _pmcid_to_doi_map() -> dict[str, str]: + """registry 的 PMCID -> DOI(斜杠形态)映射,懒构建;失败返回空(降级 PMC 兜底)。""" + global _registry_pmcid_doi + if _registry_pmcid_doi is not None: + return _registry_pmcid_doi + m: dict[str, str] = {} + try: + if REGISTRY_PATH.exists(): + with REGISTRY_PATH.open(encoding="utf-8") as f: + for line in f: + line = line.strip() + if not line: + continue + try: + rec = _json.loads(line) + except Exception: + continue + pmcid = (rec.get("pmcid") or "").strip().upper() + doi = _norm_doi_key(rec.get("doi") or "") + if pmcid and doi: + m[pmcid] = doi + except Exception: + pass + _registry_pmcid_doi = m + return m + + +def _scan_local_entries() -> dict[str, Path]: + """扫描标准图库 images 入口(仅列名,不跟软链不读文件),DOI 下划线形态 -> results 目录。 + + 新库按此布局落盘即自动生效,无需改代码: + /images/ -> /results// + """ + global _local_entries + if _local_entries is not None: + return _local_entries + entries: dict[str, Path] = {} + try: + img_root = FIGURES_ROOT / "images" + if img_root.is_dir(): + for entry in img_root.iterdir(): + results_dir = FIGURES_ROOT / "results" / entry.name + if results_dir.is_dir(): + entries[entry.name] = results_dir + except Exception: + pass + _local_entries = entries + return entries + + +def _local_figures(doi: str) -> list[dict]: + """按 DOI 从标准图库解析图片清单(缓存;content_list 缺图时返回空)。""" + if doi in _local_figures_cache: + return _local_figures_cache[doi] + figures: list[dict] = [] + try: + entries = _scan_local_entries() + for variant in _doi_variants(doi): + results_dir = entries.get(variant) + if not results_dir: + continue + candidates = [ + results_dir / f"{variant}_content_list.json", + results_dir / f"{variant}_content_list_v2.json", + ] + for cl in candidates: + if not cl.exists(): + continue + try: + data = _json.loads(cl.read_text(encoding="utf-8")) + except Exception: + continue + if not isinstance(data, list): + continue + for item in data: + if not isinstance(item, dict) or item.get("type") != "image": + continue + img_path = item.get("img_path") or "" + if not img_path: + continue + abs_path = results_dir / img_path + if not abs_path.exists(): + continue + caption = "" + caps = [str(c).strip() for c in (item.get("image_caption") or []) + if isinstance(c, str) and str(c).strip()] + if not caps: + caps = [str(c).strip() for c in (item.get("image_footnote") or []) + if isinstance(c, str) and str(c).strip()] + if caps: + caption = caps[0][:300] + figures.append({ + "url": str(abs_path), + "caption": caption, + "page_idx": item.get("page_idx", 0), + "source": "local", + "file": Path(img_path).name, + }) + if figures: + figures.sort(key=lambda x: x["page_idx"]) + break # content_list 命中即止(v2 仅兜底) + if figures: + break + except Exception as exc: + logger.warning("标准图库解析失败（%s）: %s", doi, exc) + _local_figures_cache[doi] = figures + return figures + + +# --------------------------------------------------------------------------- +# 遗留本地索引（methodology_upload_jobs 体系,次级兜底,懒构建） +# --------------------------------------------------------------------------- + +def _build_legacy_index() -> dict[str, list[dict]]: """扫描 methodology_upload_jobs 的 completed 任务，双键（pmcid数字 + doi 原串） -> 图片清单。无 registry 时跳过映射（返回空索引），不影响 PMC 兜底。""" - global _local_index - if _local_index is not None: - return _local_index + global _legacy_index + if _legacy_index is not None: + return _legacy_index idx: dict[str, list[dict]] = {} try: # registry: 双向映射 —— paper_id(DOI风格) <-> pmcid / doi @@ -97,7 +234,6 @@ def _build_local_index() -> dict[str, list[dict]]: line = line.strip() if not line: continue - import json as _json rec = _json.loads(line) pmcid = rec.get("pmcid") or "" paper = rec.get("paper_id") or "" @@ -107,7 +243,7 @@ def _build_local_index() -> dict[str, list[dict]]: if doi and paper: doi_to_paper[doi] = paper if not MINERU_ROOT.is_dir(): - _local_index = idx + _legacy_index = idx return idx for job_dir in MINERU_ROOT.iterdir(): if not job_dir.is_dir(): @@ -121,7 +257,6 @@ def _build_local_index() -> dict[str, list[dict]]: if not pmcid: continue figures: list[dict] = [] - import json as _json for cf in content_files: try: data = _json.loads(cf.read_text(encoding="utf-8")) @@ -157,8 +292,8 @@ def _build_local_index() -> dict[str, list[dict]]: if paper == paper_ref or paper_ref.startswith(paper.split("_")[0]): idx[doi] = figures except Exception as exc: # 索引构建失败不阻断服务 - logger.warning("本地图片索引构建失败（降级为纯 PMC）: %s", exc) - _local_index = idx + logger.warning("遗留本地图片索引构建失败（降级为纯 PMC）: %s", exc) + _legacy_index = idx return idx @@ -201,6 +336,13 @@ async def _epmc_figures_once(client: httpx.AsyncClient, pmcid: str) -> tuple[lis r']*content-type="image"[^>]*xlink:href="([^"]+\.(?:jpg|jpeg|png|gif))"', xml, ) + # 宽松兜底（BUG-FIX 2026-09-18）：部分文章（如 PMC11557432）的 graphic + # 标签无 content-type 属性，仅 position/orientation + xlink:href，被上面 + # 窄正则漏判为 0 图。无命中时用宽松正则补抓。 + if not files: + files = re.findall( + r']*xlink:href="([^"]+\.(?:jpg|jpeg|png|gif))"', xml + ) # caption 提取： 后的

文本

# 简化策略：全部

文本按顺序与 files 配对 caption_texts = re.findall(r"\s*

([^<]{10,400})

", xml, re.S) @@ -256,13 +398,24 @@ async def _epmc_figures_once(client: httpx.AsyncClient, pmcid: str) -> tuple[lis async def _epmc_figures(client: httpx.AsyncClient, pmcid: str) -> tuple[list[dict], str]: - """PMC 解析重试包装：2026-09-12 部署质检发现 176 出网到 EBI 冷连接 - 首次 TLS 握手 10s+ 超时、重连即秒回（实测 1.6-2.2s），失败重试一次。""" - for attempt in (1, 2): - figs, license_ = await _epmc_figures_once(client, pmcid) - if figs or attempt == 2: + """PMC 解析重试包装：176 出网到 EBI 间歇性 ConnectError（TLS 握手失败, + 异常消息常为空,且同一连接池连续失败——2026-09-18 PMC6689393 三连败复现）。 + + 修复:每次尝试用全新 client(避免池污染)+ 递增退避(0.5s/1s/2s), + 最多 4 次;空结果不缓存,下次请求自然重试。 + """ + for attempt, backoff in ((1, 0.0), (2, 0.5), (3, 1.0), (4, 2.0)): + if backoff: + await asyncio.sleep(backoff) + fresh = httpx.AsyncClient() + try: + figs, license_ = await _epmc_figures_once(fresh, pmcid) + finally: + await fresh.aclose() + if figs: return figs, license_ - logger.info("PMC 首次解析失败（%s）——重试一次（176 冷连接实测）", pmcid) + if attempt < 4: + logger.info("PMC 解析失败（%s，第 %d 次）——重试（176 冷连接/ConnectError 实测）", pmcid, attempt) return [], "" @@ -270,13 +423,36 @@ async def _epmc_figures(client: httpx.AsyncClient, pmcid: str) -> tuple[list[dic # 对外 API # --------------------------------------------------------------------------- +def _local_figures_for_key(pmcid: str, doi: str | None) -> list[dict]: + """纯本地解析(无任何外网调用):标准图库 + 遗留体系。 + + 供 sse_relay 的 doi 别名路径使用——本地未命中时由调用方改走 PMC 形态 + (单次外网调用),避免 DOI 形态外网兜底的两次调用拖爆预取预算。 + """ + if doi: + local = _local_figures(doi) + if local: + return local + else: + _local_doi = _pmcid_to_doi_map().get(pmcid.strip().upper()) or "" + if _local_doi: + local = _local_figures(_local_doi) + if local: + return local + legacy_idx = _build_legacy_index() + return legacy_idx.get(doi or _pmcid_num(pmcid)) or [] + + async def resolve_figures(pmcid: str, limit: int = 4) -> dict[str, Any]: """按 PMCID 或 DOI 形态入参返回 {pmcid, license, figures:[{url,caption,source}]}。 - - PMCID 形态（PMC123）：原路径不变 —— 本地双键索引 + PMC fullTextXML 兜底。 - - DOI 形态（DOI:10.xxx/... 或裸 10.xxx/...）：pmcid 参数填 DOI 原串传入， - 先查本地双键索引；未命中再经 EPMC search（DOI→PMCID）走原 PMC 兜底路径。 - 本地优先，PMC 兜底；缓存 TTL 1 小时。失败静默降级为空清单。""" + - 标准图库（默认主路径,2026-09-20 起共通）:DOI 形态直查;PMCID 形态经 + registry 反查 DOI 后直查。新库数据按 DOI 布局落盘 + PLANTAI_FIGURES_ROOT + 即自动生效,无需改代码。 + - 遗留体系(次级兜底):methodology_upload_jobs 双键索引(DOI 键 / pmcid 数字键)。 + - PMC 外网兜底(最后):PMCID 形态走 EPMC fullTextXML;DOI 形态先经 + EPMC search(DOI→PMCID)再走 EPMC fullTextXML。 + 缓存 TTL 1 小时;失败静默降级为空清单。""" doi = _doi_key(pmcid) cache_key = doi if doi else pmcid now = time.monotonic() @@ -287,9 +463,8 @@ async def resolve_figures(pmcid: str, limit: int = 4) -> dict[str, Any]: figures: list[dict] = [] license_ = "" - # 1. 本地优先（双键索引：DOI 形态查 doi 键，PMCID 形态查 pmcid 数字键） - idx = _build_local_index() - local = idx.get(doi or _pmcid_num(pmcid)) or [] + # 1. 本地优先(标准图库 + 遗留体系,纯本地无外网) + local = _local_figures_for_key(pmcid, doi) for f in local[:limit]: figures.append({ "url": PROXY_PATH.format(pmcid=pmcid, url=quote(f["url"], safe="")), @@ -313,7 +488,11 @@ async def resolve_figures(pmcid: str, limit: int = 4) -> dict[str, Any]: }) data = {"pmcid": pmcid, "license": license_, "figures": figures} - _cache[cache_key] = {"ts": now, "data": data} + # BUG-FIX 2026-09-18：空结果不缓存。176 冷连接偶发失败(超时/ConnectError) + # 时若缓存空清单,1 小时 TTL 内后续问答全部命中空缓存,图片永远注入不了; + # 175 出网快很少产生空结果,故未暴露。失败留给下次请求重试。 + if figures: + _cache[cache_key] = {"ts": now, "data": data} return data diff --git a/sdk/app/services/normalize.py b/sdk/app/services/normalize.py index 334d538..ac73fed 100644 --- a/sdk/app/services/normalize.py +++ b/sdk/app/services/normalize.py @@ -300,9 +300,39 @@ def normalize_qa_response(raw: dict[str, Any] | None) -> dict[str, Any]: # 图片注入（2026-08-28 图文并茂方案） # --------------------------------------------------------------------------- -# 双键形态：[[FIG:PMC123]]（原形态）与 [[FIG:DOI:10.xxx/...]]（2026-09-12 通用化） -_FIG_MARKER = re.compile(r"\[\[FIG:([A-Za-z0-9]+|DOI:[^\]]+)\]\]") -_MAX_FIGURES = 8 +# 标记形态(2026-09-22 三修):LLM 输出不稳定(PMC/DOI/PMID、带不带冒号空格都出现过), +# 改为宽松捕获 [[FIG:任意内容]] + _marker_figures 归一化识别,不再逐形态猜正则。 +_FIG_MARKER = re.compile(r"\[\[FIG:([^\]]+)\]\]") +_MAX_FIGURES = 24 + + +def _normalize_marker_token(token: str) -> str: + """把 [[FIG:...]] 内容归一化为裸标识:去空白、去 PMCID:/PMID:/DOI: 前缀。""" + value = (token or "").strip() + value = re.sub(r"^(?:PMCID|PMID|DOI)\s*[:：]\s*", "", value, flags=re.I) + return value.strip() + + +def _marker_figures(token: str, figures_map: dict[str, dict[str, Any]]) -> tuple[str, list[dict]]: + """标记内容 → (map 键, 图清单)。三种形态:PMCxxxx / 纯数字 PMID / DOI。""" + norm = _normalize_marker_token(token) + upper = norm.upper() + if re.match(r"^PMC\d+$", upper): + return upper, ((figures_map.get(upper) or {}).get("figures") or []) + if re.match(r"^\d+$", norm): + for k, v in figures_map.items(): + if str((v or {}).get("pmid") or "") == norm: + return k, ((v or {}).get("figures") or []) + key = f"PMID:{norm}" + return key, ((figures_map.get(key) or {}).get("figures") or []) + if norm.startswith("10.") or upper.startswith("DOI:"): + doi_val = norm[4:].strip() if upper.startswith("DOI:") else norm + for cand in (f"DOI:{doi_val}", f"DOI:{doi_val.replace('/', '_')}"): + if cand in figures_map: + return cand, ((figures_map.get(cand) or {}).get("figures") or []) + key = f"DOI:{doi_val}" + return key, ((figures_map.get(key) or {}).get("figures") or []) + return norm, [] def inject_figure_markdown( @@ -317,25 +347,23 @@ def inject_figure_markdown( 规则： - 标记 → 该键对应前 2 张图；无图删标记 - - 兜底：无任何标记时，按引用 [PMID: xxx] 首次出现位置插入对应论文前 2 张图 - （pmcid 从 sources 的 paper_id/pmcid 推导） + - 兜底：无任何标记时，按引用 [PMID: xxx] / [PMCID: xxx] / [DOI: xxx] 首次 + 出现位置插入对应论文前 2 张图（2026-09-22 起三种引文形态都注入） - 总图数上限 8，超出删除标记/跳过 """ answer = answer or "" if not answer: return answer - # 1. 解析标记 + # 1. 解析标记(宽松捕获 + _marker_figures 归一化识别) markers = _FIG_MARKER.findall(answer) inserted = 0 if markers: used_pmcids: dict[str, int] = {} - for pmcid in markers: + for token in markers: if inserted >= _MAX_FIGURES: break - key = pmcid if pmcid.upper().startswith("DOI:") else pmcid.upper() - data = figures_map.get(key) or figures_map.get(pmcid) - figures = (data or {}).get("figures") or [] + key, figures = _marker_figures(token, figures_map) take = min(2, len(figures), _MAX_FIGURES - inserted) if not take: continue @@ -343,25 +371,43 @@ def inject_figure_markdown( markdown_lines = _figure_lines(figures[:take], base_url) replacement = "\n\n" + "\n\n".join(markdown_lines) + "\n\n" # 只替换第一个出现的该标记（每论文一个标记由 prompt 保证） - answer = answer.replace(f"[[FIG:{pmcid}]]", replacement, 1) + answer = answer.replace(f"[[FIG:{token}]]", replacement, 1) inserted += take # 清理残余标记（无图/超限的） answer = _FIG_MARKER.sub("", answer) - return answer + if inserted >= _MAX_FIGURES: + return answer + # 2026-09-22 修复:标记未用满配额(含一张未注入)时继续引文兜底。 + # 此前标记存在即 return——「有标记但无图」时 PMID/PMCID/DOI 引文永不注入。 - # 2. 兜底：无标记 → 按引用位置注入 + # 2. 兜底：标记未用满配额时,按引用位置注入(PMID/PMCID/DOI 三种引文形态) + citation_entries: list[tuple[str, str]] = [] # (citation 文本, figures_map 键) pmcid_by_pmid = _pmid_to_pmcid_map(figures_map) for pmid, pmcid in pmcid_by_pmid.items(): + citation_entries.append((f"[PMID: {pmid}]", pmcid)) + for key, data in figures_map.items(): + upper = str(key).upper() + if upper.startswith("PMC"): + citation_entries.append((f"[PMCID: {upper}]", key)) + elif str(key).startswith("DOI:"): + citation_entries.append((f"[DOI: {str(key)[4:].strip()}]", key)) + alias = (data or {}).get("doi_alias") + if alias and str(alias).startswith("DOI:"): + citation_entries.append((f"[DOI: {str(alias)[4:].strip()}]", key)) + seen_keys: set[str] = set() + for citation, key in citation_entries: if inserted >= _MAX_FIGURES: break - data = figures_map.get(pmcid) or {} + if key in seen_keys: + continue + data = figures_map.get(key) or {} figures = (data.get("figures") or [])[:2] if not figures: continue - citation = f"[PMID: {pmid}]" idx = answer.find(citation) if idx < 0: continue + seen_keys.add(key) end = idx + len(citation) markdown_lines = _figure_lines(figures, base_url) block = "\n\n" + "\n\n".join(markdown_lines) + "\n\n" diff --git a/sdk/app/services/sse_relay.py b/sdk/app/services/sse_relay.py index 3e8d062..6dd7ea5 100644 --- a/sdk/app/services/sse_relay.py +++ b/sdk/app/services/sse_relay.py @@ -51,8 +51,11 @@ INVALID_CITATION_ERROR_CODE = "invalid_llm_citation" # 图文并茂图片注入预算（缺陷 #7，2026-09-10）：complete 帧端到端延迟上限。 # 总预算毫秒数；PMCID 只取前 3 个；预算按 PMCID 均分，单篇/整体 wait_for 截断， # 超时静默降级（图片不影响答案主链）。 -FIGURE_INJECT_BUDGET_MS = float(os.environ.get("FIGURE_INJECT_BUDGET_MS", "2500")) -FIGURE_INJECT_MAX_PMCIDS = 3 +# 2026-09-22:2500→45000。175 实测 EBI 单次 fullTextXML 2.1-4.4s,PMC 形态是 +# 两次串行调用 + DOI 形态两段解析,每篇 15s(3 篇并发,总墙钟 15s)才稳妥。 +# 2026-09-22 二次:预取上限 3→10 篇后预算 90000(每篇 9s,10 篇并发,总墙钟 90s)。 +FIGURE_INJECT_BUDGET_MS = float(os.environ.get("FIGURE_INJECT_BUDGET_MS", "90000")) +FIGURE_INJECT_MAX_PMCIDS = int(os.environ.get("FIGURE_INJECT_MAX_PMCIDS", "10")) # API.md §4：SSE 响应头 sse_headers = { @@ -568,26 +571,57 @@ async def _inject_figures_async(normalized: dict) -> dict: # 由其内部识别并走双键本地索引 + EPMC 兜底；/api/figures/{pmcid} 路由兼容不变。） import re - marker_ids = re.findall(r"\[\[FIG:([A-Za-z0-9.:/_-]+)\]\]", answer) + marker_ids = re.findall(r"\[\[FIG:([^\]]+)\]\]", answer) + figures_map: dict = {} - # 来源里 pmcid 优先（有 pmid 供兜底） + # 来源里 pmcid 优先（有 pmid 供兜底）；同时记录 doi 别名（2026-09-20： + # 175 yz 图片库按 DOI 组织，PMC 标记经 doi 别名转 DOI 形态可命中本地图） for s in sources: pmcid = (s.get("pmcid") or "").strip().upper() pmid = s.get("pmid") if not pmcid: continue - figures_map.setdefault(pmcid, {"pmid": pmid, "figures": [], "license": ""}) - # 标记里的 PMCID / DOI 也加入（即使 sources 没有） - for marker in marker_ids: - upper = marker.upper() - if upper.startswith("PMC"): - figures_map.setdefault(upper, {"pmid": None, "figures": [], "license": ""}) - elif upper.startswith("DOI:"): - doi_key = marker - figures_map.setdefault(doi_key, {"pmid": None, "figures": [], "license": ""}) + entry = figures_map.setdefault(pmcid, {"pmid": pmid, "figures": [], "license": ""}) + doi_raw = (s.get("doi") or "").strip() + if doi_raw.lower().startswith("doi:"): + doi_raw = doi_raw[4:].strip() + if doi_raw and "doi_alias" not in entry: + entry["doi_alias"] = f"DOI:{doi_raw}" + # 2026-09-22:doi-only sources 补入 map 的 DOI 键(LLM 引 [DOI: ...]/[[FIG:PMID:]] 时兜底) + for s in sources: + doi_raw = (s.get("doi") or "").strip() + if not doi_raw: + continue + if doi_raw.lower().startswith("doi:"): + doi_raw = doi_raw[4:].strip() + if doi_raw: + figures_map.setdefault(f"DOI:{doi_raw}", {"pmid": s.get("pmid"), "figures": [], "license": ""}) - # 标记裁 top 3（按出现顺序：sources 在前，标记补充在后） - pmcids = list(figures_map.keys())[:FIGURE_INJECT_MAX_PMCIDS] + # 标记 → map 键解析(2026-09-22 三修:与 normalize._normalize_marker_token 同款归一化, + # 宽松捕获 [[FIG:任意内容]] 后识别 PMC/PMID/DOI,预取键与替换键保证一致) + def _marker_key(marker: str) -> str: + norm = re.sub(r"^(?:PMCID|PMID|DOI)\s*[:：]\s*", "", (marker or "").strip(), flags=re.I).strip() + upper = norm.upper() + if re.match(r"^PMC\d+$", upper): + figures_map.setdefault(upper, {"pmid": None, "figures": [], "license": ""}) + return upper + if re.match(r"^\d+$", norm): + for k, v in figures_map.items(): + if str((v or {}).get("pmid") or "") == norm: + return k + key = f"PMID:{norm}" + figures_map.setdefault(key, {"pmid": norm, "figures": [], "license": ""}) + return key + doi_val = norm + figures_map.setdefault(f"DOI:{doi_val}", {"pmid": None, "figures": [], "license": ""}) + return f"DOI:{doi_val}" + + marker_keys = [_marker_key(m) for m in marker_ids] + + # 标记裁 top 3(标记键优先,其次 sources 键——LLM 显式要的图不能被 sources 挤掉) + marked_keys = list(dict.fromkeys(k for k in marker_keys if k in figures_map)) + other_keys = [k for k in figures_map if k not in marked_keys] + pmcids = (marked_keys + other_keys)[:FIGURE_INJECT_MAX_PMCIDS] if not pmcids: base_url = "" # figures.resolve 已返回代理端点绝对路径，无需拼接 normalized["answer"] = inject_figure_markdown(answer, figures_map, base_url) @@ -598,10 +632,24 @@ async def _inject_figures_async(normalized: dict) -> dict: async def _fetch_one(pmcid: str) -> None: data = figures_map[pmcid] + # 2026-09-22 修复回归:doi 别名只做纯本地解析(标准图库+遗留体系,零外网); + # 本地未命中则退回原 PMCID 走完整 resolve(单次 EPMC 调用),不再用 DOI 形态 + # 的两段式外网兜底(DOI→PMCID 查询+fullTextXML 会拖爆预取预算)。 + alias = data.get("doi_alias") + local_figs: list = [] + if alias: + try: + from .figures import _doi_key, _local_figures_for_key + local_figs = _local_figures_for_key(pmcid, _doi_key(alias)) + except Exception: + local_figs = [] try: - resolved = await asyncio.wait_for( - resolve_figures(pmcid, limit=2), timeout=per_pmcid_s - ) + if local_figs: + resolved = {"figures": local_figs, "license": ""} + else: + resolved = await asyncio.wait_for( + resolve_figures(pmcid, limit=2), timeout=per_pmcid_s + ) data["figures"] = resolved.get("figures") or [] data["license"] = resolved.get("license") or "" except asyncio.TimeoutError: diff --git a/tests/frontend/unit/bioflow/bioflow.store.test.js b/tests/frontend/unit/bioflow/bioflow.store.test.js index 5577abc..a12d874 100644 --- a/tests/frontend/unit/bioflow/bioflow.store.test.js +++ b/tests/frontend/unit/bioflow/bioflow.store.test.js @@ -9,6 +9,7 @@ import { useBioflowStore, applyEvent, applySnapshot, + capLogTail, STAGE_STATUS_BY_EVENT, } from '../../../../ui/src/stores/bioflow.js'; @@ -50,6 +51,10 @@ vi.mock('../../../../ui/src/api/bioflow.js', () => ({ ok: true, data: { manifest: { status: 'success' }, log_tail: 'line1\nline2' }, })), + getBioflowStageLogs: vi.fn(async (jobId, stage, params) => ({ + ok: true, + data: { lines: ['L0', 'L1', 'L2'], offset: 3, limit: 500, eof: false, log_tail: '' }, + })), listBioflowArchives: vi.fn(async () => ({ ok: true, data: { @@ -278,6 +283,32 @@ describe('bioflow store —— 探活 / 动作 / 面板 / 复用', () => { expect(store.manifests['j1:identify'].loading).toBe(false); }); + it('loadStageLogs 分页增量追加 + 轮询启停', async () => { + const store = useBioflowStore(); + const rec = await store.loadStageLogs('j1', 'identify'); + expect(rec.lines).toEqual(['L0', 'L1', 'L2']); + expect(rec.offset).toBe(3); + expect(rec.eof).toBe(false); + // 轮询启动（fake timers） + vi.useFakeTimers(); + store.startStageLogPolling('j1', 'identify'); + expect(store.stageLogs['j1:identify'].timer).toBeTruthy(); + await vi.advanceTimersByTimeAsync(3100); + vi.useRealTimers(); + store.stopStageLogPolling('j1', 'identify'); + expect(store.stageLogs['j1:identify'].timer).toBeNull(); + }); + + it('requestPanelLocate/clearPanelLocate 定位请求（nonce 递增）', () => { + const store = useBioflowStore(); + store.requestPanelLocate({ jobId: 'j1', stage: 'identify' }); + expect(store.panelLocate).toEqual({ jobId: 'j1', stage: 'identify', nonce: 1 }); + store.requestPanelLocate({ jobId: 'j1', stage: 'align' }); + expect(store.panelLocate.nonce).toBe(2); + store.clearPanelLocate(); + expect(store.panelLocate).toBeNull(); + }); + it('会话切换 resetSession 隔离（TC-BIO-STORE-021）', () => { const store = useBioflowStore(); store.currentJob.jobId = 'j1'; @@ -313,8 +344,45 @@ describe('bioflow store —— 探活 / 动作 / 面板 / 复用', () => { describe('STAGE_STATUS_BY_EVENT —— 事件类型全集映射', () => { it('12 类事件中的阶段事件映射完整', () => { expect(Object.keys(STAGE_STATUS_BY_EVENT)).toEqual([ - 'stage_started', 'stage_log', 'stage_done', 'stage_failed', 'stage_skipped', - 'stage_await_confirm', 'stage_await_files', + 'stage_started', 'stage_log', 'stage_progress', 'stage_done', 'stage_failed', + 'stage_skipped', 'stage_await_confirm', 'stage_await_files', ]); }); + + it('stage_progress 事件推进进度条（防回退 + 100 清空 message）', () => { + const state = { + jobId: 'j1', jobStatus: 'running', lastSeq: 0, + stages: [{ stage: 'rnaseq_qc', status: 'running', progress: null, progressMessage: null }], + }; + let s = applyEvent(state, { type: 'stage_progress', seq: 1, data: { job_id: 'j1', stage: 'rnaseq_qc', percent: 33, message: 'SRR1' } }); + expect(s.stages[0].progress).toBe(33); + expect(s.stages[0].progressMessage).toBe('SRR1'); + expect(s.stages[0].status).toBe('running'); + // 回退帧忽略 + s = applyEvent(s, { type: 'stage_progress', seq: 2, data: { job_id: 'j1', stage: 'rnaseq_qc', percent: 10, message: 'old' } }); + expect(s.stages[0].progress).toBe(33); + // 100 → message 清空 + s = applyEvent(s, { type: 'stage_progress', seq: 3, data: { job_id: 'j1', stage: 'rnaseq_qc', percent: 100, message: '' } }); + expect(s.stages[0].progress).toBe(100); + expect(s.stages[0].progressMessage).toBe(''); + }); + + it('stage_started 清零进度（重跑新 attempt）', () => { + const state = { + jobId: 'j1', jobStatus: 'running', lastSeq: 0, + stages: [{ stage: 'rnaseq_qc', status: 'failed', progress: 80, progressMessage: 'x' }], + }; + // failed → rerun → stage_started(attempt 2) + const s = applyEvent(state, { type: 'stage_started', seq: 1, data: { job_id: 'j1', stage: 'rnaseq_qc', attempt: 2 } }); + expect(s.stages[0].status).toBe('running'); + expect(s.stages[0].progress).toBeNull(); + expect(s.stages[0].progressMessage).toBeNull(); + }); + + it('log_tail 4000 字符截断（capLogTail）', () => { + expect(capLogTail('abc')).toBe('abc'); + const long = 'x'.repeat(5000); + expect(capLogTail(long).length).toBe(4000); + expect(capLogTail(long)).toBe('x'.repeat(4000)); + }); }); diff --git a/tests/frontend/unit/bioflow/panes.test.js b/tests/frontend/unit/bioflow/panes.test.js index 52c22e1..79c4635 100644 --- a/tests/frontend/unit/bioflow/panes.test.js +++ b/tests/frontend/unit/bioflow/panes.test.js @@ -32,6 +32,10 @@ vi.mock('../../../../ui/src/api/bioflow.js', () => ({ ok: true, data: { manifest: { status: 'success', artifacts: [] }, log_tail: 'log-line-1\nlog-line-2' }, })), + getBioflowStageLogs: vi.fn(async () => ({ + ok: true, + data: { lines: ['log-line-1', 'log-line-2'], offset: 2, limit: 500, eof: true, log_tail: '' }, + })), listBioflowArchives: vi.fn(async () => ({ ok: true, data: { @@ -158,18 +162,24 @@ describe('BioJobsPane —— TC-BIO-JOB-001~016（L4）', () => { expect(dots[2].text()).toBe('◌'); }); - it('点击 DAG 阶段行：打开 manifest + 日志尾（code-block 渲染；TC-BIO-JOB-006/ACT-094）', async () => { + it('点击 DAG 阶段行：展开滚动日志区（logs 端点轮询；log 区头部 manifest 按钮切 manifest）', async () => { const w = mount(BioJobsPane); await flushPromises(); await w.findAll('.bj-item')[0].trigger('click'); await flushPromises(); await w.findAll('.bj-dag-row')[0].trigger('click'); await flushPromises(); - expect(bioflowApi.getBioflowStageManifest).toHaveBeenCalledWith('job-1', 'identify'); - expect(w.find('.bj-manifest').exists()).toBe(true); - const code = w.find('.bj-manifest .code-block'); + // 新交互：DAG 行点击 → 展开日志滚动区（getBioflowStageLogs 分页取数） + expect(bioflowApi.getBioflowStageLogs).toHaveBeenCalledWith('job-1', 'identify', { offset: 0, limit: 500 }); + expect(w.find('.bj-log').exists()).toBe(true); + const code = w.find('.bj-log .code-block'); expect(code.text()).toContain('log-line-1'); - expect(code.text()).toContain('manifest'); + expect(code.text()).toContain('log-line-2'); + // log 区头部 manifest 按钮 → 切到 manifest 区 + await w.find('.bj-log-actions .pane-btn').trigger('click'); + await flushPromises(); + expect(w.find('.bj-manifest').exists()).toBe(true); + expect(w.find('.bj-manifest .code-block').text()).toContain('manifest'); }); it('重跑按钮：选中含 failed 阶段任务时发出 rerun（TC-BIO-JOB-007/ACT-063）', async () => { diff --git a/ui/src/api/bioflow.js b/ui/src/api/bioflow.js index 942b3e6..1ae26e9 100644 --- a/ui/src/api/bioflow.js +++ b/ui/src/api/bioflow.js @@ -84,6 +84,11 @@ export function getBioflowStageManifest(jobId, stage, opts = {}) { return get(`/api/bioflow/jobs/${jobId}/stages/${stage}/manifest`, {}, opts); } +// ---------- 阶段日志分页（进度条滚动 log；offset=已读行数,执行中可读） ---------- +export function getBioflowStageLogs(jobId, stage, params = {}, opts = {}) { + return get(`/api/bioflow/jobs/${jobId}/stages/${stage}/logs`, params, opts); +} + // ---------- 任务事件（SSE；FR-BIO-014 断线重连回放） ---------- /** GET /api/bioflow/jobs/{id}/events —— 返回原始 fetch Response（SSE 由调用方消费）。 * params: {stage, last_seq} */ diff --git a/ui/src/components/bench/BioJobsPane.vue b/ui/src/components/bench/BioJobsPane.vue index bd70254..56d9116 100644 --- a/ui/src/components/bench/BioJobsPane.vue +++ b/ui/src/components/bench/BioJobsPane.vue @@ -3,12 +3,13 @@ // 任务列表（session-item 形态：标题 13px + 状态徽章）、阶段 DAG（纵向 2px hairline // 连接线 + 8px 圆点按阶段状态取色）、进度数字、manifest/日志尾浏览（code-block： // mono 12px code-bg 滚动）、重跑失败阶段、删除（running 禁用 + 二次确认）。 -import { computed, onMounted, ref } from 'vue'; +import { computed, nextTick, onMounted, ref, watch } from 'vue'; import { BF, stageStatusText, jobStatusText } from '../../i18n/bioflow.js'; import { useBioflowStore } from '../../stores/bioflow.js'; const bioflow = useBioflowStore(); +const rootEl = ref(null); const selectedJobId = ref(null); const confirmDelete = ref(null); // 待确认删除的任务 const rerunning = ref(false); @@ -18,6 +19,20 @@ const jobs = computed(() => bioflow.jobs || []); const selectedJob = computed(() => jobs.value.find((j) => j.id === selectedJobId.value) || null); const detailStages = computed(() => (selectedJob.value && selectedJob.value.stages) || []); +/** 实时阶段状态：当前任务选中时用 SSE 推进的 currentJob.stages 覆盖详情快照 */ +const liveStages = computed(() => { + const base = detailStages.value; + const live = bioflow.currentJob; + if (!live || live.jobId !== selectedJobId.value || !Array.isArray(live.stages)) return base; + const byStage = new Map(base.map((s) => [s.stage, s])); + for (const ls of live.stages) { + if (byStage.has(ls.stage)) { + byStage.set(ls.stage, { ...byStage.get(ls.stage), ...ls }); + } + } + return [...byStage.values()]; +}); + onMounted(() => { bioflow.loadJobs({ force: true }).catch(() => { /* jobsError 已标记，失败必须可见 */ }); }); @@ -97,6 +112,11 @@ const manifest = computed(() => { return bioflow.manifests[key] || null; }); async function openManifest(st) { + // 与日志展开区互斥 + if (logStage.value && selectedJob.value) { + bioflow.stopStageLogPolling(selectedJob.value.id, logStage.value.stage); + logStage.value = null; + } manifestStage.value = st; const m = bioflow.manifests[`${selectedJob.value.id}:${st.stage}`]; if (!m || !m.data) { @@ -109,6 +129,82 @@ function closeManifest() { manifestStage.value = null; } +// ---- 阶段日志滚动区（logs 端点轮询；点击 DAG 行展开/收起；与 manifest 互斥） ---- +const logStage = ref(null); +const logEl = ref(null); +const logStages = computed(() => bioflow.stageLogs || {}); +const logRec = computed(() => { + if (!logStage.value || !selectedJob.value) return null; + return logStages.value[`${selectedJob.value.id}:${logStage.value.stage}`] || null; +}); +async function toggleStageLog(st) { + const key = `${selectedJob.value.id}:${st.stage}`; + if (logStage.value && logStage.value.stage === st.stage) { + // 再点同阶段 → 收起 + bioflow.stopStageLogPolling(selectedJob.value.id, st.stage); + logStage.value = null; + return; + } + if (logStage.value) bioflow.stopStageLogPolling(selectedJob.value.id, logStage.value.stage); + manifestStage.value = null; // 与 manifest 区互斥 + logStage.value = st; + if (!bioflow.stageLogs[key]) { + bioflow.stageLogs[key] = { lines: [], offset: 0, eof: false, loading: false, error: false, timer: null }; + } + await bioflow.loadStageLogs(selectedJob.value.id, st.stage).catch(() => {}); + nextTick(() => { + if (logEl.value) logEl.value.scrollTop = logEl.value.scrollHeight; + }); + bioflow.startStageLogPolling(selectedJob.value.id, st.stage); +} +function closeStageLog() { + if (logStage.value && selectedJob.value) { + bioflow.stopStageLogPolling(selectedJob.value.id, logStage.value.stage); + } + logStage.value = null; +} + +// ---- 面板定位消费（聊天进度条点击 → scrollIntoView + hl-fade + 展开 log） ---- +const hlStage = ref(null); +function consumeLocate(payload) { + if (!payload || !payload.jobId) return; + const job = jobs.value.find((j) => j.id === payload.jobId); + if (!job) return; + selectJob(job); + hlStage.value = payload.stage || null; + // 展开对应阶段日志滚动区 + nextTick(async () => { + const stage = (liveStages.value || []).find((s) => s.stage === payload.stage); + if (stage) { + logStage.value = null; + await toggleStageLog(stage).catch(() => {}); + } + // 滚动到该任务行 + const node = rootEl.value && rootEl.value.querySelector(`[data-bj-job="${payload.jobId}"]`); + if (node && typeof node.scrollIntoView === 'function') { + node.scrollIntoView({ behavior: 'smooth', block: 'center' }); + } + // 阶段行高亮 + 滚动 + nextTick(() => { + const row = rootEl.value && rootEl.value.querySelector(`[data-bj-stage="${payload.stage}"]`); + if (row && typeof row.scrollIntoView === 'function') { + row.scrollIntoView({ behavior: 'smooth', block: 'center' }); + } + setTimeout(() => { hlStage.value = null; }, 2600); + }); + }); +} +watch( + () => bioflow.panelLocate && bioflow.panelLocate.nonce, + (nonce) => { + if (nonce == null) return; + const payload = bioflow.panelLocate; + bioflow.clearPanelLocate(); + consumeLocate(payload); + }, + { immediate: true }, +); + function formatSize(bytes) { if (bytes == null) return ''; if (bytes >= 1024 * 1024 * 1024) return `${(bytes / 1024 / 1024 / 1024).toFixed(1)} GB`; @@ -119,7 +215,7 @@ function formatSize(bytes) {