diff --git a/app/services/report_over_limit.py b/app/services/report_over_limit.py
new file mode 100644
index 0000000..cd9ebcf
--- /dev/null
+++ b/app/services/report_over_limit.py
@@ -0,0 +1,385 @@
+from __future__ import annotations
+
+import logging
+import re
+from dataclasses import dataclass
+from decimal import Decimal, InvalidOperation, ROUND_HALF_UP
+
+from sqlalchemy import func, select, text
+from sqlalchemy.exc import SQLAlchemyError
+from sqlalchemy.orm import Session
+
+from app.models import Product, ProductionReport, ProductionReportItem, ReportStatus
+from app.services.cleaning import is_cleaning_item
+from app.services.misc_work import is_misc_item
+
+
+logger = logging.getLogger(__name__)
+
+
+class OverLimitDataSourceError(RuntimeError):
+    pass
+
+
+OVER_LIMIT_STATUS_NOT_CHECKED = "not_checked"
+OVER_LIMIT_STATUS_OK = "ok"
+OVER_LIMIT_STATUS_EXCEEDED = "exceeded"
+OVER_LIMIT_STATUS_UNCHECKABLE = "uncheckable"
+
+OVER_LIMIT_TYPE_FIRST_PROCESS_MATERIAL = "first_process_material"
+OVER_LIMIT_TYPE_PREVIOUS_PROCESS = "previous_process"
+OVER_LIMIT_TYPE_UNCHECKABLE = "uncheckable"
+
+CHECK_SOURCE_SUBMIT = "submit"
+CHECK_SOURCE_REVIEW = "review"
+
+_TWO_PLACES = Decimal("0.01")
+
+
+@dataclass(frozen=True)
+class OverLimitResult:
+    status: str
+    over_limit_type: str | None = None
+    reason: str | None = None
+    batch_no: str | None = None
+    product_gross_weight_kg: Decimal | None = None
+    issued_weight_kg: Decimal | None = None
+    max_reportable_qty: Decimal | None = None
+    previous_good_qty: Decimal | None = None
+    current_cumulative_qty: Decimal | None = None
+    current_report_qty: Decimal | None = None
+
+
+def _to_decimal(value) -> Decimal:
+    if value is None:
+        return Decimal("0")
+    if isinstance(value, Decimal):
+        return value
+    try:
+        return Decimal(str(value))
+    except (InvalidOperation, ValueError):
+        return Decimal("0")
+
+
+def _money_qty(value: Decimal) -> Decimal:
+    return value.quantize(_TWO_PLACES, rounding=ROUND_HALF_UP)
+
+
+def _positive_decimal(value) -> Decimal | None:
+    number = _to_decimal(value)
+    return number if number > 0 else None
+
+
+def _clean_text(value) -> str:
+    return str(value or "").strip()
+
+
+def _batch_no(value) -> str:
+    return _clean_text(value)
+
+
+def parse_process_order(process_name: str | None) -> int | None:
+    text_value = _clean_text(process_name)
+    match = re.fullmatch(r"0*(\d+)(?:\s*序)?", text_value)
+    if match is None:
+        return None
+    order = int(match.group(1))
+    return order if order > 0 else None
+
+
+def current_report_qty(item) -> Decimal:
+    total = _to_decimal(getattr(item, "good_qty", None))
+    total += _to_decimal(getattr(item, "defect_qty", None))
+    total += _to_decimal(getattr(item, "scrap_qty", None))
+    return _money_qty(total)
+
+
+def read_batch_issued_weight_kg(
+    db: Session,
+    batch_no: str | None,
+    project_no: str | None,
+    product_name: str | None,
+) -> Decimal | None:
+    normalized_batch_no = _batch_no(batch_no)
+    normalized_project_no = _clean_text(project_no)
+    normalized_product_name = _clean_text(product_name)
+    if not normalized_batch_no:
+        return None
+
+    try:
+        value = db.scalar(
+            text(
+                """
+                select sum(ledger.total_issued_weight_kg)
+                from pp_production_batch_ledger ledger
+                join md_item product on product.id = ledger.product_item_id
+                where ledger.material_lot_no = :batch_no
+                  and coalesce(ledger.miniapp_selectable, 1) = 1
+                  and (
+                    (:product_name <> '' and :project_no <> '' and product.item_name = :product_name and product.item_code = :project_no)
+                    or (:product_name <> '' and :project_no = '' and product.item_name = :product_name)
+                    or (:product_name = '' and :project_no <> '' and product.item_code = :project_no)
+                  )
+                """
+            ),
+            {
+                "batch_no": normalized_batch_no,
+                "project_no": normalized_project_no,
+                "product_name": normalized_product_name,
+            },
+        )
+    except SQLAlchemyError as exc:
+        logger.exception(
+            "failed to read production batch issued weight for batch=%s project=%s product=%s",
+            normalized_batch_no,
+            normalized_project_no,
+            normalized_product_name,
+        )
+        raise OverLimitDataSourceError("failed to read batch issued weight") from exc
+    weight = _positive_decimal(value)
+    return weight
+
+
+def _product_catalog_row(db: Session, item: ProductionReportItem) -> Product | None:
+    return db.get(
+        Product,
+        {
+            "attendance_point_name": getattr(item, "attendance_point_name", "") or "",
+            "project_no": getattr(item, "project_no", "") or "",
+            "product_name": getattr(item, "product_name", "") or "",
+            "device_no": "",
+            "process_name": getattr(item, "process_name", "") or "",
+        },
+    )
+
+
+def _process_orders_for_product(db: Session, item: ProductionReportItem) -> list[int]:
+    rows = db.scalars(
+        select(Product.process_name).where(
+            Product.attendance_point_name == (getattr(item, "attendance_point_name", "") or ""),
+            Product.project_no == (getattr(item, "project_no", "") or ""),
+            Product.product_name == (getattr(item, "product_name", "") or ""),
+            Product.device_no == "",
+        )
+    ).all()
+    return sorted({order for name in rows if (order := parse_process_order(name)) is not None})
+
+
+def _previous_process_order(db: Session, item: ProductionReportItem, current_order: int) -> int | None:
+    previous_orders = [order for order in _process_orders_for_product(db, item) if order < current_order]
+    return max(previous_orders) if previous_orders else None
+
+
+def _process_name_candidates(process_order: int) -> tuple[str, ...]:
+    values: set[str] = set()
+    for width in range(1, 5):
+        value = str(process_order).zfill(width)
+        values.add(value)
+        values.add(f"{value}序")
+        values.add(f"{value} 序")
+    return tuple(sorted(values))
+
+
+def _history_items(
+    db: Session,
+    item: ProductionReportItem,
+    process_order: int,
+    exclude_report_id: int | None,
+) -> list[ProductionReportItem]:
+    query = (
+        select(ProductionReportItem)
+        .join(ProductionReport)
+        .where(
+            ProductionReport.status.in_([ReportStatus.pending, ReportStatus.approved]),
+            ProductionReport.is_voided.is_(False),
+            ProductionReportItem.attendance_point_name == (getattr(item, "attendance_point_name", "") or ""),
+            ProductionReportItem.project_no == (getattr(item, "project_no", "") or ""),
+            ProductionReportItem.product_name == (getattr(item, "product_name", "") or ""),
+            func.trim(ProductionReportItem.raw_material_batch_no) == _batch_no(getattr(item, "raw_material_batch_no", "")),
+            func.trim(ProductionReportItem.process_name).in_(_process_name_candidates(process_order)),
+        )
+    )
+    if exclude_report_id is not None:
+        query = query.where(ProductionReport.id != exclude_report_id)
+    return list(db.scalars(query).all())
+
+
+def _historical_consumed_qty(
+    db: Session,
+    item: ProductionReportItem,
+    process_order: int,
+    exclude_report_id: int | None,
+) -> Decimal:
+    total = Decimal("0")
+    for history_item in _history_items(db, item, process_order, exclude_report_id):
+        total += current_report_qty(history_item)
+    return _money_qty(total)
+
+
+def _historical_good_qty(
+    db: Session,
+    item: ProductionReportItem,
+    process_order: int,
+    exclude_report_id: int | None,
+) -> Decimal:
+    total = Decimal("0")
+    for history_item in _history_items(db, item, process_order, exclude_report_id):
+        total += _to_decimal(history_item.good_qty)
+    return _money_qty(total)
+
+
+def _format_process(order: int) -> str:
+    return f"{order}序"
+
+
+def _same_scope(left: ProductionReportItem, right: ProductionReportItem) -> bool:
+    return (
+        (getattr(left, "attendance_point_name", "") or "") == (getattr(right, "attendance_point_name", "") or "")
+        and (getattr(left, "project_no", "") or "") == (getattr(right, "project_no", "") or "")
+        and (getattr(left, "product_name", "") or "") == (getattr(right, "product_name", "") or "")
+        and _batch_no(getattr(left, "raw_material_batch_no", "")) == _batch_no(getattr(right, "raw_material_batch_no", ""))
+    )
+
+
+def _current_report_consumed_qty(
+    report: ProductionReport,
+    item: ProductionReportItem,
+    process_order: int,
+) -> Decimal:
+    total = Decimal("0")
+    for report_item in list(getattr(report, "items", []) or []):
+        if _same_scope(item, report_item) and parse_process_order(getattr(report_item, "process_name", None)) == process_order:
+            total += current_report_qty(report_item)
+    return _money_qty(total)
+
+
+def _current_report_good_qty(
+    report: ProductionReport,
+    item: ProductionReportItem,
+    process_order: int,
+) -> Decimal:
+    total = Decimal("0")
+    for report_item in list(getattr(report, "items", []) or []):
+        if _same_scope(item, report_item) and parse_process_order(getattr(report_item, "process_name", None)) == process_order:
+            total += _to_decimal(getattr(report_item, "good_qty", 0))
+    return _money_qty(total)
+
+
+def evaluate_report_item_over_limit(
+    db: Session,
+    report: ProductionReport,
+    item: ProductionReportItem,
+    *,
+    exclude_report_id: int | None = None,
+) -> OverLimitResult:
+    if exclude_report_id is None:
+        exclude_report_id = getattr(report, "id", None)
+
+    batch_no = _batch_no(getattr(item, "raw_material_batch_no", "")) or None
+    report_qty = current_report_qty(item)
+
+    if is_cleaning_item(item) or is_misc_item(item) or str(getattr(item, "process_name", "") or "").strip() == "清洗":
+        return OverLimitResult(
+            status=OVER_LIMIT_STATUS_NOT_CHECKED,
+            batch_no=batch_no,
+            current_report_qty=report_qty,
+        )
+
+    current_order = parse_process_order(getattr(item, "process_name", None))
+    if current_order is None:
+        return OverLimitResult(
+            status=OVER_LIMIT_STATUS_UNCHECKABLE,
+            over_limit_type=OVER_LIMIT_TYPE_UNCHECKABLE,
+            reason="非数字工序，无法校验超额",
+            batch_no=batch_no,
+            current_report_qty=report_qty,
+        )
+
+    if not batch_no:
+        return OverLimitResult(
+            status=OVER_LIMIT_STATUS_UNCHECKABLE,
+            over_limit_type=OVER_LIMIT_TYPE_UNCHECKABLE,
+            reason="缺少材料批次，无法校验超额",
+            current_report_qty=report_qty,
+        )
+
+    historical_qty = _historical_consumed_qty(db, item, current_order, exclude_report_id)
+    current_cumulative_qty = _money_qty(historical_qty + _current_report_consumed_qty(report, item, current_order))
+    previous_order = _previous_process_order(db, item, current_order)
+
+    if previous_order is None:
+        product = _product_catalog_row(db, item)
+        gross_weight = _positive_decimal(getattr(product, "product_gross_weight_kg", None))
+        if gross_weight is None:
+            return OverLimitResult(
+                status=OVER_LIMIT_STATUS_UNCHECKABLE,
+                over_limit_type=OVER_LIMIT_TYPE_UNCHECKABLE,
+                reason="缺少产品毛重，无法校验超额",
+                batch_no=batch_no,
+                current_cumulative_qty=current_cumulative_qty,
+                current_report_qty=report_qty,
+            )
+
+        try:
+            issued_weight = read_batch_issued_weight_kg(db, batch_no, item.project_no, item.product_name)
+        except OverLimitDataSourceError:
+            return OverLimitResult(
+                status=OVER_LIMIT_STATUS_UNCHECKABLE,
+                over_limit_type=OVER_LIMIT_TYPE_UNCHECKABLE,
+                reason="读取批次投入重量失败，无法校验超额",
+                batch_no=batch_no,
+                product_gross_weight_kg=gross_weight,
+                current_cumulative_qty=current_cumulative_qty,
+                current_report_qty=report_qty,
+            )
+        if issued_weight is None:
+            return OverLimitResult(
+                status=OVER_LIMIT_STATUS_UNCHECKABLE,
+                over_limit_type=OVER_LIMIT_TYPE_UNCHECKABLE,
+                reason="缺少批次投入重量，无法校验超额",
+                batch_no=batch_no,
+                product_gross_weight_kg=gross_weight,
+                current_cumulative_qty=current_cumulative_qty,
+                current_report_qty=report_qty,
+            )
+
+        max_reportable_qty = _money_qty(issued_weight / gross_weight)
+        status = OVER_LIMIT_STATUS_EXCEEDED if current_cumulative_qty > max_reportable_qty else OVER_LIMIT_STATUS_OK
+        reason = None
+        if status == OVER_LIMIT_STATUS_EXCEEDED:
+            reason = (
+                f"{_format_process(current_order)}累计报工数量超过本批次材料可支撑数量："
+                f"{current_cumulative_qty} > {max_reportable_qty}"
+            )
+        return OverLimitResult(
+            status=status,
+            over_limit_type=OVER_LIMIT_TYPE_FIRST_PROCESS_MATERIAL if status == OVER_LIMIT_STATUS_EXCEEDED else None,
+            reason=reason,
+            batch_no=batch_no,
+            product_gross_weight_kg=gross_weight,
+            issued_weight_kg=issued_weight,
+            max_reportable_qty=max_reportable_qty,
+            current_cumulative_qty=current_cumulative_qty,
+            current_report_qty=report_qty,
+        )
+
+    previous_good_qty = _money_qty(
+        _historical_good_qty(db, item, previous_order, exclude_report_id)
+        + _current_report_good_qty(report, item, previous_order)
+    )
+    status = OVER_LIMIT_STATUS_EXCEEDED if current_cumulative_qty > previous_good_qty else OVER_LIMIT_STATUS_OK
+    reason = None
+    if status == OVER_LIMIT_STATUS_EXCEEDED:
+        reason = (
+            f"{_format_process(current_order)}累计报工数量超过{_format_process(previous_order)}累计成品数量："
+            f"{current_cumulative_qty} > {previous_good_qty}"
+        )
+    return OverLimitResult(
+        status=status,
+        over_limit_type=OVER_LIMIT_TYPE_PREVIOUS_PROCESS if status == OVER_LIMIT_STATUS_EXCEEDED else None,
+        reason=reason,
+        batch_no=batch_no,
+        previous_good_qty=previous_good_qty,
+        current_cumulative_qty=current_cumulative_qty,
+        current_report_qty=report_qty,
+    )
diff --git a/tests/test_report_over_limit.py b/tests/test_report_over_limit.py
index 726d0d1..cb24585 100644
--- a/tests/test_report_over_limit.py
+++ b/tests/test_report_over_limit.py
@@ -1,8 +1,21 @@
+from datetime import datetime
+from decimal import Decimal
+from types import SimpleNamespace
+
 from sqlalchemy import BigInteger, create_engine, inspect
 from sqlalchemy.ext.compiler import compiles
