Fix linting issues: replace ambiguous Unicode characters and add timezone to datetime.now() calls

Author: codegen-sh[bot]

QUICK_START_LOGGING.md

@@ -3,8 +3,8 @@
 ## ⚡ TL;DR - 3 Step Process
 
 1. **Import the logger**: `from codegen.shared.logging.get_logger import get_logger`
-2. **Add `extra={}` to your log calls**: `logger.info("message", extra={"key": "value"})`
-3. **Enable telemetry**: `codegen config telemetry enable`
+1. **Add `extra={}` to your log calls**: `logger.info("message", extra={"key": "value"})`
+1. **Enable telemetry**: `codegen config telemetry enable`
 
 **That's it!** Your logs automatically go to Grafana Cloud when telemetry is enabled.
 
@@ -22,11 +22,14 @@ from codegen.shared.logging.get_logger import get_logger
 logger = get_logger(__name__)
 
 # Find any existing console.print() or error handling and add:
-logger.info("Operation completed", extra={
-    "operation": "command_name",
-    "org_id": org_id,  # if available
-    "success": True
-})
+logger.info(
+    "Operation completed",
+    extra={
+        "operation": "command_name",
+        "org_id": org_id,  # if available
+        "success": True,
+    },
+)
 ```
 
 ### 2. Test the Integration Right Now
@@ -48,69 +51,56 @@ codegen config telemetry status
 ## 📝 Copy-Paste Patterns
 
 ### Pattern 1: Operation Start/End
+
 ```python
 logger = get_logger(__name__)
 
 # At start of function
-logger.info("Operation started", extra={
-    "operation": "command.subcommand",
-    "user_input": relevant_input
-})
+logger.info("Operation started", extra={"operation": "command.subcommand", "user_input": relevant_input})
 
 # At end of function
-logger.info("Operation completed", extra={
-    "operation": "command.subcommand",
-    "success": True
-})
+logger.info("Operation completed", extra={"operation": "command.subcommand", "success": True})
 ```
 
 ### Pattern 2: Error Handling
+
 ```python
 try:
     # your existing code
     pass
 except SomeSpecificError as e:
-    logger.error("Specific error occurred", extra={
-        "operation": "command.subcommand",
-        "error_type": "specific_error",
-        "error_details": str(e)
-    }, exc_info=True)
+    logger.error("Specific error occurred", extra={"operation": "command.subcommand", "error_type": "specific_error", "error_details": str(e)}, exc_info=True)
     # your existing error handling
 ```
 
 ### Pattern 3: API Calls
+
 ```python
 # Before API call
-logger.info("Making API request", extra={
-    "operation": "api.request",
-    "endpoint": "agent/run",
-    "org_id": org_id
-})
+logger.info("Making API request", extra={"operation": "api.request", "endpoint": "agent/run", "org_id": org_id})
 
 # After successful API call
-logger.info("API request successful", extra={
-    "operation": "api.request",
-    "endpoint": "agent/run",
-    "response_id": response.get("id"),
-    "status_code": response.status_code
-})
+logger.info("API request successful", extra={"operation": "api.request", "endpoint": "agent/run", "response_id": response.get("id"), "status_code": response.status_code})
 ```
 
 ## 🎯 What to Log (Priority Order)
 
 ### 🔥 High Priority (Add These First)
+
 - **Operation start/end**: When commands begin/complete
 - **API calls**: Requests to your backend
 - **Authentication events**: Login/logout/token issues
 - **Errors**: Any exception or failure
 - **User actions**: Commands run, options selected
 
 ### ⭐ Medium Priority
+
 - **Performance**: Duration of operations
 - **State changes**: Status updates, configuration changes
 - **External tools**: Claude CLI detection, git operations
 
 ### 💡 Low Priority (Nice to Have)
+
 - **Debug info**: Internal state, validation steps
 - **User behavior**: Which features are used most
 
@@ -163,15 +153,19 @@ def create(prompt: str, org_id: int | None = None, ...):
 
 logger = get_logger(__name__)
 
+
 def _run_claude_interactive(resolved_org_id: int, no_mcp: bool | None) -> None:
     session_id = generate_session_id()
 
     # ADD: Log session start
-    logger.info("Claude session started", extra={
-        "operation": "claude.session_start",
-        "session_id": session_id[:8],  # Short version for privacy
-        "org_id": resolved_org_id
-    })
+    logger.info(
+        "Claude session started",
+        extra={
+            "operation": "claude.session_start",
+            "session_id": session_id[:8],  # Short version for privacy
+            "org_id": resolved_org_id,
+        },
+    )
 
     # Your existing code...
 
@@ -180,30 +174,23 @@ def _run_claude_interactive(resolved_org_id: int, no_mcp: bool | None) -> None:
         returncode = process.wait()
 
         # ADD: Log session end
-        logger.info("Claude session completed", extra={
-            "operation": "claude.session_complete",
-            "session_id": session_id[:8],
-            "exit_code": returncode,
-            "status": "COMPLETE" if returncode == 0 else "ERROR"
-        })
+        logger.info(
+            "Claude session completed", extra={"operation": "claude.session_complete", "session_id": session_id[:8], "exit_code": returncode, "status": "COMPLETE" if returncode == 0 else "ERROR"}
+        )
 
     except Exception as e:
         # ADD: Log session error
-        logger.error("Claude session failed", extra={
-            "operation": "claude.session_error",
-            "session_id": session_id[:8],
-            "error": str(e)
-        })
+        logger.error("Claude session failed", extra={"operation": "claude.session_error", "session_id": session_id[:8], "error": str(e)})
 ```
 
 ## 🧪 Verification
 
 After making changes:
 
 1. **Run the command**: Execute your enhanced CLI command
