rapid-structure-rs 1.0.0

use std::sync::Arc;
use std::time::Instant;

use crate::{
    api::OcrEngine as _,
    error::Result,
    input::image_loader::{LoadImage, OcrInput, ensure_decode_pixels},
    ocr::cls::classifier::Classifier,
    ocr::config::RecognizeOptions,
    ocr::det::detector::Detector,
    ocr::pipeline::{
        config::EngineConfig,
        image_ops::{
            PreprocessRecord, apply_vertical_padding, crop_text_regions, map_boxes_to_original,
            map_img_to_original, resize_image_within_bounds,
        },
        types::{ExecutionOptions, ExecutionOutput},
    },
    ocr::rec::recognizer::Recognizer,
    ocr::types::{LineResult, WordBox},
    runtime::{profile::RuntimeProfile, provider::ProviderResolution},
};

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct PipelineProviderResolutions {
    pub det: ProviderResolution,
    pub cls: Option<ProviderResolution>,
    pub rec: ProviderResolution,
}

#[derive(Debug, Clone, Copy)]
struct RunSwitches {
    use_det: bool,
    use_cls: bool,
    apply_cls_rotation: bool,
    use_rec: bool,
    need_stage_images: bool,
    return_word_box: bool,
    return_single_char_box: bool,
    text_score: f32,
}

#[derive(Debug)]
struct PreparedImage {
    ori_h: usize,
    ori_w: usize,
    ratio_h: f32,
    ratio_w: f32,
    preprocess_record: PreprocessRecord,
    proc_img: crate::config::RecImage,
    decode_ms: f32,
    resize_ms: f32,
}

#[derive(Debug, Default)]
struct RunBuffers {
    det_boxes: Vec<crate::Quad>,
    det_scores: Vec<f32>,
    stage_images: Vec<crate::config::RecImage>,
    lines: Vec<LineResult>,
}

#[derive(Debug)]
pub struct RapidOcr {
    config: EngineConfig,
    profile: RuntimeProfile,
    detector: Detector,
    classifier: Option<Classifier>,
    recognizer: Recognizer,
    loader: LoadImage,
    /// 性能剖析时激活的阶段级内存采样开关。
    mem_probe_enabled: bool,
    /// 每次 run() 期间采集的分阶段内存记录；调用前由 take_stage_mem_records() 取出。
    stage_mem_records: Vec<crate::runtime::memory::StageMemRecord>,
    rec_batch_records: Vec<crate::ocr::rec::recognizer::BatchRecord>,
    rec_batch_probe_enabled: bool,
    last_det_input_size: Option<(u32, u32)>,
}

impl RapidOcr {
    pub fn new(config: EngineConfig) -> Result<Self> {
        // 唯一一次解析运行时档案：线程 plan 与 provider/arena/公式批大小都从这里出发。
        // Rayon 全局线程池的初始化也在这里，失败会向上传播（过去是 `let _ = ...` 静默吞掉）。
        let profile = RuntimeProfile::resolve(&config.runtime, config.global.use_cls)?;
        let session_runtime = profile.session_runtime();
        let det = Detector::new(config.det.clone(), &session_runtime)?;
        let cls = if config.global.use_cls {
            Some(Classifier::new(config.cls.clone(), &session_runtime)?)
        } else {
            None
        };
        let rec = Recognizer::new(config.rec.clone(), &session_runtime)?;
        Ok(Self {
            config,
            profile,
            detector: det,
            classifier: cls,
            recognizer: rec,
            loader: LoadImage::default(),
            mem_probe_enabled: false,
            stage_mem_records: Vec::new(),
            rec_batch_records: Vec::new(),
            rec_batch_probe_enabled: false,
            last_det_input_size: None,
        })
    }

    /// 引擎唯一的运行时档案：基准报告与公式会话都从这里取设置。
    pub fn runtime_profile(&self) -> &RuntimeProfile {
        &self.profile
    }

    /// 激活阶段级内存采样（profiling 专用）。
    pub fn enable_mem_probe(&mut self) {
        self.mem_probe_enabled = true;
    }

    /// 取出并清空累计的阶段内存记录。
    pub fn take_stage_mem_records(&mut self) -> Vec<crate::runtime::memory::StageMemRecord> {
        std::mem::take(&mut self.stage_mem_records)
    }

    pub(crate) fn run(
        &mut self,
        input: OcrInput,
        opts: ExecutionOptions,
    ) -> Result<ExecutionOutput> {
        let e2e_start = Instant::now();
        let mut output = ExecutionOutput::default();
        self.rec_batch_records.clear();
        self.last_det_input_size = None;
        let switches = self.resolve_run_switches(&opts);
        if self.mem_probe_enabled {
            crate::runtime::memory::tl_probe_begin();
        }
        crate::runtime::memory::tl_probe("input");
        let mut prepared = self.prepare_image(input)?;
        output.decode_ms = Some(prepared.decode_ms);
        output.resize_ms = Some(prepared.resize_ms);
        output.processed_size = Some((
            prepared.proc_img.width() as u32,
            prepared.proc_img.height() as u32,
        ));
        let mut buffers = RunBuffers::default();
        crate::runtime::memory::tl_probe("input_end");

        crate::runtime::memory::tl_probe("detect_pre");
        if !self.run_detection_stage(&opts, switches, &mut prepared, &mut buffers, &mut output)? {
            output.e2e_ms = Some(e2e_start.elapsed().as_secs_f32() * 1000.0);
            crate::runtime::memory::tl_probe("output_end");
            if self.mem_probe_enabled {
                self.stage_mem_records
                    .extend(crate::runtime::memory::tl_probe_end());
            }
            self.last_det_input_size = output.det_input_size;
            return Ok(output);
        }

        crate::runtime::memory::tl_probe("classifier");
        self.run_classification_stage(switches, &mut buffers, &mut output)?;

        crate::runtime::memory::tl_probe("rec_pre");
        self.run_recognition_stage(switches, &mut buffers, &mut output)?;

        crate::runtime::memory::tl_probe("page_post");
        let postprocess_start = Instant::now();
        self.finalize_detection_outputs(switches, &prepared, &mut buffers, &mut output)?;
        self.finalize_recognition_outputs(switches, &buffers.lines, &mut output);
        output.postprocess_ms = Some(postprocess_start.elapsed().as_secs_f32() * 1000.0);

        output.e2e_ms = Some(e2e_start.elapsed().as_secs_f32() * 1000.0);
        crate::runtime::memory::tl_probe("output_end");
        if self.mem_probe_enabled {
            self.stage_mem_records
                .extend(crate::runtime::memory::tl_probe_end());
        }
        self.rec_batch_records = output.rec_batch_records.clone();
        self.last_det_input_size = output.det_input_size;
        Ok(output)
    }

    fn resolve_run_switches(&self, opts: &ExecutionOptions) -> RunSwitches {
        let use_det = opts.use_det;
        let use_cls = opts.use_cls;
        let use_rec = opts.use_rec;
        RunSwitches {
            use_det,
            use_cls,
            apply_cls_rotation: opts.apply_cls_rotation,
            use_rec,
            need_stage_images: use_cls || use_rec,
            return_word_box: opts.return_word_box,
            return_single_char_box: opts.return_single_char_box,
            text_score: opts.text_score.unwrap_or(self.config.global.text_score),
        }
    }

    fn prepare_image(&mut self, input: OcrInput) -> Result<PreparedImage> {
        let decode_start = Instant::now();
        let ori_img = self.loader.load(input)?;
        let decode_ms = decode_start.elapsed().as_secs_f32() * 1000.0;
        let ori_h = ori_img.height();
        let ori_w = ori_img.width();
        let resize_start = Instant::now();
        let (proc_img, ratio_h, ratio_w) = resize_image_within_bounds(
            ori_img,
            self.config.global.min_side_len,
            self.config.global.max_side_len,
        )?;
        let resize_ms = resize_start.elapsed().as_secs_f32() * 1000.0;

        Ok(PreparedImage {
            ori_h,
            ori_w,
            ratio_h,
            ratio_w,
            preprocess_record: PreprocessRecord {
                ratio_h,
                ratio_w,
                ..PreprocessRecord::default()
            },
            proc_img,
            decode_ms,
            resize_ms,
        })
    }

    fn run_detection_stage(
        &mut self,
        opts: &ExecutionOptions,
        switches: RunSwitches,
        prepared: &mut PreparedImage,
        buffers: &mut RunBuffers,
        output: &mut ExecutionOutput,
    ) -> Result<bool> {
        if switches.use_det {
            let (padded, pad_top) = apply_vertical_padding(
                prepared.proc_img.clone(),
                self.config.global.width_height_ratio,
                self.config.global.min_height,
            )?;
            prepared.proc_img = padded;
            prepared.preprocess_record.pad_top = pad_top;

            self.detector
                .update_postprocess(opts.box_thresh, opts.unclip_ratio);
            let det_out = self.detector.detect(&prepared.proc_img)?;
            output.det_input_size = Some(det_out.input_size);
            if det_out.boxes.is_empty() {
                return Ok(false);
            }

            output.elapsed_ms[0] = Some(det_out.elapsed_ms);
            output.det_breakdown_ms = det_out.breakdown;
            buffers.det_boxes = det_out.boxes;
            buffers.det_scores = det_out.scores;

            if switches.need_stage_images {
                let crop_start = Instant::now();
                buffers.stage_images = crop_text_regions(&prepared.proc_img, &buffers.det_boxes)?;
                output.crop_ms = Some(crop_start.elapsed().as_secs_f32() * 1000.0);
            }
        } else if switches.need_stage_images {
            buffers.stage_images.push(prepared.proc_img.clone());
        }

        Ok(true)
    }

    fn run_classification_stage(
        &mut self,
        switches: RunSwitches,
        buffers: &mut RunBuffers,
        output: &mut ExecutionOutput,
    ) -> Result<()> {
        if switches.use_cls {
            let classifier = self.classifier.as_mut().ok_or_else(|| {
                crate::error::RapidOcrError::Config(
                    "classification requested but no classifier model is configured".to_string(),
                )
            })?;
            let cls_result = classifier
                .classify_in_place(&mut buffers.stage_images, switches.apply_cls_rotation)?;
            output.cls_res = Some(cls_result.cls_res);
            output.elapsed_ms[1] = Some(cls_result.elapsed_ms);
            output.cls_breakdown_ms = Some([
                cls_result.preprocess_ms,
                cls_result.infer_ms,
                cls_result.postprocess_ms,
            ]);
        }
        Ok(())
    }

    fn run_recognition_stage(
        &mut self,
        switches: RunSwitches,
        buffers: &mut RunBuffers,
        output: &mut ExecutionOutput,
    ) -> Result<()> {
        if !switches.use_rec {
            return Ok(());
        }

        if self.rec_batch_probe_enabled {
            self.recognizer.batch_log = Some(Vec::new());
        }
        let rec = self.recognizer.recognize(
            &buffers.stage_images,
            RecognizeOptions {
                return_word_box: switches.return_word_box,
                return_single_char_box: switches.return_single_char_box,
            },
        )?;
        output.rec_batch_records = self.recognizer.batch_log.take().unwrap_or_default();
        output.elapsed_ms[2] = Some(rec.elapsed.as_secs_f32() * 1000.0);
        output.rec_breakdown_ms = Some([rec.preprocess_ms, rec.infer_ms, rec.postprocess_ms]);
        buffers.lines = rec.lines;
        Ok(())
    }

    fn finalize_detection_outputs(
        &mut self,
        switches: RunSwitches,
        prepared: &PreparedImage,
        buffers: &mut RunBuffers,
        output: &mut ExecutionOutput,
    ) -> Result<()> {
        if !switches.use_det {
            return Ok(());
        }

        let mut mapped_boxes = std::mem::take(&mut buffers.det_boxes);
        map_boxes_to_original(
            &mut mapped_boxes,
            prepared.preprocess_record,
            prepared.ori_h,
            prepared.ori_w,
        );

        if !switches.use_rec {
            output.boxes = Some(mapped_boxes);
            output.det_scores = Some(std::mem::take(&mut buffers.det_scores));
            return Ok(());
        }

        let lines = std::mem::take(&mut buffers.lines);
        let det_scores = std::mem::take(&mut buffers.det_scores);
        let (filtered_boxes, filtered_scores, filtered_lines, kept_indices) =
            filter_empty_lines_boxes_and_scores(mapped_boxes, det_scores, lines);
        buffers.lines = filtered_lines;

        let mut computed_word_boxes = None;
        if switches.return_word_box && !filtered_boxes.is_empty() && !buffers.lines.is_empty() {
            let mapped_crops =
                map_img_to_original(&buffers.stage_images, prepared.ratio_h, prepared.ratio_w)?;
            let filtered_crops = select_items_by_indices(mapped_crops, &kept_indices);
            let word_boxes = crate::ocr::rec::word_boxes::compute_word_boxes(
                &filtered_crops,
                &filtered_boxes,
                &buffers.lines,
                switches.return_single_char_box,
            )?;
            computed_word_boxes = Some(word_boxes);
        }

        let (
            score_filtered_boxes,
            score_filtered_scores,
            score_filtered_lines,
            score_filtered_words,
        ) = filter_by_text_score_for_full(
            filtered_boxes,
            filtered_scores,
            std::mem::take(&mut buffers.lines),
            computed_word_boxes,
            switches.text_score,
        );

        buffers.lines = score_filtered_lines;
        output.boxes = Some(score_filtered_boxes);
        output.det_scores = Some(score_filtered_scores);
        output.word_boxes = score_filtered_words;
        Ok(())
    }

    fn finalize_recognition_outputs(
        &self,
        switches: RunSwitches,
        lines: &[LineResult],
        output: &mut ExecutionOutput,
    ) {
        if !switches.use_rec || lines.is_empty() {
            return;
        }

        // Keep parity with Python: text_score only filters full outputs (det+rec),
        // rec-only mode returns raw recognition results.
        output.txts = Some(lines.iter().map(|v| v.text.clone()).collect());
        output.scores = Some(lines.iter().map(|v| v.score).collect());
        output.lines = Some(lines.to_vec());
    }

    pub fn provider_resolutions(&self) -> PipelineProviderResolutions {
        PipelineProviderResolutions {
            det: self.detector.provider_resolution(),
            cls: self
                .classifier
                .as_ref()
                .map(Classifier::provider_resolution),
            rec: self.recognizer.provider_resolution(),
        }
    }

    pub fn model_paths(
        &self,
    ) -> (
        Option<&std::path::Path>,
        Option<&std::path::Path>,
        Option<&std::path::Path>,
    ) {
        (
            Some(self.detector.model_path()),
            self.classifier.as_ref().map(Classifier::model_path),
            Some(self.recognizer.model_path()),
        )
    }

    pub fn take_rec_batch_records(&mut self) -> Vec<crate::ocr::rec::recognizer::BatchRecord> {
        std::mem::take(&mut self.rec_batch_records)
    }

    pub fn enable_rec_batch_probe(&mut self) {
        self.rec_batch_probe_enabled = true;
    }

    pub fn last_detector_input_size(&self) -> Option<(u32, u32)> {
        self.last_det_input_size
    }

    pub fn set_max_side_len(&mut self, max_side_len: u32) {
        self.config.global.max_side_len = max_side_len.max(32) as usize;
    }

    pub fn set_image_bounds(&mut self, min_side_len: u32, max_side_len: u32) {
        self.config.global.min_side_len = min_side_len.max(1) as usize;
        self.config.global.max_side_len = max_side_len.max(32) as usize;
    }
}

type FullFilterOutput = (
    Vec<crate::Quad>,
    Vec<f32>,
    Vec<LineResult>,
    Option<Vec<Vec<WordBox>>>,
);

fn filter_empty_lines_boxes_and_scores(
    boxes: Vec<crate::Quad>,
    det_scores: Vec<f32>,
    lines: Vec<LineResult>,
) -> (Vec<crate::Quad>, Vec<f32>, Vec<LineResult>, Vec<usize>) {
    if boxes.len() != lines.len() || det_scores.len() != lines.len() {
        return (boxes, det_scores, lines, Vec::new());
    }
    let mut out_boxes = Vec::new();
    let mut out_scores = Vec::new();
    let mut out_lines = Vec::new();
    let mut kept_indices = Vec::new();
    for (idx, ((b, s), l)) in boxes.into_iter().zip(det_scores).zip(lines).enumerate() {
        if l.text.trim().is_empty() {
            continue;
        }
        out_boxes.push(b);
        out_scores.push(s);
        out_lines.push(l);
        kept_indices.push(idx);
    }
    (out_boxes, out_scores, out_lines, kept_indices)
}

fn select_items_by_indices<T>(items: Vec<T>, indices: &[usize]) -> Vec<T> {
    if indices.is_empty() {
        return Vec::new();
    }

    let mut out = Vec::with_capacity(indices.len());
    let mut keep_iter = indices.iter().copied().peekable();
    for (idx, item) in items.into_iter().enumerate() {
        let Some(target_idx) = keep_iter.peek().copied() else {
            break;
        };
        if idx == target_idx {
            out.push(item);
            keep_iter.next();
        }
    }
    out
}

fn filter_by_text_score_for_full(
    boxes: Vec<crate::Quad>,
    det_scores: Vec<f32>,
    lines: Vec<LineResult>,
    word_boxes: Option<Vec<Vec<WordBox>>>,
    text_score: f32,
) -> FullFilterOutput {
    let mut out_boxes = Vec::new();
    let mut out_scores = Vec::new();
    let mut out_lines = Vec::new();
    let mut out_word_boxes = Vec::new();

    for (idx, line) in lines.into_iter().enumerate() {
        if idx >= boxes.len() || idx >= det_scores.len() {
            break;
        }
        if line.score < text_score {
            continue;
        }
        out_boxes.push(boxes[idx]);
        out_scores.push(det_scores[idx]);
        if let Some(word_line) = word_boxes.as_ref().and_then(|v| v.get(idx)) {
            out_word_boxes.push(word_line.clone());
        }
        out_lines.push(line);
    }

    let out_word_boxes = if word_boxes.is_some() {
        Some(out_word_boxes)
    } else {
        None
    };

    (out_boxes, out_scores, out_lines, out_word_boxes)
}

fn rec_image_input(
    image: crate::config::RecImage,
    roi: Option<crate::api::RectU32>,
    max_decode_pixels: u64,
) -> Result<(OcrInput, crate::api::ImageSize, (u32, u32))> {
    let original_size = crate::api::ImageSize {
        width: image.width() as u32,
        height: image.height() as u32,
    };
    ensure_decode_pixels(
        original_size.width as usize,
        original_size.height as usize,
        max_decode_pixels,
    )?;
    let Some(roi) = roi else {
        return Ok((OcrInput::Image(image), original_size, (0, 0)));
    };
    roi.validate_against(original_size.width, original_size.height)?;
    let src = image.as_bgr_bytes();
    let row_bytes = roi.width as usize * 3;
    let mut out = vec![0_u8; row_bytes * roi.height as usize];
    for y in 0..roi.height as usize {
        let src_start = ((roi.y as usize + y) * image.width() + roi.x as usize) * 3;
        let dst_start = y * row_bytes;
        out[dst_start..dst_start + row_bytes]
            .copy_from_slice(&src[src_start..src_start + row_bytes]);
    }
    let cropped =
        crate::config::RecImage::from_bgr_u8(roi.width as usize, roi.height as usize, out)?;
    Ok((OcrInput::Image(cropped), original_size, (roi.x, roi.y)))
}

/// 一个已物化的请求输入：**整页**（未裁 roi）的 `RecImage` + 页尺寸。
///
/// 这是"把请求变成像素"的唯一入口（P1-5）。roi 不在这里应用：文本与分块两条路径各自
/// 把 `rec` 裁到 roi（`rec_image_input` / `crop_imm`），而页面级阶段的 `decode_source_image`
/// 需要**未抹白、未裁的整页像素**。`Encoded` 输入在此只解码一次（`load_with_limit`）；
/// 页面级阶段的第二次解码由 `decode_source_image` 保持，故整条页级路径的解码次数仍为 2。
struct MaterializedInput {
    rec: crate::config::RecImage,
    original_size: crate::api::ImageSize,
}

