fix(r4-38): mineru 启动期失败自动重试+中文可读错误+PYTHONUNBUFFERED

缺陷 #38(验收 §4):任务在最早阶段(pipeline 日志出现前)抛出无消息异常,
mineru API 侧 task.error 空串 → CLI 拼出空 message;API 异常日志打向 stdout
被块缓冲,CLI 失败后立即杀 API 进程,真因永久丢失。

修复(agent/methodology 与 agent/legacy-rag 两处同改):
- build_mineru_runtime 的 env 注入 PYTHONUNBUFFERED=1(子进程异常日志即写即落)
- returncode≠0 且 job.log 无 doc_analyze_streaming 行 = 启动期失败 → 自动重试
  1 次(stdout 追加同一 job.log,日志留 [retry] 标记,重试块含重复启动命令行)
- 重试仍失败且仍无 pipeline 行 → status failed + 中文可读错误
  「提取启动阶段失败(重试后),可能受主机 GPU/显示重配置干扰,请重试」,
  附 job.log 原始错误尾部
- 已有 pipeline 行的中途失败不得触发重试,走原错误路径(含 CUDA OOM 中文错误)

测试:agent/{methodology,legacy-rag}/api/test_mineru_retry.py(两树 8 用例):
env 含 PYTHONUNBUFFERED;首败无 pipeline 行 → 重试;重试仍败 → 中文可读错误;
有 pipeline 行失败 → 不重试;json 直投跳过 mineru 段不重试;G6-2 边界。

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
2026-09-11 14:12:46 +08:00
co-authored by Claude Fable 5
parent 7840527f82
commit 3889a2a29b
4 changed files with 793 additions and 1 deletions
+69 -1
View File
@@ -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:
+328
View File
@@ -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 条 `$ <cmd>`;无重试追加
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
+68
View File
@@ -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:
+328
View File
@@ -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 条 `$ <cmd>`;无重试追加
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