Blueprint — Audit Trail

Status: planned · Slug: audit · Kind: Cross-cutting · Module #16 Foundation module. Other blueprints (Statutory, Compliance, Vitta, Asset Management, Members Suite, Staff Details) write to it. Designed and migrated FIRST.

1. Module summary

Audit Trail is the immutable, append-only, two-key-aware event log of Aayojana. It is not the same as AuditMixin (which only carries created/updated/createdBy/updatedBy on a row). It captures event-level history — every change to legal-weight entities, with before/after blobs, the actor, an optional second signer, and a free-text reason. Entities flagged "two-key" do not commit directly: writes land in pending_writes and a second authorised user must co-sign before they hit the live row. The module also publishes a single service-layer hook record_audit_event(...) that every other module calls, plus a SQLAlchemy event-listener bridge that auto-logs ordinary CRUD on covered tables. A small admin dashboard surfaces the pending queue, the signed-off log, and a suspicious-activity heatmap.

2. Data model

All tables live in aayojana.models.audit. Tenant-scoped (carry tenant_id). None of them are mutable in the ordinary sense — audit_events and audit_signoff_log are append-only at the service layer (DB has no triggers preventing UPDATE/DELETE because Postgres-on-Neon is shared; service layer enforces it instead, and a nightly job checksums new rows for tamper-detection).

# src/aayojana/models/audit.py
from datetime import datetime
from sqlalchemy import (
    JSON, Boolean, DateTime, ForeignKey, Integer, String, Text,
    UniqueConstraint, Index, func,
)
from sqlalchemy.orm import Mapped, mapped_column

from aayojana.models.base import AuditMixin, Base, TenantMixin, uuid4_str


class AuditEvent(Base, TenantMixin, AuditMixin):
    """Append-only event log. NEVER UPDATE, NEVER DELETE in service layer.

    One row per state-changing action on a covered (table, row) pair.
    `before_blob` and `after_blob` carry the column subset that changed
    (full snapshot for INSERT, full snapshot for DELETE, diff-only for UPDATE).
    """

    __tablename__ = "audit_events"
    __table_args__ = (
        Index("ix_audit_events_entity", "entity_type", "entity_id"),
        Index("ix_audit_events_actor", "actor_user_id"),
        Index("ix_audit_events_tenant_created", "tenant_id", "created"),
    )

    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=uuid4_str)

    # Provenance
    provider: Mapped[str] = mapped_column(String(32), nullable=False)
    # 'aayojana.statutory' | 'aayojana.compliance' | 'aayojana.vitta' | 'aayojana.assets' | ...
    entity_type: Mapped[str] = mapped_column(String(64), nullable=False)
    # SQL table name OR a logical name like 'jewel_vault'
    entity_id: Mapped[str] = mapped_column(String(64), nullable=False)
    # The PK of the row (stringified; supports int/uuid)

    # Action
    action: Mapped[str] = mapped_column(String(32), nullable=False)
    # 'create' | 'update' | 'delete' | 'view-pii' | 'period-lock-override' |
    # 'fcra-transfer' | 'cosign' | 'reject' | 'export'

    # State capture
    before_blob: Mapped[dict | None] = mapped_column(JSON, nullable=True)
    after_blob: Mapped[dict | None] = mapped_column(JSON, nullable=True)
    diff_keys: Mapped[list | None] = mapped_column(JSON, nullable=True)
    # array of column names that changed — fast filtering without parsing blobs

    # Actors
    actor_user_id: Mapped[int | None] = mapped_column(
        Integer, ForeignKey("users.id"), nullable=True, index=True
    )
    second_signer_user_id: Mapped[int | None] = mapped_column(
        Integer, ForeignKey("users.id"), nullable=True
    )

    # Reason / context
    reason: Mapped[str | None] = mapped_column(Text, nullable=True)
    request_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
    # request-id from middleware so an event can be traced to an HTTP call
    ip_address: Mapped[str | None] = mapped_column(String(64), nullable=True)
    user_agent: Mapped[str | None] = mapped_column(String(255), nullable=True)

    # Linkage to two-key flow (null for direct writes)
    pending_write_id: Mapped[str | None] = mapped_column(
        String(36), ForeignKey("pending_writes.id"), nullable=True
    )

    # Tamper-evident checksum (nightly batch fills this)
    row_hash: Mapped[str | None] = mapped_column(String(64), nullable=True)


