vent_agent_2.0/tests/test_airflow_tools.py

"""风量研判工具层与真实资料回归测试。

真实资料（`矿井风量研判资料/`）不入库，缺失时相关用例跳过；合成数据与基准 fixture
始终可用，保证 CI 与离线环境仍能覆盖解析、核验、比对与报告全链路。
"""
from __future__ import annotations

import json
import re
from pathlib import Path

import pytest

import config
from algorithms.airflow_review.engine.findings import FindingFactory
from algorithms.airflow_review.engine.verify import verify_plan
from algorithms.airflow_review.parsers.measurement_report import parse_measurement_report
from algorithms.airflow_review.parsers.plan_preprocess import preprocess_plan_document
from algorithms.airflow_review.rules.loader import load_rules
from extensions.registry import load_extensions
from extensions.vent_airflow import monitoring, tools, workspace_paths

ROOT = Path(__file__).resolve().parents[1]
def raw_materials_dir() -> Path | None:
    """真实资料目录：位置由使用者决定，因此可探测或显式指定。

    依次看：AIRFLOW_RAW_DIR（本地便利开关）→ 仓库根 矿井风量研判资料。
    真实资料不入库，缺失时调用方应 skip。
    """
    import os

    candidates = []
    override = os.getenv("AIRFLOW_RAW_DIR", "").strip()
    if override:
        candidates.append(Path(override) if Path(override).is_absolute() else ROOT / override)
    candidates.append(ROOT / "矿井风量研判资料")
    for candidate in candidates:
        if candidate.is_dir() and (any(candidate.glob("*.pdf")) or any(candidate.glob("*.xlsx"))):
            return candidate
    return None


def _real_file(name: str) -> Path:
    directory = raw_materials_dir()
    return (directory / name) if directory else (ROOT / name)


REAL_REPORT = _real_file("测风报表20260528 .xlsx")
REAL_PLAN = _real_file("布尔台矿矿井风量分配计划20241201(1).pdf")
TABLE9_FIXTURE = ROOT / "tests" / "fixtures" / "airflow" / "bolitai_plan_table9.json"


@pytest.fixture(autouse=True)
def isolate_outputs(tmp_path, monkeypatch):
    """工具产物落到临时目录，避免测试污染仓库。"""
    monkeypatch.setattr(config, "AIRFLOW_OUTPUT_DIR", str(tmp_path / "airflow"))
    monkeypatch.setattr(config, "AIRFLOW_MONITORING_DIR", str(tmp_path / "monitoring"))


# ---- 扩展与路径 -----------------------------------------------------------

def test_extension_registers_all_tools_and_agent():
    catalog = load_extensions()
    expected = {
        "parse_airflow_measurement_report", "query_monitoring_airflow",
        "preprocess_air_plan_document", "validate_plan_extraction",
        "verify_and_extract_plan_airflow", "reconcile_airflow_sources",
        "compare_airflow_actual_vs_required", "trace_airflow_evidence",
        "render_airflow_review_report",
    }
    assert expected <= set(catalog.tools)
    assert "airflow-review-orchestrator" in catalog.agents
    agent = catalog.agents["airflow-review-orchestrator"]
    assert set(agent.tools) == expected
    assert "风量研判" in agent.description


def test_workspace_path_parity_with_agents_layer():
    """工具侧的沙箱路径映射必须与 agents.workspace 完全一致，否则线上找不到上传文件。"""
    from agents import workspace
    for user, session in (("alice", "s-1"), ("矿长", "会话-2")):
        assert workspace_paths.session_root(user, session) == workspace.session_root(user, session)


def test_resolve_input_rejects_escape_and_missing(tmp_path):
    with pytest.raises(workspace_paths.WorkspacePathError):
        workspace_paths.resolve_input("")
    with pytest.raises(workspace_paths.WorkspacePathError, match="无法解析|不存在"):
        workspace_paths.resolve_input("does/not/exist.xlsx")


# ---- 工具错误路径 ---------------------------------------------------------

