argus-ledger 0.1.0-alpha.2

Tamper-evident append-only SHA-256 provenance and audit ledger for autonomous agents.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

"""Interface protocols for Argus append-only audit ledger."""

from __future__ import annotations

from typing import Any, Dict, Iterator, Optional, Protocol, runtime_checkable
from .models import AuditEvent, ChainVerificationReport


class LedgerError(Exception):
    """Base exception for ledger operations."""
    pass


class TamperDetectedError(LedgerError):
    """Raised when ledger verification detects hash mismatch or chain tampering."""
    pass


@runtime_checkable
class AuditLedgerWriter(Protocol):
    """Writer interface for appending audit events."""

    def append(
        self,
        event_type: str,
        actor: str,
        payload: Dict[str, Any],
        event_id: Optional[str] = None,
    ) -> AuditEvent:
        """Append an event to the ledger with hash chaining."""
        ...


@runtime_checkable
class AuditLedgerReader(Protocol):
    """Reader interface for inspecting audit events."""

    def get_event(self, sequence: int) -> Optional[AuditEvent]:
        """Fetch event by sequence number."""
        ...

    def iter_events(self, start: int = 1, end: Optional[int] = None) -> Iterator[AuditEvent]:
        """Iterate events within sequence range."""
        ...

    def head(self) -> Optional[AuditEvent]:
        """Return the latest event in the chain."""
        ...


@runtime_checkable
class AuditVerifier(Protocol):
    """Verifier interface for validating hash chain continuity."""

    def verify_chain(self) -> ChainVerificationReport:
        """Traverse the entire ledger and verify SHA-256 hash continuity and integrity."""
        ...