core.dlq module

Dead Letter Queue (DLQ) handling for failed Redis Stream messages.

Provides the parking lot for inbound/outbound envelopes that the stream consumers could not process after repeated retries, plus the tooling to inspect and replay them. Failed messages are acknowledged on their source stream and re-published to the append-only sg:stream:dlq stream so they are removed from the live pipeline without silently trimming older recovery triggers, and operators can later resurrect them via replay_dlq_entry().

The functions here are the seam between the consumers in core/stream_consumer.py and core/outbound_consumer.py (which call handle_failed_message() from their error paths) and the operational scripts/dlq_replay.py CLI (which drives replay_dlq_entry()). All payloads are msgpack-framed by core.serialization.

class core.dlq.DLQProjectionAction(*values)

Bases: str, Enum

Outcome of one lane-local outbox projection.

PROJECTED = 'PROJECTED'
CACHED = 'CACHED'
class core.dlq.DLQProjectionResult(action, target_msg_id, acknowledged, deleted)

Bases: object

Global target identity plus source cleanup counts.

Parameters:
action: DLQProjectionAction
target_msg_id: str
acknowledged: int
deleted: int
class core.dlq.DLQPublicationResult(action, target_msg_id)

Bases: object

First-writer result for one idempotent global DLQ publication.

Parameters:
action: DLQProjectionAction
target_msg_id: str
class core.dlq.DLQSourceRetirement(acknowledged, deleted, attempts_cleared)

Bases: object

Counts returned by one atomic source-slot terminal retirement.

Parameters:

  • acknowledged (int)

  • deleted (int)

  • attempts_cleared (int)

acknowledged: int
deleted: int
attempts_cleared: int
exception core.dlq.DLQProjectionError

Bases: RuntimeError

The outbox projector rejected or could not decode a durable row.

exception core.dlq.DLQReplayError

Bases: RuntimeError

A durable operator replay could not be planned or completed safely.

exception core.dlq.DLQReplayConflict

Bases: DLQReplayError

A replay request conflicts with the immutable first-writer plan.

core.dlq.dlq_projection_ledger_key(source_outbox, projection_identity, *, target_stream='sg:stream:dlq')

Return one independently expiring target-slot projection ledger.

Return type:

str

Parameters:

  • source_outbox (str)

  • projection_identity (str)

  • target_stream (str)

core.dlq.dlq_replay_ledger_key(dlq_msg_id, *, dlq_stream='sg:stream:dlq')

Return the DLQ-slot receipt key for one operator replay request.

Return type:

str

Parameters:

  • dlq_msg_id (str)

  • dlq_stream (str)

class core.dlq.FailureDisposition(*values)

Bases: str, Enum

Durable outcome of a failed stream-processing attempt.

RETRYING leaves the source entry in its pending-entry list. Terminal dispositions move the entry to the DLQ and acknowledge the source entry; AMBIGUOUS additionally records that replay safety could not be proven.

RETRYING = 'RETRYING'
DEAD_LETTERED = 'DEAD_LETTERED'
AMBIGUOUS = 'AMBIGUOUS'
property terminal: bool

Return whether the source entry has reached a terminal parking state.

core.dlq.classify_failure_disposition(error, attempt, *, durable_proof_status='')

Classify retry, dead-letter, and replay-ambiguous failures consistently.

Return type:

FailureDisposition

Parameters:
core.dlq.extract_stream_payload_bytes(raw)

Extract the raw msgpack-framed payload from a Redis Stream entry.

Pulls the data field out of a single XRANGE/XREADGROUP entry, tolerating both the bytes-keyed (b"data") and string-keyed ("data") forms that redis.asyncio may return depending on decode settings, and coercing a string value back to utf-8 bytes. This keeps the rest of the DLQ code agnostic to how the connection was configured. A missing payload yields b"" rather than raising, which callers treat as an empty entry.

A pure helper that touches no Redis or other I/O. Called by handle_failed_message(), inspect_dlq_entry(), and replay_dlq_entry() in this module, and exercised directly by tests/core/test_context_assembly_hardening.py.

Parameters:

raw (dict[Any, Any]) – The field/value mapping of one stream entry, keyed by either bytes or str.

Return type:

bytes

Returns:

The msgpack-encoded payload bytes, or b"" when no data field is present.

core.dlq.extract_stream_aux_fields(raw)

Extract durable transport sidecars from a stream entry.

Collects the non-payload metadata that travels alongside the msgpack body so it can be carried forward when a message is moved to the DLQ, preserving the original enqueue timestamp and the cross-service trace id used for grep-based correlation across the gateway / inference / agents pipeline. Both bytes- and str-keyed variants are probed, missing values are skipped, and every surviving value is normalised to a str (decoding bytes with errors="replace").

A pure helper with no I/O. Called by handle_failed_message() to populate the ts and trace_id fields of the DLQ record.

Parameters:

raw (dict[Any, Any]) – The field/value mapping of one stream entry, keyed by either bytes or str.

Return type:

dict[str, str]

Returns:

A mapping of the present auxiliary field names to their string values; empty when neither field is set.

class core.dlq.DLQOutboxProjector(redis, *, source_stream, consumer_name, group_name='sg:dlq-outbox-projector:v1', target_stream='sg:stream:dlq', batch_size=64, max_reclaim_pages=4, min_idle_ms=30000, reclaim_interval_seconds=15.0, read_block_ms=1000, ledger_ttl_seconds=2592000)

Bases: object

Project the inbound-slot DLQ outbox into the global operator DLQ.

