Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 30 additions & 11 deletions api/cmd/server/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -173,6 +173,7 @@ func main() {
agentRepo := postgres.NewAgentRepository(pool)
mcpRepo := postgres.NewMCPServerRepository(pool)
skillRepo := postgres.NewSkillRepository(pool)
auditEventRepo := postgres.NewAuditEventRepository(pool)
auditRepo := postgres.NewAuditRepository(pool)
llmRepo := postgres.NewLLMModelRepository(pool, dataCipher)
providerRepo := postgres.NewLLMProviderRepository(pool, dataCipher)
Expand Down Expand Up @@ -261,8 +262,10 @@ func main() {
logger.Info("MCP OAuth2 manager initialized", zap.String("callback_url", callbackURL))
}

// Start chat_events cleanup goroutine
if cfg.Metrics.Enabled && cfg.Metrics.RetentionDays > 0 {
// Start cleanup goroutine for chat_events (observability) and audit_events.
// Retention is now read from runtime settings on each tick (admin-editable),
// falling back to the settings defaults; the cleanup interval stays in YAML.
{
cleanupInterval := time.Hour
if cfg.Metrics.CleanupInterval != "" {
if d, err := time.ParseDuration(cfg.Metrics.CleanupInterval); err == nil {
Expand All @@ -273,18 +276,33 @@ func main() {
ticker := time.NewTicker(cleanupInterval)
defer ticker.Stop()
for range ticker.C {
cleanCtx, cleanCancel := context.WithTimeout(context.Background(), 30*time.Second)
deleted, err := chatEventRepo.Cleanup(cleanCtx, cfg.Metrics.RetentionDays)
cleanCancel()
if err != nil {
logger.Warn("chat_events cleanup failed", zap.Error(err))
} else if deleted > 0 {
logger.Info("chat_events cleanup", zap.Int64("deleted", deleted))
// Observability metrics (only if metrics are enabled at all).
if cfg.Metrics.Enabled {
if days := settingsService.Int(settings.KeyObservabilityRetentionDays); days > 0 {
cleanCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
deleted, err := chatEventRepo.Cleanup(cleanCtx, days)
cancel()
if err != nil {
logger.Warn("chat_events cleanup failed", zap.Error(err))
} else if deleted > 0 {
logger.Info("chat_events cleanup", zap.Int64("deleted", deleted))
}
}
}
// Audit events (independent retention).
if days := settingsService.Int(settings.KeyAuditRetentionDays); days > 0 {
cleanCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
deleted, err := auditEventRepo.Cleanup(cleanCtx, days)
cancel()
if err != nil {
logger.Warn("audit_events cleanup failed", zap.Error(err))
} else if deleted > 0 {
logger.Info("audit_events cleanup", zap.Int64("deleted", deleted))
}
}
}
}()
logger.Info("chat_events cleanup started",
zap.Int("retention_days", cfg.Metrics.RetentionDays),
logger.Info("cleanup job started (chat_events + audit_events)",
zap.Duration("interval", cleanupInterval))
}

Expand Down Expand Up @@ -332,6 +350,7 @@ func main() {
AgentRepo: agentRepo,
MCPRepo: mcpRepo,
SkillRepo: skillRepo,
AuditEventRepo: auditEventRepo,
UserRepo: userRepo,
AuditRepo: auditRepo,
LLMRepo: llmRepo,
Expand Down
50 changes: 50 additions & 0 deletions api/internal/audit/event.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,50 @@
package audit

import (
"context"
"time"

"github.com/dfradehubs/agentgram-api/internal/models"
"github.com/dfradehubs/agentgram-api/internal/repository"
"go.uber.org/zap"
)

// auditInsertTimeout bounds the async DB write so a slow DB can't leak goroutines.
const auditInsertTimeout = 5 * time.Second

// RecordEvent truncates the event's prompt/response to maxContentChars and
// inserts it asynchronously, so auditing never blocks the request hot-path.
// A nil repo makes this a no-op (auditing disabled / not wired).
func RecordEvent(repo repository.AuditEventRepository, ev *models.AuditEvent, maxContentChars int, logger *zap.Logger) {
if repo == nil || ev == nil {
return
}
if maxContentChars > 0 {
ev.Prompt = truncateRunes(ev.Prompt, maxContentChars)
ev.Response = truncateRunes(ev.Response, maxContentChars)
for i := range ev.ToolCalls {
ev.ToolCalls[i].Arguments = truncateRunes(ev.ToolCalls[i].Arguments, maxContentChars)
ev.ToolCalls[i].Result = truncateRunes(ev.ToolCalls[i].Result, maxContentChars)
}
}
go func() {
ctx, cancel := context.WithTimeout(context.Background(), auditInsertTimeout)
defer cancel()
if err := repo.Insert(ctx, ev); err != nil && logger != nil {
logger.Warn("failed to insert audit_event", zap.Error(err))
}
}()
}

// truncateRunes cuts s to at most max runes (not bytes) so UTF-8 stays valid,
// appending an ellipsis marker when it actually truncates.
func truncateRunes(s string, max int) string {
if max <= 0 {
return s
}
r := []rune(s)
if len(r) <= max {
return s
}
return string(r[:max]) + "…[truncated]"
}
74 changes: 74 additions & 0 deletions api/internal/audit/event_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,74 @@
package audit

import (
"context"
"strings"
"testing"

"github.com/dfradehubs/agentgram-api/internal/models"
)

func TestTruncateRunes(t *testing.T) {
tests := []struct {
name string
in string
max int
want string
}{
{"under limit", "hello", 10, "hello"},
{"exact limit", "hello", 5, "hello"},
{"over limit", "hello world", 5, "hello…[truncated]"},
{"zero max is no-op", "hello", 0, "hello"},
{"multibyte not split", "áéíóú€", 3, "áéí…[truncated]"},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
if got := truncateRunes(tt.in, tt.max); got != tt.want {
t.Errorf("truncateRunes(%q,%d) = %q, want %q", tt.in, tt.max, got, tt.want)
}
})
}
}

// captureRepo records the last inserted event so we can assert on it.
type captureRepo struct{ last *models.AuditEvent }

func (c *captureRepo) Insert(_ context.Context, e *models.AuditEvent) error { c.last = e; return nil }
func (c *captureRepo) List(context.Context, models.AuditEventFilter) ([]*models.AuditEvent, int, error) {
return nil, 0, nil
}
func (c *captureRepo) Cleanup(context.Context, int) (int64, error) { return 0, nil }

func TestRecordEventTruncatesPromptResponseAndToolCalls(t *testing.T) {
// RecordEvent truncates synchronously (before the async insert), so we can
// assert on the mutated event immediately.
ev := &models.AuditEvent{
Prompt: strings.Repeat("a", 100),
Response: strings.Repeat("b", 100),
ToolCalls: []models.AuditToolCall{
{Name: "t", Arguments: strings.Repeat("c", 100), Result: strings.Repeat("d", 100)},
},
}
audit := &captureRepo{}
RecordEvent(audit, ev, 10, nil)

for _, f := range []struct {
name string
got string
}{
{"prompt", ev.Prompt},
{"response", ev.Response},
{"tool args", ev.ToolCalls[0].Arguments},
{"tool result", ev.ToolCalls[0].Result},
} {
if !strings.HasSuffix(f.got, "…[truncated]") || len([]rune(strings.TrimSuffix(f.got, "…[truncated]"))) != 10 {
t.Errorf("%s not truncated to 10 runes: %q", f.name, f.got)
}
}
}

func TestRecordEventNoopGuards(t *testing.T) {
// Must not panic on nil repo or nil event.
RecordEvent(nil, &models.AuditEvent{Prompt: "x"}, 10, nil)
RecordEvent(&captureRepo{}, nil, 10, nil)
}
70 changes: 70 additions & 0 deletions api/internal/handlers/admin_audit.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,70 @@
package handlers

import (
"encoding/json"
"net/http"
"strconv"
"time"

"github.com/dfradehubs/agentgram-api/internal/models"
"github.com/dfradehubs/agentgram-api/internal/repository"
"go.uber.org/zap"
)

// AdminAuditHandler serves the admin audit log (detailed activity with content).
type AdminAuditHandler struct {
repo repository.AuditEventRepository
logger *zap.Logger
}

// NewAdminAuditHandler creates a new admin audit handler.
func NewAdminAuditHandler(repo repository.AuditEventRepository, logger *zap.Logger) *AdminAuditHandler {
return &AdminAuditHandler{repo: repo, logger: logger}
}

// ListAuditEvents handles GET /api/admin/audit with filters:
// from, to (RFC3339), user, group, resource_type, request_id, limit, offset.
func (h *AdminAuditHandler) ListAuditEvents(w http.ResponseWriter, r *http.Request) {
q := r.URL.Query()
f := models.AuditEventFilter{
UserEmail: q.Get("user"),
Group: q.Get("group"),
ResourceType: q.Get("resource_type"),
SessionID: q.Get("session"),
}
if v := q.Get("from"); v != "" {
if t, err := time.Parse(time.RFC3339, v); err == nil {
f.From = &t
}
}
if v := q.Get("to"); v != "" {
if t, err := time.Parse(time.RFC3339, v); err == nil {
f.To = &t
}
}
f.Limit = 50
if v, err := strconv.Atoi(q.Get("limit")); err == nil && v > 0 {
f.Limit = v
}
if v, err := strconv.Atoi(q.Get("offset")); err == nil && v >= 0 {
f.Offset = v
}

events, total, err := h.repo.List(r.Context(), f)
if err != nil {
h.logger.Error("list audit events failed", zap.Error(err))
http.Error(w, `{"error":"internal server error"}`, http.StatusInternalServerError)
return
}
if events == nil {
events = []*models.AuditEvent{}
}

w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(map[string]interface{}{
"events": events,
"total": total,
"limit": f.Limit,
"offset": f.Offset,
})
}
81 changes: 81 additions & 0 deletions api/internal/handlers/admin_audit_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,81 @@
package handlers

import (
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"testing"

"github.com/dfradehubs/agentgram-api/internal/models"
"go.uber.org/zap"
)

type fakeAuditEventRepo struct {
events []*models.AuditEvent
total int
gotFilter models.AuditEventFilter
}

func (f *fakeAuditEventRepo) Insert(context.Context, *models.AuditEvent) error { return nil }
func (f *fakeAuditEventRepo) List(_ context.Context, filter models.AuditEventFilter) ([]*models.AuditEvent, int, error) {
f.gotFilter = filter
return f.events, f.total, nil
}
func (f *fakeAuditEventRepo) Cleanup(context.Context, int) (int64, error) { return 0, nil }

func TestListAuditEventsParsesFiltersAndResponds(t *testing.T) {
repo := &fakeAuditEventRepo{
events: []*models.AuditEvent{{ID: "1", UserEmail: "a@b.com", ResourceType: "agent", Status: "ok"}},
total: 1,
}
h := NewAdminAuditHandler(repo, zap.NewNop())

req := httptest.NewRequest(http.MethodGet,
"/api/admin/audit?user=a@b.com&resource_type=agent&session=s1&group=g1&limit=25&offset=50&from=2026-01-01T00:00:00Z", nil)
rec := httptest.NewRecorder()
h.ListAuditEvents(rec, req)

if rec.Code != http.StatusOK {
t.Fatalf("status = %d, want 200", rec.Code)
}

var body struct {
Events []*models.AuditEvent `json:"events"`
Total int `json:"total"`
Limit int `json:"limit"`
Offset int `json:"offset"`
}
if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil {
t.Fatalf("decode body: %v", err)
}
if body.Total != 1 || len(body.Events) != 1 || body.Limit != 25 || body.Offset != 50 {
t.Errorf("unexpected body: %+v", body)
}

f := repo.gotFilter
if f.UserEmail != "a@b.com" || f.ResourceType != "agent" || f.SessionID != "s1" || f.Group != "g1" {
t.Errorf("filter not parsed: %+v", f)
}
if f.Limit != 25 || f.Offset != 50 {
t.Errorf("paging not parsed: limit=%d offset=%d", f.Limit, f.Offset)
}
if f.From == nil {
t.Error("from timestamp not parsed")
}
}

func TestListAuditEventsEmptyResult(t *testing.T) {
h := NewAdminAuditHandler(&fakeAuditEventRepo{}, zap.NewNop())
req := httptest.NewRequest(http.MethodGet, "/api/admin/audit", nil)
rec := httptest.NewRecorder()
h.ListAuditEvents(rec, req)

if rec.Code != http.StatusOK {
t.Fatalf("status = %d, want 200", rec.Code)
}
// events must serialize as [] not null
if !json.Valid(rec.Body.Bytes()) || !contains(rec.Body.String(), `"events":[]`) {
t.Errorf("expected empty events array, got: %s", rec.Body.String())
}
}
28 changes: 26 additions & 2 deletions api/internal/handlers/group_chat.go
Original file line number Diff line number Diff line change
Expand Up @@ -314,7 +314,7 @@ func (h *ProxyHandler) GroupChat(w http.ResponseWriter, r *http.Request) {
}

// Observability + audit per turn (same shape as single-agent chat)
h.recordGroupTurnEvent(agent, session.SessionID, userEmail, result, err, int(time.Since(turnStart).Milliseconds()), len(messagesToSend))
h.recordGroupTurnEvent(agent, session.SessionID, userEmail, userGroups, lastMessageContent(messagesToSend), result, err, int(time.Since(turnStart).Milliseconds()), len(messagesToSend))
h.audit.Log(userEmail, audit.ActionChat,
zap.String("agent_id", agentID),
zap.String("group_id", groupID),
Expand Down Expand Up @@ -618,7 +618,7 @@ func (h *ProxyHandler) persistModeratorMessage(ctx context.Context, sessionID, a
}

// recordGroupTurnEvent records one debate turn as a chat event for observability.
func (h *ProxyHandler) recordGroupTurnEvent(agent *models.Agent, sessionID, userEmail string, result *proxy.ProxyResult, err error, durationMs, messageCount int) {
func (h *ProxyHandler) recordGroupTurnEvent(agent *models.Agent, sessionID, userEmail string, userGroups []string, prompt string, result *proxy.ProxyResult, err error, durationMs, messageCount int) {
status := "ok"
var errType, errMsg string
if err != nil {
Expand Down Expand Up @@ -650,6 +650,30 @@ func (h *ProxyHandler) recordGroupTurnEvent(agent *models.Agent, sessionID, user
MessageCount: messageCount,
ToolCalls: toolCallInfos,
}, h.logger)

var auditResp string
var auditTools []models.AuditToolCall
if result != nil {
auditResp = proxy.TranscriptText(result)
auditTools = auditToolCalls(result.ToolCalls)
}
recordAuditEvent(h.auditRepo, h.settings, &models.AuditEvent{
UserEmail: userEmail,
UserGroups: userGroups,
ResourceType: models.AuditResourceAgent,
ResourceID: agent.ID,
ResourceName: agent.Name,
Source: models.AuditSourceWeb,
SessionID: sessionID,
Action: models.AuditActionGroupDebate,
Prompt: prompt,
Response: auditResp,
ToolCalls: auditTools,
Status: status,
ErrorType: errType,
ErrorMsg: errMsg,
DurationMs: durationMs,
}, h.logger)
}

// distinctRespondingSpeakers counts agents that produced a response, even when
Expand Down
Loading
Loading