From f3870b4e2f34c21eb8e57871a2dc55aa7612513a Mon Sep 17 00:00:00 2001 From: Brian Krafft Date: Mon, 6 Apr 2026 10:36:16 -0700 Subject: [PATCH 1/2] feat(grounding): extract tools, visual, workspace modules (Epic 5.9) Extract mid-layer methods from grounding_agent.py into 3 new modules: - grounding/tools.py: _get_available_tools, _load_all_tools - grounding/visual.py: _visual_analysis_callback, _enhance_result_with_visual_context, _select_key_screenshots - grounding/workspace.py: _get_workspace_path, _scan_workspace_files, _check_workspace_artifacts grounding_agent.py delegates to extracted functions via _impl aliases. Pure functions (_select_key_screenshots, _get_workspace_path, _scan_workspace_files) bound as staticmethod. Removes unused BackendType and ScreenshotClient imports from grounding_agent.py. Reduces file from ~796 to 391 lines. 47 new tests across 3 test files. All 107 grounding tests pass. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- openspace/agents/grounding/tools.py | 101 ++++++ openspace/agents/grounding/visual.py | 215 +++++++++++++ openspace/agents/grounding/workspace.py | 128 ++++++++ openspace/agents/grounding_agent.py | 399 ++---------------------- tests/test_grounding_tools.py | 174 +++++++++++ tests/test_grounding_visual.py | 209 +++++++++++++ tests/test_grounding_workspace.py | 201 ++++++++++++ 7 files changed, 1055 insertions(+), 372 deletions(-) create mode 100644 openspace/agents/grounding/tools.py create mode 100644 openspace/agents/grounding/visual.py create mode 100644 openspace/agents/grounding/workspace.py create mode 100644 tests/test_grounding_tools.py create mode 100644 tests/test_grounding_visual.py create mode 100644 tests/test_grounding_workspace.py diff --git a/openspace/agents/grounding/tools.py b/openspace/agents/grounding/tools.py new file mode 100644 index 00000000..1c1590be --- /dev/null +++ b/openspace/agents/grounding/tools.py @@ -0,0 +1,101 @@ +"""Tool retrieval helpers for GroundingAgent. + +Handles tool discovery, filtering, and fallback loading from backends. +Extracted from grounding_agent.py (Epic 5.9). +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, List, Optional + +from openspace.grounding.core.types import BackendType +from openspace.utils.logging import Logger + +if TYPE_CHECKING: + from openspace.grounding.core.grounding_client import GroundingClient + +logger = Logger.get_logger("openspace.agents.grounding_agent") + + +async def _get_available_tools(agent, task_description: Optional[str]) -> List: + """Retrieve tools for the current execution phase. + + Both skill-augmented and normal modes use the same + ``get_tools_with_auto_search`` pipeline: + - Non-MCP tools (shell, gui, web, system) are always included. + - MCP tools are filtered by relevance only when their count + exceeds ``max_tools``. + + When skills are active, the shell backend is guaranteed to be in + scope (skills commonly reference ``shell_agent``). + + Falls back to returning all tools if anything fails. + """ + grounding_client = agent.grounding_client + if not grounding_client: + return [] + + backends = [BackendType(name) for name in agent._backend_scope] + + # Ensure shell backend is available when skills are active + # (skills commonly reference shell_agent, read_file, etc.) + if agent.has_skill_context: + shell_bt = BackendType.SHELL + if shell_bt not in backends: + backends = list(backends) + [shell_bt] + logger.info("Added Shell backend to scope for skill file I/O") + + try: + retrieval_llm = agent._tool_retrieval_llm or agent._llm_client + tools = await grounding_client.get_tools_with_auto_search( + task_description=task_description, + backend=backends, + use_cache=True, + llm_callable=retrieval_llm, + ) + logger.info( + f"GroundingAgent selected {len(tools)} tools (auto-search) " + f"from {len(backends)} backends" + + (" [skill-augmented]" if agent.has_skill_context else "") + ) + except Exception as e: + logger.warning(f"Auto-search tools failed, falling back to full list: {e}") + tools = await _load_all_tools(agent, grounding_client) + + # Append retrieve_skill tool when skill registry is available + if agent._skill_registry and agent._skill_registry.list_skills(): + from openspace.skill_engine.retrieve_tool import RetrieveSkillTool + + retrieve_llm = agent._tool_retrieval_llm or agent._llm_client + retrieve_tool = RetrieveSkillTool( + agent._skill_registry, + backends=[b.value for b in backends], + llm_client=retrieve_llm, + skill_store=getattr(agent, "_skill_store", None), + ) + retrieve_tool.bind_runtime_info( + backend=BackendType.SYSTEM, + session_name="internal", + ) + tools.append(retrieve_tool) + logger.info("Added retrieve_skill tool for mid-iteration skill retrieval") + + return tools + + +async def _load_all_tools(agent, grounding_client: "GroundingClient") -> List: + """Fallback: load all tools from all backends without search.""" + all_tools = [] + for backend_name in agent._backend_scope: + try: + backend_type = BackendType(backend_name) + tools = await grounding_client.list_tools(backend=backend_type) + all_tools.extend(tools) + logger.debug(f"Retrieved {len(tools)} tools from backend: {backend_name}") + except Exception as e: + logger.debug(f"Could not get tools from {backend_name}: {e}") + + logger.info( + f"GroundingAgent fallback retrieved {len(all_tools)} tools from {len(agent._backend_scope)} backends" + ) + return all_tools diff --git a/openspace/agents/grounding/visual.py b/openspace/agents/grounding/visual.py new file mode 100644 index 00000000..bccf8444 --- /dev/null +++ b/openspace/agents/grounding/visual.py @@ -0,0 +1,215 @@ +"""Visual analysis helpers for GroundingAgent. + +Handles screenshot capture, visual analysis callbacks, and screenshot +selection for grounding agent workflows. +Extracted from grounding_agent.py (Epic 5.9). +""" + +from __future__ import annotations + +import json +from typing import TYPE_CHECKING, Any, Dict, List + +from openspace.grounding.core.types import ToolResult +from openspace.platforms.screenshot import ScreenshotClient +from openspace.prompts import GroundingAgentPrompts +from openspace.utils.logging import Logger + +if TYPE_CHECKING: + pass + +logger = Logger.get_logger("openspace.agents.grounding_agent") + + +async def _visual_analysis_callback( + agent, result: ToolResult, tool_name: str, tool_call: Dict, backend: str +) -> ToolResult: + """Callback for LLMClient to handle visual analysis after tool execution.""" + # 1. Check if LLM requested to skip visual analysis + skip_visual_analysis = False + try: + arguments = tool_call.function.arguments + if isinstance(arguments, str): + args = json.loads(arguments.strip() or "{}") + else: + args = arguments + + if isinstance(args, dict) and args.get("skip_visual_analysis"): + skip_visual_analysis = True + logger.info(f"Visual analysis skipped for {tool_name} (meta-parameter set by LLM)") + except Exception as e: + logger.debug(f"Could not parse tool arguments: {e}") + + # 2. If skip requested, return original result + if skip_visual_analysis: + return result + + # 3. Check if this backend needs visual analysis + if backend != "gui": + return result + + # 4. Check if tool has visual data + metadata = getattr(result, "metadata", None) + has_screenshots = metadata and (metadata.get("screenshot") or metadata.get("screenshots")) + + # 5. If no visual data, try to capture a screenshot + if not has_screenshots: + try: + logger.info(f"No visual data from {tool_name}, capturing screenshot...") + screenshot_client = ScreenshotClient() + screenshot_bytes = await screenshot_client.capture() + + if screenshot_bytes: + # Add screenshot to result metadata + if metadata is None: + result.metadata = {} + metadata = result.metadata + metadata["screenshot"] = screenshot_bytes + has_screenshots = True + logger.info("Screenshot captured for visual analysis") + else: + logger.warning("Failed to capture screenshot") + except Exception as e: + logger.warning(f"Error capturing screenshot: {e}") + + # 6. If still no screenshots, return original result + if not has_screenshots: + logger.debug(f"No visual data available for {tool_name}") + return result + + # 7. Perform visual analysis + return await _enhance_result_with_visual_context(agent, result, tool_name) + + +async def _enhance_result_with_visual_context( + agent, result: ToolResult, tool_name: str +) -> ToolResult: + """Enhance tool result with visual analysis for grounding agent workflows.""" + import asyncio + import base64 + + import litellm + + try: + metadata = getattr(result, "metadata", None) + if not metadata: + return result + + # Collect all screenshots + screenshots_bytes = [] + + # Check for multiple screenshots first + if metadata.get("screenshots"): + screenshots_list = metadata["screenshots"] + if isinstance(screenshots_list, list): + screenshots_bytes = [s for s in screenshots_list if s] + # Fall back to single screenshot + elif metadata.get("screenshot"): + screenshots_bytes = [metadata["screenshot"]] + + if not screenshots_bytes: + return result + + # Select key screenshots if there are too many + selected_screenshots = _select_key_screenshots(screenshots_bytes, max_count=3) + + # Convert to base64 + visual_b64_list = [] + for visual_data in selected_screenshots: + if isinstance(visual_data, bytes): + visual_b64_list.append(base64.b64encode(visual_data).decode("utf-8")) + else: + visual_b64_list.append(visual_data) # Already base64 + + # Build prompt based on number of screenshots + num_screenshots = len(visual_b64_list) + + prompt = GroundingAgentPrompts.visual_analysis( + tool_name=tool_name, + num_screenshots=num_screenshots, + task_description=getattr(agent, "_current_instruction", ""), + ) + + # Build content with text prompt + all images + content: List[Dict[str, Any]] = [{"type": "text", "text": prompt}] + for visual_b64 in visual_b64_list: + content.append({"type": "image_url", "image_url": {"url": f"data:image/png;base64,{visual_b64}"}}) + + # Use dedicated visual analysis model if configured, otherwise use main LLM model + visual_model = agent._visual_analysis_model or ( + agent._llm_client.model if agent._llm_client else "openrouter/anthropic/claude-sonnet-4.5" + ) + response = await asyncio.wait_for( + litellm.acompletion( + model=visual_model, + messages=[{"role": "user", "content": content}], + timeout=agent._visual_analysis_timeout, + ), + timeout=agent._visual_analysis_timeout + 5, + ) + + analysis = response.choices[0].message.content.strip() + + # Inject visual analysis into content + original_content = result.content or "(no text output)" + enhanced_content = f"{original_content}\n\n**Visual content**: {analysis}" + + # Create enhanced result + enhanced_result = ToolResult( + status=result.status, + content=enhanced_content, + error=result.error, + metadata={**metadata, "visual_analyzed": True, "visual_analysis": analysis}, + execution_time=result.execution_time, + ) + + logger.info(f"Enhanced {tool_name} result with visual analysis ({num_screenshots} screenshot(s))") + return enhanced_result + + except asyncio.TimeoutError: + logger.warning(f"Visual analysis timed out for {tool_name}, returning original result") + return result + except Exception as e: + logger.warning(f"Failed to analyze visual content for {tool_name}: {e}") + return result + + +def _select_key_screenshots(screenshots: List[bytes], max_count: int = 3) -> List[bytes]: + """Select key screenshots if there are too many. + + Pure function — does not require an agent instance. + """ + if len(screenshots) <= max_count: + return screenshots + + selected_indices: set[int] = set() + + # Always include last (final state) + selected_indices.add(len(screenshots) - 1) + + # If room, include first (initial state) + if max_count >= 2: + selected_indices.add(0) + + # Fill remaining slots with evenly spaced middle screenshots + remaining_slots = max_count - len(selected_indices) + if remaining_slots > 0: + # Calculate spacing + available_indices = [i for i in range(1, len(screenshots) - 1) if i not in selected_indices] + + if available_indices: + step = max(1, len(available_indices) // (remaining_slots + 1)) + for i in range(remaining_slots): + idx = min((i + 1) * step, len(available_indices) - 1) + if idx < len(available_indices): + selected_indices.add(available_indices[idx]) + + # Return screenshots in original order + selected = [screenshots[i] for i in sorted(selected_indices)] + + logger.debug( + f"Selected {len(selected)} screenshots at indices {sorted(selected_indices)} " + f"from total of {len(screenshots)}" + ) + + return selected diff --git a/openspace/agents/grounding/workspace.py b/openspace/agents/grounding/workspace.py new file mode 100644 index 00000000..38d92095 --- /dev/null +++ b/openspace/agents/grounding/workspace.py @@ -0,0 +1,128 @@ +"""Workspace scanning helpers for GroundingAgent. + +Provides workspace path resolution, file scanning, and artifact +checking for grounding agent task execution. +Extracted from grounding_agent.py (Epic 5.9). +""" + +from __future__ import annotations + +import os +import re +import time +from typing import Any, Dict, Optional + +from openspace.utils.logging import Logger + +logger = Logger.get_logger("openspace.agents.grounding_agent") + + +def _get_workspace_path(context: Dict[str, Any]) -> Optional[str]: + """Get workspace directory path from context. + + Pure function — does not require an agent instance. + """ + return context.get("workspace_dir") + + +def _scan_workspace_files( + workspace_path: str, + recent_threshold: int = 600, # seconds +) -> Dict[str, Any]: + """Scan workspace directory and collect file information. + + Pure function — does not require an agent instance. + + Args: + workspace_path: Path to workspace directory + recent_threshold: Threshold in seconds for recent files + + Returns: + Dictionary with file information: + - files: List of all filenames + - file_details: Dict mapping filename to file info (size, modified, age_seconds) + - recent_files: List of recently modified filenames + """ + result: Dict[str, Any] = {"files": [], "file_details": {}, "recent_files": []} + + if not workspace_path or not os.path.exists(workspace_path): + return result + + # Recording system files to exclude from workspace scanning + excluded_files = {"metadata.json", "traj.jsonl"} + + try: + current_time = time.time() + + for filename in os.listdir(workspace_path): + filepath = os.path.join(workspace_path, filename) + if os.path.isfile(filepath) and filename not in excluded_files: + result["files"].append(filename) + + # Get file stats + stat = os.stat(filepath) + file_info = { + "size": stat.st_size, + "modified": stat.st_mtime, + "age_seconds": current_time - stat.st_mtime, + } + result["file_details"][filename] = file_info + + # Track recently created/modified files + if file_info["age_seconds"] < recent_threshold: + result["recent_files"].append(filename) + + result["files"] = sorted(result["files"]) + + except Exception as e: + logger.debug(f"Error scanning workspace files: {e}") + + return result + + +async def _check_workspace_artifacts(agent, context: Dict[str, Any]) -> Dict[str, Any]: + """Check workspace directory for existing artifacts that might be relevant to the task. + + Enhanced to detect if task might already be completed. + """ + workspace_info: Dict[str, Any] = {"has_files": False, "files": [], "file_details": {}, "recent_files": []} + + try: + # Get workspace path — call module-level function directly + workspace_path = _get_workspace_path(context) + + # Scan workspace files — call module-level function directly + scan_result = _scan_workspace_files(workspace_path, recent_threshold=600) + + if scan_result["files"]: + workspace_info["has_files"] = True + workspace_info["files"] = scan_result["files"] + workspace_info["file_details"] = scan_result["file_details"] + workspace_info["recent_files"] = scan_result["recent_files"] + + logger.info( + f"Grounding Agent: Found {len(scan_result['files'])} existing files in workspace " + f"({len(scan_result['recent_files'])} recent)" + ) + + # Check if instruction mentions specific filenames + instruction = context.get("instruction", "") + if instruction: + # Look for potential file references in instruction + potential_outputs = [] + # Match common file patterns: filename.ext, "filename", 'filename' + file_patterns = re.findall(r'["\']?([a-zA-Z0-9_\-]+\.[a-zA-Z0-9]+)["\']?', instruction) + for pattern in file_patterns: + if pattern in scan_result["files"]: + potential_outputs.append(pattern) + + if potential_outputs: + workspace_info["matching_files"] = potential_outputs + logger.info( + f"Grounding Agent: Found {len(potential_outputs)} files matching task: {potential_outputs}" + ) + + except Exception as e: + logger.debug(f"Could not check workspace artifacts: {e}") + + return workspace_info diff --git a/openspace/agents/grounding_agent.py b/openspace/agents/grounding_agent.py index 4af1b19b..4d550e06 100644 --- a/openspace/agents/grounding_agent.py +++ b/openspace/agents/grounding_agent.py @@ -23,8 +23,21 @@ construct_messages as _construct_messages, default_system_prompt as _default_system_prompt, ) -from openspace.grounding.core.types import BackendType, ToolResult -from openspace.platforms.screenshot import ScreenshotClient +from openspace.agents.grounding.tools import ( + _get_available_tools as _get_available_tools_impl, + _load_all_tools as _load_all_tools_impl, +) +from openspace.agents.grounding.visual import ( + _enhance_result_with_visual_context as _enhance_result_with_visual_context_impl, + _select_key_screenshots as _select_key_screenshots_impl, + _visual_analysis_callback as _visual_analysis_callback_impl, +) +from openspace.agents.grounding.workspace import ( + _check_workspace_artifacts as _check_workspace_artifacts_impl, + _get_workspace_path as _get_workspace_path_impl, + _scan_workspace_files as _scan_workspace_files_impl, +) +from openspace.grounding.core.types import ToolResult from openspace.prompts import GroundingAgentPrompts from openspace.utils.logging import Logger @@ -145,390 +158,32 @@ def construct_messages(self, context: Dict[str, Any]) -> List[Dict[str, Any]]: return _construct_messages(self, context) async def _get_available_tools(self, task_description: Optional[str]) -> List: - """ - Retrieve tools for the current execution phase. - - Both skill-augmented and normal modes use the same - ``get_tools_with_auto_search`` pipeline: - - Non-MCP tools (shell, gui, web, system) are always included. - - MCP tools are filtered by relevance only when their count - exceeds ``max_tools``. - - When skills are active, the shell backend is guaranteed to be in - scope (skills commonly reference ``shell_agent``). - - Falls back to returning all tools if anything fails. - """ - grounding_client = self.grounding_client - if not grounding_client: - return [] - - backends = [BackendType(name) for name in self._backend_scope] - - # Ensure shell backend is available when skills are active - # (skills commonly reference shell_agent, read_file, etc.) - if self.has_skill_context: - shell_bt = BackendType.SHELL - if shell_bt not in backends: - backends = list(backends) + [shell_bt] - logger.info("Added Shell backend to scope for skill file I/O") - - try: - retrieval_llm = self._tool_retrieval_llm or self._llm_client - tools = await grounding_client.get_tools_with_auto_search( - task_description=task_description, - backend=backends, - use_cache=True, - llm_callable=retrieval_llm, - ) - logger.info( - f"GroundingAgent selected {len(tools)} tools (auto-search) " - f"from {len(backends)} backends" + (" [skill-augmented]" if self.has_skill_context else "") - ) - except Exception as e: - logger.warning(f"Auto-search tools failed, falling back to full list: {e}") - tools = await self._load_all_tools(grounding_client) - - # Append retrieve_skill tool when skill registry is available - if self._skill_registry and self._skill_registry.list_skills(): - from openspace.skill_engine.retrieve_tool import RetrieveSkillTool - - retrieve_llm = self._tool_retrieval_llm or self._llm_client - retrieve_tool = RetrieveSkillTool( - self._skill_registry, - backends=[b.value for b in backends], - llm_client=retrieve_llm, - skill_store=getattr(self, "_skill_store", None), - ) - retrieve_tool.bind_runtime_info( - backend=BackendType.SYSTEM, - session_name="internal", - ) - tools.append(retrieve_tool) - logger.info("Added retrieve_skill tool for mid-iteration skill retrieval") - - return tools + """Retrieve tools for the current execution phase.""" + return await _get_available_tools_impl(self, task_description) async def _load_all_tools(self, grounding_client: "GroundingClient") -> List: """Fallback: load all tools from all backends without search.""" - all_tools = [] - for backend_name in self._backend_scope: - try: - backend_type = BackendType(backend_name) - tools = await grounding_client.list_tools(backend=backend_type) - all_tools.extend(tools) - logger.debug(f"Retrieved {len(tools)} tools from backend: {backend_name}") - except Exception as e: - logger.debug(f"Could not get tools from {backend_name}: {e}") - - logger.info( - f"GroundingAgent fallback retrieved {len(all_tools)} tools from {len(self._backend_scope)} backends" - ) - return all_tools + return await _load_all_tools_impl(self, grounding_client) async def _visual_analysis_callback( self, result: ToolResult, tool_name: str, tool_call: Dict, backend: str ) -> ToolResult: - """ - Callback for LLMClient to handle visual analysis after tool execution. - """ - # 1. Check if LLM requested to skip visual analysis - skip_visual_analysis = False - try: - arguments = tool_call.function.arguments - if isinstance(arguments, str): - args = json.loads(arguments.strip() or "{}") - else: - args = arguments - - if isinstance(args, dict) and args.get("skip_visual_analysis"): - skip_visual_analysis = True - logger.info(f"Visual analysis skipped for {tool_name} (meta-parameter set by LLM)") - except Exception as e: - logger.debug(f"Could not parse tool arguments: {e}") - - # 2. If skip requested, return original result - if skip_visual_analysis: - return result - - # 3. Check if this backend needs visual analysis - if backend != "gui": - return result - - # 4. Check if tool has visual data - metadata = getattr(result, "metadata", None) - has_screenshots = metadata and (metadata.get("screenshot") or metadata.get("screenshots")) - - # 5. If no visual data, try to capture a screenshot - if not has_screenshots: - try: - logger.info(f"No visual data from {tool_name}, capturing screenshot...") - screenshot_client = ScreenshotClient() - screenshot_bytes = await screenshot_client.capture() - - if screenshot_bytes: - # Add screenshot to result metadata - if metadata is None: - result.metadata = {} - metadata = result.metadata - metadata["screenshot"] = screenshot_bytes - has_screenshots = True - logger.info("Screenshot captured for visual analysis") - else: - logger.warning("Failed to capture screenshot") - except Exception as e: - logger.warning(f"Error capturing screenshot: {e}") - - # 6. If still no screenshots, return original result - if not has_screenshots: - logger.debug(f"No visual data available for {tool_name}") - return result - - # 7. Perform visual analysis - return await self._enhance_result_with_visual_context(result, tool_name) + """Callback for LLMClient to handle visual analysis after tool execution.""" + return await _visual_analysis_callback_impl(self, result, tool_name, tool_call, backend) async def _enhance_result_with_visual_context(self, result: ToolResult, tool_name: str) -> ToolResult: - """ - Enhance tool result with visual analysis for grounding agent workflows. - """ - import asyncio - import base64 - - import litellm - - try: - metadata = getattr(result, "metadata", None) - if not metadata: - return result - - # Collect all screenshots - screenshots_bytes = [] - - # Check for multiple screenshots first - if metadata.get("screenshots"): - screenshots_list = metadata["screenshots"] - if isinstance(screenshots_list, list): - screenshots_bytes = [s for s in screenshots_list if s] - # Fall back to single screenshot - elif metadata.get("screenshot"): - screenshots_bytes = [metadata["screenshot"]] - - if not screenshots_bytes: - return result - - # Select key screenshots if there are too many - selected_screenshots = self._select_key_screenshots(screenshots_bytes, max_count=3) - - # Convert to base64 - visual_b64_list = [] - for visual_data in selected_screenshots: - if isinstance(visual_data, bytes): - visual_b64_list.append(base64.b64encode(visual_data).decode("utf-8")) - else: - visual_b64_list.append(visual_data) # Already base64 - - # Build prompt based on number of screenshots - num_screenshots = len(visual_b64_list) - - prompt = GroundingAgentPrompts.visual_analysis( - tool_name=tool_name, - num_screenshots=num_screenshots, - task_description=getattr(self, "_current_instruction", ""), - ) - - # Build content with text prompt + all images - content = [{"type": "text", "text": prompt}] - for visual_b64 in visual_b64_list: - content.append({"type": "image_url", "image_url": {"url": f"data:image/png;base64,{visual_b64}"}}) - - # Use dedicated visual analysis model if configured, otherwise use main LLM model - visual_model = self._visual_analysis_model or ( - self._llm_client.model if self._llm_client else "openrouter/anthropic/claude-sonnet-4.5" - ) - response = await asyncio.wait_for( - litellm.acompletion( - model=visual_model, - messages=[{"role": "user", "content": content}], - timeout=self._visual_analysis_timeout, - ), - timeout=self._visual_analysis_timeout + 5, - ) - - analysis = response.choices[0].message.content.strip() - - # Inject visual analysis into content - original_content = result.content or "(no text output)" - enhanced_content = f"{original_content}\n\n**Visual content**: {analysis}" - - # Create enhanced result - enhanced_result = ToolResult( - status=result.status, - content=enhanced_content, - error=result.error, - metadata={**metadata, "visual_analyzed": True, "visual_analysis": analysis}, - execution_time=result.execution_time, - ) - - logger.info(f"Enhanced {tool_name} result with visual analysis ({num_screenshots} screenshot(s))") - return enhanced_result - - except asyncio.TimeoutError: - logger.warning(f"Visual analysis timed out for {tool_name}, returning original result") - return result - except Exception as e: - logger.warning(f"Failed to analyze visual content for {tool_name}: {e}") - return result - - def _select_key_screenshots(self, screenshots: List[bytes], max_count: int = 3) -> List[bytes]: - """ - Select key screenshots if there are too many. - """ - if len(screenshots) <= max_count: - return screenshots - - selected_indices = set() - - # Always include last (final state) - selected_indices.add(len(screenshots) - 1) - - # If room, include first (initial state) - if max_count >= 2: - selected_indices.add(0) - - # Fill remaining slots with evenly spaced middle screenshots - remaining_slots = max_count - len(selected_indices) - if remaining_slots > 0: - # Calculate spacing - available_indices = [i for i in range(1, len(screenshots) - 1) if i not in selected_indices] - - if available_indices: - step = max(1, len(available_indices) // (remaining_slots + 1)) - for i in range(remaining_slots): - idx = min((i + 1) * step, len(available_indices) - 1) - if idx < len(available_indices): - selected_indices.add(available_indices[idx]) - - # Return screenshots in original order - selected = [screenshots[i] for i in sorted(selected_indices)] - - logger.debug( - f"Selected {len(selected)} screenshots at indices {sorted(selected_indices)} " - f"from total of {len(screenshots)}" - ) - - return selected - - def _get_workspace_path(self, context: Dict[str, Any]) -> Optional[str]: - """ - Get workspace directory path from context. - """ - return context.get("workspace_dir") - - def _scan_workspace_files( - self, - workspace_path: str, - recent_threshold: int = 600, # seconds - ) -> Dict[str, Any]: - """ - Scan workspace directory and collect file information. - - Args: - workspace_path: Path to workspace directory - recent_threshold: Threshold in seconds for recent files - - Returns: - Dictionary with file information: - - files: List of all filenames - - file_details: Dict mapping filename to file info (size, modified, age_seconds) - - recent_files: List of recently modified filenames - """ - import os - import time + """Enhance tool result with visual analysis for grounding agent workflows.""" + return await _enhance_result_with_visual_context_impl(self, result, tool_name) - result = {"files": [], "file_details": {}, "recent_files": []} + _select_key_screenshots = staticmethod(_select_key_screenshots_impl) - if not workspace_path or not os.path.exists(workspace_path): - return result + _get_workspace_path = staticmethod(_get_workspace_path_impl) - # Recording system files to exclude from workspace scanning - excluded_files = {"metadata.json", "traj.jsonl"} - - try: - current_time = time.time() - - for filename in os.listdir(workspace_path): - filepath = os.path.join(workspace_path, filename) - if os.path.isfile(filepath) and filename not in excluded_files: - result["files"].append(filename) - - # Get file stats - stat = os.stat(filepath) - file_info = { - "size": stat.st_size, - "modified": stat.st_mtime, - "age_seconds": current_time - stat.st_mtime, - } - result["file_details"][filename] = file_info - - # Track recently created/modified files - if file_info["age_seconds"] < recent_threshold: - result["recent_files"].append(filename) - - result["files"] = sorted(result["files"]) - - except Exception as e: - logger.debug(f"Error scanning workspace files: {e}") - - return result + _scan_workspace_files = staticmethod(_scan_workspace_files_impl) async def _check_workspace_artifacts(self, context: Dict[str, Any]) -> Dict[str, Any]: - """ - Check workspace directory for existing artifacts that might be relevant to the task. - Enhanced to detect if task might already be completed. - """ - import re - - workspace_info = {"has_files": False, "files": [], "file_details": {}, "recent_files": []} - - try: - # Get workspace path - workspace_path = self._get_workspace_path(context) - - # Scan workspace files - scan_result = self._scan_workspace_files(workspace_path, recent_threshold=600) - - if scan_result["files"]: - workspace_info["has_files"] = True - workspace_info["files"] = scan_result["files"] - workspace_info["file_details"] = scan_result["file_details"] - workspace_info["recent_files"] = scan_result["recent_files"] - - logger.info( - f"Grounding Agent: Found {len(scan_result['files'])} existing files in workspace " - f"({len(scan_result['recent_files'])} recent)" - ) - - # Check if instruction mentions specific filenames - instruction = context.get("instruction", "") - if instruction: - # Look for potential file references in instruction - potential_outputs = [] - # Match common file patterns: filename.ext, "filename", 'filename' - file_patterns = re.findall(r'["\']?([a-zA-Z0-9_\-]+\.[a-zA-Z0-9]+)["\']?', instruction) - for pattern in file_patterns: - if pattern in scan_result["files"]: - potential_outputs.append(pattern) - - if potential_outputs: - workspace_info["matching_files"] = potential_outputs - logger.info( - f"Grounding Agent: Found {len(potential_outputs)} files matching task: {potential_outputs}" - ) - - except Exception as e: - logger.debug(f"Could not check workspace artifacts: {e}") - - return workspace_info + """Check workspace for existing artifacts relevant to the task.""" + return await _check_workspace_artifacts_impl(self, context) def _build_iteration_feedback( self, iteration: int, llm_summary: Optional[str] = None, add_guidance: bool = True diff --git a/tests/test_grounding_tools.py b/tests/test_grounding_tools.py new file mode 100644 index 00000000..fe540741 --- /dev/null +++ b/tests/test_grounding_tools.py @@ -0,0 +1,174 @@ +"""Tests for openspace.agents.grounding.tools — tool retrieval helpers. + +Epic 5.9 extraction: _get_available_tools, _load_all_tools. +""" + +from __future__ import annotations + +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from openspace.agents.grounding.tools import _get_available_tools, _load_all_tools + + +class _FakeAgent: + """Minimal stand-in for GroundingAgent instance state.""" + + def __init__(self): + self.grounding_client = None + self._backend_scope = ["gui", "shell", "mcp"] + self._skill_context = None + self._active_skill_ids = [] + self._tool_retrieval_llm = None + self._llm_client = MagicMock() + self._skill_registry = None + self._skill_store = None + + @property + def has_skill_context(self) -> bool: + return self._skill_context is not None + + +# ── _get_available_tools ─────────────────────────────────────────── + + +class TestGetAvailableTools: + @pytest.mark.asyncio + async def test_returns_empty_when_no_grounding_client(self): + agent = _FakeAgent() + agent.grounding_client = None + result = await _get_available_tools(agent, "do something") + assert result == [] + + @pytest.mark.asyncio + async def test_calls_auto_search(self): + agent = _FakeAgent() + mock_gc = AsyncMock() + mock_gc.get_tools_with_auto_search = AsyncMock(return_value=["tool1", "tool2"]) + agent.grounding_client = mock_gc + + result = await _get_available_tools(agent, "open browser") + assert result == ["tool1", "tool2"] + mock_gc.get_tools_with_auto_search.assert_awaited_once() + + @pytest.mark.asyncio + async def test_adds_shell_backend_when_skill_context(self): + agent = _FakeAgent() + agent._backend_scope = ["gui", "mcp"] # no shell + agent._skill_context = "some skill guidance" + mock_gc = AsyncMock() + mock_gc.get_tools_with_auto_search = AsyncMock(return_value=[]) + agent.grounding_client = mock_gc + + await _get_available_tools(agent, "task") + + call_kwargs = mock_gc.get_tools_with_auto_search.call_args + backends = call_kwargs.kwargs.get("backend") or call_kwargs[1].get("backend") + from openspace.grounding.core.types import BackendType + + assert BackendType.SHELL in backends + + @pytest.mark.asyncio + async def test_fallback_on_auto_search_failure(self): + agent = _FakeAgent() + mock_gc = AsyncMock() + mock_gc.get_tools_with_auto_search = AsyncMock(side_effect=RuntimeError("boom")) + mock_gc.list_tools = AsyncMock(return_value=["fallback_tool"]) + agent.grounding_client = mock_gc + + result = await _get_available_tools(agent, "task") + assert "fallback_tool" in result + + @pytest.mark.asyncio + async def test_appends_retrieve_skill_tool(self): + agent = _FakeAgent() + mock_gc = AsyncMock() + mock_gc.get_tools_with_auto_search = AsyncMock(return_value=[]) + agent.grounding_client = mock_gc + + mock_registry = MagicMock() + mock_registry.list_skills.return_value = ["skill_a"] + agent._skill_registry = mock_registry + + with patch("openspace.agents.grounding.tools.RetrieveSkillTool", create=True) as MockRST: + # Patch the lazy import inside the function + mock_tool = MagicMock() + MockRST.return_value = mock_tool + + # Need to patch the import inside the function + import openspace.agents.grounding.tools as tools_mod + + with patch.dict("sys.modules", {"openspace.skill_engine.retrieve_tool": MagicMock(RetrieveSkillTool=MockRST)}): + result = await _get_available_tools(agent, "task") + assert mock_tool in result + + @pytest.mark.asyncio + async def test_uses_tool_retrieval_llm_when_set(self): + agent = _FakeAgent() + retrieval_llm = MagicMock() + agent._tool_retrieval_llm = retrieval_llm + mock_gc = AsyncMock() + mock_gc.get_tools_with_auto_search = AsyncMock(return_value=[]) + agent.grounding_client = mock_gc + + await _get_available_tools(agent, "task") + call_kwargs = mock_gc.get_tools_with_auto_search.call_args + assert call_kwargs.kwargs.get("llm_callable") is retrieval_llm + + +# ── _load_all_tools ──────────────────────────────────────────────── + + +class TestLoadAllTools: + @pytest.mark.asyncio + async def test_loads_from_all_backends(self): + agent = _FakeAgent() + agent._backend_scope = ["gui", "shell"] + mock_gc = AsyncMock() + mock_gc.list_tools = AsyncMock(side_effect=[["t1"], ["t2", "t3"]]) + + result = await _load_all_tools(agent, mock_gc) + assert result == ["t1", "t2", "t3"] + assert mock_gc.list_tools.await_count == 2 + + @pytest.mark.asyncio + async def test_skips_failed_backends(self): + agent = _FakeAgent() + agent._backend_scope = ["gui", "shell"] + mock_gc = AsyncMock() + mock_gc.list_tools = AsyncMock(side_effect=[RuntimeError("fail"), ["ok_tool"]]) + + result = await _load_all_tools(agent, mock_gc) + assert result == ["ok_tool"] + + @pytest.mark.asyncio + async def test_empty_when_all_backends_fail(self): + agent = _FakeAgent() + agent._backend_scope = ["gui"] + mock_gc = AsyncMock() + mock_gc.list_tools = AsyncMock(side_effect=RuntimeError("fail")) + + result = await _load_all_tools(agent, mock_gc) + assert result == [] + + +# ── Delegation seam tests ────────────────────────────────────────── + + +class TestToolsDelegationSeams: + """Verify grounding_agent.py properly delegates to tools module.""" + + def test_get_available_tools_delegates(self): + from openspace.agents.grounding_agent import GroundingAgent + + assert "_get_available_tools" in dir(GroundingAgent) + method = getattr(GroundingAgent, "_get_available_tools") + assert callable(method) + + def test_load_all_tools_delegates(self): + from openspace.agents.grounding_agent import GroundingAgent + + assert "_load_all_tools" in dir(GroundingAgent) + method = getattr(GroundingAgent, "_load_all_tools") + assert callable(method) diff --git a/tests/test_grounding_visual.py b/tests/test_grounding_visual.py new file mode 100644 index 00000000..6b021f2e --- /dev/null +++ b/tests/test_grounding_visual.py @@ -0,0 +1,209 @@ +"""Tests for openspace.agents.grounding.visual — visual analysis helpers. + +Epic 5.9 extraction: _visual_analysis_callback, _enhance_result_with_visual_context, +_select_key_screenshots. +""" + +from __future__ import annotations + +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from openspace.agents.grounding.visual import ( + _enhance_result_with_visual_context, + _select_key_screenshots, + _visual_analysis_callback, +) +from openspace.grounding.core.types import ToolResult, ToolStatus + + +def _make_result(**kwargs): + defaults = {"status": ToolStatus.SUCCESS, "content": "ok", "metadata": {}} + defaults.update(kwargs) + return ToolResult(**defaults) + + +def _make_tool_call(args_dict=None, args_str=None): + """Build a mock tool_call with .function.arguments.""" + tc = MagicMock() + if args_str is not None: + tc.function.arguments = args_str + elif args_dict is not None: + tc.function.arguments = args_dict + else: + tc.function.arguments = "{}" + return tc + + +class _FakeAgent: + """Minimal stand-in for GroundingAgent instance state.""" + + def __init__(self): + self._visual_analysis_model = None + self._llm_client = MagicMock(model="test-model") + self._visual_analysis_timeout = 5.0 + self._current_instruction = "test instruction" + + +# ── _select_key_screenshots (pure function) ──────────────────────── + + +class TestSelectKeyScreenshots: + def test_returns_all_when_under_max(self): + shots = [b"a", b"b"] + assert _select_key_screenshots(shots, max_count=3) == [b"a", b"b"] + + def test_returns_all_when_equal_max(self): + shots = [b"a", b"b", b"c"] + assert _select_key_screenshots(shots, max_count=3) == shots + + def test_selects_first_and_last(self): + shots = [b"0", b"1", b"2", b"3", b"4"] + result = _select_key_screenshots(shots, max_count=2) + assert result[0] == b"0" + assert result[-1] == b"4" + assert len(result) == 2 + + def test_selects_three_from_many(self): + shots = [bytes([i]) for i in range(10)] + result = _select_key_screenshots(shots, max_count=3) + assert len(result) == 3 + # First and last are always included + assert result[0] == shots[0] + assert result[-1] == shots[-1] + + def test_preserves_order(self): + shots = [bytes([i]) for i in range(7)] + result = _select_key_screenshots(shots, max_count=3) + indices = [shots.index(r) for r in result] + assert indices == sorted(indices) + + def test_max_count_one_returns_last(self): + shots = [b"a", b"b", b"c"] + result = _select_key_screenshots(shots, max_count=1) + assert len(result) == 1 + assert result[0] == b"c" + + +# ── _visual_analysis_callback ────────────────────────────────────── + + +class TestVisualAnalysisCallback: + @pytest.mark.asyncio + async def test_skip_when_meta_parameter_set(self): + agent = _FakeAgent() + result = _make_result() + tc = _make_tool_call(args_dict={"skip_visual_analysis": True}) + out = await _visual_analysis_callback(agent, result, "click", tc, "gui") + assert out is result # unchanged + + @pytest.mark.asyncio + async def test_skip_for_non_gui_backend(self): + agent = _FakeAgent() + result = _make_result() + tc = _make_tool_call() + out = await _visual_analysis_callback(agent, result, "run_shell", tc, "shell") + assert out is result + + @pytest.mark.asyncio + async def test_returns_original_when_no_visual_data_and_capture_fails(self): + agent = _FakeAgent() + result = _make_result(metadata={}) + tc = _make_tool_call() + + with patch("openspace.agents.grounding.visual.ScreenshotClient") as MockSC: + mock_client = AsyncMock() + mock_client.capture.return_value = None + MockSC.return_value = mock_client + out = await _visual_analysis_callback(agent, result, "click", tc, "gui") + assert out is result + + @pytest.mark.asyncio + async def test_captures_and_enhances_when_no_initial_visual(self): + agent = _FakeAgent() + result = _make_result(metadata={}) + tc = _make_tool_call() + + with patch("openspace.agents.grounding.visual.ScreenshotClient") as MockSC: + mock_client = AsyncMock() + mock_client.capture.return_value = b"screenshot_bytes" + MockSC.return_value = mock_client + + with patch( + "openspace.agents.grounding.visual._enhance_result_with_visual_context", + new_callable=AsyncMock, + ) as mock_enhance: + enhanced = _make_result(content="enhanced!") + mock_enhance.return_value = enhanced + out = await _visual_analysis_callback(agent, result, "click", tc, "gui") + assert out is enhanced + + @pytest.mark.asyncio + async def test_skip_visual_analysis_from_string_args(self): + agent = _FakeAgent() + result = _make_result() + tc = _make_tool_call(args_str='{"skip_visual_analysis": true}') + out = await _visual_analysis_callback(agent, result, "click", tc, "gui") + assert out is result + + @pytest.mark.asyncio + async def test_tolerates_unparseable_arguments(self): + """Should not crash when tool_call arguments are invalid JSON.""" + agent = _FakeAgent() + result = _make_result(metadata={}) + tc = _make_tool_call(args_str="not-json{{{") + + with patch("openspace.agents.grounding.visual.ScreenshotClient") as MockSC: + mock_client = AsyncMock() + mock_client.capture.return_value = None + MockSC.return_value = mock_client + out = await _visual_analysis_callback(agent, result, "click", tc, "gui") + assert out is result + + +# ── _enhance_result_with_visual_context ──────────────────────────── + + +class TestEnhanceResultWithVisualContext: + @pytest.mark.asyncio + async def test_returns_original_when_empty_metadata(self): + agent = _FakeAgent() + result = _make_result(metadata={}) + with patch.dict("sys.modules", {"litellm": MagicMock()}): + out = await _enhance_result_with_visual_context(agent, result, "click") + # No screenshots → returns original + assert out is result + + @pytest.mark.asyncio + async def test_returns_original_when_no_screenshots(self): + agent = _FakeAgent() + result = _make_result(metadata={"some_key": "val"}) + with patch.dict("sys.modules", {"litellm": MagicMock()}): + out = await _enhance_result_with_visual_context(agent, result, "click") + assert out is result + + +# ── Delegation seam tests ────────────────────────────────────────── + + +class TestVisualDelegationSeams: + """Verify grounding_agent.py properly delegates to visual module.""" + + def test_visual_analysis_callback_delegates(self): + from openspace.agents.grounding_agent import GroundingAgent + + assert "_visual_analysis_callback" in dir(GroundingAgent) + + def test_enhance_result_delegates(self): + from openspace.agents.grounding_agent import GroundingAgent + + assert "_enhance_result_with_visual_context" in dir(GroundingAgent) + + def test_select_key_screenshots_is_static(self): + from openspace.agents.grounding_agent import GroundingAgent + + # staticmethod binding — callable on both class and instance + assert callable(GroundingAgent._select_key_screenshots) + # Verify it's the same function + assert GroundingAgent._select_key_screenshots is _select_key_screenshots diff --git a/tests/test_grounding_workspace.py b/tests/test_grounding_workspace.py new file mode 100644 index 00000000..d225a126 --- /dev/null +++ b/tests/test_grounding_workspace.py @@ -0,0 +1,201 @@ +"""Tests for openspace.agents.grounding.workspace — workspace helpers. + +Epic 5.9 extraction: _get_workspace_path, _scan_workspace_files, +_check_workspace_artifacts. +""" + +from __future__ import annotations + +import os +import tempfile +import time + +import pytest + +from openspace.agents.grounding.workspace import ( + _check_workspace_artifacts, + _get_workspace_path, + _scan_workspace_files, +) + + +# ── _get_workspace_path (pure function) ──────────────────────────── + + +class TestGetWorkspacePath: + def test_returns_value_from_context(self): + assert _get_workspace_path({"workspace_dir": "/some/path"}) == "/some/path" + + def test_returns_none_when_missing(self): + assert _get_workspace_path({}) is None + + def test_returns_none_for_empty_context(self): + assert _get_workspace_path({"other": 123}) is None + + +# ── _scan_workspace_files (pure function) ────────────────────────── + + +class TestScanWorkspaceFiles: + def test_returns_empty_for_none_path(self): + result = _scan_workspace_files(None) + assert result["files"] == [] + assert result["file_details"] == {} + assert result["recent_files"] == [] + + def test_returns_empty_for_nonexistent_path(self): + result = _scan_workspace_files("/nonexistent/path/xyz") + assert result["files"] == [] + + def test_scans_real_directory(self): + with tempfile.TemporaryDirectory() as tmpdir: + # Create test files + for name in ["alpha.txt", "beta.py", "gamma.md"]: + with open(os.path.join(tmpdir, name), "w") as f: + f.write("content") + + result = _scan_workspace_files(tmpdir, recent_threshold=600) + assert sorted(result["files"]) == ["alpha.txt", "beta.py", "gamma.md"] + assert len(result["file_details"]) == 3 + # All files just created → should be recent + assert len(result["recent_files"]) == 3 + + def test_excludes_metadata_and_traj(self): + with tempfile.TemporaryDirectory() as tmpdir: + for name in ["metadata.json", "traj.jsonl", "real.txt"]: + with open(os.path.join(tmpdir, name), "w") as f: + f.write("x") + + result = _scan_workspace_files(tmpdir) + assert result["files"] == ["real.txt"] + + def test_skips_directories(self): + with tempfile.TemporaryDirectory() as tmpdir: + os.makedirs(os.path.join(tmpdir, "subdir")) + with open(os.path.join(tmpdir, "file.txt"), "w") as f: + f.write("data") + + result = _scan_workspace_files(tmpdir) + assert result["files"] == ["file.txt"] + + def test_old_files_not_in_recent(self): + with tempfile.TemporaryDirectory() as tmpdir: + filepath = os.path.join(tmpdir, "old.txt") + with open(filepath, "w") as f: + f.write("data") + # Set mtime to 2 hours ago + old_time = time.time() - 7200 + os.utime(filepath, (old_time, old_time)) + + result = _scan_workspace_files(tmpdir, recent_threshold=600) + assert "old.txt" in result["files"] + assert "old.txt" not in result["recent_files"] + + def test_file_details_contain_expected_keys(self): + with tempfile.TemporaryDirectory() as tmpdir: + with open(os.path.join(tmpdir, "test.txt"), "w") as f: + f.write("hello world") + + result = _scan_workspace_files(tmpdir) + details = result["file_details"]["test.txt"] + assert "size" in details + assert "modified" in details + assert "age_seconds" in details + assert details["size"] > 0 + + def test_files_sorted_alphabetically(self): + with tempfile.TemporaryDirectory() as tmpdir: + for name in ["zebra.txt", "apple.txt", "mango.txt"]: + with open(os.path.join(tmpdir, name), "w") as f: + f.write("x") + + result = _scan_workspace_files(tmpdir) + assert result["files"] == ["apple.txt", "mango.txt", "zebra.txt"] + + +# ── _check_workspace_artifacts ───────────────────────────────────── + + +class _FakeAgent: + """Minimal stand-in — workspace module calls module-level functions directly, + so agent is only used as first parameter for _check_workspace_artifacts.""" + pass + + +class TestCheckWorkspaceArtifacts: + @pytest.mark.asyncio + async def test_empty_when_no_workspace_dir(self): + agent = _FakeAgent() + result = await _check_workspace_artifacts(agent, {}) + assert result["has_files"] is False + assert result["files"] == [] + + @pytest.mark.asyncio + async def test_finds_files_in_workspace(self): + agent = _FakeAgent() + with tempfile.TemporaryDirectory() as tmpdir: + for name in ["report.pdf", "data.csv"]: + with open(os.path.join(tmpdir, name), "w") as f: + f.write("content") + + result = await _check_workspace_artifacts(agent, {"workspace_dir": tmpdir}) + assert result["has_files"] is True + assert "report.pdf" in result["files"] + assert "data.csv" in result["files"] + + @pytest.mark.asyncio + async def test_detects_matching_files_from_instruction(self): + agent = _FakeAgent() + with tempfile.TemporaryDirectory() as tmpdir: + with open(os.path.join(tmpdir, "output.png"), "w") as f: + f.write("image data") + + context = { + "workspace_dir": tmpdir, + "instruction": 'Generate "output.png" from the data', + } + result = await _check_workspace_artifacts(agent, context) + assert result.get("matching_files") == ["output.png"] + + @pytest.mark.asyncio + async def test_no_matching_files_when_instruction_empty(self): + agent = _FakeAgent() + with tempfile.TemporaryDirectory() as tmpdir: + with open(os.path.join(tmpdir, "file.txt"), "w") as f: + f.write("x") + + result = await _check_workspace_artifacts(agent, {"workspace_dir": tmpdir}) + assert "matching_files" not in result + + @pytest.mark.asyncio + async def test_resilient_to_errors(self): + """Should not raise even when context is weird.""" + agent = _FakeAgent() + result = await _check_workspace_artifacts(agent, {"workspace_dir": "/nonexistent/xyz"}) + assert result["has_files"] is False + + +# ── Delegation seam tests ────────────────────────────────────────── + + +class TestWorkspaceDelegationSeams: + """Verify grounding_agent.py properly delegates to workspace module.""" + + def test_get_workspace_path_is_static(self): + from openspace.agents.grounding_agent import GroundingAgent + + assert callable(GroundingAgent._get_workspace_path) + assert GroundingAgent._get_workspace_path is _get_workspace_path + + def test_scan_workspace_files_is_static(self): + from openspace.agents.grounding_agent import GroundingAgent + + assert callable(GroundingAgent._scan_workspace_files) + assert GroundingAgent._scan_workspace_files is _scan_workspace_files + + def test_check_workspace_artifacts_delegates(self): + from openspace.agents.grounding_agent import GroundingAgent + + assert "_check_workspace_artifacts" in dir(GroundingAgent) + method = getattr(GroundingAgent, "_check_workspace_artifacts") + assert callable(method) From 1b4604cac9ad62b3bba281fc6741391a5c35d7d6 Mon Sep 17 00:00:00 2001 From: Brian Krafft Date: Mon, 6 Apr 2026 10:47:47 -0700 Subject: [PATCH 2/2] fix(5.9): route internal calls through agent._method() for MRO safety - tools.py: _get_available_tools fallback calls agent._load_all_tools() - visual.py: _visual_analysis_callback calls agent._enhance_result_with_visual_context() - workspace.py: _check_workspace_artifacts calls agent._get_workspace_path/scan - Update test _FakeAgent classes with delegate methods - Found by GPT-5.4 /collab review - 1,683 passed, 127 skipped Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- openspace/agents/grounding/tools.py | 2 +- openspace/agents/grounding/visual.py | 2 +- openspace/agents/grounding/workspace.py | 8 ++++---- tests/test_grounding_tools.py | 5 +++++ tests/test_grounding_visual.py | 5 +++++ tests/test_grounding_workspace.py | 14 +++++++++++--- 6 files changed, 27 insertions(+), 9 deletions(-) diff --git a/openspace/agents/grounding/tools.py b/openspace/agents/grounding/tools.py index 1c1590be..b93b03ef 100644 --- a/openspace/agents/grounding/tools.py +++ b/openspace/agents/grounding/tools.py @@ -60,7 +60,7 @@ async def _get_available_tools(agent, task_description: Optional[str]) -> List: ) except Exception as e: logger.warning(f"Auto-search tools failed, falling back to full list: {e}") - tools = await _load_all_tools(agent, grounding_client) + tools = await agent._load_all_tools(grounding_client) # Append retrieve_skill tool when skill registry is available if agent._skill_registry and agent._skill_registry.list_skills(): diff --git a/openspace/agents/grounding/visual.py b/openspace/agents/grounding/visual.py index bccf8444..0ab97d30 100644 --- a/openspace/agents/grounding/visual.py +++ b/openspace/agents/grounding/visual.py @@ -78,7 +78,7 @@ async def _visual_analysis_callback( return result # 7. Perform visual analysis - return await _enhance_result_with_visual_context(agent, result, tool_name) + return await agent._enhance_result_with_visual_context(result, tool_name) async def _enhance_result_with_visual_context( diff --git a/openspace/agents/grounding/workspace.py b/openspace/agents/grounding/workspace.py index 38d92095..7dddbee8 100644 --- a/openspace/agents/grounding/workspace.py +++ b/openspace/agents/grounding/workspace.py @@ -88,11 +88,11 @@ async def _check_workspace_artifacts(agent, context: Dict[str, Any]) -> Dict[str workspace_info: Dict[str, Any] = {"has_files": False, "files": [], "file_details": {}, "recent_files": []} try: - # Get workspace path — call module-level function directly - workspace_path = _get_workspace_path(context) + # Get workspace path — route through agent for MRO preservation + workspace_path = agent._get_workspace_path(context) - # Scan workspace files — call module-level function directly - scan_result = _scan_workspace_files(workspace_path, recent_threshold=600) + # Scan workspace files — route through agent for MRO preservation + scan_result = agent._scan_workspace_files(workspace_path, recent_threshold=600) if scan_result["files"]: workspace_info["has_files"] = True diff --git a/tests/test_grounding_tools.py b/tests/test_grounding_tools.py index fe540741..e84ce43f 100644 --- a/tests/test_grounding_tools.py +++ b/tests/test_grounding_tools.py @@ -29,6 +29,11 @@ def __init__(self): def has_skill_context(self) -> bool: return self._skill_context is not None + async def _load_all_tools(self, grounding_client): + """Delegate for MRO — calls module function.""" + from openspace.agents.grounding.tools import _load_all_tools + return await _load_all_tools(self, grounding_client) + # ── _get_available_tools ─────────────────────────────────────────── diff --git a/tests/test_grounding_visual.py b/tests/test_grounding_visual.py index 6b021f2e..76908884 100644 --- a/tests/test_grounding_visual.py +++ b/tests/test_grounding_visual.py @@ -45,6 +45,11 @@ def __init__(self): self._visual_analysis_timeout = 5.0 self._current_instruction = "test instruction" + async def _enhance_result_with_visual_context(self, result, tool_name): + """Delegate for MRO — calls module function.""" + from openspace.agents.grounding.visual import _enhance_result_with_visual_context + return await _enhance_result_with_visual_context(self, result, tool_name) + # ── _select_key_screenshots (pure function) ──────────────────────── diff --git a/tests/test_grounding_workspace.py b/tests/test_grounding_workspace.py index d225a126..87bda497 100644 --- a/tests/test_grounding_workspace.py +++ b/tests/test_grounding_workspace.py @@ -117,9 +117,17 @@ def test_files_sorted_alphabetically(self): class _FakeAgent: - """Minimal stand-in — workspace module calls module-level functions directly, - so agent is only used as first parameter for _check_workspace_artifacts.""" - pass + """Minimal stand-in — workspace module now routes through agent._method().""" + + @staticmethod + def _get_workspace_path(context): + from openspace.agents.grounding.workspace import _get_workspace_path + return _get_workspace_path(context) + + @staticmethod + def _scan_workspace_files(workspace_path, recent_threshold=600): + from openspace.agents.grounding.workspace import _scan_workspace_files + return _scan_workspace_files(workspace_path, recent_threshold) class TestCheckWorkspaceArtifacts: