core.turn_checkpoint module

Durable, atomic whole-turn and tool-call checkpoints.

The inbound and outbound Redis Streams are at-least-once transports. This module journals the canonical model transcript and every logical tool effect in one Redis Cluster-safe hash so a reclaimed turn resumes instead of regenerating provider decisions or repeating effects.

Every mutation is one single-key Lua EVAL. Large or structured values are msgpack encoded, zlib compressed, and base64 framed before entering the hash. The readable scalar fields exist only for Lua compare-and-set transitions.

class core.turn_checkpoint.ToolCallAction(*values)

Bases: str, Enum

Outcome of a tool-slot claim or state transition.

ACQUIRED = 'ACQUIRED'
RECLAIMED = 'RECLAIMED'
BUSY = 'BUSY'
BEGUN = 'BEGUN'
ALREADY_EXECUTING = 'ALREADY_EXECUTING'
RENEWED = 'RENEWED'
ABANDONED = 'ABANDONED'
COMPLETED = 'COMPLETED'
CACHED = 'CACHED'
AMBIGUOUS = 'AMBIGUOUS'
CONFLICT = 'CONFLICT'
STALE_OWNER = 'STALE_OWNER'
LEASE_EXPIRED = 'LEASE_EXPIRED'
INVALID_STATE = 'INVALID_STATE'
MISSING = 'MISSING'
SCHEMA_CONFLICT = 'SCHEMA_CONFLICT'
class core.turn_checkpoint.ToolCallDecision(action, state, round_index, slot_index, operation_id, provider_call_id, tool_name, raw_arguments, owner_token, fence, replay_safe, result=None, payload_fingerprint='')

Bases: object

Canonical state observed after a tool-slot operation.

Parameters:
action: ToolCallAction
state: ToolCallState | None
round_index: int
slot_index: int
operation_id: str
provider_call_id: str
tool_name: str
raw_arguments: Any
owner_token: str | None
fence: int
replay_safe: bool
result: dict[str, Any] | None = None
payload_fingerprint: str = ''
property should_begin: bool

True only when this decision authorizes the external effect.

property should_execute: bool

Alias for integrations that name the effect barrier explicitly.

property terminal: bool
property cached_result: dict[str, Any] | None
class core.turn_checkpoint.ToolCallState(*values)

Bases: str, Enum

Durable states for one logical tool slot.

CLAIMED = 'CLAIMED'
EXECUTING = 'EXECUTING'
COMPLETED = 'COMPLETED'
AMBIGUOUS = 'AMBIGUOUS'
exception core.turn_checkpoint.TurnCheckpointBusy

Bases: TurnCheckpointError

Another fenced worker currently owns the logical turn or tool call.

exception core.turn_checkpoint.TurnCheckpointConflict

Bases: TurnCheckpointError

The logical turn/slot already contains different canonical input.

exception core.turn_checkpoint.TurnCheckpointError

Bases: RuntimeError

Base class for an unusable or invalid durable turn checkpoint.

class core.turn_checkpoint.TurnCheckpointSession(redis, trace_id, scope='primary', ttl_seconds=604800, claim_lease_seconds=30, execution_lease_seconds=900, proof_deadline_ms=0, lifecycle=None)

Bases: object

Redis-backed journal for one trace/scope pair.

Parameters:

  • redis (_AsyncRedis)

  • trace_id (str)

  • scope (str)

  • ttl_seconds (int)

  • claim_lease_seconds (int)

  • execution_lease_seconds (int)

  • proof_deadline_ms (int)

  • lifecycle (Any | None)

property scope: str
property key: str
property redis: _AsyncRedis

Binary-safe Redis client shared by checkpoint sidecar stores.

property trace_hash: str
property execution_lease_seconds: int
property ttl_seconds: int
async load()

Load the canonical turn snapshot, or None before start.

Return type:

TurnSnapshot | None

async reconcile_lifecycle(snapshot=None)

Project the durable primary-turn phase into the transport lifecycle.

The turn checkpoint remains the recovery authority. Lifecycle writes happen only after the corresponding checkpoint mutation commits, so a crash can leave the projection behind but can never move it ahead of durable proof. Replaying this method is safe because lifecycle transitions are predecessor-checked and same-state idempotent.

COMPLETED checkpoint state remains projected at DELIVERED rather than moving the lifecycle directly to a terminal state. The inbound consumer owns the final single-slot Lua operation that combines XACK with the COMPLETED transition and active-index removal.

Return type:

TurnSnapshot | None

Parameters:

snapshot (TurnSnapshot | None)

async start(messages, tool_names, runtime)

Create the canonical base transcript, or recover the first writer.

