Add centralized audit log system

- Created internal/audit package for centralized logging
- Replaced per-session log files with single agent/audit file
- Updated all UIs (TUI, web, Emacs) to use centralized audit log
- Modified mailbox manager to log all message deliveries with aliases
- Updated Alog script to stream centralized audit log
- Removed session-specific log streaming in favor of global audit log
This commit is contained in:
Levi Neely 2026-02-17 22:07:39 +01:00
parent 359258f8c5
commit 314d347e76
9 changed files with 245 additions and 104 deletions

View File

@ -586,23 +586,17 @@ func killSelectedSession() {
}
func showLogViewer() {
sess := getSelectedSession()
if sess == nil {
updateStatus("[yellow]No session selected")
return
}
// Read the log from the 'log' file
path := filepath.Join(sess.ID, "log")
// Read the centralized audit log
path := "audit"
logData, err := readLogFile(path)
if err != nil {
updateStatus(fmt.Sprintf("[red]Error reading log: %v", err))
updateStatus(fmt.Sprintf("[red]Error reading audit log: %v", err))
return
}
logText := string(logData)
if logText == "" {
logText = "[gray]No log output yet[-]"
logText = "[gray]No audit log entries yet[-]"
}
textView := tview.NewTextView().
@ -612,7 +606,7 @@ func showLogViewer() {
SetWordWrap(true)
textView.SetBorder(true).
SetTitle(fmt.Sprintf(" Log: %s ", sess.ID[:8])).
SetTitle(" Audit Log ").
SetTitleAlign(tview.AlignLeft)
// Scroll to bottom
@ -632,7 +626,7 @@ func showLogViewer() {
}
logText := string(logData)
if logText == "" {
logText = "[gray]No log output yet[-]"
logText = "[gray]No audit log entries yet[-]"
}
textView.SetText(logText)
textView.ScrollToEnd()

View File

@ -51,7 +51,7 @@ func main() {
http.Handle("/", http.FileServer(http.FS(staticFS)))
http.HandleFunc("/api/sessions", handleSessions)
http.HandleFunc("/api/session/", handleSession)
http.HandleFunc("/api/log/", handleLogStream)
http.HandleFunc("/api/log", handleLogStream)
http.HandleFunc("/api/inbox", handleInbox)
http.HandleFunc("/api/inbox/count", handleInboxCount)
http.HandleFunc("/api/inbox/reply", handleInboxReply)
@ -502,13 +502,7 @@ func updateSessionFile(w http.ResponseWriter, r *http.Request, id, file string)
}
func handleLogStream(w http.ResponseWriter, r *http.Request) {
// Extract session ID from path
id := strings.TrimPrefix(r.URL.Path, "/api/log/")
if id == "" {
http.Error(w, "Invalid session ID", http.StatusBadRequest)
return
}
// Stream the centralized audit log
// Set headers for SSE
w.Header().Set("Content-Type", "text/event-stream")
w.Header().Set("Cache-Control", "no-cache")
@ -521,8 +515,8 @@ func handleLogStream(w http.ResponseWriter, r *http.Request) {
}
defer fs.Close()
// Open log file for streaming
fid, err := fs.Open(filepath.Join(id, "log"), plan9.OREAD)
// Open audit log file for streaming
fid, err := fs.Open("audit", plan9.OREAD)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return

View File

@ -707,17 +707,13 @@
// Close existing stream if any
closeLogPane();
// Find session info for title
const session = allSessions.find(s => s.id === id);
const displayName = session ? (session.alias || session.id.substring(0, 8)) : id.substring(0, 8);
// Update UI
document.getElementById('log-title').textContent = `Log: ${displayName}`;
// Update UI for centralized audit log
document.getElementById('log-title').textContent = 'Audit Log';
document.getElementById('log-content').textContent = '';
document.getElementById('log-pane').classList.remove('hidden');
// Start streaming
logEventSource = new EventSource(`/api/log/${id}`);
// Start streaming centralized audit log
logEventSource = new EventSource('/api/log');
logEventSource.onmessage = (event) => {
const logContent = document.getElementById('log-content');

View File

@ -416,25 +416,18 @@ Backends:
nil t))
(defun anvillm-view-log ()
"View the log for the selected session."
"View the centralized audit log."
(interactive)
(if-let ((session-id (anvillm--get-selected-session)))
(let* ((session-info (anvillm--get-session-info session-id))
(display-name (or (plist-get session-info :alias)
(substring session-id 0 (min 8 (length session-id)))))
(buffer-name (format "*AnviLLM Log: %s*" display-name))
(buffer (get-buffer-create buffer-name)))
(with-current-buffer buffer
(anvillm-log-mode)
(setq anvillm--log-session-id session-id)
(anvillm--refresh-log-buffer))
(pop-to-buffer buffer))
(message "No session selected")))
(let* ((buffer-name "*AnviLLM Audit Log*")
(buffer (get-buffer-create buffer-name)))
(with-current-buffer buffer
(anvillm-log-mode)
(setq anvillm--log-session-id nil) ; Not session-specific anymore
(anvillm--refresh-log-buffer))
(pop-to-buffer buffer)))
(defun anvillm--refresh-log-buffer ()
"Refresh the log content in the current buffer by starting a streaming read."
(unless anvillm--log-session-id
(error "No session ID associated with this buffer"))
;; Kill existing process if any
(when (and anvillm--log-process
@ -444,10 +437,10 @@ Backends:
(let ((inhibit-read-only t)
(log-buffer (current-buffer)))
(erase-buffer)
(insert "Loading log...\n")
(insert "Loading audit log...\n")
;; Start streaming process
(let* ((path (concat anvillm-agent-path "/" anvillm--log-session-id "/log"))
;; Start streaming process for centralized audit log
(let* ((path (concat anvillm-agent-path "/audit"))
(proc (start-process "9p-log-stream" log-buffer anvillm-9p-command "read" path)))
(setq anvillm--log-process proc)
@ -459,10 +452,10 @@ Backends:
(with-current-buffer log-buffer
(let ((inhibit-read-only t)
(at-end (= (point) (point-max))))
;; Clear "Loading log..." on first output
;; Clear "Loading audit log..." on first output
(when (save-excursion
(goto-char (point-min))
(looking-at "Loading log..."))
(looking-at "Loading audit log..."))
(erase-buffer))
;; Insert new output
(goto-char (point-max))
@ -479,7 +472,7 @@ Backends:
(let ((inhibit-read-only t))
(when (and (= (point-min) (point-max))
(not (string-match-p "^run" event)))
(insert "No log output yet.\n"))))))))))
(insert "No audit log entries yet.\n"))))))))))
(defun anvillm-log-refresh ()
"Refresh the log content."

112
internal/audit/audit.go Normal file
View File

@ -0,0 +1,112 @@
package audit
import (
"fmt"
"strings"
"sync"
"time"
)
const (
maxSize = 64 * 1024 * 1024 // 64MB
trimPercent = 25 // Remove 25% when limit reached
)
// Log maintains a centralized audit log of all messages
type Log struct {
entries []string
totalBytes int
mu sync.RWMutex
notifyCh chan struct{}
}
// NewLog creates a new audit log
func NewLog() *Log {
return &Log{
entries: make([]string, 0, 1000),
notifyCh: make(chan struct{}, 1),
}
}
// Append adds a new entry to the audit log
// Format: {timestamp} {message-type} {sender}->{recipient} {subject}: {body}
func (l *Log) Append(msgType, sender, recipient, subject, body string) {
l.mu.Lock()
defer l.mu.Unlock()
// Format timestamp
timestamp := time.Now().Format(time.RFC3339)
// Escape newlines in subject and body for single-line grep/awk
// But for human readability, we'll use multi-line format with blank line separator
entry := fmt.Sprintf("%s %s %s->%s\nSubject: %s\n%s\n",
timestamp, msgType, sender, recipient, subject, body)
l.entries = append(l.entries, entry)
l.totalBytes += len(entry)
// Trim if needed
if l.totalBytes > maxSize {
l.trim()
}
// Notify waiters
select {
case l.notifyCh <- struct{}{}:
default:
}
}
// trim removes the oldest 25% of entries
func (l *Log) trim() {
if len(l.entries) == 0 {
return
}
removeCount := len(l.entries) * trimPercent / 100
if removeCount == 0 {
removeCount = 1
}
// Calculate bytes to remove
removedBytes := 0
for i := 0; i < removeCount && i < len(l.entries); i++ {
removedBytes += len(l.entries[i])
}
// Remove entries
l.entries = l.entries[removeCount:]
l.totalBytes -= removedBytes
}
// Read returns the entire audit log as a string
func (l *Log) Read() string {
l.mu.RLock()
defer l.mu.RUnlock()
return strings.Join(l.entries, "\n")
}
// ReadFrom returns audit log content from a specific byte offset
func (l *Log) ReadFrom(offset int64) (string, bool) {
l.mu.RLock()
defer l.mu.RUnlock()
fullLog := strings.Join(l.entries, "\n")
if offset >= int64(len(fullLog)) {
return "", false
}
return fullLog[offset:], true
}
// WaitForData returns a channel that signals when new data is available
func (l *Log) WaitForData() <-chan struct{} {
return l.notifyCh
}
// Size returns the current size in bytes
func (l *Log) Size() int {
l.mu.RLock()
defer l.mu.RUnlock()
return l.totalBytes
}

View File

@ -1,11 +1,17 @@
package mailbox
import (
"anvillm/internal/audit"
"fmt"
"sync"
"time"
)
// SessionGetter provides access to session aliases
type SessionGetter interface {
GetAlias(id string) string
}
// Manager handles mailbox operations for all sessions (in-memory)
type Manager struct {
// sessionID -> list of messages
@ -14,6 +20,8 @@ type Manager struct {
completed map[string][]*Message
mu sync.RWMutex
idCounter uint64
auditLog *audit.Log
sessions SessionGetter
}
// NewManager creates a new mailbox manager
@ -22,6 +30,7 @@ func NewManager() *Manager {
inboxes: make(map[string][]*Message),
outboxes: make(map[string][]*Message),
completed: make(map[string][]*Message),
auditLog: audit.NewLog(),
}
// Initialize user mailbox
m.inboxes["user"] = []*Message{}
@ -30,6 +39,13 @@ func NewManager() *Manager {
return m
}
// SetSessionGetter sets the session getter for alias lookup
func (m *Manager) SetSessionGetter(sg SessionGetter) {
m.mu.Lock()
defer m.mu.Unlock()
m.sessions = sg
}
// EnsureMailbox initializes mailbox for a session (no-op for in-memory)
func (m *Manager) EnsureMailbox(sessionID string) error {
m.mu.Lock()
@ -98,9 +114,32 @@ func (m *Manager) DeliverToInbox(sessionID string, msg *Message) error {
defer m.mu.Unlock()
m.inboxes[sessionID] = append(m.inboxes[sessionID], msg)
// Format sender and receiver with aliases
sender := m.formatParticipant(msg.From)
receiver := m.formatParticipant(msg.To)
// Audit log the delivery
m.auditLog.Append(string(msg.Type), sender, receiver, msg.Subject, msg.Body)
return nil
}
// formatParticipant formats a participant ID with alias if available
// Returns "session-id" if no alias, or "alias(session-id)" if alias is set
func (m *Manager) formatParticipant(id string) string {
if m.sessions == nil {
return id
}
alias := m.sessions.GetAlias(id)
if alias == "" {
return id
}
return fmt.Sprintf("%s(%s)", alias, id)
}
// GetInbox returns all messages in inbox (copy to prevent modification)
func (m *Manager) GetInbox(sessionID string) []*Message {
m.mu.RLock()
@ -223,3 +262,8 @@ func (m *Manager) GetMessage(sessionID, msgID string) (*Message, error) {
return nil, fmt.Errorf("message not found")
}
// GetAuditLog returns the audit log
func (m *Manager) GetAuditLog() *audit.Log {
return m.auditLog
}

View File

@ -66,6 +66,7 @@ const (
qidCtl
qidList
qidStatus
qidAudit // agent/audit
qidUser // user directory
qidUserInbox // user/inbox
qidUserOutbox // user/outbox
@ -85,7 +86,6 @@ const (
fileCtl = iota
fileIn
fileOut
fileLog
fileState
filePid
fileCwd
@ -99,7 +99,7 @@ const (
fileCount
)
var fileNames = []string{"ctl", "in", "out", "log", "state", "pid", "cwd", "alias", "backend", "context", "role", "tasks", "tmux", "mail"}
var fileNames = []string{"ctl", "in", "out", "state", "pid", "cwd", "alias", "backend", "context", "role", "tasks", "tmux", "mail"}
// Directory names in session
var dirNames = []string{"inbox", "outbox", "completed"}
@ -262,6 +262,9 @@ func (s *Server) walk(cs *connState, fc *plan9.Fcall) *plan9.Fcall {
case "list":
qid = plan9.Qid{Type: QTFile, Path: qidList}
newPath = "/list"
case "audit":
qid = plan9.Qid{Type: QTFile, Path: qidAudit}
newPath = "/audit"
case "user":
qid = plan9.Qid{Type: QTDir, Path: qidUser}
newPath = "/user"
@ -378,19 +381,14 @@ func (s *Server) read(cs *connState, fc *plan9.Fcall) *plan9.Fcall {
if f.qid.Type&QTDir != 0 {
data = s.readDir(f.path, fc.Offset, fc.Count)
} else if f.path == "/audit" {
// Streaming audit log (like tail -f)
data = s.readAuditLog(fc.Offset, fc.Count)
} else {
// Check if this is a log file for streaming behavior
parts := strings.SplitN(f.path, "/", 3)
isLogFile := len(parts) == 3 && parts[2] == "log"
if isLogFile {
data = s.readLogFile(parts[1], fc.Offset, fc.Count)
} else {
content := s.readFile(f.path)
if fc.Offset < uint64(len(content)) {
end := min(int(fc.Offset)+int(fc.Count), len(content))
data = []byte(content[fc.Offset:end])
}
content := s.readFile(f.path)
if fc.Offset < uint64(len(content)) {
end := min(int(fc.Offset)+int(fc.Count), len(content))
data = []byte(content[fc.Offset:end])
}
}
@ -824,6 +822,10 @@ func (s *Server) readDir(path string, offset uint64, count uint32) []byte {
Qid: plan9.Qid{Type: QTFile, Path: qidStatus}, Mode: 0444, Name: "status",
Uid: "q", Gid: "q", Muid: "q",
})
dirs = append(dirs, plan9.Dir{
Qid: plan9.Qid{Type: QTFile, Path: qidAudit}, Mode: 0444, Name: "audit",
Uid: "q", Gid: "q", Muid: "q",
})
dirs = append(dirs, plan9.Dir{
Qid: plan9.Qid{Type: QTDir, Path: qidUser},
Mode: plan9.DMDIR | 0555, Name: "user", Uid: "q", Gid: "q", Muid: "q",
@ -980,19 +982,19 @@ func (s *Server) readDir(path string, offset uint64, count uint32) []byte {
return data[offset:end]
}
func (s *Server) readLogFile(sessionID string, offset uint64, count uint32) []byte {
sess := s.mgr.Get(sessionID)
if sess == nil {
func (s *Server) readAuditLog(offset uint64, count uint32) []byte {
mailMgr := s.mgr.GetMailManager()
if mailMgr == nil {
return nil
}
tmuxSess, ok := sess.(*tmux.Session)
if !ok {
auditLog := mailMgr.GetAuditLog()
if auditLog == nil {
return nil
}
// Try to read from current offset
logData, hasMore := tmuxSess.GetChatLogFrom(int64(offset))
logData, hasMore := auditLog.ReadFrom(int64(offset))
if hasMore && len(logData) > 0 {
// Return available data
end := min(int(count), len(logData))
@ -1000,16 +1002,16 @@ func (s *Server) readLogFile(sessionID string, offset uint64, count uint32) []by
}
// No data available at this offset, wait for new data
waitCh := tmuxSess.WaitForLogData()
waitCh := auditLog.WaitForData()
<-waitCh
// Check again after new data arrived
logData, hasMore = tmuxSess.GetChatLogFrom(int64(offset))
logData, hasMore = auditLog.ReadFrom(int64(offset))
if hasMore && len(logData) > 0 {
end := min(int(count), len(logData))
return []byte(logData[:end])
}
return nil
}
@ -1030,6 +1032,18 @@ func (s *Server) readFile(path string) string {
return strings.Join(lines, "\n") + "\n"
}
if path == "/audit" {
mailMgr := s.mgr.GetMailManager()
if mailMgr == nil {
return ""
}
auditLog := mailMgr.GetAuditLog()
if auditLog == nil {
return ""
}
return auditLog.Read()
}
if path == "/status" {
var lines []string
mailMgr := s.mgr.GetMailManager()
@ -1120,12 +1134,6 @@ func (s *Server) getSessionFile(sess backend.Session, idx int) string {
// Output from last command - not currently exposed
// Bots write to this via 9P, which appends to chat log
return ""
case fileLog:
// Full chat history (USER:/ASSISTANT: with --- separators)
if tmuxSess, ok := sess.(*tmux.Session); ok {
return tmuxSess.GetChatLog()
}
return ""
case fileState:
return sess.State()
case filePid:

View File

@ -33,6 +33,9 @@ func NewManager(backends map[string]backend.Backend) *Manager {
stopCh: make(chan struct{}),
}
// Set session getter for alias lookup
mailMgr.SetSessionGetter(m)
// Start mail processing loop
m.wg.Add(1)
go m.mailProcessingLoop()
@ -72,6 +75,19 @@ func (m *Manager) Get(id string) backend.Session {
return m.sessions[id]
}
// GetAlias returns the alias for a session ID, or empty string if not set
func (m *Manager) GetAlias(id string) string {
m.mu.RLock()
defer m.mu.RUnlock()
sess := m.sessions[id]
if sess == nil {
return ""
}
return sess.Metadata().Alias
}
// List returns all session IDs
func (m *Manager) List() []string {
m.mu.RLock()

View File

@ -1,28 +1,12 @@
#!/usr/bin/env rc
# Alog - Stream AnviLLM session log to Acme window
# Alog - Stream AnviLLM audit log to Acme window
if(~ $#* 1) {
sess=$1
# Create window and get its ID
winid=`{9p read acme/new/ctl | awk '{print $1}'}
echo name /AnviLLM/Audit | 9p write acme/$winid/ctl
echo clean | 9p write acme/$winid/ctl
# Get alias for display name
alias=`{9p read agent/$sess/alias >[2]/dev/null}
if(~ $alias '') {
name=$sess
}
if not {
name=$alias
}
# Stream audit log to window
9p read agent/audit | 9p write acme/$winid/data
# Create window and get its ID
winid=`{9p read acme/new/ctl | awk '{print $1}'}
echo name /AnviLLM/Log.$name | 9p write acme/$winid/ctl
echo clean | 9p write acme/$winid/ctl
# Stream log to window
9p read agent/$sess/log | 9p write acme/$winid/data
}
if not {
echo 'Usage: Alog <session-id>' >[1=2]
exit 1
}