Compliance & Audit Trail Logging for Work Order State Transitions

Compliance and audit trail logging is the evidentiary layer of the Technician Routing & SLA Enforcement pipeline — the component that records every work-order state transition and routing decision as an immutable, tamper-evident fact rather than a mutable database row.

A dispatch engine that cannot prove what it did, when, and on whose authority is a liability the moment an auditor, a regulator, or a disputed SLA credit arrives. Maintenance events touch safety-critical assets, contractual response windows, and warranty positions, so “the ticket says it was closed on Tuesday” is not evidence — a row that anyone with database access can silently edit proves nothing. This component closes that gap by treating each state change as an AuditEvent that is structured, attributed to an actor, and cryptographically chained to the event before it, then persisted to an append-only store so a later edit or deletion is mathematically detectable. Aligned to ISO 55001 asset-management traceability, it lets facilities managers demonstrate control of their maintenance record and gives Python automation teams a verifiable log they can export on demand. This guide implements that layer end to end — prerequisites, the input/output data contract, a step-by-step build, a configuration reference, chain-verification tests, and the failure modes that quietly destroy an audit trail’s integrity in production.

Prerequisites

This component runs as a synchronous side-effect of every state transition emitted by the routing and SLA engines. It never blocks dispatch, but it must never drop an event either. Before you deploy it, confirm the following are in place.

• Python 3.11+ with pydantic>=2.6 for the immutable event contract and orjson>=3.9 for canonical, byte-stable JSON serialization (ordinary json does not guarantee a stable byte layout across processes, which breaks hashing). The standard-library hashlib supplies the hash primitive; no third-party crypto library is required for the chain itself.
• An append-only store. Use a WORM-capable object store (S3 Object Lock in compliance mode, or an equivalent write-once volume) or an append-only table with UPDATE/DELETE revoked at the database-role level. The store must guarantee that once an event is committed it cannot be rewritten in place — the hash chain detects tampering, but the store is what makes tampering hard in the first place.
• A source of state transitions. Audit events are emitted when the routing engine assigns a technician and when escalation events fire from the SLA timer escalation component. Each transition must carry the before-state and after-state so the audit event can record both.
• Resolved actor identity. Every event needs an authenticated actor — a human user, a service account, or an automation worker — sourced from the identity and role model in security & access boundaries. An event with a null or “system” actor is an attribution gap, not an audit record.
• Environment variables: AUDIT_STORE_URI (the append-only destination), AUDIT_HASH_ALGO (default sha256), AUDIT_RETENTION_DAYS (default 2555, roughly seven years), and AUDIT_PII_REDACTION (default on). The write role must hold append-only permission and must not hold update or delete on the audit store.

Architecture and Data Contract

The logger sits beside the state machine, not inside it. A transition happens, the engine hands the logger the before/after states and the acting identity, and the logger produces exactly one AuditEvent that is hashed against the previous event’s hash and appended. It never mutates a prior event and it never decides routing — it only witnesses. Four boundaries keep the trail trustworthy.

• Capture boundary: an AuditEvent is built only from an authenticated transition; a change with no attributable actor is rejected before it can enter the chain as an anonymous fact.
• Chaining boundary: each event’s hash is computed over its own canonical bytes plus the prev_hash of the event before it, so the log is a linked chain where any retroactive edit invalidates every hash downstream.
• Persistence boundary: events are appended to a WORM or append-only store in commit order; the writer holds append-only permission and cannot rewrite history even if compromised.
• Export boundary: a read-only verifier walks the chain and can hand a regulator a provably-intact slice — see exporting ISO 55001 audit trails for regulators for the packaging and attestation detail this component feeds.

The canonical shape entering the logger is the site-wide WorkOrderPayload (its fields are defined in work order schema standards); the shape leaving it is an immutable AuditEvent carrying the actor, the action, the before/after snapshot, and the two hashes that bind it into the chain.

The contract is explicit and one-directional: a WorkOrderPayload and an authenticated actor go in, and an immutable AuditEvent comes out. Freezing the event with pydantic’s frozen=True means the in-memory record cannot be mutated after construction, and computing the hash over canonical bytes means any change to a persisted event is detectable no matter which field was touched.

from datetime import datetime, timezone
from enum import Enum
from typing import Any, Dict, Optional
from pydantic import BaseModel, Field, ConfigDict


class Action(str, Enum):
    """The transition classes that must be recorded for traceability."""
    CREATED = "created"
    ASSIGNED = "assigned"
    STATUS_CHANGED = "status_changed"
    ESCALATED = "escalated"
    SLA_PAUSED = "sla_paused"
    CLOSED = "closed"


