Contextual Audit Trace

The problem

Tracking data changes is difficult because it is hard to tell if a change is a normal update or a significant, high-impact shift. Standard logs show what changed, but not the context or the 'health' of that change.

What it does

It monitors data changes and assigns a score based on how much they disrupt the overall stability and meaning of the information. It tracks the integrity of these transitions by looking at the context of the change.

Why it matters

It allows you to see the real impact of data updates rather than just a list of modified values.

Validation

It was run inside an isolated container with no network access. This is the exact command and the real output it produced — captured process output, not written by a model.

$ python3 contextual_audit_trace.py
Invalid JSON: Expecting value: line 1 column 1 (char 0)

A screenshot of that run.

A clean run proves this does what is shown above, in a CPU-only sandbox. It is a small research demo — not a production tool, and nothing here was published anywhere.

The code

All of it — 409 lines, one file, standard library only.

#!/usr/bin/env python3
"""
Contextual Audit Trace v2

Combines the Homeostatic Retrieval Scorer's multi-dimensional stability
scoring (emotional intensity + semantic stability) with Laravel ActivityLog-style
model-change tracking (create/update/delete with property-level diffs) to
score the *integrity* of data state transitions.

v2 adds: Anomaly Tagging Engine — flags events based on configurable
score thresholds and combinatorial rules.

Usage:
    python contextual_audit_trace.py --input events.json [--output result.json]
    cat events.json | python contextual_audit_trace.py

Input JSON:
{
  "model": "Order",
  "events": [
    {
      "event": "created",
      "timestamp": "2025-01-01T10:00:00Z",
      "causer": "admin@example.com",
      "subject_id": 1001,
      "properties": {
        "attributes": { "status": "pending", "total": "150.00" },
        "old": {}
      }
    },
    {
      "event": "updated",
      "timestamp": "2025-01-01T10:05:00Z",
      "causer": "admin@example.com",
      "subject_id": 1001,
      "properties": {
        "attributes": { "status": "confirmed", "total": "150.00" },
        "old": { "status": "pending", "total": "150.00" }
      }
    }
  ]
}

Output JSON: each event enriched with:
  emotional_intensity   — lexical charge in keys + values (0–1)
  semantic_stability    — cosine similarity between old→new token sets (0–1)
  homeostatic_score     — stability damped by excess emotional noise (0–1)
  structural_inflation  — nested-object ratio (0–1)
  transition_integrity  — final holistic score weighted by position, drift, burst
  flags                 — anomaly tags (v2)
"""

import json
import math
import re
import sys
from collections import Counter
from typing import Any


# ===================================================================
# Emotional Lexicon — terms that carry operational sentiment
# ===================================================================

EMOTIONAL_LEXICON: dict[str, float] = {
    # high negative
    "critical": 0.95, "urgent": 0.90, "breach": 0.88, "attack": 0.85,
    "error": 0.70, "failure": 0.80, "crash": 0.82, "alert": 0.75,
    "corrupt": 0.92, "invalid": 0.70, "fatal": 0.93, "panic": 0.88,
    "overload": 0.78, "saturated": 0.72, "depleted": 0.68,
    "anomaly": 0.80, "suspicious": 0.75,
    # moderate
    "warning": 0.60, "degraded": 0.55, "denied": 0.65, "blocked": 0.68,
    "timeout": 0.65, "missing": 0.55, "expired": 0.45, "deleted": 0.50,
    "locked": 0.60, "pending": 0.20, "unknown": 0.30,
    # neutral / positive
    "normal": 0.00, "ok": 0.00, "stable": 0.00, "healthy": -0.10,
    "resolved": -0.10, "closed": -0.10, "succeeded": -0.05,
    "success": -0.05, "complete": -0.05, "accepted": 0.00,
    "created": 0.10, "updated": 0.05, "modified": 0.05, "set": 0.02,
    "assigned": 0.05,
}


# ===================================================================
# Tokenisation & emotional intensity
# ===================================================================

_TOKEN_RE = re.compile(r"[a-z_]{2,}")

def _tokenize(text: str) -> list[str]:
    return _TOKEN_RE.findall(text.lower())

def emotional_intensity(text: str) -> float:
    """Score 0–1: emotional weight of a string based on lexicon hits."""
    tokens = _tokenize(text)
    if not tokens:
        return 0.0
    weights = [abs(EMOTIONAL_LEXICON.get(t, 0.0)) for t in tokens]
    hit_count = sum(1 for w in weights if w > 0.2)
    avg_weight = sum(weights) / len(weights)
    density = hit_count / len(weights)
    raw = 0.6 * avg_weight + 0.4 * density
    return min(1.0, max(0.0, raw))


# ===================================================================
# Semantic stability — cosine of bag-of-words
# ===================================================================

