calute.audit.emitter

calute.audit.emitter#

High-level audit emitter with convenience methods.

AuditEmitter wraps an AuditCollector and exposes one dedicated emit_* method for every event type defined in calute.audit.events. This keeps call-sites in the executor and response loop clean – callers never need to construct event dataclasses directly.

The emitter is fully thread-safe: an internal threading.Lock serializes all writes to the underlying collector, and an optional session_id is automatically stamped onto every event.

Example

from calute.audit import AuditEmitter, InMemoryCollector

collector = InMemoryCollector()
emitter = AuditEmitter(collector=collector, session_id="sess-01")
tid = emitter.emit_turn_start(agent_id="agent-1", prompt="Hello")
emitter.emit_tool_call_attempt("web_search", args='{"q":"hi"}', turn_id=tid)
emitter.emit_turn_end(agent_id="agent-1", turn_id=tid, fc_count=1)
emitter.flush()
class calute.audit.emitter.AuditEmitter(collector: calute.audit.collector.AuditCollector | calute.audit.collector.InMemoryCollector | None = None, session_id: str | None = None)[source]#

Bases: object

Thread-safe emitter that converts method calls into typed audit events.

The emitter provides a dedicated emit_* convenience method for each event type so that call-sites do not need to import or construct event dataclasses directly. Every event is automatically stamped with the emitter’s session_id before being forwarded to the underlying collector.

_collector#

The audit collector that receives forwarded events.

_session_id#

Optional session-level identifier stamped onto every event.

_lock#

Internal threading lock serializing writes.

Parameters

  • collector – The audit collector that receives all emitted events. Defaults to a new InMemoryCollector when None.

  • session_id – An optional session identifier that is stamped onto every event emitted through this instance.

Example

from calute.audit import AuditEmitter, InMemoryCollector

collector = InMemoryCollector()
emitter = AuditEmitter(collector=collector, session_id="s1")
tid = emitter.emit_turn_start(agent_id="a1", prompt="Hi")
emitter.emit_turn_end(agent_id="a1", turn_id=tid)
assert len(collector) == 2
property collector: Any#

Return the underlying audit collector instance.

Returns

The collector that this emitter forwards events to.

emit_error(error_type: str = '', error_msg: str = '', context: str = '', agent_id: str | None = None, turn_id: str | None = None) → None[source]#

Emit a generic ErrorEvent for non-tool-specific errors.

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

Parameters

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

  • error_msg – The human-readable error description or stringified exception message.

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

  • agent_id – Optional identifier of the agent that encountered the error.

  • turn_id – Optional turn identifier for event correlation.

emit_hook_mutation(hook_name: str, tool_name: str = '', agent_id: str | None = None, field: str = '', turn_id: str | None = None) → None[source]#

Emit a HookMutationEvent when a hook alters data.

Records which hook mutated which field of a tool call or its result, providing a full audit trail for data transformations applied outside the tool’s own logic.

Parameters

  • hook_name – The identifier of the hook that performed the mutation (e.g. "sanitize_output").

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

  • agent_id – Optional identifier of the agent whose pipeline includes the hook.

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

  • turn_id – Optional turn identifier for event correlation.

emit_sandbox_decision(tool_name: str, context: str = '', reason: str = '', backend_type: str = '', agent_id: str | None = None, turn_id: str | None = None) → None[source]#

Emit a SandboxDecisionEvent for a sandbox routing choice.

Records which execution backend the sandbox router selected for a given tool call, along with the context and reasoning behind the decision.

Parameters

  • tool_name – The registered name of the tool being routed.

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

  • reason – A human-readable explanation of why the particular backend was chosen.

  • backend_type – The identifier of the selected sandbox backend (e.g. "local", "docker", "subprocess").

  • agent_id – Optional identifier of the agent whose tool call is being routed.

  • turn_id – Optional turn identifier for event correlation.

emit_tool_call_attempt(tool_name: str, args: str = '', agent_id: str | None = None, turn_id: str | None = None) → None[source]#

Emit a ToolCallAttemptEvent before a tool is dispatched.

Should be called immediately before the tool executor runs the tool, so that the audit trail records the intent even if the tool subsequently fails.

Parameters

  • tool_name – The registered name of the tool about to be invoked.

  • args – A string representation of the tool arguments. Only the first 200 characters are stored in the event.

  • agent_id – Optional identifier of the agent making the call.

  • turn_id – Optional turn identifier for event correlation.

emit_tool_call_complete(tool_name: str, status: str = 'success', duration_ms: float = 0.0, result: str = '', agent_id: str | None = None, turn_id: str | None = None) → None[source]#

