fathom.AuditLog

fathom.AuditLog

Records audit entries from evaluation results via a pluggable sink.

Source code in src/fathom/audit.py

class AuditLog:
    """Records audit entries from evaluation results via a pluggable sink."""

    def __init__(self, sink: AuditSink) -> None:
        """Create an audit log backed by the given sink.

        Args:
            sink: Pluggable sink that receives serialised audit records.
        """
        self._sink = sink

    @property
    def is_recording(self) -> bool:
        """True when the configured sink actually persists records.

        Lets :meth:`fathom.engine.Engine.evaluate` skip the working-memory
        snapshot that only a real sink would ever consume.
        """
        return not isinstance(self._sink, NullSink)

    def record(
        self,
        result: EvaluationResult,
        session_id: str,
        input_facts: list[dict[str, object]] | None = None,
        modules_traversed: list[str] | None = None,
        *,
        asserted_facts: list[AssertedFact] | None = None,
        log_level: LogLevel = LogLevel.SUMMARY,
    ) -> None:
        """Write one audit record, honouring the winning rule's ``then.log``.

        Args:
            result: The evaluation result being recorded.
            session_id: Session the evaluation ran under.
            input_facts: Working-memory snapshot taken before inference.
                Written only at :attr:`LogLevel.FULL`.
            modules_traversed: Overrides ``result.module_trace``.
            asserted_facts: Facts the rules themselves asserted.
            log_level: The ``then.log`` level of the winning decision.
                :attr:`LogLevel.NONE` writes nothing at all;
                :attr:`LogLevel.SUMMARY` omits *input_facts*;
                :attr:`LogLevel.FULL` includes them.
        """
        if log_level is LogLevel.NONE:
            return
        if log_level is not LogLevel.FULL:
            input_facts = None
        audit = AuditRecord(
            timestamp=datetime.now(UTC).isoformat(),
            session_id=session_id,
            input_facts=input_facts,
            modules_traversed=modules_traversed or result.module_trace,
            rules_fired=result.rule_trace,
            decision=result.decision,
            reason=result.reason,
            duration_us=result.duration_us,
            metadata=result.metadata,
            asserted_facts=asserted_facts,
            match_evidence=result.match_evidence or None,
            attestation_token=result.attestation_token,
        )
        self._sink.write(audit)

is_recording property

True when the configured sink actually persists records.

Lets :meth:fathom.engine.Engine.evaluate skip the working-memory snapshot that only a real sink would ever consume.

__init__(sink)

Create an audit log backed by the given sink.

Parameters:

Name	Type	Description	Default
sink	AuditSink	Pluggable sink that receives serialised audit records.	required

Source code in src/fathom/audit.py

def __init__(self, sink: AuditSink) -> None:
    """Create an audit log backed by the given sink.

    Args:
        sink: Pluggable sink that receives serialised audit records.
    """
    self._sink = sink

record(result, session_id, input_facts=None, modules_traversed=None, *, asserted_facts=None, log_level=LogLevel.SUMMARY)

Write one audit record, honouring the winning rule's then.log.

Parameters:

Name	Type	Description	Default
result	EvaluationResult	The evaluation result being recorded.	required
session_id	str	Session the evaluation ran under.	required
input_facts	list[dict[str, object]] | None	Working-memory snapshot taken before inference. Written only at :attr:LogLevel.FULL.	None
modules_traversed	list[str] | None	Overrides result.module_trace.	None
asserted_facts	list[AssertedFact] | None	Facts the rules themselves asserted.	None
log_level	LogLevel	The then.log level of the winning decision. :attr:LogLevel.NONE writes nothing at all; :attr:LogLevel.SUMMARY omits input_facts; :attr:LogLevel.FULL includes them.	SUMMARY

Source code in src/fathom/audit.py

def record(
    self,
    result: EvaluationResult,
    session_id: str,
    input_facts: list[dict[str, object]] | None = None,
    modules_traversed: list[str] | None = None,
    *,
    asserted_facts: list[AssertedFact] | None = None,
    log_level: LogLevel = LogLevel.SUMMARY,
) -> None:
    """Write one audit record, honouring the winning rule's ``then.log``.

    Args:
        result: The evaluation result being recorded.
        session_id: Session the evaluation ran under.
        input_facts: Working-memory snapshot taken before inference.
            Written only at :attr:`LogLevel.FULL`.
        modules_traversed: Overrides ``result.module_trace``.
        asserted_facts: Facts the rules themselves asserted.
        log_level: The ``then.log`` level of the winning decision.
            :attr:`LogLevel.NONE` writes nothing at all;
            :attr:`LogLevel.SUMMARY` omits *input_facts*;
            :attr:`LogLevel.FULL` includes them.
    """
    if log_level is LogLevel.NONE:
        return
    if log_level is not LogLevel.FULL:
        input_facts = None
    audit = AuditRecord(
        timestamp=datetime.now(UTC).isoformat(),
        session_id=session_id,
        input_facts=input_facts,
        modules_traversed=modules_traversed or result.module_trace,
        rules_fired=result.rule_trace,
        decision=result.decision,
        reason=result.reason,
        duration_us=result.duration_us,
        metadata=result.metadata,
        asserted_facts=asserted_facts,
        match_evidence=result.match_evidence or None,
        attestation_token=result.attestation_token,
    )
    self._sink.write(audit)