fix(pl): reject unknown plan product ids on upsert with 422.

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

backend/app/api/pl/plan.py

@@ -130,6 +130,7 @@ from app.services.pl_plan_service import (
     assert_contract_selectable_for_plan,
     apply_plan_product_fields,
     build_plan_product,
+    resolve_plan_product_id_for_upsert,
     calc_plan_total_amount,
     clear_draft_transport_for_plan,
     clear_hazard_cert_ready,
@@ -1645,7 +1646,10 @@ def update_plan(
     if payload.products is not None:
         existing_all = (
             db.query(PlPlanProduct)
-            .filter(PlPlanProduct.plan_id == id)
+            .filter(
+                PlPlanProduct.plan_id == id,
+                PlPlanProduct.tenant_no == tenant_no,
+            )
             .all()
         )
         existing_all_by_id = {row.id: row for row in existing_all}
@@ -1685,8 +1689,11 @@ def update_plan(
                     "legal_use_cert_url": legal_use_cert,
                 }
             )
-            row_id = (prod.id or "").strip() or None
-            existing = existing_all_by_id.get(row_id) if row_id else None
+            row_id, existing = resolve_plan_product_id_for_upsert(
+                plan_id=id,
+                payload_id=prod.id,
+                existing_by_id=existing_all_by_id,
+            )
             if existing is not None:
                 apply_plan_product_fields(
                     existing,
@@ -1706,6 +1713,7 @@ def update_plan(
                     order_num=idx,
                     user_id=user.id,
                     dept_id=user.dept_id,
+                    product_row_id=row_id,
                 )
                 db.add(pp)
                 kept_ids.add(pp.id)

backend/app/services/pl_plan_service.py

@@ -198,6 +198,25 @@ def apply_plan_product_fields(
         setattr(row, key, value)
 
 
+def resolve_plan_product_id_for_upsert(
+    *,
+    plan_id: str,
+    payload_id: Optional[str],
+    existing_by_id: dict,
+) -> tuple[str, Optional[object]]:
+    """解析计划产品行 upsert 主键：有合法 id 则保留；无 id 则新建；非法 id 则 422。"""
+    pid = (payload_id or "").strip()
+    if not pid:
+        return generate_uuid(), None
+    row = existing_by_id.get(pid)
+    if row is None or getattr(row, "plan_id", None) != plan_id:
+        raise HTTPException(
+            status_code=422,
+            detail=f"计划产品行 id 无效或不属于本计划：{pid}",
+        )
+    return pid, row
+
+
 def _apply_vehicle_snapshot_to_plan_transport(
     db: Session,
     *,

backend/tests/test_pl_plan_product_upsert.py

@@ -1,7 +1,15 @@
-"""build_plan_product 应支持传入 product_row_id，编辑时保持 pl_plan_product.id 稳定。"""
+"""build_plan_product / upsert 应保持 id 稳定，并拒绝外键无效的产品行 id。"""
+from unittest.mock import MagicMock
+
+import pytest
+from fastapi import HTTPException
+
 from app.schemas.pl.plan import PlanProductCreate
 from app.services import pl_plan_service as svc
-from app.services.pl_plan_service import build_plan_product
+from app.services.pl_plan_service import (
+    build_plan_product,
+    resolve_plan_product_id_for_upsert,
+)
 
 
 def _minimal_prod(**kwargs) -> PlanProductCreate:
@@ -40,3 +48,37 @@ def test_build_plan_product_generates_when_missing(monkeypatch):
         dept_id="d1",
     )
     assert result.id == "gen-1"
+
+
+def test_upsert_plan_product_rejects_foreign_id():
+    with pytest.raises(HTTPException) as ei:
+        resolve_plan_product_id_for_upsert(
+            plan_id="plan-1",
+            payload_id="pp-other-plan",
+            existing_by_id={},
+        )
+    assert ei.value.status_code == 422
+
+
+def test_upsert_plan_product_keeps_payload_id():
+    existing = MagicMock()
+    existing.id = "pp-keep"
+    existing.plan_id = "plan-1"
+    row_id, row = resolve_plan_product_id_for_upsert(
+        plan_id="plan-1",
+        payload_id="pp-keep",
+        existing_by_id={"pp-keep": existing},
+    )
+    assert row_id == "pp-keep"
+    assert row is existing
+
+
+def test_upsert_plan_product_generates_when_missing(monkeypatch):
+    monkeypatch.setattr(svc, "generate_uuid", lambda: "new-pp")
+    row_id, row = resolve_plan_product_id_for_upsert(
+        plan_id="plan-1",
+        payload_id=None,
+        existing_by_id={},
+    )
+    assert row_id == "new-pp"
+    assert row is None