fn rec_image_to_rgba(image: &crate::config::RecImage) -> Result<image::RgbaImage> {
    let bgr = image.as_bgr_cow();
    let mut rgba = vec![0_u8; image.width() * image.height() * 4];
    for (src, dst) in bgr
        .as_chunks::<3>()
        .0
        .iter()
        .zip(rgba.as_chunks_mut::<4>().0.iter_mut())
    {
        dst[0] = src[2];
        dst[1] = src[1];
        dst[2] = src[0];
        dst[3] = 255;
    }
    image::RgbaImage::from_raw(image.width() as u32, image.height() as u32, rgba)
        .ok_or_else(|| crate::error::RapidOcrError::InvalidImage("invalid image buffer".into()))
}

#[derive(Debug)]
pub struct RapidOcrEngine {
    inner: RapidOcr,
    model_id: String,
    base_min_side_len: usize,
    base_max_side_len: usize,
    /// 懒加载的公式识别器（**会话缓存键** -> 实例），避免每次请求重建 594 MB session。
    ///
    /// 键见 [`SessionCacheKey`]：文件身份之外还必须带上调用方这一次**声明的期望摘要**。
    formula_recognizer: Option<(
        SessionCacheKey,
        crate::formula::recognizer::FormulaRecognizer,
    )>,
    /// 懒加载的页面公式检测器（键同上）。
    formula_detector: Option<(SessionCacheKey, crate::formula::detect::FormulaDetector)>,
    /// 懒加载的表格结构会话（键同上）。
    ///
    /// 会话创建含图优化，量级远大于一次推理，所以**每张图开一次**是不可接受的
    /// （`docs/11-table-page-pipeline-execution.md` §2.1 E14.5）。用会话缓存键而不是
    /// 路径，理由与公式识别器相同：同一路径被换成另一份内容时必须重建。
    ///
    /// 词表跟着会话走（[`crate::table::structure::TableStructureSession::vocabulary`]），
    /// 因此"用这个会话解码"不需要第二个信息来源，也不需要第二次探针。
    table_session: Option<(
        SessionCacheKey,
        crate::table::structure::TableStructureSession,
    )>,
}

/// 会话缓存的**键**：文件身份 + 这次的**有效期望摘要**（规范化后）+ 请求带来的会话构造参数。
///
/// # 为什么摘要必须是键的一部分
///
/// 只按文件身份做键会留下一条与“给出摘要即校验”（[`crate::api::FormulaPolicy::expected_model_sha256`]、
/// [`crate::api::TablePolicy::expected_model_sha256`]）直接冲突的路径：
///
/// 1. 第一次请求用**正确**的摘要加载模型，会话进入缓存；
/// 2. 第二次请求走同一路径、文件一个字节没变（身份因此完全相同），但声明换一个**错误**的摘要；
/// 3. 身份没变 -> 复用旧会话 -> 那条声明从未被比对，错误摘要静默通过。
///
/// 期望摘要原先只在**重建分支**里被读取，所以这个洞不是“再检查一次路径”能补的：
/// 声明值进入键以后，声明一变就必然重建，而重建路径上的凭证构造（[`crate::model_verify::VerifiedModel::verify`]）
/// 会按**这次的声明**重新校验并按需返回 [`crate::error::RapidOcrError::HashMismatch`]。
///
/// 同一条论证也适用于**登记回落**（复审 P1-b）：表格域里 `None` 不等于“这一次没有期望摘要”，
/// 它还可能来自模型同目录的清单。所以进键的是**解析后**的有效摘要，而不是策略字段的原始写法 ——
/// 否则“清单里的登记值被改掉”这一整个类别都落在键外，新登记值一次也不会被重校。
///
/// # 规范化与判定同形
///
/// 声明按 [`crate::model_verify::normalize_expected_sha256`] 折叠大小写（校验本来就是
/// 忽略大小写的），此外不做任何加工。键不能比判定更宽松：把带空白、带前缀的写法折叠成
/// 同一个键，等于让第二个错误声明因为命中缓存而根本不参与比对。
///
/// # 为什么会话构造参数也是键的一部分
///
/// 表格画布边长、形状固定策略与 intra 线程数都来自 [`crate::api::TablePolicy`]，而它们改变的是
/// **已创建的那个会话本身**：ORT 的线程选项与 free-dimension override 都在 `open` 时一次性施加，
/// 之后没有第二次机会。于是“文件与摘要都没变、只改配置”的请求若命中旧会话，第二次请求的配置
/// 根本没生效（复审 P1）。更糟的是同一张请求里预处理仍按**新**边长生成张量，喂进按**旧**边长
/// 固定形状的会话 —— 那不是一个“稍慢一点”的会话，而是一次形状不匹配或几何错位的推理。
///
/// # 键里的文件身份是启发式，不是安全边界
///
/// `identity` 只能说“**没有观察到变化**”：体积与 mtime 写回原值、首尾 64 KiB 保持不变的
/// **中段修改**仍在盲区里（登记于 [`crate::model_verify::FileIdentity`]）。声明摘要进入键
/// 保证的是“给出摘要即校验”这条契约跨请求也成立，**不保证**磁盘上的字节与那个摘要一致；
/// 把“被校验的”与“被使用的”钉成同一份字节的是 [`crate::model_verify::VerifiedModel`] 持有的
/// 内容快照（会话一律从它打开）。不得把本缓存描述成内容完整性证明。
#[derive(Debug, Clone, PartialEq, Eq)]
struct SessionCacheKey {
    /// 文件身份：路径 + 体积 + mtime + 首尾各 64 KiB 的局部摘要。
    identity: crate::model_verify::FileIdentity,
    /// 规范化后的**有效期望摘要**：调用方声明的非空值优先，否则是模型同目录清单里登记的
    /// 摘要（表格路径的回落，见
    /// [`crate::table::registration::resolve_registered_expected`]）。
    ///
    /// 键里放的是**解析后**的那一个，不是调用方原始声明的 `Option`（复审 P1-b）：只存
    /// 原始声明会让“登记摘要变了”落在键外 —— 声明仍是 `None`、身份仍相同，于是旧会话
    /// 继续回答，新登记值一次也没被重新校验。这与“给出声明即校验”是同一个漏洞的另一半。
    declared: Option<String>,
    /// 由请求决定的会话构造参数（见类型文档里“为什么会话构造参数也是键的一部分”）。
    config: SessionConfigKey,
}

/// [`SessionCacheKey`] 的构造参数分量。
#[derive(Debug, Clone, PartialEq, Eq)]
enum SessionConfigKey {
    /// 该域的会话参数全部来自引擎唯一的运行时档案（同一引擎实例内恒定），请求里没有
    /// 可变的设置：公式识别器与公式检测器。
    EngineOnly,
    /// 表格结构会话：存**整个** [`crate::table::structure::TableStructureConfig`]，而不是手挑
    /// 三个字段 —— 将来新增一只会影响会话的旋钮会自动进键，不需要记得改这里。
    Table(crate::table::structure::TableStructureConfig),
}

impl SessionCacheKey {
    /// 公式两站点：构造参数不来自请求。
    fn engine_only(identity: crate::model_verify::FileIdentity, declared: Option<&str>) -> Self {
        Self {
            identity,
            declared: declared.map(crate::model_verify::normalize_expected_sha256),
            config: SessionConfigKey::EngineOnly,
        }
    }

    /// 表格站点：`config` 必须是**即将交给 `TableStructureSession::open` 的那一份**，
    /// 这样“键里比较的”与“会话实际用到的”不可能分叉。
    fn table(
        identity: crate::model_verify::FileIdentity,
        declared: Option<&str>,
        config: &crate::table::structure::TableStructureConfig,
    ) -> Self {
        Self {
            identity,
            declared: declared.map(crate::model_verify::normalize_expected_sha256),
            config: SessionConfigKey::Table(config.clone()),
        }
    }
}

/// 要不要重建会话 —— 三个缓存点共用的**唯一**判据。
///
/// 抽成函数有两个理由：判定只有一处（新增第四个会话时不会忘记带上声明摘要）；
/// 以及这条规则本身可以在**不加载任何模型**的情况下被单测覆盖（真实模型用例在缺少
/// 资产的环境里会静默 skip，规则本身不能只靠那种用例作证）。
fn needs_rebuild(cached: Option<&SessionCacheKey>, current: &SessionCacheKey) -> bool {
    cached != Some(current)
}

impl RapidOcrEngine {
    pub fn new(config: EngineConfig) -> Result<Self> {
        let model_id = format!(
            "{}-{}-{}",
            config.rec.model.ocr_version.as_str(),
            config.rec.model.model_type.as_str(),
            config.rec.model.lang.as_str()
        );
        let base_min_side_len = config.global.min_side_len;
        let base_max_side_len = config.global.max_side_len;
        Ok(Self {
            inner: RapidOcr::new(config)?,
            model_id,
            base_min_side_len,
            base_max_side_len,
            formula_recognizer: None,
            formula_detector: None,
            table_session: None,
        })
    }

    pub fn provider_resolutions(&self) -> PipelineProviderResolutions {
        self.inner.provider_resolutions()
    }

    /// 引擎解析出的运行时档案（线程 plan + provider/arena/公式批大小）。
    ///
    /// 基准报告直接内嵌 `runtime_profile().threads`，因此报告里的线程数就是**生效值**，
    /// 不是配置里写了什么。
    pub fn runtime_profile(&self) -> &RuntimeProfile {
        self.inner.runtime_profile()
    }

    fn provider_info(&self) -> crate::api::ProviderInfo {
        let resolutions = self.inner.provider_resolutions();
        let convert = |resolution: ProviderResolution| crate::api::ProviderResolutionInfo {
            requested: match resolution.requested {
                crate::config::ProviderPreference::Cpu => crate::api::ProviderPreference::Cpu,
                crate::config::ProviderPreference::DirectMl { device_id } => {
                    crate::api::ProviderPreference::DirectMl { device_id }
                }
                crate::config::ProviderPreference::Cuda { device_id } => {
                    crate::api::ProviderPreference::Cuda { device_id }
                }
            },
            selected_ep: match resolution.selected_ep {
                crate::runtime::provider::ResolvedExecutionProvider::Cpu => {
                    crate::api::ResolvedProvider::Cpu
                }
                crate::runtime::provider::ResolvedExecutionProvider::DirectMl => {
                    crate::api::ResolvedProvider::DirectMl
                }
                crate::runtime::provider::ResolvedExecutionProvider::Cuda => {
                    crate::api::ResolvedProvider::Cuda
                }
            },
            fallback_to_cpu: resolution.fallback_used,
        };
        crate::api::ProviderInfo {
            detector: convert(resolutions.det),
            classifier: resolutions.cls.map(convert),
            recognizer: convert(resolutions.rec),
        }
    }
}

impl crate::api::OcrEngine for RapidOcrEngine {
    fn model_id(&self) -> &str {
        &self.model_id
    }
    fn provider(&self) -> crate::api::ProviderInfo {
        self.provider_info()
    }

    fn recognize(&mut self, request: crate::api::OcrRequest) -> Result<crate::api::OcrOutput> {
        // 校验与路由一次完成，产物交给调度器；调度器不再看策略字段。
        let planned = crate::api::ValidatedRequest::plan(request)?;
        self.recognize_page(planned)
    }
}

impl RapidOcrEngine {
    /// 激活内部管线的阶段级内存采样。
    pub fn enable_mem_probe(&mut self) {
        self.inner.enable_mem_probe();
    }

    /// 取出并清空累计的阶段内存记录。
    pub fn take_stage_mem_records(&mut self) -> Vec<crate::runtime::memory::StageMemRecord> {
        self.inner.take_stage_mem_records()
    }

    pub fn model_paths(
        &self,
    ) -> (
        Option<&std::path::Path>,
        Option<&std::path::Path>,
        Option<&std::path::Path>,
    ) {
        self.inner.model_paths()
    }

    pub fn take_rec_batch_records(&mut self) -> Vec<crate::ocr::rec::recognizer::BatchRecord> {
        self.inner.take_rec_batch_records()
    }

    pub fn enable_rec_batch_probe(&mut self) {
        self.inner.enable_rec_batch_probe();
    }

    pub fn last_detector_input_size(&self) -> Option<(u32, u32)> {
        self.inner.last_detector_input_size()
    }
}

impl RapidOcrEngine {
    /// **阶段顺序在整个 crate 里只出现这一次。**
    ///
    /// 表格在公式之外的理由：表格的定位器要吃"公式抹白之后"的页级文本行作为回填来源，
    /// 而公式自己的文本通道也会先跑。把文本通道（含公式）先产出，再往它上面**追加**表格区域，
    /// 两个阶段各自至多被调用一次（不再有递归，因此也不再有防递归改写）。
    fn recognize_page(
        &mut self,
        planned: crate::api::ValidatedRequest,
    ) -> Result<crate::api::OcrOutput> {
        let request = planned.request();
        let mut output = match planned.route().formula() {
            Some(policy) => self.recognize_text_with_formula(request, policy)?,
            None => self.recognize_text(request.clone())?,
        };
        if let Some(policy) = planned.route().table() {
            self.attach_table_regions(&mut output, request, policy)?;
        }
        output.validate()?;
        Ok(output)
    }
}

impl RapidOcrEngine {
    /// 公式路由关闭时的普通文本识别路径。
    ///
    /// 提取为独立函数是为了让“公式关闭时行为与不含公式功能时一致”成为结构性保证：
    /// 未启用公式时 `recognize` 直接走这里，不经过任何公式代码。
    fn recognize_text(&mut self, request: crate::api::OcrRequest) -> Result<crate::api::OcrOutput> {
        use crate::api::{
            ClassifierPolicy, OcrRegion, RegionSource, StageReport, StageState, StageTiming,
            TextOrientation,
        };
        if let Some(tile) = request.preprocess.tile {
            return self.recognize_tiled_request(request, tile);
        }
        let preprocess_start = Instant::now();
        let roi_size = request.roi.map(|roi| crate::api::ImageSize {
            width: roi.width,
            height: roi.height,
        });
        let m = self.materialize(&request)?;
        let (input, original_size, offset) =
            rec_image_input(m.rec, request.roi, request.preprocess.max_decode_pixels)?;
        let mut input = input;
        if matches!(
            request.preprocess.enhance,
            crate::api::EnhancementPolicy::ScreenAdaptive
        ) {
            let img = self.inner.loader.load(input)?;
            input = OcrInput::Image(crate::ocr::pipeline::image_ops::enhance_screen_adaptive(
                &img,
            )?);
        }
        let max_side = request
            .preprocess
            .max_side
            .unwrap_or(self.base_max_side_len as u32);
        self.inner
            .set_image_bounds(self.base_min_side_len as u32, max_side);
        let effective_scale = match (request.scale_hint, request.preprocess.min_text_scale) {
            (Some(a), Some(b)) => Some(a.max(b)),
            (Some(a), None) | (None, Some(a)) => Some(a),
            (None, None) => None,
        };
        if let Some(scale) = effective_scale {
            let hinted = ((max_side as f32) * scale).round().clamp(32.0, 8192.0) as u32;
            let min_hint = ((self.base_min_side_len as f32) * scale)
                .round()
                .clamp(1.0, hinted as f32) as u32;
            self.inner.set_image_bounds(min_hint, hinted);
        }
        let mut use_cls = match request.stages.classify.policy {
            ClassifierPolicy::Off => false,
            ClassifierPolicy::IfAvailable => self.inner.classifier.is_some(),
            ClassifierPolicy::Required => true,
        };
        if use_cls && self.inner.classifier.is_none() {
            if matches!(request.stages.classify.policy, ClassifierPolicy::Required) {
                return Err(crate::error::RapidOcrError::Config(
                    "classification requested but no classifier model is configured".into(),
                ));
            }
            use_cls = false;
        }
        if !request.stages.detect && !request.stages.recognize && !use_cls {
            return Err(crate::error::RapidOcrError::InvalidInput(
                "requested stages are unavailable".into(),
            ));
        }
        let preprocess_ms = preprocess_start.elapsed().as_secs_f32() * 1000.0;
        let exec = self.inner.run(
            input,
            ExecutionOptions {
                use_det: request.stages.detect,
                use_cls,
                use_rec: request.stages.recognize,
                apply_cls_rotation: request.stages.classify.apply_rotation,
                return_word_box: !matches!(
                    request.recognition.words,
                    crate::api::WordOutputMode::Off
                ),
                return_single_char_box: matches!(
                    request.recognition.words,
                    crate::api::WordOutputMode::Chars
                ),
                text_score: request.detection.text_score,
                box_thresh: request.detection.box_thresh,
                unclip_ratio: request.detection.unclip_ratio,
            },
        )?;
        let local_source_size = roi_size.unwrap_or(original_size);
        let boxes = exec.boxes.clone().unwrap_or_default();
        let det_scores = exec.det_scores.clone().unwrap_or_default();
        let lines = exec.lines.clone().unwrap_or_default();
        let words = exec.word_boxes.clone().unwrap_or_default();
        let cls = exec.cls_res.clone().unwrap_or_default();
        let cls_count = cls.len();
        let mut regions = Vec::new();
        if request.stages.detect {
            for (i, det_box) in boxes.iter().enumerate() {
                let mut polygon = *det_box;
                for p in &mut polygon {
                    p[0] += offset.0 as f32;
                    p[1] += offset.1 as f32;
                }
                let recognition = lines.get(i).map(|line| crate::api::RecognitionOutcome {
                    text: line.text.clone(),
                    score: line.score,
                    words: words.get(i).map(|items| {
                        items
                            .iter()
                            .map(|w| crate::api::OcrWord {
                                text: w.text.clone(),
                                score: w.score,
                                polygon: crate::api::Polygon {
                                    points: w.bbox.map(|mut p| {
                                        p[0] += offset.0 as f32;
                                        p[1] += offset.1 as f32;
                                        p
                                    }),
                                },
                                kind: request.recognition.words.kind_for(),
                            })
                            .collect()
                    }),
                });
                let classification =
                    cls.get(i)
                        .map(|(orientation, score)| crate::api::ClassificationOutcome {
                            orientation: if orientation == "180" {
                                TextOrientation::Deg180
                            } else {
                                TextOrientation::Deg0
                            },
                            score: *score,
                            applied_rotation: request.stages.classify.apply_rotation
                                && orientation == "180",
                        });
                regions.push(OcrRegion {
                    source: RegionSource::Detected { detector_index: i },
                    kind: crate::api::RegionKind::Text,
                    polygon: Some(crate::api::Polygon { points: polygon }),
                    detection: Some(crate::api::DetectionOutcome {
                        score: det_scores.get(i).copied().unwrap_or_default(),
                    }),
                    classification,
                    recognition,
                    formula: None,
                    table: None,
                });
            }
        } else if request.stages.recognize {
            for (i, line) in lines.iter().enumerate() {
                let classification = cls.get(i).map(|(o, s)| crate::api::ClassificationOutcome {
                    orientation: if o == "180" {
                        TextOrientation::Deg180
                    } else {
                        TextOrientation::Deg0
                    },
                    score: *s,
                    applied_rotation: request.stages.classify.apply_rotation && o == "180",
                });
                regions.push(OcrRegion {
                    source: RegionSource::Input,
                    kind: crate::api::RegionKind::Text,
                    polygon: None,
                    detection: None,
                    classification,
                    recognition: Some(crate::api::RecognitionOutcome {
                        text: line.text.clone(),
                        score: line.score,
                        words: words.get(i).map(|items| {
                            items
                                .iter()
                                .map(|w| crate::api::OcrWord {
                                    text: w.text.clone(),
                                    score: w.score,
                                    polygon: crate::api::Polygon {
                                        points: w.bbox.map(|mut p| {
                                            let sx = exec
                                                .processed_size
                                                .map(|(w, _)| {
                                                    local_source_size.width as f32 / w.max(1) as f32
                                                })
                                                .unwrap_or(1.0);
                                            let sy = exec
                                                .processed_size
                                                .map(|(_, h)| {
                                                    local_source_size.height as f32
                                                        / h.max(1) as f32
                                                })
                                                .unwrap_or(1.0);
                                            p[0] = p[0] * sx + offset.0 as f32;
                                            p[1] = p[1] * sy + offset.1 as f32;
                                            p
                                        }),
                                    },
                                    kind: request.recognition.words.kind_for(),
                                })
                                .collect()
                        }),
                    }),
                    formula: None,
                    table: None,
                });
            }
        } else if use_cls {
            for (o, s) in &cls {
                regions.push(OcrRegion {
                    source: RegionSource::Input,
                    kind: crate::api::RegionKind::Text,
                    polygon: None,
                    detection: None,
                    classification: Some(crate::api::ClassificationOutcome {
                        orientation: if o == "180" {
                            TextOrientation::Deg180
                        } else {
                            TextOrientation::Deg0
                        },
                        score: *s,
                        applied_rotation: request.stages.classify.apply_rotation && o == "180",
                    }),
                    recognition: None,
                    formula: None,
                    table: None,
                });
            }
        }
        let timing = crate::api::OcrTimings {
            decode_ms: exec.decode_ms.unwrap_or_default(),
            resize_ms: exec.resize_ms.unwrap_or_default(),
            crop_ms: exec.crop_ms.unwrap_or_default(),
            preprocess_ms,
            detector_preprocess_ms: exec
                .det_breakdown_ms
                .map(|v| v.preprocess_ms)
                .unwrap_or_default(),
            detector_infer_ms: exec
                .det_breakdown_ms
                .map(|v| v.infer_ms)
                .unwrap_or_default(),
            detector_postprocess_ms: exec
                .det_breakdown_ms
                .map(|v| v.postprocess_ms)
                .unwrap_or_default(),
            detect_ms: exec.elapsed_ms[0].unwrap_or_default(),
            classifier_preprocess_ms: exec.cls_breakdown_ms.map(|v| v[0]).unwrap_or_default(),
            classifier_infer_ms: exec.cls_breakdown_ms.map(|v| v[1]).unwrap_or_default(),
            classifier_postprocess_ms: exec.cls_breakdown_ms.map(|v| v[2]).unwrap_or_default(),
            classify_ms: exec.elapsed_ms[1].unwrap_or_default(),
            recognizer_preprocess_ms: exec.rec_breakdown_ms.map(|v| v[0]).unwrap_or_default(),
            recognizer_infer_ms: exec.rec_breakdown_ms.map(|v| v[1]).unwrap_or_default(),
            recognizer_postprocess_ms: exec.rec_breakdown_ms.map(|v| v[2]).unwrap_or_default(),
            recognize_ms: exec.elapsed_ms[2].unwrap_or_default(),
            formula_ms: 0.0,
            // 表格阶段尚未接入：三个命名分量先固定为 0，使它们出现在账本与 bench JSON 里，
            // 从而让后续接入有可比的前后基线（09 文档 E1）。
            table_detect_ms: 0.0,
            table_structure_infer_ms: 0.0,
            table_backfill_ms: 0.0,
            postprocess_ms: exec.postprocess_ms.unwrap_or_default(),
            total_ms: exec.e2e_ms.unwrap_or_default() + preprocess_ms,
        };
        let detector = if !request.stages.detect {
            StageReport {
                state: StageState::Disabled,
                timing: None,
            }
        } else if boxes.is_empty() {
            StageReport {
                state: StageState::SkippedNoInput,
                timing: Some(StageTiming {
                    preprocess_ms: timing.detector_preprocess_ms,
                    infer_ms: timing.detector_infer_ms,
                    postprocess_ms: timing.detector_postprocess_ms,
                }),
            }
        } else {
            StageReport {
                state: StageState::Completed { items: boxes.len() },
                timing: Some(StageTiming {
                    preprocess_ms: timing.detector_preprocess_ms,
                    infer_ms: timing.detector_infer_ms,
                    postprocess_ms: timing.detector_postprocess_ms,
                }),
            }
        };
        let classifier = if matches!(request.stages.classify.policy, ClassifierPolicy::Off) {
            StageReport {
                state: StageState::Disabled,
                timing: None,
            }
        } else if self.inner.classifier.is_none() || !use_cls {
            StageReport {
                state: StageState::SkippedUnavailable,
                timing: None,
            }
        } else {
            StageReport {
                state: StageState::Completed { items: cls_count },
                timing: exec.cls_breakdown_ms.map(|v| StageTiming {
                    preprocess_ms: v[0],
                    infer_ms: v[1],
                    postprocess_ms: v[2],
                }),
            }
        };
        let recognizer = if !request.stages.recognize {
            StageReport {
                state: StageState::Disabled,
                timing: None,
            }
        } else if lines.is_empty() {
            StageReport {
                state: StageState::SkippedNoInput,
                timing: exec.rec_breakdown_ms.map(|v| StageTiming {
                    preprocess_ms: v[0],
                    infer_ms: v[1],
                    postprocess_ms: v[2],
                }),
            }
        } else {
            StageReport {
                state: StageState::Completed { items: lines.len() },
                timing: exec.rec_breakdown_ms.map(|v| StageTiming {
                    preprocess_ms: v[0],
                    infer_ms: v[1],
                    postprocess_ms: v[2],
                }),
            }
        };
        Ok(crate::api::OcrOutput {
            schema_version: 1,
            image: crate::api::ImageInfo {
                original_size,
                processed_size: exec
                    .processed_size
                    .map(|(width, height)| crate::api::ImageSize { width, height })
                    .unwrap_or(original_size),
                coordinate_space: request.output.coordinate_space,
            },
            stages: crate::api::StageReports {
                input: crate::api::InputTimings {
                    decode_ms: exec.decode_ms,
                    resize_ms: exec.resize_ms,
                    crop_ms: exec.crop_ms,
                },
                detector,
                classifier,
                recognizer,
                formula: crate::api::StageReport::default(),
                table: crate::api::StageReport::default(),
            },
            regions,
            timings: timing,
            engine: crate::api::EngineInfo {
                model_id: self.model_id.clone(),
                provider: self.provider_info(),
            },
        })
    }
}

