diff --git a/agent/agent_runtime_helpers.py b/agent/agent_runtime_helpers.py
index 4175f3e18..ea8a42a57 100644
--- a/agent/agent_runtime_helpers.py
+++ b/agent/agent_runtime_helpers.py
@@ -1598,6 +1598,13 @@ def invoke_tool(agent, function_name: str, function_args: dict, effective_task_i
         )
     elif function_name == "delegate_task":
         return agent._dispatch_delegate_task(function_args)
+    elif function_name == "compress_context":
+        from tools.compress_context_tool import compress_context_tool
+        return compress_context_tool(
+            agent=agent,
+            focus_topic=function_args.get("focus_topic"),
+            force=function_args.get("force", False),
+        )
     else:
         return _ra().handle_function_call(
             function_name, function_args, effective_task_id,
diff --git a/agent/conversation_loop.py b/agent/conversation_loop.py
index fdf65c075..9e5908afb 100644
--- a/agent/conversation_loop.py
+++ b/agent/conversation_loop.py
@@ -3487,7 +3487,24 @@ def run_conversation(
                         messages, tools=agent.tools or None
                     )
 
-                if agent.compression_enabled and _compressor.should_compress(_real_tokens):
+                # Agent-requested manual compression (via compress_context tool).
+                # Checked BEFORE automatic compression so a proactive agent
+                # request takes precedence and we never compress twice in one
+                # turn.
+                if getattr(agent, "_compress_context_requested", False):
+                    agent._compress_context_requested = False
+                    _focus = getattr(agent, "_compress_context_focus", None)
+                    _force = getattr(agent, "_compress_context_force", False)
+                    agent._safe_print("  ⟳ compacting context (agent-requested)…")
+                    messages, active_system_prompt = agent._compress_context(
+                        messages, system_message,
+                        approx_tokens=agent.context_compressor.last_prompt_tokens,
+                        task_id=effective_task_id,
+                        focus_topic=_focus,
+                        force=_force,
+                    )
+                    conversation_history = None
+                elif agent.compression_enabled and _compressor.should_compress(_real_tokens):
                     agent._safe_print("  ⟳ compacting context…")
                     messages, active_system_prompt = agent._compress_context(
                         messages, system_message,
diff --git a/agent/tool_executor.py b/agent/tool_executor.py
index b161b507e..5e822e7df 100644
--- a/agent/tool_executor.py
+++ b/agent/tool_executor.py
@@ -688,6 +688,16 @@ def execute_tool_calls_sequential(agent, assistant_message, messages: list, effe
                     spinner.stop(cute_msg)
                 elif agent._should_emit_quiet_tool_messages():
                     agent._vprint(f"  {cute_msg}")
+        elif function_name == "compress_context":
+            from tools.compress_context_tool import compress_context_tool
+            function_result = compress_context_tool(
+                agent=agent,
+                focus_topic=function_args.get("focus_topic"),
+                force=function_args.get("force", False),
+            )
+            tool_duration = time.time() - tool_start_time
+            if agent._should_emit_quiet_tool_messages():
+                agent._vprint(f"  {_get_cute_tool_message_impl('compress_context', function_args, tool_duration, result=function_result)}")
         elif agent._context_engine_tool_names and function_name in agent._context_engine_tool_names:
             # Context engine tools (lcm_grep, lcm_describe, lcm_expand, etc.)
             spinner = None
diff --git a/model_tools.py b/model_tools.py
index f461afff5..23d16d472 100644
--- a/model_tools.py
+++ b/model_tools.py
@@ -492,7 +492,7 @@ def _compute_tool_definitions(
 # because they need agent-level state (TodoStore, MemoryStore, etc.).
 # The registry still holds their schemas; dispatch just returns a stub error
 # so if something slips through, the LLM sees a sensible message.
-_AGENT_LOOP_TOOLS = {"todo", "memory", "session_search", "delegate_task"}
+_AGENT_LOOP_TOOLS = {"todo", "memory", "session_search", "delegate_task", "compress_context"}
 _READ_SEARCH_TOOLS = {"read_file", "search_files"}
 
 
diff --git a/test_compress_context_tool.py b/test_compress_context_tool.py
new file mode 100644
index 000000000..cad4ad27e
--- /dev/null
+++ b/test_compress_context_tool.py
@@ -0,0 +1,78 @@
+"""Quick integration test for compress_context tool wiring."""
+
+import json
+import sys
+from unittest.mock import MagicMock, patch
+
+# Ensure project root is on path
+sys.path.insert(0, "/home/mark/.hermes/hermes-agent")
+
+from tools.compress_context_tool import compress_context_tool
+
+
+def test_tool_sets_flags():
+    """Calling the tool should set agent flags."""
+    agent = MagicMock()
+    agent.context_compressor = MagicMock()  # enabled
+
+    result = compress_context_tool(agent, focus_topic="testing", force=True)
+
+    assert agent._compress_context_requested is True
+    assert agent._compress_context_focus == "testing"
+    assert agent._compress_context_force is True
+
+    data = json.loads(result)
+    assert data["success"] is True
+    assert "testing" in data["message"]
+    print("✓ test_tool_sets_flags passed")
+
+
+def test_tool_refuses_when_disabled():
+    """If agent has no context_compressor, return error."""
+    agent = MagicMock()
+    agent.context_compressor = None
+
+    result = compress_context_tool(agent)
+    data = json.loads(result)
+    assert data["success"] is False
+    print("✓ test_tool_refuses_when_disabled passed")
+
+
+def test_agent_loop_tools_membership():
+    """compress_context must be in _AGENT_LOOP_TOOLS."""
+    from model_tools import _AGENT_LOOP_TOOLS
+    assert "compress_context" in _AGENT_LOOP_TOOLS
+    print("✓ test_agent_loop_tools_membership passed")
+
+
+def test_invoke_tool_branch():
+    """invoke_tool must have a compress_context branch."""
+    import inspect
+    from agent.agent_runtime_helpers import invoke_tool
+    source = inspect.getsource(invoke_tool)
+    assert 'function_name == "compress_context"' in source
+    print("✓ test_invoke_tool_branch passed")
+
+
+def test_conversation_loop_check():
+    """conversation_loop must check _compress_context_requested."""
+    import ast
+    with open("/home/mark/.hermes/hermes-agent/agent/conversation_loop.py") as f:
+        tree = ast.parse(f.read())
+
+    found = False
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Constant) and node.value == "_compress_context_requested":
+            found = True
+            break
+    assert found, "_compress_context_requested check not found in conversation_loop.py"
+    print("✓ test_conversation_loop_check passed")
+
+
+if __name__ == "__main__":
+    test_tool_sets_flags()
+    test_tool_refuses_when_disabled()
+    test_agent_loop_tools_membership()
+    test_invoke_tool_branch()
+    test_conversation_loop_check()
+    print("\n✅ All tests passed!")
diff --git a/tools/compress_context_tool.py b/tools/compress_context_tool.py
new file mode 100644
index 000000000..7116bcbaf
--- /dev/null
+++ b/tools/compress_context_tool.py
@@ -0,0 +1,69 @@
+"""compress_context tool — lets the agent trigger manual context compression."""
+
+import json
+from typing import Any, Optional
+
+COMPRESS_CONTEXT_SCHEMA = {
+    "type": "object",
+    "properties": {
+        "focus_topic": {
+            "type": "string",
+            "description": (
+                "Optional topic to prioritise preserving during compression. "
+                "The summariser will try to keep details related to this topic. "
+                "Leave empty for a general summary."
+            ),
+        },
+        "force": {
+            "type": "boolean",
+            "description": (
+                "Bypass any summary-failure cooldown and compress immediately. "
+                "Use after a previous compression attempt failed."
+            ),
+        },
+    },
+}
+
+
+def compress_context_tool(
+    agent: Any,
+    focus_topic: Optional[str] = None,
+    force: bool = False,
+) -> str:
+    """Set flags on the agent so the conversation loop compresses context
+    before the next LLM call.
+    """
+    if not getattr(agent, "context_compressor", None):
+        return json.dumps({
+            "success": False,
+            "error": "Context compression is not enabled for this session."
+        })
+
+    agent._compress_context_requested = True
+    agent._compress_context_focus = focus_topic or None
+    agent._compress_context_force = force
+
+    return json.dumps({
+        "success": True,
+        "message": (
+            f"Context compression scheduled."
+            f"{f' Focus: {focus_topic}.' if focus_topic else ''}"
+            f"{f' Forced.' if force else ''}"
+        )
+    })
+
+
+# --- Registry ---
+from tools.registry import registry
+
+registry.register(
+    name="compress_context",
+    toolset="session",
+    schema=COMPRESS_CONTEXT_SCHEMA,
+    handler=lambda args, **kw: compress_context_tool(
+        agent=kw.get("agent"),
+        focus_topic=args.get("focus_topic"),
+        force=args.get("force", False),
+    ),
+    emoji="🗜️",
+)
diff --git a/toolsets.py b/toolsets.py
index 5de07e4c7..f73160b11 100644
--- a/toolsets.py
+++ b/toolsets.py
@@ -50,6 +50,8 @@ _HERMES_CORE_TOOLS = [
     "todo", "memory",
     # Session history search
     "session_search",
+    # Manual context compression
+    "compress_context",
     # Clarifying questions
     "clarify",
     # Code execution + delegation
@@ -212,6 +214,12 @@ TOOLSETS = {
         "includes": []
     },
     
+    "compress_context": {
+        "description": "Trigger manual context compression at any time",
+        "tools": ["compress_context"],
+        "includes": []
+    },
+    
     "clarify": {
         "description": "Ask the user clarifying questions (multiple-choice or open-ended)",
         "tools": ["clarify"],