Epic 5.7: SkillContext + MessageSafety — grounding/ package

openspace/agents/grounding/__init__.py

@@ -0,0 +1 @@
+# openspace.agents.grounding — GroundingAgent subsystem package

openspace/agents/grounding/context.py

@@ -0,0 +1,65 @@
+"""Skill-context and skill-registry helpers for GroundingAgent.
+
+Pure side-effect functions that operate on agent instance state.
+Extracted from grounding_agent.py (Epic 5.7).
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, List, Optional
+
+from openspace.utils.logging import Logger
+
+if TYPE_CHECKING:
+    from openspace.skill_engine import SkillRegistry
+
+logger = Logger.get_logger("openspace.agents.grounding_agent")
+
+
+def set_skill_context(
+    agent,
+    context: str,
+    skill_ids: Optional[List[str]] = None,
+) -> None:
+    """Inject skill guidance into the agent's system prompt.
+
+    Called by ``OpenSpace.execute()`` before ``process()`` when skills
+    are matched.  The context is a formatted string built by
+    ``SkillRegistry.build_context_injection()``.
+
+    Args:
+        agent: GroundingAgent instance.
+        context: Formatted skill content for system prompt injection.
+        skill_ids: skill_id values of injected skills.
+    """
+    agent._skill_context = context if context else None
+    agent._active_skill_ids = skill_ids or []
+    if agent._skill_context:
+        logger.info(
+            f"Skill context set: {', '.join(agent._active_skill_ids) or '(unnamed)'}"
+        )
+
+
+def clear_skill_context(agent) -> None:
+    """Remove skill guidance (used before fallback execution)."""
+    if agent._skill_context:
+        logger.info(
+            f"Skill context cleared (was: {', '.join(agent._active_skill_ids)})"
+        )
+    agent._skill_context = None
+    agent._active_skill_ids = []
+
+
+def has_skill_context(agent) -> bool:
+    """Return True if skill context is currently set."""
+    return agent._skill_context is not None
+
+
+def set_skill_registry(agent, registry: Optional["SkillRegistry"]) -> None:
+    """Attach a SkillRegistry so the agent can offer ``retrieve_skill`` as a tool."""
+    agent._skill_registry = registry
+    if registry:
+        count = len(registry.list_skills())
+        logger.info(
+            f"Skill registry attached ({count} skill(s) available for mid-iteration retrieval)"
+        )

openspace/agents/grounding/messages.py

@@ -0,0 +1,123 @@
+"""Message safety helpers for GroundingAgent.
+
+Functions to cap oversized message content and truncate long
+conversation histories before LLM calls.
+Extracted from grounding_agent.py (Epic 5.7).
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Any, Dict, List
+
+from openspace.utils.logging import Logger
+
+logger = Logger.get_logger("openspace.agents.grounding_agent")
+
+# Maximum characters allowed in a single message content field.
+_MAX_SINGLE_CONTENT_CHARS = 30_000
+
+
+def cap_message_content(
+    messages: List[Dict[str, Any]],
+    cap: int = _MAX_SINGLE_CONTENT_CHARS,
+) -> List[Dict[str, Any]]:
+    """Truncate oversized individual message contents in-place.
+
+    Targets tool-result messages and assistant messages that can
+    carry enormous file contents (read_file on large CSVs/scripts).
+    System messages and the first user instruction are never touched.
+
+    Args:
+        messages: The message list (mutated in-place).
+        cap: Maximum character count per message.
+
+    Returns:
+        The same *messages* list (for chaining).
+    """
+    trimmed = 0
+    for msg in messages:
+        content = msg.get("content")
+        if not isinstance(content, str) or len(content) <= cap:
+            continue
+        if msg.get("role") == "system":
+            continue
+        original_len = len(content)
+        msg["content"] = (
+            content[: cap // 2]
+            + f"\n\n... [truncated {original_len - cap:,} chars] ...\n\n"
+            + content[-(cap // 2) :]
+        )
+        trimmed += 1
+    if trimmed:
+        logger.info(f"Capped {trimmed} oversized message(s) to {cap:,} chars each")
+    return messages
+
+
+def truncate_messages(
+    messages: List[Dict[str, Any]],
+    keep_recent: int = 8,
+    max_tokens_estimate: int = 120_000,
+    cap: int = _MAX_SINGLE_CONTENT_CHARS,
+) -> List[Dict[str, Any]]:
+    """Trim conversation history to fit within token budget.
+
+    Steps:
+    1. Cap any single oversized message (via :func:`cap_message_content`).
+    2. If total estimated tokens exceed *max_tokens_estimate*, keep only
+       the system messages, the first user instruction, and the most
+       recent *keep_recent* conversation rounds.
+
+    Args:
+        messages: Full message list.
+        keep_recent: Number of recent conversation rounds to preserve.
+        max_tokens_estimate: Approximate token budget.
+        cap: Per-message character cap (forwarded to cap_message_content).
+
+    Returns:
+        Possibly shortened message list.
+    """
+    messages = cap_message_content(messages, cap)
+
+    if len(messages) <= keep_recent + 2:  # +2 for system and initial user
+        return messages
+
+    total_text = json.dumps(messages, ensure_ascii=False)
+    estimated_tokens = len(total_text) // 4
+
+    if estimated_tokens < max_tokens_estimate:
+        return messages
+
+    logger.info(
+        f"Truncating message history: {len(messages)} messages, "
+        f"~{estimated_tokens:,} tokens -> keeping recent {keep_recent} rounds"
+    )
+
+    system_messages: List[Dict[str, Any]] = []
+    user_instruction = None
+    conversation_messages: List[Dict[str, Any]] = []
+
+    for msg in messages:
+        role = msg.get("role")
+        if role == "system":
+            system_messages.append(msg)
+        elif role == "user" and user_instruction is None:
+            user_instruction = msg
+        else:
+            conversation_messages.append(msg)
+
+    recent_messages = (
+        conversation_messages[-(keep_recent * 2) :] if conversation_messages else []
+    )
+
+    truncated = system_messages.copy()
+    if user_instruction:
+        truncated.append(user_instruction)
+    truncated.extend(recent_messages)
+
+    logger.info(
+        f"After truncation: {len(truncated)} messages, "
+        f"~{len(json.dumps(truncated, ensure_ascii=False)) // 4:,} tokens (estimated)"
+    )
+
+    return truncated

openspace/agents/grounding_agent.py

@@ -5,6 +5,17 @@
 from typing import TYPE_CHECKING, Any, Dict, List, Optional
 
 from openspace.agents.base import BaseAgent
+from openspace.agents.grounding.context import (
+    clear_skill_context as _clear_skill_context,
+    has_skill_context as _has_skill_context,
+    set_skill_context as _set_skill_context,
+    set_skill_registry as _set_skill_registry,
+)
+from openspace.agents.grounding.messages import (
+    _MAX_SINGLE_CONTENT_CHARS,
+    cap_message_content as _cap_message_content,
+    truncate_messages as _truncate_messages_impl,
+)
 from openspace.grounding.core.types import BackendType, ToolResult
 from openspace.platforms.screenshot import ScreenshotClient
 from openspace.prompts import GroundingAgentPrompts
@@ -86,116 +97,35 @@ def set_skill_context(
         context: str,
         skill_ids: Optional[List[str]] = None,
     ) -> None:
-        """Inject skill guidance into the agent's system prompt.
-
-        Called by ``OpenSpace.execute()`` before ``process()`` when skills
-        are matched.  The context is a formatted string built by
-        ``SkillRegistry.build_context_injection()``.
-
-        Args:
-            context: Formatted skill content for system prompt injection.
-            skill_ids: skill_id values of injected skills.
-        """
-        self._skill_context = context if context else None
-        self._active_skill_ids = skill_ids or []
-        if self._skill_context:
-            logger.info(f"Skill context set: {', '.join(self._active_skill_ids) or '(unnamed)'}")
+        """Inject skill guidance into the agent's system prompt."""
+        return _set_skill_context(self, context, skill_ids)
 
     def clear_skill_context(self) -> None:
         """Remove skill guidance (used before fallback execution)."""
-        if self._skill_context:
-            logger.info(f"Skill context cleared (was: {', '.join(self._active_skill_ids)})")
-        self._skill_context = None
-        self._active_skill_ids = []
+        return _clear_skill_context(self)
 
     @property
     def has_skill_context(self) -> bool:
-        return self._skill_context is not None
+        return _has_skill_context(self)
 
     def set_skill_registry(self, registry: Optional["SkillRegistry"]) -> None:
         """Attach a SkillRegistry so the agent can offer ``retrieve_skill`` as a tool."""
-        self._skill_registry = registry
-        if registry:
-            count = len(registry.list_skills())
-            logger.info(f"Skill registry attached ({count} skill(s) available for mid-iteration retrieval)")
+        return _set_skill_registry(self, registry)
 
-    _MAX_SINGLE_CONTENT_CHARS = 30_000
+    _MAX_SINGLE_CONTENT_CHARS = _MAX_SINGLE_CONTENT_CHARS
 
     @classmethod
     def _cap_message_content(cls, messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
-        """Truncate oversized individual message contents in-place.
-
-        Targets tool-result messages and assistant messages that can
-        carry enormous file contents (read_file on large CSVs/scripts).
-        System messages and the first user instruction are never touched.
-        """
-        cap = cls._MAX_SINGLE_CONTENT_CHARS
-        trimmed = 0
-        for msg in messages:
-            content = msg.get("content")
-            if not isinstance(content, str) or len(content) <= cap:
-                continue
-            if msg.get("role") == "system":
-                continue
-            original_len = len(content)
-            msg["content"] = (
-                content[: cap // 2]
-                + f"\n\n... [truncated {original_len - cap:,} chars] ...\n\n"
-                + content[-(cap // 2) :]
-            )
-            trimmed += 1
-        if trimmed:
-            logger.info(f"Capped {trimmed} oversized message(s) to {cap:,} chars each")
-        return messages
+        """Truncate oversized individual message contents in-place."""
+        return _cap_message_content(messages, cls._MAX_SINGLE_CONTENT_CHARS)
 
     def _truncate_messages(
         self, messages: List[Dict[str, Any]], keep_recent: int = 8, max_tokens_estimate: int = 120000
     ) -> List[Dict[str, Any]]:
-        # First: cap any single oversized message to prevent one huge
-        # tool-result from dominating the context window.
-        messages = self._cap_message_content(messages)
-
-        if len(messages) <= keep_recent + 2:  # +2 for system and initial user
-            return messages
-
-        total_text = json.dumps(messages, ensure_ascii=False)
-        estimated_tokens = len(total_text) // 4
-
-        if estimated_tokens < max_tokens_estimate:
-            return messages
-
-        logger.info(
-            f"Truncating message history: {len(messages)} messages, "
-            f"~{estimated_tokens:,} tokens -> keeping recent {keep_recent} rounds"
-        )
-
-        system_messages = []
-        user_instruction = None
-        conversation_messages = []
-
-        for msg in messages:
-            role = msg.get("role")
-            if role == "system":
-                system_messages.append(msg)
-            elif role == "user" and user_instruction is None:
-                user_instruction = msg
-            else:
-                conversation_messages.append(msg)
-
-        recent_messages = conversation_messages[-(keep_recent * 2) :] if conversation_messages else []
-
-        truncated = system_messages.copy()
-        if user_instruction:
-            truncated.append(user_instruction)
-        truncated.extend(recent_messages)
-
-        logger.info(
-            f"After truncation: {len(truncated)} messages, "
-            f"~{len(json.dumps(truncated, ensure_ascii=False)) // 4:,} tokens (estimated)"
+        return _truncate_messages_impl(
+            messages, keep_recent, max_tokens_estimate, cap=self._MAX_SINGLE_CONTENT_CHARS
         )
 
-        return truncated
-
     async def process(self, context: Dict[str, Any]) -> Dict[str, Any]:
         """
         Process a task execution request with multi-round iteration control.