def test_tools_return_readable_errors_instead_of_raising():
    assert tools.parse_airflow_measurement_report("nope.xlsx")["ok"] is False
    result = tools.validate_plan_extraction_tool("{不是 json")
    assert result["ok"] is False and "JSON" in result["error"]
    result = tools.compare_airflow_actual_vs_required(targets="[]")
    assert result["ok"] is False
    result = tools.render_airflow_review_report(findings='"string"')
    assert result["ok"] is False
    result = tools.trace_airflow_evidence(evidence_path="missing.json")
    assert result["ok"] is False


# ---- 监控适配器 -----------------------------------------------------------

def test_monitoring_source_missing_returns_insufficient(tmp_path):
    result = tools.query_monitoring_airflow(mine_id="M1", target_ids="L1",
                                            start_time="2026-05-01T00:00:00+08:00",
                                            end_time="2026-05-02T00:00:00+08:00")
    data = result["data"]
    assert data["status"] == "data_insufficient"
    assert data["manual_review_required"] is True
    assert "不得用测风报表代替" in data["note"]


def test_monitoring_source_aggregates_and_flags_anomalies(tmp_path):
    directory = tmp_path / "monitoring"
    directory.mkdir()
    samples = [{"at": f"2026-05-01T{h:02d}:{m:02d}:00+08:00", "airflow_m3_min": 800}
               for h in range(3) for m in range(0, 60, 10)]
    samples.append({"at": "2026-05-01T03:00:00+08:00", "airflow_m3_min": 400})   # 突跳
    samples.append({"at": "2026-05-01T07:00:00+08:00", "airflow_m3_min": 900})   # 断线后恢复
    (directory / "M1.json").write_text(json.dumps({
        "series": [{"sensor_id": "S1", "target_id": "L1", "target_type": "workplace",
                    "interval_minutes": 10, "samples": samples}]
    }, ensure_ascii=False), encoding="utf-8")
    result = monitoring.LocalFileMonitoringSource(directory).query(
        mine_id="M1", target_ids=["L1"], sensor_ids=[],
        start_time="2026-05-01T00:00:00+08:00", end_time="2026-05-01T23:59:59+08:00",
        aggregation="hour")
    series = result["series"][0]
    statistics = series["statistics"]
    assert statistics["min_m3_min"] == 400
    assert statistics["max_m3_min"] == 900
    assert statistics["valid_sample_rate"] < 1.0
    kinds = {anomaly["kind"] for anomaly in series["anomalies"]}
    assert "突跳" in kinds and "断线" in kinds
    assert series["sensor_status"] in ("abnormal", "offline")


# ---- 真实资料回归 ---------------------------------------------------------

@pytest.mark.skipif(not REAL_REPORT.is_file(), reason="真实测风报表未提供")
def test_real_measurement_report_totals_match_source():
    """真实报表的关键读数必须与原始文件一致（硬编码自 2026-05 旬报表头）。"""
    dataset = parse_measurement_report(REAL_REPORT, mine_id="BLT", report_period="2026-05")
    assert dataset["mine_total_intake"]["airflow_m3_min"] == 35181
    assert dataset["mine_total_return"]["airflow_m3_min"] == 35713
    assert dataset["effective_airflow"]["value"] == 32208
    assert round(dataset["effective_airflow_rate"]["value"], 2) == 91.55
    assert {fan["location_name"]: fan["airflow_m3_min"] for fan in dataset["main_fans"]} == {
        "明安木独回风立井总回风": pytest.approx(22512.6801, rel=1e-4),
        "松定霍洛回风立井总回风": pytest.approx(13716.1408, rel=1e-4),
    }
    assert [system["airflow_m3_min"] for system in dataset["return_systems"]] == [22193.0, 13520.0]
    # 主扇字段不能混用：运转机号是机组编号，功率另有输入/输出两列
    first_fan = dataset["main_fans"][0]
    assert first_fan["fan_unit"] == "一号主扇"
    assert first_fan["input_power_kw"] == pytest.approx(949.7437, rel=1e-4)
    assert first_fan["output_power_kw"] == pytest.approx(469.0142, rel=1e-4)
    assert first_fan["blade_angle_and_speed"] == "568/0"
    assert "motor_power" not in first_fan
    assert dataset["data_quality"]["intake_return_difference_m3_min"] == 532.0
    assert dataset["data_quality"]["record_count"] > 80
    # 未判定地点类型的测点必须进待复核，不能静默丢弃
    assert dataset["data_quality"]["unmapped_location_count"] == len(dataset["pending_review"]) - 0 or \
        dataset["data_quality"]["unmapped_location_count"] >= 1