impl RapidOcrEngine {
    /// 页面级公式路由：先仲裁检测候选，再抹白保留区域并单独识别公式。
    ///
    /// 顺序说明：
    ///
    /// 1. 用共享输入层解码**原图**（公式裁剪需要未抹白的像素）；
    /// 2. 解析公式区域（检测模型 + 显式区域，见 `formula::route`）；
    /// 3. 检测候选存在时先在原图上获取普通 OCR 文本证据，拒绝明显代码误检；
    /// 4. 把保留的公式区域抹白后交给普通文本管线：公式像素根本不进入最终检测/识别；
    /// 5. 从原图裁剪公式区域，批量送入 `FormulaRecognizer`；
    /// 6. 公式区域作为独立 `RegionKind::Formula` 追加到输出，携带 polygon 与模型标识。
    fn recognize_text_with_formula(
        &mut self,
        request: &crate::api::OcrRequest,
        policy: &crate::api::FormulaPolicy,
    ) -> Result<crate::api::OcrOutput> {
        use crate::api::{FormulaOutcome, ImageSize, OcrRegion, Polygon, RegionKind, RegionSource};

        let formula_start = Instant::now();
        let original = self.decode_source_image(request)?;
        let size = ImageSize {
            width: original.width(),
            height: original.height(),
        };

        let mut candidates = self.detect_formula_candidates(&original, policy)?;
        for polygon in &policy.input_regions {
            polygon.validate(size)?;
            candidates.push(crate::formula::route::FormulaCandidate {
                polygon: *polygon,
                score: 1.0,
                detected: false,
            });
        }
        let regions = crate::formula::route::resolve_formula_regions(
            &candidates,
            size,
            policy.min_area_ratio,
            policy.iou_threshold,
            policy.max_regions,
        );
        let formula_ms = formula_start.elapsed().as_secs_f32() * 1000.0;

        // Detector confidence is not semantic evidence: code lines can score as formulas.
        // Use one original-image OCR pass to reject only candidates whose covered text is
        // clearly code. If all candidates are rejected, reuse that pass and avoid a second
        // OCR run; retained candidates keep the whitening path below.
        let mut evidence_output = None;
        let regions = if regions.iter().any(|candidate| candidate.detected) {
            let original_output = self.recognize_text(request.clone())?;
            let evidence: Vec<_> = original_output
                .regions
                .iter()
                .filter_map(|region| {
                    let polygon = region.polygon?;
                    let text = region.recognition.as_ref()?.text.as_str();
                    Some(crate::formula::route::TextEvidence { polygon, text })
                })
                .collect();
            let arbitrated = crate::formula::route::arbitrate_formula_regions(regions, &evidence);
            evidence_output = Some(original_output);
            arbitrated
        } else {
            regions
        };

        let mut text_request = request.clone();
        if !regions.is_empty() {
            // 抹白公式像素：普通文本管线看不到公式，也就不会对它们跑 CTC。
            text_request.input = crate::api::ImageInput::Image(
                crate::formula::route::whiten_regions(&original, &regions)?,
            );
            text_request.roi = None;
        }
        let mut output = match (regions.is_empty(), evidence_output) {
            (true, Some(original_output)) => original_output,
            (_, _) => self.recognize_text(text_request)?,
        };

        if regions.is_empty() {
            output.stages.formula = crate::api::StageReport {
                state: crate::api::StageState::SkippedNoInput,
                timing: Some(crate::api::StageTiming {
                    preprocess_ms: formula_ms,
                    infer_ms: 0.0,
                    postprocess_ms: 0.0,
                }),
            };
            return Ok(output);
        }

        let crops: Vec<image::DynamicImage> = regions
            .iter()
            .map(|candidate| crop_polygon_bounds(&original, candidate.polygon))
            .collect();
        let recognize_start = Instant::now();
        let recognitions = self.formula_recognizer(policy)?.recognize_batch(&crops)?;
        let recognize_ms = recognize_start.elapsed().as_secs_f32() * 1000.0;
        output.timings.formula_ms = recognize_ms + formula_ms;
        output.timings.total_ms += recognize_ms + formula_ms;

        for (index, (candidate, recognition)) in regions.iter().zip(recognitions).enumerate() {
            let detected = candidate.detected;
            let source = if detected {
                RegionSource::Detected {
                    detector_index: index,
                }
            } else {
                RegionSource::Input
            };
            let polygon = Some(Polygon {
                points: canonical_polygon_points(candidate.polygon),
            });
            output.regions.push(OcrRegion {
                source,
                kind: RegionKind::Formula,
                polygon,
                detection: detected.then_some(crate::api::DetectionOutcome {
                    score: candidate.score,
                }),
                classification: None,
                recognition: None,
                formula: Some(FormulaOutcome {
                    latex: recognition.latex,
                    eos_index: recognition.eos_index,
                    truncated: recognition.truncated,
                    model_id: recognition.model_id,
                    token_ids: policy.include_token_ids.then_some(recognition.token_ids),
                }),
                table: None,
            });
        }

        // `items` 是**公式阶段处理的公式数量**，不是页面区域总数：
        // 一页 20 个文本区域 + 2 个公式区域时必须报告 2。
        output.stages.formula = crate::api::StageReport {
            state: crate::api::StageState::Completed {
                items: output.formula_count(),
            },
            timing: Some(crate::api::StageTiming {
                preprocess_ms: formula_ms,
                infer_ms: recognize_ms,
                postprocess_ms: 0.0,
            }),
        };
        output.validate()?;
        Ok(output)
    }

    /// 把请求输入物化成**整页**像素（P1-5 的唯一入口）。
    ///
    /// 文本与分块两条路径原本各自有一段五变体的 `match request.input`，逐像素语义完全
    /// 一致，只是产出形态不同（`OcrInput` vs `RgbaImage`）。这里统一收成一次：
    /// 校验/限长顺序取 `view.validate() -> ensure_decode_pixels -> to_bgr(None)`（与
    /// `decode_source_image`、分块原路径一致），`Encoded`/`File`/`Url` 经 `load_with_limit`
    /// 解码一次。roi **不在这里裁剪**——`offset` 只携带偏移，裁剪仍由调用方
    /// （`rec_image_input` / `crop_imm`）完成，页面级阶段仍需未裁的整页像素。
    fn materialize(&self, request: &crate::api::OcrRequest) -> Result<MaterializedInput> {
        use crate::api::ImageInput;
        let pixels = request.preprocess.max_decode_pixels;
        let encoded = request.preprocess.max_encoded_bytes;
        let rec = match &request.input {
            ImageInput::Pixels(view) => {
                view.validate()?;
                ensure_decode_pixels(view.width as usize, view.height as usize, pixels)?;
                let (bgr, size, _) = view.to_bgr(None)?;
                crate::config::RecImage::from_bgr_u8(
                    size.width as usize,
                    size.height as usize,
                    bgr,
                )?
            }
            ImageInput::Encoded(bytes) => self.inner.loader.load_with_limit(
                OcrInput::Bytes(bytes.to_vec()),
                pixels,
                encoded,
            )?,
            ImageInput::File(path) => {
                self.inner
                    .loader
                    .load_with_limit(OcrInput::Path(path.clone()), pixels, encoded)?
            }
            ImageInput::Url(url) => {
                self.inner
                    .loader
                    .load_with_limit(OcrInput::Url(url.clone()), pixels, encoded)?
            }
            ImageInput::Image(image) => {
                ensure_decode_pixels(image.width(), image.height(), pixels)?;
                image.clone()
            }
        };
        let original_size = crate::api::ImageSize {
            width: rec.width() as u32,
            height: rec.height() as u32,
        };
        Ok(MaterializedInput { rec, original_size })
    }

    /// 解码请求输入为原图 `DynamicImage`，复用共享输入层的限制与错误语义。
    ///
    /// 公式与表格两条路由都用它：两者都需要**未抹白的原始像素**（公式裁剪、表格结构推理），
    /// 而共享输入层的像素上限与错误语义只能有一份。
    fn decode_source_image(&self, request: &crate::api::OcrRequest) -> Result<image::DynamicImage> {
        use crate::api::ImageInput;
        let pixels = request.preprocess.max_decode_pixels;
        let encoded = request.preprocess.max_encoded_bytes;
        let image = match &request.input {
            ImageInput::Encoded(bytes) => self.inner.loader.load_dynamic_with_limit(
                OcrInput::Bytes(bytes.to_vec()),
                pixels,
                encoded,
            )?,
            ImageInput::File(path) => self.inner.loader.load_dynamic_with_limit(
                OcrInput::Path(path.clone()),
                pixels,
                encoded,
            )?,
            ImageInput::Url(url) => self.inner.loader.load_dynamic_with_limit(
                OcrInput::Url(url.clone()),
                pixels,
                encoded,
            )?,
            ImageInput::Pixels(view) => {
                view.validate()?;
                // 内存像素输入同样受 `max_decode_pixels` 约束：公式路由不能成为
                // 绕过普通 OCR 限制的旁路。
                ensure_decode_pixels(
                    view.width as usize,
                    view.height as usize,
                    request.preprocess.max_decode_pixels,
                )?;
                let (bgr, size, _) = view.to_bgr(None)?;
                let rec = crate::config::RecImage::from_bgr_u8(
                    size.width as usize,
                    size.height as usize,
                    bgr,
                )?;
                rec_image_to_dynamic(&rec)?
            }
            ImageInput::Image(rec) => {
                ensure_decode_pixels(
                    rec.width(),
                    rec.height(),
                    request.preprocess.max_decode_pixels,
                )?;
                rec_image_to_dynamic(rec)?
            }
        };
        Ok(image)
    }

    /// 运行公式检测模型；未配置检测模型时返回空候选（只处理显式区域）。
    fn detect_formula_candidates(
        &mut self,
        image: &image::DynamicImage,
        policy: &crate::api::FormulaPolicy,
    ) -> Result<Vec<crate::formula::route::FormulaCandidate>> {
        let Some(path) = policy.detector_path.clone() else {
            return Ok(Vec::new());
        };
        let options = crate::formula::detect::FormulaDetectOptions {
            confidence_threshold: policy.confidence_threshold,
            iou_threshold: policy.iou_threshold,
            max_detections: policy.max_regions,
        };
        // 公式检测器与三个阶段共享同一份运行时档案，而不是再读一遍 rec 阶段配置。
        let runtime = self.inner.runtime_profile().session_runtime();
        // 缓存键（文件身份 + 这次声明的期望摘要）决定要不要重建：文件被替换后不会继续用
        // 旧会话；文件一个字节没变但**声明换了**，也不会让新声明落空。
        let key = SessionCacheKey::engine_only(
            crate::model_verify::FileIdentity::of(&path)?,
            policy.expected_detector_sha256.as_deref(),
        );
        if needs_rebuild(
            self.formula_detector.as_ref().map(|(cached, _)| cached),
            &key,
        ) {
            let detector_model = crate::model_verify::VerifiedModel::verify(
                &path,
                policy.expected_detector_sha256.as_deref(),
            )?;
            self.formula_detector = Some((
                key,
                crate::formula::detect::FormulaDetector::from_model(&detector_model, &runtime)?,
            ));
        }
        let detector = self
            .formula_detector
            .as_mut()
            .map(|(_, detector)| detector)
            .ok_or_else(|| {
                crate::error::RapidOcrError::Config("formula detector is not initialized".into())
            })?;
        Ok(detector
            .detect(image, &options)?
            .into_iter()
            .map(|detected| crate::formula::route::FormulaCandidate {
                polygon: crate::api::Polygon {
                    points: detected.polygon,
                },
                score: detected.score,
                detected: true,
            })
            .collect())
    }

    /// 懒加载公式识别器；**会话缓存键**（文件身份 + 声明摘要）变化时重建。
    ///
    /// 会话设置与批大小都来自引擎唯一的运行时档案：公式识别不是一条可以自己决定
    /// 线程数与批大小的旁路。期望摘要由策略传入，由库在加载时校验——**磁盘上的那份**
    /// 必须是声明的那一份，而不是"第一次加载时是哪一份就永远是哪一份"。
    fn formula_recognizer(
        &mut self,
        policy: &crate::api::FormulaPolicy,
    ) -> Result<&mut crate::formula::recognizer::FormulaRecognizer> {
        let path = policy
            .model_path
            .clone()
            .ok_or_else(|| crate::model_verify::missing_model_path_error("formula"))?;
        let runtime = self.inner.runtime_profile().session_runtime();
        let key = SessionCacheKey::engine_only(
            crate::model_verify::FileIdentity::of(&path)?,
            policy.expected_model_sha256.as_deref(),
        );
        if needs_rebuild(
            self.formula_recognizer.as_ref().map(|(cached, _)| cached),
            &key,
        ) {
            let recognizer_model = crate::model_verify::VerifiedModel::verify(
                &path,
                policy.expected_model_sha256.as_deref(),
            )?;
            let mut recognizer = crate::formula::recognizer::FormulaRecognizer::from_model(
                &recognizer_model,
                &runtime,
            )?;
            recognizer.set_max_batch_size(runtime.formula_batch)?;
            self.formula_recognizer = Some((key, recognizer));
        }
        self.formula_recognizer
            .as_mut()
            .map(|(_, recognizer)| recognizer)
            .ok_or_else(|| {
                crate::error::RapidOcrError::Config("formula recognizer is not initialized".into())
            })
    }

