Skip to content

Instantly share code, notes, and snippets.

@Carleslc

Carleslc/search_sessions.py

Last active March 2, 2026 10:51
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save Carleslc/88f781370da53bcd16ecd97811bc25b2 to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save Carleslc/88f781370da53bcd16ecd97811bc25b2 to your computer and use it in GitHub Desktop.
Download ZIP
Search Messages OpenClaw Script
Raw
search_sessions.py
#!/usr/bin/env python3
"""
Search OpenClaw session messages
Usage:
./search_sessions.py "search text" # Search in all sessions (exact match)
./search_sessions.py "search text" --session current # Search only current session
./search_sessions.py "search text" --session <UUID> # Search specific session
./search_sessions.py "search text" --role assistant # Filter by role
./search_sessions.py "search text" --limit 5 # Limit results (newest first by default)
./search_sessions.py --list # List all sessions (newest first)
./search_sessions.py --message <id> --session <UUID> # Find specific message
./search_sessions.py --message <id> --export # Export message + context to .md
./search_sessions.py --history # Show conversation history (newest first by default)
./search_sessions.py --history --thinking # Include all intermediate steps
./search_sessions.py --history --from 2026-01-01 # Filter from date
./search_sessions.py --history --to 2026-12-31 # Filter to date
./search_sessions.py --history --role user --limit 10 # Filter history by role
./search_sessions.py --history --role system # Show only system notifications (model changes, etc.)
./search_sessions.py --history --sort asc # Oldest first instead of newest
"""
import json
import sys
import argparse
import re
from pathlib import Path
from datetime import datetime, timezone, timedelta
from typing import List, Dict, Optional, Tuple
def get_sessions_dir(agent: str = "main") -> Path:
"""Get the sessions directory for a specific agent"""
return Path.home() / ".openclaw" / "agents" / agent / "sessions"
def get_sessions_json(agent: str = "main") -> Path:
"""Get the sessions.json file for a specific agent"""
return get_sessions_dir(agent) / "sessions.json"
def parse_date(date_str: str) -> Optional[datetime]:
"""Parse date string in various formats"""
if not date_str:
return None
# Try more specific formats first (with time), then less specific (date only)
formats = [
"%Y-%m-%dT%H:%M:%S", # 2026-02-01T10:30:00
"%Y-%m-%d %H:%M:%S", # 2026-02-01 10:30:00
"%Y-%m-%d %H:%M", # 2026-02-01 10:30
"%Y-%m-%d", # 2026-02-01
"%d/%m/%Y", # 01/02/2026
"%d-%m-%Y", # 01-02-2026
]
for fmt in formats:
try:
return datetime.strptime(date_str, fmt)
except ValueError:
continue
return None
def parse_timestamp(timestamp_str: str) -> Optional[datetime]:
"""Parse ISO timestamp to datetime"""
if not timestamp_str:
return None
try:
if timestamp_str.endswith("Z"):
dt = datetime.fromisoformat(timestamp_str.replace("Z", "+00:00"))
else:
dt = datetime.fromisoformat(timestamp_str)
return dt.astimezone() # Convert to local time
except:
return None
def get_local_tz_offset() -> str:
"""Get local timezone offset as string (e.g., GMT+1, GMT-5)"""
now = datetime.now()
utc_now = datetime.now(timezone.utc).replace(tzinfo=None)
offset_seconds = (now - utc_now).total_seconds()
offset_hours = round(offset_seconds / 3600) # Round to handle floating point errors
if offset_hours >= 0:
return f"GMT+{offset_hours}"
else:
return f"GMT{offset_hours}"
def format_timestamp(timestamp_str: str, include_tz: bool = True) -> str:
"""Format timestamp to local timezone with optional timezone indicator"""
if not timestamp_str:
return "unknown"
try:
# Parse ISO timestamp
if timestamp_str.endswith("Z"):
dt = datetime.fromisoformat(timestamp_str.replace("Z", "+00:00"))
else:
dt = datetime.fromisoformat(timestamp_str)
# Convert to local time
local_dt = dt.astimezone()
if include_tz:
tz_offset = get_local_tz_offset()
return f"{local_dt.strftime('%Y-%m-%d %H:%M:%S')} {tz_offset}"
else:
return local_dt.strftime("%Y-%m-%d %H:%M:%S")
except:
return timestamp_str
def get_current_session(agent: str = "main") -> Optional[Path]:
"""Find the current active session from sessions.json"""
try:
sessions_json = get_sessions_json(agent)
sessions_dir = get_sessions_dir(agent)
if not sessions_json.exists():
return None
with open(sessions_json) as f:
sessions_data = json.load(f)
# Get main session (key is "agent:<agent>:main")
main_session = sessions_data.get(f"agent:{agent}:main")
if not main_session:
return None
session_id = main_session.get("sessionId")
if not session_id:
return None
# Build path to session file
session_file = sessions_dir / f"{session_id}.jsonl"
if session_file.exists():
return session_file
return None
except Exception:
return None
def find_session_by_id(session_id: str, agent: str = "main") -> Optional[Path]:
"""Find session file by UUID (partial match supported)"""
sessions_dir = get_sessions_dir(agent)
for jsonl_file in sessions_dir.glob("*.jsonl"):
if session_id in jsonl_file.stem:
return jsonl_file
return None
def get_session_start_timestamp(jsonl_file: Path) -> datetime:
"""Get the timestamp of the first message in a session file"""
try:
with open(jsonl_file) as f:
first_line = json.loads(f.readline())
timestamp = first_line.get("timestamp", "")
if timestamp:
if timestamp.endswith("Z"):
dt = datetime.fromisoformat(timestamp.replace("Z", "+00:00"))
else:
dt = datetime.fromisoformat(timestamp)
return dt.astimezone()
except:
pass
# Fallback to file modification time
return datetime.fromtimestamp(jsonl_file.stat().st_mtime)
def get_session_end_timestamp(jsonl_file: Path) -> datetime:
"""Get the timestamp of the last message in a session file"""
last_timestamp = None
try:
with open(jsonl_file) as f:
for line in f:
try:
data = json.loads(line)
timestamp = data.get("timestamp", "")
if timestamp:
last_timestamp = timestamp
except:
continue
if last_timestamp:
if last_timestamp.endswith("Z"):
dt = datetime.fromisoformat(last_timestamp.replace("Z", "+00:00"))
else:
dt = datetime.fromisoformat(last_timestamp)
return dt.astimezone()
except:
pass
# Fallback to file modification time
return datetime.fromtimestamp(jsonl_file.stat().st_mtime)
def get_sorted_sessions(sort_order: str = "desc", agent: str = "main") -> List[Path]:
"""Get all session files sorted by timestamp.
For 'asc' (oldest first): sort by first message timestamp
For 'desc' (newest first): sort by last message timestamp
"""
sessions_dir = get_sessions_dir(agent)
session_files = list(sessions_dir.glob("*.jsonl"))
reverse_order = (sort_order == "desc")
# Use appropriate timestamp based on sort order
if sort_order == "desc":
# For newest first, sort by last message (so session with most recent message comes first)
key_func = get_session_end_timestamp
else:
# For oldest first, sort by first message
key_func = get_session_start_timestamp
return sorted(session_files, key=key_func, reverse=reverse_order)
def list_sessions(limit: Optional[int] = None, date_from: Optional[str] = None,
date_to: Optional[str] = None, agent: str = "main", sort_order: str = "desc"):
"""List all sessions with basic info, sorted by date"""
sessions = []
tz_offset = get_local_tz_offset()
sessions_dir = get_sessions_dir(agent)
# Parse date filters
from_dt = parse_date(date_from) if date_from else None
to_dt = parse_date(date_to) if date_to else None
# Make to_dt inclusive
if to_dt:
if to_dt.hour == 0 and to_dt.minute == 0 and to_dt.second == 0:
to_dt = to_dt + timedelta(days=1)
else:
to_dt = to_dt + timedelta(seconds=1)
# Get current session UUID
current_session = get_current_session(agent)
current_uuid = current_session.stem if current_session else None
for jsonl_file in sessions_dir.glob("*.jsonl"):
stat = jsonl_file.stat()
is_current = jsonl_file.stem == current_uuid
# Try to read first line for timestamp
try:
with open(jsonl_file) as f:
first_line = json.loads(f.readline())
timestamp = first_line.get("timestamp", "")
if timestamp:
if timestamp.endswith("Z"):
dt = datetime.fromisoformat(timestamp.replace("Z", "+00:00"))
else:
dt = datetime.fromisoformat(timestamp)
local_dt = dt.astimezone()
date_str = local_dt.strftime("%Y-%m-%d %H:%M")
sort_key = local_dt
else:
date_str = datetime.fromtimestamp(stat.st_mtime).strftime("%Y-%m-%d %H:%M")
sort_key = datetime.fromtimestamp(stat.st_mtime)
except:
date_str = datetime.fromtimestamp(stat.st_mtime).strftime("%Y-%m-%d %H:%M")
sort_key = datetime.fromtimestamp(stat.st_mtime)
# Apply date filters
if from_dt and sort_key.replace(tzinfo=None) < from_dt:
continue
if to_dt and sort_key.replace(tzinfo=None) >= to_dt:
continue
size_mb = stat.st_size / (1024 * 1024)
sessions.append({
"file": jsonl_file.stem, # UUID without .jsonl
"date": date_str,
"size": f"{size_mb:.1f}M",
"current": "← CURRENT" if is_current else "",
"sort_key": sort_key
})
# Sort by date according to sort_order
reverse_order = (sort_order == "desc")
sessions.sort(key=lambda s: s["sort_key"], reverse=reverse_order)
# Apply limit
if limit:
sessions = sessions[:limit]
print(f"\n{'Session ID':<40} {'Date':<18} {'Size':<8} {''}")
print(f"{'':40} {'(' + tz_offset + ')':<18}")
print("=" * 80)
for s in sessions:
print(f"{s['file']:<40} {s['date']:<18} {s['size']:<8} {s['current']}")
print()
def has_real_text_content(content) -> bool:
"""Check if content has actual text (not just thinking/tool_use)"""
if isinstance(content, str):
return bool(content.strip())
elif isinstance(content, list):
for item in content:
if isinstance(item, dict):
if item.get("type") == "text":
text = item.get("text", "")
if text.strip():
return True
elif isinstance(item, str) and item.strip():
return True
return False
def extract_text_content(content, include_thinking: bool = False, show_full: bool = True) -> str:
"""Extract readable text from message content"""
if isinstance(content, str):
return content
elif isinstance(content, list):
texts = []
had_thinking_or_tools = False
for item in content:
if isinstance(item, dict):
item_type = item.get("type", "")
if item_type == "text":
text_content = item.get("text", "")
# Add separator before text if we had thinking/tools before
if include_thinking and had_thinking_or_tools and text_content.strip():
texts.append("\n💬 **RESPONSE:**\n")
texts.append(text_content)
elif item_type == "thinking" and include_thinking:
thinking_text = item.get("thinking", "")
texts.append(f"\n💭 **THINKING:**\n\n{thinking_text}\n")
had_thinking_or_tools = True
elif item_type == "tool_use" and include_thinking:
tool_name = item.get("name", "unknown")
tool_input = json.dumps(item.get("input", {}), indent=2, ensure_ascii=False)
# Truncate long tool inputs if not show_full
if not show_full and len(tool_input) > 1000:
tool_input = tool_input[:1000] + "\n... [truncated]"
texts.append(f"\n🔧 **TOOL CALL: {tool_name}**\n```json\n{tool_input}\n```\n")
had_thinking_or_tools = True
elif item_type == "tool_result" and include_thinking:
tool_content = item.get("content", "")
if isinstance(tool_content, list):
tool_content = extract_text_content(tool_content, False, show_full)
# Truncate long tool results only if not show_full
if not show_full and len(str(tool_content)) > 2000:
tool_content = str(tool_content)[:2000] + "\n... [truncated]"
texts.append(f"\n📤 **TOOL RESULT:**\n```\n{tool_content}\n```\n")
had_thinking_or_tools = True
elif isinstance(item, str):
texts.append(item)
return "\n".join(texts) if include_thinking else " ".join(texts)
return str(content)
def format_system_notifications(text: str) -> str:
"""Format System: lines with emoji for better visibility"""
lines = text.split('\n')
formatted_lines = []
for line in lines:
# Match "System: [timestamp] message" pattern
if line.strip().startswith("System:"):
# Add emoji and format
formatted_line = "🛠️ " + line.strip()
formatted_lines.append(formatted_line)
else:
formatted_lines.append(line)
return '\n'.join(formatted_lines)
def load_session_messages(jsonl_file: Path) -> List[Tuple[int, Dict]]:
"""Load all messages from a session file with line numbers"""
messages = []
try:
with open(jsonl_file) as f:
for line_num, line in enumerate(f, 1):
try:
data = json.loads(line)
if data.get("type") == "message":
messages.append((line_num, data))
except json.JSONDecodeError:
continue
except Exception as e:
print(f"Error reading {jsonl_file.name}: {e}", file=sys.stderr)
return messages
def get_conversation_context(jsonl_file: Path, target_line: int, target_role: str,
include_thinking: bool = False) -> Dict:
"""
Get conversation context around a message.
- If target is user message: get the LAST assistant response that follows
- If target is assistant message: get the user message that preceded it
Returns dict with 'before' and 'after' lists of messages
"""
messages = load_session_messages(jsonl_file)
context = {
'before': [], # Messages before target (user question if target is assistant)
'after': [] # Messages after target (assistant responses if target is user)
}
target_idx = None
for idx, (line_num, data) in enumerate(messages):
if line_num == target_line:
target_idx = idx
break
if target_idx is None:
return context
if target_role == 'user':
# Find ALL assistant responses after this user message (until next user message)
# Filter: skip injected messages (system notifications) and empty responses
for idx in range(target_idx + 1, len(messages)):
line_num, data = messages[idx]
message = data.get("message", {})
role = message.get("role", "")
content = message.get("content", [])
stop_reason = message.get("stopReason", "")
if role == "user":
break # Stop at next user message
elif role == "assistant":
# Skip system-injected messages and responses without real text
if stop_reason != "injected" and has_real_text_content(content):
context['after'].append(data)
elif target_role == 'assistant':
# Find the user message that preceded this assistant response
for idx in range(target_idx - 1, -1, -1):
line_num, data = messages[idx]
message = data.get("message", {})
role = message.get("role", "")
if role == "user":
context['before'].append(data)
break
return context
def get_tool_call_for_result(jsonl_file: Path, target_line: int, tool_use_id: str) -> Optional[Dict]:
"""Find the tool_use call that corresponds to a tool_result"""
messages = load_session_messages(jsonl_file)
for line_num, data in messages:
if line_num >= target_line:
break
message = data.get("message", {})
content = message.get("content", [])
if isinstance(content, list):
for item in content:
if isinstance(item, dict) and item.get("type") == "tool_use":
if item.get("id") == tool_use_id:
return {
"name": item.get("name", "unknown"),
"input": item.get("input", {}),
"id": tool_use_id
}
return None
def search_session(jsonl_file: Path, query: str = None, role_filter: Optional[str] = None,
message_id: Optional[str] = None, include_thinking: bool = False) -> List[Dict]:
"""Search for messages in a single session file (exact match, case-insensitive)"""
results = []
query_normalized = query.strip().lower() if query else None
try:
with open(jsonl_file) as f:
for line_num, line in enumerate(f, 1):
try:
data = json.loads(line)
# Only process message types
if data.get("type") != "message":
continue
message = data.get("message", {})
role = message.get("role", "")
msg_id = data.get("id", "")
# Filter by message ID if specified
if message_id:
if message_id not in msg_id:
continue
# Apply role filter (normalize: tool -> toolResult)
effective_filter = "toolResult" if role_filter == "tool" else role_filter
if effective_filter and role != effective_filter:
continue
content = message.get("content", [])
text_content = extract_text_content(content, include_thinking)
# Search in content (if query provided) - EXACT match (case-insensitive)
if query_normalized:
text_normalized = text_content.strip().lower()
if query_normalized not in text_normalized:
continue
timestamp = data.get("timestamp", "")
time_str = format_timestamp(timestamp)
results.append({
"session": jsonl_file.stem,
"session_file": jsonl_file,
"line": line_num,
"timestamp": time_str,
"timestamp_raw": timestamp,
"role": role,
"message_id": msg_id,
"content": text_content,
"raw": data
})
except json.JSONDecodeError:
continue
except Exception as e:
print(f"Error reading {jsonl_file.name}: {e}", file=sys.stderr)
return results
def search_all_sessions(query: str = None, session_filter: Optional[str] = None,
role_filter: Optional[str] = None, limit: Optional[int] = None,
message_id: Optional[str] = None, include_thinking: bool = False,
sort_order: str = "desc", date_from: Optional[str] = None,
date_to: Optional[str] = None, agent: str = "main") -> List[Dict]:
"""Search across all or specific session(s)"""
all_results = []
sessions_dir = get_sessions_dir(agent)
# Parse date filters
from_dt = parse_date(date_from) if date_from else None
to_dt = parse_date(date_to) if date_to else None
# Make to_dt inclusive: add 1 second so we can use < comparison
# If only date specified, set to start of next day
if to_dt:
if to_dt.hour == 0 and to_dt.minute == 0 and to_dt.second == 0:
to_dt = to_dt + timedelta(days=1)
else:
to_dt = to_dt + timedelta(seconds=1)
if session_filter == "current":
current = get_current_session(agent)
if not current:
print("No current session found!", file=sys.stderr)
return []
session_files = [current]
elif session_filter and session_filter != "all":
# Treat as UUID
session_file = find_session_by_id(session_filter, agent)
if not session_file:
print(f"Session not found: {session_filter}", file=sys.stderr)
return []
session_files = [session_file]
else:
# Sort sessions by date according to sort_order (desc = newest first, asc = oldest first)
reverse_session_order = (sort_order == "desc")
session_files = sorted(sessions_dir.glob("*.jsonl"),
key=lambda p: p.stat().st_mtime, reverse=reverse_session_order)
for jsonl_file in session_files:
results = search_session(jsonl_file, query, role_filter, message_id, include_thinking)
# Apply date filters
if from_dt or to_dt:
filtered_results = []
for result in results:
msg_dt = parse_timestamp(result.get("timestamp_raw", ""))
if msg_dt:
if from_dt and msg_dt.replace(tzinfo=None) < from_dt:
continue
if to_dt and msg_dt.replace(tzinfo=None) >= to_dt:
continue
filtered_results.append(result)
results = filtered_results
all_results.extend(results)
if limit and len(all_results) >= limit:
break
# Sort by timestamp
reverse_order = (sort_order == "desc")
all_results.sort(key=lambda x: x.get("timestamp_raw", ""), reverse=reverse_order)
return all_results[:limit] if limit else all_results
def format_context_message(data: Dict, include_thinking: bool, show_full: bool,
is_preview: bool = False) -> List[str]:
"""Format a context message for display"""
lines = []
message = data.get("message", {})
role = message.get("role", "")
content = message.get("content", [])
text = extract_text_content(content, include_thinking)
msg_id = data.get("id", "")
timestamp = data.get("timestamp", "")
time_str = format_timestamp(timestamp)
role_icon = "👤" if role == "user" else "🤖"
lines.append(f"\n{'-'*40}")
lines.append(f"[{role_icon} {role.upper()}] {time_str}")
lines.append(f"Message ID: {msg_id}")
lines.append(f"{'-'*40}")
if show_full or not is_preview:
lines.append(text)
else:
max_len = 300 if role == "user" else 500
preview = text[:max_len]
if len(text) > max_len:
preview += "... [truncated]"
lines.append(preview)
return lines
def print_results(results: List[Dict], show_full: bool = False, include_thinking: bool = False,
include_context: bool = False):
"""Print search results with bidirectional context"""
if not results:
print("\nNo results found.")
return
print(f"\nFound {len(results)} result(s):\n")
for i, result in enumerate(results, 1):
print(f"{'='*80}")
print(f"Result #{i}")
print(f"Session: {result['session']}")
print(f"Line: {result['line']}")
print(f"Time: {result['timestamp']}")
print(f"Role: {result['role']}")
print(f"Message ID: {result['message_id']}")
# Get conversation context
if include_context:
context = get_conversation_context(
result['session_file'], result['line'], result['role'], include_thinking
)
# Show messages BEFORE (user question if this is assistant response)
for ctx_msg in context['before']:
for line in format_context_message(ctx_msg, include_thinking, show_full, True):
print(line)
# Show the matched message
print(f"\n{'-'*40}")
role_icon = "👤" if result['role'] == "user" else "🤖"
print(f"[{role_icon} {result['role'].upper()} - MATCHED]")
print(f"{'-'*40}")
if show_full:
print(result['content'])
else:
preview = result['content'][:300]
if len(result['content']) > 300:
preview += "... [truncated]"
print(preview)
# Show messages AFTER (assistant response if this is user message)
if include_context:
# Show the LAST assistant response (not the first)
if context['after']:
last_response = context['after'][-1] # Get the LAST one
for line in format_context_message(last_response, include_thinking, show_full, not show_full):
print(line)
print()
def show_history(session_filter: str = "current", role_filter: Optional[str] = None,
limit: Optional[int] = None, include_thinking: bool = False,
date_from: Optional[str] = None, date_to: Optional[str] = None,
sort_order: str = "desc", show_full: bool = False,
message_id: Optional[str] = None, agent: str = "main"):
"""
Show conversation history in chronological order.
Without --thinking: Shows user messages + their final assistant response
With --thinking: Shows all messages including intermediate steps (thinking, tool calls)
sort_order: "asc" (oldest first) or "desc" (newest first, default)
show_full: If True, don't truncate any content
"""
# Resolve session(s)
if session_filter == "current":
session_file = get_current_session(agent)
if not session_file:
print("No current session found!", file=sys.stderr)
return
session_files = [session_file]
elif session_filter == "all":
# Get all sessions sorted by first message timestamp according to sort_order
session_files = get_sorted_sessions(sort_order, agent)
if not session_files:
print("No sessions found!", file=sys.stderr)
return
else:
# Treat as UUID
session_file = find_session_by_id(session_filter, agent)
if not session_file:
print(f"Session not found: {session_filter}", file=sys.stderr)
return
session_files = [session_file]
# Parse date filters
from_dt = parse_date(date_from) if date_from else None
to_dt = parse_date(date_to) if date_to else None
# Make to_dt inclusive: add 1 second so we can use < comparison
# If only date specified, set to start of next day
if to_dt:
if to_dt.hour == 0 and to_dt.minute == 0 and to_dt.second == 0:
to_dt = to_dt + timedelta(days=1)
else:
to_dt = to_dt + timedelta(seconds=1)
tz_offset = get_local_tz_offset()
print(f"\n📜 Conversation History")
if len(session_files) == 1:
print(f"Session: {session_files[0].stem}")
else:
print(f"Sessions: {len(session_files)} sessions")
print(f"Timezone: {tz_offset}")
if from_dt:
print(f"From: {from_dt.strftime('%Y-%m-%d %H:%M')}")
if to_dt:
print(f"To: {to_dt.strftime('%Y-%m-%d %H:%M')}")
print(f"{'='*80}\n")
displayed_count = 0
# Process each session
for session_file in session_files:
if limit and displayed_count >= limit:
break
messages = load_session_messages(session_file)
if not messages:
continue
# Apply sort order
if sort_order == "desc":
messages = list(reversed(messages))
# Lazy session header: only print when we have messages to show
session_header_printed = False
def print_session_header():
nonlocal session_header_printed
if len(session_files) > 1 and not session_header_printed:
print(f"\n{'─'*80}")
print(f"📂 Session: {session_file.stem}")
print(f"{'─'*80}\n")
session_header_printed = True
i = 0
while i < len(messages):
if limit and displayed_count >= limit:
break
line_num, data = messages[i]
message = data.get("message", {})
role = message.get("role", "")
timestamp_raw = data.get("timestamp", "")
msg_id = data.get("id", "")
msg_dt = parse_timestamp(timestamp_raw)
# Apply message_id filter
if message_id and message_id not in msg_id:
i += 1
continue
# Apply date filters
if msg_dt:
if from_dt and msg_dt.replace(tzinfo=None) < from_dt:
i += 1
continue
if to_dt and msg_dt.replace(tzinfo=None) >= to_dt:
i += 1
continue
# Apply role filter
stop_reason = message.get("stopReason", "")
is_system_injected = stop_reason == "injected"
if role_filter:
if role_filter == "system":
# --role system: only show injected messages
if not is_system_injected:
i += 1
continue
elif role_filter == "tool":
if role != "toolResult":
i += 1
continue
else:
# For user/assistant, skip injected messages unless explicitly requested
if role != role_filter:
i += 1
continue
content = message.get("content", [])
msg_id = data.get("id", "")
time_str = format_timestamp(timestamp_raw)
if role == "user":
# User message
print_session_header()
text = extract_text_content(content, False, show_full)
# Format system notifications with emoji
text = format_system_notifications(text)
print(f"👤 [{time_str}]")
print(f" ID: {msg_id[:12]}...")
print(f" {'-'*70}")
# Indent user message
for line in text.split('\n'):
print(f" {line}")
print()
displayed_count += 1
# Find assistant response(s)
if not include_thinking:
# Without --thinking: find the LAST real assistant response before next user
# Skip system-injected messages (model change notifications, etc.)
#
# IMPORTANT: Search direction depends on sort order!
# - asc (oldest first): responses are AFTER user in array (search forward)
# - desc (newest first): array is reversed, responses are BEFORE user (search backward)
last_assistant = None
if sort_order == "asc":
# Forward search: responses come after user message
j = i + 1
while j < len(messages):
_, next_data = messages[j]
next_message = next_data.get("message", {})
next_role = next_message.get("role", "")
next_content = next_message.get("content", [])
stop_reason = next_message.get("stopReason", "")
if next_role == "user":
break
elif next_role == "assistant" and stop_reason != "injected":
# Only consider responses with real text content
# (skip intermediate thinking/tool_use only messages)
if has_real_text_content(next_content):
last_assistant = (j, next_data)
j += 1
else:
# Backward search (desc): in reversed list, responses are before user
# We want the LAST assistant response chronologically, which is the
# FIRST one in the reversed array (lowest index when going backward)
j = i - 1
while j >= 0:
_, next_data = messages[j]
next_message = next_data.get("message", {})
next_role = next_message.get("role", "")
next_content = next_message.get("content", [])
stop_reason = next_message.get("stopReason", "")
if next_role == "user":
break
elif next_role == "assistant" and stop_reason != "injected":
# Only consider responses with real text content
# (skip intermediate thinking/tool_use only messages)
if has_real_text_content(next_content):
# Always update - last one we find (lowest index) is the
# most recent response chronologically
last_assistant = (j, next_data)
j -= 1
if last_assistant and (not role_filter or role_filter == "assistant"):
_, asst_data = last_assistant
asst_message = asst_data.get("message", {})
asst_content = asst_message.get("content", [])
asst_text = extract_text_content(asst_content, False, show_full)
asst_id = asst_data.get("id", "")
asst_time = format_timestamp(asst_data.get("timestamp", ""))
print(f"🤖 [{asst_time}]")
print(f" ID: {asst_id[:12]}...")
print(f" {'-'*70}")
for line in asst_text.split('\n'):
print(f" {line}")
print()
# Skip to next user message (forward in current array direction)
i += 1
else:
# With --thinking: show all intermediate messages
i += 1
elif role == "assistant":
# Show assistant messages if:
# 1. --thinking is active (show all including system injections)
# 2. --role system is active (only system injections, already filtered above)
# 3. --role assistant is active (explicitly filtering for assistant)
# 4. --message filter is active (show specific message regardless of type)
should_show = include_thinking or role_filter == "system" or role_filter == "assistant" or message_id
if should_show:
print_session_header()
text = extract_text_content(content, include_thinking and not is_system_injected, show_full)
# Use different emoji for system-injected messages
if is_system_injected:
print(f"🛠️ [{time_str}] SYSTEM")
else:
print(f"🤖 [{time_str}]")
print(f" ID: {msg_id[:12]}...")
print(f" {'-'*70}")
for line in text.split('\n'):
print(f" {line}")
print()
displayed_count += 1
i += 1
elif role == "toolResult":
# Show tool results if:
# 1. --thinking is active (show all intermediate steps)
# 2. --role tool is specified (explicitly filtering for tool results)
# 3. --message filter is active (show specific message regardless of type)
if include_thinking or role_filter == "tool" or message_id:
# Find the tool_use that originated this result
# (if --thinking OR --role tool is specified)
tool_use_id = None
tool_use_data = None
if include_thinking or role_filter == "tool":
tool_use_id = message.get("toolCallId") # Get from message, not content
# Search for the corresponding tool_use
# If sort is desc, tool calls are AFTER results (search forward)
# If sort is asc, tool calls are BEFORE results (search backward)
if tool_use_id:
if sort_order == "desc":
# Search forward when list is reversed
search_range = range(i + 1, len(messages))
else:
# Search backward in normal order
search_range = range(i - 1, -1, -1)
for j in search_range:
_, prev_data = messages[j]
prev_message = prev_data.get("message", {})
prev_content = prev_message.get("content", [])
if isinstance(prev_content, list):
for item in prev_content:
if isinstance(item, dict) and item.get("type") == "toolCall":
if item.get("id") == tool_use_id:
tool_use_data = {
"name": item.get("name", "unknown"),
"arguments": item.get("arguments", {}),
"timestamp": prev_data.get("timestamp", "")
}
break
if tool_use_data:
break
# Show the tool_use first (if found and --thinking OR --role tool)
if tool_use_data and (include_thinking or role_filter == "tool"):
print_session_header()
tool_time = format_timestamp(tool_use_data["timestamp"])
tool_name = tool_use_data["name"]
tool_input = json.dumps(tool_use_data["arguments"], indent=2, ensure_ascii=False)
print(f"🔧 [{tool_time}] Tool Call: {tool_name}")
print(f" {'-'*70}")
# Show input parameters (truncate if too long, unless show_full)
if not show_full and len(tool_input) > 800:
tool_input = tool_input[:800] + "\n ... [truncated]"
for line in tool_input.split('\n'):
print(f" {line}")
print()
# Show tool result
print_session_header()
text = extract_text_content(content, True, show_full)
tool_name = message.get("toolName", "unknown")
print(f"📤 [{time_str}] Tool Result: {tool_name}")
print(f" ID: {msg_id[:12]}...")
print(f" {'-'*70}")
# Truncate long tool results (unless show_full)
if not show_full and len(text) > 1500:
text = text[:1500] + "\n ... [truncated]"
for line in text.split('\n'):
print(f" {line}")
print()
displayed_count += 1
i += 1
else:
i += 1
print(f"{'='*80}")
print(f"Displayed {displayed_count} message(s)")
def ensure_md_extension(filename: str) -> str:
"""Ensure filename has .md extension and sanitize for filesystem compatibility"""
if not filename:
return filename
# First, sanitize the entire filename to replace problematic characters
# Replace / and \ to avoid path interpretation issues
# Allow: letters, numbers, spaces, hyphens, underscores, parentheses
filename_sanitized = re.sub(r'[/\\]', '-', filename) # Replace slashes with dash
filename_sanitized = re.sub(r'[^\w\s\-()áéíóúñÁÉÍÓÚÑ.]', '_', filename_sanitized) # Replace other special chars
# Now get the base filename without extension
path = Path(filename_sanitized)
base_name = path.stem # Filename without extension
# Always return with .md extension
base_name_md = f"{base_name}.md"
# If path has a parent directory specified, preserve it
if path.parent.name:
return str(path.parent / base_name_md)
else:
# If no directory specified, use current working directory
return base_name_md
def export_history_to_markdown(session_filter: str = "current", role_filter: Optional[str] = None,
limit: Optional[int] = None, include_thinking: bool = False,
date_from: Optional[str] = None, date_to: Optional[str] = None,
sort_order: str = "desc", show_full: bool = False,
export_path: Optional[str] = None, message_id: Optional[str] = None,
agent: str = "main"):
"""
Export conversation history to a properly formatted markdown file.
Uses the same format as export_to_markdown (with ### headers for user/assistant).
"""
# Resolve session(s)
if session_filter == "current":
session_file = get_current_session(agent)
if not session_file:
print("No current session found!", file=sys.stderr)
return
session_files = [session_file]
elif session_filter == "all":
session_files = get_sorted_sessions(sort_order, agent)
if not session_files:
print("No sessions found!", file=sys.stderr)
return
else:
session_file = find_session_by_id(session_filter, agent)
if not session_file:
print(f"Session not found: {session_filter}", file=sys.stderr)
return
session_files = [session_file]
# Parse date filters
from_dt = parse_date(date_from) if date_from else None
to_dt = parse_date(date_to) if date_to else None
# Make to_dt inclusive: add 1 second so we can use < comparison
# If only date specified, set to start of next day
if to_dt:
if to_dt.hour == 0 and to_dt.minute == 0 and to_dt.second == 0:
to_dt = to_dt + timedelta(days=1)
else:
to_dt = to_dt + timedelta(seconds=1)
# Generate filename (always ensure .md extension)
if export_path is True or export_path == "" or export_path is None:
timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
filename = f"history_export_{timestamp}.md"
else:
filename = ensure_md_extension(export_path)
tz_offset = get_local_tz_offset()
now_str = f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} {tz_offset}"
lines = []
lines.append("# Conversation History Export")
lines.append("")
if len(session_files) == 1:
lines.append(f"**Session:** `{session_files[0].stem}`")
else:
lines.append(f"**Sessions:** {len(session_files)} sessions")
lines.append(f"**Exported:** {now_str}")
lines.append(f"**Timezone:** {tz_offset}")
if from_dt:
lines.append(f"**From:** {from_dt.strftime('%Y-%m-%d %H:%M')}")
if to_dt:
lines.append(f"**To:** {to_dt.strftime('%Y-%m-%d %H:%M')}")
lines.append("")
lines.append("---")
lines.append("")
displayed_count = 0
conversation_num = 0
session_lines = {} # Track lines per session for lazy headers
for session_file in session_files:
if limit and displayed_count >= limit:
break
messages = load_session_messages(session_file)
if not messages:
continue
if sort_order == "desc":
messages = list(reversed(messages))
# Lazy session header: collect lines, only add header if we have content
current_session_lines = []
session_has_content = False
i = 0
while i < len(messages):
if limit and displayed_count >= limit:
break
line_num, data = messages[i]
message = data.get("message", {})
role = message.get("role", "")
timestamp_raw = data.get("timestamp", "")
msg_id = data.get("id", "")
msg_dt = parse_timestamp(timestamp_raw)
# Apply message_id filter
if message_id and message_id not in msg_id:
i += 1
continue
# Apply date filters (use >= for to_dt since we added 1 second)
if msg_dt:
if from_dt and msg_dt.replace(tzinfo=None) < from_dt:
i += 1
continue
if to_dt and msg_dt.replace(tzinfo=None) >= to_dt:
i += 1
continue
# Apply role filter
stop_reason = message.get("stopReason", "")
is_system_injected = stop_reason == "injected"
if role_filter:
if role_filter == "system":
if not is_system_injected:
i += 1
continue
elif role_filter == "tool":
if role != "toolResult":
i += 1
continue
else:
if role != role_filter:
i += 1
continue
content = message.get("content", [])
time_str = format_timestamp(timestamp_raw)
if role == "user":
conversation_num += 1
session_has_content = True
text = extract_text_content(content, False, show_full)
text = format_system_notifications(text)
current_session_lines.append(f"### 👤 User")
current_session_lines.append("")
current_session_lines.append(f"- **Time:** {time_str}")
current_session_lines.append(f"- **Message ID:** `{msg_id}`")
current_session_lines.append("")
current_session_lines.append(text)
current_session_lines.append("")
displayed_count += 1
# Find assistant response(s)
if not include_thinking:
last_assistant = None
if sort_order == "asc":
j = i + 1
while j < len(messages):
_, next_data = messages[j]
next_message = next_data.get("message", {})
next_role = next_message.get("role", "")
next_content = next_message.get("content", [])
stop_reason = next_message.get("stopReason", "")
if next_role == "user":
break
elif next_role == "assistant" and stop_reason != "injected":
if has_real_text_content(next_content):
last_assistant = (j, next_data)
j += 1
else:
j = i - 1
while j >= 0:
_, next_data = messages[j]
next_message = next_data.get("message", {})
next_role = next_message.get("role", "")
next_content = next_message.get("content", [])
stop_reason = next_message.get("stopReason", "")
if next_role == "user":
break
elif next_role == "assistant" and stop_reason != "injected":
if has_real_text_content(next_content):
last_assistant = (j, next_data)
j -= 1
if last_assistant and (not role_filter or role_filter == "assistant"):
_, asst_data = last_assistant
asst_message = asst_data.get("message", {})
asst_content = asst_message.get("content", [])
asst_text = extract_text_content(asst_content, False, show_full)
asst_id = asst_data.get("id", "")
asst_time = format_timestamp(asst_data.get("timestamp", ""))
current_session_lines.append(f"### 🤖 Assistant")
current_session_lines.append("")
current_session_lines.append(f"- **Time:** {asst_time}")
current_session_lines.append(f"- **Message ID:** `{asst_id}`")
current_session_lines.append("")
current_session_lines.append(asst_text)
current_session_lines.append("")
i += 1
else:
i += 1
current_session_lines.append("---")
current_session_lines.append("")
elif role == "assistant":
should_show = include_thinking or role_filter == "system" or role_filter == "assistant" or message_id
if should_show:
session_has_content = True
text = extract_text_content(content, include_thinking and not is_system_injected, show_full)
if is_system_injected:
current_session_lines.append(f"### 🛠️ System")
else:
current_session_lines.append(f"### 🤖 Assistant")
current_session_lines.append("")
current_session_lines.append(f"- **Time:** {time_str}")
current_session_lines.append(f"- **Message ID:** `{msg_id}`")
current_session_lines.append("")
current_session_lines.append(text)
current_session_lines.append("")
current_session_lines.append("---")
current_session_lines.append("")
displayed_count += 1
i += 1
elif role == "toolResult":
if include_thinking or role_filter == "tool" or message_id:
tool_use_id = message.get("toolCallId")
tool_use_data = None
if tool_use_id:
if sort_order == "desc":
search_range = range(i + 1, len(messages))
else:
search_range = range(i - 1, -1, -1)
for j in search_range:
_, prev_data = messages[j]
prev_message = prev_data.get("message", {})
prev_content = prev_message.get("content", [])
if isinstance(prev_content, list):
for item in prev_content:
if isinstance(item, dict) and item.get("type") == "toolCall":
if item.get("id") == tool_use_id:
tool_use_data = {
"name": item.get("name", "unknown"),
"arguments": item.get("arguments", {}),
"timestamp": prev_data.get("timestamp", "")
}
break
if tool_use_data:
break
if tool_use_data:
tool_time = format_timestamp(tool_use_data["timestamp"])
tool_name = tool_use_data["name"]
tool_input = json.dumps(tool_use_data["arguments"], indent=2, ensure_ascii=False)
current_session_lines.append(f"### 🔧 Tool Call: {tool_name}")
current_session_lines.append("")
current_session_lines.append(f"- **Time:** {tool_time}")
current_session_lines.append("")
if not show_full and len(tool_input) > 800:
tool_input = tool_input[:800] + "\n... [truncated]"
current_session_lines.append("```json")
current_session_lines.append(tool_input)
current_session_lines.append("```")
current_session_lines.append("")
session_has_content = True
text = extract_text_content(content, True, show_full)
tool_name = message.get("toolName", "unknown")
current_session_lines.append(f"### 📤 Tool Result: {tool_name}")
current_session_lines.append("")
current_session_lines.append(f"- **Time:** {time_str}")
current_session_lines.append(f"- **Message ID:** `{msg_id}`")
current_session_lines.append("")
if not show_full and len(text) > 1500:
text = text[:1500] + "\n... [truncated]"
current_session_lines.append("```")
current_session_lines.append(text)
current_session_lines.append("```")
current_session_lines.append("")
current_session_lines.append("---")
current_session_lines.append("")
displayed_count += 1
i += 1
else:
i += 1
# Add session header + content only if we have something to show (lazy header)
if session_has_content:
if len(session_files) > 1:
lines.append(f"## Session: `{session_file.stem}`")
lines.append("")
lines.extend(current_session_lines)
lines.append(f"**Total messages:** {displayed_count}")
md_content = "\n".join(lines)
with open(filename, "w", encoding="utf-8") as f:
f.write(md_content)
print(f"✅ Exported to: {filename}")
return filename
def export_to_markdown(results: List[Dict], include_thinking: bool = False,
include_context: bool = True, output_path: Optional[str] = None,
show_full: bool = False) -> str:
"""Export results to a markdown file
By default (without --full): user/assistant text is complete, tool results are truncated.
With --full: everything is complete including tool results.
"""
if not results:
print("No results to export.")
return None
# Generate filename based on first result (always ensure .md extension)
first = results[0]
# Use raw timestamp for filename (without timezone string)
timestamp_for_file = format_timestamp(first.get('timestamp_raw', ''), include_tz=False)
timestamp_for_file = timestamp_for_file.replace(":", "-").replace(" ", "_")
msg_id_short = first['message_id'][:8] if first['message_id'] else "unknown"
if output_path:
filename = ensure_md_extension(output_path)
else:
filename = f"export_{timestamp_for_file}_{msg_id_short}.md"
tz_offset = get_local_tz_offset()
now_str = f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} {tz_offset}"
lines = []
lines.append(f"# Session Export")
lines.append(f"")
lines.append(f"**Session:** `{first['session']}`")
lines.append(f"**Exported:** {now_str}")
lines.append(f"")
lines.append("---")
lines.append("")
for i, result in enumerate(results, 1):
# Get conversation context
context = None
if include_context:
context = get_conversation_context(
result['session_file'], result['line'], result['role'], include_thinking
)
lines.append(f"## Conversation {i}")
lines.append(f"")
# Show messages BEFORE (user question if this is assistant response)
if context and context['before']:
for ctx_msg in context['before']:
message = ctx_msg.get("message", {})
role = message.get("role", "")
content = message.get("content", [])
text = extract_text_content(content, include_thinking, show_full)
msg_id = ctx_msg.get("id", "")
timestamp = ctx_msg.get("timestamp", "")
time_str = format_timestamp(timestamp)
lines.append(f"### 👤 User Question")
lines.append(f"")
lines.append(f"- **Time:** {time_str}")
lines.append(f"- **Message ID:** `{msg_id}`")
lines.append(f"")
lines.append(text)
lines.append(f"")
# Show the matched message
role_icon = "👤" if result['role'] == "user" else "🤖"
role_label = "User" if result['role'] == "user" else "Assistant"
matched_label = " (Matched)" if not context or (not context['before'] and not context['after']) else ""
lines.append(f"### {role_icon} {role_label}{matched_label}")
lines.append(f"")
lines.append(f"- **Time:** {result['timestamp']}")
lines.append(f"- **Message ID:** `{result['message_id']}`")
lines.append(f"- **Line:** {result['line']}")
lines.append(f"")
lines.append(result['content'])
lines.append(f"")
# Show messages AFTER (LAST assistant response if this is user message)
if context and context['after']:
# Only show the LAST response
last_response = context['after'][-1]
message = last_response.get("message", {})
role = message.get("role", "")
content = message.get("content", [])
text = extract_text_content(content, include_thinking, show_full)
msg_id = last_response.get("id", "")
timestamp = last_response.get("timestamp", "")
time_str = format_timestamp(timestamp)
lines.append(f"### 🤖 Assistant Response")
lines.append(f"")
lines.append(f"- **Time:** {time_str}")
lines.append(f"- **Message ID:** `{msg_id}`")
lines.append(f"")
lines.append(text)
lines.append(f"")
lines.append("---")
lines.append("")
content = "\n".join(lines)
with open(filename, "w", encoding="utf-8") as f:
f.write(content)
print(f"✅ Exported to: {filename}")
return filename
def main():
parser = argparse.ArgumentParser(
description="Search OpenClaw session messages",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Examples:
# List sessions
%(prog)s --list # List all sessions (newest first, main agent)
%(prog)s --list --limit 10 # List last 10 sessions
%(prog)s --list --sort asc --limit 1 # Show first session created (oldest first)
%(prog)s --list --from 2026-01-01 # List sessions from specific date onwards
%(prog)s --list --from 2026-02-01 --to 2026-02-31 # List sessions in date range
# Search messages
%(prog)s "error message" # Search all sessions (exact, case-insensitive)
%(prog)s "config" --session current # Search current session only
%(prog)s "debug" --session a1b2c3d4 # Search specific session (partial UUID)
%(prog)s "completed" --role assistant --limit 5 # Last 5 assistant messages with "completed"
%(prog)s "system prompt" --full # Show full content with context
%(prog)s "fix bug" --sort asc # Sort search results oldest to newest
# Find and export specific messages
%(prog)s --message a1b2c3d4 # Find message by ID
%(prog)s --message a1b2c3d4 --thinking # Include thinking/tool calls
%(prog)s --message a1b2c3d4 --export # Export to markdown file
%(prog)s --session a1b2c3d4 --message e5f6g7h8 --export # Export specific message
# View conversation history
%(prog)s --history # Show last 100 messages (newest first)
%(prog)s --history --sort asc # Show last 100 messages (oldest first)
%(prog)s --history --thinking # Include thinking blocks + tool calls
%(prog)s --history --role user --limit 20 # Last 20 user messages
%(prog)s --history --role assistant --limit 50 # Last 50 assistant responses
%(prog)s --history --role tool --limit 30 # Last 30 tool calls with results
%(prog)s --history --role system --limit 10 # Last 10 system notifications (e.g. model changes)
%(prog)s --history --from 2026-01-01 # Messages from specific date onwards
%(prog)s --history --from 2026-01-01 --to 2026-12-31 # Messages in date range
%(prog)s --history --thinking --limit 30 --sort asc # Detailed history (oldest first)
%(prog)s --history --session current --limit 5 --sort asc --export # Export last 5 messages from current session to markdown file (oldest first)
%(prog)s -H -s a1b2c3d4 --from 2026-02-15 10:30 --to 2026-02-15 12:30 --sort asc --export "analysis" # Export messages in datetime range from session to "analysis.md" (oldest first)
"""
)
parser.add_argument("query", nargs="?", help="Search query text (exact match, case-insensitive)")
parser.add_argument("--session", "-s", default="all",
help="Session filter: 'current', 'all', or UUID (partial match)")
parser.add_argument("--role", "-r", choices=["user", "assistant", "tool", "system"],
help="Filter by message role (tool = tool results, system = injected notifications)")
parser.add_argument("--limit", "-l", type=int, default=100,
help="Limit number of results (default: 100)")
parser.add_argument("--full", "-f", action="store_true",
help="Show full message content (do not truncate)")
parser.add_argument("--list", action="store_true",
help="List all sessions with basic info")
parser.add_argument("--message", "-m",
help="Filter by message ID (partial match)")
parser.add_argument("--thinking", "-t", action="store_true",
help="Include thinking blocks and tool calls")
parser.add_argument("--export", "-e", nargs="?", const=True,
help="Export results to markdown file (optional: specify filename)")
parser.add_argument("--history", "-H", action="store_true",
help="Show conversation history (chronological order, newest first by default)")
parser.add_argument("--from", dest="date_from", metavar="DATE", nargs='+',
help="Filter messages from date (YYYY-MM-DD or YYYY-MM-DD HH:MM:SS)")
parser.add_argument("--to", dest="date_to", metavar="DATE", nargs='+',
help="Filter messages until date (YYYY-MM-DD or YYYY-MM-DD HH:MM:SS)")
parser.add_argument("--sort", choices=["asc", "desc"], default="desc",
help="Sort order: 'asc' (oldest first) or 'desc' (newest first, default)")
parser.add_argument("--agent", "-a", default="main",
help="Agent to search in (default: main)")
args = parser.parse_args()
# Join date parts if provided without quotes (e.g., --from 2026-02-03 19:57:13)
date_from = ' '.join(args.date_from) if args.date_from else None
date_to = ' '.join(args.date_to) if args.date_to else None
if args.list:
list_sessions(limit=args.limit, date_from=date_from, date_to=date_to,
agent=args.agent, sort_order=args.sort)
return
if args.history:
if args.export:
# Export to markdown file with proper formatting
export_path = args.export if isinstance(args.export, str) else None
export_history_to_markdown(
session_filter=args.session,
role_filter=args.role,
limit=args.limit,
include_thinking=args.thinking,
date_from=date_from,
date_to=date_to,
sort_order=args.sort,
show_full=args.full,
export_path=export_path,
message_id=args.message,
agent=args.agent
)
else:
# Normal console display
show_history(
session_filter=args.session,
role_filter=args.role,
limit=args.limit,
include_thinking=args.thinking,
date_from=date_from,
date_to=date_to,
sort_order=args.sort,
show_full=args.full,
message_id=args.message,
agent=args.agent
)
return
# Allow search by date range even without query or message
has_date_filter = date_from or date_to
if not args.query and not args.message and not has_date_filter:
parser.print_help()
return
# Determine if we should include context (when no role filter specified)
include_context = args.role is None
results = search_all_sessions(
args.query,
session_filter=args.session,
role_filter=args.role,
limit=args.limit,
message_id=args.message,
include_thinking=args.thinking,
sort_order=args.sort,
date_from=date_from,
date_to=date_to,
agent=args.agent
)
if args.export:
output_path = args.export if isinstance(args.export, str) else None
# Always include context in exports (show question + answer pairs)
export_to_markdown(results, include_thinking=args.thinking,
include_context=True, output_path=output_path,
show_full=args.full)
else:
print_results(results, show_full=args.full, include_thinking=args.thinking,
include_context=include_context)
if __name__ == "__main__":
main()
@Carleslc

Carleslc commented Mar 2, 2026

Copy link
Copy Markdown
Author

Description + Quick Install:

https://www.reddit.com/r/openclaw/comments/1r4inck/search_messages_openclaw_script/

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment