diff --git a/agent/legacy-rag/api/methodology_routes.py b/agent/legacy-rag/api/methodology_routes.py old mode 100755 new mode 100644 index 07b43d4..06137a4 --- a/agent/legacy-rag/api/methodology_routes.py +++ b/agent/legacy-rag/api/methodology_routes.py @@ -29,7 +29,7 @@ PAPERS_CSV = PROJECT_ROOT / "data" / "evaluation" / "methodology_papers.csv" GOLD_CSV = PROJECT_ROOT / "data" / "evaluation" / "methodology_gold.csv" REVIEW_RESULTS_JSON = BATCH_DIR / "review_results.json" REVIEW_RESULTS_CSV = BATCH_DIR / "review_results.csv" -# 2026-08-31 新工程适配:上传目录/产出根支持 env 覆盖 +# 2026-08-31 新工程适配:上传目录/产出根支持 env 覆盖(数据根迁移时无需改代码) UPLOAD_JOBS_DIR = Path(os.environ.get("METHODOLOGY_UPLOAD_JOBS_DIR", PROJECT_ROOT / "data" / "evaluation" / "methodology_upload_jobs")) UPLOAD_OUTPUT_ROOT = Path(os.environ.get("METHODOLOGY_UPLOAD_OUTPUT_ROOT", PROJECT_ROOT / "data" / "evaluation" / "0505")) DEFAULT_MINERU_CMD = str(conda_env_path("minerupy312") / "bin" / "mineru") @@ -351,6 +351,22 @@ def mineru_gpu_preflight_error( ) +# 缺陷 #38(2026-09-11):成功任务的 job.log 必然有 doc_analyze_streaming 行(176 实测)。 +# returncode≠0 且日志中无此行 = 启动期失败(pipeline 起来前就死了,真因被块缓冲吞掉)。 +# 判定条件(methodology 与 legacy-rag 两处 build_mineru_runtime/run_uploaded_methodology_job 同改) +MINERU_PIPELINE_LOG_MARKER = "doc_analyze_streaming" +MINERU_STARTUP_RETRIES = 1 + + +def job_log_has_pipeline_line(log_path: Path) -> bool: + """job.log 是否已出现 pipeline 日志行(def _job_log_has_pipeline_line)。""" + try: + text = log_path.read_text(encoding="utf-8", errors="replace") + except OSError: + return False + return MINERU_PIPELINE_LOG_MARKER in text + + def summarize_mineru_failure(log_path: Path, returncode: int) -> str: message = f"MinerU failed with exit code {returncode}; see {log_path}" if not log_path.exists(): @@ -368,6 +384,28 @@ def summarize_mineru_failure(log_path: Path, returncode: int) -> str: return message +def summarize_mineru_startup_failure(log_path: Path, returncode: int, retried: bool) -> str: + """启动期失败(无 pipeline 行)的中文可读错误(缺陷 #38)。 + + 附上 job.log 中可见的原始错误尾部,让用户能看到真因(修复前真因 + 被 stdout 块缓冲 + CLI 立即杀 API 进程永久丢失)。 + """ + tail = "" + try: + text = log_path.read_text(encoding="utf-8", errors="replace") + lines = [ln for ln in text.splitlines() if ln.strip()] + if lines: + tail = " | ".join(ln.strip() for ln in lines[-3:])[:500] + except OSError: + tail = "" + detail = f"原始日志尾部:{tail}" if tail else f"原始错误见日志:{log_path}" + action = "重试后" if retried else "" + return ( + f"提取启动阶段失败{action},可能受主机 GPU/显示重配置干扰,请重试" + f"(MinerU 退出码 {returncode};{detail})。" + ) + + def build_mineru_runtime() -> tuple[list[str], dict[str, str], dict[str, str]]: env = os.environ.copy() env.setdefault("MINERU_MODEL_SOURCE", "modelscope") @@ -384,6 +422,9 @@ def build_mineru_runtime() -> tuple[list[str], dict[str, str], dict[str, str]]: else: env.setdefault("CUDA_VISIBLE_DEVICES", "0") env.setdefault("TORCH_CUDNN_V8_API_DISABLED", "1") + # 缺陷 #38(2026-09-11):子进程 API 的异常日志即写即落(不再被 stdout + # 块缓冲吞掉)——CLI 失败后立即杀 API 进程前真因已落到 job.log。 + env["PYTHONUNBUFFERED"] = "1" # mineru 3.1.13 CLI 陷阱(部署 agent 实测):带 MINERU_API_SHUTDOWN_ON_STDIN_EOF=1 # 时,stdin EOF 会让 API 进程静默 shutdown——subprocess 调用必须 stdin=DEVNULL env.setdefault("MINERU_API_SHUTDOWN_ON_STDIN_EOF", "0") @@ -566,7 +607,34 @@ def run_uploaded_methodology_job( stdout=handle, stderr=subprocess.STDOUT, ) + # 缺陷 #38:启动期失败自动重试 1 次——returncode≠0 且 job.log 无 + # doc_analyze_streaming 行 = pipeline 起来前就失败(真因曾因块缓冲 + # 丢失)。重试的 stdout 继续 append 到同一 job.log,日志/状态留 + # retry 标记(G6-2:已有 pipeline 行的中途失败不得触发重试)。 + startup_retried = False + if result.returncode != 0 and not job_log_has_pipeline_line(log_path): + startup_retried = True + with log_path.open("a", encoding="utf-8") as handle: + handle.write( + f"[retry] MinerU 启动期失败(exit {result.returncode}," + "无 pipeline 阶段日志),自动重试 1/1\n" + ) + handle.write("$ " + " ".join(cmd) + "\n") + handle.flush() + result = subprocess.run( + cmd, + cwd=PROJECT_ROOT, + env=mineru_env, + text=True, + stdin=subprocess.DEVNULL, + stdout=handle, + stderr=subprocess.STDOUT, + ) if result.returncode != 0: + if not job_log_has_pipeline_line(log_path): + raise RuntimeError( + summarize_mineru_startup_failure(log_path, result.returncode, retried=startup_retried) + ) raise RuntimeError(summarize_mineru_failure(log_path, result.returncode)) matches = sorted(mineru_out.rglob("*_content_list.json")) if not matches: diff --git a/agent/legacy-rag/api/test_mineru_retry.py b/agent/legacy-rag/api/test_mineru_retry.py new file mode 100644 index 0000000..cad974c --- /dev/null +++ b/agent/legacy-rag/api/test_mineru_retry.py @@ -0,0 +1,328 @@ +"""test_mineru_retry.py —— MinerU 启动期失败重试与可读错误(缺陷 #38,验收 §4 + 回归 G6)。 + +运行(worktree 根目录,无需部署、不触服务): + /home/xzq/5090/teamWorkspace/zh/plantai-rag-new-venv/bin/python -m pytest agent/methodology/api/test_mineru_retry.py -v + /home/xzq/5090/teamWorkspace/zh/plantai-rag-new-venv/bin/python -m pytest agent/legacy-rag/api/test_mineru_retry.py -v +说明:agent/methodology 与 agent/legacy-rag 共用同一份 routes(文件名同、内容同), +同一测试文件复制两处、两处同跑(各用各树的 src/)。 + +断言: +1. build_mineru_runtime 返回 env 含 PYTHONUNBUFFERED=1; +2. subprocess.run 首次失败 + job.log 无 pipeline 行 → 自动重试 1 次 + (两次启动行、job.log 有 [retry] 标记,第二次 stdout 追加同一 job.log); +3. 重试仍失败 → status failed + 中文可读「提取启动阶段失败」+ 原始错误尾部; +4. 已有 pipeline 行的中途失败 → 不重试,走原错误路径(含 CUDA OOM 中文错误)。 +""" +from __future__ import annotations + +import json +import subprocess +from dataclasses import dataclass, field +from pathlib import Path + +import pytest + +import api.methodology_routes as mr + +from api.methodology_routes import ( + MINERU_PIPELINE_LOG_MARKER, + build_mineru_runtime, + job_log_has_pipeline_line, + run_uploaded_methodology_job, + summarize_mineru_startup_failure, +) + + +# --------------------------------------------------------------------------- +# 固定夹具 +# --------------------------------------------------------------------------- + +@dataclass +class FakeResult: + returncode: int + + +@dataclass +class RunnerRecorder: + """按调用顺序返回预设 returncode;每次调用把 stdout 写进所给 handle(模拟子进程日志)。""" + + sequence: list + calls: list = field(default_factory=list) + + def run(self, cmd, **kwargs): + n = len(self.calls) + self.calls.append({"cmd": cmd, "kwargs": kwargs}) + rc = int(self.sequence[min(n, len(self.sequence) - 1)]) + handle = kwargs.get("stdout") + if handle is not None and kwargs.get("text"): + handle.write(f"[fake stdout call#{n}] rc={rc}\n") + handle.flush() + return FakeResult(returncode=rc) + + +@pytest.fixture +def fake_success(monkeypatch): + """LLM 批次与 bbav5 配置阶段全部假成功(不触真实脚本/目录/真实函数)。 + + - gf_python_path / _preflight_executable 桩掉(不依赖真实 python 路径) + - build_bbav5_methodology_config 桩掉(不要求真实批次产物目录) + - figure_table 阶段走 routes 内部 try/except,不阻断 + """ + captured = {} + + def fake_batch(cmd, **kwargs): + captured["batch_cmd"] = cmd + return FakeResult(returncode=0) + + monkeypatch.setattr(mr, "gf_python_path", lambda: Path("/usr/bin/python3")) + monkeypatch.setattr(mr, "_preflight_executable", lambda cmd, hint, name: None) + monkeypatch.setattr(mr.subprocess, "run", fake_batch) + monkeypatch.setattr( + mr, + "build_bbav5_methodology_config", + lambda output_dir: captured.setdefault("config", {}), + ) + return captured + + +@pytest.fixture +def job_root(monkeypatch, tmp_path): + """把 UPLOAD_JOBS_DIR 重定向到 tmp(绝不写真实数据目录)。""" + root = tmp_path / "upload_jobs" + monkeypatch.setattr(mr, "UPLOAD_JOBS_DIR", root) + return root + + +def read_status(job_dir: Path) -> dict: + return json.loads((job_dir / "status.json").read_text(encoding="utf-8")) + + +def run_job(monkeypatch, tmp_path, runner: RunnerRecorder, source_path: Path, fake_success, job_root): + """以 runner 重绑 subprocess.run 跑完整任务流(fake_success 保证后续阶段假成功)。 + + 注意:fake_success 已把 mr.subprocess.run 桩为假批次;这里再次 setattr + 为 runner(同一模块对象),此后所有子进程调用(mineru 两次 + LLM 批次) + 都由 runner 按 sequence 依次应答——测试 sequence 需覆盖全部调用。 + """ + job_dir = job_root / "testjob" + # 模拟 mineru 成功产出:在 job/mineru 下伪造 *_content_list.json + # (测试不真正执行 mineru) + (job_dir / "mineru").mkdir(parents=True, exist_ok=True) + (job_dir / "mineru" / "fake_content_list.json").write_text("{}", encoding="utf-8") + # GPU 预检禁用(否则 nvidia-smi 会消耗 runner 的 sequence 槽位) + monkeypatch.setenv("METHODOLOGY_MINERU_GPU_PREFLIGHT", "0") + monkeypatch.setattr(subprocess, "run", runner.run) + run_uploaded_methodology_job( + job_id="testjob", + source_path=source_path, + paper_id="paper1", + family="", + output_dir=tmp_path / "out", + llm_model="deepseek-v4-flash", + llm_request_timeout=75, + sandbox=False, + sandbox_run_id="", + ) + return job_dir + + +@pytest.fixture +def pdf_source(tmp_path): + src = tmp_path / "sample.pdf" + src.write_bytes(b"%PDF-1.4\nfake pdf bytes for startup-failure test\n") + return src + + +@pytest.fixture +def json_source(tmp_path): + src = tmp_path / "sample.json" + src.write_text("{}", encoding="utf-8") + return src + + +# --------------------------------------------------------------------------- +# 修复点 1:env 注入 PYTHONUNBUFFERED +# --------------------------------------------------------------------------- + +def test_build_mineru_runtime_env_unbuffered(monkeypatch): + monkeypatch.delenv("MINERU_CMD", raising=False) + cmd, env, runtime = build_mineru_runtime() + assert env.get("PYTHONUNBUFFERED") == "1" + assert cmd # 默认 MINERU_CMD 路径(不要求存在,build 阶段不校验) + + +# --------------------------------------------------------------------------- +# 修复点 2:启动期失败自动重试 1 次 +# --------------------------------------------------------------------------- + +def test_startup_failure_retries_once(monkeypatch, tmp_path, pdf_source, fake_success, job_root): + runner = RunnerRecorder(sequence=[1, 0, 0]) # mineru 首调失败 → 重试成功 → LLM 批次成功 + job_dir = run_job(monkeypatch, tmp_path, runner, pdf_source, fake_success, job_root) + + log_text = (job_dir / "job.log").read_text(encoding="utf-8") + startup_lines = [ln for ln in log_text.splitlines() if ln.startswith("$ ")] + assert len(startup_lines) >= 2, f"应出现两次 mineru 启动(1 次自动重试),实际:{startup_lines}" + assert "[retry]" in log_text, "job.log 应有可识别重试标记" + # 第二次 subprocess.run 的 stdout 追加到同一 job.log + assert "[fake stdout call#0]" in log_text + assert "[fake stdout call#1]" in log_text + + status = read_status(job_dir) + assert status.get("status") == "completed" + + +def test_startup_failure_retry_then_fail_readable_error(monkeypatch, tmp_path, pdf_source, fake_success, job_root): + runner = RunnerRecorder(sequence=[1, 1]) # 两次都失败,且无 pipeline 行 + job_dir = run_job(monkeypatch, tmp_path, runner, pdf_source, fake_success, job_root) + + log_text = (job_dir / "job.log").read_text(encoding="utf-8") + assert len([ln for ln in log_text.splitlines() if ln.startswith("$ /")]) == 2 # 两次 mineru 启动 + assert "[retry]" in log_text + # pipeline 判定:仅检查子进程输出行(启动命令行本身含 marker 关键字) + stdout_lines = [ln for ln in log_text.splitlines() if "[fake stdout" in ln or ln.startswith("[retry]")] + assert MINERU_PIPELINE_LOG_MARKER not in "\n".join(stdout_lines) + + status = read_status(job_dir) + assert status.get("status") == "failed" + message = str(status.get("message") or "") + assert "提取启动阶段失败" in message, f"中文可读错误缺失:{message}" + assert "退出码" in message and "1" in message + assert "原始日志尾部" in message or "原始错误见日志" in message + assert message # 非空 + + +def test_startup_failure_with_marker_retries_once_then_no_second_retry(monkeypatch, tmp_path, pdf_source, job_root): + """G6-2 边界:第一次失败无 marker → 触发重试;重试后才有 marker 但 rc≠0 → + 不二次重试,走 summarize_mineru_failure(有 marker 的中途失败路径)。""" + sequence = [1, 1] + calls = [] + + def run_with_marker(cmd, **kwargs): + n = len(calls) + rc = int(sequence[min(n, len(sequence) - 1)]) + calls.append({"cmd": cmd, "kwargs": kwargs}) + handle = kwargs.get("stdout") + if handle is not None and kwargs.get("text"): + if n == 1: # 第二次调用(重试)时写入 pipeline 行 + handle.write(f"[fake] {MINERU_PIPELINE_LOG_MARKER} started\n") + handle.write(f"[fake stdout call#{n}] rc={rc}\n") + handle.flush() + return FakeResult(returncode=rc) + + monkeypatch.setenv("METHODOLOGY_MINERU_GPU_PREFLIGHT", "0") + monkeypatch.setattr(subprocess, "run", run_with_marker) + monkeypatch.setattr(mr, "_preflight_executable", lambda cmd, hint, name: None) + monkeypatch.setattr( + mr, + "summarize_mineru_failure", + lambda log_path, rc: f"CUDA out of memory fake path rc={rc}", + ) + job_dir = job_root / "testjob" + run_uploaded_methodology_job( + job_id="testjob", + source_path=pdf_source, + paper_id="paper1", + family="", + output_dir=tmp_path / "out", + llm_model="deepseek-v4-flash", + llm_request_timeout=75, + sandbox=False, + sandbox_run_id="", + ) + log_text = (job_dir / "job.log").read_text(encoding="utf-8") + assert MINERU_PIPELINE_LOG_MARKER in log_text + assert log_text.count("[retry]") == 1, "有 pipeline 行后不得再次重试" + # 中途失败(有 marker)→ 原错误路径(CUDA 中文错误) + status = read_status(job_dir) + assert status.get("status") == "failed" + assert "CUDA" in str(status.get("message") or "") + + +# --------------------------------------------------------------------------- +# 修复点 4:有 pipeline 行的中途失败不重试 +# --------------------------------------------------------------------------- + +def test_mid_failure_with_pipeline_line_no_retry(monkeypatch, tmp_path, pdf_source, job_root): + sequence = [1] + calls = [] + + def run_with_marker(cmd, **kwargs): + n = len(calls) + rc = int(sequence[min(n, len(sequence) - 1)]) + calls.append({"cmd": cmd, "kwargs": kwargs}) + handle = kwargs.get("stdout") + if handle is not None and kwargs.get("text"): + if n == 0: # 第一次调用即写 pipeline 行 + handle.write(f"[fake] {MINERU_PIPELINE_LOG_MARKER} started\n") + handle.write(f"[fake stdout call#{n}] rc={rc}\n") + handle.flush() + return FakeResult(returncode=rc) + + monkeypatch.setenv("METHODOLOGY_MINERU_GPU_PREFLIGHT", "0") + monkeypatch.setattr(subprocess, "run", run_with_marker) + monkeypatch.setattr(mr, "_preflight_executable", lambda cmd, hint, name: None) + monkeypatch.setattr( + mr, + "summarize_mineru_failure", + lambda log_path, rc: f"CUDA out of memory fake path rc={rc}", + ) + job_dir = job_root / "testjob" + run_uploaded_methodology_job( + job_id="testjob", + source_path=pdf_source, + paper_id="paper1", + family="", + output_dir=tmp_path / "out", + llm_model="deepseek-v4-flash", + llm_request_timeout=75, + sandbox=False, + sandbox_run_id="", + ) + log_text = (job_dir / "job.log").read_text(encoding="utf-8") + assert MINERU_PIPELINE_LOG_MARKER in log_text + assert "[retry]" not in log_text, "已有 pipeline 行的中途失败不得触发重试" + # 原始启动块 = 1 条 `$ env ...` + 1 条 `$ `;无重试追加 + assert len([ln for ln in log_text.splitlines() if ln.startswith("$ /")]) == 1 + status = read_status(job_dir) + assert status.get("status") == "failed" + assert "CUDA" in str(status.get("message") or "") + + +# --------------------------------------------------------------------------- +# JSON 直接投递(跳过 mineru 段)不重试、不破坏状态机 +# --------------------------------------------------------------------------- + +def test_json_source_skips_mineru_no_retry(monkeypatch, tmp_path, json_source, fake_success, job_root): + runner = RunnerRecorder(sequence=[0]) # 仅 LLM 批次一次 + job_dir = run_job(monkeypatch, tmp_path, runner, json_source, fake_success, job_root) + log_text = (job_dir / "job.log").read_text(encoding="utf-8") + assert "[retry]" not in log_text + assert MINERU_PIPELINE_LOG_MARKER not in log_text + status = read_status(job_dir) + assert status.get("status") == "completed" + + +# --------------------------------------------------------------------------- +# 工具函数单元 +# --------------------------------------------------------------------------- + +def test_job_log_has_pipeline_line(tmp_path): + log_path = tmp_path / "job.log" + assert job_log_has_pipeline_line(log_path) is False + log_path.write_text("noise\n", encoding="utf-8") + assert job_log_has_pipeline_line(log_path) is False + log_path.write_text(f"2026-09-11 [INFO] {MINERU_PIPELINE_LOG_MARKER} begin\n", encoding="utf-8") + assert job_log_has_pipeline_line(log_path) is True + + +def test_summarize_startup_failure_readable(tmp_path): + log_path = tmp_path / "job.log" + log_path.write_text("line one\nline two\nTraceback: boom\n", encoding="utf-8") + msg = summarize_mineru_startup_failure(log_path, 1, retried=True) + assert "提取启动阶段失败" in msg + assert "重试后" in msg + assert "退出码 1" in msg + assert "Traceback: boom" in msg + missing = summarize_mineru_startup_failure(tmp_path / "nope.log", 1, retried=False) + assert "提取启动阶段失败" in missing + assert "重试后" not in missing diff --git a/agent/methodology/api/methodology_routes.py b/agent/methodology/api/methodology_routes.py old mode 100755 new mode 100644 index e9f1f7a..06137a4 --- a/agent/methodology/api/methodology_routes.py +++ b/agent/methodology/api/methodology_routes.py @@ -351,6 +351,22 @@ def mineru_gpu_preflight_error( ) +# 缺陷 #38(2026-09-11):成功任务的 job.log 必然有 doc_analyze_streaming 行(176 实测)。 +# returncode≠0 且日志中无此行 = 启动期失败(pipeline 起来前就死了,真因被块缓冲吞掉)。 +# 判定条件(methodology 与 legacy-rag 两处 build_mineru_runtime/run_uploaded_methodology_job 同改) +MINERU_PIPELINE_LOG_MARKER = "doc_analyze_streaming" +MINERU_STARTUP_RETRIES = 1 + + +def job_log_has_pipeline_line(log_path: Path) -> bool: + """job.log 是否已出现 pipeline 日志行(def _job_log_has_pipeline_line)。""" + try: + text = log_path.read_text(encoding="utf-8", errors="replace") + except OSError: + return False + return MINERU_PIPELINE_LOG_MARKER in text + + def summarize_mineru_failure(log_path: Path, returncode: int) -> str: message = f"MinerU failed with exit code {returncode}; see {log_path}" if not log_path.exists(): @@ -368,6 +384,28 @@ def summarize_mineru_failure(log_path: Path, returncode: int) -> str: return message +def summarize_mineru_startup_failure(log_path: Path, returncode: int, retried: bool) -> str: + """启动期失败(无 pipeline 行)的中文可读错误(缺陷 #38)。 + + 附上 job.log 中可见的原始错误尾部,让用户能看到真因(修复前真因 + 被 stdout 块缓冲 + CLI 立即杀 API 进程永久丢失)。 + """ + tail = "" + try: + text = log_path.read_text(encoding="utf-8", errors="replace") + lines = [ln for ln in text.splitlines() if ln.strip()] + if lines: + tail = " | ".join(ln.strip() for ln in lines[-3:])[:500] + except OSError: + tail = "" + detail = f"原始日志尾部:{tail}" if tail else f"原始错误见日志:{log_path}" + action = "重试后" if retried else "" + return ( + f"提取启动阶段失败{action},可能受主机 GPU/显示重配置干扰,请重试" + f"(MinerU 退出码 {returncode};{detail})。" + ) + + def build_mineru_runtime() -> tuple[list[str], dict[str, str], dict[str, str]]: env = os.environ.copy() env.setdefault("MINERU_MODEL_SOURCE", "modelscope") @@ -384,6 +422,9 @@ def build_mineru_runtime() -> tuple[list[str], dict[str, str], dict[str, str]]: else: env.setdefault("CUDA_VISIBLE_DEVICES", "0") env.setdefault("TORCH_CUDNN_V8_API_DISABLED", "1") + # 缺陷 #38(2026-09-11):子进程 API 的异常日志即写即落(不再被 stdout + # 块缓冲吞掉)——CLI 失败后立即杀 API 进程前真因已落到 job.log。 + env["PYTHONUNBUFFERED"] = "1" # mineru 3.1.13 CLI 陷阱(部署 agent 实测):带 MINERU_API_SHUTDOWN_ON_STDIN_EOF=1 # 时,stdin EOF 会让 API 进程静默 shutdown——subprocess 调用必须 stdin=DEVNULL env.setdefault("MINERU_API_SHUTDOWN_ON_STDIN_EOF", "0") @@ -566,7 +607,34 @@ def run_uploaded_methodology_job( stdout=handle, stderr=subprocess.STDOUT, ) + # 缺陷 #38:启动期失败自动重试 1 次——returncode≠0 且 job.log 无 + # doc_analyze_streaming 行 = pipeline 起来前就失败(真因曾因块缓冲 + # 丢失)。重试的 stdout 继续 append 到同一 job.log,日志/状态留 + # retry 标记(G6-2:已有 pipeline 行的中途失败不得触发重试)。 + startup_retried = False + if result.returncode != 0 and not job_log_has_pipeline_line(log_path): + startup_retried = True + with log_path.open("a", encoding="utf-8") as handle: + handle.write( + f"[retry] MinerU 启动期失败(exit {result.returncode}," + "无 pipeline 阶段日志),自动重试 1/1\n" + ) + handle.write("$ " + " ".join(cmd) + "\n") + handle.flush() + result = subprocess.run( + cmd, + cwd=PROJECT_ROOT, + env=mineru_env, + text=True, + stdin=subprocess.DEVNULL, + stdout=handle, + stderr=subprocess.STDOUT, + ) if result.returncode != 0: + if not job_log_has_pipeline_line(log_path): + raise RuntimeError( + summarize_mineru_startup_failure(log_path, result.returncode, retried=startup_retried) + ) raise RuntimeError(summarize_mineru_failure(log_path, result.returncode)) matches = sorted(mineru_out.rglob("*_content_list.json")) if not matches: diff --git a/agent/methodology/api/test_mineru_retry.py b/agent/methodology/api/test_mineru_retry.py new file mode 100644 index 0000000..cad974c --- /dev/null +++ b/agent/methodology/api/test_mineru_retry.py @@ -0,0 +1,328 @@ +"""test_mineru_retry.py —— MinerU 启动期失败重试与可读错误(缺陷 #38,验收 §4 + 回归 G6)。 + +运行(worktree 根目录,无需部署、不触服务): + /home/xzq/5090/teamWorkspace/zh/plantai-rag-new-venv/bin/python -m pytest agent/methodology/api/test_mineru_retry.py -v + /home/xzq/5090/teamWorkspace/zh/plantai-rag-new-venv/bin/python -m pytest agent/legacy-rag/api/test_mineru_retry.py -v +说明:agent/methodology 与 agent/legacy-rag 共用同一份 routes(文件名同、内容同), +同一测试文件复制两处、两处同跑(各用各树的 src/)。 + +断言: +1. build_mineru_runtime 返回 env 含 PYTHONUNBUFFERED=1; +2. subprocess.run 首次失败 + job.log 无 pipeline 行 → 自动重试 1 次 + (两次启动行、job.log 有 [retry] 标记,第二次 stdout 追加同一 job.log); +3. 重试仍失败 → status failed + 中文可读「提取启动阶段失败」+ 原始错误尾部; +4. 已有 pipeline 行的中途失败 → 不重试,走原错误路径(含 CUDA OOM 中文错误)。 +""" +from __future__ import annotations + +import json +import subprocess +from dataclasses import dataclass, field +from pathlib import Path + +import pytest + +import api.methodology_routes as mr + +from api.methodology_routes import ( + MINERU_PIPELINE_LOG_MARKER, + build_mineru_runtime, + job_log_has_pipeline_line, + run_uploaded_methodology_job, + summarize_mineru_startup_failure, +) + + +# --------------------------------------------------------------------------- +# 固定夹具 +# --------------------------------------------------------------------------- + +@dataclass +class FakeResult: + returncode: int + + +@dataclass +class RunnerRecorder: + """按调用顺序返回预设 returncode;每次调用把 stdout 写进所给 handle(模拟子进程日志)。""" + + sequence: list + calls: list = field(default_factory=list) + + def run(self, cmd, **kwargs): + n = len(self.calls) + self.calls.append({"cmd": cmd, "kwargs": kwargs}) + rc = int(self.sequence[min(n, len(self.sequence) - 1)]) + handle = kwargs.get("stdout") + if handle is not None and kwargs.get("text"): + handle.write(f"[fake stdout call#{n}] rc={rc}\n") + handle.flush() + return FakeResult(returncode=rc) + + +@pytest.fixture +def fake_success(monkeypatch): + """LLM 批次与 bbav5 配置阶段全部假成功(不触真实脚本/目录/真实函数)。 + + - gf_python_path / _preflight_executable 桩掉(不依赖真实 python 路径) + - build_bbav5_methodology_config 桩掉(不要求真实批次产物目录) + - figure_table 阶段走 routes 内部 try/except,不阻断 + """ + captured = {} + + def fake_batch(cmd, **kwargs): + captured["batch_cmd"] = cmd + return FakeResult(returncode=0) + + monkeypatch.setattr(mr, "gf_python_path", lambda: Path("/usr/bin/python3")) + monkeypatch.setattr(mr, "_preflight_executable", lambda cmd, hint, name: None) + monkeypatch.setattr(mr.subprocess, "run", fake_batch) + monkeypatch.setattr( + mr, + "build_bbav5_methodology_config", + lambda output_dir: captured.setdefault("config", {}), + ) + return captured + + +@pytest.fixture +def job_root(monkeypatch, tmp_path): + """把 UPLOAD_JOBS_DIR 重定向到 tmp(绝不写真实数据目录)。""" + root = tmp_path / "upload_jobs" + monkeypatch.setattr(mr, "UPLOAD_JOBS_DIR", root) + return root + + +def read_status(job_dir: Path) -> dict: + return json.loads((job_dir / "status.json").read_text(encoding="utf-8")) + + +def run_job(monkeypatch, tmp_path, runner: RunnerRecorder, source_path: Path, fake_success, job_root): + """以 runner 重绑 subprocess.run 跑完整任务流(fake_success 保证后续阶段假成功)。 + + 注意:fake_success 已把 mr.subprocess.run 桩为假批次;这里再次 setattr + 为 runner(同一模块对象),此后所有子进程调用(mineru 两次 + LLM 批次) + 都由 runner 按 sequence 依次应答——测试 sequence 需覆盖全部调用。 + """ + job_dir = job_root / "testjob" + # 模拟 mineru 成功产出:在 job/mineru 下伪造 *_content_list.json + # (测试不真正执行 mineru) + (job_dir / "mineru").mkdir(parents=True, exist_ok=True) + (job_dir / "mineru" / "fake_content_list.json").write_text("{}", encoding="utf-8") + # GPU 预检禁用(否则 nvidia-smi 会消耗 runner 的 sequence 槽位) + monkeypatch.setenv("METHODOLOGY_MINERU_GPU_PREFLIGHT", "0") + monkeypatch.setattr(subprocess, "run", runner.run) + run_uploaded_methodology_job( + job_id="testjob", + source_path=source_path, + paper_id="paper1", + family="", + output_dir=tmp_path / "out", + llm_model="deepseek-v4-flash", + llm_request_timeout=75, + sandbox=False, + sandbox_run_id="", + ) + return job_dir + + +@pytest.fixture +def pdf_source(tmp_path): + src = tmp_path / "sample.pdf" + src.write_bytes(b"%PDF-1.4\nfake pdf bytes for startup-failure test\n") + return src + + +@pytest.fixture +def json_source(tmp_path): + src = tmp_path / "sample.json" + src.write_text("{}", encoding="utf-8") + return src + + +# --------------------------------------------------------------------------- +# 修复点 1:env 注入 PYTHONUNBUFFERED +# --------------------------------------------------------------------------- + +def test_build_mineru_runtime_env_unbuffered(monkeypatch): + monkeypatch.delenv("MINERU_CMD", raising=False) + cmd, env, runtime = build_mineru_runtime() + assert env.get("PYTHONUNBUFFERED") == "1" + assert cmd # 默认 MINERU_CMD 路径(不要求存在,build 阶段不校验) + + +# --------------------------------------------------------------------------- +# 修复点 2:启动期失败自动重试 1 次 +# --------------------------------------------------------------------------- + +def test_startup_failure_retries_once(monkeypatch, tmp_path, pdf_source, fake_success, job_root): + runner = RunnerRecorder(sequence=[1, 0, 0]) # mineru 首调失败 → 重试成功 → LLM 批次成功 + job_dir = run_job(monkeypatch, tmp_path, runner, pdf_source, fake_success, job_root) + + log_text = (job_dir / "job.log").read_text(encoding="utf-8") + startup_lines = [ln for ln in log_text.splitlines() if ln.startswith("$ ")] + assert len(startup_lines) >= 2, f"应出现两次 mineru 启动(1 次自动重试),实际:{startup_lines}" + assert "[retry]" in log_text, "job.log 应有可识别重试标记" + # 第二次 subprocess.run 的 stdout 追加到同一 job.log + assert "[fake stdout call#0]" in log_text + assert "[fake stdout call#1]" in log_text + + status = read_status(job_dir) + assert status.get("status") == "completed" + + +def test_startup_failure_retry_then_fail_readable_error(monkeypatch, tmp_path, pdf_source, fake_success, job_root): + runner = RunnerRecorder(sequence=[1, 1]) # 两次都失败,且无 pipeline 行 + job_dir = run_job(monkeypatch, tmp_path, runner, pdf_source, fake_success, job_root) + + log_text = (job_dir / "job.log").read_text(encoding="utf-8") + assert len([ln for ln in log_text.splitlines() if ln.startswith("$ /")]) == 2 # 两次 mineru 启动 + assert "[retry]" in log_text + # pipeline 判定:仅检查子进程输出行(启动命令行本身含 marker 关键字) + stdout_lines = [ln for ln in log_text.splitlines() if "[fake stdout" in ln or ln.startswith("[retry]")] + assert MINERU_PIPELINE_LOG_MARKER not in "\n".join(stdout_lines) + + status = read_status(job_dir) + assert status.get("status") == "failed" + message = str(status.get("message") or "") + assert "提取启动阶段失败" in message, f"中文可读错误缺失:{message}" + assert "退出码" in message and "1" in message + assert "原始日志尾部" in message or "原始错误见日志" in message + assert message # 非空 + + +def test_startup_failure_with_marker_retries_once_then_no_second_retry(monkeypatch, tmp_path, pdf_source, job_root): + """G6-2 边界:第一次失败无 marker → 触发重试;重试后才有 marker 但 rc≠0 → + 不二次重试,走 summarize_mineru_failure(有 marker 的中途失败路径)。""" + sequence = [1, 1] + calls = [] + + def run_with_marker(cmd, **kwargs): + n = len(calls) + rc = int(sequence[min(n, len(sequence) - 1)]) + calls.append({"cmd": cmd, "kwargs": kwargs}) + handle = kwargs.get("stdout") + if handle is not None and kwargs.get("text"): + if n == 1: # 第二次调用(重试)时写入 pipeline 行 + handle.write(f"[fake] {MINERU_PIPELINE_LOG_MARKER} started\n") + handle.write(f"[fake stdout call#{n}] rc={rc}\n") + handle.flush() + return FakeResult(returncode=rc) + + monkeypatch.setenv("METHODOLOGY_MINERU_GPU_PREFLIGHT", "0") + monkeypatch.setattr(subprocess, "run", run_with_marker) + monkeypatch.setattr(mr, "_preflight_executable", lambda cmd, hint, name: None) + monkeypatch.setattr( + mr, + "summarize_mineru_failure", + lambda log_path, rc: f"CUDA out of memory fake path rc={rc}", + ) + job_dir = job_root / "testjob" + run_uploaded_methodology_job( + job_id="testjob", + source_path=pdf_source, + paper_id="paper1", + family="", + output_dir=tmp_path / "out", + llm_model="deepseek-v4-flash", + llm_request_timeout=75, + sandbox=False, + sandbox_run_id="", + ) + log_text = (job_dir / "job.log").read_text(encoding="utf-8") + assert MINERU_PIPELINE_LOG_MARKER in log_text + assert log_text.count("[retry]") == 1, "有 pipeline 行后不得再次重试" + # 中途失败(有 marker)→ 原错误路径(CUDA 中文错误) + status = read_status(job_dir) + assert status.get("status") == "failed" + assert "CUDA" in str(status.get("message") or "") + + +# --------------------------------------------------------------------------- +# 修复点 4:有 pipeline 行的中途失败不重试 +# --------------------------------------------------------------------------- + +def test_mid_failure_with_pipeline_line_no_retry(monkeypatch, tmp_path, pdf_source, job_root): + sequence = [1] + calls = [] + + def run_with_marker(cmd, **kwargs): + n = len(calls) + rc = int(sequence[min(n, len(sequence) - 1)]) + calls.append({"cmd": cmd, "kwargs": kwargs}) + handle = kwargs.get("stdout") + if handle is not None and kwargs.get("text"): + if n == 0: # 第一次调用即写 pipeline 行 + handle.write(f"[fake] {MINERU_PIPELINE_LOG_MARKER} started\n") + handle.write(f"[fake stdout call#{n}] rc={rc}\n") + handle.flush() + return FakeResult(returncode=rc) + + monkeypatch.setenv("METHODOLOGY_MINERU_GPU_PREFLIGHT", "0") + monkeypatch.setattr(subprocess, "run", run_with_marker) + monkeypatch.setattr(mr, "_preflight_executable", lambda cmd, hint, name: None) + monkeypatch.setattr( + mr, + "summarize_mineru_failure", + lambda log_path, rc: f"CUDA out of memory fake path rc={rc}", + ) + job_dir = job_root / "testjob" + run_uploaded_methodology_job( + job_id="testjob", + source_path=pdf_source, + paper_id="paper1", + family="", + output_dir=tmp_path / "out", + llm_model="deepseek-v4-flash", + llm_request_timeout=75, + sandbox=False, + sandbox_run_id="", + ) + log_text = (job_dir / "job.log").read_text(encoding="utf-8") + assert MINERU_PIPELINE_LOG_MARKER in log_text + assert "[retry]" not in log_text, "已有 pipeline 行的中途失败不得触发重试" + # 原始启动块 = 1 条 `$ env ...` + 1 条 `$ `;无重试追加 + assert len([ln for ln in log_text.splitlines() if ln.startswith("$ /")]) == 1 + status = read_status(job_dir) + assert status.get("status") == "failed" + assert "CUDA" in str(status.get("message") or "") + + +# --------------------------------------------------------------------------- +# JSON 直接投递(跳过 mineru 段)不重试、不破坏状态机 +# --------------------------------------------------------------------------- + +def test_json_source_skips_mineru_no_retry(monkeypatch, tmp_path, json_source, fake_success, job_root): + runner = RunnerRecorder(sequence=[0]) # 仅 LLM 批次一次 + job_dir = run_job(monkeypatch, tmp_path, runner, json_source, fake_success, job_root) + log_text = (job_dir / "job.log").read_text(encoding="utf-8") + assert "[retry]" not in log_text + assert MINERU_PIPELINE_LOG_MARKER not in log_text + status = read_status(job_dir) + assert status.get("status") == "completed" + + +# --------------------------------------------------------------------------- +# 工具函数单元 +# --------------------------------------------------------------------------- + +def test_job_log_has_pipeline_line(tmp_path): + log_path = tmp_path / "job.log" + assert job_log_has_pipeline_line(log_path) is False + log_path.write_text("noise\n", encoding="utf-8") + assert job_log_has_pipeline_line(log_path) is False + log_path.write_text(f"2026-09-11 [INFO] {MINERU_PIPELINE_LOG_MARKER} begin\n", encoding="utf-8") + assert job_log_has_pipeline_line(log_path) is True + + +def test_summarize_startup_failure_readable(tmp_path): + log_path = tmp_path / "job.log" + log_path.write_text("line one\nline two\nTraceback: boom\n", encoding="utf-8") + msg = summarize_mineru_startup_failure(log_path, 1, retried=True) + assert "提取启动阶段失败" in msg + assert "重试后" in msg + assert "退出码 1" in msg + assert "Traceback: boom" in msg + missing = summarize_mineru_startup_failure(tmp_path / "nope.log", 1, retried=False) + assert "提取启动阶段失败" in missing + assert "重试后" not in missing