@pytest.mark.skipif(not REAL_PLAN.is_file(), reason="真实配风计划未提供")
def test_real_plan_preprocess_is_faithful():
    """预处理产出的表格线索必须能对上计划声称的两套回风系统合计，否则说明文本被吞。"""
    document = preprocess_plan_document(REAL_PLAN, mine_id="BLT", month="2024-12")
    assert document["status"] == "ok"
    assert document["page_count"] == 63
    assert document["noise_removed"]["removed_header_lines"] > 50
    assert document["noise_removed"]["page_offset"] == 4
    sums = sorted(round(table["row_count"] and sum(row["value"] for row in table["rows"]), 2)
                  for table in document["tables"])
    assert 9228.0 in sums          # 松定霍洛回风系统
    assert 7282.0 in sums          # 采掘工作面计划风量
    assert any(abs(total - 17853.0) < 0.01 or 12230.0 == total or 5623.0 == total for total in sums)
    template = document["extraction_template"]
    assert template["rule_version"] == load_rules().rule_version
    assert len(template["formula_ids"]["采煤工作面"]) == 5
    assert template["target_schema"]["items"]


# ---- 全链路 ---------------------------------------------------------------

def test_verify_real_table9_fixture_is_exact():
    plan = json.loads(TABLE9_FIXTURE.read_text(encoding="utf-8"))
    result = verify_plan(plan, load_rules(), finding_factory=FindingFactory())
    assert result["plan_authoritative"] is True
    assert result["plan_calculation_status"] == "manual_review"   # 89 个地点缺计算过程
    required = {item["target_id"]: item["required_airflow_m3_min"] for item in result["verified_items"]}
    assert required == {"R-SDHL": 11073.6, "R-MAMD": 21423.6, "MINE_TOTAL": 32497.2}
    assert result["hierarchy"]["single_return_system"] is False
    assert result["hierarchy"]["issues"] == []
    assert len(result["hierarchy"]["levels"]["workplace"]) == 89