    /// 懒加载表格结构会话；**会话缓存键**（文件身份 + 有效期望摘要 + 会话构造参数）变化时重建。
    ///
    /// 会话的设置全部来自策略与引擎的运行时档案：表格结构不是一条可以自己决定线程数、
    /// 画布边长与形状固定策略的旁路。期望摘要同样由策略传入、由库在加载时校验。
    ///
    /// 构造参数**必须进键**（复审 P1）：它们决定的是已创建的那个会话本身（线程选项与
    /// free-dimension override 都在 `open` 时一次性施加），而同一张请求里的预处理与解码几何
    /// 却按**这次的** `input_side_len` 走 —— 复用旧会话等于让两边各自用两个不同的边长。
    ///
    /// 摘要也必须先解析再进键（复审 P1-b）：策略里写 `None` 不等于“这一次没有期望摘要”，
    /// 表格域还有一条登记回落。因此本函数在取身份之前就把**有效值**算出来，同一个值
    /// 既进键又交给凭证构造 —— 清单里的登记值换了人就必然重建并重校。
    fn table_session(
        &mut self,
        policy: &crate::api::TablePolicy,
    ) -> Result<&mut crate::table::structure::TableStructureSession> {
        let path = policy
            .model_path
            .clone()
            .ok_or_else(|| crate::model_verify::missing_model_path_error("table"))?;
        // “期望从哪来”只有这一个解析点（显式非空优先，否则取同目录清单的登记）。
        // 拿到以后向下只有一份：键、凭证都用它，所以“被重校的”与“被复用的”不会分叉。
        let expected = crate::table::registration::resolve_registered_expected(
            &path,
            policy.expected_model_sha256.as_deref(),
        )?;
        // 先组装构造参数，再让它**同时**成为键的一部分与 `open` 的入参：
        // “键里比较的”与“会话实际用到的”因此是同一个值，不可能分叉。
        let config = crate::table::structure::TableStructureConfig {
            input_side_len: policy.input_side_len,
            fix_input_shape: policy.fix_input_shape,
            intra_threads: policy.intra_threads,
        };
        let key = SessionCacheKey::table(
            crate::model_verify::FileIdentity::of(&path)?,
            expected.as_deref(),
            &config,
        );
        if needs_rebuild(self.table_session.as_ref().map(|(cached, _)| cached), &key) {
            let runtime = self.inner.runtime_profile().session_runtime();
            // 凭证在打开会话之前构造：期望摘要的“从哪来”（显式优先、否则取登记）留在
            // registration 分层，会话本身只收凭证（审核点 2：与文本/公式同一契约）。
            // 这里传的是**已解析**的值，所以 registration 层内部的回落分支不会重读清单。
            let credential =
                crate::table::registration::verified_registered_model(&path, expected.as_deref())?;
            let session = crate::table::structure::TableStructureSession::open(
                &credential,
                &runtime,
                &config,
            )?;
            self.table_session = Some((key, session));
        }
        self.table_session
            .as_mut()
            .map(|(_, session)| session)
            .ok_or_else(|| {
                crate::error::RapidOcrError::Config("table session is not initialized".into())
            })
    }

    /// 表格路由：往**已完成的页级文本输出**上追加表格区域（文本分布定位 → 结构推理 → 几何
    /// 匹配回填 → 发射）。
    ///
    /// 与 [`Self::recognize_text_with_formula`] 的关键差别：
    ///
    /// - **表格像素不抹白**。单元格文本正是要保留的页级识别结果，抹白会把它删掉。
    /// - **不重新识别单元格**。文本通过几何匹配从页级结果里取（门禁 G2/G7 的结构不变量
    ///   `BackfillOutcome::recognizer_calls == 0`），因此多一张表不会多一次 rec 调用。
    /// - 定位零新增整页 CNN（门禁 G4）：输入是**已有文本框的分布**。
    ///
    /// 页级文本通道在此之前由调度器 [`Self::recognize_page`] 产出：若公式启用，`output` 里
    /// 已经追加了公式区域、且其文本通道的公式像素已被抹白。于是定位与回填用的是同一组
    /// 抹白后的文本行（落在公式区域内的单元格因此表现为 `Unmatched`），而结构模型裁剪自
    /// **原图**、看到的是真实像素而不是被挖空的图。
    fn attach_table_regions(
        &mut self,
        output: &mut crate::api::OcrOutput,
        request: &crate::api::OcrRequest,
        policy: &crate::api::TablePolicy,
    ) -> Result<()> {
        use crate::api::{
            OcrRegion, Polygon, RegionKind, RegionSource, StageReport, StageState, StageTiming,
            TableDegradation, TableOutcome, TableStructureConfidence,
        };
        use crate::table::detection::TableDetector;

        // 定位。输入是已有文本框的分布（零新增整页 CNN），像素来自**原图**。
        let original = self.decode_source_image(request)?;
        let source = table_source_lines(&output.regions);
        let image = rec_image_from_dynamic(&original)?;

        let detect_start = Instant::now();
        let mut detector = crate::table::detection::TextDistributionDetector::new(
            crate::table::detection::TableDetectorConfig::default(),
        )?;
        let mut regions = detector.detect(&image, &source.lines)?;
        // 低分与超量区域在这里被丢掉，而不是留给下游"少输出一点"。
        regions.retain(|region| region.score >= policy.min_region_score);
        regions.sort_by(|a, b| {
            b.score
                .partial_cmp(&a.score)
                .unwrap_or(std::cmp::Ordering::Equal)
        });
        regions.truncate(policy.max_regions);
        let detect_ms = detect_start.elapsed().as_secs_f32() * 1000.0;

        if regions.is_empty() {
            output.timings.table_detect_ms = detect_ms;
            output.timings.total_ms += detect_ms;
            output.stages.table = StageReport {
                state: StageState::SkippedNoInput,
                timing: Some(StageTiming {
                    preprocess_ms: detect_ms,
                    infer_ms: 0.0,
                    postprocess_ms: 0.0,
                }),
            };
            return Ok(());
        }

        // 会话在整个请求里只取一次：`table_session` 每次都要按会话缓存键（文件身份 +
        // 有效期望摘要 + 会话构造参数）确认缓存是否仍然有效，那是每张表一次的 `stat`；更重要的是
        // “一张表一次结构推理”（门禁 G3）应当由**结构**保证，而不是靠调用点自觉。
        let session = self.table_session(policy)?;
        let vocabulary = session.vocabulary().clone();
        let mut infer_ms = 0.0_f32;
        let mut backfill_ms = 0.0_f32;
        let mut table_regions: Vec<OcrRegion> = Vec::new();

        for (index, region) in regions.iter().enumerate() {
            let (tensor, _canvas) = crate::table::preprocess::table_structure_input(
                &region.crop,
                policy.input_side_len,
            )?;

            let infer_start = Instant::now();
            let (boxes, logits) = session.run_with(tensor.view(), |outputs| {
                Ok((outputs.boxes.to_owned(), outputs.tokens.to_owned()))
            })?;
            infer_ms += infer_start.elapsed().as_secs_f32() * 1000.0;

            let backfill_start = Instant::now();
            let decoded = table_decode_and_backfill(&TableDecodeInput {
                vocabulary: &vocabulary,
                boxes: &boxes,
                logits: &logits,
                side_len: policy.input_side_len,
                region,
                source: &source,
            })?;
            backfill_ms += backfill_start.elapsed().as_secs_f32() * 1000.0;

            let text_policy = match policy.text_policy {
                crate::api::TableTextPolicy::Escape => crate::table::emitter::TextPolicy::Escape,
                crate::api::TableTextPolicy::Raw => crate::table::emitter::TextPolicy::Raw,
            };
            let rendered =
                crate::table::emitter::emit_html(&decoded.result, &decoded.texts, text_policy)?;
            let table_html = crate::table::emitter::emit_table_html(
                &decoded.result,
                &decoded.texts,
                text_policy,
            )?
            .rendered;
            let markdown =
                crate::table::emitter::emit_markdown(&decoded.result, &decoded.texts)?.rendered;

            // `docs/11` §3 E15.2 / §4 G8：面向用户的 HTML 必须**唯一**经过净化，
            // 而且净化发生在引擎出口 —— 这样 `/api/ocr` 拿到的就已经是净化后的 HTML，
            // 网页侧（E16.3）不需要第二份实现（两份实现必然漂移，白名单会失效）。
            let (html, table_html) =
                finalize_table_html(rendered.rendered, table_html, policy.text_policy);

            table_regions.push(OcrRegion {
                source: RegionSource::Detected {
                    detector_index: index,
                },
                kind: RegionKind::Table,
                // `Quad` 与 `Polygon::points` 是同一个类型，因此区域多边形**没有换算**：
                // 定位器吃的是整页文本框，吐出的就是整页坐标。
                polygon: Some(Polygon {
                    points: region.polygon,
                }),
                detection: Some(crate::api::DetectionOutcome {
                    score: region.score,
                }),
                classification: None,
                recognition: None,
                formula: None,
                table: Some(TableOutcome {
                    html,
                    table_html,
                    markdown,
                    rows: decoded.result.grid.rows,
                    cols: decoded.result.grid.cols,
                    confidence: match rendered.confidence {
                        crate::table::emitter::StructureConfidence::High => {
                            TableStructureConfidence::High
                        }
                        crate::table::emitter::StructureConfidence::Low => {
                            TableStructureConfidence::Low
                        }
                    },
                    holes: decoded.result.coverage.holes(),
                    degradations: rendered
                        .degradations
                        .iter()
                        .map(|degradation| match degradation {
                            crate::table::emitter::StructureDegradation::GridHoles { holes } => {
                                TableDegradation::GridHoles { holes: *holes }
                            }
                            crate::table::emitter::StructureDegradation::Truncated => {
                                TableDegradation::Truncated
                            }
                        })
                        .collect(),
                    cells: decoded.cells,
                    source_line_indices: decoded.source_line_indices,
                    model_id: policy
                        .model_path
                        .as_ref()
                        .map(|path| path.display().to_string())
                        .unwrap_or_default(),
                }),
            });
        }

        output.timings.table_detect_ms = detect_ms;
        output.timings.table_structure_infer_ms = infer_ms;
        output.timings.table_backfill_ms = backfill_ms;
        output.timings.total_ms += detect_ms + infer_ms + backfill_ms;
        let items = table_regions.len();
        output.regions.extend(table_regions);
        output.stages.table = StageReport {
            state: StageState::Completed { items },
            timing: Some(StageTiming {
                preprocess_ms: detect_ms,
                infer_ms,
                postprocess_ms: backfill_ms,
            }),
        };
        Ok(())
    }
}

/// 以多边形包围盒裁剪原图；旋转四边形同样使用轴对齐包围盒。
fn crop_polygon_bounds(
    image: &image::DynamicImage,
    polygon: crate::api::Polygon,
) -> image::DynamicImage {
    use image::GenericImageView;
    let (width, height) = image.dimensions();
    let (mut x0, mut y0) = (f32::INFINITY, f32::INFINITY);
    let (mut x1, mut y1) = (f32::NEG_INFINITY, f32::NEG_INFINITY);
    for [x, y] in polygon.points {
        x0 = x0.min(x);
        y0 = y0.min(y);
        x1 = x1.max(x);
        y1 = y1.max(y);
    }
    let x0 = x0.floor().clamp(0.0, width as f32) as u32;
    let y0 = y0.floor().clamp(0.0, height as f32) as u32;
    let x1 = x1.ceil().clamp(0.0, width as f32) as u32;
    let y1 = y1.ceil().clamp(0.0, height as f32) as u32;
    let w = (x1.saturating_sub(x0))
        .max(1)
        .min(width.saturating_sub(x0).max(1));
    let h = (y1.saturating_sub(y0))
        .max(1)
        .min(height.saturating_sub(y0).max(1));
    image.crop_imm(x0, y0, w, h)
}

/// 输出的多边形统一为左上起顺时针四角，便于下游稳定消费。
fn canonical_polygon_points(polygon: crate::api::Polygon) -> [[f32; 2]; 4] {
    let (mut x0, mut y0) = (f32::INFINITY, f32::INFINITY);
    let (mut x1, mut y1) = (f32::NEG_INFINITY, f32::NEG_INFINITY);
    for [x, y] in polygon.points {
        x0 = x0.min(x);
        y0 = y0.min(y);
        x1 = x1.max(x);
        y1 = y1.max(y);
    }
    [[x0, y0], [x1, y0], [x1, y1], [x0, y1]]
}

fn rec_image_to_dynamic(image: &crate::config::RecImage) -> Result<image::DynamicImage> {
    use crate::config::ColorOrder;
    let width = image.width() as u32;
    let height = image.height() as u32;
    let rgb: Vec<u8> = match image.color_order() {
        ColorOrder::Rgb => image.as_bytes().to_vec(),
        ColorOrder::Bgr => image
            .as_bytes()
            .as_chunks::<3>()
            .0
            .iter()
            .flat_map(|pixel| [pixel[2], pixel[1], pixel[0]])
            .collect(),
    };
    let buffer = image::RgbImage::from_raw(width, height, rgb).ok_or_else(|| {
        crate::error::RapidOcrError::InvalidImage("invalid RGB buffer for formula routing".into())
    })?;
    Ok(image::DynamicImage::ImageRgb8(buffer))
}

/// 把已解码的 `DynamicImage` 转回共享输入层的 `RecImage`（BGR 序）。
///
/// 表格定位器与结构预处理的输入类型都是 [`crate::config::RecImage`]，而输入层交出来的是
/// `DynamicImage`。这一层转换刻意走**公开构造器**而不是重新解码字节：像素只解码一次，
/// 上限与 EXIF 语义只有一份。
fn rec_image_from_dynamic(image: &image::DynamicImage) -> Result<crate::config::RecImage> {
    let rgb = image.to_rgb8();
    let (width, height) = (rgb.width() as usize, rgb.height() as usize);
    let bgr: Vec<u8> = rgb
        .as_raw()
        .as_chunks::<3>()
        .0
        .iter()
        .flat_map(|pixel| [pixel[2], pixel[1], pixel[0]])
        .collect();
    crate::config::RecImage::from_bgr_u8(width, height, bgr)
}

/// 页级文本区域 → 表格定位与回填所需的两种行表示。
///
/// 同一个文本区域在两条路径上出现：定位器（[`crate::table::detection::PageLine`]）需要
/// 四边形，回填（[`crate::table::cell_content_backfill::TextLine`]）需要轴对齐盒。
/// 现在两者都由**同一个** [`crate::table::page_line::PageTextLine`] 显式导出：bbox 的裁剪规则
/// 收敛到规范类型的一处，装配点不再各拼一份，"定位看到的行"与"回填用的行"因此结构上不可能分叉。
struct TableSourceLines {
    lines: Vec<crate::table::detection::PageLine>,
    text_lines: Vec<crate::table::cell_content_backfill::TextLine>,
    /// `lines[i]` / `text_lines[i]` 对应的区域在 `OcrOutput::regions` 中的下标。
    region_indices: Vec<usize>,
}

fn table_source_lines(regions: &[crate::api::OcrRegion]) -> TableSourceLines {
    let mut canonical: Vec<crate::table::page_line::PageTextLine> = Vec::new();
    for (index, region) in regions.iter().enumerate() {
        // 只有**文本**区域参与：公式区域没有文本语义，表格区域更不可能（它是本函数的消费者）。
        if region.kind != crate::api::RegionKind::Text {
            continue;
        }
        let (Some(polygon), Some(recognition)) = (region.polygon, region.recognition.as_ref())
        else {
            continue;
        };
        canonical.push(crate::table::page_line::PageTextLine::new(
            index,
            polygon.points,
            recognition.text.clone(),
            recognition.score,
        ));
    }
    TableSourceLines {
        lines: canonical.iter().map(|line| line.to_page_line()).collect(),
        text_lines: canonical.iter().map(|line| line.to_text_line()).collect(),
        region_indices: canonical.iter().map(|line| line.index).collect(),
    }
}

/// 一次表格区域的解码 + 回填所需的全部输入。
///
/// 打包成结构体而不是八个位置参数：这些量**必须来自同一张裁剪图**（框、logits、裁剪尺寸、
/// 区域原点），位置参数会让"传错了那张图的尺寸"变成一次静默的坐标错位。
struct TableDecodeInput<'a> {
    vocabulary: &'a crate::table::decode::StructureVocabulary,
    boxes: &'a ndarray::Array3<f32>,
    logits: &'a ndarray::Array3<f32>,
    side_len: usize,
    region: &'a crate::table::detection::TableRegion,
    source: &'a TableSourceLines,
}

struct TableDecoded {
    result: crate::table::decode::StructureResult,
    texts: Vec<String>,
    cells: Vec<crate::api::TableCellOutcome>,
    source_line_indices: Vec<usize>,
}

/// 把发射器产出的表格 HTML 变成 API 出口形态（`docs/11` §3 E15.2 / §4 G8）。
///
/// 返回 `(html, table_html)`：`html` 是评测口径需要的整页外壳，
/// `table_html` 是 `OcrTableItem` 里给页面渲染用的那一份。
///
/// **净化的唯一落点就在这里**。引擎出口净化一次，网页侧就没有第二份白名单，
/// 也就不存在"两份实现漂移后其中一份放行了 `<script>`"的窗口。
///
/// 评测口径（`TableTextPolicy::Raw`）**刻意不净化**：消费者是 lxml + 官方
/// `metric.py`，改一个字节都会破坏与 PubTabNet 真值的可比性。这就是
/// [`crate::api::TableTextPolicy`] 存在的理由 —— 它不是"更严格/更宽松"的开关，
/// 而是"面向用户 / 面向评测"的开关。
///
/// 生产上两条路径目前**产出相同字节**：`TextPolicy::Escape`（用户默认）已经让
/// 单元格文本里的 `<` 变成 `&lt;`，而发射器只会写出白名单内的标签与
/// `rowspan`/`colspan` 两种属性，因此净化是空操作。净化在这里是
/// **纵深防御**：它约束的是未来任何新的发射路径，而不是今天的输出。
fn finalize_table_html(
    html: String,
    table_html: String,
    policy: crate::api::TableTextPolicy,
) -> (String, String) {
    match policy {
        crate::api::TableTextPolicy::Escape => {
            // 发射器的两条输出必须满足这个恒等式（`emit_html` 就是
            // `<html><body>` + `emit_table_html` + `</body></html>`）。在这里钉住它，
            // 是为了让"净化后的 `html` 重新拼外壳"这件事有依据 —— 否则哪天发射器换了
            // 外壳，`html` 与 `table_html` 会静默分叉。
            debug_assert_eq!(html, format!("<html><body>{table_html}</body></html>"));
            let cleaned = crate::output::html::sanitize_table_html(&table_html);
            let shell = format!("<html><body>{cleaned}</body></html>");
            (shell, cleaned)
        }
        crate::api::TableTextPolicy::Raw => (html, table_html),
    }
}

fn table_decode_and_backfill(input: &TableDecodeInput<'_>) -> Result<TableDecoded> {
    use crate::api::{Polygon, TableCellMatch, TableCellOutcome};
    use crate::table::cell_content_backfill::CellMatchKind;

    if input.logits.shape()[0] != 1 || input.boxes.shape()[0] != 1 {
        return Err(crate::error::RapidOcrError::Decode(format!(
            "the table structure model returned a batch of {} box(es) / {} structure(s), but the page \
             pipeline runs one region at a time",
            input.boxes.shape()[0],
            input.logits.shape()[0]
        )));
    }
    let steps_count = input.logits.shape()[1];
    let classes = input.logits.shape()[2];
    // 先查类别数再解码：类别数不符时 `decode_token_steps` 报的是"某个 class index 越界"，
    // 那是症状而不是原因。这里报的是"模型与词表不配套"。
    if classes != input.vocabulary.class_count() {
        return Err(crate::error::RapidOcrError::Decode(format!(
            "the table structure model's logits have {classes} classes but the vocabulary has {}",
            input.vocabulary.class_count()
        )));
    }
    let argmaxed = crate::table::decode::argmax_structure(
        input.logits.as_slice().expect("contiguous"),
        steps_count,
    )?;
    let (steps, termination) =
        crate::table::decode::decode_token_steps(input.vocabulary, &argmaxed)?;

    // 单元格几何从**归一化**坐标回到**裁剪图**像素：`denormalize` 与 `rescale` 两步复合后
    // 等价于统一乘 `max(h, w)`（E4 的代数不变量），量纲是裁剪图像素而不是 488 画布。
    let height = input.region.crop.height() as f32;
    let width = input.region.crop.width() as f32;
    let side = input.side_len as f32;
    let shape = crate::table::decode::TableShape::new(height, width, side)?;
    let ratios = crate::table::decode::rescale_ratios(width, height, side);
    let box_flat = input.boxes.as_slice().expect("contiguous");
    let cell_polygons: Vec<[f32; 8]> = steps
        .iter()
        .filter(|step| step.is_cell_start)
        .map(|step| {
            let offset = step.step_index * 8;
            let mut raw = [0.0_f32; 8];
            raw.copy_from_slice(&box_flat[offset..offset + 8]);
            crate::table::decode::rescale_polygon(&shape.denormalize_polygon(&raw), ratios)
        })
        .collect();
    let result = crate::table::decode::tokens_to_structure(
        input.vocabulary,
        &steps,
        termination,
        &cell_polygons,
    )?;

    // 回填：单元格几何在**裁剪图**坐标系里，页级文本行在**整页**坐标系里，因此偏移必须
    // 加上区域原点。参考实现喂的是同一张整图（没有 crop），所以那里 offset 是零；
    // 本仓库走"整页里的小表"路线，这一项正是 E11 §12.4.7 登记的待办。
    let outcome = crate::table::cell_content_backfill::backfill(
        &result.cells,
        crate::table::cell_content_backfill::BackfillSource::Reuse {
            lines: &input.source.text_lines,
            offset: input.region.origin,
        },
    )?;
    // 复用路径恒为 0，是门禁 G2/G7 的结构不变量。这里断言而不是"记得不要调"：
    // 将来有人把 `Reuse` 换成 `Crop`，这一行会在页面路径上立刻失败。
    debug_assert_eq!(outcome.recognizer_calls, 0);

    let mut cells = Vec::with_capacity(result.cells.len());
    let mut consumed: Vec<usize> = Vec::new();
    for (index, slot) in result.cells.iter().enumerate() {
        if let Some(matched) = outcome.matched_lines.get(index) {
            for line_index in matched {
                if let Some(region_index) = input.source.region_indices.get(*line_index) {
                    consumed.push(*region_index);
                }
            }
        }
        cells.push(TableCellOutcome {
            row: slot.logic.row_start,
            col: slot.logic.col_start,
            rowspan: slot.rowspan,
            colspan: slot.colspan,
            text: outcome.texts.get(index).cloned().unwrap_or_default(),
            score: outcome.text_scores.get(index).copied().flatten(),
            match_kind: match outcome.kinds.get(index) {
                Some(CellMatchKind::BlankBox) => TableCellMatch::BlankBox,
                Some(CellMatchKind::Unmatched) => TableCellMatch::Unmatched,
                Some(CellMatchKind::Single) => TableCellMatch::Single,
                Some(CellMatchKind::Multiple) => TableCellMatch::Multiple,
                // 回填契约是"与 cells 等长同序"，缺项说明契约被破坏了。
                None => {
                    return Err(crate::error::RapidOcrError::Decode(format!(
                        "the backfill returned {} cell kind(s) for {} slot(s)",
                        outcome.kinds.len(),
                        result.cells.len()
                    )));
                }
            },
            polygon: slot.bbox.map(|bbox| {
                let mut quad = crate::table::cell_content_backfill::bbox_to_quad(bbox);
                for point in &mut quad {
                    point[0] += input.region.origin[0];
                    point[1] += input.region.origin[1];
                }
                Polygon { points: quad }
            }),
        });
    }
    consumed.sort_unstable();
    consumed.dedup();

    Ok(TableDecoded {
        result,
        texts: outcome.texts,
        cells,
        source_line_indices: consumed,
    })
}

