Adolfo Delorenzo
30b9f60668
- AuditEvent ORM model with tenant_id, action_type, latency_ms, metadata - KnowledgeBaseDocument and KBChunk ORM models for vector KB - Migration 003: audit_events (immutable via REVOKE), kb_documents, kb_chunks with HNSW index and RLS on all tables - AuditLogger with log_llm_call, log_tool_call, log_escalation methods - audit_events immutability enforced at DB level (UPDATE/DELETE rejected) - [Rule 1 - Bug] Fixed CAST(:metadata AS jsonb) for asyncpg compatibility
100 lines
3.1 KiB
Python
100 lines
3.1 KiB
Python
"""
|
|
SQLAlchemy 2.0 ORM model for the immutable audit_events table.
|
|
|
|
Design:
|
|
- Append-only: konstruct_app role has SELECT + INSERT only (enforced via REVOKE in migration)
|
|
- Tenant-scoped via RLS — every query sees only the current tenant's rows
|
|
- action_type discriminates between llm_call, tool_invocation, and escalation events
|
|
|
|
Important: The DB-level REVOKE UPDATE/DELETE on audit_events means that even if
|
|
application code accidentally attempts an UPDATE or DELETE, the database will reject it.
|
|
This is a hard compliance guarantee.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from datetime import datetime
|
|
from typing import Any
|
|
|
|
from sqlalchemy import DateTime, Integer, Text, func
|
|
from sqlalchemy.dialects.postgresql import JSONB, UUID
|
|
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
|
|
|
|
|
|
class AuditBase(DeclarativeBase):
|
|
"""Separate declarative base for audit models to avoid conflicts with tenant Base."""
|
|
|
|
pass
|
|
|
|
|
|
class AuditEvent(AuditBase):
|
|
"""
|
|
Immutable record of every LLM call, tool invocation, and escalation event.
|
|
|
|
RLS is enabled — rows are scoped to the current tenant via app.current_tenant.
|
|
The konstruct_app role has SELECT + INSERT only — UPDATE and DELETE are revoked.
|
|
|
|
action_type values:
|
|
'llm_call' — LLM completion request/response
|
|
'tool_invocation' — Tool execution (success or failure)
|
|
'escalation' — Agent handoff to human or another agent
|
|
"""
|
|
|
|
__tablename__ = "audit_events"
|
|
|
|
id: Mapped[uuid.UUID] = mapped_column(
|
|
UUID(as_uuid=True),
|
|
primary_key=True,
|
|
server_default=func.gen_random_uuid(),
|
|
)
|
|
tenant_id: Mapped[uuid.UUID] = mapped_column(
|
|
UUID(as_uuid=True),
|
|
nullable=False,
|
|
index=True,
|
|
)
|
|
agent_id: Mapped[uuid.UUID | None] = mapped_column(
|
|
UUID(as_uuid=True),
|
|
nullable=True,
|
|
)
|
|
user_id: Mapped[str | None] = mapped_column(
|
|
Text,
|
|
nullable=True,
|
|
comment="Channel-native user identifier",
|
|
)
|
|
action_type: Mapped[str] = mapped_column(
|
|
Text,
|
|
nullable=False,
|
|
comment="llm_call | tool_invocation | escalation",
|
|
)
|
|
input_summary: Mapped[str | None] = mapped_column(
|
|
Text,
|
|
nullable=True,
|
|
comment="Truncated input for audit readability (not full content)",
|
|
)
|
|
output_summary: Mapped[str | None] = mapped_column(
|
|
Text,
|
|
nullable=True,
|
|
comment="Truncated output for audit readability",
|
|
)
|
|
latency_ms: Mapped[int | None] = mapped_column(
|
|
Integer,
|
|
nullable=True,
|
|
comment="Duration of the operation in milliseconds",
|
|
)
|
|
metadata: Mapped[dict[str, Any]] = mapped_column(
|
|
JSONB,
|
|
nullable=False,
|
|
server_default="{}",
|
|
default=dict,
|
|
comment="Additional structured context (model name, tool args hash, etc.)",
|
|
)
|
|
created_at: Mapped[datetime] = mapped_column(
|
|
DateTime(timezone=True),
|
|
nullable=False,
|
|
server_default=func.now(),
|
|
)
|
|
|
|
def __repr__(self) -> str:
|
|
return f"<AuditEvent id={self.id} action={self.action_type} tenant={self.tenant_id}>"
|