class PendingWrite(Base, TenantMixin, AuditMixin):
    """Two-key staging area. A change submitted by actor lives here until
    a second authorised user co-signs (or rejects). The ORM *never* writes
    the live row directly — only this row is created. On co-sign the
    service layer applies `changes` to the target table and writes an
    AuditEvent linking back via `pending_write_id`.
    """

    __tablename__ = "pending_writes"
    __table_args__ = (
        Index("ix_pending_writes_entity", "entity_type", "entity_id"),
        Index("ix_pending_writes_status_tenant", "status", "tenant_id"),
    )

    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=uuid4_str)

    # Target
    entity_type: Mapped[str] = mapped_column(String(64), nullable=False)
    entity_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
    # nullable for create-actions where ID is allocated on commit
    action: Mapped[str] = mapped_column(String(32), nullable=False)
    # 'create' | 'update' | 'delete'

    # Proposed state
    changes: Mapped[dict] = mapped_column(JSON, nullable=False)
    # full proposed-row for create, {col: new_value} diff for update,
    # null/empty for delete
    before_snapshot: Mapped[dict | None] = mapped_column(JSON, nullable=True)
    # snapshot of the row at submission time (for race-condition detection)

    # Submission
    submitted_by_user_id: Mapped[int] = mapped_column(
        Integer, ForeignKey("users.id"), nullable=False, index=True
    )
    submitted_reason: Mapped[str | None] = mapped_column(Text, nullable=True)
    submitted_at: Mapped[datetime] = mapped_column(
        DateTime, nullable=False, server_default=func.now()
    )

    # Lifecycle
    status: Mapped[str] = mapped_column(
        String(20), nullable=False, server_default="pending"
    )
    # 'pending' | 'cosigned' | 'rejected' | 'expired' | 'superseded'
    expires_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
    # default 7 days from submission; configurable per coverage entry

    # Co-sign / reject
    cosigned_by_user_id: Mapped[int | None] = mapped_column(
        Integer, ForeignKey("users.id"), nullable=True
    )
    cosigned_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
    cosign_note: Mapped[str | None] = mapped_column(Text, nullable=True)
    rejected_reason: Mapped[str | None] = mapped_column(Text, nullable=True)


class AuditCoverage(Base, AuditMixin):
    """Config table — declares which (entity_type, action, [column]) tuples
    require two-key, and which simply emit a passive AuditEvent. Loaded
    into an in-memory cache at app startup; reload-on-write so admin
    edits take effect on next request.

    Tenant-nullable: a row with tenant_id=NULL is the *global default*;
    a tenant-specific row overrides it.
    """

    __tablename__ = "audit_coverage"
    __table_args__ = (
        UniqueConstraint(
            "tenant_id", "entity_type", "action", "column_name",
            name="uq_audit_coverage_scope",
        ),
    )

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    tenant_id: Mapped[str | None] = mapped_column(
        String(36), ForeignKey("organisations.id"), nullable=True, index=True
    )
    entity_type: Mapped[str] = mapped_column(String(64), nullable=False)
    action: Mapped[str] = mapped_column(String(32), nullable=False)
    # 'create' | 'update' | 'delete' | '*' (all)
    column_name: Mapped[str | None] = mapped_column(String(64), nullable=True)
    # if set, two-key only when *this* column is in the diff;
    # null = any column triggers it
    requires_two_key: Mapped[bool] = mapped_column(
        Boolean, nullable=False, server_default="false"
    )
    requires_reason: Mapped[bool] = mapped_column(
        Boolean, nullable=False, server_default="false"
    )
    cosign_window_hours: Mapped[int] = mapped_column(
        Integer, nullable=False, server_default="168"
    )  # 7 days
    eligible_cosigner_roles: Mapped[list | None] = mapped_column(JSON, nullable=True)
    # ['tenant-admin', 'module-admin'] etc. null = any tenant-admin
    notes: Mapped[str | None] = mapped_column(Text, nullable=True)