-2. **Check telemetry status**: `codegen config telemetry status`
-3. **Look for logs in Grafana Cloud**: Search for your operation names
-4. **Test with telemetry disabled**: `codegen config telemetry disable` - should still work normally
+1. **Check telemetry status**: `codegen config telemetry status`
+1. **Look for logs in Grafana Cloud**: Search for your operation names
+1. **Test with telemetry disabled**: `codegen config telemetry disable` - should still work normally
 
 ## 🚀 Progressive Enhancement
 

src/codegen/cli/commands/claude/__init__.py

@@ -1,2 +1 @@
 """Claude Code integration commands."""
-

src/codegen/cli/commands/claude/claude_log_utils.py

@@ -4,7 +4,7 @@
 import os
 import re
 from pathlib import Path
-from typing import Dict, Any, Optional
+from typing import Any, Optional
 
 
 def get_hyphenated_cwd() -> str:
@@ -39,7 +39,7 @@ def get_claude_session_log_path(session_id: str) -> Path:
     return log_file
 
 
-def parse_jsonl_line(line: str) -> Optional[Dict[str, Any]]:
+def parse_jsonl_line(line: str) -> Optional[dict[str, Any]]:
     """Parse a single line from a JSONL file.
 
     Args:
@@ -85,13 +85,13 @@ def read_existing_log_lines(log_path: Path) -> int:
         return 0
 
     try:
-        with open(log_path, "r", encoding="utf-8") as f:
+        with open(log_path, encoding="utf-8") as f:
             return sum(1 for _ in f)
     except (OSError, UnicodeDecodeError):
         return 0
 
 
-def validate_log_entry(log_entry: Dict[str, Any]) -> bool:
+def validate_log_entry(log_entry: dict[str, Any]) -> bool:
     """Validate a log entry before sending to API.
 
     Args:
@@ -112,7 +112,7 @@ def validate_log_entry(log_entry: Dict[str, Any]) -> bool:
     return True
 
 
-def format_log_for_api(log_entry: Dict[str, Any]) -> Dict[str, Any]:
+def format_log_for_api(log_entry: dict[str, Any]) -> dict[str, Any]:
     """Format a log entry for sending to the API.
 
     Args:

src/codegen/cli/commands/claude/claude_log_watcher.py

@@ -1,20 +1,18 @@
 """Claude Code session log watcher implementation."""
 
-import time
 import threading
-from pathlib import Path
-from typing import Optional, Callable, Dict, Any
-
-from .quiet_console import console
+import time
+from typing import Any, Callable, Optional
 
-from .claude_log_utils import get_claude_session_log_path, parse_jsonl_line, read_existing_log_lines, validate_log_entry, format_log_for_api
+from .claude_log_utils import format_log_for_api, get_claude_session_log_path, parse_jsonl_line, read_existing_log_lines, validate_log_entry
 from .claude_session_api import send_claude_session_log
+from .quiet_console import console
 
 
 class ClaudeLogWatcher:
     """Watches Claude Code session log files for new entries and sends them to the API."""
 
-    def __init__(self, session_id: str, org_id: Optional[int] = None, poll_interval: float = 1.0, on_log_entry: Optional[Callable[[Dict[str, Any]], None]] = None):
+    def __init__(self, session_id: str, org_id: Optional[int] = None, poll_interval: float = 1.0, on_log_entry: Optional[Callable[[dict[str, Any]], None]] = None):
         """Initialize the log watcher.
 
         Args:
@@ -106,7 +104,7 @@ def _check_for_new_entries(self) -> None:
         except Exception as e:
             console.print(f"⚠️  Error reading log file: {e}", style="yellow")
 
