fix(workflow): drop shared context on fork and legacy pickle context

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

app/application/workflow.py

@@ -659,7 +659,12 @@ class WorkflowDefinitionCommand:
         payload: Mapping[str, Any],
         share_id: Optional[int] = None,
     ) -> WorkflowMutationResult:
-        """解析共享工作流内容并在提交后更新远程复用次数。"""
+        """
+        解析共享工作流内容并在提交后更新远程复用次数。
+
+        只复用动作、流程和触发配置；执行上下文属于分享者本机的运行状态，对新工作流没有意义，
+        且来自外部不可信，因此忽略载荷中的 ``context``，新工作流固定从空上下文开始。
+        """
         values = dict(payload)
         if not values.get("name"):
             return WorkflowMutationResult(False, "工作流名称不能为空")
@@ -667,7 +672,6 @@ class WorkflowDefinitionCommand:
         for field, default, error_message in (
             ("actions", "[]", "actions字段JSON格式错误"),
             ("flows", "[]", "flows字段JSON格式错误"),
-            ("context", "{}", "context字段JSON格式错误"),
             ("event_conditions", "{}", "event_conditions字段JSON格式错误"),
         ):
             raw = values.get(field)
@@ -684,7 +688,7 @@ class WorkflowDefinitionCommand:
             "event_conditions": parsed["event_conditions"],
             "actions": parsed["actions"],
             "flows": parsed["flows"],
-            "context": parsed["context"],
+            "context": {},
             "state": "P",
         }
         if await self._repository.async_get_by_name(workflow_values["name"]):

app/chain/workflow.py

@@ -1,8 +1,6 @@
 import ast
-import base64
 import copy
 import inspect
-import pickle
 import threading
 from collections import defaultdict, deque
 from contextvars import Context, copy_context
@@ -353,7 +351,10 @@ class WorkflowExecutor:
 
     def restore_context(self) -> ActionContext:
         """
-        恢复工作流上下文，兼容旧版 Base64 Pickle 存储格式。
+        恢复工作流上下文。
+
+        只接受结构化 JSON 上下文；旧版 Base64 Pickle 格式（``content`` 字段）不再反序列化，
+        按空上下文从头执行，避免执行数据库中来源不可信的序列化载荷。
         """
         context = ActionContext()
         if self.workflow.context:
@@ -361,8 +362,7 @@ class WorkflowExecutor:
                 logger.info(f"工作流已执行动作：{self.workflow.current_action}")
             try:
                 if isinstance(self.workflow.context, dict) and self.workflow.context.get("content"):
-                    decoded_data = base64.b64decode(self.workflow.context["content"])
-                    context = pickle.loads(decoded_data)
+                    logger.warning(f"工作流 {self.workflow.name} 的上下文为已废弃的旧版格式，将从空上下文开始执行")
                 elif isinstance(self.workflow.context, dict):
                     context = ActionContext.model_validate(self.workflow.context)
             except Exception:

tests/test_workflow_execution.py

@@ -33,11 +33,19 @@ def _build_workflow(current_action=None, context=None, actions=None, flows=None,
     )
 
 
-def _encoded_context(context: ActionContext) -> dict:
-    """编码工作流恢复上下文。"""
-    return {
-        "content": base64.b64encode(pickle.dumps(context)).decode("utf-8"),
-    }
+_LEGACY_PAYLOAD_CALLS = []
+
+
+def _record_legacy_payload_call():
+    """记录旧版 pickle 载荷被反序列化时触发的调用。"""
+    _LEGACY_PAYLOAD_CALLS.append("loaded")
+
+
+class _LegacyPayload:
+    """反序列化时会执行回调的旧版上下文载荷。"""
+
+    def __reduce__(self):
+        return _record_legacy_payload_call, ()
 
 
 class _FakeWorkflowManager:
@@ -188,7 +196,7 @@ def test_workflow_executor_resumes_downstream_nodes(monkeypatch):
     fake_manager = _FakeWorkflowManager(calls)
     workflow = _build_workflow(
         current_action="A",
-        context=_encoded_context(ActionContext()),
+        context={"progress": 50},
     )
 
     monkeypatch.setattr(workflow_module, "get_workflow_manager", lambda: fake_manager)
@@ -203,6 +211,27 @@ def test_workflow_executor_resumes_downstream_nodes(monkeypatch):
     assert executor.context.progress == 100
 
 
+def test_workflow_executor_ignores_legacy_pickle_context(monkeypatch):
+    """旧版 Base64 Pickle 上下文不得被反序列化，应按空上下文继续执行。"""
+    _LEGACY_PAYLOAD_CALLS.clear()
+    calls = []
+    fake_manager = _FakeWorkflowManager(calls)
+    workflow = _build_workflow(
+        current_action="A",
+        context={"content": base64.b64encode(pickle.dumps(_LegacyPayload())).decode("utf-8")},
+    )
+
+    monkeypatch.setattr(workflow_module, "get_workflow_manager", lambda: fake_manager)
+    monkeypatch.setattr(workflow_module.runtime_stop_state, "resume_workflow", lambda workflow_id: None)
+    monkeypatch.setattr(workflow_module.runtime_stop_state, "is_workflow_stopped", lambda workflow_id: False)
+
+    executor = workflow_module.WorkflowExecutor(workflow)
+
+    assert _LEGACY_PAYLOAD_CALLS == []
+    assert executor.context.workflow_context == {}
+    assert executor.context.node_outputs == {}
+
+
 def test_workflow_executor_restores_structured_context(monkeypatch):
     """恢复执行时应兼容新版结构化上下文存储格式。"""
     calls = []

tests/test_workflow_mutation_command.py

@@ -312,6 +312,36 @@ async def test_fork_commits_before_reporting_remote_count():
     assert calls == ["commit", "report"]
 
 
+@pytest.mark.asyncio
+async def test_fork_ignores_shared_execution_context():
+    """复用共享工作流时不得导入分享载荷中的执行上下文。"""
+    command, dependencies = _definition_command()
+
+    result = await command.fork(
+        {
+            "name": "Forked",
+            "actions": "[]",
+            "flows": "[]",
+            "context": '{"content": "bGVnYWN5"}',
+        },
+    )
+
+    assert result.success is True
+    created_values = dependencies["repository"].stage_create.await_args.args[0]
+    assert created_values["context"] == {}
+
+
+@pytest.mark.asyncio
+async def test_fork_accepts_malformed_shared_context():
+    """分享载荷中的上下文不参与复用，其 JSON 格式错误也不应阻止创建。"""
+    command, dependencies = _definition_command()
+
+    result = await command.fork({"name": "Forked", "context": "{"})
+
+    assert result.success is True
+    dependencies["repository"].stage_create.assert_awaited_once()
+
+
 @pytest.mark.asyncio
 async def test_fork_invalid_json_stops_before_database_write():
     """共享内容 JSON 无效时不得创建半成品工作流。"""