core.gateway_rpc_ledger module

Redis-backed effect ledger for gateway RPC operations.

The outbound Redis Stream is an at-least-once transport. This module supplies the state machine needed to make an effectful gateway RPC safe at the delivery boundary: a logical operation is claimed, explicitly moved to EXECUTING, and then completed or failed. A worker may reclaim a stale CLAIMED entry, but an expired EXECUTING entry is permanently marked AMBIGUOUS because Redis cannot determine whether an external platform effect happened.

All state transitions are single-key Lua scripts. The operation identifier is SHA-256 hashed before it is used in a Redis key, and the hash is also used as a Redis Cluster hash tag. Callers must execute the external effect only when begin returns LedgerAction.BEGUN.

class core.gateway_rpc_ledger.GatewayRpcLedger(redis, *, namespace='sg:gateway:rpc', terminal_ttl_seconds=691200, claim_lease_seconds=30, execution_lease_seconds=300)

Bases: object

Atomic Redis effect ledger for at-least-once gateway RPC delivery.

owner_token must uniquely identify one processing attempt and must be reused only when retrying an unobserved response from the same ledger call. The returned fence must be supplied to begin and the terminal method. Execute the external effect only after begin(...).should_execute is true.

Parameters:

  • redis (_AsyncRedis)

  • namespace (str)

  • terminal_ttl_seconds (int)

  • claim_lease_seconds (int)

  • execution_lease_seconds (int)

static fingerprint(payload)

Expose canonical payload fingerprinting for envelope construction.

Return type:

str

Parameters:

payload (Any)

static operation_hash(operation_id)

Expose the non-reversible Redis-key suffix for observability.

Return type:

str

Parameters:

operation_id (str)

key_for(operation_id)

Return the cluster-safe Redis hash key for an operation.

Return type:

str

Parameters:

operation_id (str)

static reply_key_for(call_id)

Return the default caller-safe reply key for a logical call id.

Return type:

str

Parameters:

call_id (str)

async claim(call_id, fingerprint, owner_token=None, *, proof_deadline_ms=0, require_existing=False)

Claim work, optionally refusing to create a missing operation.

require_existing is used by rolling-upgrade recovery paths: a request bound to an older catalog may resume or replay an existing operation, but it must never create fresh work under that stale catalog generation.

Return type:

LedgerResult

Parameters:

  • call_id (str)

  • fingerprint (str)

  • owner_token (str | None)

  • proof_deadline_ms (int)

  • require_existing (bool)

async begin(call_id, fingerprint, owner_token, fence)

Move a valid claim to EXECUTING and authorize one effect attempt.

Return type:

LedgerResult

Parameters:

  • call_id (str)

  • fingerprint (str)

  • owner_token (str)

  • fence (int)

async renew(call_id, fingerprint, owner_token, fence)

Extend the lease of the exact fenced EXECUTING owner.

Return type:

LedgerResult

Parameters:

  • call_id (str)

  • fingerprint (str)

  • owner_token (str)

  • fence (int)

async complete(call_id, fingerprint, owner_token, fence, result)

Atomically cache a successful result for the configured terminal TTL.

Return type:

LedgerResult

Parameters:

  • call_id (str)

  • fingerprint (str)

  • owner_token (str)

  • fence (int)

  • result (Any)

async fail(call_id, fingerprint, owner_token, fence, error)

Atomically cache a definite pre-response failure as FAILED.

Return type:

LedgerResult

Parameters:

  • call_id (str)

  • fingerprint (str)

  • owner_token (str)

  • fence (int)

  • error (Any)

async mark_ambiguous(call_id, fingerprint, owner_token, fence, error)

Atomically tombstone an executing effect whose outcome is unknown.

This transition is used when adapter code raises or task cancellation occurs after begin. The external platform may already have accepted the effect, so automatic replay is permanently forbidden for the ledger retention horizon.

Return type:

LedgerResult

Parameters:

  • call_id (str)

  • fingerprint (str)

  • owner_token (str)

  • fence (int)

  • error (Any)

async get(operation_id)

Read and decode the current durable record without changing its TTL.

Return type:

LedgerRecord | None

Parameters:

operation_id (str)

core.gateway_rpc_ledger.GatewayRpcDecision

alias of LedgerResult

