diff --git a/backend/app/crud/production_work_order.py b/backend/app/crud/production_work_order.py
index e582f86a..f7774807 100644
--- a/backend/app/crud/production_work_order.py
+++ b/backend/app/crud/production_work_order.py
@@ -924,12 +924,51 @@ def complete_production_work_order(
         raise ValueError("工单不存在")
     if wo.work_order_status == "ywc":
         raise ValueError("工单已完成")
+    if not data:
+        raise ValueError("须选择生产结论（成功或失败）")
+    concl = (getattr(data, "conclusion", None) or "").strip()
+    if concl not in ("cg", "sb"):
+        raise ValueError("须选择生产结论（成功或失败）")
+    remark_val = None
+    if getattr(data, "remark", None) is not None:
+        remark_val = (data.remark or "").strip() or None
+
+    now = datetime.now()
+
+    incomplete_tasks = (
+        db.query(BizProductionTask)
+        .filter(
+            BizProductionTask.work_order_id == wo.id,
+            BizProductionTask.tenant_no == tenant_no,
+            BizProductionTask.del_flag == 0,
+            BizProductionTask.task_status != "ywc",
+        )
+        .all()
+    )
+    for task in incomplete_tasks:
+        production_task_svc._force_complete_task_steps(
+            db, task.id, tenant_no, user_id, now
+        )
+        task.task_status = "ywc"
+        task.conclusion = concl
+        if remark_val is not None:
+            task.remark = remark_val
+        if not task.completed_time:
+            task.completed_time = now
+        task.update_by = user_id
+        task.update_time = now
+        db.add(task)
+        production_task_svc._consume_task_input_cell_products(
+            db, task.id, user_id, now
+        )
+
     wo.work_order_status = "ywc"
     wo.update_by = user_id
-    wo.update_time = datetime.now()
-    if data and data.remark is not None:
-        wo.remark = data.remark
+    wo.update_time = now
+    if remark_val is not None:
+        wo.remark = remark_val
     db.add(wo)
+
     from app.crud.pilot_outbound_status_sync import sync_schedule_from_production_work_order
 
     sync_schedule_from_production_work_order(db, wo, user_id=user_id)
diff --git a/backend/tests/test_production_cascade_complete.py b/backend/tests/test_production_cascade_complete.py
index 661c12b0..be620564 100644
--- a/backend/tests/test_production_cascade_complete.py
+++ b/backend/tests/test_production_cascade_complete.py
@@ -1,14 +1,13 @@
 """生产完成任务/工单状态级联。"""
 import unittest
 from datetime import datetime
-from unittest.mock import MagicMock, patch
+from unittest.mock import ANY, MagicMock, patch
 
 from pydantic import ValidationError
 
-from app.schemas.production_work_order import ProductionWorkOrderComplete
-
-
 from app.crud import production_task as production_task_svc
+from app.crud import production_work_order as production_work_order_svc
+from app.schemas.production_work_order import ProductionWorkOrderComplete
 
 
 class TestProductionWorkOrderCompleteSchema(unittest.TestCase):
@@ -82,5 +81,70 @@ class TestUpdateTaskStatusCascadeSteps(unittest.TestCase):
         self.assertIs(result, task)
 
 
+class TestCompleteProductionWorkOrderCascade(unittest.TestCase):
+    @patch("app.crud.pilot_outbound_status_sync.sync_schedule_from_production_work_order")
+    @patch("app.crud.production_task._consume_task_input_cell_products")
+    @patch("app.crud.production_task._force_complete_task_steps")
+    @patch("app.crud.production_work_order.get_production_work_order")
+    def test_cascades_incomplete_tasks_and_steps(
+        self, mock_get_wo, mock_force, mock_consume, mock_sync_sched
+    ):
+        db = MagicMock()
+        wo = MagicMock(id="wo-1", work_order_status="jxz", remark=None)
+        mock_get_wo.return_value = wo
+
+        task_a = MagicMock(
+            id="ta",
+            task_status="dzx",
+            completed_time=None,
+            conclusion=None,
+            remark=None,
+        )
+        q = MagicMock()
+        db.query.return_value = q
+        q.filter.return_value = q
+        q.all.return_value = [task_a]
+
+        body = ProductionWorkOrderComplete(conclusion="sb", remark="batch fail")
+        result = production_work_order_svc.complete_production_work_order(
+            db, "wo-1", "t1", "user-1", body
+        )
+
+        mock_force.assert_called_once_with(db, "ta", "t1", "user-1", ANY)
+        self.assertEqual(task_a.task_status, "ywc")
+        self.assertEqual(task_a.conclusion, "sb")
+        self.assertEqual(task_a.remark, "batch fail")
+        mock_consume.assert_called_once()
+        self.assertEqual(wo.work_order_status, "ywc")
+        self.assertEqual(wo.remark, "batch fail")
+        mock_sync_sched.assert_called_once()
+        db.commit.assert_called_once()
+        self.assertIs(result, wo)
+
+    @patch("app.crud.production_work_order.get_production_work_order")
+    def test_rejects_invalid_conclusion(self, mock_get_wo):
+        db = MagicMock()
+        mock_get_wo.return_value = MagicMock(work_order_status="jxz")
+        bad = MagicMock(conclusion="xx", remark=None)
+        with self.assertRaises(ValueError):
+            production_work_order_svc.complete_production_work_order(
+                db, "wo-1", "t1", "user-1", bad
+            )
+
+    @patch("app.crud.production_work_order.get_production_work_order")
+    def test_rejects_already_completed(self, mock_get_wo):
+        db = MagicMock()
+        mock_get_wo.return_value = MagicMock(work_order_status="ywc")
+        with self.assertRaises(ValueError) as ctx:
+            production_work_order_svc.complete_production_work_order(
+                db,
+                "wo-1",
+                "t1",
+                "user-1",
+                ProductionWorkOrderComplete(conclusion="cg"),
+            )
+        self.assertIn("已完成", str(ctx.exception))
+
+
 if __name__ == "__main__":
-    unittest.main()
\ No newline at end of file
+    unittest.main()