From 9bca0c690fb85aa4282a90114dbd520fdcfa2312 Mon Sep 17 00:00:00 2001
From: ViperEkura <3081035982@qq.com>
Date: Thu, 30 Apr 2026 15:32:16 +0800
Subject: [PATCH] =?UTF-8?q?feat:=20=E5=A2=9E=E5=8A=A0plan=20=E7=AD=89?=
 =?UTF-8?q?=E9=83=A8=E5=88=86?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 assets/ARCHITECTURE.md                        |  87 ++++-
 dashboard/src/App.vue                         |   5 +
 dashboard/src/components/MessageBubble.vue    |   5 +-
 dashboard/src/components/room/RoomSidebar.vue |   5 +-
 dashboard/src/utils/api.js                    |   8 +-
 dashboard/src/utils/parallelStreamManager.js  |  10 +-
 dashboard/src/utils/useConversations.js       |   8 +-
 dashboard/src/views/ConversationView.vue      |  81 +++-
 luxx/routes/__init__.py                       |   3 +-
 luxx/routes/tasks.py                          | 351 ++++++++++++++++++
 luxx/services/task.py                         | 317 ++++++++++++++--
 luxx/services/task_executor.py                | 257 +++++++++++++
 12 files changed, 1067 insertions(+), 70 deletions(-)
 create mode 100644 luxx/routes/tasks.py
 create mode 100644 luxx/services/task_executor.py

diff --git a/assets/ARCHITECTURE.md b/assets/ARCHITECTURE.md
index 34586f3..8889d7d 100644
--- a/assets/ARCHITECTURE.md
+++ b/assets/ARCHITECTURE.md
@@ -598,7 +598,7 @@ LLM API 客户端：
 
 ### 7. 任务系统 (`services/task.py`)
 
-用于自主任务执行和依赖管理：
+用于自主任务执行和计划管理：
 
 ```mermaid
 classDiagram
@@ -608,19 +608,24 @@ classDiagram
         +str goal
         +TaskStatus status
         +List~Step~ steps
-        +List~Task~ subtasks
+        +str metadata
+        +get_step() Step?
+        +is_complete() bool
     }
 
     class Step {
         +str id
         +str name
+        +str goal
         +List~str~ depends_on
         +StepStatus status
+        +Dict tool_calls
     }
 
     class TaskGraph {
         +topological_sort() List~Step~
         +get_ready_steps() List~Step~
+        +get_parallel_levels() List~List~Step~~
         +detect_cycles() List~List~str~~
         +validate() tuple
     }
@@ -630,20 +635,26 @@ classDiagram
         +get_task() Task
         +update_task_status() Task
         +add_steps() List~Step~
+        +set_steps_from_plan() List~Step~
         +build_graph() TaskGraph
     }
 
+    class Planner {
+        +plan() List~Dict~
+        +plan_sync() List~Dict~
+    }
+
     Task "1" o-- "*" Step
-    Task "1" o-- "*" Task
     TaskService ..> TaskGraph
+    Planner ..> TaskService
 ```
 
-**任务状态 (TaskStatus)：**
-- `PENDING` - 待处理
-- `READY` - 就绪
+**任务状态 (TaskStatus) - 五状态模型：**
+- `PENDING` - 待处理/就绪
 - `RUNNING` - 运行中
 - `BLOCK` - 阻塞
-- `TERMINATED` - 已终止
+- `TERMINATED` - 终止
+- `COMPLETED` - 完成
 
 **步骤状态 (StepStatus)：**
 - `PENDING` - 待执行
@@ -652,6 +663,56 @@ classDiagram
 - `FAILED` - 失败
 - `SKIPPED` - 跳过
 
+**TaskGraph 功能：**
+- `topological_sort()` - 拓扑排序获取执行顺序
+- `get_ready_steps()` - 获取就绪步骤（依赖已满足）
+- `get_parallel_levels()` - 获取并行执行层级
+- `detect_cycles()` - 检测循环依赖
+- `validate()` - 验证图结构
+
+**Planner 功能：**
+- 调用 LLM 自动分解目标为步骤
+- 支持上下文（可用工具、约束条件）
+- 返回结构化步骤列表
+
+### 8. 任务执行服务 (`services/task_executor.py`)
+
+将 Task/Plan 集成到 Agent 执行流程：
+
+```mermaid
+classDiagram
+    class TaskExecutorService {
+        +llm: LLMClient
+        +generate_plan() List~Dict~
+        +create_task_with_plan() Task
+        +get_execution_plan() Dict
+    }
+
+    class AgentWithPlan {
+        +chat_service
+        +execute_with_plan() AsyncGenerator
+    }
+
+    TaskExecutorService --> TaskService
+    AgentWithPlan --> TaskExecutorService
+    AgentWithPlan --> ChatService
+```
+
+**执行流程：**
+1. 用户输入目标 → AgentWithPlan 接收
+2. TaskExecutorService.generate_plan() 调用 LLM 分解目标
+3. 创建 Task 并设置步骤
+4. 按 TaskGraph 拓扑顺序执行每个步骤
+5. 每个步骤作为独立 Agent 对话执行
+6. 步骤完成后更新状态，继续下一个
+
+**SSE 事件：**
+- `planning` - 计划生成中
+- `plan_created` - 计划已创建
+- `step_start` - 步骤开始
+- `step_complete` - 步骤完成
+- `task_complete` - 任务完成
+
 ### 7. 认证系统 (`routes/auth.py`)
 - JWT Bearer Token
 - Bcrypt 密码哈希