def test_tool_chain_verify_compare_trace_report(tmp_path):
    """工具串起来跑：核验真实 fixture → 比对（时间不可对齐）→ 证据链 → 报告。"""
    verify_result = tools.verify_and_extract_plan_airflow(str(TABLE9_FIXTURE), mine_id="BLT")
    assert verify_result["ok"] is not False
    verified = verify_result["data"]
    assert verified["plan_authoritative"] is True
    assert len(verified["verified_items"]) == 3

    dataset = {
        "dataset_id": "AIR-DATASET-BLT-2024-12", "report_period": "2024-12",
        "mine_total_intake": {"airflow_m3_min": 33000, "evidence_id": "e001"},
        "mine_total_return": {"airflow_m3_min": 32500, "evidence_id": "e002"},
        "return_systems": [{"location_name": "明安木独回风立井总回风", "airflow_m3_min": 10000,
                            "evidence_id": "e003"}],
        "locations": [], "records": [{"observed_at": "2024-12-15T09:00:00+08:00"}],
        "data_quality": {"record_count": 0, "unmapped_location_count": 0},
        "evidence": {
            "e001": {"evidence_id": "e001", "source_file": "测风报表.xlsx", "source_type": "measurement_report",
                     "normalized_value": 33000, "unit": "m3/min", "locator": {"sheet": "测风报表", "cell": "C11"},
                     "confidence": 0.98},
            "e002": {"evidence_id": "e002", "source_file": "测风报表.xlsx", "source_type": "measurement_report",
                     "normalized_value": 32500, "unit": "m3/min", "locator": {"sheet": "测风报表", "cell": "J7"},
                     "confidence": 0.98},
            "e003": {"evidence_id": "e003", "source_file": "测风报表.xlsx", "source_type": "measurement_report",
                     "normalized_value": 10000, "unit": "m3/min", "locator": {"sheet": "测风报表", "cell": "J5"},
                     "confidence": 0.98},
        },
    }
    dataset_path = tmp_path / "dataset.json"
    dataset_path.write_text(json.dumps(dataset, ensure_ascii=False), encoding="utf-8")

    comparison = tools.compare_airflow_actual_vs_required(
        dataset_path=str(dataset_path), verification_path=verified["plan_path"], mine_id="BLT")
    body = comparison["data"]
    # 计划生效窗口是 2024-12，合成数据集也在 2024-12 → 应能对齐并给出结论
    statuses = {record["target_name"]: record["comparison_status"] for record in body["comparisons"]}
    # 明安木独回风系统实测 10000 < 需风 11073.6 → 回风系统风量不足；矿井总风量正常
    assert statuses["明安木独回风立井回风系统"] == "major_risk"
    assert statuses["矿井总进风量"] == "normal"
    assert all(record["time_alignment_status"] == "aligned" for record in body["comparisons"])
    assert body["finding_summary"]["by_finding_type"]["return_system_airflow_shortage"] == 1

    trace = tools.trace_airflow_evidence(str(dataset_path), finding_id="F001")
    assert trace["data"]["evidence_chain"]

    report = tools.render_airflow_review_report(
        findings=body["result_path"], mine_id="BLT", period="2024-12",
        comparisons=body["result_path"], verified_items=verified["plan_path"],
        evidence_path=str(dataset_path), report_format="both")
    report_data = report["data"]
    assert report_data["report_id"] == "AIR-REVIEW-BLT-2024-12"
    assert report_data["finding_count"] >= 1
    assert Path(report_data["file_paths"]["md"]).is_file()
    text = Path(report_data["file_paths"]["md"]).read_text(encoding="utf-8")
    assert "本报告为智能辅助审查结果" in text
    assert "## 七、证据索引" in text
    if not report_data["unsupported_formats"]:
        assert Path(report_data["file_paths"]["docx"]).is_file()


def test_validate_then_verify_round_trip():
    """校验工具写出的规范化计划可以直接喂给核验工具。"""
    extraction = {
        "mine_id": "M1", "month": "2026-05", "plan_id": "P1", "plan_version": "V1",
        "approved_at": "2026-04-28",
        "effective_window": {"start": "2026-05-01T00:00:00+08:00", "end": "2026-05-31T23:59:59+08:00"},
        "items": [
            {"target_id": "L1", "target_type": "workplace", "target_name": "22210 综采工作面",
             "location_type": "采煤工作面", "parent_return_system_id": "R1",
             "planned_required_airflow_m3_min": 1221, "planned_formula": "WF-CFI-01",
             "planned_parameters": {"v_cfi": 1.0, "S_cfi": 11.3, "k_chi": 1.2, "k_cli": 1.5,
                                    "section_area_basis": "有效断面积", "q_cgi": 3.4, "k_cgi": 1.6,
                                    "q_cci": 0.5, "k_cci": 1.5, "l_cbi": 5.0, "l_csi": 4.4, "h_cfi": 3.0},
             "page": 10},
            {"target_id": "R1", "target_type": "return_air_system", "target_name": "唯一回风系统",
             "component_workplace_ids": ["L1"], "parent_mine_total_id": "MINE_TOTAL",
             "planned_required_airflow_m3_min": 1465.2,
             "planned_parameters": {"reserve_coefficient": 1.2}},
            {"target_id": "MINE_TOTAL", "target_type": "mine_total", "target_name": "矿井总需风量",
             "planned_required_airflow_m3_min": 1465.2, "component_return_system_ids": ["R1"]},
        ],
    }
    validated = tools.validate_plan_extraction_tool(json.dumps(extraction, ensure_ascii=False), mine_id="M1")
    assert validated["data"]["status"] in ("ok", "warning")
    assert validated["data"]["plan_path"]

    verified = tools.verify_and_extract_plan_airflow(validated["data"]["plan_path"], mine_id="M1")
    assert verified["data"]["plan_calculation_status"] == "valid"
    reuse = {item["target_id"]: item["required_reuse_key"] for item in verified["data"]["verified_items"]}
    assert reuse["R1"] == reuse["MINE_TOTAL"] != ""