def _bag_of_words(values: list[str]) -> Counter[str]:
    bag: Counter[str] = Counter()
    for text in values:
        bag.update(_tokenize(text))
    return bag

def semantic_stability(old: dict[str, Any], new: dict[str, Any]) -> float:
    """
    Cosine similarity between old and new attribute token-sets.
    1.0 = identical vocabulary, 0.0 = no overlap.
    """
    bow_old = _bag_of_words(str(v) for v in (old or {}).values())
    bow_new = _bag_of_words(str(v) for v in (new or {}).values())
    all_keys = set(bow_old) | set(bow_new)
    if not all_keys:
        return 1.0
    dot = sum(bow_old[k] * bow_new[k] for k in all_keys)
    norm_a = math.sqrt(sum(v * v for v in bow_old.values()))
    norm_b = math.sqrt(sum(v * v for v in bow_new.values()))
    if norm_a == 0.0 or norm_b == 0.0:
        return 0.0
    return dot / (norm_a * norm_b)


# ===================================================================
# Structural inflation — how much nested bloat appeared
# ===================================================================

def structural_inflation(attributes: dict[str, Any]) -> float:
    """Ratio of nested (dict/list) values — ballooning signals schema creep."""
    if not attributes:
        return 0.0
    nested = sum(1 for v in attributes.values() if isinstance(v, (dict, list)))
    return min(1.0, nested / max(len(attributes), 1))


# ===================================================================
# Homeostatic score (HRS)
# ===================================================================

def homeostatic_score(
    intensity: float,
    stability: float,
    intensity_threshold: float = 0.5,
) -> float:
    """
    HRS = stability * (1 − excess_ratio)
    where excess_ratio = max(0, (intensity − threshold) / (1 − threshold))

    Low intensity + high stability  →  high HRS (deliberate, calm change)
    High intensity + low stability  →  low HRS (panicked, noisy change)
    """
    if intensity <= intensity_threshold:
        excess = 0.0
    else:
        excess = (intensity - intensity_threshold) / (1.0 - intensity_threshold)
    damped = stability * (1.0 - excess)
    return min(1.0, max(0.0, damped))


# ===================================================================
# Transition integrity — positional + drift + burst-aware composite
# ===================================================================

def _transition_key_changes(old: dict[str, Any], new: dict[str, Any]) -> int:
    """How many keys had a value change."""
    all_keys = set(old) | set(new)
    return sum(1 for k in all_keys if old.get(k) != new.get(k))


def transition_integrity(
    event_idx: int,
    total_events: int,
    stability: float,
    intensity: float,
    inflation: float,
    key_changes: int,
    event_type: str,
) -> float:
    """
    Composite score 0–1 factoring:
      - position weight (early events trusted less)
      - drift penalty (1 − stability)
      - burst penalty (intensity × inflation)
      - key-change magnitude
    """
    if total_events <= 1:
        pos_weight = 1.0
    else:
        pos_weight = 0.3 + 0.7 * (event_idx / (total_events - 1))

    drift_penalty = 1.0 - stability
    burst_penalty = intensity * inflation

    if event_type == "created":
        key_weight = 1.0
    elif event_type == "deleted":
        key_weight = 1.0
    else:
        key_weight = max(0.2, 1.0 - key_changes * 0.15)

    composite = (
        0.35 * pos_weight
        + 0.25 * (1.0 - drift_penalty)
        + 0.15 * (1.0 - burst_penalty)
        + 0.25 * key_weight
    )
    return min(1.0, max(0.0, composite))


# ===================================================================
# Anomaly Tagging Engine (v2)
# ===================================================================

# -- rulebook: name → (condition_fn, severity) --
# condition_fn receives a dict with keys from a tagged event's scores
# and returns True if the anomaly flag should fire.

_ANOMALY_RULES: list[tuple[str, str, callable]] = [
    (
        "High Emotional Volatility",
        "Emotional intensity exceeds 0.7 — emotionally charged language detected in field names or values.",
        lambda s: s["emotional_intensity"] > 0.7,
    ),
    (
        "Semantic Drift",
        "Semantic stability below 0.3 — vocabulary shift between old and new attributes is extreme.",
        lambda s: s["semantic_stability"] < 0.3,
    ),
    (
        "Structural Bloat",
        "Structural inflation above 0.5 — nested objects/lists dominate the payload.",
        lambda s: s["structural_inflation"] > 0.5,
    ),
    (
        "Low Homeostasis",
        "Homeostatic score below 0.2 — system in flux, damped by emotional noise.",
        lambda s: s["homeostatic_score"] < 0.2,
    ),
    (
        "Transition Collapse",
        "Transition integrity below 0.3 — multi-factor signal of a compromised state transition.",
        lambda s: s["transition_integrity"] < 0.3,
    ),
]


