diff --git a/openspace/agents/grounding/results.py b/openspace/agents/grounding/results.py new file mode 100644 index 00000000..9dc1e517 --- /dev/null +++ b/openspace/agents/grounding/results.py @@ -0,0 +1,285 @@ +"""Results, telemetry, and iteration helpers for GroundingAgent. + +Pure functions and agent-delegating helpers for building iteration +feedback, final summaries, task-completion checks, and recording. +Extracted from grounding_agent.py (Epic 5.10). +""" + +from __future__ import annotations + +import copy +import json +from typing import Any, Dict, List, Optional + +from openspace.prompts import GroundingAgentPrompts +from openspace.utils.logging import Logger + +logger = Logger.get_logger("openspace.agents.grounding_agent") + + +# ── Pure functions (no agent parameter) ──────────────────────────── + + +def build_iteration_feedback( + iteration: int, + llm_summary: Optional[str] = None, + add_guidance: bool = True, +) -> Optional[Dict[str, str]]: + """Build feedback message to add to next iteration. + + Returns ``None`` when *llm_summary* is falsy. + """ + if not llm_summary: + return None + + feedback_content = GroundingAgentPrompts.iteration_feedback( + iteration=iteration, llm_summary=llm_summary, add_guidance=add_guidance + ) + + return {"role": "system", "content": feedback_content} + + +def remove_previous_guidance(messages: List[Dict[str, Any]]) -> None: + """Remove guidance section from previous iteration feedback messages. + + Mutates *messages* in-place. + """ + for msg in messages: + if msg.get("role") == "system": + content = msg.get("content", "") + # Check if this is an iteration feedback message with guidance + if "## Iteration" in content and "Summary" in content and "---" in content: + # Remove everything from "---" onwards (the guidance part) + summary_only = content.split("---")[0].strip() + msg["content"] = summary_only + + +def format_tool_executions(all_tool_results: List[Dict]) -> List[Dict]: + """Format raw tool-result dicts into a serialisable execution list.""" + executions = [] + for tr in all_tool_results: + tool_result_obj = tr.get("result") + tool_call = tr.get("tool_call") + + status = "unknown" + if hasattr(tool_result_obj, "status"): + status_obj = tool_result_obj.status + status = getattr(status_obj, "value", status_obj) + + # Extract tool_name and arguments from tool_call object (litellm format) + tool_name = "unknown" + arguments: dict = {} + if tool_call is not None: + if hasattr(tool_call, "function"): + # tool_call is an object with .function attribute + tool_name = getattr(tool_call.function, "name", "unknown") + args_raw = getattr(tool_call.function, "arguments", "{}") + if isinstance(args_raw, str): + try: + arguments = json.loads(args_raw) if args_raw.strip() else {} + except json.JSONDecodeError: + arguments = {} + else: + arguments = args_raw if isinstance(args_raw, dict) else {} + elif isinstance(tool_call, dict): + # Fallback: tool_call is a dict + func = tool_call.get("function", {}) + tool_name = func.get("name", "unknown") + args_raw = func.get("arguments", "{}") + if isinstance(args_raw, str): + try: + arguments = json.loads(args_raw) if args_raw.strip() else {} + except json.JSONDecodeError: + arguments = {} + else: + arguments = args_raw if isinstance(args_raw, dict) else {} + + executions.append( + { + "tool_name": tool_name, + "arguments": arguments, + "backend": tr.get("backend"), + "server_name": tr.get("server_name"), + "status": status, + "content": tool_result_obj.content if hasattr(tool_result_obj, "content") else None, + "error": tool_result_obj.error if hasattr(tool_result_obj, "error") else None, + "execution_time": tool_result_obj.execution_time + if hasattr(tool_result_obj, "execution_time") + else None, + "metadata": tool_result_obj.metadata if hasattr(tool_result_obj, "metadata") else {}, + } + ) + return executions + + +def check_task_completion(messages: List[Dict]) -> bool: + """Return True if the last assistant message contains the COMPLETE token.""" + for msg in reversed(messages): + if msg.get("role") == "assistant": + content = msg.get("content", "") + return GroundingAgentPrompts.TASK_COMPLETE in content + return False + + +def extract_last_assistant_message(messages: List[Dict]) -> str: + """Return content of the last assistant message, or ``""``.""" + for msg in reversed(messages): + if msg.get("role") == "assistant": + return msg.get("content", "") + return "" + + +# ── Agent-dependent functions ────────────────────────────────────── + + +async def generate_final_summary( + agent, + instruction: str, + messages: List[Dict], + iterations: int, +) -> tuple[str, bool, List[Dict]]: + """Generate final summary across all iterations for reporting to upper layer. + + Returns: + tuple[str, bool, List[Dict]]: (summary_text, success_flag, context_used) + """ + final_summary_prompt = { + "role": "user", + "content": GroundingAgentPrompts.final_summary(instruction=instruction, iterations=iterations), + } + + clean_messages = [] + for msg in messages: + # Skip tool result messages + if msg.get("role") == "tool": + continue + # Copy message and remove tool_calls if present + clean_msg = msg.copy() + if "tool_calls" in clean_msg: + del clean_msg["tool_calls"] + clean_messages.append(clean_msg) + + clean_messages.append(final_summary_prompt) + + # Save context for return + context_for_return = copy.deepcopy(clean_messages) + + try: + # Call LLMClient to generate final summary (without tools) + summary_response = await agent._llm_client.complete( + messages=clean_messages, tools=None, execute_tools=False + ) + + final_summary = summary_response.get("message", {}).get("content", "") + + if final_summary: + logger.info(f"Generated final summary: {final_summary[:200]}...") + return final_summary, True, context_for_return + else: + logger.warning("LLM returned empty final summary") + return ( + f"Task completed after {iterations} iteration(s). Check execution history for details.", + True, + context_for_return, + ) + + except Exception as e: + logger.error(f"Error generating final summary: {e}") + return ( + f"Task completed after {iterations} iteration(s), but failed to generate summary.", + False, + context_for_return, + ) + + +async def build_final_result( + agent, + instruction: str, + messages: List[Dict], + all_tool_results: List[Dict], + iterations: int, + max_iterations: int, + iteration_contexts: List[Dict] = None, + retrieved_tools_list: List[Dict] = None, + search_debug_info: Dict[str, Any] = None, +) -> Dict[str, Any]: + """Build the final execution result dict. + + Routes ``_check_task_completion``, ``_format_tool_executions``, and + ``_extract_last_assistant_message`` through *agent* to preserve MRO. + """ + is_complete = agent._check_task_completion(messages) + + tool_executions = agent._format_tool_executions(all_tool_results) + + result = { + "instruction": instruction, + "step": agent.step, + "iterations": iterations, + "tool_executions": tool_executions, + "messages": messages, + "iteration_contexts": iteration_contexts or [], + "retrieved_tools_list": retrieved_tools_list or [], + "search_debug_info": search_debug_info, + "active_skills": list(agent._active_skill_ids), + "keep_session": True, + } + + if is_complete: + logger.info("Task completed with marker") + last_response = agent._extract_last_assistant_message(messages) + result["response"] = last_response.replace(GroundingAgentPrompts.TASK_COMPLETE, "").strip() + result["status"] = "success" + else: + result["response"] = agent._extract_last_assistant_message(messages) + result["status"] = "incomplete" + result["warning"] = ( + f"Task reached max iterations ({max_iterations}) without completion. " + f"This may indicate the task needs more steps or clarification." + ) + + return result + + +async def record_agent_execution( + agent, + result: Dict[str, Any], + instruction: str, +) -> None: + """Record agent execution to recording manager. + + No-op when ``agent._recording_manager`` is falsy. + """ + if not agent._recording_manager: + return + + # Extract tool execution summary + tool_summary = [] + if result.get("tool_executions"): + for exec_info in result["tool_executions"]: + tool_summary.append( + { + "tool": exec_info.get("tool_name", "unknown"), + "backend": exec_info.get("backend", "unknown"), + "status": exec_info.get("status", "unknown"), + } + ) + + await agent._recording_manager.record_agent_action( + agent_name=agent.name, + action_type="execute", + input_data={"instruction": instruction}, + reasoning={ + "response": result.get("response", ""), + "tools_selected": tool_summary, + }, + output_data={ + "status": result.get("status", "unknown"), + "iterations": result.get("iterations", 0), + "num_tool_executions": len(result.get("tool_executions", [])), + }, + metadata={ + "step": agent.step, + "instruction": instruction, + }, + ) diff --git a/openspace/agents/grounding_agent.py b/openspace/agents/grounding_agent.py index 4d550e06..6128fc38 100644 --- a/openspace/agents/grounding_agent.py +++ b/openspace/agents/grounding_agent.py @@ -1,7 +1,5 @@ from __future__ import annotations -import copy -import json from typing import TYPE_CHECKING, Any, Dict, List, Optional from openspace.agents.base import BaseAgent @@ -23,6 +21,16 @@ construct_messages as _construct_messages, default_system_prompt as _default_system_prompt, ) +from openspace.agents.grounding.results import ( + build_final_result as _build_final_result_impl, + build_iteration_feedback as _build_iteration_feedback_impl, + check_task_completion as _check_task_completion_impl, + extract_last_assistant_message as _extract_last_assistant_message_impl, + format_tool_executions as _format_tool_executions_impl, + generate_final_summary as _generate_final_summary_impl, + record_agent_execution as _record_agent_execution_impl, + remove_previous_guidance as _remove_previous_guidance_impl, +) from openspace.agents.grounding.tools import ( _get_available_tools as _get_available_tools_impl, _load_all_tools as _load_all_tools_impl, @@ -38,7 +46,6 @@ _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 if TYPE_CHECKING: @@ -185,91 +192,22 @@ async def _check_workspace_artifacts(self, context: Dict[str, Any]) -> Dict[str, """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 - ) -> Optional[Dict[str, str]]: - """ - Build feedback message to add to next iteration. - """ - if not llm_summary: - return None + # ── Results / telemetry delegates (Epic 5.10) ──────────────────── - feedback_content = GroundingAgentPrompts.iteration_feedback( - iteration=iteration, llm_summary=llm_summary, add_guidance=add_guidance - ) + _build_iteration_feedback = staticmethod(_build_iteration_feedback_impl) - return {"role": "system", "content": feedback_content} + _remove_previous_guidance = staticmethod(_remove_previous_guidance_impl) - def _remove_previous_guidance(self, messages: List[Dict[str, Any]]) -> None: - """ - Remove guidance section from previous iteration feedback messages. - """ - for msg in messages: - if msg.get("role") == "system": - content = msg.get("content", "") - # Check if this is an iteration feedback message with guidance - if "## Iteration" in content and "Summary" in content and "---" in content: - # Remove everything from "---" onwards (the guidance part) - summary_only = content.split("---")[0].strip() - msg["content"] = summary_only + _format_tool_executions = staticmethod(_format_tool_executions_impl) + + _check_task_completion = staticmethod(_check_task_completion_impl) + + _extract_last_assistant_message = staticmethod(_extract_last_assistant_message_impl) async def _generate_final_summary( self, instruction: str, messages: List[Dict], iterations: int ) -> tuple[str, bool, List[Dict]]: - """ - Generate final summary across all iterations for reporting to upper layer. - - Returns: - tuple[str, bool, List[Dict]]: (summary_text, success_flag, context_used) - - summary_text: The generated summary or error message - - success_flag: True if summary was generated successfully, False otherwise - - context_used: The cleaned messages used for generating summary - """ - final_summary_prompt = { - "role": "user", - "content": GroundingAgentPrompts.final_summary(instruction=instruction, iterations=iterations), - } - - clean_messages = [] - for msg in messages: - # Skip tool result messages - if msg.get("role") == "tool": - continue - # Copy message and remove tool_calls if present - clean_msg = msg.copy() - if "tool_calls" in clean_msg: - del clean_msg["tool_calls"] - clean_messages.append(clean_msg) - - clean_messages.append(final_summary_prompt) - - # Save context for return - context_for_return = copy.deepcopy(clean_messages) - - try: - # Call LLMClient to generate final summary (without tools) - summary_response = await self._llm_client.complete(messages=clean_messages, tools=None, execute_tools=False) - - final_summary = summary_response.get("message", {}).get("content", "") - - if final_summary: - logger.info(f"Generated final summary: {final_summary[:200]}...") - return final_summary, True, context_for_return - else: - logger.warning("LLM returned empty final summary") - return ( - f"Task completed after {iterations} iteration(s). Check execution history for details.", - True, - context_for_return, - ) - - except Exception as e: - logger.error(f"Error generating final summary: {e}") - return ( - f"Task completed after {iterations} iteration(s), but failed to generate summary: {str(e)}", - False, - context_for_return, - ) + return await _generate_final_summary_impl(self, instruction, messages, iterations) async def _build_final_result( self, @@ -282,170 +220,18 @@ async def _build_final_result( retrieved_tools_list: List[Dict] = None, search_debug_info: Dict[str, Any] = None, ) -> Dict[str, Any]: - """ - Build final execution result. - - Args: - instruction: Original instruction - messages: Complete conversation history (including all iteration summaries) - all_tool_results: All tool execution results - iterations: Number of iterations performed - max_iterations: Maximum allowed iterations - iteration_contexts: Context snapshots for each iteration - retrieved_tools_list: List of tools retrieved for this task - search_debug_info: Debug info from tool search (similarity scores, LLM selections) - """ - is_complete = self._check_task_completion(messages) - - tool_executions = self._format_tool_executions(all_tool_results) - - result = { - "instruction": instruction, - "step": self.step, - "iterations": iterations, - "tool_executions": tool_executions, - "messages": messages, - "iteration_contexts": iteration_contexts or [], - "retrieved_tools_list": retrieved_tools_list or [], - "search_debug_info": search_debug_info, - "active_skills": list(self._active_skill_ids), - "keep_session": True, - } - - if is_complete: - logger.info("Task completed with marker") - # Use LLM's own completion response directly (no extra LLM call needed) - # LLM already generates a summary before outputting - last_response = self._extract_last_assistant_message(messages) - # Remove the token from response for cleaner output - result["response"] = last_response.replace(GroundingAgentPrompts.TASK_COMPLETE, "").strip() - result["status"] = "success" - - # [DISABLED] Extra LLM call to generate final summary - # final_summary, summary_success, final_summary_context = await self._generate_final_summary( - # instruction=instruction, - # messages=messages, - # iterations=iterations - # ) - # result["response"] = final_summary - # result["final_summary_context"] = final_summary_context - else: - result["response"] = self._extract_last_assistant_message(messages) - result["status"] = "incomplete" - result["warning"] = ( - f"Task reached max iterations ({max_iterations}) without completion. " - f"This may indicate the task needs more steps or clarification." - ) - - return result - - def _format_tool_executions(self, all_tool_results: List[Dict]) -> List[Dict]: - executions = [] - for tr in all_tool_results: - tool_result_obj = tr.get("result") - tool_call = tr.get("tool_call") - - status = "unknown" - if hasattr(tool_result_obj, "status"): - status_obj = tool_result_obj.status - status = getattr(status_obj, "value", status_obj) - - # Extract tool_name and arguments from tool_call object (litellm format) - tool_name = "unknown" - arguments = {} - if tool_call is not None: - if hasattr(tool_call, "function"): - # tool_call is an object with .function attribute - tool_name = getattr(tool_call.function, "name", "unknown") - args_raw = getattr(tool_call.function, "arguments", "{}") - if isinstance(args_raw, str): - try: - arguments = json.loads(args_raw) if args_raw.strip() else {} - except json.JSONDecodeError: - arguments = {} - else: - arguments = args_raw if isinstance(args_raw, dict) else {} - elif isinstance(tool_call, dict): - # Fallback: tool_call is a dict - func = tool_call.get("function", {}) - tool_name = func.get("name", "unknown") - args_raw = func.get("arguments", "{}") - if isinstance(args_raw, str): - try: - arguments = json.loads(args_raw) if args_raw.strip() else {} - except json.JSONDecodeError: - arguments = {} - else: - arguments = args_raw if isinstance(args_raw, dict) else {} - - executions.append( - { - "tool_name": tool_name, - "arguments": arguments, - "backend": tr.get("backend"), - "server_name": tr.get("server_name"), - "status": status, - "content": tool_result_obj.content if hasattr(tool_result_obj, "content") else None, - "error": tool_result_obj.error if hasattr(tool_result_obj, "error") else None, - "execution_time": tool_result_obj.execution_time - if hasattr(tool_result_obj, "execution_time") - else None, - "metadata": tool_result_obj.metadata if hasattr(tool_result_obj, "metadata") else {}, - } - ) - return executions - - def _check_task_completion(self, messages: List[Dict]) -> bool: - for msg in reversed(messages): - if msg.get("role") == "assistant": - content = msg.get("content", "") - return GroundingAgentPrompts.TASK_COMPLETE in content - return False - - def _extract_last_assistant_message(self, messages: List[Dict]) -> str: - for msg in reversed(messages): - if msg.get("role") == "assistant": - return msg.get("content", "") - return "" + return await _build_final_result_impl( + self, + instruction, + messages, + all_tool_results, + iterations, + max_iterations, + iteration_contexts, + retrieved_tools_list, + search_debug_info, + ) async def _record_agent_execution(self, result: Dict[str, Any], instruction: str) -> None: - """ - Record agent execution to recording manager. + return await _record_agent_execution_impl(self, result, instruction) - Args: - result: Execution result - instruction: Original instruction - """ - if not self._recording_manager: - return - - # Extract tool execution summary - tool_summary = [] - if result.get("tool_executions"): - for exec_info in result["tool_executions"]: - tool_summary.append( - { - "tool": exec_info.get("tool_name", "unknown"), - "backend": exec_info.get("backend", "unknown"), - "status": exec_info.get("status", "unknown"), - } - ) - - await self._recording_manager.record_agent_action( - agent_name=self.name, - action_type="execute", - input_data={"instruction": instruction}, - reasoning={ - "response": result.get("response", ""), - "tools_selected": tool_summary, - }, - output_data={ - "status": result.get("status", "unknown"), - "iterations": result.get("iterations", 0), - "num_tool_executions": len(result.get("tool_executions", [])), - }, - metadata={ - "step": self.step, - "instruction": instruction, - }, - ) diff --git a/tests/test_grounding_results.py b/tests/test_grounding_results.py new file mode 100644 index 00000000..9955d8da --- /dev/null +++ b/tests/test_grounding_results.py @@ -0,0 +1,538 @@ +"""Tests for openspace.agents.grounding.results — results & telemetry helpers. + +Epic 5.10 extraction: build_iteration_feedback, remove_previous_guidance, +format_tool_executions, check_task_completion, extract_last_assistant_message, +generate_final_summary, build_final_result, record_agent_execution. +""" + +from __future__ import annotations + +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from openspace.agents.grounding.results import ( + build_final_result, + build_iteration_feedback, + check_task_completion, + extract_last_assistant_message, + format_tool_executions, + generate_final_summary, + record_agent_execution, + remove_previous_guidance, +) +from openspace.prompts import GroundingAgentPrompts + + +# ── helpers ──────────────────────────────────────────────────────── + + +class _FakeAgent: + """Minimal stand-in for GroundingAgent instance state.""" + + def __init__(self): + self.step = 3 + self.name = "TestAgent" + self._llm_client = AsyncMock() + self._recording_manager = None + self._active_skill_ids = ["skill-a"] + + # These are the staticmethod bindings on real GroundingAgent; + # for _build_final_result tests we attach module-level functions. + self._check_task_completion = check_task_completion + self._format_tool_executions = format_tool_executions + self._extract_last_assistant_message = extract_last_assistant_message + + +# ══════════════════════════════════════════════════════════════════════ +# build_iteration_feedback +# ══════════════════════════════════════════════════════════════════════ + + +class TestBuildIterationFeedback: + def test_returns_none_when_no_summary(self): + assert build_iteration_feedback(1, llm_summary=None) is None + + def test_returns_none_for_empty_string(self): + assert build_iteration_feedback(1, llm_summary="") is None + + def test_returns_system_message_with_summary(self): + result = build_iteration_feedback(2, llm_summary="Did stuff") + assert result is not None + assert result["role"] == "system" + assert "## Iteration 2" in result["content"] + assert "Did stuff" in result["content"] + + def test_guidance_included_by_default(self): + result = build_iteration_feedback(1, llm_summary="progress") + assert "---" in result["content"] + + def test_guidance_excluded_when_disabled(self): + result = build_iteration_feedback(1, llm_summary="progress", add_guidance=False) + assert "---" not in result["content"] + + +# ══════════════════════════════════════════════════════════════════════ +# remove_previous_guidance +# ══════════════════════════════════════════════════════════════════════ + + +class TestRemovePreviousGuidance: + def test_strips_guidance_from_iteration_feedback(self): + msgs = [ + { + "role": "system", + "content": "## Iteration 1 Summary\nDid things\n---\nContinue with iteration 2.", + } + ] + remove_previous_guidance(msgs) + assert "---" not in msgs[0]["content"] + assert "Did things" in msgs[0]["content"] + + def test_leaves_non_feedback_messages_alone(self): + msgs = [ + {"role": "user", "content": "Hello"}, + {"role": "system", "content": "You are an assistant"}, + ] + remove_previous_guidance(msgs) + assert msgs[0]["content"] == "Hello" + assert msgs[1]["content"] == "You are an assistant" + + def test_handles_empty_list(self): + msgs = [] + remove_previous_guidance(msgs) + assert msgs == [] + + +# ══════════════════════════════════════════════════════════════════════ +# format_tool_executions +# ══════════════════════════════════════════════════════════════════════ + + +class TestFormatToolExecutions: + def test_empty_list(self): + assert format_tool_executions([]) == [] + + def test_litellm_object_format(self): + """tool_call is an object with .function attribute (litellm style).""" + func = SimpleNamespace(name="click", arguments='{"x": 10}') + tool_call = SimpleNamespace(function=func) + result_obj = SimpleNamespace( + status=SimpleNamespace(value="success"), + content="clicked", + error=None, + execution_time=0.5, + metadata={"k": "v"}, + ) + raw = [{"result": result_obj, "tool_call": tool_call, "backend": "gui", "server_name": "s1"}] + + out = format_tool_executions(raw) + assert len(out) == 1 + assert out[0]["tool_name"] == "click" + assert out[0]["arguments"] == {"x": 10} + assert out[0]["status"] == "success" + assert out[0]["backend"] == "gui" + + def test_dict_format(self): + """tool_call is a dict (fallback path).""" + tool_call = {"function": {"name": "type_text", "arguments": '{"text": "hi"}'}} + result_obj = SimpleNamespace( + status="ok", + content="typed", + error=None, + execution_time=0.1, + metadata={}, + ) + raw = [{"result": result_obj, "tool_call": tool_call}] + + out = format_tool_executions(raw) + assert out[0]["tool_name"] == "type_text" + assert out[0]["arguments"] == {"text": "hi"} + assert out[0]["status"] == "ok" + + def test_none_tool_call(self): + """tool_call is None — should produce 'unknown' tool_name.""" + result_obj = SimpleNamespace() + raw = [{"result": result_obj, "tool_call": None}] + out = format_tool_executions(raw) + assert out[0]["tool_name"] == "unknown" + + def test_invalid_json_arguments(self): + """Malformed JSON arguments should produce empty dict.""" + func = SimpleNamespace(name="bad", arguments="not-json") + tool_call = SimpleNamespace(function=func) + result_obj = SimpleNamespace() + raw = [{"result": result_obj, "tool_call": tool_call}] + out = format_tool_executions(raw) + assert out[0]["arguments"] == {} + + +# ══════════════════════════════════════════════════════════════════════ +# check_task_completion +# ══════════════════════════════════════════════════════════════════════ + + +class TestCheckTaskCompletion: + def test_true_when_complete_in_last_assistant(self): + msgs = [ + {"role": "user", "content": "do it"}, + {"role": "assistant", "content": f"Done {GroundingAgentPrompts.TASK_COMPLETE}"}, + ] + assert check_task_completion(msgs) is True + + def test_false_when_not_present(self): + msgs = [ + {"role": "assistant", "content": "Still working..."}, + ] + assert check_task_completion(msgs) is False + + def test_false_when_no_assistant_messages(self): + msgs = [{"role": "user", "content": "hello"}] + assert check_task_completion(msgs) is False + + def test_false_on_empty_list(self): + assert check_task_completion([]) is False + + +# ══════════════════════════════════════════════════════════════════════ +# extract_last_assistant_message +# ══════════════════════════════════════════════════════════════════════ + + +class TestExtractLastAssistantMessage: + def test_extracts_last_assistant(self): + msgs = [ + {"role": "assistant", "content": "first"}, + {"role": "user", "content": "next"}, + {"role": "assistant", "content": "second"}, + ] + assert extract_last_assistant_message(msgs) == "second" + + def test_returns_empty_when_none(self): + msgs = [{"role": "user", "content": "hello"}] + assert extract_last_assistant_message(msgs) == "" + + def test_returns_empty_for_empty_list(self): + assert extract_last_assistant_message([]) == "" + + +# ══════════════════════════════════════════════════════════════════════ +# generate_final_summary +# ══════════════════════════════════════════════════════════════════════ + + +class TestGenerateFinalSummary: + @pytest.mark.asyncio + async def test_returns_summary_text(self): + agent = _FakeAgent() + agent._llm_client.complete.return_value = { + "message": {"content": "Summary of execution"} + } + msgs = [{"role": "assistant", "content": "did things"}] + + text, success, ctx = await generate_final_summary(agent, "do stuff", msgs, 2) + + assert text == "Summary of execution" + assert success is True + assert len(ctx) > 0 + agent._llm_client.complete.assert_awaited_once() + + @pytest.mark.asyncio + async def test_handles_empty_response(self): + agent = _FakeAgent() + agent._llm_client.complete.return_value = {"message": {"content": ""}} + msgs = [{"role": "assistant", "content": "work"}] + + text, success, _ = await generate_final_summary(agent, "inst", msgs, 3) + + assert "3 iteration(s)" in text + assert success is True + + @pytest.mark.asyncio + async def test_handles_exception(self): + agent = _FakeAgent() + agent._llm_client.complete.side_effect = RuntimeError("LLM down") + msgs = [{"role": "assistant", "content": "work"}] + + text, success, _ = await generate_final_summary(agent, "inst", msgs, 1) + + assert "failed to generate summary" in text + assert "LLM down" not in text # str(e) must NOT leak to caller (security) + assert success is False + + @pytest.mark.asyncio + async def test_strips_tool_messages_and_tool_calls(self): + agent = _FakeAgent() + agent._llm_client.complete.return_value = {"message": {"content": "ok"}} + msgs = [ + {"role": "system", "content": "sys"}, + {"role": "assistant", "content": "resp", "tool_calls": [{"id": "tc1"}]}, + {"role": "tool", "content": "result"}, + ] + + _, _, ctx = await generate_final_summary(agent, "inst", msgs, 1) + + # tool messages stripped, tool_calls key removed + roles = [m["role"] for m in ctx] + assert "tool" not in roles + assert not any("tool_calls" in m for m in ctx) + + +# ══════════════════════════════════════════════════════════════════════ +# build_final_result +# ══════════════════════════════════════════════════════════════════════ + + +class TestBuildFinalResult: + @pytest.mark.asyncio + async def test_complete_case(self): + agent = _FakeAgent() + msgs = [ + {"role": "assistant", "content": f"All done. {GroundingAgentPrompts.TASK_COMPLETE}"}, + ] + + result = await build_final_result( + agent, + instruction="do it", + messages=msgs, + all_tool_results=[], + iterations=2, + max_iterations=10, + ) + + assert result["status"] == "success" + assert GroundingAgentPrompts.TASK_COMPLETE not in result["response"] + assert "All done." in result["response"] + assert result["step"] == 3 + assert result["active_skills"] == ["skill-a"] + + @pytest.mark.asyncio + async def test_incomplete_case(self): + agent = _FakeAgent() + msgs = [ + {"role": "assistant", "content": "Still going..."}, + ] + + result = await build_final_result( + agent, + instruction="do it", + messages=msgs, + all_tool_results=[], + iterations=10, + max_iterations=10, + ) + + assert result["status"] == "incomplete" + assert "max iterations" in result["warning"] + assert result["response"] == "Still going..." + + @pytest.mark.asyncio + async def test_passes_optional_fields(self): + agent = _FakeAgent() + msgs = [{"role": "assistant", "content": "x"}] + + result = await build_final_result( + agent, + instruction="i", + messages=msgs, + all_tool_results=[], + iterations=1, + max_iterations=5, + iteration_contexts=[{"iter": 1}], + retrieved_tools_list=[{"name": "t"}], + search_debug_info={"scores": []}, + ) + + assert result["iteration_contexts"] == [{"iter": 1}] + assert result["retrieved_tools_list"] == [{"name": "t"}] + assert result["search_debug_info"] == {"scores": []} + + +# ══════════════════════════════════════════════════════════════════════ +# record_agent_execution +# ══════════════════════════════════════════════════════════════════════ + + +class TestRecordAgentExecution: + @pytest.mark.asyncio + async def test_calls_recording_manager(self): + agent = _FakeAgent() + agent._recording_manager = AsyncMock() + + result = { + "response": "done", + "status": "success", + "iterations": 2, + "tool_executions": [ + {"tool_name": "click", "backend": "gui", "status": "success"}, + ], + } + + await record_agent_execution(agent, result, "do it") + + agent._recording_manager.record_agent_action.assert_awaited_once() + call_kwargs = agent._recording_manager.record_agent_action.call_args.kwargs + assert call_kwargs["agent_name"] == "TestAgent" + assert call_kwargs["action_type"] == "execute" + assert call_kwargs["metadata"]["step"] == 3 + + @pytest.mark.asyncio + async def test_skips_when_no_recording_manager(self): + agent = _FakeAgent() + agent._recording_manager = None + + # Should not raise + await record_agent_execution(agent, {"tool_executions": []}, "do it") + + @pytest.mark.asyncio + async def test_handles_empty_tool_executions(self): + agent = _FakeAgent() + agent._recording_manager = AsyncMock() + + result = {"response": "x", "status": "ok", "iterations": 1, "tool_executions": []} + await record_agent_execution(agent, result, "inst") + + call_kwargs = agent._recording_manager.record_agent_action.call_args.kwargs + assert call_kwargs["reasoning"]["tools_selected"] == [] + + +# ══════════════════════════════════════════════════════════════════════ +# Delegation seam tests (GroundingAgent → results module) +# ══════════════════════════════════════════════════════════════════════ + + +class TestGroundingAgentDelegation: + """Verify GroundingAgent thin delegates route to the results module.""" + + def _make_agent(self): + """Build a GroundingAgent with mocked dependencies.""" + from openspace.agents.grounding_agent import GroundingAgent + + agent = GroundingAgent.__new__(GroundingAgent) + # Minimal state to avoid __init__ side-effects + agent._backend_scope = ["shell"] + agent._llm_client = AsyncMock() + agent._grounding_client = None + agent._recording_manager = None + agent._system_prompt = "sys" + agent._max_iterations = 5 + agent._visual_analysis_timeout = 10.0 + agent._tool_retrieval_llm = None + agent._visual_analysis_model = None + agent._skill_context = None + agent._active_skill_ids = [] + agent._skill_registry = None + agent._last_tools = [] + agent._step = 0 + agent._name = "test" + return agent + + # Pure staticmethod bindings + + def test_build_iteration_feedback_delegates(self): + agent = self._make_agent() + result = agent._build_iteration_feedback(1, llm_summary="ok") + assert result["role"] == "system" + assert "## Iteration 1" in result["content"] + + def test_remove_previous_guidance_delegates(self): + agent = self._make_agent() + msgs = [ + {"role": "system", "content": "## Iteration 1 Summary\nX\n---\nGuidance"}, + ] + agent._remove_previous_guidance(msgs) + assert "---" not in msgs[0]["content"] + + def test_format_tool_executions_delegates(self): + agent = self._make_agent() + assert agent._format_tool_executions([]) == [] + + def test_check_task_completion_delegates(self): + agent = self._make_agent() + msgs = [{"role": "assistant", "content": f"x {GroundingAgentPrompts.TASK_COMPLETE}"}] + assert agent._check_task_completion(msgs) is True + + def test_extract_last_assistant_message_delegates(self): + agent = self._make_agent() + msgs = [{"role": "assistant", "content": "hi"}] + assert agent._extract_last_assistant_message(msgs) == "hi" + + # Async thin delegates + + @pytest.mark.asyncio + async def test_generate_final_summary_delegates(self): + agent = self._make_agent() + agent._llm_client.complete.return_value = {"message": {"content": "summary"}} + msgs = [{"role": "assistant", "content": "work"}] + + text, ok, _ = await agent._generate_final_summary("inst", msgs, 1) + assert text == "summary" + assert ok is True + + @pytest.mark.asyncio + async def test_build_final_result_delegates(self): + agent = self._make_agent() + msgs = [{"role": "assistant", "content": f"Done {GroundingAgentPrompts.TASK_COMPLETE}"}] + + result = await agent._build_final_result( + instruction="i", + messages=msgs, + all_tool_results=[], + iterations=1, + max_iterations=5, + ) + assert result["status"] == "success" + + @pytest.mark.asyncio + async def test_mro_override_respected_in_build_final_result(self): + """Adversarial MRO test: subclass overrides _check_task_completion + and _build_final_result MUST call the override, not the module function.""" + from openspace.agents.grounding_agent import GroundingAgent + + SENTINEL = "SUBCLASS_OVERRIDE_SENTINEL" + + class SubGroundingAgent(GroundingAgent): + @staticmethod + def _check_task_completion(messages): + # Always returns True regardless of content + return True + + @staticmethod + def _extract_last_assistant_message(messages): + return SENTINEL + + agent = SubGroundingAgent.__new__(SubGroundingAgent) + agent._backend_scope = ["shell"] + agent._llm_client = AsyncMock() + agent._recording_manager = None + agent._active_skill_ids = [] + agent._step = 0 + agent._name = "sub-test" + # Give it a staticmethod binding for format (not overridden) + agent._format_tool_executions = format_tool_executions + + # Messages WITHOUT the TASK_COMPLETE token — base would return False + msgs = [{"role": "assistant", "content": "no complete token here"}] + result = await agent._build_final_result( + instruction="test", + messages=msgs, + all_tool_results=[], + iterations=1, + max_iterations=5, + ) + # Subclass override makes _check_task_completion return True + assert result["status"] == "success" + # Subclass override makes _extract_last_assistant_message return sentinel + assert result["response"] == SENTINEL + + @pytest.mark.asyncio + async def test_record_agent_execution_delegates(self): + agent = self._make_agent() + agent._recording_manager = AsyncMock() + + await agent._record_agent_execution( + {"response": "x", "status": "ok", "iterations": 1, "tool_executions": []}, + "inst", + ) + agent._recording_manager.record_agent_action.assert_awaited_once()