def test_preprocess_tool_returns_template_and_cleaned_text(tmp_path):
    """预处理工具必须走通完整返回路径（含来源标注构造），输入用纯文本以脱离 PDF 依赖。"""
    plan_text = "\n".join([
        "布尔台煤矿矿井风量分配计划", "1", "一、主要配风地点：",
        "42 上205 综放工作面", "1523", "综放",
        "42 上208 主运顺槽工作面局扇安设点", "929", "掘进",
        "中央水仓变电所", "200", "硐室",
        "Q 采 1＝60×70%v 采 S 采 K 采高 K 采面长", "=1221m3/min",
    ])
    source = tmp_path / "plan.txt"
    source.write_text(plan_text, encoding="utf-8")
    result = tools.preprocess_air_plan_document(str(source), mine_id="M1", month="2026-05")
    assert result.get("ok") is True, result
    body = result["data"]
    assert body["cleaned_text_path"]
    assert Path(body["cleaned_text_path"]).is_file()
    assert body["extraction_template"]["formula_ids"]["采煤工作面"]
    assert result["sources"], "预处理结果必须带来源标注"
    assert body["extraction_template"]["target_schema"]["items"]


# ---- 技能包结构 -----------------------------------------------------------

SKILL_DIR = ROOT / "skills" / "airflow-review"
OLD_SKILL_NAMES = (
    "airflow-data-preparation", "air-distribution-plan-verification",
    "mine-system-airflow-assessment", "workplace-airflow-assessment",
    "multi-source-airflow-reconciliation", "airflow-review-reporting",
    "airflow-review-orchestrator",
)


def test_skill_is_discovered_by_framework_and_old_ones_are_gone():
    """框架只扫 skills/<name>/SKILL.md 一层；合并后应只发现一个风量研判技能。"""
    from deepagents.backends import FilesystemBackend
    from deepagents.middleware.skills import _list_skills

    backend = FilesystemBackend(root_dir=str(ROOT), virtual_mode=True)
    listed = _list_skills(backend, "/skills/")
    # 不同 deepagents 版本返回 dict 或 SkillMetadata，两种都兼容
    skills = {entry["name"] if isinstance(entry, dict) else entry.name: entry for entry in listed}
    assert "airflow-review" in skills
    assert not (set(OLD_SKILL_NAMES) & set(skills)), f"旧技能目录应已删除：{sorted(set(OLD_SKILL_NAMES) & set(skills))}"
    for name in OLD_SKILL_NAMES:
        assert not (ROOT / "skills" / name).exists(), f"旧技能目录残留：{name}"
    entry = skills["airflow-review"]
    path = entry["path"] if isinstance(entry, dict) else entry.path
    assert path == "/skills/airflow-review/SKILL.md"


