calute.audit.events

calute.audit.events#

Typed audit event definitions for structured event export.

This module defines a hierarchy of dataclass-based audit events that capture every significant decision and transition in a Calute agent run. All events inherit from AuditEvent and are fully JSON-serializable via their AuditEvent.to_dict() and AuditEvent.to_json() methods.

The event taxonomy covers the following domains:

Turn lifecycle – TurnStartEvent, TurnEndEvent

Tool execution – ToolCallAttemptEvent, ToolCallCompleteEvent, ToolCallFailureEvent

Policy & security – ToolPolicyDecisionEvent, SandboxDecisionEvent

Loop detection – ToolLoopWarningEvent, ToolLoopBlockEvent

Hook introspection – HookMutationEvent

General errors – ErrorEvent

Example

Creating and serializing a tool-call event:

event = ToolCallAttemptEvent(
    tool_name="web_search",
    arguments_preview='{"query": "calute docs"}',
    agent_id="agent-1",
    turn_id="abc123",
)
print(event.to_json())

class calute.audit.events.AuditEvent(event_type: str = 'base', timestamp: str = <factory>, agent_id: str | None = None, turn_id: str | None = None, session_id: str | None = None, severity: str = 'info', metadata: dict[str, typing.Any] = <factory>)[source]#

Bases: object

Base audit event carrying the common metadata envelope.

Every audit event in the Calute audit system inherits from this dataclass. It provides the shared envelope fields (timestamps, IDs, severity) and the serialization helpers that collectors rely on.

Subclasses should not override event_type via __init__; instead they declare it as a field(default=..., init=False) so that the type tag is automatically set and immutable.

event_type#

A short string tag identifying the event kind (e.g. "turn_start", "tool_call_attempt").

Type: str

timestamp#

ISO-8601 UTC timestamp captured at construction time.

Type: str

agent_id#

Optional identifier of the agent that produced the event.

Type: str | None

turn_id#

Optional identifier of the conversational turn.

Type: str | None

session_id#

Optional session-level identifier, typically stamped by the AuditEmitter.

Type: str | None

severity#

Log-style severity level. One of "info", "warning", or "error".

Type: str

metadata#

Free-form dictionary for attaching additional context that does not warrant its own field.

Type: dict[str, Any]

Example

Constructing a bare base event:

event = AuditEvent(agent_id="agent-1", severity="warning")
payload = event.to_dict()
assert payload["event_type"] == "base"

agent_id: str | None = None#

event_type: str = 'base'#

metadata: dict[str, Any]#

session_id: str | None = None#

severity: str = 'info'#

timestamp: str#

to_dict() → dict[str, Any][source]#

Serialize the event to a plain dictionary.

Converts the entire dataclass tree (including nested dataclasses) into a dictionary of primitive types suitable for JSON encoding.

Returns

A recursively-unwrapped dictionary of all: fields and their values.

Return type

dict[str, Any]

to_json() → str[source]#

Serialize the event to a compact JSON string.

Non-serializable values (e.g. datetime objects stored in metadata) are coerced to strings via the default=str fallback.

Returns: A single-line JSON representation of the event.
Return type: str

turn_id: str | None = None#

class calute.audit.events.ErrorEvent(timestamp: str = <factory>, agent_id: str | None = None, turn_id: str | None = None, session_id: str | None = None, severity: str = 'error', metadata: dict[str, typing.Any] = <factory>, error_type: str = '', error_message: str = '', error_context: str = '')[source]#

Bases: AuditEvent

Emitted for generic errors not tied to a specific tool call.

Use this event type for infrastructure-level or agent-level errors that cannot be attributed to a single tool invocation (e.g. LLM API failures, serialization errors, or unexpected state transitions). For tool-specific failures, prefer ToolCallFailureEvent.

event_type#

Fixed to "error" (not settable via init).

Type: str

severity#

Defaults to "error" (overrides the base "info").

Type: str

error_type#

A short classifier for the error, typically the exception class name (e.g. "RuntimeError").

Type: str

error_message#

The human-readable error description or the stringified exception message.

Type: str

error_context#

