adminandClaude Fable 5
3889a2a29b
缺陷 #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>
329 lines
14 KiB
Python
329 lines
14 KiB
Python
"""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
|