diff --git a/cmd/anvillm/main.go b/cmd/anvillm/main.go index 9da6e93..5c489d7 100644 --- a/cmd/anvillm/main.go +++ b/cmd/anvillm/main.go @@ -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() diff --git a/cmd/anvilweb/main.go b/cmd/anvilweb/main.go index 8cef9fe..12b4d68 100644 --- a/cmd/anvilweb/main.go +++ b/cmd/anvilweb/main.go @@ -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 diff --git a/cmd/anvilweb/static/index.html b/cmd/anvilweb/static/index.html index 0d9216f..100e4a0 100644 --- a/cmd/anvilweb/static/index.html +++ b/cmd/anvilweb/static/index.html @@ -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'); diff --git a/elisp/anvillm.el b/elisp/anvillm.el index 8f2519c..d865494 100644 --- a/elisp/anvillm.el +++ b/elisp/anvillm.el @@ -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." diff --git a/internal/audit/audit.go b/internal/audit/audit.go new file mode 100644 index 0000000..ef043b0 --- /dev/null +++ b/internal/audit/audit.go @@ -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 +} diff --git a/internal/mailbox/manager.go b/internal/mailbox/manager.go index dbba57d..d8bd4dd 100644 --- a/internal/mailbox/manager.go +++ b/internal/mailbox/manager.go @@ -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 +} diff --git a/internal/p9/server.go b/internal/p9/server.go index 5440baf..09782e1 100644 --- a/internal/p9/server.go +++ b/internal/p9/server.go @@ -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: diff --git a/internal/session/manager.go b/internal/session/manager.go index 619ac11..932d189 100644 --- a/internal/session/manager.go +++ b/internal/session/manager.go @@ -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() diff --git a/scripts/Alog b/scripts/Alog index 538da07..929defd 100755 --- a/scripts/Alog +++ b/scripts/Alog @@ -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 ' >[1=2] - exit 1 -}