Repository navigation
Fix/cancel task error - #517
Conversation
📝 WalkthroughWalkthroughThis PR refactors the Claude agent executor to support task interruption and resume workflows. It introduces a new INTERRUPTED task state, replaces complex retry-with-resume logic with simpler cancellation-aware handling, adds unified thread-safe state management, and makes numerous formatting adjustments across backend services. Changes
Estimated code review effort🎯 4 (Complex) | ⏱️ ~55 minutes Possibly related PRs
Suggested reviewers
Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 inconclusive)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing touches
🧪 Generate unit tests (beta)
No actionable comments were generated in the recent review. 🎉 Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
dc29d5c to
0d216f5
Compare
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 6
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (3)
executor/services/agent_service.py (1)
19-19:⚠️ Potential issue | 🟡 MinorRemove unused
TaskStateManagerimport: this is a leftover from the removed cleanup logic.The import on line 19 is never referenced anywhere in the file and should be removed.
Proposed fix
-from executor.tasks.task_state_manager import TaskStateManagerexecutor/modes/local/runner.py (1)
528-538:⚠️ Potential issue | 🟡 MinorMissing new logger names in
_cleanup_file_logging.Lines 503-504 add
"progress_state_manager"and"claude_response_processor"to_setup_file_logging, but_cleanup_file_logging(lines 528-538) doesn't include them. On shutdown the file handler won't be removed from those loggers, which can cause write-after-close errors or handler leaks.🐛 Proposed fix
logger_names = [ "local_runner", "websocket_client", "local_heartbeat", "local_handlers", "websocket_progress_reporter", "task_executor", "executor.config.config_loader", "claude_code_agent", "executor.services.attachment_downloader", + "progress_state_manager", + "claude_response_processor", ]executor/agents/claude_code/session_manager.py (1)
399-404:⚠️ Potential issue | 🟡 MinorMissing type hint for
task_state_managerparameter.Per coding guidelines, type hints are required in Python code. The parameter should be typed.
Proposed fix
+from __future__ import annotations +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from executor.tasks.task_state_manager import TaskStateManager + def resolve_session_id( task_id: int, bot_id: Optional[int] = None, new_session: bool = False, - task_state_manager=None, + task_state_manager: Optional["TaskStateManager"] = None, ) -> tuple[str, str]:If circular imports are a concern, the
TYPE_CHECKINGguard keeps it import-time safe.As per coding guidelines, "Type hints are required in Python code."
🤖 Fix all issues with AI agents
In `@executor/agents/claude_code/progress_state_manager.py`:
- Line 99: The parameter annotation in update_workbench_status uses an implicit
Optional (result_value: str = None); update its type to an explicit Optional
type (e.g., result_value: Optional[str] or result_value: str | None) and add the
corresponding import (from typing import Optional) if necessary, ensuring the
function signature and any callers remain type-correct.
In `@executor/agents/claude_code/response_processor.py`:
- Around line 362-388: The local flag saw_sdk_interrupt_messages set inside
_handle_user_message (when handling UserMessage text/ToolResultBlock) never gets
propagated to the caller, so change _handle_user_message to return that boolean
(update its signature and return value), update the caller process_response to
capture the returned saw_sdk_interrupt_messages, and pass that updated flag into
the subsequent call to _process_result_message; also update any other call sites
of _handle_user_message to handle the new return value and remove the unused
local assignment (fixing the F841 static analysis warning).
- Around line 859-908: Initialize task_id = None before the interruption/resume
logic so it is always defined (e.g., set task_id = None just before computing
is_resuming_from_interruption). In the block that checks task_state_manager and
state_manager use state_manager.task_data.get("task_id") to set task_id as
needed, but remove the later redundant reassignments of task_id (the one inside
the is_resuming_from_interruption branch and any duplicate before logging).
Ensure subsequent references (logs and task_state_manager.set_interrupted) use
the now-initialized task_id and guard actions when task_id is None; the relevant
symbols are is_resuming_from_interruption, task_state_manager, state_manager,
task_id, cancellation_in_progress, and TaskStatus.
- Around line 896-908: The cancellation branch handles internal interruption via
task_state_manager.set_interrupted(task_id) but fails to notify the frontend;
before returning TaskStatus.CANCELLED when cancellation_in_progress is True,
call state reporting the terminal status and progress: invoke
task_state_manager.set_task_status(TaskStatus.CANCELLED.value) (or
state_manager.set_task_status if that symbol is used in this file) and
task_state_manager.report_progress(task_id, progress=100,
status=TaskStatus.CANCELLED.value) (or the existing report_progress signature
used elsewhere) so the frontend receives the CANCELLED terminal state, then
return TaskStatus.CANCELLED; keep the existing set_interrupted call to preserve
internal session.
- Around line 878-887: The code path that handles is_resuming_from_interruption
currently returns TaskStatus.SUCCESS which incorrectly signals completion;
change it to return a status that indicates continuation (e.g.,
TaskStatus.RUNNING or a dedicated RESUMING status) so callers like
HeartbeatService and claude_code_agent.py don't treat initialization as
finished. Locate the block checking is_resuming_from_interruption in
response_processor.py (uses state_manager, task_id, logger) and replace the
return value TaskStatus.SUCCESS with the chosen continuation status from the
TaskStatus enum in shared/status.py, and update any logging text to reflect
"resuming/continuing" rather than "SUCCESS".
In `@executor/services/agent_service.py`:
- Around line 318-323: The comment points out that cancelled/failed tasks leave
entries in _agent_sessions because response_processor only calls
task_state_manager.cleanup() and does not remove sessions; fix by making
response_processor (the function handling cancelled tasks) also call the agent
session cleanup path (e.g., invoke delete_session_async(task_id) or remove the
session from the _agent_sessions mapping) whenever a task is cleaned up, or
alternatively add a timeout-based reaper that periodically scans _agent_sessions
and removes sessions that have been idle beyond a threshold; ensure the fix
touches response_processor and uses the existing delete_session_async /
close_all_agent_sessions APIs (or a new remove_session_by_task_id helper) and
ties cleanup to task_id so stale sessions are reliably removed.
🧹 Nitpick comments (11)
chat_shell/tests/tools/test_knowledge_base_abc.py (1)
72-75: The first condition is logically redundant.
"abstract method"being present implies"abstract"is present, so theoralways reduces to just the second branch. You could simplify to:assert "abstract" in str(exc_info.value).lower()That said, the overall change to make the assertion more robust across Python versions is a good idea.
chat_shell/chat_shell/tools/builtin/knowledge_base.py (1)
21-21: Import relocation looks fine.No semantic change. Worth noting: this file is ~1,571 lines, which exceeds the 1,000-line guideline. Consider splitting it into sub-modules (e.g., separating persistence logic, HTTP helpers, and injection strategy coordination) in a follow-up. As per coding guidelines, "File size limit: If a file exceeds 1000 lines, split it into multiple sub-modules".
executor/services/agent_service.py (1)
321-322: Redundant exception object inlogging.exceptioncall.
logger.exceptionautomatically includes the traceback; passingein the f-string is redundant.Proposed fix
- logger.exception(f"[{task_id}] Error sending cancel callback: {e}") + logger.exception(f"[{task_id}] Error sending cancel callback")executor/executor.spec (1)
74-84: Redundant explicit imports aftercollect_submodules.Line 75 already calls
collect_submodules('executor.wecode'), which recursively collects all submodules. The explicit list on lines 78-84 duplicates those entries. This is harmless (PyInstaller deduplicates), but you could remove the explicit list ifcollect_submodulesis reliable for these modules.executor/tasks/task_state_manager.py (2)
107-121: Inconsistent log format and duplicated logic withset_state.Two observations:
Line 120 logs
"TaskState.INTERRUPTED"as a literal string, whereasset_state(line 75) logs the actual{state}enum value (which renders asTaskState.INTERRUPTED). Functionally equivalent here, but if the enum value ever changes, this hardcoded string won't follow.
set_interruptedlargely duplicatesset_statelogic. Consider delegating:♻️ Reduce duplication by delegating to set_state
def set_interrupted(self, task_id: int) -> None: """ Mark task as interrupted (resumable) Args: task_id: Task ID """ with self._state_lock: - old_state = self._states.get(task_id) - self._states[task_id] = TaskState.INTERRUPTED if task_id not in self._cancel_timestamps: self._cancel_timestamps[task_id] = datetime.now() - logger.info( - f"Task {task_id} state changed: {old_state} -> TaskState.INTERRUPTED" - ) + self.set_state(task_id, TaskState.INTERRUPTED)Note: This requires reordering the lock acquisition since
set_statealso acquires_state_lock. You'd need to set the timestamp first (under lock) then callset_state, or fold the timestamp logic intoset_statefor the INTERRUPTED case.
90-105:is_cancelledreturningTrueforINTERRUPTEDmay cause confusion.
INTERRUPTEDis documented as "resumable" (line 29), butis_cancelled()returningTruefor it meansshould_continue()returnsFalse, effectively halting the task. This is likely intentional for the current execution loop, but the naming is misleading — callers may not expect an "interrupted but resumable" task to report as "cancelled."Consider either renaming
is_cancelledto something likeis_stoppedor adding a separateis_interruptedcheck for callers that need to distinguish the two.executor/agents/claude_code/session_manager.py (1)
438-456: Verbose logging branches can be simplified.The "has cache, using cached session_id" log message is duplicated across two branches (lines 450-452 and 454-456). Consider collapsing:
♻️ Simplified logging
else: # Has cache + new_session=False -> use cached session_id session_id = cached_session_id - # Check if resuming from interruption - if task_state_manager: - task_state = task_state_manager.get_state(task_id) - if task_state == TaskState.INTERRUPTED: - logger.info( - f"Resuming interrupted session {session_id} for task {task_id}" - ) - else: - logger.info( - f"Has cache, using cached session_id {session_id} (bot_id={bot_id})" - ) - else: + if ( + task_state_manager + and task_state_manager.get_state(task_id) == TaskState.INTERRUPTED + ): logger.info( - f"Has cache, using cached session_id {session_id} (bot_id={bot_id})" + f"Resuming interrupted session {session_id} for task {task_id}" ) + else: + logger.info( + f"Has cache, using cached session_id {session_id} (bot_id={bot_id})" + )executor/agents/claude_code/progress_state_manager.py (1)
171-174:mark_cancellingname is misleading — it setsCANCELLED, notCANCELLING.The method name suggests an intermediate "cancelling" state, but it calls
set_task_status(TaskStatus.CANCELLED.value), which is the terminal state. This could confuse callers who expect aCANCELLING → CANCELLEDtransition. Consider renaming tomark_cancelledto match the actual behavior.♻️ Proposed rename
- def mark_cancelling(self) -> None: - """Mark that cancellation is in progress""" - self.set_task_status(TaskStatus.CANCELLED.value) - logger.info("Task marked as cancelled") + def mark_cancelled(self) -> None: + """Mark that the task has been cancelled""" + self.set_task_status(TaskStatus.CANCELLED.value) + logger.info("Task marked as cancelled")executor/agents/claude_code/response_processor.py (3)
113-120: Nit: f-string without placeholders on line 115.Static analysis (Ruff F541) flags this. Use a plain string instead.
Fix
- logger.info( - f"Detected SDK interrupt message - marking cancellation_in_progress=True" - ) + logger.info( + "Detected SDK interrupt message - marking cancellation_in_progress=True" + )
208-223: Redundantif state_managercheck on line 218.
state_manageris already checked on line 211. The inner check on line 218 is alwaysTrue.Fix
if task_id and task_state_manager.is_cancelled(task_id): logger.info( f"Exception due to user cancellation for task {task_id}, " f"treating as CANCELLED not FAILED" ) - if state_manager: - state_manager.set_task_status(TaskStatus.CANCELLED.value) + state_manager.set_task_status(TaskStatus.CANCELLED.value)
266-282: Duplicate session ID saving logic.The session ID is saved from
SystemMessageinit (lines 266–282) and again fromResultMessage(lines 153–169). Both use the sameSessionManager.save_session_idcall. This is likely intentional for resilience (save early from init, update from result), but worth noting the duplication. The lazy import ofSessionManageralso appears in both places.Consider extracting a small helper like
_save_claude_session_id(task_id, session_id)to deduplicate the try/except/import pattern used in both locations.
| self.workbench_data["lastUpdated"] = datetime.now().isoformat() | ||
| logger.debug(f"Updated workbench summary: {len(summary)} chars") | ||
|
|
||
| def update_workbench_status(self, status: str, result_value: str = None) -> None: |
There was a problem hiding this comment.
PEP 484 prohibits implicit Optional.
result_value: str = None should use an explicit Optional[str] (or str | None). Confirmed by static analysis (Ruff RUF013).
Proposed fix
- def update_workbench_status(self, status: str, result_value: str = None) -> None:
+ def update_workbench_status(self, status: str, result_value: Optional[str] = None) -> None:📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| def update_workbench_status(self, status: str, result_value: str = None) -> None: | |
| def update_workbench_status(self, status: str, result_value: Optional[str] = None) -> None: |
🧰 Tools
🪛 Ruff (0.14.14)
[warning] 99-99: PEP 484 prohibits implicit Optional
Convert to T | None
(RUF013)
🤖 Prompt for AI Agents
In `@executor/agents/claude_code/progress_state_manager.py` at line 99, The
parameter annotation in update_workbench_status uses an implicit Optional
(result_value: str = None); update its type to an explicit Optional type (e.g.,
result_value: Optional[str] or result_value: str | None) and add the
corresponding import (from typing import Optional) if necessary, ensuring the
function signature and any callers remain type-correct.
| # Check if this is SDK interruption text block | ||
| if "[Request interrupted by user for tool use]" in block.text: | ||
| logger.info( | ||
| f"Skipping SDK interruption text block for task {state_manager.task_data.get('task_id') if state_manager else 'unknown'}" | ||
| ) | ||
| saw_sdk_interrupt_messages = ( | ||
| True # Mark that we saw interrupt messages | ||
| ) | ||
| continue # Skip this interruption message | ||
|
|
||
| # 文本内容详情，符合目标格式 | ||
| text_detail = {"type": "text", "text": block.text} | ||
| message_details["message"]["content"].append(text_detail) | ||
|
|
||
| logger.info(f"UserMessage TextBlock: {len(block.text)} chars") | ||
|
|
||
| elif isinstance(block, ToolResultBlock): | ||
| # Check if this is SDK interrupt error from previous cancellation | ||
| if block.is_error and "Request interrupted" in str(block.content): | ||
| # This is expected on resume - skip this error block | ||
| logger.info( | ||
| f"Skipping SDK interrupt error on resume for task {state_manager.task_data.get('task_id') if state_manager else 'unknown'}" | ||
| ) | ||
| saw_sdk_interrupt_messages = ( | ||
| True # Mark that we saw interrupt messages | ||
| ) | ||
| continue # Skip processing this error block |
There was a problem hiding this comment.
Bug: saw_sdk_interrupt_messages is a local variable, never propagated to caller.
Inside _handle_user_message, saw_sdk_interrupt_messages is assigned (lines 367, 385) but it's a local variable — it's never returned or otherwise communicated back to the caller. The caller process_response maintains its own saw_sdk_interrupt_messages (line 76) which stays False, and that's what gets passed to _process_result_message (line 183). This means the resume-from-interruption detection in _process_result_message (line 862) will never trigger based on SDK interrupt messages seen in UserMessage.
The static analysis hint also flags saw_sdk_interrupt_messages as assigned but never used (F841).
Proposed fix: return `saw_sdk_interrupt_messages` from `_handle_user_message`
Update the return type and return value:
def _handle_user_message(
msg: UserMessage, thinking_manager=None, state_manager=None
-) -> tuple[bool, str]:
+) -> tuple[bool, str, bool]:
"""处理用户消息，提取详细信息
Args:
msg: UserMessage to process
thinking_manager: Optional ThinkingStepManager instance
state_manager: Optional ProgressStateManager instance
Returns:
- Tuple of (silent_exit_detected, silent_exit_reason)
+ Tuple of (silent_exit_detected, silent_exit_reason, saw_sdk_interrupt_messages)
"""
from executor.tools.silent_exit import detect_silent_exit
# Track silent exit detection
silent_exit_detected = False
silent_exit_reason = ""
+ saw_sdk_interrupt_messages = FalseAt the end of the function:
- return silent_exit_detected, silent_exit_reason
+ return silent_exit_detected, silent_exit_reason, saw_sdk_interrupt_messagesAnd update the call site in process_response:
- is_silent, reason = _handle_user_message(
- msg, thinking_manager, state_manager
- )
+ is_silent, reason, saw_interrupt = _handle_user_message(
+ msg, thinking_manager, state_manager
+ )
+ if saw_interrupt:
+ saw_sdk_interrupt_messages = True🧰 Tools
🪛 Ruff (0.14.14)
[warning] 372-372: Comment contains ambiguous ， (FULLWIDTH COMMA). Did you mean , (COMMA)?
(RUF003)
[error] 385-385: Local variable saw_sdk_interrupt_messages is assigned to but never used
Remove assignment to unused variable saw_sdk_interrupt_messages
(F841)
🤖 Prompt for AI Agents
In `@executor/agents/claude_code/response_processor.py` around lines 362 - 388,
The local flag saw_sdk_interrupt_messages set inside _handle_user_message (when
handling UserMessage text/ToolResultBlock) never gets propagated to the caller,
so change _handle_user_message to return that boolean (update its signature and
return value), update the caller process_response to capture the returned
saw_sdk_interrupt_messages, and pass that updated flag into the subsequent call
to _process_result_message; also update any other call sites of
_handle_user_message to handle the new return value and remove the unused local
assignment (fixing the F841 static analysis warning).
| # Check if this is resuming from an interrupted task | ||
| # When resuming from interruption, SDK may legitimately return error_during_execution | ||
| # Check both: 1) if we saw SDK interrupt messages in this cycle, OR 2) task state is INTERRUPTED | ||
| is_resuming_from_interruption = saw_sdk_interrupt_messages | ||
|
|
||
| # Check if this is a user-initiated cancellation | ||
| # When user cancels, we get error_during_execution but it's not a real failure | ||
| is_user_cancellation = False | ||
| if task_state_manager and state_manager: | ||
| if not is_resuming_from_interruption and task_state_manager and state_manager: | ||
| task_id = state_manager.task_data.get("task_id") | ||
| if task_id and task_state_manager.is_cancelled(task_id): | ||
| is_user_cancellation = True | ||
| logger.info( | ||
| f"Task {task_id} error_during_execution is due to user cancellation, treating as CANCELLED" | ||
| ) | ||
|
|
||
| if is_user_cancellation: | ||
| # User cancelled the task, treat as CANCELLED not FAILED | ||
| if thinking_manager: | ||
| thinking_manager.add_thinking_step( | ||
| title="thinking.task_cancelled", | ||
| report_immediately=False, | ||
| use_i18n_keys=True, | ||
| details=masked_result_details, | ||
| ) | ||
|
|
||
| # Update workbench status to cancelled | ||
| state_manager.update_workbench_status("cancelled") | ||
| if task_id: | ||
| from executor.tasks.task_state_manager import TaskState | ||
|
|
||
| # Report cancelled status | ||
| state_manager.report_progress( | ||
| progress=100, | ||
| status=TaskStatus.CANCELLED.value, | ||
| message="Task cancelled by user", | ||
| task_state = task_state_manager.get_state(task_id) | ||
| if task_state == TaskState.INTERRUPTED: | ||
| is_resuming_from_interruption = True | ||
| logger.info( | ||
| f"Task {task_id} resumed from interruption based on task state" | ||
| ) | ||
| # Clear interrupted state and set to RUNNING | ||
| task_state_manager.set_state(task_id, TaskState.RUNNING) | ||
|
|
||
| if is_resuming_from_interruption: | ||
| # This is expected - SDK is reporting the previous interruption | ||
| # Return SUCCESS to indicate initialization can continue | ||
| # The actual new message will be processed in execute phase | ||
| if state_manager: | ||
| task_id = state_manager.task_data.get("task_id") | ||
| logger.info( | ||
| f"Ignoring error_during_execution on resume for task {task_id}, returning SUCCESS" | ||
| ) | ||
| return TaskStatus.CANCELLED | ||
| return TaskStatus.SUCCESS | ||
|
|
||
| # Check if this error subtype can be retried by resuming session | ||
| if ( | ||
| is_retryable_error_subtype(msg.subtype) | ||
| and session_id | ||
| and error_subtype_retry_count < MAX_ERROR_SUBTYPE_RETRIES | ||
| ): | ||
| # Get actual Claude session ID (UUID) for resume | ||
| # msg.session_id is the Claude UUID, session_id parameter is our internal key | ||
| actual_claude_session_id = msg.session_id or session_id | ||
| logger.warning( | ||
| f"Retryable error subtype detected: {msg.subtype}, " | ||
| f"retry {error_subtype_retry_count + 1}/{MAX_ERROR_SUBTYPE_RETRIES}, " | ||
| f"internal_session={session_id}, claude_session={actual_claude_session_id}" | ||
| # Check if this is due to current cancellation (cancellation_in_progress flag) | ||
| # When user cancels during execution, SDK sends error_during_execution | ||
| # This is the final message in the interrupt sequence - we can now return CANCELLED | ||
| logger.info( | ||
| f"Checking cancellation status: cancellation_in_progress={cancellation_in_progress}, " | ||
| f"is_resuming={is_resuming_from_interruption}" | ||
| ) | ||
| if cancellation_in_progress: | ||
| logger.info( | ||
| f"Task {task_id} error_during_execution due to current cancellation - marking as INTERRUPTED internally" | ||
| ) | ||
|
|
||
| if thinking_manager: | ||
| thinking_manager.add_thinking_step( | ||
| title="thinking.error_subtype_retry", | ||
| report_immediately=True, | ||
| use_i18n_keys=True, | ||
| details={ | ||
| "error_subtype": msg.subtype, | ||
| "retry_count": error_subtype_retry_count + 1, | ||
| "max_retries": MAX_ERROR_SUBTYPE_RETRIES, | ||
| "session_id": actual_claude_session_id, | ||
| }, | ||
| # Mark task as INTERRUPTED internally (preserve session for resumption) | ||
| if task_state_manager and task_id: | ||
| task_state_manager.set_interrupted(task_id) | ||
| logger.info( | ||
| f"Task {task_id} marked as INTERRUPTED internally (session preserved for resumption)" | ||
| ) | ||
|
|
||
| # Signal to caller that session should be resumed for retry | ||
| # Return tuple with Claude SDK's actual session_id (UUID) for proper resume | ||
| return ("RETRY_WITH_RESUME", actual_claude_session_id) | ||
|
|
||
| elif ( | ||
| is_retryable_error_subtype(msg.subtype) | ||
| and error_subtype_retry_count >= MAX_ERROR_SUBTYPE_RETRIES | ||
| ): | ||
| logger.error( | ||
| f"Max error subtype retries ({MAX_ERROR_SUBTYPE_RETRIES}) reached " | ||
| f"for session {session_id}, error_subtype={msg.subtype}" | ||
| ) | ||
| if thinking_manager: | ||
| thinking_manager.add_thinking_step( | ||
| title="thinking.error_subtype_max_retries", | ||
| report_immediately=True, | ||
| use_i18n_keys=True, | ||
| details={ | ||
| "error_subtype": msg.subtype, | ||
| "retry_count": error_subtype_retry_count, | ||
| "max_retries": MAX_ERROR_SUBTYPE_RETRIES, | ||
| "session_id": session_id, | ||
| }, | ||
| ) | ||
| # Return CANCELLED to frontend (INTERRUPTED is internal state only) | ||
| return TaskStatus.CANCELLED |
There was a problem hiding this comment.
Bug: task_id may be unbound, causing UnboundLocalError.
If is_resuming_from_interruption is True from line 862 (via saw_sdk_interrupt_messages), the block at lines 864–876 is skipped. Then at line 882, if state_manager is falsy, task_id is never assigned — yet line 885 references it unconditionally.
Similarly, if is_resuming_from_interruption is False and the condition at line 864 (task_state_manager and state_manager) is False, task_id is never assigned. Then at line 896, if cancellation_in_progress is True, lines 898 and 901 reference task_id, resulting in UnboundLocalError.
Proposed fix: initialize `task_id` at the top of this block
if msg.subtype and msg.subtype.startswith("error_"):
result_str = (
str(msg.result) if msg.result is not None else f"Task ended: {msg.subtype}"
)
+ task_id = state_manager.task_data.get("task_id") if state_manager else None
+
# Check if this is resuming from an interrupted taskThen remove the redundant task_id assignments at lines 865 and 883.
🤖 Prompt for AI Agents
In `@executor/agents/claude_code/response_processor.py` around lines 859 - 908,
Initialize task_id = None before the interruption/resume logic so it is always
defined (e.g., set task_id = None just before computing
is_resuming_from_interruption). In the block that checks task_state_manager and
state_manager use state_manager.task_data.get("task_id") to set task_id as
needed, but remove the later redundant reassignments of task_id (the one inside
the is_resuming_from_interruption branch and any duplicate before logging).
Ensure subsequent references (logs and task_state_manager.set_interrupted) use
the now-initialized task_id and guard actions when task_id is None; the relevant
symbols are is_resuming_from_interruption, task_state_manager, state_manager,
task_id, cancellation_in_progress, and TaskStatus.
| if is_resuming_from_interruption: | ||
| # This is expected - SDK is reporting the previous interruption | ||
| # Return SUCCESS to indicate initialization can continue | ||
| # The actual new message will be processed in execute phase | ||
| if state_manager: | ||
| task_id = state_manager.task_data.get("task_id") | ||
| logger.info( | ||
| f"Ignoring error_during_execution on resume for task {task_id}, returning SUCCESS" | ||
| ) | ||
| return TaskStatus.CANCELLED | ||
| return TaskStatus.SUCCESS |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
# First, let's find and examine the TaskStatus enum definition
find . -type f -name "*.py" | grep -E "(status|shared)" | head -20Repository: wecode-ai/Wegent
Length of output: 719
🏁 Script executed:
# Search for callers of process_response to understand the context
rg -n --type=py "\.process_response\(" --max-count=20Repository: wecode-ai/Wegent
Length of output: 42
🏁 Script executed:
# Check the context around line 194 in response_processor.py
rg -n --type=py -B5 -A10 "process_response" executor/agents/claude_code/response_processor.py | head -100Repository: wecode-ai/Wegent
Length of output: 473
🏁 Script executed:
# Search for TaskStatus enum definition
rg -n "class TaskStatus" --type=pyRepository: wecode-ai/Wegent
Length of output: 299
🏁 Script executed:
# Read the area around line 194 in response_processor.py
sed -n '185,210p' executor/agents/claude_code/response_processor.pyRepository: wecode-ai/Wegent
Length of output: 1162
🏁 Script executed:
# Search for calls to process_response with different patterns
rg -n "process_response" --type=py | head -30Repository: wecode-ai/Wegent
Length of output: 566
🏁 Script executed:
# Check the TaskStatus enum definition
cat -n shared/status.pyRepository: wecode-ai/Wegent
Length of output: 817
🏁 Script executed:
# Read how the result from process_response is used in claude_code_agent.py
sed -n '700,740p' executor/agents/claude_code/claude_code_agent.pyRepository: wecode-ai/Wegent
Length of output: 1710
🏁 Script executed:
# Find who calls the execute or main methods of claude_code_agent
rg -n "claude_code_agent\|ClaudeCodeAgent" --type=py | grep -E "(import|from|=)" | head -20Repository: wecode-ai/Wegent
Length of output: 42
🏁 Script executed:
# Search for callers of the function that calls process_response
rg -n "result = await process_response\|from.*claude_code_agent" --type=pyRepository: wecode-ai/Wegent
Length of output: 42
🏁 Script executed:
# Search for where ClaudeCodeAgent is instantiated or its methods called
rg -n "ClaudeCodeAgent\(" --type=pyRepository: wecode-ai/Wegent
Length of output: 729
🏁 Script executed:
# Look for the execute method being called
rg -n "\.execute\(" executor/agents/claude_code/ --type=pyRepository: wecode-ai/Wegent
Length of output: 42
🏁 Script executed:
# Check what happens after process_response returns result in claude_code_agent.py
sed -n '730,750p' executor/agents/claude_code/claude_code_agent.pyRepository: wecode-ai/Wegent
Length of output: 954
🏁 Script executed:
# Check how executor/modes/local/runner.py handles the result from ClaudeCodeAgent.execute()
sed -n '340,380p' executor/modes/local/runner.pyRepository: wecode-ai/Wegent
Length of output: 1521
🏁 Script executed:
# Also check executor/services/heartbeat_service.py
sed -n '360,400p' executor/services/heartbeat_service.pyRepository: wecode-ai/Wegent
Length of output: 1402
🏁 Script executed:
# Let's also look at the full context of lines 878-887 to understand the condition better
sed -n '870,895p' executor/agents/claude_code/response_processor.pyRepository: wecode-ai/Wegent
Length of output: 1436
Returning TaskStatus.SUCCESS on resume indicates task completion but should signal resumption continuation.
When resuming from an interrupted task (lines 878-887), the code returns TaskStatus.SUCCESS with a comment saying "Return SUCCESS to indicate initialization can continue." However, per the TaskStatus enum in shared/status.py, SUCCESS means "Task executed successfully, subtask completed," not "continue processing." This semantic mismatch will confuse callers—HeartbeatService line 376 already interprets init_status == TaskStatus.SUCCESS as "initialization completed," and claude_code_agent.py line 735 returns the result directly without special handling, propagating this misleading status upstream. Consider returning TaskStatus.RUNNING or a different status that accurately reflects the resumption state.
🤖 Prompt for AI Agents
In `@executor/agents/claude_code/response_processor.py` around lines 878 - 887,
The code path that handles is_resuming_from_interruption currently returns
TaskStatus.SUCCESS which incorrectly signals completion; change it to return a
status that indicates continuation (e.g., TaskStatus.RUNNING or a dedicated
RESUMING status) so callers like HeartbeatService and claude_code_agent.py don't
treat initialization as finished. Locate the block checking
is_resuming_from_interruption in response_processor.py (uses state_manager,
task_id, logger) and replace the return value TaskStatus.SUCCESS with the chosen
continuation status from the TaskStatus enum in shared/status.py, and update any
logging text to reflect "resuming/continuing" rather than "SUCCESS".
| if cancellation_in_progress: | ||
| logger.info( | ||
| f"Task {task_id} error_during_execution due to current cancellation - marking as INTERRUPTED internally" | ||
| ) | ||
|
|
||
| if thinking_manager: | ||
| thinking_manager.add_thinking_step( | ||
| title="thinking.error_subtype_retry", | ||
| report_immediately=True, | ||
| use_i18n_keys=True, | ||
| details={ | ||
| "error_subtype": msg.subtype, | ||
| "retry_count": error_subtype_retry_count + 1, | ||
| "max_retries": MAX_ERROR_SUBTYPE_RETRIES, | ||
| "session_id": actual_claude_session_id, | ||
| }, | ||
| # Mark task as INTERRUPTED internally (preserve session for resumption) | ||
| if task_state_manager and task_id: | ||
| task_state_manager.set_interrupted(task_id) | ||
| logger.info( | ||
| f"Task {task_id} marked as INTERRUPTED internally (session preserved for resumption)" | ||
| ) | ||
|
|
||
| # Signal to caller that session should be resumed for retry | ||
| # Return tuple with Claude SDK's actual session_id (UUID) for proper resume | ||
| return ("RETRY_WITH_RESUME", actual_claude_session_id) | ||
|
|
||
| elif ( | ||
| is_retryable_error_subtype(msg.subtype) | ||
| and error_subtype_retry_count >= MAX_ERROR_SUBTYPE_RETRIES | ||
| ): | ||
| logger.error( | ||
| f"Max error subtype retries ({MAX_ERROR_SUBTYPE_RETRIES}) reached " | ||
| f"for session {session_id}, error_subtype={msg.subtype}" | ||
| ) | ||
| if thinking_manager: | ||
| thinking_manager.add_thinking_step( | ||
| title="thinking.error_subtype_max_retries", | ||
| report_immediately=True, | ||
| use_i18n_keys=True, | ||
| details={ | ||
| "error_subtype": msg.subtype, | ||
| "retry_count": error_subtype_retry_count, | ||
| "max_retries": MAX_ERROR_SUBTYPE_RETRIES, | ||
| "session_id": session_id, | ||
| }, | ||
| ) | ||
| # Return CANCELLED to frontend (INTERRUPTED is internal state only) | ||
| return TaskStatus.CANCELLED |
There was a problem hiding this comment.
Missing set_task_status(CANCELLED) and report_progress before returning CANCELLED on cancellation path.
When cancellation_in_progress is True, the function returns TaskStatus.CANCELLED (line 908) but never calls state_manager.set_task_status(TaskStatus.CANCELLED.value) or state_manager.report_progress(...) to notify the frontend. Compare with other terminal paths (e.g., lines 837–843 for FAILED, lines 736–745 for COMPLETED) which all call both. The caller process_response also just does return result_status at line 195 without additional reporting.
Proposed fix
if cancellation_in_progress:
logger.info(
f"Task {task_id} error_during_execution due to current cancellation - marking as INTERRUPTED internally"
)
# Mark task as INTERRUPTED internally (preserve session for resumption)
if task_state_manager and task_id:
task_state_manager.set_interrupted(task_id)
logger.info(
f"Task {task_id} marked as INTERRUPTED internally (session preserved for resumption)"
)
+ # Notify frontend of cancellation
+ if state_manager:
+ state_manager.set_task_status(TaskStatus.CANCELLED.value)
+ state_manager.report_progress(
+ progress=100,
+ status=TaskStatus.CANCELLED.value,
+ message=f"Task cancelled by user",
+ )
+
# Return CANCELLED to frontend (INTERRUPTED is internal state only)
return TaskStatus.CANCELLED🤖 Prompt for AI Agents
In `@executor/agents/claude_code/response_processor.py` around lines 896 - 908,
The cancellation branch handles internal interruption via
task_state_manager.set_interrupted(task_id) but fails to notify the frontend;
before returning TaskStatus.CANCELLED when cancellation_in_progress is True,
call state reporting the terminal status and progress: invoke
task_state_manager.set_task_status(TaskStatus.CANCELLED.value) (or
state_manager.set_task_status if that symbol is used in this file) and
task_state_manager.report_progress(task_id, progress=100,
status=TaskStatus.CANCELLED.value) (or the existing report_progress signature
used elsewhere) so the frontend receives the CANCELLED terminal state, then
return TaskStatus.CANCELLED; keep the existing set_interrupted call to preserve
internal session.
| # DO NOT cleanup task state here - SDK interrupt messages still need to be processed | ||
| # State will be cleaned up in response_processor after all messages are processed | ||
|
|
||
| except Exception as e: | ||
| logger.exception(f"[{task_id}] Error sending cancel callback: {e}") | ||
| # Still attempt to cleanup task state and session 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" | ||
| ) | ||
| # Try to cleanup session even on error | ||
| session = self._agent_sessions.get(task_id) | ||
| if session: | ||
| await self._cleanup_cancelled_session(task_id, session) | ||
| except Exception as cleanup_error: | ||
| logger.warning( | ||
| f"[{task_id}] Failed to cleanup task state or session: {cleanup_error}" | ||
| ) | ||
|
|
||
| async def _cleanup_cancelled_session( | ||
| self, task_id: int, session: AgentSession | ||
| ) -> None: | ||
| """ | ||
| Clean up cancelled session by closing client and removing from session map. | ||
|
|
||
| This method is called after cancel callback is sent to ensure: | ||
| 1. Agent client connection is properly closed | ||
| 2. Session is removed from _agent_sessions map | ||
| 3. Task ID is no longer reported in heartbeat (freeing up device slot) | ||
|
|
||
| Args: | ||
| task_id: Task ID | ||
| session: Agent session to clean up | ||
| """ | ||
| try: | ||
| agent = session.agent | ||
| agent_name = agent.get_name() | ||
|
|
||
| # Close client connection if agent supports it | ||
| if hasattr(agent, "close_client") and hasattr(agent, "session_id"): | ||
| session_id = getattr(agent, "session_id", None) | ||
| if session_id: | ||
| logger.info( | ||
| f"[{_format_task_log(task_id, -1)}] Closing {agent_name} client for session_id: {session_id}" | ||
| ) | ||
| close_status = await agent.close_client(session_id) | ||
| if close_status == TaskStatus.SUCCESS: | ||
| logger.info( | ||
| f"[{_format_task_log(task_id, -1)}] Successfully closed {agent_name} client" | ||
| ) | ||
| else: | ||
| logger.warning( | ||
| f"[{_format_task_log(task_id, -1)}] Failed to close {agent_name} client" | ||
| ) | ||
| else: | ||
| logger.warning( | ||
| f"[{_format_task_log(task_id, -1)}] No session_id found for {agent_name} agent" | ||
| ) | ||
| else: | ||
| logger.debug( | ||
| f"[{_format_task_log(task_id, -1)}] {agent_name} agent does not support close_client" | ||
| ) | ||
|
|
||
| # Remove session from map to stop reporting in heartbeat | ||
| if task_id in self._agent_sessions: | ||
| del self._agent_sessions[task_id] | ||
| logger.info( | ||
| f"[{_format_task_log(task_id, -1)}] Removed session from agent_sessions map" | ||
| ) | ||
|
|
||
| except Exception as e: | ||
| logger.exception(f"[{task_id}] Error cleaning up cancelled session: {e}") | ||
| # DO NOT cleanup task state on error - let response_processor handle it |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Check how response_processor handles cleanup for cancelled/interrupted tasks
rg -n -C5 'CANCELLED\|INTERRUPTED\|cleanup\|delete_session\|_agent_sessions' --type=py -g '*response_processor*'Repository: wecode-ai/Wegent
Length of output: 42
🏁 Script executed:
#!/bin/bash
# Find response_processor files
fd -t f 'response_processor' --type py
# Also search for any files containing response_processor in their name or directory
find . -type f -name '*response_processor*' 2>/dev/nullRepository: wecode-ai/Wegent
Length of output: 343
🏁 Script executed:
#!/bin/bash
# Check agent_service.py around lines 318-323
sed -n '300,350p' executor/services/agent_service.py
# Also check for response_processor usage in agent_service.py
echo "--- Response processor usage in agent_service.py ---"
rg -n 'response_processor\|_agent_sessions' executor/services/agent_service.py | head -50Repository: wecode-ai/Wegent
Length of output: 2253
🏁 Script executed:
#!/bin/bash
# Search for where response_processor is imported/used
echo "--- Searching for response_processor imports/usage ---"
rg -n 'response_processor\|ResponseProcessor' executor/services/agent_service.py
# Check the beginning of agent_service.py for imports
echo "--- Imports in agent_service.py ---"
head -50 executor/services/agent_service.py
# Check how response_processor is defined/called in the codebase
echo "--- All response_processor references ---"
rg -n 'response_processor\|ResponseProcessor' --type=py | head -30Repository: wecode-ai/Wegent
Length of output: 1624
🏁 Script executed:
#!/bin/bash
# Check the size of response_processor files
echo "--- Response processor file sizes ---"
wc -l executor/agents/agno/response_processor.py executor/agents/claude_code/response_processor.py
# Read the agno response_processor
echo "--- Agno response_processor.py ---"
cat -n executor/agents/agno/response_processor.py
# Read the claude_code response_processor (check size first)
wc -l executor/agents/claude_code/response_processor.pyRepository: wecode-ai/Wegent
Length of output: 8825
🏁 Script executed:
#!/bin/bash
# Search for cleanup-related code in claude_code response_processor
echo "--- Searching for cleanup in claude_code response_processor ---"
rg -n 'cleanup\|delete_session\|_agent_sessions\|remove.*session' executor/agents/claude_code/response_processor.py
# Check the entire codebase for where _agent_sessions is modified
echo "--- Where _agent_sessions is modified ---"
rg -n '_agent_sessions' --type=py
# Look for timeout or periodic cleanup
echo "--- Looking for timeout/cleanup mechanisms ---"
rg -n 'timeout\|cleanup\|reap\|expire' executor/services/agent_service.py | head -30Repository: wecode-ai/Wegent
Length of output: 1456
🏁 Script executed:
#!/bin/bash
# Check the context around line 198 where session is deleted
echo "--- Context around line 198 (session deletion) ---"
sed -n '180,210p' executor/services/agent_service.py
# Check close_all_agent_sessions and surrounding code
echo "--- close_all_agent_sessions implementation (lines 354-380) ---"
sed -n '354,385p' executor/services/agent_service.py
# Search for where the function containing line 198 is called
echo "--- Function name at line 198 ---"
sed -n '170,210p' executor/services/agent_service.py | grep -B 30 'del self._agent_sessions'Repository: wecode-ai/Wegent
Length of output: 3843
🏁 Script executed:
#!/bin/bash
# Find all callers of delete_session
echo "--- Where delete_session/delete_session_async is called ---"
rg -n 'delete_session' --type=py | grep -v 'def delete_session'
# Check the full cancel_task function
echo "--- Full cancel_task function ---"
sed -n '250,330p' executor/services/agent_service.py
# Also check if there's any cleanup after cancel_task is called
echo "--- Search for cancel_task callers ---"
rg -n 'cancel_task' --type=py | head -20Repository: wecode-ai/Wegent
Length of output: 5090
🏁 Script executed:
#!/bin/bash
# Verify that response_processor is actually called during cancellation flow
echo "--- Search for where response_processor functions are called ---"
rg -n 'process_response\|process_team_chunk\|process_final_result' --type=py | head -20
# Check if there's any automatic cleanup mechanism we might have missed
echo "--- Search for automatic cleanup/garbage collection ---"
rg -n '__del__\|__exit__\|contextmanager\|atexit' executor/services/agent_service.py
# Verify the full cancellation and message processing flow
echo "--- Check how messages trigger response_processor ---"
rg -n 'response_processor\|SDK.*interrupt\|interrupt.*message' --type=pyRepository: wecode-ai/Wegent
Length of output: 234
🏁 Script executed:
#!/bin/bash
# Search for any imports of response_processor
echo "--- Imports of response_processor anywhere ---"
rg -n 'from.*response_processor|import.*response_processor' --type=py
# Check agno_agent.py and claude_code_agent.py
echo "--- Search in agno_agent.py ---"
fd -type f '*agno_agent*' --extension py
echo "--- Search in claude_code_agent.py ---"
fd -type f '*claude_code_agent*' --extension py
# Check if response_processor functions are defined but never called
echo "--- All function definitions in response_processor files ---"
rg -n '^(async )?def ' executor/agents/agno/response_processor.py executor/agents/claude_code/response_processor.pyRepository: wecode-ai/Wegent
Length of output: 1845
🏁 Script executed:
#!/bin/bash
# Check the imports from response_processor in claude_code_agent.py
echo "--- Imports from response_processor in claude_code_agent.py ---"
sed -n '40,60p' executor/agents/claude_code/claude_code_agent.py
# Search for where process_response is called in claude_code_agent.py
echo "--- Where process_response is called ---"
rg -n 'process_response\|process_team_chunk' executor/agents/claude_code/claude_code_agent.py
# Check what happens after response processing
echo "--- Context around process_response calls ---"
rg -n -B5 -A10 'await.*process_response' executor/agents/claude_code/claude_code_agent.py | head -50Repository: wecode-ai/Wegent
Length of output: 1574
🏁 Script executed:
#!/bin/bash
# Check what happens after process_response
echo "--- After process_response (lines 715-750) ---"
sed -n '715,750p' executor/agents/claude_code/claude_code_agent.py
# Check the claude_code_agent.py structure to understand the flow
echo "--- Method containing process_response call ---"
sed -n '650,720p' executor/agents/claude_code/claude_code_agent.py | head -40Repository: wecode-ai/Wegent
Length of output: 3747
🏁 Script executed:
#!/bin/bash
# Now let's check what process_response actually does in detail, especially for CANCELLED status
echo "--- process_response implementation (claude_code version) ---"
sed -n '48,150p' executor/agents/claude_code/response_processor.py
# Check if it handles CANCELLED status and what it does
echo "--- Search for CANCELLED handling in claude_code response_processor ---"
rg -n 'CANCELLED\|delete_session\|cleanup' executor/agents/claude_code/response_processor.pyRepository: wecode-ai/Wegent
Length of output: 5088
🏁 Script executed:
#!/bin/bash
# Search for where CANCELLED is returned or handled in response_processor
echo "--- Full process_response function (check file size first) ---"
wc -l executor/agents/claude_code/response_processor.py
# Read the end of process_response to see all return paths
echo "--- End of process_response function (lines 200-250) ---"
sed -n '200,250p' executor/agents/claude_code/response_processor.py
# Look for all return statements in process_response
echo "--- All return statements in process_response ---"
rg -n 'return.*TaskStatus' executor/agents/claude_code/response_processor.py | head -30
# Most importantly, search if session deletion happens anywhere in response_processor
echo "--- Session/cleanup operations in all response_processor files ---"
rg -n 'del \|remove\|delete\|_agent_sessions' executor/agents/claude_code/response_processor.py executor/agents/agno/response_processor.pyRepository: wecode-ai/Wegent
Length of output: 2618
🏁 Script executed:
#!/bin/bash
# Critical finding: response_processor calls task_state_manager.cleanup but does it clean sessions?
echo "--- What does task_state_manager.cleanup do? ---"
rg -n 'class TaskStateManager\|def cleanup' executor/tasks/task_state_manager.py | head -10
# Check the cleanup method in task_state_manager
echo "--- TaskStateManager cleanup method ---"
rg -n -A20 'def cleanup' executor/tasks/task_state_manager.py
# Most importantly: does task_state_manager.cleanup delete from _agent_sessions?
echo "--- Does task_state_manager interact with _agent_sessions? ---"
rg -n '_agent_sessions' executor/tasks/task_state_manager.py
# Back to agent_service: does the execute flow ever delete the session?
echo "--- Full execute flow in agent_service ---"
sed -n '100,150p' executor/services/agent_service.pyRepository: wecode-ai/Wegent
Length of output: 2921
🏁 Script executed:
#!/bin/bash
# Check if agent.handle() or execute flow ever deletes sessions
echo "--- Check agent implementations for cleanup ---"
rg -n 'def handle\|delete_session\|_agent_sessions' executor/agents/agent.py executor/agents/base_agent.py 2>/dev/null | head -20
# Check if execute_task ever calls delete_session
echo "--- Full execute_task function ---"
rg -n -A50 'def execute_task' executor/services/agent_service.py | head -70
# Check if there's any cleanup after agent execution completes
echo "--- Is there any cleanup after agent.handle() returns? ---"
rg -n -B5 -A15 'agent.handle\|agent_service.execute' executor/app.py | head -50Repository: wecode-ai/Wegent
Length of output: 2549
Cleanup gap confirmed: response_processor does not actually clean up _agent_sessions.
The code comments claim state cleanup is deferred to response_processor, but verification shows this is not implemented. While response_processor calls task_state_manager.cleanup() for cancelled tasks (line 223 in claude_code version), this only cleans the task state (_states dict), not the agent session from _agent_sessions. Sessions are only deleted via explicit delete_session_async() or close_all_agent_sessions() calls—neither of which is triggered automatically by response processing.
This creates a resource leak: cancelled or failed tasks with no subsequent explicit cleanup will leave their sessions indefinitely in _agent_sessions. A safety-net mechanism (timeout-based cleanup or periodic session reaper) is not optional but necessary.
Either:
- Implement actual session cleanup in
response_processor, or - Add a timeout-based mechanism to expire stale sessions (e.g., on agent task completion)
🧰 Tools
🪛 Ruff (0.14.14)
[warning] 322-322: Redundant exception object included in logging.exception call
(TRY401)
🤖 Prompt for AI Agents
In `@executor/services/agent_service.py` around lines 318 - 323, The comment
points out that cancelled/failed tasks leave entries in _agent_sessions because
response_processor only calls task_state_manager.cleanup() and does not remove
sessions; fix by making response_processor (the function handling cancelled
tasks) also call the agent session cleanup path (e.g., invoke
delete_session_async(task_id) or remove the session from the _agent_sessions
mapping) whenever a task is cleaned up, or alternatively add a timeout-based
reaper that periodically scans _agent_sessions and removes sessions that have
been idle beyond a threshold; ensure the fix touches response_processor and uses
the existing delete_session_async / close_all_agent_sessions APIs (or a new
remove_session_by_task_id helper) and ties cleanup to task_id so stale sessions
are reliably removed.
* fix: cancel task error * fix: black and isort
feat: add Socket.IO real-time communication system for chat (#517) See merge request weibo_rd/common/wecode/wegent!821
Fix/cancel task error (#517) See merge request weibo_rd/common/wecode/wegent!1427
Summary by CodeRabbit
New Features
Bug Fixes
Refactoring
Chores