Additional context about where or why the error occurred (e.g. "during response parsing").

Type: str

Example

event = ErrorEvent(
    error_type="TimeoutError",
    error_message="LLM request timed out after 30s",
    error_context="during turn execution",
    agent_id="agent-1",
)

error_context: str = ''#

error_message: str = ''#

error_type: str = ''#

event_type: str = 'error'#

severity: str = 'error'#

class calute.audit.events.HookMutationEvent(timestamp: str = <factory>, agent_id: str | None = None, turn_id: str | None = None, session_id: str | None = None, severity: str = 'info', metadata: dict[str, typing.Any] = <factory>, hook_name: str = '', tool_name: str = '', mutated_field: str = '')[source]#

Bases: AuditEvent

Emitted when a hook mutates a tool call or its result.

Hooks can intercept tool calls before execution (pre-hooks) or modify results after execution (post-hooks). This event records which hook altered which field, providing an audit trail for any data transformations applied outside the tool’s own logic.

event_type#

Fixed to "hook_mutation" (not settable via init).

Type: str

hook_name#

The identifier of the hook that performed the mutation (e.g. "sanitize_output", "inject_context").

Type: str

tool_name#

The registered name of the tool whose call or result was mutated.

Type: str

mutated_field#

The specific field that was changed (e.g. "arguments", "result").

Type: str

Example

event = HookMutationEvent(
    hook_name="sanitize_output",
    tool_name="shell_exec",
    mutated_field="result",
    agent_id="agent-1",
)

event_type: str = 'hook_mutation'#

hook_name: str = ''#

mutated_field: str = ''#

tool_name: str = ''#

class calute.audit.events.SandboxDecisionEvent(timestamp: str = <factory>, agent_id: str | None = None, turn_id: str | None = None, session_id: str | None = None, severity: str = 'info', metadata: dict[str, typing.Any] = <factory>, tool_name: str = '', context: str = '', reason: str = '', backend_type: str = '')[source]#

Bases: AuditEvent

Emitted when the sandbox router selects an execution backend.

The sandbox router inspects the tool call and its context to decide which isolation backend (e.g. local, Docker, subprocess jail) should execute the call. This event captures that routing decision.

event_type#

Fixed to "sandbox_decision" (not settable via init).

Type: str

tool_name#

The registered name of the tool being routed.

Type: str

context#

A short description of the execution context that influenced the routing decision.

Type: str

reason#

A human-readable explanation of why the particular backend was selected.

Type: str

backend_type#

The identifier of the chosen sandbox backend (e.g. "local", "docker", "subprocess").

Type: str

Example

event = SandboxDecisionEvent(
    tool_name="shell_exec",
    context="untrusted_input",
    reason="Input contains user-supplied shell commands",
    backend_type="docker",
)

backend_type: str = ''#

context: str = ''#

event_type: str = 'sandbox_decision'#

reason: str = ''#

tool_name: str = ''#

class calute.audit.events.ToolCallAttemptEvent(timestamp: str = <factory>, agent_id: str | None = None, turn_id: str | None = None, session_id: str | None = None, severity: str = 'info', metadata: dict[str, typing.Any] = <factory>, tool_name: str = '', arguments_preview: str = '')[source]#

Bases: AuditEvent

Emitted just before a tool call is dispatched to its executor.

This event is created before the tool actually runs, making it useful for pre-execution auditing and for correlating with the subsequent ToolCallCompleteEvent or ToolCallFailureEvent.

event_type#

Fixed to "tool_call_attempt" (not settable via init).

Type: str

tool_name#

The registered name of the tool being invoked.

Type: str

arguments_preview#

A truncated (up to 200 characters) string representation of the arguments passed to the tool.

Type: str

Example

event = ToolCallAttemptEvent(
    tool_name="web_search",
    arguments_preview='{"query": "Python dataclasses"}',
    agent_id="agent-1",
)

arguments_preview: str = ''#

event_type: str = 'tool_call_attempt'#

tool_name: str = ''#