impl RapidOcrEngine {
    fn recognize_tiled_request(
        &mut self,
        request: crate::api::OcrRequest,
        tile: crate::api::TilePolicy,
    ) -> Result<crate::api::OcrOutput> {
        if matches!(
            request.stages.classify.policy,
            crate::api::ClassifierPolicy::Required
        ) && self.inner.classifier.is_none()
        {
            return Err(crate::error::RapidOcrError::Config(
                "classification requested but no classifier model is configured".into(),
            ));
        }
        if tile.max_width == 0
            || tile.max_height == 0
            || tile.overlap >= tile.max_width
            || tile.overlap >= tile.max_height
        {
            return Err(crate::error::RapidOcrError::InvalidInput(
                "invalid tile policy".into(),
            ));
        }
        let m = self.materialize(&request)?;
        let rgba = rec_image_to_rgba(&m.rec)?;
        let original_size = m.original_size;
        if let Some(roi) = request.roi {
            roi.validate_against(original_size.width, original_size.height)?;
        }
        let roi = request.roi.unwrap_or(crate::api::RectU32 {
            x: 0,
            y: 0,
            width: original_size.width,
            height: original_size.height,
        });
        let view = image::imageops::crop_imm(&rgba, roi.x, roi.y, roi.width, roi.height).to_image();
        let step_x = tile.max_width - tile.overlap;
        let step_y = tile.max_height - tile.overlap;
        let mut regions = Vec::new();
        let mut timings = crate::api::OcrTimings::default();
        let mut y = 0;
        while y < view.height() {
            let h = tile.max_height.min(view.height() - y);
            let mut x = 0;
            while x < view.width() {
                let w = tile.max_width.min(view.width() - x);
                let crop = image::imageops::crop_imm(&view, x, y, w, h).to_image();
                let mut preprocess = request.preprocess.clone();
                preprocess.tile = None;
                let sub = crate::api::OcrRequest {
                    input: crate::api::ImageInput::Pixels(crate::api::OwnedPixelBuffer {
                        width: w,
                        height: h,
                        stride: w as usize * 4,
                        format: crate::api::PixelFormat::Rgba8,
                        bottom_up: false,
                        data: Arc::from(crop.into_raw()),
                    }),
                    roi: None,
                    scale_hint: request.scale_hint,
                    stages: request.stages,
                    preprocess,
                    detection: request.detection,
                    recognition: request.recognition,
                    output: request.output,
                    // tiling 与公式路由互斥（`request.validate()` 已拒绝同时启用）。
                    formula: crate::api::FormulaPolicy::default(),
                    // 表格路由只在顶层跑：分块子请求必须显式关闭，否则会逐块重复定位。
                    table: crate::api::TablePolicy::default(),
                };
                let mut out = self.recognize(sub)?;
                let region_offset = regions.len();
                timings.decode_ms += out.timings.decode_ms;
                timings.resize_ms += out.timings.resize_ms;
                timings.crop_ms += out.timings.crop_ms;
                timings.preprocess_ms += out.timings.preprocess_ms;
                timings.detector_preprocess_ms += out.timings.detector_preprocess_ms;
                timings.detector_infer_ms += out.timings.detector_infer_ms;
                timings.detector_postprocess_ms += out.timings.detector_postprocess_ms;
                timings.detect_ms += out.timings.detect_ms;
                timings.classifier_preprocess_ms += out.timings.classifier_preprocess_ms;
                timings.classifier_infer_ms += out.timings.classifier_infer_ms;
                timings.classifier_postprocess_ms += out.timings.classifier_postprocess_ms;
                timings.classify_ms += out.timings.classify_ms;
                timings.recognizer_preprocess_ms += out.timings.recognizer_preprocess_ms;
                timings.recognizer_infer_ms += out.timings.recognizer_infer_ms;
                timings.recognizer_postprocess_ms += out.timings.recognizer_postprocess_ms;
                timings.recognize_ms += out.timings.recognize_ms;
                timings.postprocess_ms += out.timings.postprocess_ms;
                for region in &mut out.regions {
                    if let Some(poly) = &mut region.polygon {
                        for p in &mut poly.points {
                            p[0] += (roi.x + x) as f32;
                            p[1] += (roi.y + y) as f32;
                        }
                    }
                    if let Some(rec) = &mut region.recognition
                        && let Some(words) = &mut rec.words
                    {
                        for word in words {
                            for p in &mut word.polygon.points {
                                p[0] += (roi.x + x) as f32;
                                p[1] += (roi.y + y) as f32;
                            }
                        }
                    }
                    if let crate::api::RegionSource::Detected { detector_index } = region.source {
                        region.source = crate::api::RegionSource::Detected {
                            detector_index: detector_index + region_offset,
                        };
                    }
                }
                regions.extend(out.regions);
                if x + w >= view.width() {
                    break;
                }
                x += step_x;
            }
            if y + h >= view.height() {
                break;
            }
            y += step_y;
        }
        regions = deduplicate_regions(regions);
        timings.total_ms = timings.preprocess_ms
            + timings.detect_ms
            + timings.classify_ms
            + timings.recognize_ms
            + timings.postprocess_ms;
        Ok(crate::api::OcrOutput {
            schema_version: 1,
            image: crate::api::ImageInfo {
                original_size,
                processed_size: original_size,
                coordinate_space: request.output.coordinate_space,
            },
            stages: crate::api::StageReports {
                input: crate::api::InputTimings {
                    decode_ms: Some(timings.decode_ms),
                    resize_ms: Some(timings.resize_ms),
                    crop_ms: Some(timings.crop_ms),
                },
                detector: crate::api::StageReport {
                    state: if !request.stages.detect {
                        crate::api::StageState::Disabled
                    } else {
                        let items = regions.iter().filter(|r| r.detection.is_some()).count();
                        if items == 0 {
                            crate::api::StageState::SkippedNoInput
                        } else {
                            crate::api::StageState::Completed { items }
                        }
                    },
                    timing: request.stages.detect.then_some(crate::api::StageTiming {
                        preprocess_ms: timings.detector_preprocess_ms,
                        infer_ms: timings.detector_infer_ms,
                        postprocess_ms: timings.detector_postprocess_ms,
                    }),
                },
                classifier: crate::api::StageReport {
                    state: if matches!(
                        request.stages.classify.policy,
                        crate::api::ClassifierPolicy::Off
                    ) {
                        crate::api::StageState::Disabled
                    } else if self.inner.classifier.is_none() {
                        crate::api::StageState::SkippedUnavailable
                    } else {
                        let items = regions
                            .iter()
                            .filter(|r| r.classification.is_some())
                            .count();
                        if items == 0 {
                            crate::api::StageState::SkippedNoInput
                        } else {
                            crate::api::StageState::Completed { items }
                        }
                    },
                    timing: if matches!(
                        request.stages.classify.policy,
                        crate::api::ClassifierPolicy::Off
                    ) || self.inner.classifier.is_none()
                    {
                        None
                    } else {
                        Some(crate::api::StageTiming {
                            preprocess_ms: timings.classifier_preprocess_ms,
                            infer_ms: timings.classifier_infer_ms,
                            postprocess_ms: timings.classifier_postprocess_ms,
                        })
                    },
                },
                recognizer: crate::api::StageReport {
                    state: if !request.stages.recognize {
                        crate::api::StageState::Disabled
                    } else {
                        let items = regions.iter().filter(|r| r.recognition.is_some()).count();
                        if items == 0 {
                            crate::api::StageState::SkippedNoInput
                        } else {
                            crate::api::StageState::Completed { items }
                        }
                    },
                    timing: request.stages.recognize.then_some(crate::api::StageTiming {
                        preprocess_ms: timings.recognizer_preprocess_ms,
                        infer_ms: timings.recognizer_infer_ms,
                        postprocess_ms: timings.recognizer_postprocess_ms,
                    }),
                },
                // 分块路径与公式路由互斥（`request.validate()` 已拒绝同时启用）。
                formula: crate::api::StageReport::default(),
                table: crate::api::StageReport::default(),
            },
            regions,
            timings,
            engine: crate::api::EngineInfo {
                model_id: self.model_id.clone(),
                provider: self.provider_info(),
            },
        })
    }
}

fn deduplicate_regions(mut regions: Vec<crate::api::OcrRegion>) -> Vec<crate::api::OcrRegion> {
    let mut unique = Vec::with_capacity(regions.len());
    for region in regions.drain(..) {
        let duplicate = region.recognition.as_ref().and_then(|recognition| {
            region.polygon.and_then(|polygon| {
                unique.iter().position(|existing: &crate::api::OcrRegion| {
                    existing
                        .recognition
                        .as_ref()
                        .is_some_and(|v| v.text == recognition.text)
                        && existing.polygon.is_some_and(|other| {
                            crate::evaluation::ocr::polygon_iou(other.points, polygon.points) >= 0.5
                        })
                })
            })
        });
        if let Some(index) = duplicate {
            let replace = region
                .recognition
                .as_ref()
                .zip(unique[index].recognition.as_ref())
                .is_some_and(|(a, b)| a.score > b.score);
            if replace {
                unique[index] = region;
            }
        } else {
            unique.push(region);
        }
    }
    unique.sort_by(|a, b| {
        let key = |region: &crate::api::OcrRegion| {
            region
                .polygon
                .map(|polygon| {
                    (
                        polygon
                            .points
                            .iter()
                            .map(|p| p[1])
                            .fold(f32::INFINITY, f32::min),
                        polygon
                            .points
                            .iter()
                            .map(|p| p[0])
                            .fold(f32::INFINITY, f32::min),
                    )
                })
                .unwrap_or((f32::INFINITY, f32::INFINITY))
        };
        key(a)
            .partial_cmp(&key(b))
            .unwrap_or(std::cmp::Ordering::Equal)
    });
    unique
}

#[cfg(test)]
mod formula_integration_tests {
    //! 页面级公式路由的集成测试。
    //!
    //! 这些测试需要真实模型（普通 OCR、公式识别、公式检测）与真实页面图片，
    //! 因此全部通过环境变量定位：缺失时显式 skip，绝不 panic、绝不回落到
    //! 开发机绝对路径。

    use std::path::PathBuf;
    use std::sync::Arc;

    use crate::api::{
        ClassifierPlan, ClassifierPolicy, DetectionPolicy, FormulaPolicy, ImageInput, OcrEngine,
        OcrRequest, OutputPolicy, PreprocessPolicy, RecognitionPolicy, RegionKind, RegionSource,
        StagePlan, StageState, TextOrder, WordOutputMode,
    };
    use crate::ocr::pipeline::config::EngineConfig;
    use crate::ocr::pipeline::rapid_structure::RapidOcrEngine;

    fn engine_config(model_root: &std::path::Path) -> EngineConfig {
        let yaml = format!(
            r#"
global:
  use_det: true
  use_cls: false
  use_rec: true
  max_side_len: 2000
  min_side_len: 30
det:
  lang: multi
  ocr_version: PP-OCRv6
  model_type: small
  model_path: {root}/small/PP-OCRv6_det_small.onnx
  allow_download: false
cls:
  lang: ch
rec:
  model:
    lang: ch
    ocr_version: PP-OCRv6
    model_type: small
    model_path: {root}/small/PP-OCRv6_rec_small.onnx
    rec_keys_path: {root}/small/ppocrv6_dict.txt
    allow_download: false
"#,
            root = model_root.display().to_string().replace('\\', "/")
        );
        EngineConfig::from_yaml_str(&yaml).expect("engine config template must parse")
    }

    struct Assets {
        engine: RapidOcrEngine,
        formula_model: PathBuf,
        detector: PathBuf,
        page_with_formula: PathBuf,
        /// 实测在该页面上检测器返回 0 个候选（无公式 / 漏检场景）。
        page_without_detections: PathBuf,
        /// 实测该页面没有公式，但默认阈值下检测器会误检（用于验证阈值可控性）。
        page_with_false_positives: PathBuf,
    }

    fn assets() -> Option<Assets> {
        let model_root = crate::test_support::ocr_model_root()?;
        let formula_model = crate::test_support::formula_model_path()?;
        let detector = crate::test_support::formula_detector_path()?;
        let page_with_formula = crate::test_support::page_fixture("08数字公式与符号.png")?;
        let page_without_detections = crate::test_support::page_fixture("09竖排文本.png")?;
        let page_with_false_positives = crate::test_support::page_fixture("05代码与等宽字体.png")?;
        let engine = RapidOcrEngine::new(engine_config(&model_root)).expect("engine should load");
        Some(Assets {
            engine,
            formula_model,
            detector,
            page_with_formula,
            page_without_detections,
            page_with_false_positives,
        })
    }

    fn request(bytes: Vec<u8>, formula: FormulaPolicy) -> OcrRequest {
        OcrRequest {
            input: ImageInput::Encoded(Arc::from(bytes)),
            roi: None,
            scale_hint: None,
            stages: StagePlan {
                detect: true,
                classify: ClassifierPlan {
                    policy: ClassifierPolicy::Off,
                    apply_rotation: false,
                },
                recognize: true,
            },
            preprocess: PreprocessPolicy::default(),
            detection: DetectionPolicy::default(),
            recognition: RecognitionPolicy {
                words: WordOutputMode::Off,
            },
            output: OutputPolicy::default(),
            formula,
            table: crate::api::TablePolicy::default(),
        }
    }

    fn formula_policy(assets: &Assets) -> FormulaPolicy {
        FormulaPolicy {
            enabled: true,
            model_path: Some(assets.formula_model.clone()),
            detector_path: Some(assets.detector.clone()),
            ..FormulaPolicy::default()
        }
    }

    /// 评审 P1-2：公式检测器的会话缓存必须按**文件身份**（路径 + 体积 + mtime）失效。
    ///
    /// 旧实现只比较**路径**：模型文件在第一次加载之后被替换/损坏，第二次请求会发现
    /// "路径没变"，于是继续用内存里那份旧会话——损坏被静默吞掉。这里用真实检测模型的
    /// 一份**副本**（绝不改动仓库资产）证明：追加一个字节之后，同一个路径上的下一次请求
    /// 必须重新校验并因摘要不匹配而失败。
    ///
    /// 用"没有公式区域的页面"（`09竖排文本.png`，实测检测器返回 0 个候选）就不需要加载
    /// 566 MB 的识别模型：本用例考察的是**检测器**的缓存失效。
    #[test]
    fn a_replaced_formula_detector_is_reverified_instead_of_reused() {
        let Some(model_root) = crate::test_support::ocr_model_root() else {
            return;
        };
        let Some(detector) = crate::test_support::formula_detector_path() else {
            return;
        };
        let Some(formula_model) = crate::test_support::formula_model_path() else {
            return;
        };
        let Some(page) = crate::test_support::page_fixture("09竖排文本.png") else {
            return;
        };
        let dir = crate::test_support::TempDir::new("pipeline-detector-identity");
        let copy = dir.path().join("pix2text-mfd-1.5.onnx");
        std::fs::copy(&detector, &copy).expect("copy the real detector");
        let declared = crate::model_store::sha256_file(&copy).expect("hash the copy");

        let mut engine =
            RapidOcrEngine::new(engine_config(&model_root)).expect("engine should load");
        let policy = FormulaPolicy {
            enabled: true,
            model_path: Some(formula_model),
            expected_model_sha256: None,
            detector_path: Some(copy.clone()),
            expected_detector_sha256: Some(declared.clone()),
            ..FormulaPolicy::default()
        };
        let bytes = std::fs::read(&page).expect("page readable");
        engine
            .recognize(request(bytes.clone(), policy.clone()))
            .expect("the first load must succeed");

        // 追加一个字节：体积变化 → 身份变化 → 必须重新校验并拒绝（而不是复用旧会话）。
        let mut file = std::fs::OpenOptions::new()
            .append(true)
            .open(&copy)
            .expect("open the copy for append");
        std::io::Write::write_all(&mut file, b"x").expect("append one byte");
        drop(file);

        let error = engine
            .recognize(request(bytes, policy))
            .expect_err("a replaced detector must not be reused");
        match error {
            crate::error::RapidOcrError::HashMismatch {
                path: reported,
                expected,
                actual,
            } => {
                assert_eq!(reported, copy, "the error must name the replaced file");
                assert_eq!(expected, declared);
                assert_ne!(actual, declared);
            }
            other => panic!("expected a hash mismatch from the re-verification, got {other:?}"),
        }
    }

