calute.session.models

Contents

calute.session.models#

Data models for session persistence and replay.

Provides serializable dataclasses for recording session turns, tool calls, agent transitions, and full session records.

class calute.session.models.AgentTransitionRecord(from_agent: str | None, to_agent: str, reason: str | None = None, turn_id: str = '', timestamp: str = '')[source]#

Bases: object

Record of a transition between agents.

from_agent#

ID of the agent being switched away from.

Type

str | None

to_agent#

ID of the agent being switched to.

Type

str

reason#

Human-readable reason for the transition.

Type

str | None

turn_id#

ID of the turn during which the transition occurred.

Type

str

timestamp#

ISO 8601 timestamp of the transition.

Type

str

from_agent: str | None#
classmethod from_dict(data: dict[str, Any]) AgentTransitionRecord[source]#

Deserialize an AgentTransitionRecord from a plain dictionary.

Parameters

data – A dictionary containing agent transition fields. Must include to_agent at minimum.

Returns

A new AgentTransitionRecord instance populated from data.

Raises

KeyError – If the required key to_agent is missing.

Example

>>> data = {"from_agent": "a", "to_agent": "b"}
>>> rec = AgentTransitionRecord.from_dict(data)
>>> rec.to_agent
'b'
reason: str | None = None#
timestamp: str = ''#
to_agent: str#
to_dict() dict[str, Any][source]#

Serialize the agent transition record to a JSON-compatible dictionary.

Returns

A dictionary containing all agent transition record fields with JSON-serializable values.

Example

>>> transition = AgentTransitionRecord(
...     from_agent="agent-a", to_agent="agent-b", reason="escalation"
... )
>>> data = transition.to_dict()
>>> data["to_agent"]
'agent-b'
turn_id: str = ''#
calute.session.models.SessionId#

A distinct string type representing a unique session identifier.

alias of str

class calute.session.models.SessionRecord(session_id: str, workspace_id: str | None = None, created_at: str = '', updated_at: str = '', agent_id: str | None = None, turns: list[calute.session.models.TurnRecord] = <factory>, agent_transitions: list[calute.session.models.AgentTransitionRecord] = <factory>, metadata: dict[str, typing.Any] = <factory>)[source]#

Bases: object

Complete record of a session.

session_id#

Unique identifier for this session.

Type

str

workspace_id#

Identifier for the workspace this session belongs to.

Type

str | None

created_at#

ISO 8601 timestamp when the session was created.

Type

str

updated_at#

ISO 8601 timestamp when the session was last updated.

Type

str

agent_id#

Initial/primary agent for the session.

Type

str | None

turns#

Ordered list of turn records.

Type

list[calute.session.models.TurnRecord]

agent_transitions#

List of agent transition events.

Type

list[calute.session.models.AgentTransitionRecord]

metadata#

Arbitrary metadata for extensibility.

Type

dict[str, Any]

agent_id: str | None = None#
agent_transitions: list[calute.session.models.AgentTransitionRecord]#
created_at: str = ''#
classmethod from_dict(data: dict[str, Any]) SessionRecord[source]#

Deserialize a SessionRecord from a plain dictionary.

Reconstructs a complete SessionRecord including all nested TurnRecord and AgentTransitionRecord objects from a dictionary previously produced by to_dict() or any compatible mapping.

Parameters

data – A dictionary containing session record fields. Must include session_id at minimum.

Returns

A new SessionRecord instance populated from data.

Raises

KeyError – If the required key session_id is missing.

Example

>>> data = {"session_id": "sess-1", "turns": [], "agent_transitions": []}
>>> session = SessionRecord.from_dict(data)
>>> session.session_id
'sess-1'
metadata: dict[str, Any]#
session_id: str#
to_dict() dict[str, Any][source]#

Serialize the session record to a JSON-compatible dictionary.

Nested TurnRecord and AgentTransitionRecord instances are recursively serialized via their own to_dict methods. Mutable containers are shallow-copied.

Returns

A dictionary containing all session record fields with JSON-serializable values.

Example

>>> session = SessionRecord(session_id="sess-1")
>>> data = session.to_dict()
>>> data["session_id"]
'sess-1'
turns: list[calute.session.models.TurnRecord]#
updated_at: str = ''#
workspace_id: str | None = None#
class calute.session.models.ToolCallRecord(call_id: str, tool_name: str, arguments: dict[str, typing.Any], result: ~typing.Any = None, status: str = 'success', error: str | None = None, duration_ms: float | None = None, sandbox_context: str | None = None, metadata: dict[str, typing.Any] = <factory>)[source]#

Bases: object

Record of a single tool/function call within a turn.

call_id#

Unique identifier for this tool call.

