From 21d8a0ccd6b7dc750d099f626b722f525742f564 Mon Sep 17 00:00:00 2001 From: devshift-stack Date: Wed, 31 Dec 2025 23:17:47 +0100 Subject: [PATCH] feat: Add persistent event logging (audit_events) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - src/db/audit-events.ts: Event table + AuditService - src/lib/events.ts: Global recordEvent() helper - Task lifecycle: created, started, finished, failed - Chat events: message sent - Auth events: login, logout - Events persisted to SQLite, survive restarts 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 --- src/api/auth.ts | 23 ++++ src/api/tasks.ts | 40 ++++++ src/chat/manager.ts | 4 + src/db/audit-events.ts | 285 +++++++++++++++++++++++++++++++++++++++++ src/index.ts | 5 + src/lib/events.ts | 104 +++++++++++++++ 6 files changed, 461 insertions(+) create mode 100644 src/db/audit-events.ts create mode 100644 src/lib/events.ts diff --git a/src/api/auth.ts b/src/api/auth.ts index 2769277..7def083 100644 --- a/src/api/auth.ts +++ b/src/api/auth.ts @@ -60,6 +60,15 @@ export function createAuthRouter(): Router { email: user.email, }); + // Log login event + db.audit.log({ + kind: "user_login", + message: `User ${user.email} logged in`, + userId: user.id, + severity: "info", + meta: { email: user.email, role: user.role }, + }); + // Return tokens res.json({ success: true, @@ -98,6 +107,9 @@ export function createAuthRouter(): Router { }); } + // Get user from token before revoking + const payload = verifyAccessToken(accessToken); + // Get refresh token from body const { refreshToken } = req.body; @@ -109,6 +121,17 @@ export function createAuthRouter(): Router { revokeToken(refreshToken); } + // Log logout event + if (payload) { + db.audit.log({ + kind: "user_logout", + message: `User ${payload.email} logged out`, + userId: payload.userId, + severity: "info", + meta: { email: payload.email }, + }); + } + res.json({ success: true, message: "Logged out successfully", diff --git a/src/api/tasks.ts b/src/api/tasks.ts index 01e8cc1..0c9bb3d 100644 --- a/src/api/tasks.ts +++ b/src/api/tasks.ts @@ -61,6 +61,16 @@ export function createTaskRouter(db: Database, queue: QueueAdapter, gate: Enforc // Task is BLOCKED - do NOT queue, require human approval db.updateTask(task.id, { status: "stopped", stop_score: gateDecision.stopScore }); + // Log blocked task event + db.audit.log({ + kind: "task_failed", + message: `Task "${task.title}" blocked by enforcement gate`, + taskId: task.id, + userId: (req as any).userId, + severity: "warn", + meta: { stopScore: gateDecision.stopScore, reasons: gateDecision.reasons }, + }); + return res.status(202).json({ id: task.id, status: "BLOCKED", @@ -74,6 +84,16 @@ export function createTaskRouter(db: Database, queue: QueueAdapter, gate: Enforc // Task passed gate - queue for processing await queue.add("process_task", { taskId: task.id }); + // Log task created event + db.audit.log({ + kind: "task_created", + message: `Task "${task.title}" created`, + taskId: task.id, + userId: (req as any).userId, + severity: "info", + meta: { priority: task.priority, assignee: task.assignee }, + }); + res.status(201).json({ id: task.id, status: "pending", @@ -111,6 +131,16 @@ export function createTaskRouter(db: Database, queue: QueueAdapter, gate: Enforc // Work is BLOCKED - cannot complete task db.updateTask(task.id, { status: "stopped", stop_score: gateDecision.stopScore }); + // Log blocked submission event + db.audit.log({ + kind: "task_failed", + message: `Task "${task.title}" submission blocked`, + taskId: task.id, + userId: (req as any).userId, + severity: "warn", + meta: { stopScore: gateDecision.stopScore, reasons: gateDecision.reasons }, + }); + return res.status(202).json({ id: task.id, status: "BLOCKED", @@ -124,6 +154,16 @@ export function createTaskRouter(db: Database, queue: QueueAdapter, gate: Enforc // Work passed gate - mark as completed db.updateTask(task.id, { status: "completed", stop_score: gateDecision.stopScore }); + // Log task completed event + db.audit.log({ + kind: "task_finished", + message: `Task "${task.title}" completed`, + taskId: task.id, + userId: (req as any).userId, + severity: "info", + meta: { stopScore: gateDecision.stopScore }, + }); + res.json({ id: task.id, status: "completed", diff --git a/src/chat/manager.ts b/src/chat/manager.ts index fd59e7f..d7f1872 100644 --- a/src/chat/manager.ts +++ b/src/chat/manager.ts @@ -14,6 +14,7 @@ import { costTracker } from "../billing/costTracker.ts"; import { selectModel } from "../billing/modelSelector.ts"; import { agentTools, executeTool } from "./tools.ts"; import { trackAICall, log, metrics } from "../monitoring/sentry.js"; +import { events } from "../lib/events.js"; import Anthropic from "@anthropic-ai/sdk"; import OpenAI from "openai"; import { GoogleGenerativeAI } from "@google/generative-ai"; @@ -163,6 +164,9 @@ export class ChatManager { } } + // Log chat event for audit trail + events.chatSent(request.userId, request.agentName, request.message); + return { chatId: chat.id, messageId: assistantMessage.id, diff --git a/src/db/audit-events.ts b/src/db/audit-events.ts new file mode 100644 index 0000000..d87c8a8 --- /dev/null +++ b/src/db/audit-events.ts @@ -0,0 +1,285 @@ +/** + * Audit Events - Centralized event logging + * Single source of truth for all system events + */ + +import type { Database as BetterSqlite3Database } from "better-sqlite3"; +import { randomUUID } from "crypto"; + +// Event kinds +export type AuditEventKind = + | "agent_heartbeat" + | "task_created" + | "task_started" + | "task_finished" + | "task_failed" + | "chat_sent" + | "deploy" + | "error" + | "brain_ingest" + | "brain_search" + | "user_login" + | "user_logout" + | "api_call"; + +export type AuditSeverity = "info" | "warn" | "error"; + +export interface AuditEvent { + id: string; + ts: string; + kind: AuditEventKind; + agentId: string | null; + taskId: string | null; + userId: string | null; + severity: AuditSeverity; + message: string; + meta: Record | null; +} + +export interface CreateAuditEventInput { + kind: AuditEventKind; + message: string; + agentId?: string; + taskId?: string; + userId?: string; + severity?: AuditSeverity; + meta?: Record; +} + +export interface ListAuditEventsOptions { + kind?: AuditEventKind; + severity?: AuditSeverity; + userId?: string; + agentId?: string; + taskId?: string; + limit?: number; + offset?: number; + since?: string; // ISO date string +} + +export interface AuditEventStats { + total: number; + byKind: Record; + bySeverity: Record; + lastEvent: string | null; +} + +/** + * Initialize audit_events table + */ +export function initAuditEventsTable(db: BetterSqlite3Database): void { + db.exec(` + CREATE TABLE IF NOT EXISTS audit_events ( + id TEXT PRIMARY KEY, + ts TEXT NOT NULL, + kind TEXT NOT NULL, + agent_id TEXT, + task_id TEXT, + user_id TEXT, + severity TEXT NOT NULL DEFAULT 'info', + message TEXT NOT NULL, + meta TEXT, + + -- Indexes for common queries + created_at TEXT GENERATED ALWAYS AS (ts) STORED + ); + + CREATE INDEX IF NOT EXISTS idx_audit_events_ts ON audit_events(ts DESC); + CREATE INDEX IF NOT EXISTS idx_audit_events_kind ON audit_events(kind); + CREATE INDEX IF NOT EXISTS idx_audit_events_severity ON audit_events(severity); + CREATE INDEX IF NOT EXISTS idx_audit_events_user_id ON audit_events(user_id); + CREATE INDEX IF NOT EXISTS idx_audit_events_agent_id ON audit_events(agent_id); + CREATE INDEX IF NOT EXISTS idx_audit_events_task_id ON audit_events(task_id); + `); +} + +/** + * Audit Events Service + */ +export function createAuditService(db: BetterSqlite3Database) { + return { + /** + * Log a new audit event + */ + log(input: CreateAuditEventInput): AuditEvent { + const id = randomUUID(); + const ts = new Date().toISOString(); + const severity = input.severity ?? "info"; + const meta = input.meta ? JSON.stringify(input.meta) : null; + + const stmt = db.prepare(` + INSERT INTO audit_events (id, ts, kind, agent_id, task_id, user_id, severity, message, meta) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + `); + + stmt.run( + id, + ts, + input.kind, + input.agentId ?? null, + input.taskId ?? null, + input.userId ?? null, + severity, + input.message, + meta + ); + + return { + id, + ts, + kind: input.kind, + agentId: input.agentId ?? null, + taskId: input.taskId ?? null, + userId: input.userId ?? null, + severity, + message: input.message, + meta: input.meta ?? null, + }; + }, + + /** + * Get single event by ID + */ + get(id: string): AuditEvent | undefined { + const stmt = db.prepare("SELECT * FROM audit_events WHERE id = ?"); + const row = stmt.get(id) as { + id: string; + ts: string; + kind: string; + agent_id: string | null; + task_id: string | null; + user_id: string | null; + severity: string; + message: string; + meta: string | null; + } | undefined; + + if (!row) return undefined; + + return { + id: row.id, + ts: row.ts, + kind: row.kind as AuditEventKind, + agentId: row.agent_id, + taskId: row.task_id, + userId: row.user_id, + severity: row.severity as AuditSeverity, + message: row.message, + meta: row.meta ? JSON.parse(row.meta) : null, + }; + }, + + /** + * List events with filters + */ + list(options: ListAuditEventsOptions = {}): AuditEvent[] { + const limit = options.limit ?? 100; + const offset = options.offset ?? 0; + const conditions: string[] = []; + const params: unknown[] = []; + + if (options.kind) { + conditions.push("kind = ?"); + params.push(options.kind); + } + if (options.severity) { + conditions.push("severity = ?"); + params.push(options.severity); + } + if (options.userId) { + conditions.push("user_id = ?"); + params.push(options.userId); + } + if (options.agentId) { + conditions.push("agent_id = ?"); + params.push(options.agentId); + } + if (options.taskId) { + conditions.push("task_id = ?"); + params.push(options.taskId); + } + if (options.since) { + conditions.push("ts >= ?"); + params.push(options.since); + } + + const whereClause = conditions.length > 0 ? `WHERE ${conditions.join(" AND ")}` : ""; + const query = `SELECT * FROM audit_events ${whereClause} ORDER BY ts DESC LIMIT ? OFFSET ?`; + params.push(limit, offset); + + const stmt = db.prepare(query); + const rows = stmt.all(...params) as Array<{ + id: string; + ts: string; + kind: string; + agent_id: string | null; + task_id: string | null; + user_id: string | null; + severity: string; + message: string; + meta: string | null; + }>; + + return rows.map((row) => ({ + id: row.id, + ts: row.ts, + kind: row.kind as AuditEventKind, + agentId: row.agent_id, + taskId: row.task_id, + userId: row.user_id, + severity: row.severity as AuditSeverity, + message: row.message, + meta: row.meta ? JSON.parse(row.meta) : null, + })); + }, + + /** + * Get event statistics + */ + stats(): AuditEventStats { + const totalStmt = db.prepare("SELECT COUNT(*) as count FROM audit_events"); + const total = (totalStmt.get() as { count: number }).count; + + const kindStmt = db.prepare("SELECT kind, COUNT(*) as count FROM audit_events GROUP BY kind"); + const kindRows = kindStmt.all() as Array<{ kind: string; count: number }>; + const byKind: Record = {}; + for (const row of kindRows) { + byKind[row.kind] = row.count; + } + + const sevStmt = db.prepare("SELECT severity, COUNT(*) as count FROM audit_events GROUP BY severity"); + const sevRows = sevStmt.all() as Array<{ severity: string; count: number }>; + const bySeverity: Record = {}; + for (const row of sevRows) { + bySeverity[row.severity] = row.count; + } + + const lastStmt = db.prepare("SELECT ts FROM audit_events ORDER BY ts DESC LIMIT 1"); + const lastRow = lastStmt.get() as { ts: string } | undefined; + + return { + total, + byKind, + bySeverity, + lastEvent: lastRow?.ts ?? null, + }; + }, + + /** + * Cleanup old events (retention policy) + * @param daysToKeep Number of days to retain (default: 30) + * @returns Number of deleted events + */ + cleanup(daysToKeep = 30): number { + const cutoffDate = new Date(); + cutoffDate.setDate(cutoffDate.getDate() - daysToKeep); + const cutoffIso = cutoffDate.toISOString(); + + const stmt = db.prepare("DELETE FROM audit_events WHERE ts < ?"); + const result = stmt.run(cutoffIso); + return result.changes; + }, + }; +} + +export type AuditService = ReturnType; diff --git a/src/index.ts b/src/index.ts index da6b168..53230f4 100644 --- a/src/index.ts +++ b/src/index.ts @@ -51,6 +51,7 @@ import { createEnforcementGate } from "./audit/enforcementGate.js"; import { setupSwagger } from "./swagger/index.js"; import { registerAllWebhookWorkers } from "./queue/workers/index.js"; import { initEmailTransporter } from "./email/mailer.js"; +import { initEventRecorder, events } from "./lib/events.js"; const __filename = fileURLToPath(import.meta.url); const __dirname = dirname(__filename); @@ -66,6 +67,10 @@ async function main() { console.log("✅ Database initialized"); log.info("Database initialized"); + // Initialize event recorder for persistent event logging + initEventRecorder(db.audit); + events.deploy("Server starting", { version: "0.1.0" }); + // Seed default admin if users table is empty await seedDefaultAdmin(db.getRawDb()); log.info("Admin seed check completed"); diff --git a/src/lib/events.ts b/src/lib/events.ts new file mode 100644 index 0000000..3a91631 --- /dev/null +++ b/src/lib/events.ts @@ -0,0 +1,104 @@ +/** + * Event Recording - Global helper for audit event logging + * Single source of truth for all system events + */ + +import type { AuditService, AuditEventKind, AuditSeverity } from "../db/audit-events.js"; + +let auditService: AuditService | null = null; + +/** + * Initialize the event recorder with the audit service + * Call this once at startup after database is initialized + */ +export function initEventRecorder(service: AuditService): void { + auditService = service; + console.log("✅ Event recorder initialized"); +} + +/** + * Record an event to the audit log + * Falls back to console.log if audit service not initialized + */ +export function recordEvent( + kind: AuditEventKind, + message: string, + options?: { + agentId?: string; + taskId?: string; + userId?: string; + severity?: AuditSeverity; + meta?: Record; + } +): void { + const severity = options?.severity ?? "info"; + const prefix = severity === "error" ? "❌" : severity === "warn" ? "⚠️" : "📝"; + + // Always log to console for visibility + console.log(`${prefix} [${kind}] ${message}`, options?.meta ? JSON.stringify(options.meta) : ""); + + // Persist to database if available + if (auditService) { + try { + auditService.log({ + kind, + message, + agentId: options?.agentId, + taskId: options?.taskId, + userId: options?.userId, + severity, + meta: options?.meta, + }); + } catch (err) { + console.error("Failed to record event:", err); + } + } +} + +// Convenience wrappers for common events + +export const events = { + // Task lifecycle + taskCreated: (taskId: string, title: string, userId?: string) => + recordEvent("task_created", `Task created: ${title}`, { taskId, userId }), + + taskStarted: (taskId: string, agentId?: string) => + recordEvent("task_started", `Task started`, { taskId, agentId }), + + taskFinished: (taskId: string, agentId?: string) => + recordEvent("task_finished", `Task completed`, { taskId, agentId }), + + taskFailed: (taskId: string, error: string, agentId?: string) => + recordEvent("task_failed", `Task failed: ${error}`, { taskId, agentId, severity: "error" }), + + // Chat + chatSent: (userId: string, agentId: string, messagePreview: string) => + recordEvent("chat_sent", `Message sent to ${agentId}`, { + userId, + agentId, + meta: { preview: messagePreview.slice(0, 100) }, + }), + + // Auth + userLogin: (userId: string, email: string) => + recordEvent("user_login", `User logged in: ${email}`, { userId }), + + userLogout: (userId: string) => + recordEvent("user_logout", `User logged out`, { userId }), + + // Agent + agentHeartbeat: (agentId: string, status: string) => + recordEvent("agent_heartbeat", `Agent ${agentId}: ${status}`, { agentId }), + + // Errors + error: (message: string, meta?: Record) => + recordEvent("error", message, { severity: "error", meta }), + + // Deploy + deploy: (message: string, meta?: Record) => + recordEvent("deploy", message, { meta }), + + // API calls (for monitoring) + apiCall: (endpoint: string, method: string, userId?: string, meta?: Record) => + recordEvent("api_call", `${method} ${endpoint}`, { userId, meta }), +};