v2 #169
v2 #169
Conversation
- Add keyword field to UsageEntry model for message tagging - Update monitor UI to display keywords alongside token counts - Implement keyword extraction from message content (no AI) - Add writer utility for manual keyword logging - Keywords auto-generated from first 3-5 words of messages 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <[email protected]>
WalkthroughThis PR extends Claude Monitor with keyword tagging and a live monitoring dashboard. It adds an optional Changes
Sequence DiagramsequenceDiagram
participant Main as Main Loop
participant KeyListener as Key Listener Thread
participant Reader as Data Reader
participant Renderer as Rich Renderer
participant Console as Console
Main->>+KeyListener: Launch background listener
Note over KeyListener: Listens for 1,2,3,q keys
loop Every 500ms (2 Hz)
Main->>Reader: get_recent_entries(hours=24)
Reader-->>Main: filtered UsageEntry list
Main->>Reader: load_recent_raw_prompts(limit)
Reader-->>Main: recent prompts with text
Main->>Renderer: render_combined_layout()
Renderer->>Renderer: render_usage_panel() + render_overlay()
Renderer-->>Main: combined Rich layout
Main->>Console: Live.update(layout)
end
alt User presses 'q' or KeyboardInterrupt
KeyListener->>Main: set stop_listen flag
Main->>Console: Clear and print goodbye
end
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~25 minutes The changes span multiple modules with new logic (two-pass reader, UI rendering with threading) and new functions. While substantial, the patterns are relatively clear: data propagation is logical, writer functions follow standard file operations, and the UI uses Rich conventions. Moderate cognitive load due to interconnected changes across layers. Poem
Pre-merge checks and finishing touches❌ Failed checks (1 warning, 1 inconclusive)
✅ Passed checks (1 passed)
✨ Finishing touches
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 5
🧹 Nitpick comments (9)
src/claude_monitor/core/models.py (1)
32-32: Model the prompt text explicitly to avoid dynamic attributes.Reader attaches entry.text at runtime. Add an Optional[str] text field here to keep types consistent and avoid ad‑hoc attrs.
@dataclass class UsageEntry: @@ request_id: str = "" - keyword: Optional[str] = None + keyword: Optional[str] = None + text: Optional[str] = Nonesrc/claude_monitor/data/reader.py (5)
130-161: Two‑pass loads entire file into memory; may spike RAM on large logs.Keep the first pass (user text index), but stream the second pass or chunk the file to bound memory.
- Build user_messages in pass 1.
- Reopen file and stream pass 2 without accumulating all_lines.
- Or process in chunks (e.g., 50k lines) to cap peak memory.
217-224: Dedup hash requires both message_id and request_id; fallback would reduce dupes.Some lines may miss one field, leading to no hash and potential duplicates.
- Fallback to message_id alone, or hash of message.id + timestamp when request_id missing.
- Consider stable hash of relevant fields when IDs absent.
269-291: Keyword derivation heuristic is inconsistent (5 words for short texts, 3 for long).Make it deterministic; e.g., always first N words capped by max length.
- if not keyword and text: - words = text.strip().split() - keyword = " ".join(words[:5]) if len(words) <= 5 else " ".join(words[:3]) - if len(keyword) > 40: - keyword = keyword[:37] + "..." + if not keyword and text: + N = 5 + MAX = 40 + words = text.strip().split() + keyword = " ".join(words[:N])[:MAX].rstrip() + if len(" ".join(words[:N])) > MAX: + keyword = keyword[:-3] + "..."
305-308: Avoid dynamic attribute injection for entry.text.Prefer modeling text on UsageEntry (see models.py suggestion) and assign via constructor.
- if hasattr(entry, "__dict__"): - entry.text = text + # UsageEntry now has `text`; set via constructor or here for back-compat + entry.text = text
311-313: Catching JSONDecodeError here is misleading.Second pass operates on dicts; JSON decoding won’t occur. Narrow exceptions (KeyError/TypeError/ValueError) or remove the block.
- except (KeyError, ValueError, TypeError, AttributeError) as e: + except (KeyError, ValueError, TypeError, AttributeError) as e: logger.debug("Failed to map entry: %s: %s", type(e).__name__, e) return Nonesrc/claude_monitor/core/monitor_ui.py (2)
120-127: Use actual cost from entries when available; fall back to estimate.Avoid hard‑coding a per‑token rate if cost_usd is present.
- cost_per_token = 0.000002 - total_cost = total_tokens * cost_per_token + cost_values = [getattr(e, "cost_usd", 0.0) for e in entries] + total_cost = sum(cost_values) if any(cost_values) else total_tokens * 0.000002
114-124: Frequent full reload of logs every 2s; consider bounding I/O.Cache entries between refreshes, use hours_back to limit reads, and avoid include_raw unless needed.
- Pass hours_back=24 to load_usage_entries.
- Keep last mtime per file and only re‑parse on change.
- Drop include_raw=True in overlay when not used directly.
Also applies to: 149-160, 216-220
src/claude_monitor/data/writer.py (1)
133-141: Non‑atomic rewrite risks data loss on crash; use temp file + atomic replace.Switch to atomic write to prevent partial files.
- try: - with open(path, "w", encoding="utf-8") as f: - for e in lines: - f.write(json.dumps(e) + "\n") - print(f"✅ Added keyword '{keyword}' to message {message_id}") - return True + try: + from tempfile import NamedTemporaryFile + tmp = NamedTemporaryFile("w", delete=False, encoding="utf-8", dir=os.path.dirname(path)) + try: + with tmp as f: + for e in lines: + f.write(json.dumps(e) + "\n") + os.replace(tmp.name, path) # atomic on POSIX/Windows + finally: + try: + if os.path.exists(tmp.name): + os.unlink(tmp.name) + except Exception: + pass + print(f"✅ Added keyword '{keyword}' to message {message_id}") + return True
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (4)
src/claude_monitor/core/models.py(1 hunks)src/claude_monitor/core/monitor_ui.py(1 hunks)src/claude_monitor/data/reader.py(9 hunks)src/claude_monitor/data/writer.py(1 hunks)
🧰 Additional context used
🧬 Code graph analysis (2)
src/claude_monitor/core/monitor_ui.py (1)
src/claude_monitor/data/reader.py (1)
load_usage_entries(37-77)
src/claude_monitor/data/reader.py (3)
src/claude_monitor/core/models.py (2)
CostMode(11-16)UsageEntry(20-32)src/claude_monitor/utils/time_utils.py (1)
TimezoneHandler(348-458)src/claude_monitor/core/pricing.py (1)
calculate_cost_for_entry(185-230)
| sys.path.append(os.path.expanduser("~/code/Claude-Code-Usage-Monitor/src")) | ||
| from claude_monitor.data.reader import load_usage_entries | ||
|
|
There was a problem hiding this comment.
Remove hardcoded sys.path modification.
This is brittle and leaks a local path into production. Rely on installed package imports.
-sys.path.append(os.path.expanduser("~/code/Claude-Code-Usage-Monitor/src"))
-from claude_monitor.data.reader import load_usage_entries
+from claude_monitor.data.reader import load_usage_entries📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| sys.path.append(os.path.expanduser("~/code/Claude-Code-Usage-Monitor/src")) | |
| from claude_monitor.data.reader import load_usage_entries | |
| from claude_monitor.data.reader import load_usage_entries | |
🤖 Prompt for AI Agents
In src/claude_monitor/core/monitor_ui.py around lines 29-31, remove the
hardcoded
sys.path.append(os.path.expanduser("~/code/Claude-Code-Usage-Monitor/src")) and
the local-path import style; instead rely on the package being installed and
import claude_monitor.data.reader normally. If this file is used in development,
move any local dev path adjustments to a separate bootstrap/dev-only script or
use editable installs (pip install -e .) and update tests/runtime to ensure the
package is on PYTHONPATH rather than mutating sys.path in production code.
| def listen_for_keys(): | ||
| global last_n_display, stop_listen | ||
| fd = system.stdin.fileno() | ||
| old = termios.tcgetattr(fd) | ||
| tty.setcbreak(fd) | ||
| try: | ||
| while not stop_listen: | ||
| if system.stdin in select([system.stdin], [], [], 0.1)[0]: | ||
| ch = system.stdin.read(1) | ||
| if ch == "1": | ||
| last_n_display = 3 | ||
| elif ch == "2": | ||
| last_n_display = 15 | ||
| elif ch == "3": | ||
| last_n_display = 30 | ||
| elif ch.lower() == "q": | ||
| stop_listen = True | ||
| finally: | ||
| termios.tcsetattr(fd, termios.TCSADRAIN, old) | ||
|
|
There was a problem hiding this comment.
Keyboard input is POSIX‑only (termios/tty); breaks on Windows/non‑TTY.
Provide a cross‑platform path (msvcrt on Windows) and guard for non‑interactive environments.
-def listen_for_keys():
+def listen_for_keys():
global last_n_display, stop_listen
- fd = system.stdin.fileno()
- old = termios.tcgetattr(fd)
- tty.setcbreak(fd)
+ if os.name == "nt":
+ import msvcrt
+ while not stop_listen:
+ if msvcrt.kbhit():
+ ch = msvcrt.getwch()
+ if ch == "1": last_n_display = 3
+ elif ch == "2": last_n_display = 15
+ elif ch == "3": last_n_display = 30
+ elif ch.lower() == "q": stop_listen = True
+ time.sleep(0.05)
+ return
+ fd = system.stdin.fileno()
+ old = termios.tcgetattr(fd)
+ tty.setcbreak(fd)
try:
while not stop_listen:
if system.stdin in select([system.stdin], [], [], 0.1)[0]:
ch = system.stdin.read(1)
if ch == "1":
last_n_display = 3
elif ch == "2":
last_n_display = 15
elif ch == "3":
last_n_display = 30
elif ch.lower() == "q":
stop_listen = True
finally:
termios.tcsetattr(fd, termios.TCSADRAIN, old)Also consider honoring TTY_INTERACTIVE/TTY_COMPATIBLE to force non‑interactive mode in CI. Based on learnings.
Committable suggestion skipped: line range outside the PR's diff.
🤖 Prompt for AI Agents
In src/claude_monitor/core/monitor_ui.py around lines 43-62, the current
listen_for_keys() uses POSIX-only termios/tty and will fail on Windows or
non-interactive CI; change it to a cross-platform implementation: at function
start check if interactive by calling sys.stdin.isatty() and respect env vars
TTY_INTERACTIVE/TTY_COMPATIBLE (allow forcing non-interactive mode); if not
interactive, return immediately or set stop_listen so it doesn’t block; for
Windows (sys.platform == "win32") import msvcrt and use
msvcrt.kbhit()/msvcrt.getwch() in the loop instead of termios/tty; for POSIX
keep the existing termios/tty logic but isolate it under the platform guard and
ensure termios.tcsetattr is always called in finally; import OS-specific modules
conditionally to avoid import errors.
| total_entries = len(entries) | ||
| recent = list(enumerate(entries[-last_n_display:], start=total_entries - last_n_display + 1)) | ||
| raw_prompts = load_recent_raw_prompts(limit=last_n_display * 2) | ||
|
|
||
| msgs = [] | ||
| for idx, e in recent: | ||
| tokens_used = e.input_tokens + e.output_tokens | ||
| t = e.timestamp.strftime("%H:%M:%S") | ||
| model = e.model | ||
| keyword = getattr(e, "keyword", None) | ||
| keyword_text = f" - {keyword}" if keyword else "" | ||
|
|
||
| # find snippet |
There was a problem hiding this comment.
🛠️ Refactor suggestion | 🟠 Major
Index can go negative when fewer entries than last_n_display.
Clamp the starting index to 1 for clean numbering.
- recent = list(enumerate(entries[-last_n_display:], start=total_entries - last_n_display + 1))
+ slice_entries = entries[-last_n_display:]
+ start_idx = max(1, total_entries - len(slice_entries) + 1)
+ recent = list(enumerate(slice_entries, start=start_idx))
@@
- total = sum(e.input_tokens + e.output_tokens for _, e in recent)
- avg = total / len(recent)
+ total = sum(e.input_tokens + e.output_tokens for _, e in recent)
+ avg = total / max(1, len(recent))Also applies to: 188-195
🤖 Prompt for AI Agents
In src/claude_monitor/core/monitor_ui.py around lines 157 to 169, the computed
start index for enumerate can be negative when total_entries < last_n_display;
change the start calculation to clamp to 1 by using max(1, total_entries -
last_n_display + 1) so numbering never goes below 1, and apply the same fix to
the similar block at lines 188-195.
| # Second pass: process assistant messages and link to user messages | ||
| for data in all_lines: | ||
| try: | ||
| if not _should_process_entry(data, cutoff_time, processed_hashes, timezone_handler): | ||
| continue | ||
|
|
||
| logger.debug( | ||
| f"File {file_path.name}: {entries_read} read, " | ||
| f"{entries_filtered} filtered out, {entries_mapped} successfully mapped" | ||
| ) | ||
| # Add user text to assistant message if available | ||
| if data.get("type") == "assistant": | ||
| parent_uuid = data.get("parentUuid") | ||
| if parent_uuid and parent_uuid in user_messages: | ||
| data["_user_text"] = user_messages[parent_uuid] | ||
|
|
||
| entry = _map_to_usage_entry(data, mode, timezone_handler, pricing_calculator) | ||
| if entry: | ||
| entries.append(entry) | ||
| _update_processed_hashes(data, processed_hashes) | ||
| if include_raw: | ||
| raw_data.append(data) | ||
| except json.JSONDecodeError: |
There was a problem hiding this comment.
include_raw is not truly “raw” (mutated with _user_text).
You append data after injecting _user_text, so callers don’t get pristine lines.
- if data.get("type") == "assistant":
- parent_uuid = data.get("parentUuid")
- if parent_uuid and parent_uuid in user_messages:
- data["_user_text"] = user_messages[parent_uuid]
+ # Work on a copy when enriching to keep raw_data pristine
+ enriched = dict(data)
+ if enriched.get("type") == "assistant":
+ parent_uuid = enriched.get("parentUuid")
+ if parent_uuid and parent_uuid in user_messages:
+ enriched["_user_text"] = user_messages[parent_uuid]
@@
- entry = _map_to_usage_entry(data, mode, timezone_handler, pricing_calculator)
+ entry = _map_to_usage_entry(enriched, mode, timezone_handler, pricing_calculator)
@@
- if include_raw:
- raw_data.append(data)
+ if include_raw:
+ raw_data.append(data) # original dict, unmodified📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| # Second pass: process assistant messages and link to user messages | |
| for data in all_lines: | |
| try: | |
| if not _should_process_entry(data, cutoff_time, processed_hashes, timezone_handler): | |
| continue | |
| logger.debug( | |
| f"File {file_path.name}: {entries_read} read, " | |
| f"{entries_filtered} filtered out, {entries_mapped} successfully mapped" | |
| ) | |
| # Add user text to assistant message if available | |
| if data.get("type") == "assistant": | |
| parent_uuid = data.get("parentUuid") | |
| if parent_uuid and parent_uuid in user_messages: | |
| data["_user_text"] = user_messages[parent_uuid] | |
| entry = _map_to_usage_entry(data, mode, timezone_handler, pricing_calculator) | |
| if entry: | |
| entries.append(entry) | |
| _update_processed_hashes(data, processed_hashes) | |
| if include_raw: | |
| raw_data.append(data) | |
| except json.JSONDecodeError: | |
| # Second pass: process assistant messages and link to user messages | |
| for data in all_lines: | |
| try: | |
| if not _should_process_entry(data, cutoff_time, processed_hashes, timezone_handler): | |
| continue | |
| # Work on a copy when enriching to keep raw_data pristine | |
| enriched = dict(data) | |
| if enriched.get("type") == "assistant": | |
| parent_uuid = enriched.get("parentUuid") | |
| if parent_uuid and parent_uuid in user_messages: | |
| enriched["_user_text"] = user_messages[parent_uuid] | |
| entry = _map_to_usage_entry(enriched, mode, timezone_handler, pricing_calculator) | |
| if entry: | |
| entries.append(entry) | |
| _update_processed_hashes(data, processed_hashes) | |
| if include_raw: | |
| raw_data.append(data) # original dict, unmodified | |
| except json.JSONDecodeError: |
🤖 Prompt for AI Agents
In src/claude_monitor/data/reader.py around lines 165-183, include_raw currently
receives the mutated `data` (it gets `_user_text` injected), so callers don't
get the original line; fix by appending a deep copy of `data` to `raw_data`
before mutating it (or explicitly create original_copy = copy.deepcopy(data) and
append that), ensure you import `copy` at the top, then proceed to inject
`_user_text` and continue using `data` for mapping so stored raw entries remain
pristine.
| if log_path is None: | ||
| # Prefer ~/.claude/projects, fallback to ~/.claude/code | ||
| log_path = next((Path(p) for p in USAGE_PATHS if os.path.exists(os.path.dirname(p))), Path(USAGE_PATHS[0])) | ||
|
|
||
| log_path.parent.mkdir(parents=True, exist_ok=True) | ||
|
|
There was a problem hiding this comment.
Type bug: log_path may be str; using .parent will fail.
Always coerce to Path before use.
- if log_path is None:
- # Prefer ~/.claude/projects, fallback to ~/.claude/code
- log_path = next((Path(p) for p in USAGE_PATHS if os.path.exists(os.path.dirname(p))), Path(USAGE_PATHS[0]))
+ if log_path:
+ log_path = Path(log_path)
+ else:
+ # Prefer ~/.claude/projects, fallback to ~/.claude/code
+ log_path = next(
+ (Path(p) for p in USAGE_PATHS if os.path.exists(os.path.dirname(p))),
+ Path(USAGE_PATHS[0]),
+ )🤖 Prompt for AI Agents
In src/claude_monitor/data/writer.py around lines 58-63, log_path may be a str
so calling .parent will fail; coerce to pathlib.Path before any attribute access
and use Path operations when checking existence. Replace
os.path.exists(os.path.dirname(p)) with Path(p).parent.exists() (or
Path(p).exists() where appropriate) inside the generator and ensure the fallback
is Path(USAGE_PATHS[0]); finally set log_path = Path(log_path) immediately after
selection and then call log_path.parent.mkdir(parents=True, exist_ok=True).
1 participant
terminal
Summary by CodeRabbit