diff --git a/inventory-backend/app/services/scrap_sources.py b/inventory-backend/app/services/scrap_sources.py
new file mode 100644
index 0000000..c9abcc1
--- /dev/null
+++ b/inventory-backend/app/services/scrap_sources.py
@@ -0,0 +1,470 @@
+"""
+报废来源适配层 —— 统一三类报废来源的解析、校验、快照与扣减。
+
+背景
+----
+报废审批流（ScrapApprovalService）原先只认三张库存表，导致「借出未还」与
+「在管不良品」两类来源只能各自走直报接口绕过审批 —— 与系统自陈的
+「报废一律需审批」（SCRAP_ALWAYS_REQUIRES_APPROVAL）规则冲突，构成职责分离
+漏洞：同一个库管可自行宣告实物销毁而无人复核。
+
+本模块把「来源差异」收敛到适配器，审批服务不必再关心来源细节。
+
+执行模式
+--------
+  scan —— 需扫码执行。实物在仓库内、有可扫标识，扫码能防「申请报 A、实际毁 B」。
+  auto —— 按批准量执行。实物不在库，物理上无法扫码。
+
+  ★ 为什么「在管不良品」保留扫码：坏件就在仓库里、有 SKU，扫码是有效的
+    实物在场核验，且成本极低。
+  ★ 为什么「借出未还」免扫码：东西在借用人手上，不可能扫到；且该来源执行
+    只改台账与总库存，**不产生任何可被挪用的可用库存**，风险等级不同量级。
+    残余风险（执行人可能在未实际销毁时点「执行」）由 scrap_execute 权限与
+    audit_logs 留痕兜底。
+
+多租户
+------
+  库存行与借出记录经查询层隔离；在管不良品用**台账自带的 company_name 快照**，
+  刻意不联表 MaterialBase —— 坏件的原库存行可能已被入库模块物理删除。
+"""
+import logging
+
+from app.extensions import db
+
+logger = logging.getLogger(__name__)
+
+# 执行模式
+SCRAP_MODE_SCAN = 'scan'   # 需扫码执行
+SCRAP_MODE_AUTO = 'auto'   # 按批准量执行
+
+
+def _stock_model_map():
+    """复用 inventory_reservation 的库存表映射，避免第四份重复定义。"""
+    from app.services.inventory_reservation import stock_model_map
+    return stock_model_map()
+
+
+def defective_unit_cost(goods):
+    """
+    取坏件单价，用于报废台账的 cost_at_scrap / total_loss（best-effort）。
+
+    取价口径与既有报废模块一致：成品取 sale_price，采购件取 pre_tax_unit_price，
+    半成品无价（返回 0）。
+
+    ★ 原库存行可能已被删除（入库模块会物理删除库存行），故取不到时返回 0。
+      刻意**不**因缺行而中断报废：实物已经销毁，台账必须先记上，
+      成本缺失是可接受的降级，记录丢失不是。
+    """
+    model = _stock_model_map().get(goods.source_table)
+    if model is None or not goods.stock_id:
+        return 0.0
+    row = model.query.get(goods.stock_id)
+    if not row:
+        return 0.0
+    if goods.source_table == 'stock_product':
+        return float(getattr(row, 'sale_price', 0) or 0)
+    if goods.source_table == 'stock_buy':
+        return float(getattr(row, 'pre_tax_unit_price', 0) or 0)
+    return 0.0
+
+
+# =============================================================================
+# 基类
+# =============================================================================
+
+class ScrapSourceAdapter:
+    """
+    报废来源适配器基类。
+
+    子类必须提供 source_table / scrap_mode / label / cap_label 四个类属性，
+    并实现 load / cap / snapshot / deduct。
+    """
+
+    source_table = ''
+    scrap_mode = SCRAP_MODE_SCAN
+    label = ''            # 中文来源名，用于报表与错误文案
+    cap_label = '可报废量'  # 上限的中文说法，用于错误文案
+
+    # --- 提交阶段（不加锁，乐观读）---
+    def load(self, sid):
+        """按主键取来源行；不存在返回 None。"""
+        raise NotImplementedError
+
+    def cap(self, row):
+        """该来源当前的可报废上限。"""
+        raise NotImplementedError
+
+    def submit_guard(self, row, qty):
+        """
+        提交期的额外校验（symmetry guard）。
+
+        ★ 存在的意义：提交与执行的校验必须对称，否则会出现「申请能过、
+          执行必失败」的单据。默认无额外约束，子类按需覆盖。
+        """
+        return None
+
+    def snapshot(self, row, qty, raw):
+        """产出写入 items_json 的明细快照。"""
+        raise NotImplementedError
+
+    # --- 执行阶段（加锁 + 扣减 + 写台账）---
+    def deduct(self, row_id, qty, req, operator_name):
+        """加锁重取 → 二次校验 → 扣减 → 写 TransScrap。不满足即抛 ValueError。"""
+        raise NotImplementedError
+
+    # --- 台账公共字段 ---
+    @staticmethod
+    def _ledger_kwargs(req, operator_name):
+        from app.models.scrap_approval import ScrapApproval
+        return {
+            'reason': req.remark or '',
+            'operator_name': operator_name,
+            'approver_name': ScrapApproval._user_name(req.actual_approver_id),
+            'approval_status': 'executed',
+            'scrap_request_no': req.request_no,
+        }
+
+
+# =============================================================================
+# 一类：库存行（三张库存表）
+# =============================================================================
+
+class StockRowAdapter(ScrapSourceAdapter):
+    """
+    常规库存行来源。语义与改造前的 execute() 逐字一致。
+
+    扣减：available_quantity 与 stock_quantity **同时扣**（报废 = 实物销毁）。
+    扫码：必需 —— 库存行是同质可替换物，在库内处于执行人物理控制之下，
+          存在以次充好、批次腾挪的空间，扫码把「批准批次」与「实毁批次」钉死。
+    """
+
+    scrap_mode = SCRAP_MODE_SCAN
+    cap_label = '可用库存'
+
+    _LABELS = {
+        'stock_buy': '采购件',
+        'stock_semi': '半成品',
+        'stock_product': '成品',
+    }
+
+    def __init__(self, source_table, model):
+        self.source_table = source_table
+        self.model = model
+        self.label = self._LABELS.get(source_table, source_table)
+
+    def load(self, sid):
+        return self.model.query.get(sid)
+
+    def cap(self, row):
+        return float(getattr(row, 'available_quantity', 0) or 0)
+
+    def submit_guard(self, row, qty):
+        # ★ 对称性修复：旧代码提交只校验 available_quantity，而执行同时校验
+        #   available 与 stock_quantity —— 会出现「申请通过、执行必然失败」。
+        #   此处提前拦下。
+        stock = float(getattr(row, 'stock_quantity', 0) or 0)
+        if qty > stock:
+            raise ValueError(f"报废数量({qty})超过实物库存({stock})")
+
+    def snapshot(self, row, qty, raw):
+        base = getattr(row, 'base', None)
+        return {
+            'source_table': self.source_table,
+            'stock_id': row.id,
+            'base_id': getattr(row, 'base_id', None),
+            'sku': getattr(row, 'sku', '') or '',
+            'name': (base.name if base else '') or raw.get('name') or '',
+            'spec_model': (base.spec_model if base else '') or raw.get('spec_model') or '',
+            'location': getattr(row, 'warehouse_location', '') or '',
+            'batch_number': (getattr(row, 'batch_number', '')
+                             or getattr(row, 'serial_number', '') or ''),
+            'scrap_qty': qty,
+            'available_at_apply': self.cap(row),
+            'scrap_mode': self.scrap_mode,
+        }
+
+    def deduct(self, row_id, qty, req, operator_name):
+        from app.models.transaction import TransScrap
+
+        row = self.model.query.with_for_update().get(row_id)
+        if not row:
+            raise ValueError(f"库存记录已不存在（{self.source_table}#{row_id}）")
+
+        avail = float(getattr(row, 'available_quantity', 0) or 0)
+        stock = float(getattr(row, 'stock_quantity', 0) or 0)
+        label = getattr(row, 'sku', '') or f"{self.source_table}#{row_id}"
+        if qty > avail:
+            raise ValueError(f"库存 SKU【{label}】可用不足（剩 {avail}），无法报废 {qty}")
+        if qty > stock:
+            raise ValueError(f"库存 SKU【{label}】实物不足（剩 {stock}），无法报废 {qty}")
+
+        # 报废 = 实物销毁：实物数与可用数同时扣减
+        row.available_quantity = avail - qty
+        row.stock_quantity = stock - qty
+        db.session.flush()
+
+        db.session.add(TransScrap(
+            sku=getattr(row, 'sku', '') or label,
+            source_table=self.source_table,
+            stock_id=row_id,
+            quantity=qty,
+            **self._ledger_kwargs(req, operator_name),
+        ))
+
+
+# =============================================================================
+# 二类：在管不良品台账
+# =============================================================================
+
+class DefectiveScrapAdapter(ScrapSourceAdapter):
+    """
+    在管不良品来源（逆向物流）。
+
+    扣减：remaining_qty -=、scrapped_qty +=、推进状态机；**完全不动任何库存表**
+          —— 坏件从未进入库存表，这是本次逆向物流的核心架构决策。
+    扫码：必需（业务方决策）—— 坏件实物在仓、有 SKU，扫码是有效核验。
+    成本：按 defective_unit_cost 取价，保持改造前直报接口的口径。
+    """
+
+    source_table = 'trans_defective_goods'
+    scrap_mode = SCRAP_MODE_SCAN
+    label = '在管不良品'
+    cap_label = '在管数量'
+
+    def load(self, sid):
+        from app.models.transaction import TransDefectiveGoods
+        return TransDefectiveGoods.query.get(sid)
+
+    def cap(self, row):
+        return float(getattr(row, 'remaining_qty', 0) or 0)
+
+    def snapshot(self, row, qty, raw):
+        # 物料名/规格取自台账自身的冗余快照，**不联表 MaterialBase** ——
+        # 原库存行可能已被物理删除，联表会取到空值。
+        return {
+            'source_table': self.source_table,
+            'stock_id': row.id,
+            'base_id': getattr(row, 'base_id', None),
+            'sku': getattr(row, 'sku', '') or '',
+            'name': getattr(row, 'material_name', '') or raw.get('name') or '',
+            'spec_model': getattr(row, 'spec_model', '') or raw.get('spec_model') or '',
+            'location': '',      # 台账无库位字段
+            'batch_number': '',
+            'scrap_qty': qty,
+            'available_at_apply': self.cap(row),
+            'scrap_mode': self.scrap_mode,
+        }
+
+    def deduct(self, row_id, qty, req, operator_name):
+        from app.models.transaction import (
+            TransScrap, TransDefectiveGoods, OPEN_DEFECTIVE_STATUSES,
+            DEFECTIVE_STATUS_IN_PROGRESS, defective_close_status,
+        )
+
+        goods = TransDefectiveGoods.query.with_for_update().get(row_id)
+        if not goods:
+            raise ValueError(f"不良品在管记录已不存在（#{row_id}）")
+
+        # 状态守门（Fail-Closed）：终态不得再处置
+        if goods.status not in OPEN_DEFECTIVE_STATUSES:
+            raise ValueError(
+                f"在管不良品【{goods.sku or row_id}】当前状态为「{goods.status}」，不可报废"
+            )
+
+        remaining = float(goods.remaining_qty or 0)
+        if qty > remaining:
+            # ★ Fail-Closed：批准后可能被另一张单先报废、或已部分回库，
+            #   此时静默夹到剩余量会让单据状态（已执行）与实际扣减不符，
+            #   审计上不可接受。整单报错，由申请人撤回后重提。
+            raise ValueError(
+                f"在管不良品【{goods.sku or row_id}】在管量不足"
+                f"（剩 {remaining}），无法报废 {qty}"
+            )
+
+        new_remaining = remaining - qty
+        goods.remaining_qty = new_remaining
+        goods.scrapped_qty = float(goods.scrapped_qty or 0) + qty
+        # 终态由「累计去向」推导而非「最后一次动作」—— 一批坏件可能既回库过
+        # 又报废过，按最后动作定状态会产生误导（见 defective_close_status）
+        goods.status = (
+            defective_close_status(goods.restocked_qty, goods.scrapped_qty)
+            if new_remaining <= 0 else DEFECTIVE_STATUS_IN_PROGRESS
+        )
+        db.session.flush()
+
+        unit_cost = defective_unit_cost(goods)
+        db.session.add(TransScrap(
+            sku=goods.sku or '',
+            source_table=self.source_table,
+            stock_id=goods.id,
+            quantity=qty,
+            cost_at_scrap=unit_cost,
+            total_loss=round(unit_cost * qty, 2),
+            **self._ledger_kwargs(req, operator_name),
+        ))
+
+
+# =============================================================================
+# 三类：借出未还（借库转报废）
+# =============================================================================
+
+class BorrowScrapAdapter(ScrapSourceAdapter):
+    """
+    借出未还来源（借库转报废）。
+
+    扣减：标记借用记录为已报废；内层库存行**只扣 stock_quantity**，不动
+          available_quantity —— 可用量已在借出时冻结（见
+          trans_service.execute_dispatch 的 deduct_stock=False 及其注释）。
+    扫码：免扫码（业务方决策，理由见模块头）。
+    成本：0/0 —— 与改造前 scrap_borrow 口径一致。
+    """
+
+    source_table = 'trans_borrow'
+    scrap_mode = SCRAP_MODE_AUTO
+    label = '借出未还'
+    cap_label = '待还数量'
+
+    def load(self, sid):
+        from app.models.transaction import TransBorrow
+        return TransBorrow.query.get(sid)
+
+    def cap(self, row):
+        return (float(getattr(row, 'quantity', 0) or 0)
+                - float(getattr(row, 'returned_quantity', 0) or 0))
+
+    def submit_guard(self, row, qty):
+        # 已归还/已报废的借用记录不得再报废。旧实现是静默 continue 并返回
+        # count=0（缺陷：用户以为成功），这里改为明确报错。
+        if getattr(row, 'is_returned', False):
+            raise ValueError(
+                f"借用记录【{getattr(row, 'borrow_no', '')}】已归还或已报废，不可再报废"
+            )
+
+    def snapshot(self, row, qty, raw):
+        name, spec = self._resolve_material(row)
+        return {
+            'source_table': self.source_table,
+            'stock_id': row.id,          # ★ 存 TransBorrow.id，与既有约定一致
+            'base_id': None,
+            'sku': getattr(row, 'sku', '') or '',
+            'name': name or raw.get('name') or '',
+            'spec_model': spec or raw.get('spec_model') or '',
+            'location': getattr(row, 'location', '') or '',
+            'batch_number': getattr(row, 'barcode', '') or '',
+            'scrap_qty': qty,
+            'available_at_apply': self.cap(row),
+            'scrap_mode': self.scrap_mode,
+        }
+
+    @staticmethod
+    def _resolve_material(record):
+        """经借用记录回查其源库存行取物料名/规格；行已删除时返回空串。"""
+        model = _stock_model_map().get(getattr(record, 'source_table', ''))
+        if model is None or not getattr(record, 'stock_id', None):
+            return '', ''
+        row = model.query.get(record.stock_id)
+        base = getattr(row, 'base', None) if row else None
+        return ((base.name if base else '') or '',
+                (base.spec_model if base else '') or '')
+
+    def deduct(self, row_id, qty, req, operator_name):
+        from datetime import datetime
+        from app.models.transaction import TransScrap, TransBorrow
+
+        record = TransBorrow.query.with_for_update().get(row_id)
+        if not record:
+            raise ValueError(f"借用记录已不存在（#{row_id}）")
+        if record.is_returned:
+            raise ValueError(
+                f"借用记录【{record.borrow_no}】已归还或已报废，不可再报废"
+            )
+
+        pending = (float(record.quantity or 0)
+                   - float(record.returned_quantity or 0))
+        if qty > pending:
+            # Fail-Closed：批准后可能已被部分归还，同不良品来源的理由
+            raise ValueError(
+                f"借用记录【{record.borrow_no}】待还量不足（剩 {pending}），无法报废 {qty}"
+            )
+
+        # 1) 标记借用记录：不再追讨归还
+        record.is_returned = True
+        record.status = 'scrapped'
+        record.return_time = datetime.now()
+        record.return_operator = operator_name
+
+        # 2) 扣总库存（该物品确认损失）；可用量已在借出时冻结，不重复扣
+        model = _stock_model_map().get(record.source_table)
+        if model is not None and record.stock_id:
+            stock = model.query.with_for_update().get(record.stock_id)
+            if stock:
+                stock_qty = float(stock.stock_quantity or 0)
+                if qty > stock_qty:
+                    raise ValueError(
+                        f"SKU {record.sku} 实物库存不足（剩 {stock_qty}），无法报废 {qty}"
+                    )
+                stock.stock_quantity = stock_qty - qty
+            else:
+                # 源库存行已被物理删除：损失无法落在库存账上。
+                # 刻意不中断 —— 实物已确认无法归还，台账必须先记上；
+                # 但显式告警，避免这种「账上无痕」的损失悄无声息。
+                logger.warning(
+                    "[报废] 借出记录 #%s 的源库存行已不存在（%s#%s），"
+                    "本次报废未在库存账上体现",
+                    row_id, record.source_table, record.stock_id,
+                )
+        db.session.flush()
+
+        # 3) 写报废台账。成本口径保持 0/0（与改造前 scrap_borrow 一致）
+        db.session.add(TransScrap(
+            sku=record.sku or '',
+            source_table=self.source_table,
+            stock_id=record.id,
+            quantity=qty,
+            cost_at_scrap=0,
+            total_loss=0,
+            **self._ledger_kwargs(req, operator_name),
+        ))
+
+
+# =============================================================================
+# 注册表
+# =============================================================================
+
+def _build_registry():
+    models = _stock_model_map()
+    registry = {
+        st: StockRowAdapter(st, model) for st, model in models.items()
+    }
+    registry[DefectiveScrapAdapter.source_table] = DefectiveScrapAdapter()
+    registry[BorrowScrapAdapter.source_table] = BorrowScrapAdapter()
+    return registry
+
+
+_REGISTRY = None
+
+
+def get_adapter(source_table):
+    """按来源表名取适配器；不支持则返回 None。"""
+    global _REGISTRY
+    if _REGISTRY is None:
+        _REGISTRY = _build_registry()
+    return _REGISTRY.get((source_table or '').strip())
+
+
+def all_source_tables():
+    global _REGISTRY
+    if _REGISTRY is None:
+        _REGISTRY = _build_registry()
+    return tuple(_REGISTRY.keys())
+
+
+def is_scan_source(source_table):
+    """
+    该来源是否走扫码执行。
+
+    ★ 未知来源一律返回 False —— execute 的扫码索引只接纳明确声明的 scan 来源，
+      避免历史上 trans_repair 那类「扫码能扫到、执行却拒绝」的错配重演。
+    """
+    adapter = get_adapter(source_table)
+    return bool(adapter and adapter.scrap_mode == SCRAP_MODE_SCAN)