def test_skill_bundle_frontmatter_and_references_resolve():
    """SKILL.md 的 name 必须等于目录名；它引用的每一份子文件都必须真实存在。"""
    import yaml

    content = (SKILL_DIR / "SKILL.md").read_text(encoding="utf-8")
    front = yaml.safe_load(content.split("---")[1])
    assert front["name"] == SKILL_DIR.name == "airflow-review"
    assert len(front["description"]) <= 1024, "description 超过框架上限 1024 字符"
    assert "75%" in front["description"] and "配风计划" in front["description"]

    referenced = set(re.findall(r"references/[\w./-]+\.md", content))
    assert len(referenced) >= 8, f"SKILL.md 应索引全部子文件，实际 {sorted(referenced)}"
    for relative in referenced:
        assert (SKILL_DIR / relative).is_file(), f"SKILL.md 引用了不存在的文件：{relative}"

    # 每份子文件都要能被 SKILL.md 索引到（防止加了文件却没人读）
    on_disk = {path.relative_to(SKILL_DIR).as_posix() for path in (SKILL_DIR / "references").rglob("*.md")}
    assert on_disk == referenced, f"未被索引的文件：{sorted(on_disk - referenced)}"


def test_contract_doc_matches_registered_tools():
    """契约文档与实际注册的工具必须一一对应，避免文档漂移。"""
    from extensions.vent_airflow.extension import TOOL_NAMES

    doc = (SKILL_DIR / "references" / "contracts" / "tools.md").read_text(encoding="utf-8")
    documented = set(re.findall(r"^## \d+\. (\w+)$", doc, flags=re.M))
    assert documented == set(TOOL_NAMES), (
        f"文档多出：{sorted(documented - set(TOOL_NAMES))}；文档缺失：{sorted(set(TOOL_NAMES) - documented)}")
    catalog = load_extensions()
    for name in TOOL_NAMES:
        assert name in catalog.tools, f"文档中声明但未注册：{name}"


def test_disclaimer_matches_renderer_across_skill_and_prompt():
    """免责声明在渲染器、技能入口与子 Agent 提示词里必须是同一句话（需求 §2.3 工具 8）。"""
    from algorithms.airflow_review.report.render import DISCLAIMER

    skill = (SKILL_DIR / "SKILL.md").read_text(encoding="utf-8")
    prompt = (ROOT / "extensions" / "vent_airflow" / "prompt.md").read_text(encoding="utf-8")
    assert DISCLAIMER in skill, "技能入口未声明免责声明原文"
    assert DISCLAIMER in prompt, "子 Agent 提示词未声明免责声明原文"


def test_release_packaging_includes_skill_references():
    """打包白名单必须包含技能包里的 references——漏了上线才会发现技能读不到子流程。"""
    import sys as _sys

    _sys.path.insert(0, str(ROOT / "release"))
    import build as release_build

    packaged = {path.relative_to(ROOT).as_posix() for path in release_build.source_files(ROOT)}
    on_disk = {path.relative_to(ROOT).as_posix()
               for path in SKILL_DIR.rglob("*") if path.is_file()}
    assert on_disk <= packaged, f"未被打包：{sorted(on_disk - packaged)}"


LLM_EXTRACTION_FIXTURE = ROOT / "tests" / "fixtures" / "airflow" / "bolitai_plan_extraction.llm.json"


@pytest.mark.skipif(not LLM_EXTRACTION_FIXTURE.is_file(), reason="未保存模型真实提取结果")
def test_real_llm_extraction_verifies_without_false_errors():
    """真实模型对真实 PDF 的提取结果必须能核验通过，不能被误判成计算错误。

    这份 fixture 是一次真实运行留下的产物：92 个对象，风量值全部来自计划原文。
    它固定住了两个曾经踩过的坑——模型把参数口径写成同义说法（"平均有效净断面积"），
    以及自洽核验只认声明口径。两者都会把正确计划整盘阻断。
    """
    validated = tools.validate_plan_extraction_tool(str(LLM_EXTRACTION_FIXTURE), mine_id="布尔台矿")
    assert validated.get("ok") is True, validated
    body = validated["data"]
    assert body["item_count"] == 92
    assert body["plan_path"]

    verified = tools.verify_and_extract_plan_airflow(body["plan_path"], mine_id="布尔台矿")
    data = verified["data"]
    assert data["plan_authoritative"] is True
    assert data["blocked_items"] == [], "正确计划不应被阻断"
    required = {item["target_id"]: item["required_airflow_m3_min"] for item in data["verified_items"]}
    assert required == {"RS-SDHL": 11073.6, "RS-MAMD": 21423.6, "MT-BLT": 32497.2}
    assert not [f for f in data["findings"] if f.get("severity") == "major"], "不应产生重大风险问题"