Projection and source cleanup are intentionally two Lua calls because the lane-local outbox and global DLQ occupy different Redis Cluster slots. The target call first-writer binds the source event to one global stream ID in an independently expiring target-slot ledger. A lost target reply therefore reuses that ID; only then does a source-slot call atomically XACK and XDEL the outbox row.

Parameters:

  • redis (redis.asyncio.Redis)

  • source_stream (str)

  • consumer_name (str)

  • group_name (str)

  • target_stream (str)

  • batch_size (int)

  • max_reclaim_pages (int)

  • min_idle_ms (int)

  • reclaim_interval_seconds (float)

  • read_block_ms (int)

  • ledger_ttl_seconds (int)

property source_stream: str
property group_name: str
async ensure_group()

Create the dedicated outbox group from the beginning, once.

Return type:

None

async project_outbox_entry(source_msg_id, raw)

Project one row idempotently, then atomically ACK+delete its source.

Return type:

DLQProjectionResult

Parameters:
async reclaim_pending()

Bound one XAUTOCLAIM sweep by pages and per-page entry count.

Return type:

int

async run()

Continuously read new entries and periodically reclaim stale PEL rows.

Return type:

None

async start()

Provision the dedicated group and start one idempotent worker task.

Return type:

None

async stop()

Stop promptly; any in-flight row remains recoverable in the PEL.

Return type:

None

async core.dlq.publish_dlq_entry_idempotently(redis, *, source_stream, source_group, source_msg_id, source_raw, fields, ledger_ttl_seconds=2592000)

Publish one terminal row through an independently keyed target receipt.

The global DLQ cannot share a Redis Cluster slot with every source lane. This first-writer receipt makes that cross-slot boundary replay-safe: a lost reply reuses the canonical DLQ stream ID, and conflicting reuse of a source identity is rejected before another row can be appended.

Return type:

DLQPublicationResult

Parameters:
async core.dlq.retire_failed_source(redis, *, source_stream, source_group, source_msg_id, delete_source=False, attempts_key=None)

Atomically retire a source PEL row and its colocated retry counter.

Return type:

DLQSourceRetirement

Parameters:

  • redis (redis.asyncio.Redis)

  • source_stream (str)

  • source_group (str)

  • source_msg_id (str)

  • delete_source (bool)

  • attempts_key (str | None)

async core.dlq.handle_failed_message(redis, source_stream, group_name, msg_id, raw, error, attempt, *, durable_proof_status='', delete_source=False, attempts_key=None)

Move a message to the DLQ after MAX_RETRIES failures.

Ordinary failures below MAX_RETRIES stay in the PEL for XAUTOCLAIM and return RETRYING. Any replay-ambiguous failure is terminal immediately, regardless of attempt; otherwise reaching MAX_RETRIES parks and acknowledges the original as DEAD_LETTERED.

Return type:

FailureDisposition

Parameters:

  • redis (redis.asyncio.Redis)

  • source_stream (str)

  • group_name (str)

  • msg_id (str)

  • raw (dict[Any, Any])

  • error (Exception)

  • attempt (int)

  • durable_proof_status (str)

  • delete_source (bool)

  • attempts_key (str | None)

async core.dlq.inspect_dlq_entry(redis, msg_id)

Fetch and decode a single DLQ entry into an operator-friendly dict.

Reads one entry from the sg:stream:dlq stream by exact id (via an XRANGE bounded to msg_id..``msg_id``) and unpacks both the failure metadata that handle_failed_message() recorded – the originating stream, original message id, error string, and attempt count – and, when a body is present, the msgpack payload itself through core.serialization.deserialize_stream_payload. A payload that fails to deserialize is logged at debug and reported as None rather than aborting, so a poison message can still be inspected.

Issues one read against Redis and performs no writes. No internal callers were found by grep; this is an inspection/diagnostic helper intended for DLQ tooling and interactive use.

Parameters:

  • redis (Redis) – Async Redis client connected to the stream backend.

  • msg_id (str) – The DLQ stream entry id to look up.

Return type:

dict[str, Any] | None

Returns:

A dict with dlq_msg_id, original_stream, original_group, original_msg_id, error, attempt, durable disposition, and the decoded payload (possibly None), or None when no entry exists at that id.

async core.dlq.replay_dlq_entry(redis, dlq_msg_id, *, target_stream=None, migration_fn=None)

Replay one safe DLQ entry as a new, durable child operation.

A terminal lifecycle is immutable, so this function never republishes its trace_id. It derives one deterministic child trace, first-writer binds the exact source row, migrated payload, and destination in a DLQ-slot plan, and delegates publication to IngressLedger. That publication atomically establishes QUEUED before making the destination row visible. A second DLQ-slot call records the canonical destination ID and deletes the parked row. Retries after a lost reply therefore recover one receipt rather than creating another turn.

AMBIGUOUS entries and checkpoint-bound terminal turns remain parked. Their prior external-effect outcome or turn journal cannot safely be rebound to the fresh child trace at this seam.

Parameters:

  • redis (Redis) – Async Redis client connected to the stream backend.

  • dlq_msg_id (str) – The DLQ stream entry id to replay.

  • target_stream (str | None) – Optional override destination stream; when None the entry returns to the stream it originally failed on.

  • migration_fn (Any | None) – Optional callable applied to the unpacked payload to transform it before re-publishing.

Return type:

str | None

Returns:

The canonical child stream message ID, including on a retry after a lost reply, or None when policy refuses the entry.

Raises:

  • DLQReplayConflict – A concurrent or later request changed the destination or migrated payload after the first writer established a plan.

  • DLQReplayError – Redis returned a corrupt or incomplete replay binding.