class calute.audit.events.ToolCallCompleteEvent(timestamp: str = <factory>, agent_id: str | None = None, turn_id: str | None = None, session_id: str | None = None, severity: str = 'info', metadata: dict[str, typing.Any] = <factory>, tool_name: str = '', status: str = 'success', duration_ms: float = 0.0, result_preview: str = '')[source]#

Bases: AuditEvent

Emitted when a tool call completes successfully.

Captures the outcome of a successful tool invocation, including its wall-clock duration and a truncated preview of the result payload.

event_type#

Fixed to "tool_call_complete" (not settable via init).

Type: str

tool_name#

The registered name of the tool that was invoked.

Type: str

status#

Completion status string, typically "success".

Type: str

duration_ms#

Wall-clock execution time in milliseconds.

Type: float

result_preview#

A truncated (up to 200 characters) string representation of the tool’s return value.

Type: str

Example

event = ToolCallCompleteEvent(
    tool_name="web_search",
    status="success",
    duration_ms=142.5,
    result_preview="Found 10 results for ...",
)

duration_ms: float = 0.0#

event_type: str = 'tool_call_complete'#

result_preview: str = ''#

status: str = 'success'#

tool_name: str = ''#

class calute.audit.events.ToolCallFailureEvent(timestamp: str = <factory>, agent_id: str | None = None, turn_id: str | None = None, session_id: str | None = None, severity: str = 'error', metadata: dict[str, typing.Any] = <factory>, tool_name: str = '', error_type: str = '', error_message: str = '')[source]#

Bases: AuditEvent

Emitted when a tool call raises an exception or otherwise fails.

The event automatically sets severity to "error" to ensure failures are prominently surfaced in any downstream log consumer.

event_type#

Fixed to "tool_call_failure" (not settable via init).

Type: str

severity#

Defaults to "error" (overrides the base "info").

Type: str

tool_name#

The registered name of the tool that failed.

Type: str

error_type#

A short classifier for the error, typically the exception class name (e.g. "ValueError").

Type: str

error_message#

The human-readable error description or the stringified exception message.

Type: str

Example

event = ToolCallFailureEvent(
    tool_name="file_read",
    error_type="FileNotFoundError",
    error_message="/tmp/missing.txt not found",
    agent_id="agent-1",
)

error_message: str = ''#

error_type: str = ''#

event_type: str = 'tool_call_failure'#

severity: str = 'error'#

tool_name: str = ''#

class calute.audit.events.ToolLoopBlockEvent(timestamp: str = <factory>, agent_id: str | None = None, turn_id: str | None = None, session_id: str | None = None, severity: str = 'error', metadata: dict[str, typing.Any] = <factory>, tool_name: str = '', pattern: str = '', call_count: int = 0)[source]#

Bases: AuditEvent

Emitted when a tool-call loop triggers a hard block.

Unlike ToolLoopWarningEvent, this event indicates that the loop detector has prevented the tool call from executing. The severity is automatically set to "error".

event_type#

Fixed to "tool_loop_block" (not settable via init).

Type: str

severity#

Defaults to "error" (overrides the base "info").

Type: str

tool_name#

The registered name of the tool that was blocked.

Type: str

pattern#

A short description of the repetitive pattern that triggered the block (e.g. "same_args_5x").

Type: str

call_count#

The number of consecutive or recent calls that matched the loop pattern before the block was imposed.

Type: int

Example

event = ToolLoopBlockEvent(
    tool_name="web_search",
    pattern="same_args_5x",
    call_count=5,
)

call_count: int = 0#

event_type: str = 'tool_loop_block'#

pattern: str = ''#

severity: str = 'error'#

tool_name: str = ''#

class calute.audit.events.ToolLoopWarningEvent(timestamp: str = <factory>, agent_id: str | None = None, turn_id: str | None = None, session_id: str | None = None, severity: str = 'warning', metadata: dict[str, typing.Any] = <factory>, tool_name: str = '', pattern: str = '', severity_level: str = '', call_count: int = 0)[source]#

Bases: AuditEvent

Emitted when the loop detector identifies a potential tool-call loop.