Emit a ToolCallCompleteEvent after a tool returns.

Should be called immediately after a successful tool execution. The result text is truncated to 200 characters for the result_preview field.

Parameters

  • tool_name – The registered name of the tool that completed.

  • status – Completion status string, typically "success".

  • duration_ms – Wall-clock execution time of the tool in milliseconds.

  • result – The full string representation of the tool result. Only the first 200 characters are stored.

  • agent_id – Optional identifier of the agent that invoked the tool.

  • turn_id – Optional turn identifier for event correlation.

emit_tool_call_failure(tool_name: str, error_type: str = '', error_msg: str = '', agent_id: str | None = None, turn_id: str | None = None) → None[source]#

Emit a ToolCallFailureEvent when a tool raises an error.

This should be called in the exception handler of the tool executor, capturing the exception class name and message for the audit trail.

Parameters

  • tool_name – The registered name of the tool that failed.

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

  • error_msg – The human-readable error description or stringified exception message.

  • agent_id – Optional identifier of the agent that invoked the tool.

  • turn_id – Optional turn identifier for event correlation.

emit_tool_loop_block(tool_name: str, pattern: str = '', count: int = 0, agent_id: str | None = None, turn_id: str | None = None) → None[source]#

Emit a ToolLoopBlockEvent when a loop causes a hard block.

Unlike emit_tool_loop_warning(), this indicates that the loop detector has prevented the tool call from executing.

Parameters

  • tool_name – The registered name of the tool that was blocked.

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

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

  • agent_id – Optional identifier of the agent exhibiting the loop behavior.

  • turn_id – Optional turn identifier for event correlation.

emit_tool_loop_warning(tool_name: str, pattern: str = '', severity: str = 'warning', count: int = 0, agent_id: str | None = None, turn_id: str | None = None) → None[source]#

Emit a ToolLoopWarningEvent for a suspected call loop.

This is a soft warning – the tool call is still dispatched, but the event alerts downstream consumers that the agent may be stuck in a repetitive invocation pattern.

Parameters

  • tool_name – The registered name of the tool involved in the suspected loop.

  • pattern – A short descriptor of the detected repetitive pattern (e.g. "same_args_3x").

  • severity – The severity qualifier assigned by the loop detector (e.g. "warning", "critical").

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

  • agent_id – Optional identifier of the agent exhibiting the loop behavior.

  • turn_id – Optional turn identifier for event correlation.

emit_tool_policy_decision(tool_name: str, agent_id: str | None = None, action: str = '', source: str = '', turn_id: str | None = None) → None[source]#

Emit a ToolPolicyDecisionEvent for a policy verdict.

Records whether the tool-policy engine allowed or denied a given tool invocation, along with the policy rule that produced the decision.

Parameters

  • tool_name – The registered name of the tool under evaluation.

  • agent_id – Optional identifier of the agent requesting the tool call.

  • action – The policy verdict, typically "allow" or "deny".

  • source – An identifier for the policy rule or configuration that produced the decision.

  • turn_id – Optional turn identifier for event correlation.

emit_turn_end(agent_id: str | None = None, turn_id: str | None = None, content: str = '', fc_count: int = 0) → None[source]#

Emit a TurnEndEvent recording turn completion details.

The assistant’s response content is truncated to 200 characters for the content_preview field.

Parameters

  • agent_id – Optional identifier of the agent that completed the turn.

  • turn_id – Optional turn identifier correlating this end event with its corresponding TurnStartEvent.

  • content – The full assistant response text. Only the first 200 characters are stored in the event.

  • fc_count – The total number of function / tool calls that were executed during this turn.

emit_turn_start(agent_id: str | None = None, turn_id: str | None = None, prompt: str = '') → str[source]#

Emit a TurnStartEvent and return the turn identifier.

If no turn_id is provided, a new random 12-hex-character ID is generated via _generate_turn_id(). The prompt is truncated to 200 characters for the prompt_preview field.

Parameters

  • agent_id – Optional identifier of the agent starting the turn.

  • turn_id – Optional pre-assigned turn identifier. When None, a new one is generated automatically.

  • prompt – The full user prompt text. Only the first 200 characters are stored in the event.

Returns

The turn identifier (either the provided turn_id

or the auto-generated one).

Return type

str

flush() → None[source]#

Flush the underlying collector’s buffered state.

Acquires the internal lock and delegates to the collector’s flush() method, ensuring that all previously emitted events have been fully persisted or transmitted.

property session_id: str | None#

Return the session identifier stamped onto every event.

Returns

The session ID, or None if not set.

Return type

str | None

Contents