Upload jarvis_core/audit.py with huggingface_hub

jarvis_core/audit.py

@@ -0,0 +1,55 @@
+"""jarvis_core.audit — append-only action audit log (Firebase: jarvis_audit).
+
+Never stores secrets (all free text redacted). Answers:
+"What did JARVIS do today?" / "Show my recent actions." / "What failed?"
+"""
+import datetime
+import time
+
+from .store import get_store, redact
+
+COL = "jarvis_audit"
+
+
+def record(user, channel, intent, tool="", args="", approval="none",
+           approval_status="", status="SUCCESS", result="", error=""):
+    return get_store().add(COL, {
+        "timestamp": time.time(), "user": user, "channel": channel,
+        "intent": redact(intent)[:300], "tool": tool,
+        "arguments_summary": redact(args)[:300],
+        "approval_required": approval, "approval_status": approval_status,
+        "execution_status": status, "result_summary": redact(result)[:300],
+        "error": redact(error)[:300],
+    })
+
+
+def recent(limit=10):
+    items = get_store().all(COL)
+    return sorted(items, key=lambda a: a.get("timestamp", 0), reverse=True)[:limit]
+
+
+def today():
+    day = datetime.datetime.now().strftime("%Y-%m-%d")
+    out = []
+    for a in recent(200):
+        ts = datetime.datetime.fromtimestamp(a.get("timestamp", 0)).strftime("%Y-%m-%d")
+        if ts == day:
+            out.append(a)
+    return out
+
+
+def failures(limit=10):
+    items = [a for a in get_store().all(COL)
+             if a.get("execution_status") == "FAILED"]
+    return sorted(items, key=lambda a: a.get("timestamp", 0), reverse=True)[:limit]
+
+
+def format_row(a):
+    ts = datetime.datetime.fromtimestamp(a.get("timestamp", 0)).strftime("%H:%M")
+    base = (f"{ts} {a.get('channel')}: {a.get('intent')} "
+            f"[{a.get('tool')}] {a.get('execution_status')}")
+    if a.get("error"):
+        base += f" — {a.get('error')}"
+    elif a.get("result_summary"):
+        base += f" — {a.get('result_summary')}"
+    return base