core.state_machine module

Atomic Redis-backed lifecycle accounting for one inbound Stargazer turn.

The lifecycle is observability/accounting state, while TurnCheckpointSession remains the execution-recovery authority. Every mutation in this module is a single Lua compare-and-set using Redis server time. Trace, history, checkpoint, active-index, and inbound stream keys are colocated with the anchor stream so the same operations remain legal under Redis Cluster.

class core.state_machine.OperationStateMachine(redis, *, anchor_stream='sg:stream:inbound', active_ttl_seconds=691200, terminal_ttl_seconds=691200, watchdog_timeout_ms=1800000)

Bases: object

Strict lifecycle mutations for one Redis/anchor-stream domain.

Parameters:

redis (_AsyncRedis)
anchor_stream (str)
active_ttl_seconds (int)
terminal_ttl_seconds (int)
watchdog_timeout_ms (int)

property anchor_stream: str

property active_key: str

classmethod is_valid_transition(current, target)

Return true for a legal edge or an idempotent same-state replay.

Return type:

bool

Parameters:

current (str)
target (str)

keys_for(trace_id)

Return type:: TraceKeys
Parameters:: trace_id (str)

async create(trace_id, initial_state=TraceState.RECEIVED, metadata=None, *, input_fingerprint='', operation_kind='inbound_turn', stream_name='', stream_msg_id='', proof_deadline_ms=0, transition_id=None, actor='trace:create')

Return type:

TransitionResult

Parameters:

trace_id (str)
initial_state (str | TraceState)
metadata (Mapping[str, Any] | None)
input_fingerprint (str)
operation_kind (str)
stream_name (str)
stream_msg_id (str)
proof_deadline_ms (int)
transition_id (str | None)
actor (str)

async ensure_queued(trace_id, metadata=None, *, input_fingerprint='', operation_kind='inbound_turn', stream_name=None, stream_msg_id='', proof_deadline_ms=0, transition_id=None, actor='trace:queue')

Return type:

TransitionResult

Parameters:

trace_id (str)
metadata (Mapping[str, Any] | None)
input_fingerprint (str)
operation_kind (str)
stream_name (str | None)
stream_msg_id (str)
proof_deadline_ms (int)
transition_id (str | None)
actor (str)

async transition_detailed(trace_id, target_state, metadata=None, *, transition_id=None, transition_fingerprint_value=None, actor='trace:transition', expected_revision=None, terminal_reason='')

Return type:

TransitionResult

Parameters:

trace_id (str)
target_state (str | TraceState)
metadata (Mapping[str, Any] | None)
transition_id (str | None)
transition_fingerprint_value (str | None)
actor (str)
expected_revision (int | None)
terminal_reason (str)

async transition(trace_id, target_state, metadata=None, **kwargs)

Compatibility wrapper preserving the former awaited boolean API.

Return type:

bool

Parameters:

trace_id (str)
target_state (str | TraceState)
metadata (Mapping[str, Any] | None)
kwargs (Any)

async heartbeat(trace_id, metadata=None, *, expected_revision=None, actor='trace:heartbeat')

Return type:

TransitionResult

Parameters:

trace_id (str)
metadata (Mapping[str, Any] | None)
expected_revision (int | None)
actor (str)

async abandon_if_due(trace_id, *, expected_revision, source_group, observed_state=None, reason='', actor='trace:watchdog')

Return type:

TransitionResult

Parameters:

trace_id (str)
expected_revision (int)
source_group (str)
observed_state (str | TraceState | None)
reason (str)
actor (str)

async checkpoint_and_transition(trace_id, gather_output, target_state, *, transition_id=None, actor='trace:checkpoint', expected_revision=None, checkpoint_purpose='', checkpoint_binding='')

Return type:

TransitionResult

Parameters:

trace_id (str)
gather_output (Mapping[str, Any])
target_state (str | TraceState)
transition_id (str | None)
actor (str)
expected_revision (int | None)
checkpoint_purpose (str)
checkpoint_binding (str)

