Struct ReliableConsumer 

pub struct ReliableConsumer {

    pub group: ConsumerGroupName,
    pub agent: Option<AgentId>,
    pub topic: String,
    pub dedup_window: usize,
    pub retry: RetryPolicy,
    pub understood_features: u64,
    pub poll_interval: Duration,
    pub shutdown_grace: Duration,
    pub concurrency: ConcurrencyPolicy,
    pub max_queued_records: usize,
    pub max_queued_bytes: usize,
    pub respond_on: Option<AgentTopic<'static>>,
    pub inbox_route: InboxRoute,
    pub ack_on_pickup: bool,
    pub deduplicator: Option<Box<dyn Deduplicator>>,
    pub warm_dedup: bool,
    pub middleware: Vec<Arc<dyn AgentMiddleware>>,
    pub on_dead_letter: Option<Arc<dyn DeadLetterSink>>,
    pub operations: Option<Vec<String>>,
    pub sessions: Option<SessionConfig>,
    pub verifier: Option<Arc<KeyRegistry>>,
    pub signing_key: Option<Arc<SigningKey>>,
}

The reliable consumer (consumer-group delivery + dedup + retry + DLQ). Most callers use Agent::builder, not this directly.

Fields

group: ConsumerGroupName

agent: Option<AgentId>

Logical identity used for target filtering and replies. It is never inferred from the deployment group name.

topic: String

dedup_window: usize

retry: RetryPolicy

understood_features: u64

Must-understand AGDX feature bits implemented by this receiver. Messages requiring any other bit are dead-lettered before dispatch.

poll_interval: Duration

Poll interval, default a reactive 10ms. Raise for throughput-bound work.

shutdown_grace: Duration

How long a graceful shutdown waits for the in-flight message to finish before dropping the consumer. run returns LaserError::Timeout if the grace elapses with a message still in flight. Default 30s.

concurrency: ConcurrencyPolicy

How message handling is scheduled across partitions. Defaults to ConcurrencyPolicy::Serial (strict one-at-a-time).

max_queued_records: usize

Maximum records buffered across partition lanes.

max_queued_bytes: usize

Maximum payload and header bytes buffered across partition lanes.

respond_on: Option<AgentTopic<'static>>

inbox_route: InboxRoute

Default inbox route for the ctx’s directed-send and fan-out helpers.

ack_on_pickup: bool

Emit a Working task status on respond_on the moment an AGDX command is picked up, before the handler runs, so a contract caller can tell the command was consumed (versus expired unconsumed). Off by default. Requires respond_on and a valid agent id.

deduplicator: Option<Box<dyn Deduplicator>>

warm_dedup: bool

middleware: Vec<Arc<dyn AgentMiddleware>>

Cross-cutting hooks wrapped around each handler dispatch, in order, for auth, metrics, and tracing without touching the handler.

on_dead_letter: Option<Arc<dyn DeadLetterSink>>

Notified on every dead-letter with the result of publishing it, so a lost poison message (a DLQ publish failure) is an observable event, not a log line.

operations: Option<Vec<String>>

The command operations the handler serves. None serves every operation. A command for another operation is reported as skipped, so a model or tool record an agent writes for itself never becomes its own work.

sessions: Option<SessionConfig>

The session configuration the runtime applies: the session lens handlers read through AgentCtx::session, and SessionConfig::fail_on_dead_letter. Defaults to SessionConfig::default.

verifier: Option<Arc<KeyRegistry>>

When set, every message’s envelope signature is verified against this registry before dispatch, and an unsigned or unverified record is dead-lettered. Set it on control and effect topics, where verification is mandatory (the enforcement chokepoint for authorship and authorization).

signing_key: Option<Arc<SigningKey>>

This consumer’s signing identity, threaded into AgentCtx so respond answers correlated commands with signed AGDX responses.

Implementations

impl ReliableConsumer

pub fn builder() -> ReliableConsumerBuilder

Create an instance of ReliableConsumer using the builder syntax

impl ReliableConsumer

pub async fn run<H>( self, laser: &Laser, handler: H, ready: Sender<()>, shutdown: Receiver<()>, ) -> Result<(), LaserError>

where H: AgentHandler + Sync + Send + 'static,

Consume until shutdown fires, dispatching each message to handler. ready fires once the consumer has joined its group and is polling.

Delivery runs on ConsumerGroup::consumer, so a server that resolves group policies reads through the group-aware engine. On agent.sessions and agent.control, when the server serves filtered reads and the filter catalog, the group is bound to the addressee filter agdx.to In [<agent>, "*"] before it reads. A group bound to another filter is refused. On open Apache Iggy the group stays unbound and records are classified on the client. A capability probe that established nothing is an error.