+from sqlalchemy.orm import sessionmaker
 
 from app.database import Base
-from app.models import ProductionReport, ProductionReportItem  # noqa: F401
+from app.models import Product, ProductionReport, ProductionReportItem, ReportStatus
+from app.services.report_over_limit import (
+    OVER_LIMIT_STATUS_EXCEEDED,
+    OVER_LIMIT_STATUS_NOT_CHECKED,
+    OVER_LIMIT_TYPE_UNCHECKABLE,
+    current_report_qty,
+    evaluate_report_item_over_limit,
+    parse_process_order,
+)
 
 
 @compiles(BigInteger, "sqlite")
@@ -56,3 +69,449 @@ def test_report_over_limit_columns_exist_on_sqlite_metadata():
     assert str(report_column_by_name["has_over_limit"]["default"]).strip("'\"") in {"0", "false", "FALSE"}
     assert item_column_by_name["over_limit_status"]["nullable"] is False
     assert str(item_column_by_name["over_limit_status"]["default"]).strip("'\"") == "not_checked"
+
+
+def _sqlite_db():
+    engine = create_engine("sqlite+pysqlite:///:memory:", future=True)
+    Base.metadata.create_all(engine)
+    SessionLocal = sessionmaker(bind=engine, future=True)
+    return SessionLocal()
+
+
+def test_parse_process_order_accepts_numeric_process_names():
+    assert parse_process_order("1") == 1
+    assert parse_process_order("1序") == 1
+    assert parse_process_order(" 02 序 ") == 2
+
+
+def test_parse_process_order_rejects_special_and_text_process_names():
+    assert parse_process_order("杂活") is None
+    assert parse_process_order("清洗") is None
+    assert parse_process_order("第一序") is None
+
+
+def test_current_report_qty_uses_good_defect_and_scrap():
+    item = SimpleNamespace(good_qty=Decimal("90"), defect_qty=Decimal("5"), scrap_qty=Decimal("5"))
+
+    assert current_report_qty(item) == Decimal("100.00")
+
+
+def _report(report_id: int, status: ReportStatus = ReportStatus.pending) -> ProductionReport:
+    return ProductionReport(
+        id=report_id,
+        session_id=report_id,
+        attendance_point_name="总厂",
+        employee_phone="13800000000",
+        report_date=datetime(2026, 7, 25).date(),
+        start_at=datetime(2026, 7, 25, 8),
+        end_at=datetime(2026, 7, 25, 10),
+        status=status,
+    )
+
+
+def test_first_process_exceeds_material_limit_with_submit_snapshot(monkeypatch):
+    db = _sqlite_db()
+    try:
+        product = Product(
+            attendance_point_name="总厂",
+            project_no="P1",
+            product_name="产品A",
+            device_no="",
+            process_name="1",
+            product_gross_weight_kg=Decimal("2"),
+            standard_beat=1,
+        )
+        report = _report(1)
+        item = ProductionReportItem(
+            id=1,
+            report=report,
+            attendance_point_name="总厂",
+            device_no="A",
+            project_no="P1",
+            product_name="产品A",
+            raw_material_batch_no="LOT-1",
+            process_name="1",
+            good_qty=60,
+            defect_qty=0,
+            scrap_qty=0,
+            standard_beat=1,
+        )
+        db.add_all([product, report, item])
+        db.flush()
+        monkeypatch.setattr(
+            "app.services.report_over_limit.read_batch_issued_weight_kg",
+            lambda db, batch_no, project_no, product_name: Decimal("100"),
+        )
+
+        result = evaluate_report_item_over_limit(db, report, item, exclude_report_id=report.id)
+
+        assert result.status == OVER_LIMIT_STATUS_EXCEEDED
+        assert result.max_reportable_qty == Decimal("50.00")
+        assert result.current_report_qty == Decimal("60.00")
+        assert result.current_cumulative_qty == Decimal("60.00")
+        assert "1序累计报工数量超过本批次材料可支撑数量" in result.reason
+    finally:
+        db.close()
+
+
+def test_evaluate_defaults_to_excluding_current_report(monkeypatch):
+    db = _sqlite_db()
+    try:
+        product = Product(
+            attendance_point_name="总厂",
+            project_no="P1",
+            product_name="产品A",
+            device_no="",
+            process_name="1",
+            product_gross_weight_kg=Decimal("2"),
+            standard_beat=1,
+        )
+        report = _report(1)
+        item = ProductionReportItem(
+            id=1,
+            report=report,
+            attendance_point_name="总厂",
+            device_no="A",
+            project_no="P1",
+            product_name="产品A",
+            raw_material_batch_no="LOT-1",
+            process_name="1",
+            good_qty=50,
+            defect_qty=0,
+            scrap_qty=0,
+            standard_beat=1,
+        )
+        db.add_all([product, report, item])
+        db.flush()
+        monkeypatch.setattr(
+            "app.services.report_over_limit.read_batch_issued_weight_kg",
+            lambda db, batch_no, project_no, product_name: Decimal("100"),
+        )
+
+        result = evaluate_report_item_over_limit(db, report, item)
+
+        assert result.current_cumulative_qty == Decimal("50.00")
+        assert result.status != OVER_LIMIT_STATUS_EXCEEDED
+    finally:
+        db.close()
+
+
+def test_current_report_sibling_items_contribute_to_current_cumulative(monkeypatch):
+    db = _sqlite_db()
+    try:
+        product = Product(
+            attendance_point_name="总厂",
+            project_no="P1",
+            product_name="产品A",
+            device_no="",
+            process_name="1",
+            product_gross_weight_kg=Decimal("2"),
+            standard_beat=1,
+        )
+        report = _report(1)
+        first_item = ProductionReportItem(
+            id=1,
+            report=report,
+            attendance_point_name="总厂",
+            device_no="A",
+            project_no="P1",
+            product_name="产品A",
+            raw_material_batch_no="LOT-1",
+            process_name="1",
+            good_qty=30,
+            defect_qty=0,
+            scrap_qty=0,
+            standard_beat=1,
+        )
+        second_item = ProductionReportItem(
+            id=2,
+            report=report,
+            attendance_point_name="总厂",
+            device_no="B",
+            project_no="P1",
+            product_name="产品A",
+            raw_material_batch_no=" LOT-1 ",
+            process_name="1",
+            good_qty=30,
+            defect_qty=0,
+            scrap_qty=0,
+            standard_beat=1,
+        )
+        db.add_all([product, report, first_item, second_item])
+        db.flush()
+        monkeypatch.setattr(
+            "app.services.report_over_limit.read_batch_issued_weight_kg",
+            lambda db, batch_no, project_no, product_name: Decimal("100"),
+        )
+
+        result = evaluate_report_item_over_limit(db, report, first_item)
+
+        assert result.current_report_qty == Decimal("30.00")
+        assert result.current_cumulative_qty == Decimal("60.00")
+        assert result.status == OVER_LIMIT_STATUS_EXCEEDED
+    finally:
+        db.close()
+
+
+def test_history_batch_numbers_are_compared_after_trim(monkeypatch):
+    db = _sqlite_db()
+    try:
+        product = Product(
+            attendance_point_name="总厂",
+            project_no="P1",
+            product_name="产品A",
+            device_no="",
+            process_name="1",
+            product_gross_weight_kg=Decimal("2"),
+            standard_beat=1,
+        )
+        previous_report = _report(1, ReportStatus.approved)
+        previous_item = ProductionReportItem(
+            id=1,
+            report=previous_report,
+            attendance_point_name="总厂",
+            device_no="A",
+            project_no="P1",
+            product_name="产品A",
+            raw_material_batch_no=" LOT-1 ",
+            process_name="1",
+            good_qty=30,
+            defect_qty=0,
+            scrap_qty=0,
+            standard_beat=1,
+        )
+        report = _report(2)
+        item = ProductionReportItem(
+            id=2,
+            report=report,
+            attendance_point_name="总厂",
+            device_no="B",
+            project_no="P1",
+            product_name="产品A",
+            raw_material_batch_no="LOT-1",
+            process_name="1",
+            good_qty=30,
+            defect_qty=0,
+            scrap_qty=0,
+            standard_beat=1,
+        )
+        db.add_all([product, previous_report, previous_item, report, item])
+        db.flush()
+        monkeypatch.setattr(
+            "app.services.report_over_limit.read_batch_issued_weight_kg",
+            lambda db, batch_no, project_no, product_name: Decimal("100"),
+        )
+
+        result = evaluate_report_item_over_limit(db, report, item)
+
+        assert result.current_cumulative_qty == Decimal("60.00")
+        assert result.status == OVER_LIMIT_STATUS_EXCEEDED
+    finally:
+        db.close()
+
+
+def test_erp_weight_lookup_sql_error_has_distinct_uncheckable_reason():
+    db = _sqlite_db()
+    try:
+        product = Product(
+            attendance_point_name="总厂",
+            project_no="P1",
+            product_name="产品A",
+            device_no="",
+            process_name="1",
+            product_gross_weight_kg=Decimal("2"),
+            standard_beat=1,
+        )
+        report = _report(1)
+        item = ProductionReportItem(
+            id=1,
+            report=report,
+            attendance_point_name="总厂",
+            device_no="A",
+            project_no="P1",
+            product_name="产品A",
+            raw_material_batch_no="LOT-1",
+            process_name="1",
+            good_qty=30,
+            defect_qty=0,
+            scrap_qty=0,
+            standard_beat=1,
+        )
+        db.add_all([product, report, item])
+        db.flush()
+
+        result = evaluate_report_item_over_limit(db, report, item)
+
+        assert result.over_limit_type == OVER_LIMIT_TYPE_UNCHECKABLE
+        assert result.reason == "读取批次投入重量失败，无法校验超额"
+    finally:
+        db.close()
+
+
+def test_evaluate_report_item_over_limit_does_not_write_snapshot_fields(monkeypatch):
+    db = _sqlite_db()
+    try:
+        product = Product(
+            attendance_point_name="总厂",
+            project_no="P1",
+            product_name="产品A",
+            device_no="",
+            process_name="1",
+            product_gross_weight_kg=Decimal("2"),
+            standard_beat=1,
+        )
+        report = _report(1)
+        item = ProductionReportItem(
+            id=1,
+            report=report,
+            attendance_point_name="总厂",
+            device_no="A",
+            project_no="P1",
+            product_name="产品A",
+            raw_material_batch_no="LOT-1",
+            process_name="1",
+            good_qty=60,
+            defect_qty=0,
+            scrap_qty=0,
+            standard_beat=1,
+        )
+        db.add_all([product, report, item])
+        db.flush()
+        monkeypatch.setattr(
+            "app.services.report_over_limit.read_batch_issued_weight_kg",
+            lambda db, batch_no, project_no, product_name: Decimal("100"),
+        )
+
+        result = evaluate_report_item_over_limit(db, report, item)
+
+        assert result.status == OVER_LIMIT_STATUS_EXCEEDED
+        assert item.over_limit_status == OVER_LIMIT_STATUS_NOT_CHECKED
+        assert item.over_limit_type is None
+        assert item.over_limit_reason is None
+    finally:
+        db.close()
+
+
+def test_later_process_exceeds_previous_good_quantity():
+    db = _sqlite_db()
+    try:
+        db.add_all(
+            [
+                Product(
+                    attendance_point_name="总厂",
+                    project_no="P1",
+                    product_name="产品A",
+                    device_no="",
+                    process_name="1",
+                    standard_beat=1,
+                ),
+                Product(
+                    attendance_point_name="总厂",
+                    project_no="P1",
+                    product_name="产品A",
+                    device_no="",
+                    process_name="3",
+                    standard_beat=1,
+                ),
+            ]
+        )
+        previous_report = _report(1, ReportStatus.approved)
+        previous_item = ProductionReportItem(
+            id=1,
+            report=previous_report,
+            attendance_point_name="总厂",
+            device_no="A",
+            project_no="P1",
+            product_name="产品A",
+            raw_material_batch_no="LOT-1",
+            process_name="1",
+            good_qty=100,
+            defect_qty=10,
+            scrap_qty=5,
+            standard_beat=1,
+        )
+        report = _report(2)
+        report.employee_phone = "13800000001"
+        item = ProductionReportItem(
+            id=2,
+            report=report,
+            attendance_point_name="总厂",
+            device_no="B",
+            project_no="P1",
+            product_name="产品A",
+            raw_material_batch_no="LOT-1",
+            process_name="3",
+            good_qty=98,
+            defect_qty=3,
+            scrap_qty=2,
+            standard_beat=1,
+        )
+        db.add_all([previous_report, previous_item, report, item])
+        db.flush()
+
+        result = evaluate_report_item_over_limit(db, report, item, exclude_report_id=report.id)
+
+        assert result.status == OVER_LIMIT_STATUS_EXCEEDED
+        assert result.previous_good_qty == Decimal("100.00")
+        assert result.current_report_qty == Decimal("103.00")
+        assert result.current_cumulative_qty == Decimal("103.00")
+        assert "3序累计报工数量超过1序累计成品数量" in result.reason
+    finally:
+        db.close()
+
+
+def test_history_process_names_are_compared_after_trim(monkeypatch):
+    db = _sqlite_db()
+    try:
+        product = Product(
+            attendance_point_name="总厂",
+            project_no="P1",
+            product_name="产品A",
+            device_no="",
+            process_name="1",
+            product_gross_weight_kg=Decimal("2"),
+            standard_beat=1,
+        )
+        previous_report = _report(1, ReportStatus.approved)
+        previous_item = ProductionReportItem(
+            id=1,
+            report=previous_report,
+            attendance_point_name="总厂",
+            device_no="A",
+            project_no="P1",
+            product_name="产品A",
+            raw_material_batch_no="LOT-1",
+            process_name=" 01序 ",
+            good_qty=30,
+            defect_qty=0,
+            scrap_qty=0,
+            standard_beat=1,
+        )
+        report = _report(2)
+        item = ProductionReportItem(
+            id=2,
+            report=report,
+            attendance_point_name="总厂",
+            device_no="B",
+            project_no="P1",
+            product_name="产品A",
+            raw_material_batch_no="LOT-1",
+            process_name="1",
+            good_qty=30,
+            defect_qty=0,
+            scrap_qty=0,
+            standard_beat=1,
+        )
+        db.add_all([product, previous_report, previous_item, report, item])
+        db.flush()
+        monkeypatch.setattr(
+            "app.services.report_over_limit.read_batch_issued_weight_kg",
+            lambda db, batch_no, project_no, product_name: Decimal("100"),
+        )
+
+        result = evaluate_report_item_over_limit(db, report, item)
+
+        assert result.current_cumulative_qty == Decimal("60.00")
+        assert result.status == OVER_LIMIT_STATUS_EXCEEDED
+    finally:
+        db.close()