def test_validate_tool_returns_validation_details_instead_of_empty_error():
    """校验发现阻断性问题时，必须把 validation 明细交给模型，不能塌缩成一句空错误。"""
    extraction = {
        "mine_id": "M1", "month": "2026-05", "plan_id": "P1", "plan_version": "V1",
        "effective_window": {"start": "2026-05-01T00:00:00+08:00", "end": "2026-05-31T23:59:59+08:00"},
        "items": [
            {"target_id": "L1", "target_type": "workplace", "target_name": "某个面",
             "parent_return_system_id": "R-missing", "planned_required_airflow_m3_min": 900},
        ],
    }
    result = tools.validate_plan_extraction_tool(json.dumps(extraction, ensure_ascii=False), mine_id="M1")
    assert result.get("ok") is True
    body = result["data"]
    assert body["status"] == "error"           # 校验判定为有阻断性问题
    assert body["item_count"] == 1
    codes = {issue["code"] for issue in body["validation"]["issues"]}
    assert "invalid_parent" in codes or "missing_aggregation" in codes
    assert body["next_step"] and "修正" in body["next_step"]


# ---- 入参容错与工具安全网 -------------------------------------------------

def test_tool_schemas_expose_documented_parameters():
    """工具包装（含异常安全网）不能吃掉入参 schema——吃到参数模型就没法调用工具了。"""
    from extensions.vent_airflow.extension import TOOL_NAMES
    from tools import registry

    catalog = load_extensions()
    expected_min = {
        "parse_airflow_measurement_report": {"file_path", "mine_id", "report_period", "scope"},
        "query_monitoring_airflow": {"mine_id", "aggregation"},
        "preprocess_air_plan_document": {"file_path", "mine_id", "month"},
        "validate_plan_extraction": {"extraction", "mine_id"},
        "verify_and_extract_plan_airflow": {"plan_path", "mine_id", "rule_version"},
        "reconcile_airflow_sources": {"targets"},
        "compare_airflow_actual_vs_required": {"targets", "dataset_path", "verification_path"},
        "trace_airflow_evidence": {"evidence_path", "finding_id", "evidence_ids"},
        "render_airflow_review_report": {"findings", "mine_id", "period", "report_format"},
    }
    assert set(expected_min) == set(TOOL_NAMES)
    for name, expected in expected_min.items():
        tool = registry.spec_as_tool(catalog.tools[name])
        assert expected <= set(tool.args), f"{name} 少了参数：{sorted(expected - set(tool.args))}"


def test_compare_tool_tolerates_windows_written_as_strings():
    """真实故障回归：模型把 required_effective_window 写成字符串，曾经直接崩掉整轮。"""
    targets = [{
        "target_id": "L1", "target_type": "workplace", "location_type": "采煤工作面",
        "target_name": "某采面", "actual_airflow_m3_min": 800, "required_airflow_m3_min": 900,
        "plan_verification_status": "valid",
        "actual_window": "2024年12月15日",
        "required_effective_window": "2024年12月1日 ~ 2024年12月31日",
    }]
    result = tools.compare_airflow_actual_vs_required(targets=json.dumps(targets, ensure_ascii=False))
    assert result.get("ok") is True, result
    record = result["data"]["comparisons"][0]
    assert record["time_alignment_status"] == "aligned"
    assert record["required_effective_window"].get("normalized_from") == "string"
    assert result["data"]["finding_summary"]["by_finding_type"] == {"workplace_airflow_below_75_percent": 1}


