logger.py

"""
logger.py - Structured JSONL audit logging for all autonomous agent actions.
"""
import json, os
from datetime import datetime, timezone
from typing import Optional, Dict, List
from pathlib import Path

G="\033[32m";C="\033[36m";Y="\033[33m";R="\033[31m";B="\033[34m";RESET="\033[0m"

class AgentLogger:
    """Color console + persistent JSONL file logging."""
    def __init__(self, log_dir="/a0/usr/workdir/agentic_wallet/logs",
                 agent_id="agentic-wallet-v1", verbose=True):
        self.log_dir=Path(log_dir); self.agent_id=agent_id; self.verbose=verbose
        self.log_dir.mkdir(parents=True, exist_ok=True)
        self._fa=self.log_dir/"actions.jsonl"
        self._ft=self.log_dir/"transactions.jsonl"
        self._fe=self.log_dir/"errors.jsonl"
        self._ev: List[Dict]=[]
        self._start=datetime.now(timezone.utc).isoformat()
        self._w({"event":"session_start","agent_id":agent_id,"time":self._start},self._fa)
        if verbose:
            print(f"\n{chr(61)*60}")
            print(f"  Agentic Wallet SDK | agent: {agent_id}")
            print(f"  session: {self._start}")
            print(f"  logs:    {self.log_dir}")
            print(f"{chr(61)*60}\n")

    def _ts(self): return datetime.now(timezone.utc).isoformat()+"Z"
    def _w(self, data, path):
        e={"timestamp":self._ts(),"agent_id":self.agent_id,**data}
        open(path,"a").write(json.dumps(e)+"\n"); self._ev.append(e)
    def _p(self, col, tag, msg, extra=None):
        if not self.verbose: return
        ts=datetime.now(timezone.utc).strftime("%H:%M:%S")
        print(f"{col}[{ts}][{tag}]{RESET} {msg}")
        if extra:
            for k,v in extra.items():
                sv=str(v); sv=sv[:80]+"..." if len(sv)>80 else sv
                print(f"  {col}  {k}:{RESET} {sv}")
    def info(self,msg,**kw): self._p(C,"INFO ",msg,kw or None); self._w({"level":"INFO","message":msg,**kw},self._fa)
    def warning(self,msg,**kw): self._p(Y,"WARN ",msg,kw or None); self._w({"level":"WARNING","message":msg,**kw},self._fa)
    def error(self,msg,**kw): self._p(R,"ERROR",msg,kw or None); self._w({"level":"ERROR","message":msg,**kw},self._fe)
    def log_action(self,action_type,intent,params=None,result=None,success=True):
        self._p(G,"ACT:"+("OK  " if success else "FAIL"),f"{action_type}: {intent}",params)
        self._w({"level":"ACTION","action_type":action_type,"intent":intent,
                 "params":params or {},"result":result or {},"success":success},self._fa)
    def log_transaction(self,tx_type,from_addr,to_addr,amount,token,
                        signature=None,success=True,error=None,metadata=None):
        icon="\u2713" if success else "\u2717"
        self._p(B,"TX   ",f"{icon} {tx_type}: {amount:.6f} {token}",
                {"sig":(signature[:20]+"...") if signature else "none","to":(to_addr or "")[:24]})
        self._w({"level":"TX","tx_type":tx_type,"from":from_addr,"to":to_addr,
                 "amount":amount,"token":token,"signature":signature,
                 "success":success,"error":error,"metadata":metadata or {}},self._ft)
    def log_llm_decision(self,prompt_summary,decision,reasoning=None):
        short=(reasoning[:120]+"...") if reasoning and len(reasoning)>120 else reasoning
        self._p(C,"LLM  ",f"Decision: {decision[:80]}",{"reasoning":short} if short else None)
        self._w({"level":"LLM","prompt":prompt_summary,"decision":decision,"reasoning":reasoning},self._fa)
    def log_safety_check(self,action,decision,reason):
        col=G if decision=="allow" else R; icon="\u2713" if decision=="allow" else "\u2717"
        self._p(col,"SAFE ",f"{icon} [{decision.upper()}] {action[:48]}: {reason[:55]}")
    def get_session_summary(self):
        a=[e for e in self._ev if e.get("level")=="ACTION"]
        t=[e for e in self._ev if e.get("level")=="TX"]
        er=[e for e in self._ev if e.get("level")=="ERROR"]
        return {"session_start":self._start,"total_events":len(self._ev),
                "actions_taken":len(a),"transactions":len(t),
                "successful_txns":sum(1 for x in t if x.get("success")),"errors":len(er)}
    def print_summary(self):
        s=self.get_session_summary()
        print("\n"+"="*50+"\n  Session Summary")
        for k,v in s.items(): print(f"  {k}: {v}")
        print("="*50+"\n")