-
Notifications
You must be signed in to change notification settings - Fork 3
Expand file tree
/
Copy pathtrace_viewer.py
More file actions
391 lines (297 loc) Β· 11.6 KB
/
trace_viewer.py
File metadata and controls
391 lines (297 loc) Β· 11.6 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
#!/usr/bin/env python3
"""
PatchPro Trace Viewer
Streamlit app for viewing and analyzing patch generation traces.
Visualizes LLM interactions, failures, and performance metrics.
Usage:
streamlit run trace_viewer.py
Or with custom trace directory:
streamlit run trace_viewer.py -- --trace-dir /path/to/traces
"""
import argparse
import json
import sqlite3
import sys
from pathlib import Path
from typing import Dict, List, Any, Optional
import streamlit as st
# Constants
DEFAULT_TRACE_DIR = Path.cwd() / ".patchpro" / "traces"
def get_trace_dir() -> Path:
"""Get trace directory from command line or use default."""
# Parse command line args (Streamlit passes them after --)
parser = argparse.ArgumentParser()
parser.add_argument("--trace-dir", type=Path, default=DEFAULT_TRACE_DIR)
try:
args = parser.parse_args()
return args.trace_dir
except SystemExit:
# Fallback if arg parsing fails
return DEFAULT_TRACE_DIR
def load_trace_db(trace_dir: Path) -> Optional[sqlite3.Connection]:
"""Load traces.db from directory."""
db_path = trace_dir / "traces.db"
if not db_path.exists():
return None
conn = sqlite3.connect(str(db_path))
conn.row_factory = sqlite3.Row # Return dict-like rows
return conn
def load_trace_json(trace_dir: Path, trace_id: str) -> Optional[Dict[str, Any]]:
"""Load full trace JSON for a specific trace_id."""
json_file = trace_dir / f"{trace_id}.json"
if not json_file.exists():
return None
with open(json_file) as f:
return json.load(f)
def query_traces(
conn: sqlite3.Connection,
rule_id: Optional[str] = None,
status: Optional[str] = None,
strategy: Optional[str] = None,
search_text: Optional[str] = None,
limit: int = 100
) -> List[Dict[str, Any]]:
"""Query traces from SQLite with filters."""
cursor = conn.cursor()
# Build query
query = "SELECT * FROM traces WHERE 1=1"
params = []
if rule_id and rule_id != "All":
query += " AND rule_id = ?"
params.append(rule_id)
if status and status != "All":
query += " AND final_status = ?"
params.append(status)
if strategy and strategy != "All":
query += " AND strategy = ?"
params.append(strategy)
if search_text:
query += " AND (finding_message LIKE ? OR file_path LIKE ?)"
search_pattern = f"%{search_text}%"
params.append(search_pattern)
params.append(search_pattern)
query += " ORDER BY timestamp DESC LIMIT ?"
params.append(limit)
cursor.execute(query, params)
rows = cursor.fetchall()
return [dict(row) for row in rows]
def get_summary_stats(conn: sqlite3.Connection) -> Dict[str, Any]:
"""Get summary statistics from all traces."""
cursor = conn.cursor()
# Total traces
cursor.execute("SELECT COUNT(*) FROM traces")
total = cursor.fetchone()[0]
if total == 0:
return {
"total_traces": 0,
"successful_traces": 0,
"success_rate": 0,
"avg_cost_usd": 0,
"avg_latency_ms": 0,
}
# Success rate
cursor.execute("SELECT COUNT(*) FROM traces WHERE validation_passed = 1")
successes = cursor.fetchone()[0]
# Average cost
cursor.execute("SELECT AVG(cost_usd) FROM traces")
avg_cost = cursor.fetchone()[0] or 0
# Average latency
cursor.execute("SELECT AVG(latency_ms) FROM traces")
avg_latency = cursor.fetchone()[0] or 0
# Total cost
cursor.execute("SELECT SUM(cost_usd) FROM traces")
total_cost = cursor.fetchone()[0] or 0
# Average retries
cursor.execute("SELECT AVG(retry_attempt) FROM traces")
avg_retries = cursor.fetchone()[0] or 1
return {
"total_traces": total,
"successful_traces": successes,
"failed_traces": total - successes,
"success_rate": successes / total if total > 0 else 0,
"avg_cost_usd": avg_cost,
"total_cost_usd": total_cost,
"avg_latency_ms": avg_latency,
"avg_retry_attempt": avg_retries,
}
def get_filter_options(conn: sqlite3.Connection) -> Dict[str, List[str]]:
"""Get unique values for filter dropdowns."""
cursor = conn.cursor()
# Unique rule IDs
cursor.execute("SELECT DISTINCT rule_id FROM traces ORDER BY rule_id")
rule_ids = ["All"] + [row[0] for row in cursor.fetchall()]
# Unique statuses
cursor.execute("SELECT DISTINCT final_status FROM traces ORDER BY final_status")
statuses = ["All"] + [row[0] for row in cursor.fetchall()]
# Unique strategies
cursor.execute("SELECT DISTINCT strategy FROM traces ORDER BY strategy")
strategies = ["All"] + [row[0] for row in cursor.fetchall()]
return {
"rule_ids": rule_ids,
"statuses": statuses,
"strategies": strategies,
}
def render_summary_metrics(stats: Dict[str, Any]):
"""Render summary metrics at top of page."""
col1, col2, col3, col4 = st.columns(4)
with col1:
st.metric("Total Traces", stats["total_traces"])
st.metric("Avg Retry", f"{stats['avg_retry_attempt']:.1f}")
with col2:
st.metric("Success Rate", f"{stats['success_rate']:.1%}")
st.metric("Successful", stats["successful_traces"])
with col3:
st.metric("Avg Cost", f"${stats['avg_cost_usd']:.4f}")
st.metric("Total Cost", f"${stats['total_cost_usd']:.3f}")
with col4:
st.metric("Avg Latency", f"{stats['avg_latency_ms']:.0f}ms")
st.metric("Failed", stats["failed_traces"])
def render_trace_card(trace: Dict[str, Any], trace_dir: Path):
"""Render a single trace as an expandable card."""
# Status badge
if trace["validation_passed"]:
status_color = "π’"
status_text = "SUCCESS"
else:
status_color = "π΄"
status_text = trace["final_status"].upper()
# Header line
header = f"{status_color} **{trace['rule_id']}** - `{Path(trace['file_path']).name}:{trace['line_number']}` - Attempt {trace['retry_attempt']}"
with st.expander(header):
# Metadata
col1, col2, col3 = st.columns(3)
with col1:
st.write("**Strategy:**", trace["strategy"])
st.write("**Model:**", trace["model"])
st.write("**File Type:**", trace["file_type"])
with col2:
st.write("**Complexity:**", trace["finding_complexity"])
st.write("**Category:**", trace["rule_category"])
st.write("**Tokens:**", trace["tokens_used"])
with col3:
st.write("**Cost:**", f"${trace['cost_usd']:.4f}")
st.write("**Latency:**", f"{trace['latency_ms']}ms")
st.write("**Status:**", status_text)
st.divider()
# Load full trace JSON for detailed info
full_trace = load_trace_json(trace_dir, trace["trace_id"])
if full_trace:
# Finding message
st.write("**Finding:**")
st.info(full_trace["finding_message"])
# Prompt (collapsed by default)
with st.expander("π View Prompt"):
st.text_area(
"System Prompt",
value=full_trace["system_prompt"],
height=150,
key=f"sys_{trace['trace_id']}"
)
st.text_area(
"User Prompt",
value=full_trace["prompt"],
height=300,
key=f"prompt_{trace['trace_id']}"
)
# LLM Response
with st.expander("π€ View LLM Response"):
st.text_area(
"Response",
value=full_trace["llm_response"],
height=300,
key=f"resp_{trace['trace_id']}"
)
# Generated Patch
if full_trace["patch_generated"]:
with st.expander("π§ View Generated Patch"):
st.code(full_trace["patch_generated"], language="diff")
else:
st.warning("β οΈ No patch generated")
# Validation Errors
if full_trace["validation_errors"]:
st.write("**Validation Errors:**")
for error in full_trace["validation_errors"]:
st.error(error)
# Previous Errors (if retry)
if full_trace["previous_errors"]:
with st.expander("π Previous Attempt Errors"):
for i, error in enumerate(full_trace["previous_errors"], 1):
st.write(f"**Error {i}:**")
st.code(error)
# Actions
st.divider()
col1, col2, col3 = st.columns(3)
with col1:
if st.button("π Copy Trace ID", key=f"copy_{trace['trace_id']}"):
st.code(trace['trace_id'])
with col2:
if st.button("πΎ Save as Good Example", key=f"good_{trace['trace_id']}"):
st.success("β
Marked as good example (feature coming soon)")
with col3:
if st.button("π Save as Bad Example", key=f"bad_{trace['trace_id']}"):
st.warning("β οΈ Marked as bad example (feature coming soon)")
def main():
"""Main Streamlit app."""
st.set_page_config(
page_title="PatchPro Trace Viewer",
page_icon="π",
layout="wide"
)
st.title("π PatchPro Trace Viewer")
st.caption("Analyze patch generation traces and LLM interactions")
# Get trace directory
trace_dir = get_trace_dir()
# Load database
conn = load_trace_db(trace_dir)
if conn is None:
st.error(f"β No traces database found at `{trace_dir / 'traces.db'}`")
st.info("""
**How to generate traces:**
1. Run PatchPro with agentic mode enabled:
```bash
patchpro analyze-pr --base main --head HEAD --with-llm
```
2. Traces will be saved to `.patchpro/traces/`
3. Rerun this viewer to see traces
""")
return
# Show trace directory
st.success(f"β
Connected to: `{trace_dir}`")
# Get summary stats
stats = get_summary_stats(conn)
# Render summary metrics
render_summary_metrics(stats)
st.divider()
# Filters
st.subheader("π Filter Traces")
filter_options = get_filter_options(conn)
col1, col2, col3, col4 = st.columns(4)
with col1:
rule_id_filter = st.selectbox("Rule ID", filter_options["rule_ids"])
with col2:
status_filter = st.selectbox("Status", filter_options["statuses"])
with col3:
strategy_filter = st.selectbox("Strategy", filter_options["strategies"])
with col4:
search_text = st.text_input("Search (message/file)")
# Query traces
traces = query_traces(
conn,
rule_id=rule_id_filter,
status=status_filter,
strategy=strategy_filter,
search_text=search_text,
limit=100
)
# Show count
st.write(f"**Found {len(traces)} traces**")
# Render traces
if traces:
for trace in traces:
render_trace_card(trace, trace_dir)
else:
st.info("No traces match your filters")
# Close connection
conn.close()
if __name__ == "__main__":
main()