def test_compare_tool_marks_manual_review_for_unparsable_window():
    """时间窗实在认不出来时转人工复核，而不是抛错、也不是硬判正常。"""
    targets = [{
        "target_id": "L1", "target_type": "workplace", "target_name": "某面",
        "actual_airflow_m3_min": 800, "required_airflow_m3_min": 900,
        "plan_verification_status": "valid",
        "actual_window": "本月中旬", "required_effective_window": {"start": "2024-12-01T00:00:00+08:00",
                                                                  "end": "2024-12-31T23:59:59+08:00"},
    }]
    result = tools.compare_airflow_actual_vs_required(targets=json.dumps(targets, ensure_ascii=False))
    record = result["data"]["comparisons"][0]
    assert record["comparison_status"] == "insufficient"
    assert result["data"]["findings"][0]["manual_review_required"] is True


def test_reconcile_tool_tolerates_string_numbers_and_dates():
    request = [{
        "target_id": "L1", "target_type": "workplace", "target_name": "某面",
        "comparison_window": "2024年12月15日 09:00 至 2024年12月15日 23:59",
        "sources": [
            {"source_type": "measurement_report", "airflow_m3_min": "860", "observed_at": "2024年12月15日 09:00"},
            {"source_type": "monitoring", "airflow_m3_min": 820, "observed_at": "2024-12-15T09:30:00+08:00"},
        ],
    }]
    result = tools.reconcile_airflow_sources_tool(targets=json.dumps(request, ensure_ascii=False))
    first = result["data"]["results"][0]
    assert first["consistency"] == "acceptable"
    assert first["deviation_percent"] == pytest.approx(4.878, rel=1e-3)


def test_tools_return_readable_error_when_engine_raises(monkeypatch):
    """安全网：引擎抛意外异常时返回可读错误，而不是把整轮对话打断。"""
    def boom(*_args, **_kwargs):
        raise RuntimeError("模拟引擎内部错误")

    monkeypatch.setattr(tools, "compare_targets", boom)
    result = tools.compare_airflow_actual_vs_required(targets=json.dumps([{"target_id": "L1"}]))
    assert result["ok"] is False
    assert "工具内部错误" in result["error"] and "RuntimeError" in result["error"]
    assert result["hint"]


def test_validate_tool_never_raises_on_garbage_items():
    """items 里混入字符串、参数写成数组等畸形输入，只能得到可读错误或校验问题。"""
    extraction = {"plan_id": "P", "plan_version": "V", "effective_window": "2024年12月",
                  "items": ["不是对象", {"target_id": "L1", "target_type": "workplace",
                                        "target_name": "面", "planned_parameters": ["a", "b"]}]}
    result = tools.validate_plan_extraction_tool(json.dumps(extraction, ensure_ascii=False), mine_id="M1")
    assert "error" not in result or "工具内部错误" not in str(result.get("error", ""))
    if result.get("ok"):
        codes = {issue["code"] for issue in result["data"]["validation"]["issues"]}
        assert "invalid_item" in codes or "unparsable_effective_window" in codes


def test_report_filename_is_sanitized_but_report_id_is_not(tmp_path):
    """真实故障回归：模型把期间写成含斜杠的一长串，斜杠被当路径分隔符 → 写文件失败。

    文件名必须安全化（斜杠、Windows 非法字符、结尾点空格），而 report_id 保持原样用于
    展示与追溯——两者不能混为一谈。
    """
    from algorithms.airflow_review.report.render import render_report

    messy = "2024-12（配风计划）/ 2026-05-28（实测）: <草稿>"
    result = render_report(mine_id="布尔台矿", period=messy, findings=[],
                           report_format="both", output_dir=tmp_path)
    assert result["report_id"].endswith(messy)          # 逻辑 ID 原样
    assert result["files"], result
    for path in result["files"].values():
        written = Path(path)
        assert written.is_file()
        assert written.parent == tmp_path               # 没有被当成子目录
        assert not set(written.name) & set('/:*?"<>|\\')
        assert result["report_id"] not in written.name  # 用的是安全化后的名字