This is a soft warning – the tool call is still allowed to proceed, but downstream consumers (dashboards, logs) are alerted that the agent may be stuck in a repetitive pattern.

event_type#

Fixed to "tool_loop_warning" (not settable via init).

Type: str

severity#

Defaults to "warning" (overrides the base "info").

Type: str

tool_name#

The registered name of the tool involved in the suspected loop.

Type: str

pattern#

A short description of the repetitive pattern that was detected (e.g. "same_args_3x").

Type: str

severity_level#

An additional severity qualifier supplied by the loop detector (e.g. "warning", "critical").

Type: str

call_count#

The number of consecutive or recent calls that matched the loop pattern.

Type: int

Example

event = ToolLoopWarningEvent(
    tool_name="web_search",
    pattern="same_args_3x",
    severity_level="warning",
    call_count=3,
)

call_count: int = 0#

event_type: str = 'tool_loop_warning'#

pattern: str = ''#

severity: str = 'warning'#

severity_level: str = ''#

tool_name: str = ''#

class calute.audit.events.ToolPolicyDecisionEvent(timestamp: str = <factory>, agent_id: str | None = None, turn_id: str | None = None, session_id: str | None = None, severity: str = 'info', metadata: dict[str, typing.Any] = <factory>, tool_name: str = '', action: str = '', policy_source: str = '')[source]#

Bases: AuditEvent

Emitted when the policy engine makes an allow/deny decision.

Records the outcome of the tool-policy evaluation step that occurs before a tool call is dispatched. This is critical for auditing which policy rules allowed or blocked a given tool invocation.

event_type#

Fixed to "tool_policy_decision" (not settable via init).

Type: str

tool_name#

The registered name of the tool under evaluation.

Type: str

action#

The policy verdict, typically "allow" or "deny".

Type: str

policy_source#

An identifier for the policy rule or configuration file that produced the decision (e.g. "default_policy" or "operator_override").

Type: str

Example

event = ToolPolicyDecisionEvent(
    tool_name="shell_exec",
    action="deny",
    policy_source="operator_override",
    agent_id="agent-1",
)

action: str = ''#

event_type: str = 'tool_policy_decision'#

policy_source: str = ''#

tool_name: str = ''#

class calute.audit.events.TurnEndEvent(timestamp: str = <factory>, agent_id: str | None = None, turn_id: str | None = None, session_id: str | None = None, severity: str = 'info', metadata: dict[str, typing.Any] = <factory>, content_preview: str = '', function_calls_count: int = 0)[source]#

Bases: AuditEvent

Emitted when an agent turn finishes.

Records the outcome of a completed turn, including a truncated preview of the assistant’s response and the total number of function calls that were executed during the turn.

event_type#

Fixed to "turn_end" (not settable via init).

Type: str

content_preview#

The first portion (up to 200 characters) of the assistant’s response content. May be empty when no textual content was produced.

Type: str

function_calls_count#

The total number of tool / function calls that were dispatched during this turn.

Type: int

Example

event = TurnEndEvent(
    agent_id="agent-1",
    turn_id="t-001",
    content_preview="The report shows ...",
    function_calls_count=3,
)

content_preview: str = ''#

event_type: str = 'turn_end'#

function_calls_count: int = 0#

class calute.audit.events.TurnStartEvent(timestamp: str = <factory>, agent_id: str | None = None, turn_id: str | None = None, session_id: str | None = None, severity: str = 'info', metadata: dict[str, typing.Any] = <factory>, prompt_preview: str = '')[source]#

Bases: AuditEvent

Emitted when a new agent turn begins.

Captures the opening of a conversational turn, including a truncated preview of the user prompt that initiated it.

event_type#

Fixed to "turn_start" (not settable via init).

Type: str

prompt_preview#

The first portion (up to 200 characters) of the user prompt that triggered this turn. May be empty if no prompt text was available.

Type: str

Example

event = TurnStartEvent(
    agent_id="agent-1",
    turn_id="t-001",
    prompt_preview="Summarize the latest report",
)

event_type: str = 'turn_start'#

prompt_preview: str = ''#

calute.audit.events

Contents

calute.audit.events#