fix(chat/sdk): address PR review feedback on security and robustness

- security_hooks: use realpath instead of normpath to resolve symlinks
- security_hooks: check tool-results as path segment, not substring
- response_adapter: emit StreamFinish for unknown ResultMessage subtypes
- tool_adapter: delete file after read (prevent accumulation in pods)
- check_operation_status: guard against None.strip() from LLM null args
- service: remove redundant ".." check (realpath already resolves)

autogpt_platform/backend/backend/api/features/chat/sdk/response_adapter.py

@@ -156,6 +156,11 @@ class SDKResponseAdapter:
                     StreamError(errorText=str(error_msg), code="sdk_error")
                 )
                 responses.append(StreamFinish())
+            else:
+                logger.warning(
+                    f"Unexpected ResultMessage subtype: {sdk_message.subtype}"
+                )
+                responses.append(StreamFinish())
 
         else:
             logger.debug(f"Unhandled SDK message type: {type(sdk_message).__name__}")

autogpt_platform/backend/backend/api/features/chat/sdk/security_hooks.py

@@ -80,21 +80,22 @@ def _validate_workspace_path(
     # naturally uses relative paths like "test.txt" instead of absolute ones).
     # Tilde paths (~/) are home-dir references, not relative — expand first.
     if path.startswith("~"):
-        resolved = os.path.normpath(os.path.expanduser(path))
+        resolved = os.path.realpath(os.path.expanduser(path))
     elif not os.path.isabs(path) and sdk_cwd:
-        resolved = os.path.normpath(os.path.join(sdk_cwd, path))
+        resolved = os.path.realpath(os.path.join(sdk_cwd, path))
     else:
-        resolved = os.path.normpath(path)
+        resolved = os.path.realpath(path)
 
     # Allow access within the SDK working directory
     if sdk_cwd:
-        norm_cwd = os.path.normpath(sdk_cwd)
+        norm_cwd = os.path.realpath(sdk_cwd)
         if resolved.startswith(norm_cwd + os.sep) or resolved == norm_cwd:
             return {}
 
     # Allow access to ~/.claude/projects/*/tool-results/ (big tool results)
-    claude_dir = os.path.normpath(os.path.expanduser("~/.claude/projects"))
-    if resolved.startswith(claude_dir + os.sep) and "tool-results" in resolved:
+    claude_dir = os.path.realpath(os.path.expanduser("~/.claude/projects"))
+    tool_results_seg = os.sep + "tool-results" + os.sep
+    if resolved.startswith(claude_dir + os.sep) and tool_results_seg in resolved:
         return {}
 
     logger.warning(

autogpt_platform/backend/backend/api/features/chat/sdk/service.py

@@ -271,9 +271,9 @@ def _cleanup_sdk_tool_results(cwd: str) -> None:
         logger.warning(f"[SDK] Rejecting cleanup for invalid path: {cwd}")
         return
 
-    # Security check 2: Ensure no path traversal in the normalized path
-    if ".." in normalized:
-        logger.warning(f"[SDK] Rejecting cleanup for traversal attempt: {cwd}")
+    # Security check 2: Verify path stayed within workspace after normalization
+    if not normalized.startswith(_SDK_CWD_PREFIX):
+        logger.warning(f"[SDK] Rejecting cleanup for path outside workspace: {cwd}")
         return
 
     # SDK encodes the cwd path by replacing '/' with '-'

autogpt_platform/backend/backend/api/features/chat/sdk/tool_adapter.py

@@ -219,6 +219,11 @@ async def _read_file_handler(args: dict[str, Any]) -> dict[str, Any]:
             lines = f.readlines()
         selected = lines[offset : offset + limit]
         content = "".join(selected)
+        # Clean up to prevent accumulation in long-running pods
+        try:
+            os.remove(real_path)
+        except OSError:
+            pass
         return {"content": [{"type": "text", "text": content}], "isError": False}
     except FileNotFoundError:
         return {

autogpt_platform/backend/backend/api/features/chat/tools/check_operation_status.py

@@ -80,8 +80,8 @@ class CheckOperationStatusTool(BaseTool):
     ) -> ToolResponseBase:
         from backend.api.features.chat import stream_registry
 
-        operation_id: str = kwargs.get("operation_id", "").strip()
-        task_id: str = kwargs.get("task_id", "").strip()
+        operation_id = (kwargs.get("operation_id") or "").strip()
+        task_id = (kwargs.get("task_id") or "").strip()
 
         if not operation_id and not task_id:
             return ErrorResponse(