class AuditEvent(BaseModel):
    """Immutable, hash-chained record of one work-order state transition."""
    model_config = ConfigDict(frozen=True)

    sequence: int = Field(..., ge=0, description="Monotonic position in the chain")
    work_order_id: str = Field(..., min_length=8, max_length=64)
    actor: str = Field(..., min_length=1, description="Authenticated identity, never null")
    action: Action
    before: Dict[str, Any] = Field(default_factory=dict)
    after: Dict[str, Any] = Field(default_factory=dict)
    recorded_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
    prev_hash: str = Field(..., description="hash of the preceding event, or the genesis seed")
    hash: Optional[str] = Field(None, description="H(canonical payload + prev_hash)")

Step-by-Step Implementation

1. Define the canonical work order payload

The logger records transitions of the same WorkOrderPayload used everywhere on this site rather than inventing its own shape. Its SLA fields — priority, requested_completion, and escalation_tier — are exactly the values that must be provable after the fact, because they determine contractual response windows and are the fields most likely to be disputed.

from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum
from typing import Dict, List, Optional


class Priority(str, Enum):
    CRITICAL = "critical"
    HIGH = "high"
    STANDARD = "standard"
    PLANNED = "planned"


@dataclass
class WorkOrderPayload:
    """Canonical CMMS work order — SLA fields are mandatory site-wide."""
    work_order_id: str
    asset_id: str
    part_skus: List[str]
    required_quantities: Dict[str, int]
    location_id: str
    priority: Priority = Priority.STANDARD
    requested_completion: Optional[datetime] = None
    escalation_tier: int = 0
    status: str = "open"
    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))


def state_snapshot(wo: WorkOrderPayload) -> Dict[str, object]:
    """The subset of fields whose changes are audit-relevant."""
    return {
        "status": wo.status,
        "priority": wo.priority.value,
        "escalation_tier": wo.escalation_tier,
        "requested_completion": (
            wo.requested_completion.isoformat() if wo.requested_completion else None
        ),
    }

2. Build a structured AuditEvent from a transition

Capture the before-state and after-state around the change, attach the authenticated actor, and refuse to build an event without one. Attribution is not optional — an event with a missing or placeholder actor is worse than no event, because it looks like a record while proving nothing about who acted.

class AttributionError(ValueError):
    """Raised when a transition has no authenticated actor."""


def build_audit_event(
    sequence: int,
    wo_before: WorkOrderPayload,
    wo_after: WorkOrderPayload,
    action: Action,
    actor: str,
    prev_hash: str,
) -> AuditEvent:
    """Construct one immutable event; the hash is filled in during chaining."""
    if not actor or actor.strip().lower() in {"", "system", "none", "null"}:
        raise AttributionError(
            f"transition for {wo_after.work_order_id} has no attributable actor"
        )
    return AuditEvent(
        sequence=sequence,
        work_order_id=wo_after.work_order_id,
        actor=actor,
        action=action,
        before=state_snapshot(wo_before),
        after=state_snapshot(wo_after),
        prev_hash=prev_hash,
    )

3. Hash-chain each entry so tampering is detectable

Serialize the event to canonical, byte-stable JSON, concatenate the previous event’s hash, and digest the result. Because each hash includes prev_hash, editing any historical event changes its hash, which breaks the prev_hash of the next event, which breaks its hash, and so on — a single silent edit invalidates the entire remainder of the chain. Redact configured PII fields before hashing so the stored bytes and the digest never contain protected data.

import hashlib
import orjson
from typing import Iterable

GENESIS_HASH = "0" * 64  # seed for the first event in a chain

_PII_FIELDS = {"actor_email", "actor_phone", "reporter_name"}


def _redact(payload: dict, pii_fields: Iterable[str]) -> dict:
    """Replace configured PII values with a stable marker before hashing."""
    redacted = dict(payload)
    for key in pii_fields:
        if key in redacted:
            redacted[key] = "[REDACTED]"
    return redacted


def compute_hash(event: AuditEvent, algo: str = "sha256") -> str:
    """H(canonical event bytes without its own hash + prev_hash)."""
    body = event.model_dump(exclude={"hash"}, mode="json")
    body = _redact(body, _PII_FIELDS)
    # orjson with sorted keys guarantees identical bytes across processes.
    canonical = orjson.dumps(body, option=orjson.OPT_SORT_KEYS)
    digest = hashlib.new(algo)
    digest.update(canonical)
    digest.update(event.prev_hash.encode("utf-8"))
    return digest.hexdigest()