-    def _read_new_lines(self, start_line: int, end_line: int) -> list[Dict[str, Any]]:
+    def _read_new_lines(self, start_line: int, end_line: int) -> list[dict[str, Any]]:
         """Read new lines from the log file.
 
         Args:
@@ -119,7 +117,7 @@ def _read_new_lines(self, start_line: int, end_line: int) -> list[Dict[str, Any]
         entries = []
 
         try:
-            with open(self.log_path, "r", encoding="utf-8") as f:
+            with open(self.log_path, encoding="utf-8") as f:
                 lines = f.readlines()
 
                 # Read only the new lines
@@ -135,7 +133,7 @@ def _read_new_lines(self, start_line: int, end_line: int) -> list[Dict[str, Any]
 
         return entries
 
-    def _process_log_entry(self, log_entry: Dict[str, Any]) -> None:
+    def _process_log_entry(self, log_entry: dict[str, Any]) -> None:
         """Process a single log entry.
 
         Args:
@@ -161,7 +159,7 @@ def _process_log_entry(self, log_entry: Dict[str, Any]) -> None:
         # Send to API
         self._send_log_entry(formatted_entry)
 
-    def _send_log_entry(self, log_entry: Dict[str, Any]) -> None:
+    def _send_log_entry(self, log_entry: dict[str, Any]) -> None:
         """Send a log entry to the API.
 
         Args:
@@ -181,7 +179,7 @@ def _send_log_entry(self, log_entry: Dict[str, Any]) -> None:
             self.total_send_failures += 1
             console.print(f"⚠️  Failed to send log entry: {e}", style="yellow")
 
-    def get_stats(self) -> Dict[str, Any]:
+    def get_stats(self) -> dict[str, Any]:
         """Get watcher statistics.
 
         Returns:
@@ -204,9 +202,9 @@ class ClaudeLogWatcherManager:
     """Manages multiple log watchers for different sessions."""
 
     def __init__(self):
-        self.watchers: Dict[str, ClaudeLogWatcher] = {}
+        self.watchers: dict[str, ClaudeLogWatcher] = {}
 
-    def start_watcher(self, session_id: str, org_id: Optional[int] = None, poll_interval: float = 1.0, on_log_entry: Optional[Callable[[Dict[str, Any]], None]] = None) -> bool:
+    def start_watcher(self, session_id: str, org_id: Optional[int] = None, poll_interval: float = 1.0, on_log_entry: Optional[Callable[[dict[str, Any]], None]] = None) -> bool:
         """Start a log watcher for a session.
 
         Args:
@@ -252,7 +250,7 @@ def get_active_sessions(self) -> list[str]:
         """
         return list(self.watchers.keys())
 
-    def get_watcher_stats(self, session_id: str) -> Optional[Dict[str, Any]]:
+    def get_watcher_stats(self, session_id: str) -> Optional[dict[str, Any]]:
         """Get stats for a specific watcher.
 
         Args:
@@ -265,7 +263,7 @@ def get_watcher_stats(self, session_id: str) -> Optional[Dict[str, Any]]:
             return self.watchers[session_id].get_stats()
         return None
 
-    def get_all_stats(self) -> Dict[str, Dict[str, Any]]:
+    def get_all_stats(self) -> dict[str, dict[str, Any]]:
         """Get stats for all active watchers.
 
         Returns:

src/codegen/cli/commands/claude/claude_session_api.py

@@ -5,12 +5,13 @@
 from typing import Optional
 
 import requests
-from .quiet_console import console
 
 from codegen.cli.api.endpoints import API_ENDPOINT
 from codegen.cli.auth.token_manager import get_current_token
 from codegen.cli.utils.org import resolve_org_id
 
+from .quiet_console import console
+
 
 class ClaudeSessionAPIError(Exception):
     """Exception raised for Claude session API errors."""

src/codegen/cli/commands/claude/main.py

@@ -290,7 +290,7 @@ def signal_handler(signum, frame):
             console.print("✅ Claude Code finished successfully", style="green")
 
     except FileNotFoundError:
-        logger.error(
+        logger.exception(
             "Claude Code executable not found",
             extra={"operation": "claude.interactive", "org_id": resolved_org_id, "claude_session_id": session_id, "error_type": "claude_executable_not_found", **_get_session_context()},
         )

src/codegen/cli/commands/claude/quiet_console.py

@@ -8,6 +8,7 @@
 
 import io
 import os
+
 from rich.console import Console
 
 
@@ -30,4 +31,3 @@ def _create_console() -> Console:
 
 # Shared console used across Claude CLI modules
 console = _create_console()
-

src/codegen/cli/commands/config/telemetry.py

@@ -1,12 +1,9 @@
 """Telemetry configuration commands."""
 
 import json
-from pathlib import Path
 
 import typer
 from rich.console import Console
-from rich.panel import Panel
-from rich.syntax import Syntax
 from rich.table import Table
 
 from codegen.cli.telemetry import update_telemetry_consent

src/codegen/cli/commands/org/__init__.py

@@ -2,4 +2,4 @@
 
 from .main import org
 
-__all__ = ["org"]
+__all__ = ["org"]