From acdaebc7e9dc55ac9f731b26737ef37cdc566df3 Mon Sep 17 00:00:00 2001 From: Micro66 Date: Sun, 11 Jan 2026 08:33:07 +0800 Subject: [PATCH 1/3] fix(executor): use correct workspace path for attachment downloads Change attachment download path from project_path to WORKSPACE_ROOT/task_id to ensure attachments are saved to the correct location. Expected: /workspace/{task_id}/{task_id}:executor:attachments/{subtask_id}/ Previous: /workspace/{task_id}/{repo_name}/{task_id}:executor:attachments/{subtask_id}/ --- executor/agents/claude_code/claude_code_agent.py | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/executor/agents/claude_code/claude_code_agent.py b/executor/agents/claude_code/claude_code_agent.py index ee40919e170..182915eb22a 100644 --- a/executor/agents/claude_code/claude_code_agent.py +++ b/executor/agents/claude_code/claude_code_agent.py @@ -1383,10 +1383,11 @@ def _download_attachments(self) -> None: logger.warning("No auth token available, cannot download attachments") return - # Determine workspace path - workspace = self.project_path or os.path.join( - config.WORKSPACE_ROOT, str(self.task_id) - ) + # Determine workspace path for attachments + # Attachments should be stored in WORKSPACE_ROOT/task_id, not in the project path + # This ensures attachments are accessible at /workspace/{task_id}:executor:attachments/... + # instead of /workspace/{task_id}/{repo_name}/{task_id}:executor:attachments/... + workspace = os.path.join(config.WORKSPACE_ROOT, str(self.task_id)) # Import and use attachment downloader from executor.services.attachment_downloader import \ From 1e6127dae658d580f5ad54875b5b5591a94faf71 Mon Sep 17 00:00:00 2001 From: Micro66 Date: Sun, 11 Jan 2026 12:33:56 +0800 Subject: [PATCH 2/3] fix(executor): clean up task state after cancel to prevent "Request interrupted" error When a task is cancelled, TaskStateManager stores the CANCELLED state. If the state is not cleaned up, the next message sent by the user will check is_cancelled() which returns true, causing the agent to return immediately with "Request interrupted by user" error. This fix adds task state cleanup in send_cancel_callback_async() after the cancel callback is sent, ensuring subsequent messages can be processed normally. Changes: - Import TaskStateManager in agent_service.py - Call task_state_manager.cleanup(task_id) after cancel callback is sent - Add cleanup in error handler as well to ensure state is cleaned up --- executor/services/agent_service.py | 186 +++++++++++++++++++++-------- 1 file changed, 137 insertions(+), 49 deletions(-) diff --git a/executor/services/agent_service.py b/executor/services/agent_service.py index 38897c80215..eadbed24c09 100644 --- a/executor/services/agent_service.py +++ b/executor/services/agent_service.py @@ -9,18 +9,21 @@ import asyncio import threading import time -from typing import Dict, Optional, Any, Tuple, List, ClassVar from dataclasses import dataclass +from typing import Any, ClassVar, Dict, List, Optional, Tuple from shared.logger import setup_logger from shared.status import TaskStatus + from executor.agents import Agent, AgentFactory -from executor.agents.claude_code.claude_code_agent import ClaudeCodeAgent from executor.agents.agno.agno_agent import AgnoAgent +from executor.agents.claude_code.claude_code_agent import ClaudeCodeAgent from executor.callback.callback_handler import send_status_callback +from executor.tasks.task_state_manager import TaskStateManager logger = setup_logger("agent_service") + def _format_task_log(task_id, subtask_id): return f"task_id: {task_id}.{subtask_id}" @@ -32,7 +35,7 @@ class AgentSession: class AgentService: - _instance: ClassVar[Optional['AgentService']] = None + _instance: ClassVar[Optional["AgentService"]] = None _lock: ClassVar[threading.Lock] = threading.Lock() def __new__(cls): @@ -57,81 +60,113 @@ def create_agent(self, task_data: Dict[str, Any]) -> Optional[Agent]: logger.info(f"task_id: [{task_id}] Creating agent") - if (existing_agent := self.get_agent(f"{task_id}")): - logger.info(f"[{_format_task_log(task_id, subtask_id)}] Reusing existing agent") + if existing_agent := self.get_agent(f"{task_id}"): + logger.info( + f"[{_format_task_log(task_id, subtask_id)}] Reusing existing agent" + ) return existing_agent try: # Determine agent type based on task type task_type = task_data.get("type", "") - + if task_type == "validation": # For validation tasks, use ImageValidatorAgent shell_type = "imagevalidator" - logger.info(f"[{_format_task_log(task_id, subtask_id)}] Validation task detected, using ImageValidatorAgent") + logger.info( + f"[{_format_task_log(task_id, subtask_id)}] Validation task detected, using ImageValidatorAgent" + ) else: # For regular tasks, get shell_type from bot config bot_config = task_data.get("bot") if isinstance(bot_config, list): - shell_type = bot_config[0].get("shell_type", "").strip().lower() if bot_config else "" + shell_type = ( + bot_config[0].get("shell_type", "").strip().lower() + if bot_config + else "" + ) elif isinstance(bot_config, dict): shell_type = bot_config.get("shell_type", "").strip().lower() else: shell_type = "" - logger.info(f"[{_format_task_log(task_id, subtask_id)}] Creating new agent '{shell_type}'") + logger.info( + f"[{_format_task_log(task_id, subtask_id)}] Creating new agent '{shell_type}'" + ) agent = AgentFactory.get_agent(shell_type, task_data) if not agent: - logger.error(f"[{_format_task_log(task_id, subtask_id)}] Failed to create agent") + logger.error( + f"[{_format_task_log(task_id, subtask_id)}] Failed to create agent" + ) return None - + init_status = agent.initialize() if init_status != TaskStatus.SUCCESS: - logger.error(f"[{_format_task_log(task_id, subtask_id)}] Failed to initialize agent: {init_status}") + logger.error( + f"[{_format_task_log(task_id, subtask_id)}] Failed to initialize agent: {init_status}" + ) return None - self._agent_sessions[task_id] = AgentSession(agent=agent, created_at=time.time()) + self._agent_sessions[task_id] = AgentSession( + agent=agent, created_at=time.time() + ) logger.info(f"task_id: [{task_id}] Agent created") return agent except Exception as e: - logger.exception(f"[{_format_task_log(task_id, subtask_id)}] Exception during agent creation: {e}") + logger.exception( + f"[{_format_task_log(task_id, subtask_id)}] Exception during agent creation: {e}" + ) return None - def execute_agent_task(self, agent: Agent, pre_executed: Optional[TaskStatus] = None) -> Tuple[TaskStatus, Optional[str]]: + def execute_agent_task( + self, agent: Agent, pre_executed: Optional[TaskStatus] = None + ) -> Tuple[TaskStatus, Optional[str]]: try: - logger.info(f"[{agent.get_name()}][{_format_task_log(agent.task_id, agent.subtask_id)}] Executing with pre_executed={pre_executed}") + logger.info( + f"[{agent.get_name()}][{_format_task_log(agent.task_id, agent.subtask_id)}] Executing with pre_executed={pre_executed}" + ) return agent.handle(pre_executed) except Exception as e: - logger.exception(f"[{agent.get_name()}][{_format_task_log(agent.task_id, agent.subtask_id)}] Execution error: {e}") + logger.exception( + f"[{agent.get_name()}][{_format_task_log(agent.task_id, agent.subtask_id)}] Execution error: {e}" + ) return TaskStatus.FAILED, str(e) - def execute_task(self, task_data: Dict[str, Any]) -> Tuple[TaskStatus, Optional[str]]: + def execute_task( + self, task_data: Dict[str, Any] + ) -> Tuple[TaskStatus, Optional[str]]: task_id = task_data.get("task_id", -1) subtask_id = task_data.get("subtask_id", -1) try: agent = self.get_agent(f"{task_id}") # If agent exists, update prompt - if agent and hasattr(agent, 'update_prompt') and "prompt" in task_data: + if agent and hasattr(agent, "update_prompt") and "prompt" in task_data: new_prompt = task_data.get("prompt", "") - logger.info(f"[{_format_task_log(task_id, subtask_id)}] Updating prompt for existing agent") + logger.info( + f"[{_format_task_log(task_id, subtask_id)}] Updating prompt for existing agent" + ) agent.update_prompt(new_prompt) # If agent doesn't exist, create new agent elif not agent: agent = self.create_agent(task_data) - + if not agent: msg = f"[{_format_task_log(task_id, subtask_id)}] Unable to get or create agent" logger.error(msg) return TaskStatus.FAILED, msg return self.execute_agent_task(agent) except Exception as e: - logger.exception(f"[{_format_task_log(task_id, subtask_id)}] Task execution error: {e}") + logger.exception( + f"[{_format_task_log(task_id, subtask_id)}] Task execution error: {e}" + ) return TaskStatus.FAILED, str(e) - async def _close_agent_session(self, task_id: str, agent: Agent) -> Tuple[TaskStatus, Optional[str]]: + async def _close_agent_session( + self, task_id: str, agent: Agent + ) -> Tuple[TaskStatus, Optional[str]]: try: agent_name = agent.get_name() if agent_name == "ClaudeCodeAgent": @@ -142,20 +177,30 @@ async def _close_agent_session(self, task_id: str, agent: Agent) -> Tuple[TaskSt logger.info(f"[{_format_task_log(task_id, -1)}] Closed Agno client") return TaskStatus.SUCCESS, None except Exception as e: - logger.exception(f"[{_format_task_log(task_id, -1)}] Error closing agent: {e}") + logger.exception( + f"[{_format_task_log(task_id, -1)}] Error closing agent: {e}" + ) return TaskStatus.FAILED, str(e) - async def delete_session_async(self, task_id: str) -> Tuple[TaskStatus, Optional[str]]: + async def delete_session_async( + self, task_id: str + ) -> Tuple[TaskStatus, Optional[str]]: session = self._agent_sessions.get(task_id) if not session: - return TaskStatus.FAILED, f"[{_format_task_log(task_id, -1)}] No session found" + return ( + TaskStatus.FAILED, + f"[{_format_task_log(task_id, -1)}] No session found", + ) try: status, error_msg = await self._close_agent_session(task_id, session.agent) if status != TaskStatus.SUCCESS: return status, error_msg del self._agent_sessions[task_id] - return TaskStatus.SUCCESS, f"[{_format_task_log(task_id, -1)}] Session deleted" + return ( + TaskStatus.SUCCESS, + f"[{_format_task_log(task_id, -1)}] Session deleted", + ) except Exception as e: logger.exception(f"[{task_id}] Error deleting session: {e}") return TaskStatus.FAILED, str(e) @@ -186,22 +231,38 @@ def cancel_task(self, task_id: int) -> Tuple[TaskStatus, Optional[str]]: logger.info(f"task_id: [{task_id}] Cancelling task") session = self._agent_sessions.get(task_id) if not session: - return TaskStatus.FAILED, f"[{_format_task_log(task_id, -1)}] No session found" + return ( + TaskStatus.FAILED, + f"[{_format_task_log(task_id, -1)}] No session found", + ) try: agent = session.agent agent_name = agent.get_name() - if hasattr(agent, 'cancel_run'): + if hasattr(agent, "cancel_run"): success = agent.cancel_run() if success: - logger.info(f"[{_format_task_log(task_id, -1)}] Successfully cancelled {agent_name} task") - return TaskStatus.SUCCESS, f"[{_format_task_log(task_id, -1)}] Task cancelled" + logger.info( + f"[{_format_task_log(task_id, -1)}] Successfully cancelled {agent_name} task" + ) + return ( + TaskStatus.SUCCESS, + f"[{_format_task_log(task_id, -1)}] Task cancelled", + ) else: - logger.warning(f"[{_format_task_log(task_id, -1)}] Failed to cancel {agent_name} task") - return TaskStatus.FAILED, f"[{_format_task_log(task_id, -1)}] Cancel failed" + logger.warning( + f"[{_format_task_log(task_id, -1)}] Failed to cancel {agent_name} task" + ) + return ( + TaskStatus.FAILED, + f"[{_format_task_log(task_id, -1)}] Cancel failed", + ) else: - return TaskStatus.FAILED, f"[{_format_task_log(task_id, -1)}] {agent_name} agent does not support cancellation" + return ( + TaskStatus.FAILED, + f"[{_format_task_log(task_id, -1)}] {agent_name} agent does not support cancellation", + ) except Exception as e: logger.exception(f"[{task_id}] Error cancelling task: {e}") @@ -211,26 +272,30 @@ async def send_cancel_callback_async(self, task_id: int) -> None: """ Asynchronously send cancel task callback This method is called in a background task and will not block the cancel API response - + Args: task_id: Task ID """ try: session = self._agent_sessions.get(task_id) if not session: - logger.warning(f"[{_format_task_log(task_id, -1)}] No session found for sending cancel callback") + logger.warning( + f"[{_format_task_log(task_id, -1)}] No session found for sending cancel callback" + ) return - + agent = session.agent - task_data = getattr(agent, 'task_data', {}) - + task_data = getattr(agent, "task_data", {}) + # Get task information subtask_id = task_data.get("subtask_id", -1) task_title = task_data.get("task_title", "") subtask_title = task_data.get("subtask_title", "") - - logger.info(f"[{_format_task_log(task_id, subtask_id)}] Sending cancel callback asynchronously") - + + logger.info( + f"[{_format_task_log(task_id, subtask_id)}] Sending cancel callback asynchronously" + ) + # Send CANCELLED status callback (not COMPLETED) result = send_status_callback( task_id=task_id, @@ -239,16 +304,39 @@ async def send_cancel_callback_async(self, task_id: int) -> None: subtask_title=subtask_title, status=TaskStatus.CANCELLED.value, message="${{tasks.cancel_task}}", - progress=100 + progress=100, ) - + if result and result.get("status") == TaskStatus.SUCCESS.value: - logger.info(f"[{_format_task_log(task_id, subtask_id)}] Cancel callback sent successfully") + logger.info( + f"[{_format_task_log(task_id, subtask_id)}] Cancel callback sent successfully" + ) else: - logger.error(f"[{_format_task_log(task_id, subtask_id)}] Failed to send cancel callback: {result}") - + logger.error( + f"[{_format_task_log(task_id, subtask_id)}] Failed to send cancel callback: {result}" + ) + + # Clean up task state after cancel callback is sent + # This allows next message to be processed without "Request interrupted" error + task_state_manager = TaskStateManager() + task_state_manager.cleanup(task_id) + logger.info( + f"[{_format_task_log(task_id, subtask_id)}] Cleaned up task state after cancel" + ) + except Exception as e: logger.exception(f"[{task_id}] Error sending cancel callback: {e}") + # Still attempt to cleanup task state on error + try: + task_state_manager = TaskStateManager() + task_state_manager.cleanup(task_id) + logger.info( + f"[{_format_task_log(task_id, -1)}] Cleaned up task state after cancel error" + ) + except Exception as cleanup_error: + logger.warning( + f"[{task_id}] Failed to cleanup task state: {cleanup_error}" + ) def list_sessions(self) -> List[Dict[str, Any]]: return [ @@ -256,7 +344,7 @@ def list_sessions(self) -> List[Dict[str, Any]]: "task_id": task_id, "agent_type": session.agent.get_name(), "pre_executed": session.agent.pre_executed, - "created_at": session.created_at + "created_at": session.created_at, } for task_id, session in self._agent_sessions.items() ] @@ -307,4 +395,4 @@ async def close_all_agent_sessions(self) -> Tuple[TaskStatus, str, Dict[str, str return TaskStatus.SUCCESS, "All agent sessions closed successfully", {} else: message = f"Some agents failed to close: {', '.join(errors)}; Successful: {', '.join(results) or 'None'}" - return TaskStatus.FAILED, message, error_detail \ No newline at end of file + return TaskStatus.FAILED, message, error_detail From 80ded67bb82b01f9a5d6b7c03057a7f914693955 Mon Sep 17 00:00:00 2001 From: hongyu9 Date: Mon, 12 Jan 2026 11:53:05 +0800 Subject: [PATCH 3/3] =?UTF-8?q?fix:=20=E4=BF=AE=E6=94=B9=E5=90=AF=E5=8A=A8?= =?UTF-8?q?=E8=84=9A=E6=9C=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- start.sh | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/start.sh b/start.sh index dea030fea4d..dba4ad44be4 100755 --- a/start.sh +++ b/start.sh @@ -770,8 +770,10 @@ start_services() { "export CHAT_SHELL_MODE=http && export CHAT_SHELL_STORAGE_TYPE=remote && export CHAT_SHELL_REMOTE_STORAGE_URL=http://localhost:8000/api/internal && source .venv/bin/activate && .venv/bin/python -m uvicorn chat_shell.main:app --reload --host 0.0.0.0 --port 8100" # 3. Start Executor Manager + # TASK_API_DOMAIN uses local IP so docker containers can access the backend + # DOCKER_HOST_ADDR=localhost so executor_manager can access docker containers start_service "executor_manager" "executor_manager" \ - "export EXECUTOR_IMAGE=$EXECUTOR_IMAGE && export TASK_API_DOMAIN=http://localhost:8000 && export NETWORK=wegent-network && source .venv/bin/activate && uvicorn main:app --reload --host 0.0.0.0 --port 8001" + "export EXECUTOR_IMAGE=$EXECUTOR_IMAGE && export TASK_API_DOMAIN=http://$(get_local_ip):8000 && export DOCKER_HOST_ADDR=localhost && export NETWORK=wegent-network && source .venv/bin/activate && uvicorn main:app --reload --host 0.0.0.0 --port 8001" # 4. Start Frontend (run in background) echo -e " Starting ${BLUE}frontend${NC}..."