systemprompt_cli/commands/infrastructure/logs/
search.rs

1//! `infra logs search`: pattern-match log messages and (optionally) MCP tool
2//! executions, returning a combined result set filtered by level, module, and
3//! time.
4//!
5//! Copyright (c) systemprompt.io — Business Source License 1.1.
6//! See <https://systemprompt.io> for licensing details.
7
8use anyhow::Result;
9use clap::Args;
10use schemars::JsonSchema;
11use serde::{Deserialize, Serialize};
12use std::sync::Arc;
13use systemprompt_identifiers::TraceId;
14use systemprompt_logging::{CliService, LogSearchItem, ToolExecutionItem, TraceQueryService};
15
16use super::duration::parse_since;
17use super::shared::display_log_row;
18use super::{LogEntryRow, LogFilters};
19use crate::CliConfig;
20use crate::shared::CommandOutput;
21
22#[derive(Debug, Args)]
23pub struct SearchArgs {
24    #[arg(help = "Search pattern (matches message content and tool names)")]
25    pub pattern: String,
26
27    #[arg(long, help = "Filter by log level (error, warn, info, debug, trace)")]
28    pub level: Option<String>,
29
30    #[arg(long, help = "Filter by module name (partial match)")]
31    pub module: Option<String>,
32
33    #[arg(
34        long,
35        help = "Only search logs since this duration (e.g., '1h', '24h', '7d') or datetime"
36    )]
37    pub since: Option<String>,
38
39    #[arg(long, short = 'n', default_value = "50", help = "Maximum results")]
40    pub limit: i64,
41
42    #[arg(long, default_value = "true", help = "Include MCP tool executions")]
43    pub include_tools: bool,
44}
45
46#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
47pub struct ToolSearchResult {
48    pub timestamp: String,
49    pub trace_id: TraceId,
50    pub tool_name: String,
51    pub server: String,
52    pub status: String,
53    #[serde(skip_serializing_if = "Option::is_none")]
54    pub duration_ms: Option<i64>,
55}
56
57#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
58pub struct CombinedSearchOutput {
59    pub logs: Vec<LogEntryRow>,
60    pub log_count: u64,
61    pub tools: Vec<ToolSearchResult>,
62    pub tool_count: u64,
63    pub filters: LogFilters,
64}
65
66crate::define_pool_command!(SearchArgs => CommandOutput, with_config);
67
68async fn execute_with_pool_inner(
69    args: SearchArgs,
70    pool: &Arc<sqlx::PgPool>,
71    config: &CliConfig,
72) -> Result<CommandOutput> {
73    let since_timestamp = parse_since(args.since.as_ref())?;
74    let level_filter = args.level.as_deref().map(str::to_uppercase);
75    let pattern = format!("%{}%", args.pattern);
76
77    let service = TraceQueryService::new(Arc::clone(pool));
78
79    let rows = service
80        .search_logs(
81            &pattern,
82            since_timestamp,
83            level_filter.as_deref(),
84            args.limit,
85        )
86        .await?;
87
88    let tool_rows = if args.include_tools {
89        service
90            .search_tool_executions(&pattern, since_timestamp, args.limit)
91            .await?
92    } else {
93        vec![]
94    };
95
96    let logs = map_log_rows(rows, args.module.as_deref());
97    let tools = map_tool_rows(tool_rows);
98
99    let filters = LogFilters {
100        level: args.level.clone(),
101        module: args.module.clone(),
102        since: args.since.clone(),
103        pattern: Some(args.pattern.clone()),
104        tail: args.limit,
105    };
106
107    let output = CombinedSearchOutput {
108        log_count: logs.len() as u64,
109        logs,
110        tool_count: tools.len() as u64,
111        tools,
112        filters,
113    };
114
115    let result = CommandOutput::table_of(
116        vec!["id", "trace_id", "timestamp", "level", "module", "message"],
117        &output.logs,
118    )
119    .with_title("Search Results");
120
121    if config.is_json_output() {
122        return Ok(result);
123    }
124
125    render_combined_results(&output.logs, &output.tools, &args.pattern, &output.filters);
126    Ok(result.with_skip_render())
127}
128
129pub fn map_log_rows(rows: Vec<LogSearchItem>, module_filter: Option<&str>) -> Vec<LogEntryRow> {
130    rows.into_iter()
131        .filter(|r| module_filter.is_none_or(|module| r.module.contains(module)))
132        .map(|r| LogEntryRow {
133            id: r.id,
134            trace_id: r.trace_id,
135            timestamp: r.timestamp.format("%Y-%m-%d %H:%M:%S%.3f").to_string(),
136            level: r.level.to_uppercase(),
137            module: r.module,
138            message: r.message,
139            metadata: r.metadata.as_ref().and_then(|m| {
140                serde_json::from_str(m)
141                    .map_err(|e| {
142                        tracing::warn!(error = %e, "Failed to parse log metadata");
143                        e
144                    })
145                    .ok()
146            }),
147        })
148        .collect()
149}
150
151pub fn map_tool_rows(rows: Vec<ToolExecutionItem>) -> Vec<ToolSearchResult> {
152    rows.into_iter()
153        .map(|r| ToolSearchResult {
154            timestamp: r.timestamp.format("%Y-%m-%d %H:%M:%S").to_string(),
155            trace_id: r.trace_id,
156            tool_name: r.tool_name,
157            server: r.server_name.unwrap_or_else(|| "unknown".to_owned()),
158            status: r.status,
159            duration_ms: r.execution_time_ms.map(i64::from),
160        })
161        .collect()
162}
163
164fn render_combined_results(
165    logs: &[LogEntryRow],
166    tools: &[ToolSearchResult],
167    pattern: &str,
168    filters: &LogFilters,
169) {
170    CliService::section(&format!("Search Results: \"{}\"", pattern));
171
172    if filters.level.is_some() || filters.module.is_some() || filters.since.is_some() {
173        if let Some(ref level) = filters.level {
174            CliService::key_value("Level", level);
175        }
176        if let Some(ref module) = filters.module {
177            CliService::key_value("Module", module);
178        }
179        if let Some(ref since) = filters.since {
180            CliService::key_value("Since", since);
181        }
182    }
183
184    if !tools.is_empty() {
185        CliService::subsection(&format!("MCP Tool Executions ({})", tools.len()));
186        for tool in tools {
187            let duration = tool.duration_ms.map(|d| format!(" ({}ms)", d));
188            let line = format!(
189                "{} {}/{} [{}]{}  trace:{}",
190                tool.timestamp,
191                tool.server,
192                tool.tool_name,
193                tool.status,
194                duration.as_deref().unwrap_or(""),
195                tool.trace_id
196            );
197            match tool.status.as_str() {
198                "error" | "failed" => CliService::error(&line),
199                _ => CliService::info(&line),
200            }
201        }
202    }
203
204    if !logs.is_empty() {
205        CliService::subsection(&format!("Log Entries ({})", logs.len()));
206        for log in logs {
207            display_log_row(log);
208        }
209    }
210
211    if logs.is_empty() && tools.is_empty() {
212        CliService::warning("No matching results found");
213        return;
214    }
215
216    CliService::info(&format!(
217        "Found {} log entries and {} tool executions",
218        logs.len(),
219        tools.len()
220    ));
221}
systemprompt_cli/commands/infrastructure/logs/search.rs

systemprompt_cli/commands/infrastructure/logs/
search.rs