    /// 复审 P1-1（检测器站点）：**热会话 + 同一路径 + 换一个错误的声明摘要**。
    ///
    /// 与上面那条的分工：上面那条改的是**文件**（身份变 → 重建），这一条文件一个字节
    /// 没改、只改**声明**。旧实现里声明只在重建分支被读，于是第二次请求的错误声明会
    /// 因为"身份没变"而被旧会话直接回答 —— 声明从未被比对，API 上"给出摘要即校验"
    /// 落空。声明进入缓存键后必须重建，重建分支按这次的声明校验并拒绝。
    #[test]
    fn a_changed_detector_declaration_is_not_answered_by_a_hot_session() {
        let Some(model_root) = crate::test_support::ocr_model_root() else {
            return;
        };
        let Some(detector) = crate::test_support::formula_detector_path() else {
            return;
        };
        let Some(formula_model) = crate::test_support::formula_model_path() else {
            return;
        };
        let Some(page) = crate::test_support::page_fixture("09竖排文本.png") else {
            return;
        };
        let dir = crate::test_support::TempDir::new("pipeline-detector-declaration");
        let copy = dir.path().join("pix2text-mfd-1.5.onnx");
        std::fs::copy(&detector, &copy).expect("copy the real detector");
        let declared = crate::model_store::sha256_file(&copy).expect("hash the copy");

        let mut engine =
            RapidOcrEngine::new(engine_config(&model_root)).expect("engine should load");
        let policy = |expected: Option<String>| FormulaPolicy {
            enabled: true,
            model_path: Some(formula_model.clone()),
            expected_model_sha256: None,
            detector_path: Some(copy.clone()),
            expected_detector_sha256: expected,
            ..FormulaPolicy::default()
        };
        let bytes = std::fs::read(&page).expect("page readable");
        engine
            .recognize(request(bytes.clone(), policy(Some(declared.clone()))))
            .expect("the first request with the correct digest must pass");

        let bogus = format!("{:064x}", 0xdead_beef_u32);
        let error = engine
            .recognize(request(bytes.clone(), policy(Some(bogus.clone()))))
            .expect_err("a changed declaration must not be answered by the hot session");
        match error {
            crate::error::RapidOcrError::HashMismatch {
                path: reported,
                expected,
                actual,
            } => {
                assert_eq!(reported, copy, "the error must name the declared file");
                assert_eq!(expected, bogus);
                assert_eq!(
                    actual, declared,
                    "the file never changed - only the declaration did"
                );
            }
            other => panic!("expected a hash mismatch from the declaration, got {other:?}"),
        }

        // 同一条声明换个写法（大小写）仍是**同一个声明**：键的规范化与判定同形，
        // 不会因为折叠不一致而白白重建会话。
        engine
            .recognize(request(bytes, policy(Some(declared.to_uppercase()))))
            .expect("an equivalently spelled declaration must still be accepted");
    }

    /// 复审 P1-1（识别器站点）：与检测器同形的一条，把声明摘要钉进缓存键。
    ///
    /// 用**显式声明的公式区域**触发识别器（`detector_path = None`），因此这条用例
    /// 只加载识别模型一份会话；文件从第一次请求之后不再改动。566 MB 会话只构造一次。
    #[test]
    fn a_changed_recognizer_declaration_is_not_answered_by_a_hot_session() {
        let Some(model_root) = crate::test_support::ocr_model_root() else {
            return;
        };
        let Some(formula_model) = crate::test_support::formula_model_path() else {
            return;
        };
        let Some(page) = crate::test_support::page_fixture("08数字公式与符号.png") else {
            return;
        };
        let dir = crate::test_support::TempDir::new("pipeline-recognizer-declaration");
        let copy = dir.path().join("pp_formulanet_plus_m.onnx");
        std::fs::copy(&formula_model, &copy).expect("copy the real recognizer");
        let declared = crate::model_store::sha256_file(&copy).expect("hash the copy");

        let mut engine =
            RapidOcrEngine::new(engine_config(&model_root)).expect("engine should load");
        let policy = |expected: Option<String>| FormulaPolicy {
            enabled: true,
            model_path: Some(copy.clone()),
            expected_model_sha256: expected,
            detector_path: None,
            input_regions: vec![crate::api::Polygon {
                points: [
                    [120.0, 420.0],
                    [900.0, 420.0],
                    [900.0, 560.0],
                    [120.0, 560.0],
                ],
            }],
            ..FormulaPolicy::default()
        };
        let bytes = std::fs::read(&page).expect("page readable");
        let output = engine
            .recognize(request(bytes.clone(), policy(Some(declared.clone()))))
            .expect("the first request with the correct digest must pass");
        assert_eq!(
            output.formula_count(),
            1,
            "the explicit region must reach the recognizer (otherwise this test proves nothing)"
        );

        let bogus = format!("{:064x}", 7u32);
        let error = engine
            .recognize(request(bytes, policy(Some(bogus.clone()))))
            .expect_err("a changed declaration must not be answered by the hot session");
        match error {
            crate::error::RapidOcrError::HashMismatch {
                path: reported,
                expected,
                actual,
            } => {
                assert_eq!(reported, copy, "the error must name the declared file");
                assert_eq!(expected, bogus);
                assert_eq!(
                    actual, declared,
                    "the file never changed - only the declaration did"
                );
            }
            other => panic!("expected a hash mismatch from the declaration, got {other:?}"),
        }
    }

    /// 公式路由与 `roi`/`tile` 互斥，必须在进入管线前结构化拒绝。
    #[test]
    fn formula_routing_rejects_roi_and_tile() {
        let base = FormulaPolicy {
            enabled: true,
            model_path: Some(PathBuf::from("model.onnx")),
            ..FormulaPolicy::default()
        };
        let mut with_roi = request(Vec::new(), base.clone());
        with_roi.roi = Some(crate::api::RectU32 {
            x: 0,
            y: 0,
            width: 10,
            height: 10,
        });
        let error = with_roi.validate().expect_err("roi must be rejected");
        assert!(error.to_string().contains("roi"), "error: {error}");

        let mut with_tile = request(Vec::new(), base);
        with_tile.preprocess.tile = Some(crate::api::TilePolicy {
            max_width: 100,
            max_height: 100,
            overlap: 10,
        });
        let error = with_tile.validate().expect_err("tile must be rejected");
        assert!(error.to_string().contains("tiled"), "error: {error}");
    }

    #[test]
    fn formula_policy_requires_a_model_path() {
        let policy = FormulaPolicy {
            enabled: true,
            ..FormulaPolicy::default()
        };
        assert!(policy.validate().is_err());
    }

    /// 公式功能关闭时：输出中不能出现公式区域，公式阶段必须是 `Disabled`。
    #[test]
    fn disabled_formula_policy_produces_no_formula_regions() {
        let Some(mut assets) = assets() else {
            return;
        };
        let bytes = std::fs::read(&assets.page_with_formula).expect("page readable");
        let output = assets
            .engine
            .recognize(request(bytes, FormulaPolicy::default()))
            .expect("baseline recognize");
        assert!(output.validate().is_ok());
        assert_eq!(output.formula_count(), 0);
        assert!(output.formula_latex(TextOrder::Reading).is_empty());
        assert_eq!(output.stages.formula.state, StageState::Disabled);
        assert!(
            output
                .regions
                .iter()
                .all(|region| region.kind == RegionKind::Text),
            "disabled formula policy must not produce formula regions"
        );
    }

    /// 漏检场景：页面上没有公式（检测器返回 0 候选）时，公式路由不得改变文本输出。
    #[test]
    fn formula_routing_without_detections_matches_the_baseline() {
        let Some(mut assets) = assets() else {
            return;
        };
        let bytes = std::fs::read(&assets.page_without_detections).expect("page readable");
        let baseline = assets
            .engine
            .recognize(request(bytes.clone(), FormulaPolicy::default()))
            .expect("baseline recognize");
        let routed = assets
            .engine
            .recognize(request(bytes, formula_policy(&assets)))
            .expect("routed recognize");
        assert_eq!(routed.formula_count(), 0, "no formula on this page");
        assert_eq!(routed.stages.formula.state, StageState::SkippedNoInput);
        assert_eq!(
            routed.plain_text(TextOrder::Reading),
            baseline.plain_text(TextOrder::Reading),
            "formula routing must not change text output when nothing is detected"
        );
        assert_eq!(
            routed.regions.len(),
            baseline.regions.len(),
            "region count must not change when nothing is detected"
        );
    }

    /// 误检回归：普通 OCR 的代码文本证据必须阻止公式检测器抹白正文。
    ///
    /// 这个页面在默认检测阈值下确实会产生候选；正确行为是仲裁后不再产生公式
    /// 区域，并保持普通 OCR 的文本结果与公式关闭时一致。
    #[test]
    fn code_text_evidence_rejects_formula_false_positives_in_the_pipeline() {
        let Some(mut assets) = assets() else {
            return;
        };
        let bytes = std::fs::read(&assets.page_with_false_positives).expect("page readable");
        let baseline = assets
            .engine
            .recognize(request(bytes.clone(), FormulaPolicy::default()))
            .expect("baseline recognize");

        let routed = assets
            .engine
            .recognize(request(bytes.clone(), formula_policy(&assets)))
            .expect("routed recognize");
        assert!(
            routed.formula_count() == 0,
            "this page is known to produce false positives at the default threshold; \
             text evidence arbitration must reject them"
        );
        assert_eq!(routed.stages.formula.state, StageState::SkippedNoInput);
        assert_eq!(
            routed.plain_text(TextOrder::Reading),
            baseline.plain_text(TextOrder::Reading),
            "rejecting a false positive must preserve the original OCR text"
        );
    }

    /// 主路径：检测到公式后，公式区域是独立 typed region，且文本区域不含公式文本。
    #[test]
    fn detected_formula_regions_are_typed_and_carry_latex() {
        let Some(mut assets) = assets() else {
            return;
        };
        let bytes = std::fs::read(&assets.page_with_formula).expect("page readable");
        let output = assets
            .engine
            .recognize(request(bytes, formula_policy(&assets)))
            .expect("routed recognize");
        output.validate().expect("routed output must validate");
        assert!(
            output.formula_count() > 0,
            "the formula page must produce at least one formula region"
        );
        assert_eq!(
            output.stages.formula.state,
            StageState::Completed {
                items: output.formula_count()
            },
            "the formula stage must count formula regions, not all regions"
        );
        assert!(
            output.formula_count() < output.regions.len(),
            "this page has text regions too, so the two counts must differ"
        );
        for region in output
            .regions
            .iter()
            .filter(|region| region.kind == RegionKind::Formula)
        {
            let formula = region.formula.as_ref().expect("formula outcome");
            assert!(
                region.recognition.is_none(),
                "formula regions must not carry CTC text: {:?}",
                region.recognition
            );
            assert!(!formula.model_id.is_empty());
            assert!(
                region.polygon.is_some(),
                "formula regions must keep their crop polygon"
            );
            assert!(
                formula.token_ids.is_none(),
                "token ids must stay opt-in outside debug mode"
            );
        }
    }

    /// 显式声明的公式区域无需检测模型即可工作，并带上 `RegionSource::Input`。
    #[test]
    fn explicit_input_regions_are_recognized_without_a_detector() {
        let Some(mut assets) = assets() else {
            return;
        };
        let bytes = std::fs::read(&assets.page_with_formula).expect("page readable");
        let policy = FormulaPolicy {
            enabled: true,
            model_path: Some(assets.formula_model.clone()),
            detector_path: None,
            input_regions: vec![crate::api::Polygon {
                points: [
                    [120.0, 420.0],
                    [900.0, 420.0],
                    [900.0, 560.0],
                    [120.0, 560.0],
                ],
            }],
            include_token_ids: true,
            ..FormulaPolicy::default()
        };
        let output = assets
            .engine
            .recognize(request(bytes, policy))
            .expect("explicit region recognize");
        output.validate().expect("output must validate");
        assert_eq!(output.formula_count(), 1);
        let region = output
            .regions
            .iter()
            .find(|region| region.kind == RegionKind::Formula)
            .expect("formula region");
        assert_eq!(region.source, RegionSource::Input);
        assert!(
            region.detection.is_none(),
            "explicit regions have no detector score"
        );
        let formula = region.formula.as_ref().expect("formula outcome");
        assert!(
            formula.token_ids.is_some(),
            "include_token_ids must keep the raw token sequence"
        );
    }

    /// 引擎必须把 `runtime.formula_batch` 交给懒加载的公式识别器，并且**区域数超过
    /// 批大小**时仍然成功（分块由识别器内部完成）。
    ///
    /// 这是根因回归：`FormulaPolicy::max_regions` 默认 64 而模型批大小默认 16，
    /// 过去的实现把整页裁剪一次性送进 `recognize_batch`，超过上限就让整个请求失败。
    /// 这里把 `formula_batch` 压到 4 并用 5 个区域，于是“区域数 > 批大小”这条路径
    /// 在引擎层面被真实走通；同时直接观察识别器实际生效的批大小，锁住接线。
    #[test]
    fn engine_applies_formula_batch_and_chunks_region_batches() {
        let Some(model_root) = crate::test_support::ocr_model_root() else {
            return;
        };
        let Some(formula_model) = crate::test_support::formula_model_path() else {
            return;
        };
        let Some(page) = crate::test_support::page_fixture("08数字公式与符号.png") else {
            return;
        };

        let mut config = engine_config(&model_root);
        config.runtime.formula_batch = 4;
        let mut engine = RapidOcrEngine::new(config).expect("engine should load");
        assert_eq!(
            engine.runtime_profile().formula_batch,
            4,
            "the resolved profile must carry the configured formula batch"
        );

        let page_image = image::open(&page).expect("page readable");
        let (width, height) = (page_image.width() as f32, page_image.height() as f32);
        // 5 个互不重叠的显式区域：显式区域不做面积过滤，但会被裁剪到图像范围内。
        let input_regions: Vec<crate::api::Polygon> = (0..5)
            .map(|index| {
                let x0 = width * (0.05 + index as f32 * 0.19);
                let x1 = x0 + width * 0.15;
                crate::api::Polygon {
                    points: [
                        [x0, height * 0.05],
                        [x1, height * 0.05],
                        [x1, height * 0.15],
                        [x0, height * 0.15],
                    ],
                }
            })
            .collect();
        let policy = FormulaPolicy {
            enabled: true,
            model_path: Some(formula_model),
            detector_path: None,
            input_regions: input_regions.clone(),
            ..FormulaPolicy::default()
        };
        let bytes = std::fs::read(&page).expect("page readable");
        let output = engine
            .recognize(request(bytes, policy))
            .expect("more regions than formula_batch must be chunked, not rejected");
        output.validate().expect("output must validate");
        assert_eq!(output.formula_count(), input_regions.len());
        assert_eq!(
            output.stages.formula.state,
            StageState::Completed {
                items: input_regions.len()
            }
        );

        let applied_batch = engine
            .formula_recognizer
            .as_ref()
            .expect("the formula recognizer is created on first use")
            .1
            .max_batch_size();
        assert_eq!(
            applied_batch, 4,
            "the engine must apply the profile's formula_batch to the recognizer"
        );
    }

    /// 内存像素/内存图像输入必须与编码输入一样受 `max_decode_pixels` 约束。
    ///
    /// 公式路由不能成为绕过解码像素上限的旁路。
    #[test]
    fn formula_routing_enforces_the_pixel_limit_for_in_memory_inputs() {
        let Some(mut assets) = assets() else {
            return;
        };
        let page = image::open(&assets.page_with_formula).expect("page readable");
        let (width, height) = (page.width(), page.height());
        let tight_limit = u64::from(width) * u64::from(height) - 1;

        // 1) 编码输入（File）——基线：已经受限制。
        let mut from_file = request(Vec::new(), formula_policy(&assets));
        from_file.input = ImageInput::File(assets.page_with_formula.clone());
        from_file.preprocess.max_decode_pixels = tight_limit;
        let error = assets
            .engine
            .recognize(from_file)
            .expect_err("encoded input must respect the pixel limit");
        assert!(
            matches!(error, crate::error::RapidOcrError::InvalidImage(_)),
            "unexpected error: {error}"
        );

        // 2) 内存像素输入（Pixels）——回归：曾经绕过限制。
        let rgb = page.to_rgb8();
        let mut from_pixels = request(Vec::new(), formula_policy(&assets));
        from_pixels.input = ImageInput::Pixels(crate::api::OwnedPixelBuffer {
            data: Arc::from(rgb.as_raw().as_slice()),
            width,
            height,
            stride: (width * 3) as usize,
            format: crate::api::PixelFormat::Rgb8,
            bottom_up: false,
        });
        from_pixels.preprocess.max_decode_pixels = tight_limit;
        let error = assets
            .engine
            .recognize(from_pixels)
            .expect_err("in-memory pixel input must respect the pixel limit");
        assert!(
            matches!(error, crate::error::RapidOcrError::InvalidImage(_)),
            "unexpected error: {error}"
        );

        // 3) 内存图像输入（Image）——回归：曾经绕过限制。
        let rec = crate::config::RecImage::from_bgr_u8(width as usize, height as usize, {
            let mut bgr = Vec::with_capacity((width * height * 3) as usize);
            for pixel in rgb.pixels() {
                bgr.extend_from_slice(&[pixel[2], pixel[1], pixel[0]]);
            }
            bgr
        })
        .expect("rec image");
        let mut from_image = request(Vec::new(), formula_policy(&assets));
        from_image.input = ImageInput::Image(rec);
        from_image.preprocess.max_decode_pixels = tight_limit;
        let error = assets
            .engine
            .recognize(from_image)
            .expect_err("in-memory image input must respect the pixel limit");
        assert!(
            matches!(error, crate::error::RapidOcrError::InvalidImage(_)),
            "unexpected error: {error}"
        );

        // 放宽限制后同一条内存输入必须能正常跑完，证明上面的失败来自限制而不是输入本身。
        let rgb_again = page.to_rgb8();
        let mut allowed = request(Vec::new(), formula_policy(&assets));
        allowed.input = ImageInput::Pixels(crate::api::OwnedPixelBuffer {
            data: Arc::from(rgb_again.as_raw().as_slice()),
            width,
            height,
            stride: (width * 3) as usize,
            format: crate::api::PixelFormat::Rgb8,
            bottom_up: false,
        });
        allowed.preprocess.max_decode_pixels = u64::from(width) * u64::from(height);
        let output = assets
            .engine
            .recognize(allowed)
            .expect("pixel input within the limit must succeed");
        output.validate().expect("output must validate");
    }

    /// 抹白是**按区域**进行的：区域内的文本消失，远处文本的内容仍然保留。
    ///
    /// 这条测试把 `FormulaPolicy` 文档里描述的“整区域抹白”语义固定下来：
    /// 它不是“按文本框覆盖比例跳过 CTC”，而是先移除区域像素再跑文本管线。
    ///
    /// 同时固定两个实测事实（`tests/` 中没有对应断言，因此写在这里）：
    ///
    /// 1. 抹白改变了文本检测的输入，因此**远处**区域的**分割**也可能变化
    ///    （实测：“42.7 ms” 被重新切成 “42.7” 与 “ms” 两个区域），所以断言按
    ///    去除空白后的内容比较，而不是要求区域边界不变；
    /// 2. 区域**内部**的文本不再出现在文本通道里——这是抹白的核心语义。
    #[test]
    fn whitening_removes_text_inside_the_region_and_keeps_distant_text() {
        let Some(mut assets) = assets() else {
            return;
        };
        let bytes = std::fs::read(&assets.page_with_formula).expect("page readable");
        let baseline = assets
            .engine
            .recognize(request(bytes.clone(), FormulaPolicy::default()))
            .expect("baseline recognize");
        let text_regions: Vec<_> = baseline
            .regions
            .iter()
            .filter(|region| region.kind == RegionKind::Text)
            .filter_map(|region| {
                let polygon = region.polygon?;
                let text = region.recognition.as_ref()?.text.clone();
                (!text.trim().is_empty()).then_some((polygon, text))
            })
            .collect();
        assert!(
            text_regions.len() >= 3,
            "baseline must produce several non-empty text regions"
        );

        let centroid = |polygon: &crate::api::Polygon| {
            let points = polygon.points;
            let sum_x: f32 = points.iter().map(|point| point[0]).sum();
            let sum_y: f32 = points.iter().map(|point| point[1]).sum();
            (sum_x / points.len() as f32, sum_y / points.len() as f32)
        };
        let compact = |text: &str| {
            text.chars()
                .filter(|c| !c.is_whitespace())
                .collect::<String>()
        };

        // 取中间一个文本框作为显式公式区域，再挑出离它最远的文本框。
        let (target_polygon, target_text) = text_regions[text_regions.len() / 2].clone();
        let target_centroid = centroid(&target_polygon);
        let (distant_polygon, distant_text) = text_regions
            .iter()
            .max_by(|left, right| {
                let distance = |candidate: &(crate::api::Polygon, String)| {
                    let (x, y) = centroid(&candidate.0);
                    (x - target_centroid.0).hypot(y - target_centroid.1)
                };
                distance(left)
                    .partial_cmp(&distance(right))
                    .unwrap_or(std::cmp::Ordering::Equal)
            })
            .expect("at least one distant region")
            .clone();
        assert_ne!(distant_polygon, target_polygon);

        let policy = FormulaPolicy {
            enabled: true,
            model_path: Some(assets.formula_model.clone()),
            detector_path: None,
            input_regions: vec![target_polygon],
            ..FormulaPolicy::default()
        };
        let routed = assets
            .engine
            .recognize(request(bytes, policy))
            .expect("routed recognize");
        let routed_text = compact(&routed.plain_text(TextOrder::Reading));

        assert!(
            !routed_text.contains(&compact(&target_text)),
            "text covered by the whitened region must disappear: {target_text:?}"
        );
        assert!(
            routed_text.contains(&compact(&distant_text)),
            "content far from the whitened region must survive (whitespace-insensitive, because \
             whitening can re-segment distant regions): {distant_text:?}"
        );
        assert_eq!(
            routed.formula_count(),
            1,
            "the explicit region must still be recognized as a formula"
        );
    }