def seal_event(event: AuditEvent, algo: str = "sha256") -> AuditEvent:
    """Return a copy of the event with its chain hash fixed in place."""
    return event.model_copy(update={"hash": compute_hash(event, algo)})

4. Persist to an append-only store

Append the sealed event in commit order and never rewrite an existing one. The reference writer below serializes each event as one JSON line (JSONL), opening the file strictly in append mode; in production the same interface fronts a WORM object store or an append-only table whose role lacks UPDATE and DELETE. The writer tracks the tail hash so the next event chains onto it without re-reading the whole log.

from pathlib import Path


class AppendOnlyAuditLog:
    """Append-only sink. Never seeks, never truncates, never overwrites."""

    def __init__(self, path: str, algo: str = "sha256") -> None:
        self.path = Path(path)
        self.algo = algo
        self._tail_hash = self._load_tail_hash()
        self._sequence = self._load_sequence()

    def _load_tail_hash(self) -> str:
        if not self.path.exists():
            return GENESIS_HASH
        last = None
        with self.path.open("rb") as fh:
            for line in fh:
                if line.strip():
                    last = line
        return orjson.loads(last)["hash"] if last else GENESIS_HASH

    def _load_sequence(self) -> int:
        if not self.path.exists():
            return 0
        with self.path.open("rb") as fh:
            return sum(1 for line in fh if line.strip())

    def append(
        self, wo_before: WorkOrderPayload, wo_after: WorkOrderPayload,
        action: Action, actor: str,
    ) -> AuditEvent:
        event = build_audit_event(
            self._sequence, wo_before, wo_after, action, actor, self._tail_hash
        )
        sealed = seal_event(event, self.algo)
        # "a" mode only — the OS refuses to move the write offset backwards.
        with self.path.open("ab") as fh:
            fh.write(orjson.dumps(sealed.model_dump(mode="json")) + b"\n")
        self._tail_hash = sealed.hash
        self._sequence += 1
        return sealed

5. Verify chain integrity

Walk the log from the genesis seed forward, recomputing each event’s hash and confirming that every event’s prev_hash equals the actual hash of its predecessor. Any mismatch pinpoints the first tampered or missing event by sequence number, which is the evidence an auditor needs — not just “the chain is broken” but exactly where.

from typing import List


class ChainBroken(Exception):
    """Raised at the first event whose hash or linkage does not verify."""

    def __init__(self, sequence: int, reason: str) -> None:
        super().__init__(f"chain broke at sequence {sequence}: {reason}")
        self.sequence = sequence


def verify_chain(events: List[AuditEvent], algo: str = "sha256") -> bool:
    """Return True only if every hash and every prev_hash link is intact."""
    expected_prev = GENESIS_HASH
    for event in events:
        if event.prev_hash != expected_prev:
            raise ChainBroken(event.sequence, "prev_hash does not match predecessor")
        recomputed = compute_hash(event, algo)
        if recomputed != event.hash:
            raise ChainBroken(event.sequence, "recomputed hash does not match stored hash")
        expected_prev = event.hash
    return True


def load_events(path: str) -> List[AuditEvent]:
    """Read the append-only log back into ordered AuditEvent records."""
    events: List[AuditEvent] = []
    with open(path, "rb") as fh:
        for line in fh:
            if line.strip():
                events.append(AuditEvent(**orjson.loads(line)))
    return events

Configuration Reference

Keep every tunable in a version-controlled configuration registry, not in the worker source. The defaults below suit a general-purpose maintenance record under a seven-year retention obligation.

Validation and Testing

The single highest-value test proves the chain is actually tamper-evident: build a short chain, mutate one persisted event, and assert that verification raises ChainBroken at exactly that event’s sequence. If this test passes, a silent edit anywhere in production is guaranteed to be caught by the verifier.

def _wo(status: str, tier: int = 0) -> WorkOrderPayload:
    return WorkOrderPayload(
        work_order_id="WO-2026-0042",
        asset_id="PUMP-014",
        part_skus=["PMP-SEAL-0042"],
        required_quantities={"PMP-SEAL-0042": 1},
        location_id="STORE-A",
        priority=Priority.CRITICAL,
        status=status,
        escalation_tier=tier,
    )


def test_tampering_breaks_the_chain(tmp_path):
    log = AppendOnlyAuditLog(str(tmp_path / "audit.jsonl"))
    log.append(_wo("open"), _wo("assigned"), Action.ASSIGNED, actor="svc-dispatch")
    log.append(_wo("assigned"), _wo("escalated", tier=1), Action.ESCALATED, actor="svc-sla")

    events = load_events(str(tmp_path / "audit.jsonl"))
    assert verify_chain(events) is True  # pristine chain verifies

    # Silently edit event 0's "after" state, as a bad actor would.
    forged = events[0].model_copy(update={"after": {"status": "closed"}})
    tampered = [forged, events[1]]

    import pytest
    with pytest.raises(ChainBroken) as exc:
        verify_chain(tampered)
    assert exc.value.sequence == 0  # the verifier pinpoints the forged event