Type

str

tool_name#

Name of the tool/function invoked.

Type

str

arguments#

Arguments passed to the tool.

Type

dict[str, Any]

result#

Result returned by the tool.

Type

Any

status#

Execution status (success, error, timeout, skipped).

Type

str

error#

Error message if the call failed.

Type

str | None

duration_ms#

Execution duration in milliseconds.

Type

float | None

sandbox_context#

Sandbox context identifier, if applicable.

Type

str | None

arguments: dict[str, Any]#
call_id: str#
duration_ms: float | None = None#
error: str | None = None#
classmethod from_dict(data: dict[str, Any]) ToolCallRecord[source]#

Deserialize a ToolCallRecord from a plain dictionary.

Reconstructs a ToolCallRecord instance from a dictionary previously produced by to_dict() or any compatible mapping. Missing optional keys fall back to sensible defaults.

Parameters

data – A dictionary containing tool call record fields. Must include call_id and tool_name at minimum.

Returns

A new ToolCallRecord instance populated from data.

Raises

KeyError – If required keys (call_id, tool_name) are missing.

Example

>>> data = {"call_id": "tc-1", "tool_name": "search", "arguments": {}}
>>> record = ToolCallRecord.from_dict(data)
>>> record.tool_name
'search'
metadata: dict[str, Any]#
result: Any = None#
sandbox_context: str | None = None#
status: str = 'success'#
to_dict() dict[str, Any][source]#

Serialize the tool call record to a JSON-compatible dictionary.

Converts all fields of this record into a plain dictionary suitable for JSON serialization. Mutable containers (e.g., metadata) are shallow-copied to prevent unintended mutation of the original record.

Returns

A dictionary containing all tool call record fields with JSON-serializable values.

Example

>>> record = ToolCallRecord(
...     call_id="tc-1", tool_name="search", arguments={"q": "hello"}
... )
>>> data = record.to_dict()
>>> data["tool_name"]
'search'
tool_name: str#
class calute.session.models.TurnRecord(turn_id: str, agent_id: str | None = None, prompt: str = '', response_content: str | None = None, tool_calls: list[calute.session.models.ToolCallRecord] = <factory>, started_at: str = '', ended_at: str | None = None, status: str = 'success', error: str | None = None, audit_event_ids: list[str] = <factory>, metadata: dict[str, typing.Any] = <factory>)[source]#

Bases: object

Record of a single conversational turn.

turn_id#

Unique identifier for this turn.

Type

str

agent_id#

Identifier of the agent handling the turn.

Type

str | None

prompt#

The user prompt for this turn.

Type

str

response_content#

The agent’s text response content.

Type

str | None

tool_calls#

List of tool calls made during this turn.

Type

list[calute.session.models.ToolCallRecord]

started_at#

ISO 8601 timestamp when the turn started.

Type

str

ended_at#

ISO 8601 timestamp when the turn ended.

Type

str | None

status#

Turn outcome (success, error, cancelled).

Type

str

error#

Error message if the turn failed.

Type

str | None

audit_event_ids#

References to audit/logging events.

Type

list[str]

metadata#

Arbitrary metadata for extensibility.

Type

dict[str, Any]

agent_id: str | None = None#
audit_event_ids: list[str]#
ended_at: str | None = None#
error: str | None = None#
classmethod from_dict(data: dict[str, Any]) TurnRecord[source]#

Deserialize a TurnRecord from a plain dictionary.

Reconstructs a TurnRecord instance, including nested ToolCallRecord objects, from a dictionary previously produced by to_dict() or any compatible mapping.

Parameters

data – A dictionary containing turn record fields. Must include turn_id at minimum.

Returns

A new TurnRecord instance populated from data.

Raises

KeyError – If the required key turn_id is missing.

Example

>>> data = {"turn_id": "t-1", "prompt": "Hello"}
>>> turn = TurnRecord.from_dict(data)
>>> turn.prompt
'Hello'
metadata: dict[str, Any]#
prompt: str = ''#
response_content: str | None = None#
started_at: str = ''#
status: str = 'success'#
to_dict() dict[str, Any][source]#

Serialize the turn record to a JSON-compatible dictionary.

Nested ToolCallRecord instances are recursively serialized via their own to_dict methods. Mutable containers are shallow-copied.

Returns

A dictionary containing all turn record fields with JSON-serializable values.

Example

>>> turn = TurnRecord(turn_id="t-1", prompt="Hello")
>>> data = turn.to_dict()
>>> data["prompt"]
'Hello'
tool_calls: list[calute.session.models.ToolCallRecord]#
turn_id: str#
calute.session.models.WorkspaceId#

A distinct string type representing a unique workspace identifier.

alias of str