core.trace module

Trace IDs, atomic creation, and bounded orphan reclamation.

class core.trace.TraceWatchdog(redis=None, *, anchor_stream='sg:stream:inbound', source_group='sg:workers', sweep_interval=300.0, orphan_threshold=1800.0, batch_size=128)

Bases: object

Reclaim due traces with a bounded ZSET scan and revision CAS.

Parameters:

redis (_WatchdogRedis | None)
anchor_stream (str)
source_group (str)
sweep_interval (float)
orphan_threshold (float)
batch_size (int)

SWEEP_INTERVAL = 300.0

ORPHAN_THRESHOLD = 1800.0

DEFAULT_BATCH_SIZE = 128

async sweep(redis=None)

Process at most batch_size traces due according to Redis TIME.

Return type:: WatchdogSweep
Parameters:: redis (_WatchdogRedis | None)

async run(redis=None)

Run sweeps until stopped; suitable for a supervised background task.

Return type:: None
Parameters:: redis (_WatchdogRedis | None)

start(redis=None)

Start the watchdog task once and return the supervised task handle.

Return type:: Task[None]
Parameters:: redis (_WatchdogRedis | None)

async stop()

Request a clean stop and wait for the current sweep to finish.

Return type:: None

class core.trace.WatchdogSweep(scanned=0, abandoned=0, ambiguous=0, cleaned=0, deferred=0, conflicted=0)

Bases: object

Parameters:

scanned (int)
abandoned (int)
ambiguous (int)
cleaned (int)
deferred (int)
conflicted (int)

scanned: int

abandoned: int

ambiguous: int

cleaned: int

deferred: int

conflicted: int

async core.trace.create_trace(redis, trace_id, initial_state, metadata)

Create a lifecycle trace through the canonical atomic kernel.

Return type:

TransitionResult

Parameters:

redis (Any)
trace_id (str)
initial_state (str | TraceState)
metadata (Mapping[str, Any] | None)

core.trace.generate_trace_id()

Generate a time-sortable UUIDv7, falling back to UUIDv4.

Return type:: str