async terminal_ack(trace_id, source_stream, group, msg_id, target_state=TraceState.COMPLETED, metadata=None, *, transition_id=None, actor='trace:terminal-ack', expected_revision=None, completion_proof='', terminal_reason='')

Return type:

TransitionResult

Parameters:

trace_id (str)
source_stream (str)
group (str)
msg_id (str)
target_state (str | TraceState)
metadata (Mapping[str, Any] | None)
transition_id (str | None)
actor (str)
expected_revision (int | None)
completion_proof (str)
terminal_reason (str)

async dead_letter_ack(trace_id, source_stream, group, msg_id, raw_payload, error, target_state=TraceState.DEAD_LETTERED, metadata=None, *, transition_id=None, actor='trace:dead-letter-ack', expected_revision=None, terminal_reason='')

Durably enqueue one DLQ event, ACK its source, and retire the trace.

The source stream, lifecycle keys, and lane-local outbox share one Redis Cluster slot. The Lua operation stores the generated outbox ID on the trace, making connection-loss retries return the original event instead of appending a duplicate.

Return type:

TransitionResult

Parameters:

trace_id (str)
source_stream (str)
group (str)
msg_id (str)
raw_payload (bytes)
error (str)
target_state (str | TraceState)
metadata (Mapping[str, Any] | None)
transition_id (str | None)
actor (str)
expected_revision (int | None)
terminal_reason (str)

async terminal_reack(trace_id, source_stream, group, msg_id, target_state, *, actor='trace:terminal-reack')

Repair a staged terminal outbox row and repeat only its source ACK.

Return type:

TransitionResult

Parameters:

trace_id (str)
source_stream (str)
group (str)
msg_id (str)
target_state (str | TraceState)
actor (str)

async get_record(trace_id)

Return type:: dict[str, Any] | None
Parameters:: trace_id (str)

class core.state_machine.TraceKeys(trace, history, checkpoint, dead_letter_stage, active, identity_hash, anchor_stream)

Bases: object

Parameters:

trace (str)
history (str)
checkpoint (str)
dead_letter_stage (str)
active (str)
identity_hash (str)
anchor_stream (str)

trace: str

history: str

checkpoint: str

dead_letter_stage: str

active: str

identity_hash: str

anchor_stream: str

exception core.state_machine.TraceLifecycleError

Bases: RuntimeError

The Redis lifecycle script returned malformed or unsupported data.

class core.state_machine.TraceState(*values)

Bases: str, Enum

RECEIVED = 'RECEIVED'

PREPROCESSING = 'PREPROCESSING'

QUEUED = 'QUEUED'

CLAIMED = 'CLAIMED'

RECLAIMED = 'RECLAIMED'

PROCESSING = 'PROCESSING'

PRE_INFERENCE_GATHER = 'PRE_INFERENCE_GATHER'

INFERRING = 'INFERRING'

TOOL_EXECUTING = 'TOOL_EXECUTING'

POSTPROCESSING = 'POSTPROCESSING'

DELIVERING = 'DELIVERING'

DELIVERED = 'DELIVERED'

DEFERRED = 'DEFERRED'

CANCEL_REQUESTED = 'CANCEL_REQUESTED'

ERRORED = 'ERRORED'

RETRYING = 'RETRYING'

COMPLETED = 'COMPLETED'

DEAD_LETTERED = 'DEAD_LETTERED'

ABANDONED = 'ABANDONED'

AMBIGUOUS = 'AMBIGUOUS'

class core.state_machine.TransitionAction(*values)

Bases: str, Enum

CREATED = 'CREATED'

APPLIED = 'APPLIED'

CACHED = 'CACHED'

ALREADY_STATE = 'ALREADY_STATE'

HEARTBEAT = 'HEARTBEAT'

NOT_DUE = 'NOT_DUE'

PROOF_LIVE = 'PROOF_LIVE'

