1111

src/langbot/pkg/workflow/executor.py

@@ -10,6 +10,8 @@ import uuid
 from datetime import datetime
 from typing import Any, Optional, TYPE_CHECKING
 
+import sqlalchemy
+
 from .entities import (
     WorkflowDefinition,
     NodeDefinition,
@@ -20,6 +22,7 @@ from .entities import (
     NodeStatus,
     ExecutionStep,
 )
+from ..entity.persistence import workflow as persistence_workflow
 from .registry import NodeTypeRegistry
 
 if TYPE_CHECKING:
@@ -346,6 +349,8 @@ class WorkflowExecutor:
             logger.warning(f"Circular dependency detected at node: {node.id}")
             context.node_states[node.id].status = NodeStatus.SKIPPED
             context.node_states[node.id].error = "Circular dependency detected"
+            context.node_states[node.id].end_time = datetime.now()
+            await self._persist_node_execution(node, context.node_states[node.id], context)
             return
         
         # Add node to current path
@@ -353,7 +358,10 @@ class WorkflowExecutor:
         
         # Check if node should be skipped
         if await self._should_skip_node(node, context):
-            context.node_states[node.id].status = NodeStatus.SKIPPED
+            existing_state = context.node_states[node.id]
+            if existing_state.status == NodeStatus.SKIPPED:
+                existing_state.end_time = existing_state.end_time or datetime.now()
+                await self._persist_node_execution(node, existing_state, context)
             path.discard(node.id)
             return
         
@@ -469,6 +477,7 @@ class WorkflowExecutor:
             node_state.error = f"Unknown node type: {node.type}"
             node_state.end_time = datetime.now()
             self._record_execution_step(node, node_state, context)
+            await self._persist_node_execution(node, node_state, context)
             return
         
         # Resolve inputs
@@ -482,6 +491,7 @@ class WorkflowExecutor:
             node_state.error = "; ".join(validation_errors)
             node_state.end_time = datetime.now()
             self._record_execution_step(node, node_state, context)
+            await self._persist_node_execution(node, node_state, context)
             return
         
         # Execute with retries
@@ -523,6 +533,7 @@ class WorkflowExecutor:
                     )
         
         self._record_execution_step(node, node_state, context)
+        await self._persist_node_execution(node, node_state, context)
     
     async def _resolve_inputs(
         self, 
@@ -738,6 +749,47 @@ class WorkflowExecutor:
         )
         context.history.append(step)
 
+    async def _persist_node_execution(
+        self,
+        node: NodeDefinition,
+        node_state: NodeState,
+        context: ExecutionContext,
+    ):
+        """Persist node execution state for execution detail and logs."""
+        if not self.ap:
+            return
+
+        values = {
+            'execution_uuid': context.execution_id,
+            'node_id': node.id,
+            'node_type': node.type,
+            'status': node_state.status.value,
+            'inputs': node_state.inputs,
+            'outputs': node_state.outputs,
+            'start_time': node_state.start_time,
+            'end_time': node_state.end_time,
+            'error': node_state.error,
+            'retry_count': node_state.retry_count,
+        }
+
+        existing_query = sqlalchemy.select(persistence_workflow.WorkflowNodeExecution).where(
+            persistence_workflow.WorkflowNodeExecution.execution_uuid == context.execution_id,
+            persistence_workflow.WorkflowNodeExecution.node_id == node.id,
+        )
+        existing_result = await self.ap.persistence_mgr.execute_async(existing_query)
+        existing = existing_result.first()
+
+        if existing is None:
+            await self.ap.persistence_mgr.execute_async(
+                sqlalchemy.insert(persistence_workflow.WorkflowNodeExecution).values(**values)
+            )
+        else:
+            await self.ap.persistence_mgr.execute_async(
+                sqlalchemy.update(persistence_workflow.WorkflowNodeExecution)
+                .where(persistence_workflow.WorkflowNodeExecution.id == existing.id)
+                .values(**values)
+            )
+
 
 class ParallelExecutor:
     """Execute multiple branches in parallel"""
@@ -997,8 +1049,8 @@ class DebugWorkflowExecutor(WorkflowExecutor):
         
         # Check if should skip
         if await self._should_skip_node(node, context):
-            context.node_states[node.id].status = NodeStatus.SKIPPED
-            debug_state.add_log('info', f'Skipping node: {node.id}', node_id=node.id)
+            if context.node_states[node.id].status == NodeStatus.SKIPPED:
+                debug_state.add_log('info', f'Skipping node: {node.id}', node_id=node.id)
             return
         
         # Check breakpoint
@@ -1076,6 +1128,7 @@ class DebugWorkflowExecutor(WorkflowExecutor):
             node_state.end_time = datetime.now()
             debug_state.add_log('error', f'Unknown node type: {node.type}', node_id=node.id)
             self._record_execution_step(node, node_state, context)