@@ -814,6 +875,18 @@ room_started → round_start → (message_start → message_chunk* → message_e
 | `/tools` | GET | 可用工具列表 |
 | `/tools/{name}` | GET | 工具详情 |
 | `/tools/{name}/execute` | POST | 执行工具 |
+| `/tasks` | GET | 任务列表 |
+| `/tasks` | POST | 创建任务 |
+| `/tasks/{id}` | GET | 任务详情 |
+| `/tasks/{id}` | PUT | 更新任务 |
+| `/tasks/{id}` | DELETE | 删除任务 |
+| `/tasks/{id}/steps` | POST | 添加步骤 |
+| `/tasks/{id}/steps/{step_id}` | PUT | 更新步骤 |
+| `/tasks/{id}/graph` | GET | 获取任务图 |
+| `/tasks/{id}/ready-steps` | GET | 获取就绪步骤 |
+| `/tasks/{id}/execute` | POST | 执行任务 |
+| `/tasks/{id}/export` | GET | 导出任务 |
+| `/tasks/import` | POST | 导入任务 |
 | `/health` | GET | 健康检查 |
 | `/` | GET | 服务信息 |
 
diff --git a/dashboard/src/App.vue b/dashboard/src/App.vue
index 07033b6..5256da6 100644
--- a/dashboard/src/App.vue
+++ b/dashboard/src/App.vue
@@ -1,8 +1,13 @@
 <script setup>
 import { useAuth } from './utils/useAuth.js'
 import AppHeader from './components/AppHeader.vue'
+import { useRouter } from 'vue-router'
 
 const { isLoggedIn } = useAuth()
+const router = useRouter()
+
+// 设置全局 router 引用，供 api.js 响应拦截器使用
+window.__VUE_ROUTER__ = router
 </script>
 
 <template>
diff --git a/dashboard/src/components/MessageBubble.vue b/dashboard/src/components/MessageBubble.vue
index e85fe1a..8ffb914 100644
--- a/dashboard/src/components/MessageBubble.vue
+++ b/dashboard/src/components/MessageBubble.vue
@@ -157,12 +157,11 @@ const regenerateIcon = `<svg viewBox="0 0 24 24" width="14" height="14" fill="no
   display: flex;
   align-items: center;
   gap: 10px;
-  padding: 8px 0 0;
+  margin-top: 8px;
+  padding-top: 10px;
   font-size: 12px;
   color: var(--text-tertiary);
   border-top: 1px solid transparent;
-  margin-top: 8px;
-  padding-top: 10px;
 }
 
 
diff --git a/dashboard/src/components/room/RoomSidebar.vue b/dashboard/src/components/room/RoomSidebar.vue
index 0d407ae..8ed7f29 100644
--- a/dashboard/src/components/room/RoomSidebar.vue
+++ b/dashboard/src/components/room/RoomSidebar.vue
@@ -194,8 +194,9 @@ function handleBack() {
     <div v-if="sidebarTab === 'roomAgents' && room" class="sidebar-tab-content">
       <div class="sidebar-header sidebar-header-row">
         <button class="btn-back" @click="handleBack">
-          <svg width="18" height="18">
-            <use href="#arrow-left-icon"/>
+          <svg width="18" height="18" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2" stroke-linecap="round" stroke-linejoin="round">
+            <line x1="19" y1="12" x2="5" y2="12"></line>
+            <polyline points="12 19 5 12 12 5"></polyline>
           </svg>
         </button>
         <span class="sidebar-title">{{ room.title }}</span>
diff --git a/dashboard/src/utils/api.js b/dashboard/src/utils/api.js
index b68af6d..70749d8 100644
--- a/dashboard/src/utils/api.js
+++ b/dashboard/src/utils/api.js
@@ -30,7 +30,12 @@ api.interceptors.response.use(
     if (error.response?.status === 401) {
       localStorage.removeItem('access_token')
       localStorage.removeItem('user')
-      window.location.href = '/auth'
+      // 使用 Vue Router 跳转，避免 SPA 路由丢失
+      if (window.__VUE_ROUTER__) {
+        window.__VUE_ROUTER__.push('/auth')
+      } else {
+        window.location.href = '/auth'
+      }
     }
     return Promise.reject(error.response?.data || error.message)
   }
@@ -106,6 +111,7 @@ export const chatRoomsAPI = {
   delete: (id) => api.delete(`/chat-rooms/${id}`),
   getMessages: (id) => api.get(`/chat-rooms/${id}/messages`),
   start: (id) => `/api/chat-rooms/${id}/start`,
+  // 注意: start 返回路径字符串，由调用方使用 fetch 处理 SSE 流
   stop: (id) => api.post(`/chat-rooms/${id}/stop`),
   reset: (id) => api.post(`/chat-rooms/${id}/reset`),
   addAgent: (roomId, data) => api.post(`/chat-rooms/${roomId}/agents`, data),
diff --git a/dashboard/src/utils/parallelStreamManager.js b/dashboard/src/utils/parallelStreamManager.js
index 638d48b..ba690d4 100644
--- a/dashboard/src/utils/parallelStreamManager.js
+++ b/dashboard/src/utils/parallelStreamManager.js
@@ -108,24 +108,24 @@ class ParallelStreamManager {
         break
         
       case 'message_start':
-        store.startAgentStream(roomId, data.agent_id || data.agentId, data)
+        store.startAgentStream(roomId, data.agent_id, data)
         break
         
       case 'message_chunk':
-        store.updateAgentContent(roomId, data.agent_id || data.agentId, {
+        store.updateAgentContent(roomId, data.agent_id, {
           content: data.content || '',
           progress: data.progress || 0
         })
         break
         
       case 'message_end':
-        store.completeAgentStream(roomId, data.agent_id || data.agentId, data)
+        store.completeAgentStream(roomId, data.agent_id, data)
         break
         
       case 'agent_error':
-        store.errorAgentStream(roomId, data.agent_id || data.agentId, {
+        store.errorAgentStream(roomId, data.agent_id, {
           message: data.error,
-          agentName: data.agent_name || data.agentName
+          agentName: data.agent_name
         })
         break
         
diff --git a/dashboard/src/utils/useConversations.js b/dashboard/src/utils/useConversations.js
index fbb48b9..d6d68d3 100644
--- a/dashboard/src/utils/useConversations.js
+++ b/dashboard/src/utils/useConversations.js
@@ -102,8 +102,8 @@ export function useConversations() {
         convMessages.value = res.data?.messages || []
         // 加载完成后强制滚动到底部（初始加载总是显示最新消息）
         nextTick(() => {
-          if (typeof onInitialScroll === 'function') {
-            onInitialScroll()
+          if (typeof initialScrollCallback === 'function') {
+            initialScrollCallback()
           }
         })
       }
@@ -278,10 +278,10 @@ export function useConversations() {
   }
   
   // 初始滚动回调（由外部设置）
-  let onInitialScroll = null
+  let initialScrollCallback = null
   
   const setOnInitialScroll = (callback) => {
-    onInitialScroll = callback
+    initialScrollCallback = callback
   }
   
   return {
diff --git a/dashboard/src/views/ConversationView.vue b/dashboard/src/views/ConversationView.vue
index 36daa37..b1e76cd 100644
--- a/dashboard/src/views/ConversationView.vue
+++ b/dashboard/src/views/ConversationView.vue
@@ -94,21 +94,23 @@
             </div>
           </div>
           <div class="chat-input-area">
-            <input 
-              v-model="newMessage" 
-              @keyup.enter="handleSend" 
-              type="text" 
-              placeholder="输入消息..." 
-              class="chat-input" 
-              :disabled="sending"
-            />
-            <button @click="handleSend" class="btn-send" :disabled="sending || !newMessage.trim()" title="发送">
-              <svg v-if="!sending" viewBox="0 0 24 24" width="18" height="18" fill="none" stroke="currentColor" stroke-width="2" stroke-linecap="round" stroke-linejoin="round">
-                <line x1="22" y1="2" x2="11" y2="13"></line>
-                <polygon points="22 2 15 22 11 13 2 9 22 2"></polygon>
-              </svg>
-              <span v-else>...</span>
-            </button>
+            <div class="input-wrapper">
+              <textarea 
+                v-model="newMessage" 
+                @keydown.enter.exact.prevent="handleSend"
+                placeholder="输入消息... (Shift+Enter 换行)" 
+                class="chat-input" 
+                :disabled="sending"
+                rows="3"
+              ></textarea>
+              <button @click="handleSend" class="btn-send" :disabled="sending || !newMessage.trim()" title="发送">
+                <svg v-if="!sending" viewBox="0 0 24 24" width="18" height="18" fill="none" stroke="currentColor" stroke-width="2" stroke-linecap="round" stroke-linejoin="round">
+                  <line x1="22" y1="2" x2="11" y2="13"></line>
+                  <polygon points="22 2 15 22 11 13 2 9 22 2"></polygon>
+                </svg>
+                <span v-else>...</span>
+              </button>
+            </div>
           </div>
         </div>
       </main>
@@ -424,10 +426,51 @@ onUnmounted(() => {
 .loading-messages .spinner-small { margin-bottom: 0.5rem; }
 
 /* 聊天输入区 */
-.chat-input-area { padding: 1rem; border-top: 1px solid var(--border-light); display: flex; gap: 0.75rem; }
-.chat-input { flex: 1; padding: 0.65rem 0.9rem; border: 1px solid var(--border-input); border-radius: 8px; background: var(--bg-input); color: var(--text-primary); font-size: 0.9rem; }
-.chat-input:focus { outline: none; border-color: var(--accent-primary); }
-.btn-send { width: 40px; height: 40px; background: var(--accent-primary); color: white; border: none; border-radius: 8px; cursor: pointer; display: flex; align-items: center; justify-content: center; transition: all 0.15s ease; }
+.chat-input-area { padding: 1rem; border-top: 1px solid var(--border-light); }
+.input-wrapper { 
+  position: relative; 
+  display: flex; 
+  background: var(--bg-input); 
+  border: 1px solid var(--border-input); 
+  border-radius: 10px; 
+  transition: border-color 0.2s ease;
+}
+.input-wrapper:focus-within { border-color: var(--accent-primary); }
+.chat-input { 
+  flex: 1; 
+  padding: 0.75rem 3rem 0.75rem 1rem; 
+  border: none; 
+  background: transparent; 
+  color: var(--text-primary); 
+  font-size: 0.9rem; 
+  font-family: inherit;
+  resize: none;
+  min-height: 72px;
+  max-height: 200px;
+  overflow-y: auto;
+  line-height: 1.5;
+  white-space: pre-wrap;
+  word-break: break-word;
+  box-sizing: border-box;
+}
+.chat-input:focus { outline: none; }
+.chat-input:disabled { opacity: 0.6; }
+.btn-send { 
+  position: absolute; 
+  right: 8px; 
+  bottom: 8px; 
+  width: 36px; 
+  height: 36px; 
+  background: var(--accent-primary); 
+  color: white; 
+  border: none; 
+  border-radius: 8px; 
+  cursor: pointer; 
+  display: flex; 
+  align-items: center; 
+  justify-content: center; 
+  transition: all 0.15s ease; 
+}
 .btn-send:hover:not(:disabled) { background: var(--accent-primary-hover); transform: translateY(-1px); box-shadow: 0 2px 8px rgba(37, 99, 235, 0.3); }
 .btn-send:disabled { opacity: 0.5; cursor: not-allowed; }
 
diff --git a/luxx/routes/__init__.py b/luxx/routes/__init__.py
index 247603a..d2e599c 100644
--- a/luxx/routes/__init__.py
+++ b/luxx/routes/__init__.py
@@ -1,7 +1,7 @@
 """API routes module"""
 from fastapi import APIRouter
 
-from luxx.routes import auth, conversations, messages, tools, providers, chat_rooms, agents
+from luxx.routes import auth, conversations, messages, tools, providers, chat_rooms, agents, tasks
 
 
 api_router = APIRouter()
@@ -14,3 +14,4 @@ api_router.include_router(tools.router)
 api_router.include_router(providers.router)
 api_router.include_router(chat_rooms.router)
 api_router.include_router(agents.router)
+api_router.include_router(tasks.router)
diff --git a/luxx/routes/tasks.py b/luxx/routes/tasks.py
new file mode 100644
index 0000000..9057ffb
--- /dev/null
+++ b/luxx/routes/tasks.py
@@ -0,0 +1,351 @@
+"""Task API routes"""
+from fastapi import APIRouter, HTTPException, Depends
+from pydantic import BaseModel
+from typing import List, Optional, Dict, Any
+from datetime import datetime
+from luxx.services.task import (
+    task_service,
+    TaskService,
+    TaskStatus,
+    StepStatus
+)
+from luxx.utils.helpers import success_response, error_response
+
+router = APIRouter(prefix="/tasks", tags=["tasks"])
+
+
+class StepCreate(BaseModel):
+    """Step creation model"""
+    name: str
+    description: str = ""
+    goal: str = ""
+    depends_on: List[str] = []
+
+    class Config:
+        json_schema_extra = {
+            "example": {
+                "name": "Read file",
+                "description": "Read the input file",
+                "goal": "Use file_read tool to read data.txt",
+                "depends_on": []
+            }
+        }
+
+
+class TaskCreate(BaseModel):
+    """Task creation model"""
+    name: str
+    goal: str
+    description: str = ""
+    steps: List[StepCreate] = []
+
+    class Config:
+        json_schema_extra = {
+            "example": {
+                "name": "Data Processing",
+                "goal": "Process and analyze the input data",
+                "description": "A complex data processing pipeline",
+                "steps": []
+            }
+        }
+
+
+class TaskUpdate(BaseModel):
+    """Task update model"""
+    name: Optional[str] = None
+    goal: Optional[str] = None
+    description: Optional[str] = None
+    status: Optional[str] = None
+
+
+class StepUpdate(BaseModel):
+    """Step update model"""
+    name: Optional[str] = None
+    description: Optional[str] = None
+    goal: Optional[str] = None
+    status: Optional[str] = None
+
+
+class PlanGenerate(BaseModel):
+    """Request model for plan generation"""
+    goal: str
+    context: Optional[Dict[str, Any]] = None
+
+    class Config:
+        json_schema_extra = {
+            "example": {
+                "goal": "Analyze sales data and generate a report",
+                "context": {
+                    "available_tools": ["file_read", "python_execute"],
+                    "constraints": "Must use Python for analysis"
+                }
+            }
+        }
+
+
+def get_task_service() -> TaskService:
+    """Dependency to get task service"""
+    return task_service
+
+
+@router.get("")
+async def list_tasks(service: TaskService = Depends(get_task_service)):
+    """List all tasks"""
+    tasks = service.list_tasks()
+    return success_response([t.to_dict() for t in tasks])
+
+
+@router.post("")
+async def create_task(
+    task_data: TaskCreate,
+    service: TaskService = Depends(get_task_service)
+):
+    """Create a new task"""
+    steps = [s.model_dump() for s in task_data.steps]
+    task = service.create_task(
+        name=task_data.name,
+        goal=task_data.goal,
+        description=task_data.description,
+        steps=steps if steps else None
+    )
+    return success_response(task.to_dict(), "Task created successfully")
+
+
+@router.get("/{task_id}")
+async def get_task(
+    task_id: str,
+    service: TaskService = Depends(get_task_service)
+):
+    """Get task by ID"""
+    task = service.get_task(task_id)
+    if not task:
+        return error_response("Task not found", code=404)
+    return success_response(task.to_dict())
+
+
+@router.put("/{task_id}")
+async def update_task(
+    task_id: str,
+    update_data: TaskUpdate,
+    service: TaskService = Depends(get_task_service)
+):
+    """Update task"""
+    task = service.get_task(task_id)
+    if not task:
+        return error_response("Task not found", code=404)
+
+    if update_data.name is not None:
+        task.name = update_data.name
+    if update_data.goal is not None:
+        task.goal = update_data.goal
+    if update_data.description is not None:
+        task.description = update_data.description
+    if update_data.status is not None:
+        try:
+            task.status = TaskStatus(update_data.status)
+        except ValueError:
+            return error_response(f"Invalid status: {update_data.status}")
+
+    task.updated_at = datetime.now()
+    return success_response(task.to_dict())
+
+
+@router.delete("/{task_id}")
+async def delete_task(
+    task_id: str,
+    service: TaskService = Depends(get_task_service)
+):
+    """Delete task"""
+    if not service.delete_task(task_id):
+        return error_response("Task not found", code=404)
+    return success_response(message="Task deleted successfully")
+
+
+# Step endpoints
+
+@router.post("/{task_id}/steps")
+async def add_steps(
+    task_id: str,
+    steps: List[StepCreate],
+    service: TaskService = Depends(get_task_service)
+):
+    """Add steps to task"""
+    steps_data = [s.model_dump() for s in steps]
+    added = service.add_steps(task_id, steps_data)
+    if added is None:
+        return error_response("Task not found", code=404)
+    return success_response([s.to_dict() for s in added], "Steps added successfully")
+
+
+@router.get("/{task_id}/graph")
+async def get_task_graph(
+    task_id: str,
+    service: TaskService = Depends(get_task_service)
+):
+    """Get task dependency graph"""
+    task = service.get_task(task_id)
+    if not task:
+        return error_response("Task not found", code=404)
+
+    graph = service.build_graph(task_id)
+    if not graph:
+        return error_response("Failed to build graph")
+
+    # Validate graph
+    is_valid, error = graph.validate()
+    if not is_valid:
+        return error_response(f"Invalid graph: {error}")
+
+    # Get execution order
+    sorted_steps = graph.topological_sort()
+
+    # Get parallel execution levels
+    parallel_levels = graph.get_parallel_levels()
+
+    return success_response({
+        "task_id": task_id,
+        "steps": [s.to_dict() for s in task.steps],
+        "execution_order": [s.id for s in sorted_steps],
+        "parallel_levels": [[s.id for s in level] for level in parallel_levels],
+        "is_valid": True
+    })
+
+
+@router.get("/{task_id}/ready-steps")
+async def get_ready_steps(
+    task_id: str,
+    service: TaskService = Depends(get_task_service)
+):
+    """Get steps that are ready to execute"""
+    task = service.get_task(task_id)
+    if not task:
+        return error_response("Task not found", code=404)
+
+    graph = service.build_graph(task_id)
+    if not graph:
+        return error_response("Failed to build graph")
+
+    ready = graph.get_ready_steps()
+    return success_response([s.to_dict() for s in ready])
+
+
+@router.put("/{task_id}/steps/{step_id}")
+async def update_step(
+    task_id: str,
+    step_id: str,
+    update_data: StepUpdate,
+    service: TaskService = Depends(get_task_service)
+):
+    """Update step"""
+    task = service.get_task(task_id)
+    if not task:
+        return error_response("Task not found", code=404)
+
+    step = task.get_step(step_id)
+    if not step:
+        return error_response("Step not found", code=404)
+
+    if update_data.name is not None:
+        step.name = update_data.name
+    if update_data.description is not None:
+        step.description = update_data.description
+    if update_data.goal is not None:
+        step.goal = update_data.goal
+    if update_data.status is not None:
+        try:
+            step.status = StepStatus(update_data.status)
+        except ValueError:
+            return error_response(f"Invalid status: {update_data.status}")
+
+    step.updated_at = datetime.now()
+    task.updated_at = datetime.now()
+    return success_response(step.to_dict())
+
+
+@router.post("/{task_id}/execute")
+async def execute_task(
+    task_id: str,
+    service: TaskService = Depends(get_task_service)
+):
+    """Mark task as running and return first ready steps"""
+    task = service.get_task(task_id)
+    if not task:
+        return error_response("Task not found", code=404)
+
+    if task.status == TaskStatus.RUNNING:
+        return error_response("Task is already running")
+
+    # Validate graph
+    graph = service.build_graph(task_id)
+    if not graph:
+        return error_response("Failed to build graph")
+
+    is_valid, error = graph.validate()
+    if not is_valid:
+        return error_response(f"Invalid task graph: {error}")
+
+    # Update status to running
+    service.update_task_status(task_id, TaskStatus.RUNNING)
+
+    # Get ready steps
+    ready = graph.get_ready_steps()
+    if not ready:
+        if task.is_complete():
+            service.update_task_status(task_id, TaskStatus.COMPLETED)
+            return success_response({
+                "status": "completed",
+                "message": "All steps completed"
+            })
+        return error_response("No steps ready to execute")
+
+    return success_response({
+        "status": "running",
+        "ready_steps": [s.to_dict() for s in ready]
+    })
+
+
+@router.post("/{task_id}/complete")
+async def complete_task(
+    task_id: str,
+    service: TaskService = Depends(get_task_service)
+):
+    """Mark task as completed"""
+    task = service.get_task(task_id)
+    if not task:
+        return error_response("Task not found", code=404)
+
+    if task.is_complete():
+        service.update_task_status(task_id, TaskStatus.COMPLETED)
+        return success_response({
+            "status": "completed",
+            "message": "Task completed successfully"
+        })
+    
+    return error_response("Task is not yet complete")
+
+
+# Export/Import endpoints
+
+@router.get("/{task_id}/export")
+async def export_task(
+    task_id: str,
+    service: TaskService = Depends(get_task_service)
+):
+    """Export task as JSON"""
+    task_json = service.export_task(task_id)
+    if not task_json:
+        return error_response("Task not found", code=404)
+    return success_response({"json": task_json})
+
+
+@router.post("/import")
+async def import_task(
+    task_json: str,
+    service: TaskService = Depends(get_task_service)
+):
+    """Import task from JSON"""
+    task = service.import_task(task_json)
+    if not task:
+        return error_response("Failed to import task")
+    return success_response(task.to_dict(), "Task imported successfully")
+
diff --git a/luxx/services/task.py b/luxx/services/task.py
index 3043ffe..10b3e34 100644
--- a/luxx/services/task.py
+++ b/luxx/services/task.py
@@ -1,25 +1,27 @@
-"""Task module for autonomous task execution"""
+"""Task module for autonomous task execution and planning"""
 from dataclasses import dataclass, field
 from datetime import datetime
 from enum import Enum
 import logging
-from typing import List, Optional, Dict, Any
+from typing import List, Optional, Dict, Any, AsyncGenerator, Callable
+import json
+
 from luxx.utils.helpers import generate_id
 
 logger = logging.getLogger(__name__)
 
 
 class TaskStatus(Enum):
-    """Task status enum"""
-    PENDING = "pending"
-    READY = "ready"
-    RUNNING = "running"
-    BLOCK = "block"
-    TERMINATED = "terminated"
+    """Task status - 5-state model"""
+    PENDING = "pending"     # 创建/就绪
+    RUNNING = "running"     # 运行中
+    BLOCK = "block"         # 阻塞
+    TERMINATED = "terminated"  # 终止
+    COMPLETED = "completed" # 完成
 
 
 class StepStatus(Enum):
-    """Step status enum"""
+    """Step status"""
     PENDING = "pending"
     RUNNING = "running"
     COMPLETED = "completed"
@@ -33,29 +35,48 @@ class Step:
     id: str
     name: str
     description: str = ""
+    goal: str = ""                    # 步骤目标（用于 LLM 执行）
     depends_on: List[str] = field(default_factory=list)
     status: StepStatus = StepStatus.PENDING
     result: Optional[Dict[str, Any]] = None
+    tool_calls: List[Dict[str, Any]] = field(default_factory=list)
+    context: Dict[str, Any] = field(default_factory=dict)
     created_at: datetime = field(default_factory=datetime.now)
     updated_at: datetime = field(default_factory=datetime.now)
 
     def to_dict(self) -> Dict[str, Any]:
-        """Convert to dictionary"""
         return {
             "id": self.id,
             "name": self.name,
             "description": self.description,
+            "goal": self.goal,
             "depends_on": self.depends_on,
             "status": self.status.value,
             "result": self.result,
+            "tool_calls": self.tool_calls,
+            "context": self.context,
             "created_at": self.created_at.isoformat() if self.created_at else None,
             "updated_at": self.updated_at.isoformat() if self.updated_at else None
         }
 
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "Step":
+        return cls(
+            id=data.get("id", generate_id("step")),
+            name=data.get("name", ""),
+            description=data.get("description", ""),
+            goal=data.get("goal", ""),
+            depends_on=data.get("depends_on", []),
+            status=StepStatus(data.get("status", "pending")),
+            result=data.get("result"),
+            tool_calls=data.get("tool_calls", []),
+            context=data.get("context", {})
+        )
+
 
 @dataclass
 class Task:
-    """Task entity"""
+    """Task entity with built-in planning capability"""
     id: str
     name: str
     description: str = ""
@@ -64,11 +85,11 @@ class Task:
     steps: List[Step] = field(default_factory=list)
     subtasks: List["Task"] = field(default_factory=list)
     result: Optional[Dict[str, Any]] = None
+    metadata: Dict[str, Any] = field(default_factory=dict)
     created_at: datetime = field(default_factory=datetime.now)
     updated_at: datetime = field(default_factory=datetime.now)
 
     def to_dict(self) -> Dict[str, Any]:
-        """Convert to dictionary"""
         return {
             "id": self.id,
             "name": self.name,
@@ -78,10 +99,40 @@ class Task:
             "steps": [s.to_dict() for s in self.steps],
             "subtasks": [t.to_dict() for t in self.subtasks],
             "result": self.result,
+            "metadata": self.metadata,
             "created_at": self.created_at.isoformat() if self.created_at else None,
             "updated_at": self.updated_at.isoformat() if self.updated_at else None
         }
 
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "Task":
+        steps = [Step.from_dict(s) if isinstance(s, dict) else s for s in data.get("steps", [])]
+        subtasks = [cls.from_dict(t) if isinstance(t, dict) else t for t in data.get("subtasks", [])]
+        
+        return cls(
+            id=data.get("id", generate_id("task")),
+            name=data.get("name", ""),
+            description=data.get("description", ""),
+            goal=data.get("goal", ""),
+            status=TaskStatus(data.get("status", "pending")),
+            steps=steps,
+            subtasks=subtasks,
+            result=data.get("result"),
+            metadata=data.get("metadata", {})
+        )
+
+    def get_step(self, step_id: str) -> Optional[Step]:
+        for step in self.steps:
+            if step.id == step_id:
+                return step
+        return None
+
+    def get_completed_step_ids(self) -> List[str]:
+        return [s.id for s in self.steps if s.status == StepStatus.COMPLETED]
+
+    def is_complete(self) -> bool:
+        return all(s.status in (StepStatus.COMPLETED, StepStatus.SKIPPED) for s in self.steps)
+
 
 class TaskGraph:
     """Task graph for managing step dependencies"""
@@ -94,7 +145,6 @@ class TaskGraph:
         self._build_graph()
 
     def _build_graph(self) -> None:
-        """Build graph from task steps"""
         for step in self.task.steps:
             self._adjacency[step.id] = []
             self._reverse_adjacency[step.id] = []
@@ -108,7 +158,6 @@ class TaskGraph:
                     self._in_degree[step.id] += 1
 
     def topological_sort(self) -> List[Step]:
-        """Get steps in topological order"""
         in_degree = self._in_degree.copy()
         queue = [step_id for step_id, degree in in_degree.items() if degree == 0]
         result = []
@@ -126,14 +175,14 @@ class TaskGraph:
 
         return result
 
-    def get_ready_steps(self, completed_step_ids: List[str]) -> List[Step]:
-        """Get steps that are ready to execute"""
+    def get_ready_steps(self, completed_step_ids: List[str] = None) -> List[Step]:
+        if completed_step_ids is None:
+            completed_step_ids = self.task.get_completed_step_ids()
+        
         step_map = {step.id: step for step in self.task.steps}
         ready = []
 
         for step in self.task.steps:
-            if step.id in completed_step_ids:
-                continue
             if step.status != StepStatus.PENDING:
                 continue
             deps_completed = all(dep_id in completed_step_ids for dep_id in step.depends_on)
@@ -142,8 +191,30 @@ class TaskGraph:
 
         return ready
 
+    def get_parallel_levels(self) -> List[List[Step]]:
+        """Get steps grouped by dependency level for parallel execution"""
+        completed = set(self.task.get_completed_step_ids())
+        levels: List[List[Step]] = []
+        remaining = {s.id for s in self.task.steps if s.status == StepStatus.PENDING}
+        
+        while remaining:
+            current_level = []
+            for step_id in list(remaining):
+                step = self.task.get_step(step_id)
+                if step and all(dep_id in completed for dep_id in step.depends_on):
+                    current_level.append(step)
+            
+            if not current_level:
+                break
+            
+            levels.append(current_level)
+            for step in current_level:
+                remaining.remove(step.id)
+                completed.add(step.id)
+        
+        return levels
+
     def detect_cycles(self) -> List[List[str]]:
-        """Detect cycles in the graph"""
         WHITE, GRAY, BLACK = 0, 1, 2
         color = {step.id: WHITE for step in self.task.steps}
         cycles = []
@@ -172,7 +243,6 @@ class TaskGraph:
         return cycles
 
     def validate(self) -> tuple[bool, Optional[str]]:
-        """Validate the graph structure"""
         cycles = self.detect_cycles()
         if cycles:
             return False, f"Circular dependency detected: {cycles[0]}"
@@ -199,7 +269,6 @@ class TaskService:
         description: str = "",
         steps: List[Dict[str, Any]] = None
     ) -> Task:
-        """Create a new task"""
         task_id = generate_id("task")
         task = Task(
             id=task_id,
@@ -213,7 +282,8 @@ class TaskService:
                 step = Step(
                     id=generate_id("step"),
                     name=step_data.get("name", ""),
-                    description=step_data.get("description", "")
+                    description=step_data.get("description", ""),
+                    goal=step_data.get("goal", "")
                 )
                 task.steps.append(step)
 
@@ -222,11 +292,9 @@ class TaskService:
         return task
 
     def get_task(self, task_id: str) -> Optional[Task]:
-        """Get task by ID"""
         return self._tasks.get(task_id)
 
     def list_tasks(self) -> List[Task]:
-        """List all tasks"""
         return list(self._tasks.values())
 
     def update_task_status(
@@ -235,7 +303,6 @@ class TaskService:
         status: TaskStatus,
         result: Any = None
     ) -> Optional[Task]:
-        """Update task status"""
         task = self._tasks.get(task_id)
         if not task:
             return None
@@ -245,12 +312,32 @@ class TaskService:
         task.updated_at = datetime.now()
         return task
 
+    def update_step_status(
+        self,
+        task_id: str,
+        step_id: str,
+        status: StepStatus,
+        result: Any = None
+    ) -> Optional[Step]:
+        task = self._tasks.get(task_id)
+        if not task:
+            return None
+        
+        step = task.get_step(step_id)
+        if not step:
+            return None
+
+        step.status = status
+        step.result = result
+        step.updated_at = datetime.now()
+        task.updated_at = datetime.now()
+        return step
+
     def add_steps(
         self,
         task_id: str,
         steps: List[Dict[str, Any]]
     ) -> Optional[List[Step]]:
-        """Add steps to task"""
         task = self._tasks.get(task_id)
         if not task:
             return None
@@ -261,6 +348,7 @@ class TaskService:
                 id=generate_id("step"),
                 name=step_data.get("name", ""),
                 description=step_data.get("description", ""),
+                goal=step_data.get("goal", ""),
                 depends_on=step_data.get("depends_on", [])
             )
             task.steps.append(step)
@@ -269,8 +357,33 @@ class TaskService:
         task.updated_at = datetime.now()
         return result
 
+    def set_steps_from_plan(
+        self,
+        task_id: str,
+        plan_steps: List[Dict[str, Any]]
+    ) -> Optional[List[Step]]:
+        """Set steps from LLM-generated plan"""
+        task = self._tasks.get(task_id)
+        if not task:
+            return None
+
+        task.steps = []
+        for step_data in plan_steps:
+            step = Step(
+                id=generate_id("step"),
+                name=step_data.get("name", ""),
+                description=step_data.get("description", ""),
+                goal=step_data.get("goal", step_data.get("name", "")),
+                depends_on=step_data.get("depends_on", [])
+            )
+            task.steps.append(step)
+
+        task.metadata["raw_plan"] = plan_steps
+        task.updated_at = datetime.now()
+        logger.info(f"Set {len(task.steps)} steps for task: {task_id}")
+        return task.steps
+
     def delete_task(self, task_id: str) -> bool:
-        """Delete task"""
         if task_id not in self._tasks:
             return False
 
@@ -278,11 +391,159 @@ class TaskService:
         return True
 
     def build_graph(self, task_id: str) -> Optional[TaskGraph]:
-        """Build task graph for a task"""
         task = self._tasks.get(task_id)
         if not task:
             return None
         return TaskGraph(task)
 
+    def export_task(self, task_id: str) -> Optional[str]:
+        task = self._tasks.get(task_id)
+        if not task:
+            return None
+        return json.dumps(task.to_dict(), ensure_ascii=False, indent=2)
 
+    def import_task(self, task_json: str) -> Optional[Task]:
+        try:
+            data = json.loads(task_json)
+            task = Task.from_dict(data)
+            self._tasks[task.id] = task
+            return task
+        except Exception as e:
+            logger.error(f"Failed to import task: {e}")
+            return None
+
+
+class Planner:
+    """LLM-based task planner - generates steps from goal"""
+
+    PLANNER_SYSTEM_PROMPT = """You are a task planning assistant. Your job is to break down complex goals into clear, actionable steps.
+
+For a given goal, you should output a JSON array of steps. Each step should have:
+- name: Short, descriptive name
+- description: What this step does
+- depends_on: (optional) List of step indices this depends on
+
+Rules:
+1. Steps should be atomic and focused
+2. Consider dependencies between steps
+3. Order steps logically
+4. Use simple dependencies when needed
+
+Output ONLY valid JSON array, no other text."""
+
+    def __init__(self, llm_client):
+        self.llm = llm_client
+
+    async def plan(
+        self,
+        goal: str,
+        context: Dict[str, Any] = None
+    ) -> List[Dict[str, Any]]:
+        """Generate a plan (list of steps) for the given goal"""
+        user_message = f"Goal: {goal}\n\n"
+        
+        if context:
+            if context.get("available_tools"):
+                user_message += f"Available tools: {', '.join(context['available_tools'])}\n"
+            if context.get("constraints"):
+                user_message += f"Constraints: {context['constraints']}\n"
+
+        messages = [
+            {"role": "system", "content": self.PLANNER_SYSTEM_PROMPT},
+            {"role": "user", "content": user_message}
+        ]
+
+        try:
+            response = await self.llm.call(messages=messages)
+            
+            # Parse JSON response
+            plan_text = response.get("content", "")
+            # Extract JSON from response
+            start = plan_text.find("[")
+            end = plan_text.rfind("]") + 1
+            if start != -1 and end > start:
+                plan_json = plan_text[start:end]
+                steps = json.loads(plan_json)
+                
+                # Normalize steps - ensure depends_on uses indices
+                normalized_steps = []
+                for i, step in enumerate(steps):
+                    normalized = {
+                        "name": step.get("name", f"Step {i+1}"),
+                        "description": step.get("description", ""),
+                        "goal": step.get("goal", step.get("description", "")),
+                        "depends_on": []
+                    }
+                    
+                    # Convert index dependencies to step IDs (placeholder, will be fixed later)
+                    deps = step.get("depends_on", [])
+                    if isinstance(deps, list):
+                        normalized["depends_on"] = deps
+                    
+                    normalized_steps.append(normalized)
+                
+                return normalized_steps
+            
+            return []
+            
+        except Exception as e:
+            logger.error(f"Planning failed: {e}")
+            return []
+
+    def plan_sync(
+        self,
+        goal: str,
+        context: Dict[str, Any] = None
+    ) -> List[Dict[str, Any]]:
+        """Synchronous version of plan"""
+        raise NotImplementedError("Use async plan() method")
+
+
+class TaskRunner:
+    """Execute tasks using TaskGraph"""
+
+    def __init__(self, task_service: TaskService):
+        self.task_service = task_service
+
+    def run_step(
+        self,
+        task_id: str,
+        step_id: str
+    ) -> Optional[Step]:
+        """Mark step as running"""
+        return self.task_service.update_step_status(
+            task_id, step_id, StepStatus.RUNNING
+        )
+
+    def complete_step(
+        self,
+        task_id: str,
+        step_id: str,
+        result: Dict[str, Any] = None
+    ) -> Optional[Step]:
+        """Mark step as completed"""
+        return self.task_service.update_step_status(
+            task_id, step_id, StepStatus.COMPLETED, result
+        )
+
+    def fail_step(
+        self,
+        task_id: str,
+        step_id: str,
+        error: str = ""
+    ) -> Optional[Step]:
+        """Mark step as failed"""
+        return self.task_service.update_step_status(
+            task_id, step_id, StepStatus.FAILED, {"error": error}
+        )
+
+    def get_next_steps(self, task_id: str) -> List[Step]:
+        """Get steps ready to execute"""
+        graph = self.task_service.build_graph(task_id)
+        if not graph:
+            return []
+        return graph.get_ready_steps()
+
+
+# Global task service instance
 task_service = TaskService()
diff --git a/luxx/services/task_executor.py b/luxx/services/task_executor.py
new file mode 100644
index 0000000..43dd91f
--- /dev/null
+++ b/luxx/services/task_executor.py
@@ -0,0 +1,257 @@
+"""Task Executor Service - Integrates Task with Agent execution"""
+import logging
+import json
+from typing import List, Dict, Any, Optional, AsyncGenerator
+from datetime import datetime
+
+from luxx.services.task import (
+    task_service, TaskService, Task, Step, StepStatus, TaskStatus,
+    TaskGraph, Planner
+)
+from luxx.services.events import sse_event
+
+logger = logging.getLogger(__name__)
+
+
+class TaskExecutorService:
+    """Integrates Task/Plan with Agent execution loop"""
+
+    # System prompt for plan generation
+    PLANNER_SYSTEM = """You are a task planning assistant. Break down goals into clear, executable steps.
+    
+For each step, provide:
+- name: Short descriptive name
+- description: What this step accomplishes
+- goal: Specific objective for the agent to achieve (use this as the actual prompt)
+
+Consider dependencies between steps. Steps can run in parallel if they don't depend on each other.
+
+Output ONLY a JSON array, no other text."""
+
+    def __init__(
+        self,
+        task_svc: TaskService = None,
+        llm_client = None  # LLM client for plan generation
+    ):
+        self.task_service = task_svc or task_service
+        self.llm = llm_client
+
+    async def generate_plan(
+        self,
+        goal: str,
+        context: Dict[str, Any] = None
+    ) -> List[Dict[str, Any]]:
+        """Generate a plan from goal using LLM"""
+        if not self.llm:
+            logger.warning("No LLM client configured for planning")
+            return []
+
+        user_msg = f"Goal: {goal}\n\n"
+        if context:
+            if context.get("available_tools"):
+                user_msg += f"Available tools: {', '.join(context['available_tools'])}\n"
+            if context.get("constraints"):
+                user_msg += f"Constraints: {context['constraints']}\n"
+
+        messages = [
+            {"role": "system", "content": self.PLANNER_SYSTEM},
+            {"role": "user", "content": user_msg}
+        ]
+
+        try:
+            response = await self.llm.sync_call(messages=messages)
+            plan_text = response.get("content", "")
+
+            # Extract JSON
+            start = plan_text.find("[")
+            end = plan_text.rfind("]") + 1
+            if start != -1 and end > start:
+                plan_json = plan_text[start:end]
+                steps = json.loads(plan_json)
+
+                # Normalize steps
+                normalized = []
+                for i, step in enumerate(steps):
+                    normalized.append({
+                        "name": step.get("name", f"Step {i+1}"),
+                        "description": step.get("description", ""),
+                        "goal": step.get("goal", step.get("description", "")),
+                        "depends_on": step.get("depends_on", [])
+                    })
+                return normalized
+
+        except Exception as e:
+            logger.error(f"Plan generation failed: {e}")
+
+        return []
+
+    async def create_task_with_plan(
+        self,
+        name: str,
+        goal: str,
+        description: str = "",
+        auto_plan: bool = True,
+        context: Dict[str, Any] = None
+    ) -> Task:
+        """Create a task and optionally auto-generate plan"""
+        task = self.task_service.create_task(
+            name=name,
+            goal=goal,
+            description=description
+        )
+
+        if auto_plan:
+            # Generate plan using LLM
+            steps = await self.generate_plan(goal, context)
+            if steps:
+                self.task_service.set_steps_from_plan(task.id, steps)
+                logger.info(f"Generated {len(steps)} steps for task {task.id}")
+            else:
+                # Fallback: single step with the goal
+                self.task_service.add_steps(task.id, [{
+                    "name": "Execute Goal",
+                    "description": description,
+                    "goal": goal
+                }])
+
+        return self.task_service.get_task(task.id)
+
+    def get_execution_plan(self, task_id: str) -> Dict[str, Any]:
+        """Get execution plan for a task"""
+        task = self.task_service.get_task(task_id)
+        if not task:
+            return {"error": "Task not found"}
+
+        graph = self.task_service.build_graph(task_id)
+        if not graph:
+            return {"error": "Failed to build graph"}
+
+        sorted_steps = graph.topological_sort()
+        parallel_levels = graph.get_parallel_levels()
+        ready_steps = graph.get_ready_steps()
+
+        return {
+            "task_id": task_id,
+            "task_name": task.name,
+            "goal": task.goal,
+            "total_steps": len(task.steps),
+            "completed_steps": len(task.get_completed_step_ids()),
+            "execution_order": [s.id for s in sorted_steps],
+            "parallel_levels": [[s.id for s in level] for level in parallel_levels],
+            "ready_steps": [s.id for s in ready_steps],
+            "can_execute": len(ready_steps) > 0,
+            "steps": [s.to_dict() for s in task.steps]
+        }
+
+
+class AgentWithPlan:
+    """Agent execution with built-in task planning"""
+
+    def __init__(
+        self,
+        chat_service,
+        task_executor: TaskExecutorService = None
+    ):
+        self.chat_service = chat_service
+        self.task_executor = task_executor or TaskExecutorService()
+
+    async def execute_with_plan(
+        self,
+        goal: str,
+        context: Dict[str, Any]
+    ) -> AsyncGenerator[Dict[str, Any], None]:
+        """Execute goal with automatic planning"""
+        # 1. Create task and generate plan
+        yield sse_event("planning", {"status": "generating", "goal": goal})
+
+        task = await self.task_executor.create_task_with_plan(
+            name=f"Task: {goal[:50]}...",
+            goal=goal,
+            context=context
+        )
+
+        if not task.steps:
+            yield sse_event("error", {"content": "Failed to generate plan"})
+            return
+
+        yield sse_event("plan_created", {
+            "task_id": task.id,
+            "steps": [s.to_dict() for s in task.steps]
+        })
+
+        # 2. Execute steps
+        graph = self.task_executor.task_service.build_graph(task.id)
+        if not graph:
+            yield sse_event("error", {"content": "Failed to build execution graph"})
+            return
+
+        # Get execution order
+        sorted_steps = graph.topological_sort()
+        completed_ids = []
+
+        for step in sorted_steps:
+            yield sse_event("step_start", {
+                "task_id": task.id,
+                "step_id": step.id,
+                "step_name": step.name,
+                "step_goal": step.goal,
+                "progress": len(completed_ids) + 1,
+                "total": len(sorted_steps)
+            })
+
+            # 3. Execute step with Agent
+            # For now, execute as a chat message
+            step_messages = [
+                {"role": "system", "content": f"You are executing step: {step.name}\n\n{step.goal}"},
+                {"role": "user", "content": f"Execute: {step.goal}"}
+            ]
+
+            step_result = {"messages": [], "tool_calls": []}
+
+            # Stream response from agent
+            async for event in self._stream_step(step_messages, context):
+                if event.startswith("data: "):
+                    data = event[6:]
+                    yield event
+                    # Collect results
+                    if "step_complete" in data or "done" in data:
+                        step_result["completed"] = True
+
+            # Mark step complete
+            self.task_executor.task_service.update_step_status(
+                task.id, step.id, StepStatus.COMPLETED,
+                {"result": step_result}
+            )
+            completed_ids.append(step.id)
+
+            yield sse_event("step_complete", {
+                "task_id": task.id,
+                "step_id": step.id,
+                "progress": len(completed_ids),
+                "total": len(sorted_steps)
+            })
+
+        # 4. Task complete
+        self.task_executor.task_service.update_task_status(
+            task.id, TaskStatus.COMPLETED
+        )
+
+        yield sse_event("task_complete", {
+            "task_id": task.id,
+            "completed_steps": len(completed_ids)
+        })
+
+    async def _stream_step(
+        self,
+        messages: List[Dict],
+        context: Dict[str, Any]
+    ) -> AsyncGenerator[str, None]:
+        """Stream response for a step (placeholder - integrate with ChatService)"""
+        # TODO: Integrate with actual ChatService streaming
+        yield sse_event("info", {"content": "Step execution not yet integrated"})
+
+
+# Factory function
+def create_task_executor(llm_client=None) -> TaskExecutorService:
+    """Create TaskExecutorService with optional LLM client"""
+    return TaskExecutorService(llm_client=llm_client)