    /// **P1-2 引擎级根因回归**：`runtime.auto_tune_threads = false` 必须真的让引擎
    /// “什么都不配置”，而不是被 `RuntimeProfile` 悄悄替换成自动策略的数字。
    ///
    /// 旧行为：`RapidOcrEngine::new` 得到
    /// `ort_intra = Some(budget)` / `ort_inter = Some(1)` / `rayon = Some(clamp(budget/4,1,8))`，
    /// 于是公开字段 `auto_tune_threads = false` 在引擎路径上被完全忽略
    /// （公式路径的 `FormulaSession` 却按 `false` 理解，同一份配置两种行为）。
    ///
    /// 需要真实 OCR 模型，因此沿用同一套外部资产定位：缺失时显式 skip。
    #[test]
    fn engine_honours_auto_tune_threads_disabled() {
        let Some(model_root) = crate::test_support::ocr_model_root() else {
            return;
        };
        let mut config = engine_config(&model_root);
        config.runtime.auto_tune_threads = false;
        config.runtime.intra_threads = None;
        config.runtime.inter_threads = None;
        config.runtime.rayon_threads = None;

        let engine = RapidOcrEngine::new(config).expect("engine should load");
        let threads = &engine.runtime_profile().threads;
        assert_eq!(threads.ort_intra, None);
        assert_eq!(threads.ort_inter, None);
        assert_eq!(threads.rayon, None);
        assert_eq!(threads.source, crate::runtime::profile::ThreadSource::Auto);
        assert_eq!(threads.sessions, 2, "det + rec = two sessions");

        // plan 与“真正下发给 ORT 的配置”必须一致：引擎不再自己造线程数。
        let session = engine.runtime_profile().session_runtime();
        assert_eq!(session.intra_threads, None);
        assert_eq!(session.inter_threads, None);
        assert_eq!(session.rayon_threads, None);
        assert!(!session.auto_tune_threads);
    }

    /// P1-5：输入物化入口收口后，各路由的**解码次数**不得改变。
    ///
    /// `materialize` 取代了文本与分块两条路径各自的五变体 `match`，`Encoded` 输入仍
    /// 在文本通道解码一次（`load_with_limit`）；页面级阶段额外由 `decode_source_image`
    /// 解码一次。这里用真实模型钉住“纯文本=1、页面级=2”，证明统一入口既没有多
    /// 解码一次，也没有把页级路径悄悄降成一次。
    #[test]
    fn materialization_preserves_the_decode_count_per_route() {
        let Some(mut assets) = assets() else {
            return;
        };
        // 纯文本（公式关闭）：`Encoded` 只在 `materialize` 里解码一次。
        let bytes = std::fs::read(&assets.page_without_detections).expect("page readable");
        crate::input::image_loader::reset_decode_calls();
        assets
            .engine
            .recognize(request(bytes.clone(), FormulaPolicy::default()))
            .expect("plain text route must succeed");
        assert_eq!(
            crate::input::image_loader::decode_calls(),
            1,
            "the text channel must decode an Encoded input exactly once"
        );

        // 页面级公式（检测器在该页返回 0 候选 → regions 为空）：`decode_source_image`
        // 与回落到同一段 `Encoded` 的文本通道各解码一次，合计仍为 2。
        crate::input::image_loader::reset_decode_calls();
        let policy = formula_policy(&assets);
        assets
            .engine
            .recognize(request(bytes, policy))
            .expect("page-level formula route must succeed");
        assert_eq!(
            crate::input::image_loader::decode_calls(),
            2,
            "the page-level route must still decode twice (source + text fallback)"
        );
    }
}

/// 表格 HTML 出口策略的纯函数测试（`docs/11` §3 E15.2 / §4 G8）。
///
/// 刻意不依赖任何模型或图片：这一层要钉死的是"哪条路径会被净化"，
/// 与表格内容无关。
#[cfg(test)]
mod table_html_policy_tests {
    use crate::api::TableTextPolicy;

    /// 面向用户的路径（默认 `Escape`）：主动标记必须被剥掉，白名单结构必须留着。
    #[test]
    fn the_user_facing_path_strips_active_markup() {
        let table_html = "<table><tr><td>ok</td></tr></table><script>alert(1)</script>";
        let (html, cleaned) = super::finalize_table_html(
            format!("<html><body>{table_html}</body></html>"),
            table_html.to_string(),
            TableTextPolicy::Escape,
        );
        assert!(!cleaned.contains("<script"), "table_html: {cleaned}");
        assert!(cleaned.contains("<table>"), "table_html: {cleaned}");
        assert!(!html.contains("<script"), "html: {html}");
        // 外壳是**重新**按净化后的内容拼的，不是复用发射器给的那一串。
        assert_eq!(html, format!("<html><body>{cleaned}</body></html>"));
    }

    /// 评测路径（`Raw`）：一个字节都不动 —— 官方 `metric.py` 的可比性依赖这个承诺。
    #[test]
    fn the_evaluation_path_is_byte_identical() {
        let raw =
            "<html><body><table><tr><td><script>alert(1)</script></td></tr></table></body></html>";
        let (html, table_html) =
            super::finalize_table_html(raw.to_string(), raw.to_string(), TableTextPolicy::Raw);
        assert_eq!(table_html, raw);
        assert_eq!(html, raw);
    }

    /// 发射器的真实输出在两条路径下**逐字节相同** —— 这是"净化是纵深防御而不是
    /// 当下的行为变更"的可执行证据。若哪天发射器开始写出白名单外的标记，这条会红，
    /// 而那正是需要知道的时候。
    #[test]
    fn sanitizing_is_a_no_op_on_whitelisted_emitter_output() {
        let table_html = "<table><thead><tr><td rowspan=\"2\" colspan=\"3\">a</td></tr></thead><tbody><tr><td colspan=\"2\">b</td></tr></tbody></table>";
        let shell = format!("<html><body>{table_html}</body></html>");
        let (escaped_html, escaped_table) = super::finalize_table_html(
            shell.clone(),
            table_html.to_string(),
            TableTextPolicy::Escape,
        );
        let (raw_html, raw_table) =
            super::finalize_table_html(shell, table_html.to_string(), TableTextPolicy::Raw);
        assert_eq!(
            escaped_table, raw_table,
            "sanitizing changed emitter output"
        );
        assert_eq!(escaped_table, table_html);
        assert_eq!(escaped_html, raw_html);
    }
}

/// 页面级表格路由的集成测试（`docs/11` §2.1 E14.3–E14.6）。
///
/// 与 `formula_integration_tests` 同一条规矩：真实模型与真实页面图片全部通过
/// 环境变量定位，缺失时**显式 skip**，绝不 panic、绝不回落到开发机绝对路径。
#[cfg(test)]
mod table_pipeline_tests {
    use std::path::PathBuf;
    use std::sync::Arc;

    use crate::api::{
        ClassifierPlan, ClassifierPolicy, DetectionPolicy, ImageInput, OcrEngine, OcrRequest,
        OutputPolicy, PreprocessPolicy, RecognitionPolicy, RegionKind, StagePlan, StageState,
        TablePolicy, TableTextPolicy, WordOutputMode,
    };
    use crate::ocr::pipeline::config::EngineConfig;
    use crate::ocr::pipeline::rapid_structure::RapidOcrEngine;

    /// 与 `formula_integration_tests::engine_config` 同一份模板：表格路径必须先跑
    /// 页级文本通道（回填的文本来源），所以必须真的启用 det + rec。
    fn engine_config(model_root: &std::path::Path) -> EngineConfig {
        let yaml = format!(
            r#"
global:
  use_det: true
  use_cls: false
  use_rec: true
  max_side_len: 2000
  min_side_len: 30
det:
  lang: multi
  ocr_version: PP-OCRv6
  model_type: small
  model_path: {root}/small/PP-OCRv6_det_small.onnx
  allow_download: false
cls:
  lang: ch
rec:
  model:
    lang: ch
    ocr_version: PP-OCRv6
    model_type: small
    model_path: {root}/small/PP-OCRv6_rec_small.onnx
    rec_keys_path: {root}/small/ppocrv6_dict.txt
    allow_download: false
"#,
            root = model_root.display().to_string().replace('\\', "/")
        );
        EngineConfig::from_yaml_str(&yaml).expect("engine config template must parse")
    }

    struct Assets {
        engine: RapidOcrEngine,
        table_model: PathBuf,
        /// `datasets/text/04表格与键值对.png`：页面上有一个 5 列统计表 + 两个并排结构。
        page_with_table: PathBuf,
    }

    fn assets() -> Option<Assets> {
        let model_root = crate::test_support::ocr_model_root()?;
        let table_model = crate::test_support::table_structure_model_path()?;
        let page_with_table = crate::test_support::page_fixture("04表格与键值对.png")?;
        let engine = RapidOcrEngine::new(engine_config(&model_root)).expect("engine should load");
        Some(Assets {
            engine,
            table_model,
            page_with_table,
        })
    }

    fn request(bytes: Vec<u8>, table: TablePolicy) -> OcrRequest {
        OcrRequest {
            input: ImageInput::Encoded(Arc::from(bytes)),
            roi: None,
            scale_hint: None,
            stages: StagePlan {
                detect: true,
                classify: ClassifierPlan {
                    policy: ClassifierPolicy::Off,
                    apply_rotation: false,
                },
                recognize: true,
            },
            preprocess: PreprocessPolicy::default(),
            detection: DetectionPolicy::default(),
            recognition: RecognitionPolicy {
                words: WordOutputMode::Off,
            },
            output: OutputPolicy::default(),
            formula: crate::api::FormulaPolicy::default(),
            table,
        }
    }

    fn table_policy(assets: &Assets) -> TablePolicy {
        TablePolicy {
            enabled: true,
            model_path: Some(assets.table_model.clone()),
            text_policy: TableTextPolicy::Escape,
            ..TablePolicy::default()
        }
    }

    /// 页级输出里所有 `RegionKind::Table` 区域。
    fn table_regions(output: &crate::api::OcrOutput) -> Vec<&crate::api::OcrRegion> {
        output
            .regions
            .iter()
            .filter(|region| region.kind == RegionKind::Table)
            .collect()
    }

    fn bounds(points: &[[f32; 2]; 4]) -> [f32; 4] {
        let mut out = [
            f32::INFINITY,
            f32::INFINITY,
            f32::NEG_INFINITY,
            f32::NEG_INFINITY,
        ];
        for point in points {
            out[0] = out[0].min(point[0]);
            out[1] = out[1].min(point[1]);
            out[2] = out[2].max(point[0]);
            out[3] = out[3].max(point[1]);
        }
        out
    }

    /// E14.3 + E14.4 + E14.6：整页路径真的产出了表格区域，且三处计时/阶段账目都有值。
    ///
    /// 这里同时覆盖 G3 的可观测面：`stages.table.items` 必须等于表格区域数，而每个表格
    /// 区域恰好携带一份 `TableOutcome` —— 结构模型每张表只被调用一次，不随单元格数增长。
    #[test]
    fn the_page_pipeline_produces_a_table_region_with_timings() {
        let Some(mut assets) = assets() else {
            return;
        };
        let bytes = std::fs::read(&assets.page_with_table).expect("the page must be readable");
        let output = assets
            .engine
            .recognize(request(bytes, table_policy(&assets)))
            .expect("the table route must not fail");

        let regions = table_regions(&output);
        assert!(
            !regions.is_empty(),
            "the fixture page contains a table; the locator found none"
        );

        // 记录实测值（不是断言）：docs/11 §2.5 的完成记录要用它，而"表格路径在
        // 整页预算里占多少"只能由真实页面回答。
        for (index, region) in regions.iter().enumerate() {
            let outcome = region.table.as_ref().expect("table outcome");
            println!(
                "TABLE_REGION[{index}] rows={} cols={} cells={} confidence={:?} holes={} \
                 consumed_lines={}",
                outcome.rows,
                outcome.cols,
                outcome.cells.len(),
                outcome.confidence,
                outcome.holes,
                outcome.source_line_indices.len(),
            );
        }
        println!(
            "TABLE_TIMING detect_ms={:.3} infer_ms={:.3} backfill_ms={:.3} page_total_ms={:.3} \
             regions={}",
            output.timings.table_detect_ms,
            output.timings.table_structure_infer_ms,
            output.timings.table_backfill_ms,
            output.timings.total_ms,
            output.regions.len(),
        );

        // E14.4：阶段账目如实报告表格数量。
        match output.stages.table.state {
            StageState::Completed { items } => assert_eq!(
                items,
                regions.len(),
                "the table stage must report exactly the table regions it produced"
            ),
            other => panic!("the table stage must be Completed, got {other:?}"),
        }
        let timing = output
            .stages
            .table
            .timing
            .expect("a completed table stage must carry a timing");
        assert_eq!(
            timing.infer_ms, output.timings.table_structure_infer_ms,
            "the stage timing and the ledger must report the same structure inference time"
        );

        // E14.6：三分量在页面路径上都有值（结构推理与定位必须真的花过时间）。
        assert!(
            output.timings.table_detect_ms > 0.0,
            "the locator must report a non-zero cost"
        );
        assert!(
            output.timings.table_structure_infer_ms > 0.0,
            "the structure model must report a non-zero cost"
        );
        assert!(
            output.timings.table_backfill_ms >= 0.0,
            "the backfill window cannot be negative"
        );
        // 三个分量是 `total_ms` 的一部分，不能超过它。
        let table_total = output.timings.table_detect_ms
            + output.timings.table_structure_infer_ms
            + output.timings.table_backfill_ms;
        assert!(
            table_total <= output.timings.total_ms + 1e-3,
            "table stages ({table_total} ms) must fit inside total_ms ({})",
            output.timings.total_ms
        );

        // 每个表格区域必须携带完整结果，且几何与逻辑网格自洽。
        for region in &regions {
            let outcome = region
                .table
                .as_ref()
                .expect("a table region must carry a table outcome");
            assert!(
                outcome.html.starts_with("<html><body><table>"),
                "the full HTML must carry the page shell: {}",
                outcome.html
            );
            assert!(
                outcome.table_html.starts_with("<table>"),
                "the table-only variant must start at the table element: {}",
                outcome.table_html
            );
            assert!(!outcome.cells.is_empty(), "a located table has cells");
            assert!(
                outcome.rows >= 1 && outcome.cols >= 1,
                "a decoded table must have at least one row and column, got {}x{}",
                outcome.rows,
                outcome.cols
            );
            for cell in &outcome.cells {
                assert!(
                    cell.row < outcome.rows && cell.col < outcome.cols,
                    "cell ({}, {}) falls outside the {}x{} grid",
                    cell.row,
                    cell.col,
                    outcome.rows,
                    outcome.cols
                );
                assert!(cell.rowspan >= 1 && cell.colspan >= 1);
            }
            // 被消费的页级文本行下标必须指向真实存在的区域。
            for index in &outcome.source_line_indices {
                assert!(
                    *index < output.regions.len(),
                    "consumed line index {index} is out of range"
                );
            }
            assert!(
                outcome
                    .source_line_indices
                    .windows(2)
                    .all(|pair| pair[0] < pair[1]),
                "consumed line indices must be sorted and deduplicated: {:?}",
                outcome.source_line_indices
            );
        }
    }

    /// 坐标一致性：表格区域的 polygon 必须**覆盖**它自己的单元格 polygon 的并集。
    ///
    /// 这条断言钉住的是 E11 §12.4.7 登记的那个偏移：单元格几何在裁剪图坐标系里，
    /// 必须加上区域原点才能回到整页坐标系。少加偏移时单元格会整体平移到表格左上角
    /// 附近，本断言立刻失败。
    #[test]
    fn the_table_polygon_covers_its_cells_in_page_coordinates() {
        let Some(mut assets) = assets() else {
            return;
        };
        let bytes = std::fs::read(&assets.page_with_table).expect("the page must be readable");
        let output = assets
            .engine
            .recognize(request(bytes, table_policy(&assets)))
            .expect("the table route must not fail");

        let mut checked = 0;
        for region in table_regions(&output) {
            let outcome = region.table.as_ref().expect("table outcome");
            let Some(table_polygon) = region.polygon else {
                panic!("a table region must carry a polygon");
            };
            let table_box = bounds(&table_polygon.points);
            let mut union: Option<[f32; 4]> = None;
            for cell in &outcome.cells {
                let Some(polygon) = cell.polygon else {
                    continue;
                };
                let cell_box = bounds(&polygon.points);
                union = Some(match union {
                    None => cell_box,
                    Some(current) => [
                        current[0].min(cell_box[0]),
                        current[1].min(cell_box[1]),
                        current[2].max(cell_box[2]),
                        current[3].max(cell_box[3]),
                    ],
                });
            }
            let Some(union) = union else {
                continue;
            };
            checked += 1;
            // 容差取 1 px 的浮点松弛：两端都来自 `floor`/`ceil` 过的整数裁剪框。
            let tolerance = 1.0;
            assert!(
                union[0] >= table_box[0] - tolerance
                    && union[1] >= table_box[1] - tolerance
                    && union[2] <= table_box[2] + tolerance
                    && union[3] <= table_box[3] + tolerance,
                "cell union {union:?} escapes the table polygon {table_box:?}"
            );
        }
        assert!(
            checked > 0,
            "no table region carried cell geometry, so this test proved nothing"
        );
    }

    /// 门禁 G2/G7 的**静态面**：整页路径只能走"复用页级文本行"的回填来源。
    ///
    /// 行为面由 `table_decode_and_backfill` 里的 `debug_assert_eq!(recognizer_calls, 0)`
    /// 保证（测试跑在 debug 构建下，任何一次走 crop 自识别的改动都会立刻失败）；
    /// 这里补一条静态守卫，让"有人把 `Reuse` 换成 `Crop`"在**不跑到那条路径**时也能被发现。
    #[test]
    fn the_page_path_never_asks_the_backfill_to_recognize() {
        let source = include_str!("rapid_structure.rs");
        let test_marker = ["#[cfg", "(test)]"].concat();
        let production = source
            .split(&test_marker)
            .next()
            .expect("the production section must come first");
        assert!(
            production.len() < source.len(),
            "the split must actually cut the test modules away"
        );
        // 探针由碎片拼出，否则本函数自己的字面量会被 `include_str!` 读进来。
        let crop_probe = ["Backfill", "Source::", "Crop"].concat();
        assert_eq!(
            production.matches(&crop_probe).count(),
            0,
            "the page pipeline must reuse page-level lines (G2/G7); crop self-recognition \
             belongs to the offline evaluator, not the page path"
        );
        let reuse_probe = ["Backfill", "Source::", "Reuse"].concat();
        assert_eq!(
            production.matches(&reuse_probe).count(),
            1,
            "the page pipeline must build exactly one reuse backfill source"
        );
        // G3：结构模型每张表恰好调用一次 `run_with`，不随单元格数增长。
        // `attach_table_regions` 已收 `&TablePolicy`，取会话时直接把该引用传入（无 `&`）。
        let run_probe = ["table_session", "(policy)?"].concat();
        assert_eq!(
            production.matches(&run_probe).count(),
            1,
            "the structure session must be obtained once per request, not per cell"
        );
    }