+            await self._persist_node_execution(node, node_state, context)
             return
         
         # Resolve inputs
@@ -1100,6 +1153,7 @@ class DebugWorkflowExecutor(WorkflowExecutor):
                 node_id=node.id
             )
             self._record_execution_step(node, node_state, context)
+            await self._persist_node_execution(node, node_state, context)
             return
         
         # Execute with retries
@@ -1147,6 +1201,7 @@ class DebugWorkflowExecutor(WorkflowExecutor):
                     )
         
         self._record_execution_step(node, node_state, context)
+        await self._persist_node_execution(node, node_state, context)
     
     async def step_execute(
         self,

src/langbot/pkg/workflow/node.py

@@ -148,6 +148,7 @@ class WorkflowNode(abc.ABC):
         'llm-model-selector': 'llm-model-selector',
         'embedding-model-selector': 'embedding-model-selector',
         'rerank-model-selector': 'rerank-model-selector',
+        'pipeline-selector': 'pipeline-selector',
         'knowledge-base-selector': 'knowledge-base-selector',
         'knowledge-base-multi-selector': 'knowledge-base-multi-selector',
         'bot-selector': 'bot-selector',

src/langbot/pkg/workflow/nodes/call_pipeline.py

@@ -7,6 +7,13 @@ from __future__ import annotations
 
 from typing import Any, ClassVar
 
+import langbot_plugin.api.definition.abstract.platform.adapter as abstract_platform_adapter
+import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query
+import langbot_plugin.api.entities.builtin.platform.entities as platform_entities
+import langbot_plugin.api.entities.builtin.platform.events as platform_events
+import langbot_plugin.api.entities.builtin.platform.message as platform_message
+import langbot_plugin.api.entities.builtin.provider.session as provider_session
+
 from ..entities import ExecutionContext
 from ..node import WorkflowNode, workflow_node, NodePort, NodeConfig
 
@@ -30,7 +37,187 @@ class CallPipelineNode(WorkflowNode):
     config_schema: ClassVar[list[NodeConfig]] = []
 
     async def execute(self, inputs: dict[str, Any], context: ExecutionContext) -> dict[str, Any]:
-        query = inputs.get("query", "")
-        pipeline_uuid = self.get_config("pipeline_uuid", "")
+        if not self.ap:
+            raise RuntimeError('Application instance not available — cannot call pipeline')
 
-        return {"response": f"[Pipeline {pipeline_uuid} response for: {query[:50]}...]", "result": {}}
+        raw_query = inputs.get('query', '')
+        query_text = str(raw_query or inputs.get('input') or '')
+        pipeline_ref = str(self.get_config('pipeline_uuid', '') or '').strip()
+
+        if not pipeline_ref:
+            raise ValueError('No pipeline configured for call pipeline node')
+
+        pipeline_data = await self.ap.pipeline_service.get_pipeline(pipeline_ref)
+        if pipeline_data is None:
+            pipeline_data = await self.ap.pipeline_service.get_pipeline_by_name(pipeline_ref)
+        if pipeline_data is None:
+            raise ValueError(f'Pipeline not found: {pipeline_ref}')
+
+        pipeline_uuid = str(pipeline_data.get('uuid', '') or '')
+        if not pipeline_uuid:
+            raise ValueError(f'Pipeline UUID missing for: {pipeline_ref}')
+
+        runtime_pipeline = await self.ap.pipeline_mgr.get_pipeline_by_uuid(pipeline_uuid)
+        if runtime_pipeline is None:
+            raise ValueError(f'Runtime pipeline not loaded: {pipeline_uuid}')
+
+        adapter = _WorkflowPipelineCaptureAdapter(context=context)
+        adapter.bot_account_id = 'workflow-call-pipeline'
+
+        message_event = self._build_message_event(query_text, context)
+        message_chain = message_event.message_chain
+        launcher_type = provider_session.LauncherTypes.GROUP if context.message_context and context.message_context.is_group else provider_session.LauncherTypes.PERSON
+        launcher_id = context.session_id or context.execution_id
+        sender_id = (
+            context.message_context.sender_id
+            if context.message_context and context.message_context.sender_id
+            else context.user_id or f'workflow_{context.execution_id}'
+        )
+
+        query = pipeline_query.Query(
+            bot_uuid=context.bot_id,
+            query_id=-1,
+            launcher_type=launcher_type,
+            launcher_id=launcher_id,
+            sender_id=sender_id,
+            message_event=message_event,
+            message_chain=message_chain,
+            variables={
+                '_called_from_workflow': True,
+                '_workflow_execution_id': context.execution_id,
+                '_workflow_id': context.workflow_id,
+                **dict(context.variables or {}),
+            },
+            resp_messages=[],
+            resp_message_chain=[],
+            adapter=adapter,
+            pipeline_uuid=pipeline_uuid,
+        )
+
+        await runtime_pipeline.run(query)
+
+        response_text = adapter.get_last_text_response()
+        result = {
+            'pipeline_uuid': pipeline_uuid,
+            'pipeline_name': pipeline_data.get('name', ''),
+            'responses': adapter.responses,
+            'query_text': query_text,
+        }
+
+        return {'response': response_text, 'result': result}
+
+    def _build_message_event(
+        self,
+        query_text: str,
+        context: ExecutionContext,
+    ) -> platform_events.MessageEvent:
+        message_chain_data = context.trigger_data.get('message_chain') or context.trigger_data.get('message', [])
+        if isinstance(message_chain_data, list) and message_chain_data:
+            message_chain = platform_message.MessageChain.model_validate(message_chain_data)
+        else:
+            message_chain = platform_message.MessageChain([platform_message.Plain(text=query_text)])
+
+        if context.message_context and context.message_context.is_group:
+            group = platform_entities.Group(
+                id=context.message_context.group_id or context.session_id or 'workflow_group',
+                name='Workflow Group',
+                permission=platform_entities.Permission.Member,
+            )
+            sender = platform_entities.GroupMember(
+                id=context.message_context.sender_id,
+                member_name=context.message_context.sender_name or 'Workflow User',
+                permission=platform_entities.Permission.Member,
+                group=group,
+            )
+            return platform_events.GroupMessage(
+                sender=sender,
+                message_chain=message_chain,
+                time=context.message_context.raw_message.get('time') if context.message_context.raw_message else None,
+            )
+
+        sender = platform_entities.Friend(
+            id=context.message_context.sender_id if context.message_context else context.user_id or 'workflow_user',
+            nickname=context.message_context.sender_name if context.message_context else 'Workflow User',
+            remark=context.message_context.sender_name if context.message_context else 'Workflow User',
+        )
+        return platform_events.FriendMessage(
+            sender=sender,
+            message_chain=message_chain,
+            time=context.message_context.raw_message.get('time') if context.message_context and context.message_context.raw_message else None,
+        )
+
+
+class _WorkflowPipelineCaptureAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
+    responses: list[dict[str, Any]] = []
+
+    def __init__(self, context: ExecutionContext):
+        super().__init__(config={}, logger=None)
+        self.context = context
+        self.responses = []
+
+    async def send_message(self, target_type: str, target_id: str, message: platform_message.MessageChain):
+        payload = {
+            'type': 'send',
+            'target_type': target_type,
+            'target_id': target_id,
+            'content': str(message),
+            'message_chain': message.model_dump(),
+        }
+        self.responses.append(payload)
+        return payload
+
+    async def reply_message(
+        self,
+        message_source: platform_events.MessageEvent,
+        message: platform_message.MessageChain,
+        quote_origin: bool = False,
+    ):
+        payload = {
+            'type': 'reply',
+            'content': str(message),
+            'message_chain': message.model_dump(),
+            'quote_origin': quote_origin,
+        }
+        self.responses.append(payload)
+        return payload
+
+    async def reply_message_chunk(
+        self,
+        message_source: platform_events.MessageEvent,
+        bot_message: dict,
+        message: platform_message.MessageChain,
+        quote_origin: bool = False,
+        is_final: bool = False,
+    ):
+        payload = {
+            'type': 'reply_chunk',
+            'content': str(message),
+            'message_chain': message.model_dump(),
+            'quote_origin': quote_origin,
+            'is_final': is_final,
+        }
+        self.responses.append(payload)
+        return payload
+
+    async def create_message_card(self, message_id, event: platform_events.MessageEvent) -> bool:
+        return False
+
+    def register_listener(self, event_type, callback):
+        return None
+
+    def unregister_listener(self, event_type, callback):
+        return None
+
+    async def run_async(self):
+        return None
+
+    async def is_stream_output_supported(self) -> bool:
+        return False
+
+    async def kill(self) -> bool:
+        return True
+
+    def get_last_text_response(self) -> str:
+        if not self.responses:
+            return ''
+        return str(self.responses[-1].get('content', '') or '')

src/langbot/pkg/workflow/nodes/reply_message.py

@@ -38,6 +38,8 @@ class ReplyMessageNode(WorkflowNode):
             message = inputs.get("input")
         if message in (None, ""):
             message = inputs.get("response")
+        if message in (None, ""):
+            message = inputs.get("content")
         if message in (None, "") and context.message_context:
             message = context.message_context.message_content
         if message is None: