diff --git a/src/timmy/thinking.py b/src/timmy/thinking.py
index f5fe297..178c0d3 100644
--- a/src/timmy/thinking.py
+++ b/src/timmy/thinking.py
@@ -1076,5 +1076,80 @@ class ThinkingEngine:
logger.debug("Failed to broadcast thought: %s", exc)
+def search_thoughts(query: str, seed_type: str | None = None, limit: int = 10) -> str:
+ """Search Timmy's thought history for reflections matching a query.
+
+ Use this tool when Timmy needs to recall his previous thoughts on a topic,
+ reflect on past insights, or build upon earlier reflections. This enables
+ self-awareness and continuity of thinking across time.
+
+ Args:
+ query: Search term to match against thought content (case-insensitive).
+ seed_type: Optional filter by thought category (e.g., 'existential',
+ 'swarm', 'sovereignty', 'creative', 'memory', 'observation').
+ limit: Maximum number of thoughts to return (default 10, max 50).
+
+ Returns:
+ Formatted string with matching thoughts, newest first, including
+ timestamps and seed types. Returns a helpful message if no matches found.
+ """
+ # Clamp limit to reasonable bounds
+ limit = max(1, min(limit, 50))
+
+ try:
+ engine = thinking_engine
+ db_path = engine._db_path
+
+ # Build query with optional seed_type filter
+ with _get_conn(db_path) as conn:
+ if seed_type:
+ rows = conn.execute(
+ """
+ SELECT id, content, seed_type, created_at
+ FROM thoughts
+ WHERE content LIKE ? AND seed_type = ?
+ ORDER BY created_at DESC
+ LIMIT ?
+ """,
+ (f"%{query}%", seed_type, limit),
+ ).fetchall()
+ else:
+ rows = conn.execute(
+ """
+ SELECT id, content, seed_type, created_at
+ FROM thoughts
+ WHERE content LIKE ?
+ ORDER BY created_at DESC
+ LIMIT ?
+ """,
+ (f"%{query}%", limit),
+ ).fetchall()
+
+ if not rows:
+ if seed_type:
+ return f'No thoughts found matching "{query}" with seed_type="{seed_type}".'
+ return f'No thoughts found matching "{query}".'
+
+ # Format results
+ lines = [f'Found {len(rows)} thought(s) matching "{query}":']
+ if seed_type:
+ lines[0] += f' [seed_type="{seed_type}"]'
+ lines.append("")
+
+ for row in rows:
+ ts = datetime.fromisoformat(row["created_at"])
+ local_ts = ts.astimezone()
+ time_str = local_ts.strftime("%Y-%m-%d %I:%M %p").lstrip("0")
+ seed = row["seed_type"]
+ content = row["content"].replace("\n", " ") # Flatten newlines for display
+ lines.append(f"[{time_str}] ({seed}) {content[:150]}")
+
+ return "\n".join(lines)
+
+ except Exception as exc:
+ logger.warning("Thought search failed: %s", exc)
+ return f"Error searching thoughts: {exc}"
+
+
# Module-level singleton
thinking_engine = ThinkingEngine()
diff --git a/src/timmy/tools.py b/src/timmy/tools.py
index e49268e..8f33686 100644
--- a/src/timmy/tools.py
+++ b/src/timmy/tools.py
@@ -619,6 +619,17 @@ def _register_gematria_tool(toolkit: Toolkit) -> None:
logger.debug("Gematria tool not available")
+def _register_thinking_tools(toolkit: Toolkit) -> None:
+ """Register thinking/introspection tools for self-reflection."""
+ try:
+ from timmy.thinking import search_thoughts
+
+ toolkit.register(search_thoughts, name="thought_search")
+ except (ImportError, AttributeError) as exc:
+ logger.warning("Tool execution failed (Thinking tools registration): %s", exc)
+ logger.debug("Thinking tools not available")
+
+
def create_full_toolkit(base_dir: str | Path | None = None):
"""Create a full toolkit with all available tools (for the orchestrator).
@@ -646,6 +657,7 @@ def create_full_toolkit(base_dir: str | Path | None = None):
_register_introspection_tools(toolkit)
_register_delegation_tools(toolkit)
_register_gematria_tool(toolkit)
+ _register_thinking_tools(toolkit)
# Gitea issue management is now provided by the gitea-mcp server
# (wired in as MCPTools in agent.py, not registered here)
@@ -837,6 +849,11 @@ def _introspection_tool_catalog() -> dict:
"description": "Search past conversation logs for messages, tool calls, errors, and decisions",
"available_in": ["orchestrator"],
},
+ "thought_search": {
+ "name": "Thought Search",
+ "description": "Query Timmy's own thought history for past reflections and insights",
+ "available_in": ["orchestrator"],
+ },
}
diff --git a/tests/timmy/test_thinking.py b/tests/timmy/test_thinking.py
index 5250c27..0acf012 100644
--- a/tests/timmy/test_thinking.py
+++ b/tests/timmy/test_thinking.py
@@ -833,6 +833,115 @@ def test_thinking_chain_api_404(client):
assert response.status_code == 404
+# ---------------------------------------------------------------------------
+# Thought search
+# ---------------------------------------------------------------------------
+
+
+def test_search_thoughts_basic(tmp_path):
+ """search_thoughts should find thoughts by content substring."""
+ from timmy import thinking
+
+ engine = _make_engine(tmp_path)
+ engine._store_thought("I wonder about sovereignty and freedom.", "existential")
+ engine._store_thought("The swarm is performing well today.", "swarm")
+ engine._store_thought("True sovereignty comes from local execution.", "sovereignty")
+
+ with patch.object(thinking, "thinking_engine", engine):
+ result = thinking.search_thoughts("sovereignty")
+ assert "Found 2 thought(s)" in result
+ assert "sovereignty" in result.lower()
+
+
+def test_search_thoughts_with_seed_type(tmp_path):
+ """search_thoughts should filter by seed_type when provided."""
+ from timmy import thinking
+
+ engine = _make_engine(tmp_path)
+ engine._store_thought("I wonder about sovereignty and freedom.", "existential")
+ engine._store_thought("True sovereignty comes from local execution.", "sovereignty")
+
+ with patch.object(thinking, "thinking_engine", engine):
+ result = thinking.search_thoughts("sovereignty", seed_type="sovereignty")
+ assert "Found 1 thought(s)" in result
+ assert '[seed_type="sovereignty"]' in result
+
+
+def test_search_thoughts_no_matches(tmp_path):
+ """search_thoughts should return helpful message when no matches found."""
+ from timmy import thinking
+
+ engine = _make_engine(tmp_path)
+ engine._store_thought("A thought about memory.", "memory")
+
+ with patch.object(thinking, "thinking_engine", engine):
+ result = thinking.search_thoughts("xyz_nonexistent")
+ assert "No thoughts found" in result
+
+
+def test_search_thoughts_limit(tmp_path):
+ """search_thoughts should respect the limit parameter."""
+ from timmy import thinking
+
+ engine = _make_engine(tmp_path)
+ for i in range(5):
+ engine._store_thought(f"Sovereignty thought number {i}.", "sovereignty")
+
+ with patch.object(thinking, "thinking_engine", engine):
+ result = thinking.search_thoughts("sovereignty", limit=3)
+ assert "Found 3 thought(s)" in result
+
+
+def test_search_thoughts_limit_bounds(tmp_path):
+ """search_thoughts should clamp limit to valid bounds."""
+ from timmy import thinking
+
+ engine = _make_engine(tmp_path)
+ engine._store_thought("A test thought.", "freeform")
+
+ with patch.object(thinking, "thinking_engine", engine):
+ # These should not raise errors - just clamp internally
+ result_low = thinking.search_thoughts("test", limit=0)
+ result_high = thinking.search_thoughts("test", limit=100)
+ # Both should execute (may return no results, but shouldn't crash)
+ assert isinstance(result_low, str)
+ assert isinstance(result_high, str)
+
+
+def test_search_thoughts_case_insensitive(tmp_path):
+ """search_thoughts should be case-insensitive (SQLite LIKE is case-insensitive)."""
+ from timmy import thinking
+
+ engine = _make_engine(tmp_path)
+ engine._store_thought("The SWARM is active today.", "swarm")
+
+ with patch.object(thinking, "thinking_engine", engine):
+ result_lower = thinking.search_thoughts("swarm")
+ result_upper = thinking.search_thoughts("SWARM")
+ result_mixed = thinking.search_thoughts("Swarm")
+
+ assert "Found 1 thought(s)" in result_lower
+ assert "Found 1 thought(s)" in result_upper
+ assert "Found 1 thought(s)" in result_mixed
+
+
+def test_search_thoughts_returns_formatted_output(tmp_path):
+ """search_thoughts should return formatted output with timestamps and seed types."""
+ from timmy import thinking
+
+ engine = _make_engine(tmp_path)
+ engine._store_thought("A memorable thought about existence.", "existential")
+
+ with patch.object(thinking, "thinking_engine", engine):
+ result = thinking.search_thoughts("memorable")
+ # Should contain timestamp-like content (year in 2026)
+ assert "2026-" in result or "2025-" in result
+ # Should contain seed type
+ assert "existential" in result
+ # Should contain the thought content
+ assert "memorable thought" in result
+
+
# ---------------------------------------------------------------------------
# _call_agent uses skip_mcp=True (#72)
# ---------------------------------------------------------------------------
@@ -884,8 +993,7 @@ async def test_call_agent_strips_think_tags(tmp_path):
mock_agent = AsyncMock()
mock_run = AsyncMock()
mock_run.content = (
- "Let me reason about this carefully..."
- "The actual thought content."
+ "Let me reason about this carefully...The actual thought content."
)
mock_agent.arun.return_value = mock_run
@@ -903,10 +1011,7 @@ async def test_call_agent_strips_multiline_think_tags(tmp_path):
mock_agent = AsyncMock()
mock_run = AsyncMock()
- mock_run.content = (
- "\nStep 1: analyze\nStep 2: synthesize\n\n"
- "Clean output here."
- )
+ mock_run.content = "\nStep 1: analyze\nStep 2: synthesize\n\nClean output here."
mock_agent.arun.return_value = mock_run
with patch("timmy.agent.create_timmy", return_value=mock_agent):