Return type:

TurnSnapshot

Parameters:
async reserve_assistant(round_index, response)

Persist a provider decision before any effect and return canonical data.

Return type:

dict[str, Any]

Parameters:
async get_assistant(round_index)

Return a previously reserved provider decision without mutation.

Return type:

dict[str, Any] | None

Parameters:

round_index (int)

logical_operation_id(round_index, slot_index)

Derive a stable effect ID using no provider-generated identity.

Return type:

str

Parameters:

  • round_index (int)

  • slot_index (int)

async claim_tool(round_index, slot_index, provider_call_id, tool_name, raw_arguments, operation_id, replay_safe=False, owner_token=None)

Claim a logical tool slot, reclaiming only under safe lease rules.

Return type:

ToolCallDecision

Parameters:

  • round_index (int)

  • slot_index (int)

  • provider_call_id (str)

  • tool_name (str)

  • raw_arguments (Any)

  • operation_id (str)

  • replay_safe (bool)

  • owner_token (str | None)

async begin_tool(round_index, slot_index, owner_token, fence)

Cross the effect barrier for the current fenced owner.

Return type:

ToolCallDecision

Parameters:

  • round_index (int)

  • slot_index (int)

  • owner_token (str)

  • fence (int)

async renew_tool(round_index, slot_index, owner_token, fence)

Renew an executing slot while preserving its owner/fence.

Return type:

ToolCallDecision

Parameters:

  • round_index (int)

  • slot_index (int)

  • owner_token (str)

  • fence (int)

async abandon_tool(round_index, slot_index, owner_token, fence)

Release an executing slot whose call provably caused no effects.

Only the live fenced owner may release, and only from EXECUTING. The slot returns to CLAIMED with an expired lease so the next attempt reclaims it through the ordinary fenced path. Callers must be certain no effect was possible — a failure that merely might have reached the provider has to keep the fail-closed lease-expiry route to AMBIGUOUS.

Return type:

ToolCallDecision

Parameters:

  • round_index (int)

  • slot_index (int)

  • owner_token (str)

  • fence (int)

async complete_tool(round_index, slot_index, owner_token, fence, result_record)

Save the exact model-facing/audit result before inference resumes.

Return type:

ToolCallDecision

Parameters:
async get_tool(round_index, slot_index)

Read a logical tool slot without changing leases or ownership.

Return type:

ToolCallDecision | None

Parameters:

  • round_index (int)

  • slot_index (int)

async commit_round(round_index, messages, runtime)

Commit the post-tool transcript/runtime and return canonical state.

Return type:

TurnSnapshot

Parameters:
async save_final_model(content)

First-writer reserve the terminal provider content.

Return type:

str

Parameters:

content (str)

async reserve_final_delivery(reply_message_key, text, delivery=None)

Reserve delivery identity, exact text, and canonical presentation.

Return type:

TurnSnapshot

Parameters:
async mark_delivered(reply_message_key, message_id)

Record successful deterministic delivery using the reserved key.

Return type:

TurnSnapshot

Parameters:

  • reply_message_key (str)

  • message_id (str)

async mark_completed(*, allow_no_output=False)

Close a turn after delivery and all tool/effect slots are terminal.

Return type:

TurnSnapshot

Parameters:

allow_no_output (bool)

class core.turn_checkpoint.TurnPhase(*values)

Bases: str, Enum

Durable progress phases for a whole model turn.

STARTED = 'STARTED'
ASSISTANT_RESERVED = 'ASSISTANT_RESERVED'
ROUND_COMMITTED = 'ROUND_COMMITTED'
FINAL_MODEL_SAVED = 'FINAL_MODEL_SAVED'
DELIVERY_RESERVED = 'DELIVERY_RESERVED'
DELIVERED = 'DELIVERED'
COMPLETED = 'COMPLETED'
class core.turn_checkpoint.TurnSnapshot(phase, messages, tool_names, runtime, final_model=None, final_text=None, final_delivery=None, reply_message_key=None, sent_message_id=None, completed=False, current_round=-1, revision=0, input_fingerprint='', assistant_responses=<factory>)

Bases: object

Decoded canonical whole-turn state returned to inference recovery.

Parameters:
phase: TurnPhase
messages: list[Any]
tool_names: list[str] | None
runtime: dict[str, Any]
final_model: str | None = None
final_text: str | None = None
final_delivery: dict[str, Any] | None = None
reply_message_key: str | None = None
sent_message_id: str | None = None
completed: bool = False
current_round: int = -1
revision: int = 0
input_fingerprint: str = ''
assistant_responses: dict[int, dict[str, Any]]