class AuditSignoffLog(Base, TenantMixin, AuditMixin):
    """Compact, fast-query log of every co-sign / reject decision —
    kept separate from `audit_events` so dashboards can filter purely
    on the human-decision stream without scanning the whole event log.
    Every row here also has a corresponding row in `audit_events`.
    """

    __tablename__ = "audit_signoff_log"
    __table_args__ = (
        Index("ix_audit_signoff_pending", "pending_write_id"),
        Index("ix_audit_signoff_signer", "signer_user_id"),
    )

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    pending_write_id: Mapped[str] = mapped_column(
        String(36), ForeignKey("pending_writes.id"), nullable=False
    )
    audit_event_id: Mapped[str] = mapped_column(
        String(36), ForeignKey("audit_events.id"), nullable=False
    )
    signer_user_id: Mapped[int] = mapped_column(
        Integer, ForeignKey("users.id"), nullable=False
    )
    decision: Mapped[str] = mapped_column(String(16), nullable=False)
    # 'cosign' | 'reject'
    decided_at: Mapped[datetime] = mapped_column(
        DateTime, nullable=False, server_default=func.now()
    )
    note: Mapped[str | None] = mapped_column(Text, nullable=True)

3. Reuse map

Concept	Use existing	Do NOT redefine
Tenant scope	TenantMixin + organisations.id	new tenant table
Created/updated stamp on rows	AuditMixin from models/base.py	duplicate timestamp logic
Actor identity	users.id (int)	new actor table
Membership / role check	memberships.role (tenant-admin, module-admin, ...)	new RBAC table
ID format	uuid4_str() for audit_events.id, pending_writes.id (varchar(36)) — matches organisations.id, branches.id	int autoincrement for these
JSON blobs	SQLAlchemy JSON (works on Postgres jsonb)	Text + manual json.dumps

4. API surface

All under /api/audit/. Multi-tenant scoped via JWT. Roles required noted in Roles.