MISSING = 'MISSING'

CONFLICT = 'CONFLICT'

INVALID_STATE = 'INVALID_STATE'

INVALID_PREDECESSOR = 'INVALID_PREDECESSOR'

INVALID_ARGUMENT = 'INVALID_ARGUMENT'

INVALID_TTL = 'INVALID_TTL'

METADATA_TOO_LARGE = 'METADATA_TOO_LARGE'

WRONGTYPE = 'WRONGTYPE'

SCHEMA_CONFLICT = 'SCHEMA_CONFLICT'

TERMINAL = 'TERMINAL'

STALE_REVISION = 'STALE_REVISION'

EXPIRED = 'EXPIRED'

ACK_REQUIRED = 'ACK_REQUIRED'

BINDING_MISSING = 'BINDING_MISSING'

BINDING_CONFLICT = 'BINDING_CONFLICT'

class core.state_machine.TransitionResult(action, previous_state, state, revision, history_id='', acknowledged=0, detail='')

Bases: object

Parameters:

action (TransitionAction)
previous_state (TraceState | None)
state (TraceState | None)
revision (int)
history_id (str)
acknowledged (int)
detail (str)

action: TransitionAction

previous_state: TraceState | None

state: TraceState | None

revision: int

history_id: str

acknowledged: int

detail: str

property ok: bool

property terminal: bool

core.state_machine.attempt_counter_key(anchor_stream='sg:stream:inbound')

Return the per-message reclaim counter hash colocated with the anchor.

Return type:: str
Parameters:: anchor_stream (str)

core.state_machine.canonical_metadata_json(metadata)

Encode bounded audit metadata without exposing reserved hash fields.

Return type:: str
Parameters:: metadata (Mapping[str, Any] | None)

core.state_machine.dead_letter_outbox_key(anchor_stream='sg:stream:inbound')

Return the durable DLQ outbox stream colocated with anchor_stream.

Return type:: str
Parameters:: anchor_stream (str)

core.state_machine.ingress_transition_id(identity_hash, target_state)

Return the deterministic first-writer ingress transition identity.

Return type:

str

Parameters:

identity_hash (str)
target_state (str | TraceState)

async core.state_machine.recover_checkpoint(redis, trace_id, *, anchor_stream='sg:stream:inbound')

Recover a current, valid compressed gather checkpoint.

The v2 colocated key is authoritative. During migration the former sg:checkpoint:{trace_id} key is read only when no v2 record exists. Corrupt, stale, or future-schema records fail closed and require regather.

Return type:

dict[str, Any] | None

Parameters:

redis (_AsyncRedis)
trace_id (str)
anchor_stream (str)

core.state_machine.trace_keys(trace_id, anchor_stream='sg:stream:inbound')

Return the v2 lifecycle keys colocated with anchor_stream.

Return type:

TraceKeys

Parameters:

trace_id (str)
anchor_stream (str)

core.state_machine.trace_keys_from_identity(identity_hash, *, anchor_stream='sg:stream:inbound')

Return type:

TraceKeys

Parameters:

identity_hash (str)
anchor_stream (str)

core.state_machine.transition_fingerprint(trace_id, from_state, target_state, actor, metadata_json, input_fingerprint='')

Hash one immutable lifecycle transition request.

Return type:

str

Parameters:

trace_id (str)
from_state (str | TraceState | None)
target_state (str | TraceState)
actor (str)
metadata_json (str)
input_fingerprint (str)

async core.state_machine.write_checkpoint_and_transition(redis, trace_id, gather_output, target_state, script_sha=None)

Compatibility wrapper around the embedded atomic checkpoint operation.

script_sha is accepted for source compatibility but intentionally unused; the canonical v2 script is loaded from scripts/state_transition.lua.

Return type:

TransitionResult

Parameters:

redis (_AsyncRedis)
trace_id (str)
gather_output (Mapping[str, Any])
target_state (str | TraceState)
script_sha (str | None)