class core.gateway_rpc_ledger.GatewayRpcState(*values)

Bases: str, Enum

Durable states of one logical gateway RPC operation.

CLAIMED = 'CLAIMED'
EXECUTING = 'EXECUTING'
COMPLETED = 'COMPLETED'
FAILED = 'FAILED'
AMBIGUOUS = 'AMBIGUOUS'
class core.gateway_rpc_ledger.LedgerAction(*values)

Bases: str, Enum

Outcome of one attempted ledger transition.

ACQUIRED = 'ACQUIRED'
RECLAIMED = 'RECLAIMED'
BUSY = 'BUSY'
BEGUN = 'BEGUN'
ALREADY_EXECUTING = 'ALREADY_EXECUTING'
RENEWED = 'RENEWED'
COMPLETED = 'COMPLETED'
FAILED = 'FAILED'
AMBIGUOUS = 'AMBIGUOUS'
CACHED = 'CACHED'
CONFLICT = 'CONFLICT'
STALE_OWNER = 'STALE_OWNER'
LEASE_EXPIRED = 'LEASE_EXPIRED'
INVALID_STATE = 'INVALID_STATE'
MISSING = 'MISSING'
PROOF_EXPIRED = 'PROOF_EXPIRED'
PROOF_HORIZON_UNSUPPORTED = 'PROOF_HORIZON_UNSUPPORTED'
class core.gateway_rpc_ledger.LedgerRecord(state, operation_hash, payload_fingerprint, owner_token, fence, created_at_ms, updated_at_ms, lease_until_ms, completed_at_ms, result=None, error=None)

Bases: object

Decoded durable record returned by GatewayRpcLedger.get().

Parameters:
state: GatewayRpcState
operation_hash: str
payload_fingerprint: str
owner_token: str
fence: int
created_at_ms: int
updated_at_ms: int
lease_until_ms: int
completed_at_ms: int
result: Any = None
error: Any = None
class core.gateway_rpc_ledger.LedgerResult(action, state, operation_hash, requested_fingerprint, stored_fingerprint, fence, owner_token, result=None, error=None)

Bases: object

Result returned by a claim or state transition.

Parameters:
action: LedgerAction
state: GatewayRpcState | None
operation_hash: str
requested_fingerprint: str
stored_fingerprint: str
fence: int
owner_token: str | None
result: Any = None
error: Any = None
property terminal: bool

Whether the stored operation is in a terminal state.

property should_execute: bool

True only for the one transition authorized to run the effect.

property status: str

Stored state, exposed under the integration-friendly status name.

property owner: str | None

Alias used by gateway dispatch integration.

property cached_result: Any

Cached successful result, if this decision observed one.

core.gateway_rpc_ledger.canonical_payload_fingerprint(payload)

Return the SHA-256 hex digest of a canonical JSON payload.

Return type:

str

Parameters:

payload (Any)

core.gateway_rpc_ledger.gateway_rpc_fingerprint(platform, method, args, trace_id='', channel_id='')

Fingerprint the complete logical gateway RPC request.

Ephemeral transport fields such as retry number, stream entry id, reply key, and owner token are deliberately excluded.

Return type:

str

Parameters:
core.gateway_rpc_ledger.hashed_operation_suffix(operation_id)

Hash a logical operation id so raw identifiers never enter Redis keys.

Return type:

str

Parameters:

operation_id (str)

core.gateway_rpc_ledger.validate_gateway_rpc_replay_horizon(*, ledger_ttl_seconds, turn_checkpoint_ttl_seconds, turn_checkpoint_execution_lease_seconds, stream_retention_seconds)

Validate the shared replay horizon before durable RPC traffic starts.

A checkpointed tool slot can be reclaimed until its checkpoint TTL plus the final execution lease has elapsed. The Gateway tombstone must outlive that complete window, otherwise a late recovery could recreate an expired RPC ledger row and repeat an external effect. The tombstone must also outlive the finite stream replay horizon that can redeliver the original request.

Returns the validated integer ledger TTL so startup call sites can use the exact value for both publisher proof stamps and Gateway consumers.

Return type:

int

Parameters:

  • ledger_ttl_seconds (int)

  • turn_checkpoint_ttl_seconds (int)

  • turn_checkpoint_execution_lease_seconds (int)

  • stream_retention_seconds (float | int | None)