feat: add thought_search tool for querying Timmy's thinking history (#260)

Co-authored-by: Kimi Agent <kimi@timmy.local>
Co-committed-by: Kimi Agent <kimi@timmy.local>
This commit is contained in:
2026-03-15 19:35:58 -04:00
committed by hermes
parent 80aba0bf6d
commit bcbdc7d7cb
3 changed files with 203 additions and 6 deletions

View File

@@ -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()

View File

@@ -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"],
},
}