docs: add plan-arrange upsert cascade implementation plan.

Break phase-1 dibang field sync and monthly-plan remount into TDD tasks; defer stop stable ids to the 2026-09-05 plan.

Co-authored-by: Cursor <cursoragent@cursor.com>

docs/superpowers/plans/2026-09-06-plan-arrange-upsert-cascade.md

@@ -0,0 +1,603 @@
+# 运输计划调整 Upsert 级联 Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** 已安排运输计划调整时，地磅出入库/过磅单业务字段与月计划按「有 id 更新、无 id 创建」原地级联；出入库 id 不变；称重数字不动。
+
+**Architecture:** 在 `pl_dibang_bridge_service._update_existing_stop_inventory` 去掉「已过磅则跳过正文」；统一回写出入库正文，并同步关联过磅单正文；产品/往来变更时对月计划 find-or-create 改挂。段稳定 ID（二期）复用既有计划 `2026-09-05-pl-stop-stable-id-reuse-links.md`，本计划不重复造轮子。
+
+**Tech Stack:** FastAPI · SQLAlchemy · Pydantic v2 · pytest
+
+**Spec:** `docs/superpowers/specs/2026-09-06-plan-arrange-upsert-cascade-design.md`
+
+**注意：** 工作区可能有无关未提交改动。只改本计划列出的文件，提交时勿夹带无关 diff。生产部署后必须重启 backend，否则仍跑旧 revoke 逻辑。
+
+---
+
+## 文件结构
+
+| 操作 | 文件 | 职责 |
+|------|------|------|
+| Modify | `backend/app/services/pl_dibang_bridge_service.py` | 强制正文同步、过磅单正文、月计划改挂 |
+| Modify | `backend/tests/test_pl_dibang_upsert_on_arrange.py` | 已过磅改车牌/产品、月计划改挂用例 |
+| Verify | `backend/app/api/pl/transport.py` | 确认安排路径无 `revoke_regeneratable` |
+| Follow | `docs/superpowers/plans/2026-09-05-pl-stop-stable-id-reuse-links.md` | 二期：段/产品稳定 id（若未完成则续做） |
+
+---
+
+### Task 1: 失败测试 — 已过磅时仍同步出入库车牌与过磅单正文
+
+**Files:**
+- Modify: `backend/tests/test_pl_dibang_upsert_on_arrange.py`
+- Test: 同文件
+
+- [ ] **Step 1: 追加失败测试**
+
+在 `backend/tests/test_pl_dibang_upsert_on_arrange.py` 文件末尾追加：
+
+```python
+from app.models.db_models import DbWeighingSlip
+
+
+@patch("app.services.pl_dibang_bridge_service._inventory_has_weighing_slip", return_value=True)
+@patch("app.services.pl_dibang_bridge_service.resolve_crm_product")
+def test_upsert_updates_inventory_and_weighing_plate_when_already_weighed(mock_crm, _mock_weighing):
+    """已过磅：仍强制回写出入库与过磅单车牌；称重不动；出入库 id 不变。"""
+    mock_crm.return_value = MagicMock(
+        product_id="prod-1",
+        product_name="硫酸",
+        specification="98%",
+    )
+    plan = PlPlan(
+        id="plan-1",
+        plan_no="PO1",
+        plan_type="outbound",
+        plan_date=date(2026, 9, 1),
+        pickup_time=datetime(2026, 9, 1, 0, 0),
+        license_plate="皖HA7429",
+        tenant_no="t1",
+    )
+    transport = PlTransport(
+        id="tr-1", tenant_no="t1", license_plate="皖HA7429", del_flag=0
+    )
+    load = PlPlanLoad(
+        id="load-1",
+        plan_id="plan-1",
+        customer_name="山峡",
+        generate_dibang=1,
+        dibang_direction="outbound",
+        load_time=datetime(2026, 9, 1, 0, 0),
+        tenant_no="t1",
+        del_flag=0,
+    )
+    delivery = PlPlanDelivery(
+        id="del-1",
+        plan_id="plan-1",
+        customer_name="江峰",
+        seq=1,
+        generate_dibang=0,
+        tenant_no="t1",
+        del_flag=0,
+    )
+    stop_product = PlPlanLoadProduct(
+        id="lp-1",
+        load_id="load-1",
+        plan_product_id="pp-1",
+        product_id="prod-1",
+        product_name="硫酸",
+        model="98%",
+        unit="吨",
+        plan_qty=Decimal("22"),
+        tenant_no="t1",
+        del_flag=0,
+        order_num=1,
+    )
+    inventory = DbInboundOutbound(
+        id="2861",
+        tenant_no="t1",
+        date=date(2026, 9, 6),
+        partner_company="江峰",
+        business_type="出库",
+        product_id="prod-1",
+        product="硫酸",
+        specification="98%",
+        quantity=22.0,
+        batch_no="CRKDH0002488",
+        vehicle_no="皖H28926",
+        status="待出库",
+        purchase_plan_id="dbp-1",
+        plan_no="SX-1",
+        del_flag=0,
+    )
+    slip = DbWeighingSlip(
+        id="ws-1",
+        tenant_no="t1",
+        vehicle_no="皖H28926",
+        plan_no="SX-1",
+        cargo_name="硫酸",
+        specification="98%",
+        business_type="出库",
+        batch_no="CRKDH0002488",
+        supplier="",
+        consignee="江峰",
+        gross_weight=0.0,
+        tare_weight=18730.0,
+        net_weight=-18730.0,
+        inbound_outbound_id="2861",
+        purchase_plan_id="dbp-1",
+        is_voided=0,
+        del_flag=0,
+    )
+    link = PlTransportDbLink(
+        id="link-1",
+        tenant_no="t1",
+        plan_id="plan-1",
+        transport_id="tr-1",
+        plan_product_id="pp-1",
+        stop_type="load",
+        stop_id="load-1",
+        db_purchase_plan_id="dbp-1",
+        db_inventory_id="2861",
+        contributed_qty=Decimal("22"),
+        del_flag=0,
+    )
+    db = MagicMock()
+
+    def link_find_q(m):
+        filtered = MagicMock()
+        filtered.first.return_value = link
+        m.filter.return_value = filtered
+        return m
+
+    def load_q(m):
+        m.filter.return_value.first.return_value = load
+        return m
+
+    def delivery_q(m):
+        m.filter.return_value.order_by.return_value.all.return_value = [delivery]
+        return m
+
+    def load_prod_q(m):
+        m.filter.return_value.order_by.return_value.all.return_value = [stop_product]
+        return m
+
+    def delivery_prod_q(m):
+        m.filter.return_value.order_by.return_value.all.return_value = []
+        return m
+
+    def inv_q(m):
+        m.filter.return_value.first.return_value = inventory
+        return m
+
+    def slip_q(m):
+        filtered = MagicMock()
+        filtered.all.return_value = [slip]
+        m.filter.return_value = filtered
+        return m
+
+    def db_plan_q(m):
+        filtered = MagicMock()
+        filtered.first.return_value = None
+        m.filter.return_value = filtered
+        return m
+
+    db.query.side_effect = _make_query_router(
+        {
+            PlTransportDbLink: link_find_q,
+            PlPlanLoad: load_q,
+            PlPlanDelivery: delivery_q,
+            PlPlanLoadProduct: load_prod_q,
+            PlPlanDeliveryProduct: delivery_prod_q,
+            DbInboundOutbound: inv_q,
+            DbWeighingSlip: slip_q,
+            DbPurchasePlan: db_plan_q,
+        }
+    )
+
+    results, warnings = upsert_stop_dibang_links(
+        db,
+        tenant_no="t1",
+        plan=plan,
+        transport=transport,
+        stop_options=[
+            StopDibangOption(stop_type="load", stop_id="load-1", generate_dibang=True),
+            StopDibangOption(stop_type="delivery", stop_id="del-1", generate_dibang=False),
+        ],
+        user_id="u1",
+        dept_id="d1",
+    )
+
+    assert results and results[0].db_inventory_id == "2861"
+    assert inventory.vehicle_no == "皖HA7429"
+    assert slip.vehicle_no == "皖HA7429"
+    assert slip.tare_weight == 18730.0
+    assert slip.net_weight == -18730.0
+    assert not any("未同步" in w for w in warnings)
+```
+
+- [ ] **Step 2: 运行测试确认失败**
+
+Run:
+
+```bash
+cd backend && source venv/bin/activate && pytest tests/test_pl_dibang_upsert_on_arrange.py::test_upsert_updates_inventory_and_weighing_plate_when_already_weighed -v
+```
+
+Expected: FAIL（当前已过磅会走「未同步…」分支，车牌仍为 `皖H28926`，或断言失败）
+
+- [ ] **Step 3: Commit 测试**
+
+```bash
+git add backend/tests/test_pl_dibang_upsert_on_arrange.py
+git commit -m "test(pl): expect dibang plate sync even after weighing"
+```
+
+---
+
+### Task 2: 实现 — 已过磅仍强制同步出入库正文 + 过磅单正文
+
+**Files:**
+- Modify: `backend/app/services/pl_dibang_bridge_service.py`
+
+- [ ] **Step 1: 新增过磅单正文同步辅助函数**
+
+在 `pl_dibang_bridge_service.py` 中 `_inventory_has_weighing_slip` 附近增加：
+
+```python
+def _sync_weighing_slip_business_fields(
+    db: Session,
+    *,
+    inventory: DbInboundOutbound,
+    user_id: Optional[str],
+) -> None:
+    """按出入库正文回写关联过磅单业务字段；称重数字一律不动。"""
+    slips = (
+        db.query(DbWeighingSlip)
+        .filter(
+            DbWeighingSlip.inbound_outbound_id == inventory.id,
+            DbWeighingSlip.del_flag == 0,
+            DbWeighingSlip.is_voided == 0,
+        )
+        .all()
+    )
+    if not slips:
+        return
+
+    partner = (inventory.partner_company or "").strip()
+    business_type = (inventory.business_type or "").strip()
+    for slip in slips:
+        plate = (inventory.vehicle_no or "").strip()
+        if plate:
+            slip.vehicle_no = plate
+        if inventory.product_id:
+            slip.product_id = inventory.product_id
+        if inventory.product:
+            slip.cargo_name = inventory.product
+        slip.specification = inventory.specification
+        if business_type:
+            slip.business_type = business_type
+        if inventory.plan_no:
+            slip.plan_no = inventory.plan_no
+        if inventory.batch_no:
+            slip.batch_no = inventory.batch_no
+        if inventory.purchase_plan_id:
+            slip.purchase_plan_id = inventory.purchase_plan_id
+        if partner:
+            if business_type == "入库":
+                slip.supplier = partner
+            else:
+                slip.consignee = partner
+        slip.update_by = user_id
+```
+
+确保文件顶部已 import `DbWeighingSlip`（若已有则不动）。
+
+- [ ] **Step 2: 改写 `_update_existing_stop_inventory` — 去掉「不可编辑则跳过正文」**
+
+删除（或绕过）这段逻辑：
+
+```python
+editable = _inventory_mutable_editable(db, inventory)
+if not editable:
+    warnings.append(...未同步...)
+    return DibangInventoryLinkResult(...)
+```
+
+改为：**始终**执行正文更新（车牌/往来/数量/日期），然后调用 `_sync_weighing_slip_business_fields`。
+
+最小结构：
+
+```python
+def _update_existing_stop_inventory(...):
+    label = inventory.batch_no or inventory.id
+
+    # 1) 往来
+    if partner and (inventory.partner_company or "").strip() != partner:
+        inventory.partner_company = partner
+
+    # 2) 车牌
+    plate = (license_plate or "").strip()
+    if plate and (inventory.vehicle_no or "").strip() != plate:
+        inventory.vehicle_no = plate
+
+    # 3) 产品快照（允许换产品；月计划改挂见 Task 3）
+    #    先用 resolve_crm_product + apply_product_snapshot_to_inventory
+    #    若 product.product_id 有值且与 inventory 不同，先记录 old 用于 Task 3 改挂
+
+    # 4) 数量 / 日期：保留现有 delta 与跨月改挂逻辑
+
+    desired_remark = f"运输计划 {plan.plan_no}"
+    inventory.remark = desired_remark
+    link.remark = desired_remark
+    inventory.update_by = user_id
+    link.update_by = user_id
+
+    _sync_weighing_slip_business_fields(db, inventory=inventory, user_id=user_id)
+
+    return DibangInventoryLinkResult(...)
+```
+
+**注意：** 删除「产品变更直接 `raise ValueError`」硬阻断（与 D6 冲突）。产品变更改挂放到 Task 3。
+
+- [ ] **Step 3: 运行 Task 1 测试**
+
+```bash
+cd backend && source venv/bin/activate && pytest tests/test_pl_dibang_upsert_on_arrange.py::test_upsert_updates_inventory_and_weighing_plate_when_already_weighed -v
+```
+
+Expected: PASS
+
+- [ ] **Step 4: 跑既有 upsert 套件**
+
+```bash
+cd backend && source venv/bin/activate && pytest tests/test_pl_dibang_upsert_on_arrange.py -v
+```
+
+Expected: 全部 PASS（含「取消勾选已过磅仍阻断」）
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add backend/app/services/pl_dibang_bridge_service.py backend/tests/test_pl_dibang_upsert_on_arrange.py
+git commit -m "fix(pl): sync dibang inventory and weighing fields after rearrange"
+```
+
+---
+
+### Task 3: 失败测试 + 实现 — 产品/往来变更时月计划利旧或新建并改挂
+
+**Files:**
+- Modify: `backend/app/services/pl_dibang_bridge_service.py`
+- Modify: `backend/tests/test_pl_dibang_upsert_on_arrange.py`
+
+- [ ] **Step 1: 写失败测试**
+
+```python
+@patch("app.services.pl_dibang_bridge_service._inventory_has_weighing_slip", return_value=False)
+@patch("app.services.pl_dibang_bridge_service._create_db_plan")
+@patch("app.services.pl_dibang_bridge_service.find_monthly_db_plan")
+@patch("app.services.pl_dibang_bridge_service.resolve_crm_product")
+def test_upsert_remounts_monthly_plan_when_partner_changes(
+    mock_crm, mock_find_plan, mock_create_plan, _mock_weighing
+):
+    """往来变更：出入库 id 不变；purchase_plan_id 改挂到利旧/新建月计划。"""
+    mock_crm.return_value = MagicMock(
+        product_id="prod-1", product_name="硫酸", specification="98%"
+    )
+    old_plan = DbPurchasePlan(
+        id="dbp-old",
+        tenant_no="t1",
+        plan_no="SX-OLD",
+        planned_quantity=22.0,
+        customer_name="旧往来",
+        product_id="prod-1",
+        del_flag=0,
+    )
+    new_plan = DbPurchasePlan(
+        id="dbp-new",
+        tenant_no="t1",
+        plan_no="SX-NEW",
+        planned_quantity=0.0,
+        customer_name="新往来",
+        product_id="prod-1",
+        del_flag=0,
+    )
+    mock_find_plan.return_value = new_plan
+
+    # ... 构造 plan/transport/load/delivery/stop_product/inventory/link
+    # inventory.partner_company = "旧往来"
+    # inventory.purchase_plan_id = "dbp-old"
+    # partner 解析结果为「新往来」（装货段用送货客户等，按现有 resolve_stop_dibang_partner）
+    # upsert_stop_dibang_links(... generate_dibang=True ...)
+
+    assert inventory.id == "inv-1"
+    assert inventory.partner_company == "新往来"
+    assert inventory.purchase_plan_id == "dbp-new"
+    assert link.db_purchase_plan_id == "dbp-new"
+```
+
+（实现时按文件内既有 mock 风格补全 router；`mock_create_plan` 不应被调用当 `find` 命中。）
+
+- [ ] **Step 2: 跑测试确认失败**
+
+```bash
+cd backend && source venv/bin/activate && pytest tests/test_pl_dibang_upsert_on_arrange.py::test_upsert_remounts_monthly_plan_when_partner_changes -v
+```
+
+Expected: FAIL
+
+- [ ] **Step 3: 在 `_update_existing_stop_inventory` 中实现同月产品/往来改挂**
+
+在更新 `partner_company` / 产品快照之后：
+
+```python
+need_remount = False
+# partner 已变 或 product_id/specification 已变 → need_remount = True
+
+if need_remount:
+    qty = _resolve_link_contributed_qty(link, inventory)
+    _subtract_plan_contribution(
+        db, tenant_no=tenant_no, link=link, inventory=inventory, user_id=user_id
+    )
+    new_db_plan = _ensure_monthly_db_plan_for_inventory(
+        db,
+        tenant_no=tenant_no,
+        plan=plan,
+        inventory=inventory,
+        dibang_date=_normalize_to_date(inventory.date) or dibang_date,
+        contributed_qty=qty,
+        user_id=user_id,
+        dept_id=dept_id or "",
+    )
+    link.db_purchase_plan_id = new_db_plan.id
+    inventory.purchase_plan_id = new_db_plan.id
+    inventory.plan_no = (new_db_plan.plan_no or "").strip() or inventory.plan_no
+elif plate:
+    # 仅车牌：回写当前月计划 vehicle_no（若存在）
+    if link.db_purchase_plan_id:
+        db_plan = (
+            db.query(DbPurchasePlan)
+            .filter(
+                DbPurchasePlan.id == link.db_purchase_plan_id,
+                DbPurchasePlan.tenant_no == tenant_no,
+                DbPurchasePlan.del_flag == 0,
+            )
+            .first()
+        )
+        if db_plan and (db_plan.vehicle_no or "").strip() != plate:
+            db_plan.vehicle_no = plate
+            db_plan.update_by = user_id
+```
+
+然后再次 `_sync_weighing_slip_business_fields`（确保过磅单 `purchase_plan_id`/`plan_no` 跟上）。
+
+复用已有 `_ensure_monthly_db_plan_for_inventory` / `find_monthly_db_plan`（内部已是利旧或 `_create_db_plan`）。
+
+- [ ] **Step 4: 跑测试**
+
+```bash
+cd backend && source venv/bin/activate && pytest tests/test_pl_dibang_upsert_on_arrange.py -v
+```
+
+Expected: PASS
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add backend/app/services/pl_dibang_bridge_service.py backend/tests/test_pl_dibang_upsert_on_arrange.py
+git commit -m "feat(pl): remount dibang monthly plan when partner or product changes"
+```
+
+---
+
+### Task 4: 校验安排入口无整批 revoke + 冒烟
+
+**Files:**
+- Verify: `backend/app/api/pl/transport.py`
+- Test: `backend/tests/test_pl_dibang_upsert_on_arrange.py`
+
+- [ ] **Step 1: 确认代码**
+
+在 `transport.py` 的 `arrange_transport` 中确认：
+
+- 存在注释「地磅关联改为按段 upsert，不再在调整前整批 revoke 再建」
+- **没有**对 `revoke_regeneratable_dibang_for_transport` 的调用
+
+若仍有 revoke 调用：删除该分支，改为仅 `sync_dibang_for_plan_stops` / `upsert_stop_dibang_links`。
+
+- [ ] **Step 2: 增加回归断言测试（可选轻量）**
+
+在测试文件增加：
+
+```python
+def test_arrange_transport_source_has_no_revoke_call():
+    import inspect
+    from app.api.pl import transport as transport_api
+
+    src = inspect.getsource(transport_api.arrange_transport)
+    assert "revoke_regeneratable_dibang_for_transport" not in src
+```
+
+- [ ] **Step 3: 跑相关测试**
+
+```bash
+cd backend && source venv/bin/activate && pytest tests/test_pl_dibang_upsert_on_arrange.py tests/test_pl_dibang_date_sync_on_plan_save.py -v
+```
+
+Expected: PASS
+
+- [ ] **Step 4: Commit（若有改动）**
+
+```bash
+git add backend/app/api/pl/transport.py backend/tests/test_pl_dibang_upsert_on_arrange.py
+git commit -m "chore(pl): guard arrange path against dibang revoke rebuild"
+```
+
+---
+
+### Task 5: 二期衔接 — 段/产品稳定 ID
+
+**Files:**
+- Follow: `docs/superpowers/plans/2026-09-05-pl-stop-stable-id-reuse-links.md`
+
+- [ ] **Step 1: 检查 09-05 计划未完成任务**
+
+对照仓库：`replace_plan_stops` 是否仍对送货段/段产品「软删 + 新 UUID」。若是，按 `2026-09-05-pl-stop-stable-id-reuse-links.md` **剩余 Task** 继续执行（本计划不复制全文）。
+
+- [ ] **Step 2: 验收交叉点**
+
+完成二期后，手工/单测确认：保存计划后 `plan_product_id` 不变 → `pl_transport_db_link` 仍能命中 → Task 2/3 的正文同步生效。
+
+- [ ] **Step 3: 若二期已全部完成**
+
+在本计划文件头部标注「二期：已由 2026-09-05 计划覆盖，跳过」。
+
+---
+
+### Task 6: 部署检查清单（人工）
+
+- [ ] **Step 1: 合并/发布后重启生产**
+
+```bash
+cd /iflytek/shanxia2.0/backend/
+PORT=9111 WORKERS=1 ENV_FILE=./.env.production ./start-prod.sh restart
+```
+
+- [ ] **Step 2: 用类似 PO20260903001 场景回归**
+
+1. 已安排 + 已有出入库（可先未过磅）改车 → 出入库车牌变、id 不变  
+2. 已过磅改车 → 出入库+过磅单车牌变、称重不变  
+3. 新勾选地磅 → 新建出入库  
+4. 改往来/产品 → 月计划改挂  
+5. 取消已过磅地磅勾选 → 仍阻断  
+
+- [ ] **Step 3: 确认 API 不再返回**
+
+`无法随运输安排自动重建`
+
+---
+
+## Spec 覆盖自检
+
+| Spec 项 | Task |
+|---------|------|
+| D2/D3 有 id 更新、出入库 id 稳定 | Task 2 |
+| D4 过磅改正文、称重不动 | Task 1–2 |
+| D5 新地磅需求新建出入库 | 既有 upsert 新建分支；Task 4 冒烟 |
+| D6 月计划利旧/新建改挂 | Task 3 |
+| D7 警告非整单 400；废除安排 revoke | Task 2（不再因已过磅跳过）、Task 4 |
+| 二期段稳定 ID | Task 5 → 09-05 计划 |
+| 三期审计摘要 | 本期不做（YAGNI，spec 三期） |
+
+---
+
+## 执行交接
+
+Plan complete and saved to `docs/superpowers/plans/2026-09-06-plan-arrange-upsert-cascade.md`.
+
+**两种执行方式：**
+
+1. **Subagent-Driven（推荐）** — 每个 Task 开新子代理，Task 间复查  
+2. **Inline Execution** — 本会话按 executing-plans 连续做，设检查点  
+
+选哪种？