feat: Add action logging system for audit and debugging

lib/action_logger.py

@@ -0,0 +1,288 @@
+"""Action logging for PolyClaw - records all operations to JSONL files.
+
+Log files are stored at ~/.openclaw/polyclaw/logs/actions-YYYY-MM-DD.jsonl
+Format: JSONL (one JSON object per line, easy to append and parse)
+Rotation: Daily, with 30-day retention
+"""
+
+import json
+import threading
+import time
+from dataclasses import dataclass, asdict
+from datetime import datetime, timezone, timedelta
+from pathlib import Path
+from typing import Optional, Any
+
+
+def get_storage_dir() -> Path:
+    """Get the storage directory for PolyClaw data."""
+    storage_dir = Path.home() / ".openclaw" / "polyclaw"
+    storage_dir.mkdir(parents=True, exist_ok=True)
+    return storage_dir
+
+
+def get_logs_dir() -> Path:
+    """Get the logs directory."""
+    logs_dir = get_storage_dir() / "logs"
+    logs_dir.mkdir(parents=True, exist_ok=True)
+    return logs_dir
+
+
+# Global lock for thread-safe log writes
+_log_lock = threading.Lock()
+
+# Log retention period in days
+LOG_RETENTION_DAYS = 30
+
+
+@dataclass
+class ActionEntry:
+    """A single action log entry."""
+
+    timestamp: str        # ISO 8601
+    action: str           # "markets.trending", "trade.buy", etc.
+    params: dict          # Input parameters (sanitized)
+    result: str           # "success" | "failure"
+    details: dict         # Result details (tx_hash, order_id, error, etc.)
+    duration_ms: int      # Execution time in milliseconds
+
+
+def sanitize_value(value: Any, key: str = "") -> Any:
+    """Sanitize a value for logging, masking sensitive data.
+
+    Sensitive keys:
+    - private_key, POLYCLAW_PRIVATE_KEY
+    - password, secret
+    - api_key, OPENROUTER_API_KEY
+    """
+    sensitive_keys = {
+        "private_key", "POLYCLAW_PRIVATE_KEY",
+        "password", "secret",
+        "api_key", "OPENROUTER_API_KEY",
+    }
+
+    key_lower = key.lower()
+
+    # Check if this key is sensitive
+    if any(s in key_lower for s in ["private", "secret", "password", "api_key"]):
+        if isinstance(value, str) and len(value) > 10:
+            # Show prefix and suffix
+            if value.startswith("0x"):
+                return f"{value[:8]}...{value[-6:]}"
+            elif value.startswith("sk-"):
+                return f"{value[:6]}...{value[-3:]}"
+            else:
+                return f"{value[:6]}...{value[-4:]}"
+
+    return value
+
+
+def sanitize_dict(d: dict) -> dict:
+    """Recursively sanitize a dictionary for logging."""
+    result = {}
+    for key, value in d.items():
+        if isinstance(value, dict):
+            result[key] = sanitize_dict(value)
+        elif isinstance(value, list):
+            result[key] = [sanitize_dict(v) if isinstance(v, dict) else sanitize_value(v, key) for v in value]
+        else:
+            result[key] = sanitize_value(value, key)
+    return result
+
+
+def get_log_file_path(date: Optional[datetime] = None) -> Path:
+    """Get the log file path for a specific date.
+
+    Args:
+        date: Date to get log file for. Defaults to today (UTC).
+
+    Returns:
+        Path to the log file.
+    """
+    if date is None:
+        date = datetime.now(timezone.utc)
+
+    filename = f"actions-{date.strftime('%Y-%m-%d')}.jsonl"
+    return get_logs_dir() / filename
+
+
+def cleanup_old_logs() -> None:
+    """Delete log files older than LOG_RETENTION_DAYS."""
+    logs_dir = get_logs_dir()
+    cutoff = datetime.now(timezone.utc) - timedelta(days=LOG_RETENTION_DAYS)
+
+    for log_file in logs_dir.glob("actions-*.jsonl"):
+        try:
+            # Extract date from filename
+            date_str = log_file.stem.replace("actions-", "")
+            file_date = datetime.strptime(date_str, "%Y-%m-%d").replace(tzinfo=timezone.utc)
+
+            if file_date < cutoff:
+                log_file.unlink()
+        except (ValueError, OSError):
+            # Ignore files with invalid names or deletion errors
+            pass
+
+
+def log_action(
+    action: str,
+    params: dict,
+    result: str,
+    details: dict,
+    duration_ms: int,
+) -> None:
+    """Log an action to the daily log file.
+
+    Thread-safe: uses a global lock for concurrent writes.
+
+    Args:
+        action: Action type (e.g., "markets.trending", "trade.buy")
+        params: Input parameters (will be sanitized)
+        result: "success" or "failure"
+        details: Result details (tx_hash, order_id, error, etc.)
+        duration_ms: Execution time in milliseconds
+    """
+    entry = ActionEntry(
+        timestamp=datetime.now(timezone.utc).isoformat(),
+        action=action,
+        params=sanitize_dict(params),
+        result=result,
+        details=sanitize_dict(details),
+        duration_ms=duration_ms,
+    )
+
+    log_file = get_log_file_path()
+
+    with _log_lock:
+        # Append to log file
+        with open(log_file, "a") as f:
+            f.write(json.dumps(asdict(entry)) + "\n")
+
+    # Periodically clean up old logs
+    # (Run cleanup ~1% of the time to avoid overhead)
+    if hash(entry.timestamp) % 100 == 0:
+        cleanup_old_logs()
+
+
+class ActionLogger:
+    """Context manager for timing and logging actions.
+
+    Usage:
+        with ActionLogger("markets.trending", {"limit": 20}) as log:
+            markets = await client.get_trending_markets(limit=20)
+            log.set_details({"count": len(markets)})
+            log.success()
+    """
+
+    def __init__(self, action: str, params: dict):
+        """Initialize the action logger.
+
+        Args:
+            action: Action type (e.g., "markets.trending", "trade.buy").
+            params: Input parameters to log.
+        """
+        self.action = action
+        self.params = params
+        self.details: dict = {}
+        self.result: str = "success"
+        self.start_time = time.perf_counter()
+
+    def set_details(self, details: dict) -> None:
+        """Set additional details for the log entry."""
+        self.details.update(details)
+
+    def success(self) -> None:
+        """Mark the action as successful."""
+        self.result = "success"
+
+    def failure(self, error: str) -> None:
+        """Mark the action as failed with an error message."""
+        self.result = "failure"
+        self.details["error"] = error
+
+    def __enter__(self) -> "ActionLogger":
+        """Enter the context manager, returning self for use in the with block."""
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
+        """Exit the context manager, logging the action with duration and result.
+
+        Args:
+            exc_type: Exception type if an exception was raised, else None.
+            exc_val: Exception value if an exception was raised, else None.
+            exc_tb: Exception traceback if an exception was raised, else None.
+        """
+        duration_ms = int((time.perf_counter() - self.start_time) * 1000)
+
+        if exc_type is not None:
+            self.result = "failure"
+            self.details["error"] = str(exc_val)
+
+        log_action(
+            action=self.action,
+            params=self.params,
+            result=self.result,
+            details=self.details,
+            duration_ms=duration_ms,
+        )
+
+
+def get_actions_by_date(date_str: str) -> list[dict]:
+    """Get all actions for a specific date.
+
+    Args:
+        date_str: Date in YYYY-MM-DD format.
+
+    Returns:
+        List of action entries.
+    """
+    try:
+        date = datetime.strptime(date_str, "%Y-%m-%d").replace(tzinfo=timezone.utc)
+    except ValueError:
+        return []
+
+    log_file = get_log_file_path(date)
+
+    if not log_file.exists():
+        return []
+
+    actions = []
+    with open(log_file) as f:
+        for line in f:
+            line = line.strip()
+            if line:
+                try:
+                    actions.append(json.loads(line))
+                except json.JSONDecodeError:
+                    pass
+
+    return actions
+
+
+def get_recent_actions(limit: int = 20) -> list[dict]:
+    """Get the most recent actions from today's log.
+
+    Args:
+        limit: Maximum number of actions to return.
+
+    Returns:
+        List of action entries, most recent first.
+    """
+    actions = get_actions_by_date(datetime.now(timezone.utc).strftime("%Y-%m-%d"))
+    return actions[-limit:][::-1]  # Return last N, reversed (most recent first)
+
+
+def get_actions_by_type(action_prefix: str, limit: int = 50) -> list[dict]:
+    """Get actions filtered by type (action prefix).
+
+    Args:
+        action_prefix: Filter actions starting with this prefix (e.g., "trade")
+        limit: Maximum number of actions to return.
+
+    Returns:
+        List of matching action entries.
+    """
+    actions = get_actions_by_date(datetime.now(timezone.utc).strftime("%Y-%m-%d"))
+
+    filtered = [a for a in actions if a.get("action", "").startswith(action_prefix)]
+    return filtered[-limit:][::-1]  # Return last N, reversed