Method	Path	Purpose	Roles
GET	/api/audit/events	List audit events (filter: entity_type, entity_id, actor, since, action)	tenant-admin, module-admin
GET	/api/audit/events/{id}	Single event with full before/after blobs	tenant-admin, module-admin
GET	/api/audit/events/by-entity/{entity_type}/{entity_id}	History trail for a specific row	tenant-admin, module-admin, owning-module-admin
POST	/api/audit/pending	Submit a two-key change (called by other modules' service layer; rarely by humans)	tenant-admin, module-admin
GET	/api/audit/pending	List pending writes (filter: entity_type, status, submitter)	tenant-admin, module-admin
GET	/api/audit/pending/{id}	Single pending write (with diff preview)	tenant-admin, module-admin
POST	/api/audit/pending/{id}/cosign	Co-sign and apply	tenant-admin OR coverage-eligible
POST	/api/audit/pending/{id}/reject	Reject with reason	tenant-admin OR coverage-eligible
GET	/api/audit/signoffs	Signed-off decision log	tenant-admin, module-admin
GET	/api/audit/coverage	List coverage rules	tenant-admin
POST	/api/audit/coverage	Add/edit a coverage rule (tenant-scoped or global)	super-admin only for global
DELETE	/api/audit/coverage/{id}	Remove a tenant-scope override	tenant-admin
GET	/api/audit/dashboard	Dashboard JSON: pending count, today's events, suspicious flags	tenant-admin
GET	/audit/ (HTML)	Audit dashboard page	tenant-admin
GET	/audit/pending (HTML)	Pending queue UI	tenant-admin
GET	/audit/log (HTML)	Signed-off log UI	tenant-admin

Cosign self-veto rule: the user identified as submitted_by_user_id cannot also be cosigned_by_user_id. Enforced at service layer with HTTP 403 — not at DB level (Postgres CHECK constraints across rows are awkward).

5. Service layer

src/aayojana/services/audit.py — single import point for the rest of the codebase.

def record_audit_event(
    *,
    tenant_id: str | None,
    provider: str,             # 'aayojana.statutory' etc.
    entity_type: str,
    entity_id: str,
    action: str,                # 'create' | 'update' | 'delete' | 'view-pii' | ...
    before_blob: dict | None,
    after_blob: dict | None,
    actor_user_id: int | None,
    reason: str | None = None,
    request: Request | None = None,    # extracts request_id, IP, UA
    second_signer_user_id: int | None = None,
    pending_write_id: str | None = None,
    db: Session,
) -> AuditEvent:
    """Single hook every module calls. Idempotent on (provider, entity_type,
    entity_id, action, request_id) when request_id present."""

def is_two_key_required(
    tenant_id: str | None,
    entity_type: str,
    action: str,
    changed_columns: list[str] | None = None,
    db: Session,
) -> bool:
    """Looks at AuditCoverage cache. Tenant-specific rule beats global."""

def submit_two_key_change(
    *,
    tenant_id: str,
    entity_type: str,
    entity_id: str | None,           # null for create
    action: str,
    changes: dict,
    actor_user_id: int,
    reason: str,
    before_snapshot: dict | None,
    db: Session,
) -> PendingWrite:
    """Stages a change. Also writes a 'submit' AuditEvent. Raises if
    coverage rule does not require two-key (caller should have checked)."""

def cosign_pending_change(
    *,
    pending_id: str,
    second_signer_user_id: int,
    note: str | None = None,
    db: Session,
) -> AuditEvent:
    """Validates: (a) status==pending, (b) signer != submitter,
    (c) signer role is in eligible_cosigner_roles, (d) not expired,
    (e) before_snapshot still matches live row (race detection).
    Then applies changes to target table, writes the final
    AuditEvent (action='update'/'create'/'delete'), updates
    PendingWrite.status='cosigned', appends AuditSignoffLog row."""

def reject_pending_change(
    *,
    pending_id: str,
    rejecter_user_id: int,
    reason: str,
    db: Session,
) -> None:
    """Marks status='rejected', writes AuditEvent (action='reject'),
    appends AuditSignoffLog row."""

def list_pending_for_user(user_id: int, tenant_id: str, db: Session) -> list[PendingWrite]:
    """Returns pending writes the user is *eligible* to co-sign
    (excludes their own submissions)."""

def expire_stale_pending(now: datetime, db: Session) -> int:
    """Cron job — flips expired pending writes to status='expired'."""

def detect_suspicious_activity(
    tenant_id: str, lookback_hours: int = 24, db: Session
) -> list[dict]:
    """Heuristics: same actor > N two-key submissions in window;
    after-hours edits to legal-weight tables; bulk delete; PII reads
    spike. Used by dashboard."""

SQLAlchemy event-listener bridge

src/aayojana/services/audit_bridge.py — wires record_audit_event into ORM flushes so individual modules need not call it explicitly for routine CRUD.

@event.listens_for(Session, "before_flush")
def capture_pre_state(session, flush_context, instances):
    # for each dirty/deleted instance whose mapped class is in the
    # COVERED_MODELS set, snapshot original column values
    ...

@event.listens_for(Session, "after_flush_postexec")
def emit_events(session, flush_context):
    # for each captured instance, build before/after diff and call
    # record_audit_event(); for two-key models it should already
    # have been routed through submit_two_key_change — bridge raises
    # IntegrityError if a covered-two-key model is being directly
    # flushed without a pending_write_id
    ...

The bridge is opt-in per model via a class-level marker:

class TrustRegistration(Base, TenantMixin, AuditMixin):
    __audit_log__ = True               # passive log on every change
    __audit_two_key__ = ("update", "delete")  # two-key for update/delete
    ...

6. UI / Templates

Templates under src/aayojana/templates/audit/. Inherits the Aayojana admin shell.

Page	Route	Highlights
Dashboard	/audit/	Tile: pending count (with overdue badge); today's event count by module; suspicious-activity flags (yellow rows); top 5 actors by event count last 7 days
Pending queue	/audit/pending	Cards: each pending write with diff render (added/changed/removed columns side-by-side), submitter, age, expiry countdown, Cosign + Reject buttons (disabled if you are the submitter)
Pending detail	/audit/pending/{id}	Full before/after JSON, reason, eligible-cosigners list, race-detection warning if before_snapshot no longer matches live
Signed-off log	/audit/log	Table: time, action, entity, actor, second signer, reason — filterable by date range, actor, entity_type
Event log	/audit/events	Same table-style as signed-off log but covers all events (auto-logged + cosigned)
Entity history	/audit/events/by-entity/{type}/{id}	Single-row biography — embedded into Statutory/Compliance/Vitta detail pages as a sidebar
Coverage admin	/audit/coverage	List of (entity_type, action, column, requires_two_key, eligible_roles) rules; add/edit/remove
Suspicious activity	/audit/suspicious	Heuristic-flagged rows with explanation strings

Pending diff renderer

Renders before_blob and after_blob as a two-column table, highlighting diff_keys rows in amber. JSON columns are pretty-printed; PII columns (Aadhaar, bank account) are redacted to last-4 unless the viewer has view-pii privilege (which itself records an event).

7. Migration plan

0007 .. 0012   — Vitta Fin (Agent 1)
0013           — AUDIT TRAIL — this module      (audit_events, pending_writes,
                                                 audit_coverage, audit_signoff_log)
0014           — Statutory Data tables
0015           — Compliance tables
0016+          — Other agents (Asset Mgmt, Staff, Members suite, etc.)

Audit goes BEFORE Statutory and Compliance because both rely on coverage entries seeded in 0013. Coverage seeds (global rows with tenant_id=NULL):

entity_type	action	column	two-key?	reason
trust_registrations	update	*	yes	trust deed metadata cannot change unilaterally
tax_registrations	update	certificate_number	yes	12A/80G/FCRA number is a legal anchor
tax_registrations	delete	*	yes	legal record
bank_loans	*	*	yes	financial liability
asset_acquisition_records	update	*	yes	sale-deed metadata
compliance_filings	update	filed_date,status	yes	filing timestamps are auditor-relevant
audit_engagements	update	*	yes	auditor identity & qualifications
document_custody	update	locker_id,custodian_user_id	yes	physical custody chain
jewel_vault	*	*	yes	(Asset Mgmt module — pre-seeded contract)
fcra_*	*	*	yes	(Vitta Fin — pre-seeded contract)
All others	update	*	no (auto-log only)	passive audit

Migration follows 0006_payments.py style: idempotent, additive-only, offline-safe inspector.

8. Cross-module dependencies

Audit Trail is uniquely inverted — every other module imports it; it imports nothing from any feature module.

Contract every module must follow

1. Mark the model. Add __audit_log__ = True (passive) and/or __audit_two_key__ = ('update', 'delete', ...) (active two-key).
2. For two-key writes, the module's service layer routes through submit_two_key_change(...) — never db.commit() directly.
3. For sensitive reads (PII view, FCRA donor read, jewel vault open), call record_audit_event(action='view-pii', ...) explicitly.
4. For bulk operations (delete-many, status-change-many), one AuditEvent per affected row OR one with a list of entity_ids — the bridge does the latter automatically when more than 50 rows.

Specific writers expected from other modules

Module	Calls	When
Statutory	record_audit_event	every doc edit, expiry change, custody transfer
Statutory	submit_two_key_change	trust-deed changes, tax-cert number edits
Compliance	record_audit_event	every filing status flip (pending→filed→accepted)
Compliance	submit_two_key_change	filed_date back-dating, audit-engagement opinion edit
Vitta Fin	submit_two_key_change	period-lock break, FCRA transfer, restricted-fund movement
Asset Mgmt	submit_two_key_change	jewel-vault any change, asset-disposal record
Members Suite	record_audit_event (PII)	Aadhaar reveal, bank-detail reveal
Staff Details	record_audit_event (PII)	medical/Aadhaar/bank reveal
Comms	record_audit_event	bulk message dispatch over N recipients

Why event-listener pattern, not module-explicit calls

VKG codebase has dozens of CRUD endpoints. Asking every endpoint to remember to call record_audit_event is brittle. The bridge in services/audit_bridge.py listens on the SQLAlchemy session and emits events automatically for any class with __audit_log__ = True. Modules only need to opt in (one class attribute) and explicitly call submit_two_key_change for the two-key path (which the bridge cannot infer because two-key requires human reason text).

9. Implementation phases

Phase A — Schemas + basic CRUD (Sprint 1, ~1 week). - Migration 0013 with all four tables + global coverage seeds. - Models in aayojana/models/audit.py. - services/audit.py with record_audit_event, is_two_key_required, list_pending_for_user. - Read-only routes: GET /api/audit/events, GET /api/audit/events/{id}, GET /api/audit/pending. - Minimal HTML page: /audit/log (read-only table).

Phase B — Two-key flow + bridge (Sprint 2, ~1 week). - submit_two_key_change, cosign_pending_change, reject_pending_change. - POST routes for cosign/reject. - SQLAlchemy event-listener bridge; opt-in via __audit_log__ class attribute. - Self-veto enforcement, race-detection on cosign. - Pending-queue HTML page with diff renderer. - Cron job for expire_stale_pending.

Phase C — Dashboards + bulk ops (Sprint 3, ~1 week). - Suspicious-activity heuristics + /audit/suspicious. - Coverage admin UI. - Bulk-event collapsing for big operations. - Nightly tamper-evidence checksum batch. - Embed entity-history sidebar into Statutory and Compliance detail pages. - Export endpoint: GET /api/audit/events.csv for auditor handoff.

10. Open questions

1. Two-key threshold. Per-tenant override of the global coverage table — yes (model supports it). But should a small tenant on Basic plan be allowed to disable a two-key requirement we ship as global default? Recommend NO — global coverage rules with requires_two_key=true are non-overridable.
2. Audit retention. Forever in Postgres, or hot 7 years + cold-storage (GCS Coldline) older? Form 10B requires 6-year retention; FCRA requires 6. Recommend: forever in Postgres for now (cheap on Neon), revisit at 100M rows.
3. Tamper evidence. Postgres has no native immutable tables. Options: (a) nightly row_hash chain (Merkle-style) checksummed by a separate cron; (b) write-only DB role. Recommend (a) — implementable on shared Neon.
4. Cosign window. Default 7 days (cosign_window_hours=168). Per-coverage override exists. Should expired writes be re-submittable as a fresh pending, or fully terminal? Recommend: re-submittable — submitter clicks "renew" which creates a new PendingWrite linking back via superseded_by.
5. PII in audit blobs. before_blob/after_blob may capture Aadhaar numbers if the column changes. Should we hash-redact at the audit layer, or trust the column-level redaction in the read endpoint? Recommend: store full value (audit must be complete), redact aggressively at read time and require view-pii privilege which itself audits.
6. Bulk operations. A delete-100-rows action — one event per row, or one event with a 100-element entity_ids array? Recommend: one event with array if the rows share entity_type AND the action is uniform; otherwise per-row. Cap at 1000 ids per event; spill above that.
7. Cosigner pool. eligible_cosigner_roles is a JSON array. Should a user need an explicit designation (e.g. "audit-cosigner") on top of role? Recommend NO for v1 — tenant-admin + role list is enough; revisit if regulators demand named cosigners.
8. Multi-tenant cosign. A user belonging to two tenants — events scoped by tenant_id, never cross-leak. Confirmed; document in service-layer tests.
9. Performance — JSON blobs on every update. For high-volume modules (Vitta journal entries) JSON capture per row is expensive. Recommend: __audit_log__ = 'sampled' mode that logs every Nth row plus all threshold-crossing rows. Vitta opts into sampled.
10. API consumers outside Aayojana. Should peer apps (SSM, Samudwaaha) write to this audit log via JWT-authenticated POST, or maintain their own? Recommend their own (federation principle), but expose a read-only GET /api/audit/events to tenant-admins for cross-app review.