    /// 关闭表格路由时**不得**触碰表格模型，且输出与不带表格功能的引擎逐位一致。
    ///
    /// 这是门禁 G6 在单元层面的对应物（整页 12 图口径由 `tools/run_windows_baseline.ps1` 覆盖）：
    /// 关闭时连模型路径都不需要存在。
    #[test]
    fn a_disabled_table_route_ignores_the_model_path_entirely() {
        let Some(mut assets) = assets() else {
            return;
        };
        let bytes = std::fs::read(&assets.page_with_table).expect("the page must be readable");
        let policy = TablePolicy {
            enabled: false,
            model_path: Some(PathBuf::from("does/not/exist/slanet-plus.onnx")),
            ..TablePolicy::default()
        };
        let output = assets
            .engine
            .recognize(request(bytes, policy))
            .expect("a disabled table route must not try to load the model");
        assert!(
            table_regions(&output).is_empty(),
            "a disabled table route must not produce table regions"
        );
        assert_eq!(
            output.stages.table.state,
            StageState::Disabled,
            "the table stage must be reported as disabled"
        );
        assert_eq!(output.timings.table_detect_ms, 0.0);
        assert_eq!(output.timings.table_structure_infer_ms, 0.0);
        assert_eq!(output.timings.table_backfill_ms, 0.0);
    }

    /// 复审 P1-1（表格站点）：热会话 + 同一路径 + 换一个错误的声明摘要。
    ///
    /// 表格是审核点 2 之后才收凭证的域，它的缓存键因此单独钉一条：第一次用正确摘要
    /// （会话进入缓存），第二次只换声明。第一次还必须有**真表格** —— 定位器没找到
    /// 区域时 `table_session` 根本不会被取用，那种用例只能证明它没有报错。
    #[test]
    fn a_changed_table_declaration_is_not_answered_by_a_hot_session() {
        let Some(model_root) = crate::test_support::ocr_model_root() else {
            return;
        };
        let Some(table_model) = crate::test_support::table_structure_model_path() else {
            return;
        };
        let Some(page) = crate::test_support::page_fixture("04表格与键值对.png") else {
            return;
        };
        let dir = crate::test_support::TempDir::new("pipeline-table-declaration");
        let copy = dir.path().join("slanet-plus.onnx");
        std::fs::copy(&table_model, &copy).expect("copy the real table model");
        let declared = crate::model_store::sha256_file(&copy).expect("hash the copy");

        let mut engine =
            RapidOcrEngine::new(engine_config(&model_root)).expect("engine should load");
        let policy = |expected: Option<String>| TablePolicy {
            enabled: true,
            model_path: Some(copy.clone()),
            expected_model_sha256: expected,
            text_policy: TableTextPolicy::Escape,
            ..TablePolicy::default()
        };
        let bytes = std::fs::read(&page).expect("the page must be readable");
        let output = engine
            .recognize(request(bytes.clone(), policy(Some(declared.clone()))))
            .expect("the first request with the correct digest must pass");
        assert!(
            !table_regions(&output).is_empty(),
            "the session must actually have been built, or this test proves nothing"
        );

        let bogus = format!("{:064x}", 0xabcd_u32);
        let error = engine
            .recognize(request(bytes, policy(Some(bogus.clone()))))
            .expect_err("a changed declaration must not be answered by the hot session");
        match error {
            crate::error::RapidOcrError::HashMismatch {
                path: reported,
                expected,
                actual,
            } => {
                assert_eq!(reported, copy, "the error must name the declared file");
                assert_eq!(expected, bogus);
                assert_eq!(
                    actual, declared,
                    "the file never changed - only the declaration did"
                );
            }
            other => panic!("expected a hash mismatch from the declaration, got {other:?}"),
        }
    }

    /// 复审 P1（表格站点）：热会话 + 同一路径同一摘要 + **只改会话配置**。
    ///
    /// 与上一条同构，钉的是键的另一半：`input_side_len` / `fix_input_shape` /
    /// `intra_threads` 都在 `open` 时一次性施加，命中旧会话时这次的配置根本不生效；
    /// 而同一张请求里的预处理与解码几何已经按**新**边长走（见 [`Self::table_session`] 的文档），
    /// 所以被吞掉不只是“配置没生效”，而是会话与预处理各用一种边长。
    ///
    /// 后半句不是修辞。把键退回“不含 config”后本用例的实测失败：
    /// `Got invalid dimensions for input: x … index: 2 Got: 600 Expected: 488` ——
    /// 预处理交出的张量与新边长一致，而被复用的会话仍钉在旧边长上。
    #[test]
    fn a_changed_table_config_is_not_answered_by_a_hot_session() {
        let Some(model_root) = crate::test_support::ocr_model_root() else {
            return;
        };
        let Some(table_model) = crate::test_support::table_structure_model_path() else {
            return;
        };
        let Some(page) = crate::test_support::page_fixture("04表格与键值对.png") else {
            return;
        };
        let dir = crate::test_support::TempDir::new("pipeline-table-config");
        let copy = dir.path().join("slanet-plus.onnx");
        std::fs::copy(&table_model, &copy).expect("copy the real table model");
        let declared = crate::model_store::sha256_file(&copy).expect("hash the copy");
        let bytes = std::fs::read(&page).expect("the page must be readable");

        let mut engine =
            RapidOcrEngine::new(engine_config(&model_root)).expect("engine should load");
        let default_side = TablePolicy::default().input_side_len;
        let policy =
            |input_side_len: usize, fix_input_shape: bool, intra_threads: Option<usize>| {
                TablePolicy {
                    enabled: true,
                    model_path: Some(copy.clone()),
                    expected_model_sha256: Some(declared.clone()),
                    input_side_len,
                    fix_input_shape,
                    intra_threads,
                    ..TablePolicy::default()
                }
            };

        // 第一次请求走默认配置，会话进入缓存。必须真的有表格区域，否则 `table_session`
        // 根本不会被取用，这条用例在表格页上就什么也没证明。
        let output = engine
            .recognize(request(bytes.clone(), policy(default_side, true, None)))
            .expect("the first request must pass");
        assert!(
            !table_regions(&output).is_empty(),
            "the session must actually have been used by a table region"
        );
        let session = engine
            .table_session(&policy(default_side, true, None))
            .expect("session with the default canvas");
        assert_eq!(session.config().input_side_len, default_side);
        assert!(session.config().fix_input_shape);
        let default_dims = session.input_dims().to_vec();

        // 第二次请求只换画布边长：会话必须反映新配置。若被热会话吞掉，这里仍是旧值。
        let wider = default_side + 112;
        engine
            .recognize(request(bytes.clone(), policy(wider, true, None)))
            .expect("a larger canvas must still infer");
        let session = engine
            .table_session(&policy(wider, true, None))
            .expect("session with the wider canvas");
        assert_eq!(
            session.config().input_side_len,
            wider,
            "配置变化被旧会话吞掉了：键里没有带 input_side_len"
        );
        assert_ne!(
            session.input_dims(),
            default_dims.as_slice(),
            "会话配置里写着新边长、实测维度却还是旧的：override 没有重新施加"
        );
        assert_eq!(
            session.input_dims(),
            [-1, 3, wider as i64, wider as i64],
            "固定形状必须等于这次的 `input_side_len`"
        );

        // 第三次请求只关形状固定：后置条件是可观察的配置本身。
        let session = engine
            .table_session(&policy(wider, false, None))
            .expect("session with a dynamic input shape");
        assert!(
            !session.config().fix_input_shape,
            "配置变化被旧会话吞掉了：键里没有带 fix_input_shape"
        );

        // 第四次请求只改 intra 线程：直接读实际下发给 ONNX Runtime 的后置条件。
        let session = engine
            .table_session(&policy(wider, false, Some(1)))
            .expect("session with a pinned intra budget");
        assert_eq!(
            session.session_threads().0,
            Some(1),
            "配置变化被旧会话吞掉了：键里没有带 intra_threads"
        );
    }

    /// 复审 P1-b（表格站点）：热会话 + 模型文件一个字节没变 + **只改清单里登记的摘要**。
    ///
    /// 策略里写 `None` 不等于“这一次没有期望摘要”：表格域还有一条登记回落。上一条用例钉的是
    /// 显式声明那一半，本条钉回落那一半 —— 旧实现把原始声明（`None`）直接进键，于是登记值
    /// 换了人也不会重校。三段：登记正确 -> 通过；登记被改掉 -> 必红；登记改回来 -> 必绿
    /// （键的变化是双向的，一次失败不会把会话钉死）。
    #[test]
    fn a_changed_table_registration_is_not_answered_by_a_hot_session() {
        let Some(model_root) = crate::test_support::ocr_model_root() else {
            return;
        };
        let Some(table_model) = crate::test_support::table_structure_model_path() else {
            return;
        };
        let Some(page) = crate::test_support::page_fixture("04表格与键值对.png") else {
            return;
        };
        let dir = crate::test_support::TempDir::new("pipeline-table-registration");
        let copy = dir.path().join("slanet-plus.onnx");
        std::fs::copy(&table_model, &copy).expect("copy the real table model");
        let registered = crate::model_store::sha256_file(&copy).expect("hash the copy");
        let size = std::fs::metadata(&copy).expect("stat the copy").len();
        let bytes = std::fs::read(&page).expect("the page must be readable");

        // 摘要只写在清单里，策略留 `None`：走的就是那条回落路径，也就是必须进键的那一份。
        let policy = TablePolicy {
            enabled: true,
            model_path: Some(copy.clone()),
            expected_model_sha256: None,
            text_policy: TableTextPolicy::Escape,
            ..TablePolicy::default()
        };
        let write_manifest = |sha256: &str| {
            dir.write(
                "manifest.json",
                format!(
                    r#"{{"schema_version":1,"id":"scratch","family":"SLANet","version":"v2.0.0","files":[
                        {{"name":"slanet-plus.onnx","role":"table_structure","sha256":"{sha256}","size_bytes":{size}}}
                    ]}}"#
                )
                    .as_bytes(),
            );
        };
        write_manifest(&registered);

        let mut engine =
            RapidOcrEngine::new(engine_config(&model_root)).expect("engine should load");
        let output = engine
            .recognize(request(bytes.clone(), policy.clone()))
            .expect("the first request must pass on the registered digest");
        assert!(
            !table_regions(&output).is_empty(),
            "the session must actually have been used by a table region"
        );

        // 模型没动，只把清单里的登记摘要改掉：旧会话不许继续回答。
        let bogus = format!("{:064x}", 0xabcd_u32);
        write_manifest(&bogus);
        let error = engine
            .recognize(request(bytes.clone(), policy.clone()))
            .expect_err("a changed registration must not be answered by the hot session");
        match error {
            crate::error::RapidOcrError::HashMismatch {
                path: reported,
                expected,
                actual,
            } => {
                assert_eq!(reported, copy, "the error must name the model file");
                assert_eq!(
                    expected, bogus,
                    "期望必须来自清单里的登记值，而不是策略里那个 `None`"
                );
                assert_eq!(
                    actual, registered,
                    "the file never changed - only the registration did"
                );
            }
            other => panic!("expected a hash mismatch from the registration, got {other:?}"),
        }

        // 改回来必须又能跑：键的变化是双向的。
        write_manifest(&registered);
        let output = engine
            .recognize(request(bytes, policy))
            .expect("restoring the registration must pass again");
        assert!(!table_regions(&output).is_empty());
    }

    /// 表格路由与 `roi`/`tile` 互斥，必须在进入管线前结构化拒绝（与公式路由同形）。
    ///
    /// 坐标**恰好**仍能对上（页级文本区域的 polygon 会被 `+ offset` 映射回原图），所以
    /// 这不是一次会崩的错，而是一次会静默返回调用方没问过的结果的错：`roi` 只裁了输入，
    /// 而定位器读的是整页文本框分布。
    #[test]
    fn table_routing_rejects_roi_and_tile() {
        let base = TablePolicy {
            enabled: true,
            model_path: Some(PathBuf::from("model.onnx")),
            ..TablePolicy::default()
        };
        let mut with_roi = request(Vec::new(), base.clone());
        with_roi.roi = Some(crate::api::RectU32 {
            x: 0,
            y: 0,
            width: 10,
            height: 10,
        });
        let error = with_roi.validate().expect_err("roi must be rejected");
        assert!(error.to_string().contains("roi"), "error: {error}");

        let mut with_tile = request(Vec::new(), base);
        with_tile.preprocess.tile = Some(crate::api::TilePolicy {
            max_width: 100,
            max_height: 100,
            overlap: 10,
        });
        let error = with_tile.validate().expect_err("tile must be rejected");
        assert!(error.to_string().contains("tiled"), "error: {error}");
    }
}

#[cfg(test)]
mod session_cache_rule_tests {
    //! 会话缓存判据本身的覆盖（复审 P1-1 / P1）：**不加载任何模型**。
    //!
    //! 三个会话点共用 [`needs_rebuild`]，而真实模型用例在缺少资产的环境里会**静默 skip**；
    //! 所以“声明变了必须重建”“会话构造参数变了必须重建”这两条规则自己必须各有一条无模型
    //! 的守卫，否则它们在干净 clone 上什么也没被证明。三个站点的端到端行为各有一条
    //! 真实模型用例钉着。
    //!
    //! 分工也要记清楚：本模块钉的是**键类型**，钉不了**调用点**。把表格站点的键退回
    //! “不含 config”或“不含登记摘要”时，这里的用例仍然全绿（它拿的是自己构造的键），
    //! 失败只会出现在两条真实模型用例上 —— 所以两边都不可省。

    use super::{SessionCacheKey, needs_rebuild};
    use crate::model_verify::FileIdentity;
    use crate::table::structure::TableStructureConfig;
    use crate::test_support::TempDir;

    const HELLO_SHA256: &str = "2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824";
    const BOGUS_SHA256: &str = "0000000000000000000000000000000000000000000000000000000000000000";

    /// 核心回归：文件身份完全相同，**只有声明换了** → 必须重建。
    ///
    /// 不重建就等于把新声明吞掉：旧会话直接回答，那条摘要从未参与比对。
    #[test]
    fn a_changed_declaration_with_an_unchanged_identity_rebuilds() {
        let dir = TempDir::new("cache-rule-declaration");
        dir.write("model.onnx", b"hello");
        let file = FileIdentity::of(&dir.path().join("model.onnx")).expect("identity");
        let first = SessionCacheKey::engine_only(file.clone(), Some(HELLO_SHA256));

        assert!(
            !needs_rebuild(Some(&first), &first),
            "the same declaration against the same identity is a hit"
        );
        assert!(
            needs_rebuild(
                Some(&first),
                &SessionCacheKey::engine_only(file.clone(), Some(BOGUS_SHA256))
            ),
            "a different declaration on an unchanged file must not be answered by the hot session"
        );
        assert!(
            !needs_rebuild(
                Some(&first),
                &SessionCacheKey::engine_only(file.clone(), Some(&HELLO_SHA256.to_uppercase()))
            ),
            "case is the only difference verification ignores, so it must not force a rebuild"
        );
        assert!(
            needs_rebuild(
                Some(&first),
                &SessionCacheKey::engine_only(file.clone(), None)
            ),
            "dropping the declaration is a different statement about the file"
        );
        // 反向（复审 P1-b 的键侧形状）：第一次什么期望都没有（无声明也无登记），后来清单
        // 登记了一个值 -> 也必须重建，否则“受清单管辖”这件事永远追不上已经活着的会话。
        let undeclared = SessionCacheKey::engine_only(file.clone(), None);
        assert!(needs_rebuild(
            Some(&undeclared),
            &SessionCacheKey::engine_only(file, Some(BOGUS_SHA256))
        ));
        assert!(needs_rebuild(None, &first), "an empty cache always builds");
    }

    /// 键不能比校验更宽松：带空白的写法在 `verify` 里是另一个（会失败的）判定，
    /// 折叠它就会让第二个错误声明因为命中缓存而根本不参与比对。
    #[test]
    fn a_padded_declaration_is_a_different_declaration() {
        let dir = TempDir::new("cache-rule-padded");
        dir.write("model.onnx", b"hello");
        let file = FileIdentity::of(&dir.path().join("model.onnx")).expect("identity");
        let first = SessionCacheKey::engine_only(file.clone(), Some(HELLO_SHA256));
        assert!(needs_rebuild(
            Some(&first),
            &SessionCacheKey::engine_only(file, Some(&format!(" {HELLO_SHA256}")))
        ));
    }

    /// 文件侧：内容变（体积变）与路径变（内容相同）都必须重建。
    #[test]
    fn a_changed_file_or_a_changed_path_rebuilds() {
        let dir = TempDir::new("cache-rule-file");
        dir.write("model.onnx", b"hello");
        let before = FileIdentity::of(&dir.path().join("model.onnx")).expect("identity");
        dir.write("model.onnx", b"hello!");
        let rewritten = FileIdentity::of(&dir.path().join("model.onnx")).expect("identity");
        assert!(needs_rebuild(
            Some(&SessionCacheKey::engine_only(
                before.clone(),
                Some(HELLO_SHA256)
            )),
            &SessionCacheKey::engine_only(rewritten, Some(HELLO_SHA256))
        ));

        dir.write("other.onnx", b"hello");
        let other_path = FileIdentity::of(&dir.path().join("other.onnx")).expect("identity");
        assert!(
            needs_rebuild(
                Some(&SessionCacheKey::engine_only(
                    before.clone(),
                    Some(HELLO_SHA256)
                )),
                &SessionCacheKey::engine_only(other_path, Some(HELLO_SHA256))
            ),
            "the same bytes under a different path are a different file"
        );
    }

    /// 同一路径、同一份内容：两次独立观察必须给出同一个键，否则每条请求都在重建会话。
    #[test]
    fn two_observations_of_an_untouched_file_are_the_same_key() {
        let dir = TempDir::new("cache-rule-stable");
        dir.write("model.onnx", b"hello");
        let path = dir.path().join("model.onnx");
        let first = SessionCacheKey::engine_only(
            FileIdentity::of(&path).expect("identity"),
            Some(HELLO_SHA256),
        );
        let second = SessionCacheKey::engine_only(
            FileIdentity::of(&path).expect("identity"),
            Some(HELLO_SHA256),
        );
        assert!(!needs_rebuild(Some(&first), &second));
    }

    /// 复审 P1（表格站点）：同一文件、同一声明，**只改会话构造参数** → 必须重建。
    ///
    /// 这三只旋钮都在 `open` 时一次性施加（线程选项与 free-dimension override 都没有
    /// 第二次机会），命中旧会话等于这次的配置根本没生效。键里存的是**整个**
    /// [`TableStructureConfig`]，所以本用例按字段逐条变异：任何一只旋钮的差异都不允许
    /// 被判成“键相同”。
    #[test]
    fn a_changed_table_config_rebuilds_without_any_file_change() {
        let dir = TempDir::new("cache-rule-table-config");
        dir.write("slanet-plus.onnx", b"hello");
        let file = FileIdentity::of(&dir.path().join("slanet-plus.onnx")).expect("identity");
        let base = TableStructureConfig::default();
        let first = SessionCacheKey::table(file.clone(), Some(HELLO_SHA256), &base);

        assert!(
            !needs_rebuild(
                Some(&first),
                &SessionCacheKey::table(file.clone(), Some(HELLO_SHA256), &base)
            ),
            "identity、声明、配置全都不变 -> 不应该重建"
        );

        let mut wider = base.clone();
        wider.input_side_len += 112;
        assert!(
            needs_rebuild(
                Some(&first),
                &SessionCacheKey::table(file.clone(), Some(HELLO_SHA256), &wider)
            ),
            "input_side_len 变了却不重建 = 新画布被旧会话吞掉"
        );

        let mut unfixed = base.clone();
        unfixed.fix_input_shape = !unfixed.fix_input_shape;
        assert!(
            needs_rebuild(
                Some(&first),
                &SessionCacheKey::table(file.clone(), Some(HELLO_SHA256), &unfixed)
            ),
            "fix_input_shape 变了却不重建 = 形状策略被旧会话吞掉"
        );

        let mut threaded = base.clone();
        threaded.intra_threads = Some(base.intra_threads.map_or(1, |threads| threads + 1));
        assert!(
            needs_rebuild(
                Some(&first),
                &SessionCacheKey::table(file.clone(), Some(HELLO_SHA256), &threaded)
            ),
            "intra_threads 变了却不重建 = 线程预算被旧会话吞掉"
        );

        // 反面守卫：config 真的参与了键的相等性。把字段从 `SessionCacheKey` 里删掉，
        // 表格键就会与“构造参数不来自请求”的公式键完全重合，而上面的三条会一起失效。
        assert_ne!(
            first,
            SessionCacheKey::engine_only(file, Some(HELLO_SHA256)),
            "两个站点即使文件身份与声明完全相同，也不是同一把键"
        );
    }
}