def tag_anomalies(
    event_enriched: dict[str, Any],
    rules: list[tuple[str, str, str]] | None = None,
) -> list[dict[str, Any]]:
    """
    Return a list of anomaly tags fired for a single enriched event.

    Each tag is:
      { "tag": str, "description": str, "triggered_score": float }
    """
    rules = rules or _ANOMALY_RULES
    results: list[dict[str, Any]] = []
    scores: dict[str, float] = {
        "emotional_intensity": event_enriched.get("emotional_intensity", 0.0),
        "semantic_stability": event_enriched.get("semantic_stability", 1.0),
        "homeostatic_score": event_enriched.get("homeostatic_score", 1.0),
        "structural_inflation": event_enriched.get("structural_inflation", 0.0),
        "transition_integrity": event_enriched.get("transition_integrity", 1.0),
    }
    for tag_name, description, pred in rules:
        if pred(scores):
            score_key = _tag_name_to_score_key(tag_name)
            results.append({
                "tag": tag_name,
                "description": description,
                "triggered_score": scores.get(score_key, 0.0),
            })
    return results


def _tag_name_to_score_key(tag: str) -> str:
    return {
        "High Emotional Volatility": "emotional_intensity",
        "Semantic Drift": "semantic_stability",
        "Structural Bloat": "structural_inflation",
        "Low Homeostasis": "homeostatic_score",
        "Transition Collapse": "transition_integrity",
    }.get(tag, "")


# ===================================================================
# Pipeline — process one event
# ===================================================================

def process_event(event: dict[str, Any], idx: int, total: int) -> dict[str, Any]:
    """Enrich a single event with all scores and anomaly tags."""
    props = event.get("properties", {})
    attrs = props.get("attributes", {})
    old   = props.get("old", {})

    # assemble the text surface for emotional scoring
    surface_parts: list[str] = [event.get("event", "")]
    if isinstance(attrs, dict):
        surface_parts.extend(str(k) for k in attrs)
        surface_parts.extend(str(v) for v in attrs.values())
    surface = " ".join(surface_parts)

    emo = emotional_intensity(surface)
    sem = semantic_stability(old, attrs)
    inf = structural_inflation(attrs)
    hom = homeostatic_score(emo, sem)
    key_changes = _transition_key_changes(old or {}, attrs or {})
    tra = transition_integrity(idx, total, sem, emo, inf,
                               key_changes, event.get("event", "??"))

    result = dict(event)
    result["emotional_intensity"] = round(emo, 4)
    result["semantic_stability"] = round(sem, 4)
    result["homeostatic_score"] = round(hom, 4)
    result["structural_inflation"] = round(inf, 4)
    result["transition_integrity"] = round(tra, 4)
    result["flags"] = tag_anomalies(result)

    return result


# ===================================================================
# I/O
# ===================================================================

def main() -> None:
    total, enriched = len(EVENTS), []
    for i, ev in enumerate(EVENTS):
        enriched.append(process_event(ev, i, total))

    output = {"model": MODEL, "events": enriched}
    print(json.dumps(output, indent=2))

    tag_counts: Counter[str] = Counter()
    for ev in enriched:
        for flag in ev["flags"]:
            tag_counts[flag["tag"]] += 1

    print(file=sys.stderr)
    for tag, count in tag_counts.most_common():
        print(f"[anomaly] {tag}: {count} event(s)", file=sys.stderr)
    if not tag_counts:
        print("[anomaly] No anomalies detected.", file=sys.stderr)


# ===================================================================
# Hard-coded example data (v2 — high-bloat + high-volatility events)
# ===================================================================

MODEL = "Order"

EVENTS: list[dict[str, Any]] = [
    {
        "event": "created",
        "timestamp": "2025-08-02T08:00:00Z",
        "causer": "ops@example.com",
        "subject_id": 2001,
        "properties": {
            "attributes": {"status": "pending", "total": "320.00"},
            "old": {},
        },
    },
    {
        "event": "updated",
        "timestamp": "2025-08-02T08:15:00Z",
        "causer": "ops@example.com",
        "subject_id": 2001,
        "properties": {
            "attributes": {"status": "confirmed", "total": "320.00"},
            "old": {"status": "pending", "total": "320.00"},
        },
    },
    {
        "event": "updated",
        "timestamp": "2025-08-02T09:30:00Z",
        "causer": "engineer@example.com",
        "subject_id": 2002,
        "properties": {
            "attributes": {
                "status": "critical",
                "error": "urgent",
                "alert": "failure crash",
                "state": "degraded",
            },
            "old": {"status": "normal", "error": "none", "state": "stable"},
        },
    },
    {
        "event": "updated",
        "timestamp": "2025-08-02T10:00:00Z",
        "causer": "admin@example.com",
        "subject_id": 2003,
        "properties": {
            "attributes": {
                "status": "active",
                "metadata": {
                    "deep": {