| """Simple structured logger.""" |
|
|
| import json |
| import sys |
| from datetime import datetime |
| from pathlib import Path |
|
|
|
|
| class Logger: |
| """Structured logging for cloud coder agents.""" |
|
|
| def __init__(self, log_file: Path = None): |
| self.log_file = log_file or Path("/tmp/cloud-coder.log") |
| self.ensure_log_dir() |
|
|
| def ensure_log_dir(self): |
| """Ensure log directory exists.""" |
| self.log_file.parent.mkdir(parents=True, exist_ok=True) |
|
|
| def log(self, level: str, message: str, **kwargs): |
| """Write a structured log entry.""" |
| entry = { |
| "timestamp": datetime.utcnow().isoformat() + "Z", |
| "level": level, |
| "message": message, |
| **kwargs, |
| } |
| line = json.dumps(entry) |
| print(line, file=sys.stderr) |
| with open(self.log_file, "a") as f: |
| f.write(line + "\n") |
|
|
| def info(self, message: str, **kwargs): |
| self.log("INFO", message, **kwargs) |
|
|
| def warning(self, message: str, **kwargs): |
| self.log("WARNING", message, **kwargs) |
|
|
| def error(self, message: str, **kwargs): |
| self.log("ERROR", message, **kwargs) |
|
|
| def task(self, task_id: str, status: str, **kwargs): |
| self.log( |
| "TASK", f"Task {task_id} status: {status}", task_id=task_id, status=status, **kwargs |
| ) |
|
|
|
|
| |
| logger = Logger() |
|
|