A missing actor must also fail fast rather than entering the chain as an anonymous fact — build_audit_event(..., actor="system") should raise AttributionError before any hash is computed. In integration, assert that the append writer emits monotonically increasing sequence values with no gaps: a break in the sequence run is the earliest signal that events were dropped between the state machine and the store.

Failure Modes and Troubleshooting

Expand each scenario for the root cause, the diagnostic signal, and the fix. The checklist items render as interactive checkboxes — work through them in order.

A mutable log lets someone silently edit closed history

• Confirm the storage role has no UPDATE/DELETE grant and the object store is in WORM compliance mode — if a row can be edited in place, the chain is your only defense and it is not enough on its own.
• Run verify_chain on a schedule and alert on any ChainBroken; the sequence number in the exception localizes the first altered event.
• Verify writes use append mode only — a code path that opens the log for r+ or seeks before writing can overwrite sealed events and must be removed.

Events with a missing actor or “system” attribution

• Grep the log for "actor": "system" or empty actors; each one is an untraceable change that fails an ISO 55001 review.
• Confirm the transition source resolves a concrete identity from security & access boundaries before calling append, and that AttributionError is not being swallowed upstream.
• For genuine automation, assign a named service-account identity (svc-dispatch, svc-sla) rather than a generic placeholder so even machine actions are attributable.

Clock skew reorders events under concurrent load

• Check for events whose recorded_at moves backwards relative to their sequence; the monotonic sequence is authoritative, but skewed recorded_at values mislead human review.
• Confirm every writer host is NTP-disciplined and clock_source is ntp, not the raw host clock.
• Order any export and any reconciliation by sequence, never by timestamp, so a skewed clock cannot rearrange the evidentiary order.

Gaps in the chain from dropped events

• Scan for non-contiguous sequence values; a jump from N to N+2 means an event never reached the store even though its transition occurred.
• Confirm the logger is invoked synchronously in the same transaction boundary as the state change so a crash cannot commit the change while losing its audit event.
• Ensure the append call does not run on a best-effort fire-and-forget task — a dropped event is an integrity hole, so a failed append must fail the transition or be retried onto the tail hash, never discarded.

Frequently Asked Questions

Why hash-chain the log instead of just write-protecting the database?

Write protection and hash chaining defend against different threats and you want both. WORM storage makes tampering hard by refusing overwrites; the hash chain makes tampering detectable even if the storage layer is bypassed, misconfigured, or restored from a doctored backup. If someone with sufficient privilege edits a record, the recomputed hash no longer matches and every downstream event breaks, so the alteration cannot pass unnoticed.

What exactly makes an edit detectable?

Each event’s hash is computed over its own canonical bytes plus the previous event’s hash. Changing any field of a historical event changes its recomputed hash, which no longer equals the prev_hash stored in the following event, which invalidates that event’s hash in turn. Verification walks the chain from the genesis seed and stops at the first mismatch, reporting the exact sequence number of the tampered or missing event.

How does this satisfy ISO 55001 traceability?

ISO 55001 requires that decisions and changes affecting managed assets be traceable end to end. An append-only, attributed, tamper-evident record of every work-order transition and routing/SLA decision provides that: who acted, what changed from what to what, when, and proof the record has not been altered since. The regulator export path packages a verified slice with its integrity attestation.

Does redaction of PII weaken the audit trail?

No, because redaction happens before hashing, so the digest is computed over the redacted bytes and remains stable and verifiable. The trail still proves who did what and when; it simply never stores protected personal data in the sealed record. The actor is retained as a stable identifier, while free-text personal fields are replaced with a marker that participates in the hash like any other value.

Should audit logging ever block a dispatch?

A failed append must not be silently dropped, but it also should not stall live dispatch indefinitely. Emit the event inside the same transaction as the state change so the two commit or fail together, and if the store is briefly unreachable, retry onto the current tail hash rather than skipping the event. Losing an event creates a chain gap, which is a worse outcome than a short retry.

Related

Route the escalation transitions recorded here from SLA timer escalation, resolve every actor identity through security & access boundaries, keep the recorded shape aligned with work order schema standards, and hand a verified, attested slice to a regulator via exporting ISO 55001 audit trails for regulators.

Part of: Technician Routing & SLA Enforcement.