Crate agentq 

An embeddable job queue for agent tool calls.

agentq runs inside your own process. There is no Redis, no separate service, and nothing to deploy. It exists to stop a retried unit of work from being executed twice, and to keep a burst of work from overwhelming whatever is on the other end of it.

What it gives you

• Idempotency keys. Every job carries a caller-supplied key. Push a key that already completed and you get that job’s output back from cache instead of running it again. Push one that is still running and you join it, receiving the same outcome when it lands.
• Per-lane concurrency. Each Priority lane has its own capacity and its own concurrency limit, so cheap work and expensive work can be given very different budgets, and a saturated lane never starves another.
• Backpressure. Lanes are bounded. When one fills, Queue::push waits rather than letting the queue grow without limit.
• Failure isolation. A job that returns an error, or panics outright, records a terminal state and leaves every other job untouched.

Getting started

Build a queue, push work into it, and wait for the result:

use agentq::{Job, LaneConfig, Priority, Queue};

let queue = Queue::builder()
    .lane(Priority::High, LaneConfig { capacity: 32, permits: 1 })
    .lane(Priority::Low, LaneConfig { capacity: 128, permits: 4 })
    .start();

let job = Job::new(
    "charge-order-4821".to_string(),
    Priority::High,
    Box::new(|| Box::pin(async { Ok("charged".to_string()) })),
);

let output = queue.push_and_wait(job).await?;

Queue::push_and_wait returns the cached output if the key already completed, joins the running job if the key is in flight, and otherwise queues the work and waits for it. A job that returns Err surfaces as WaitError::Failed.

If you only want to enqueue without waiting, Queue::push returns an Accepted describing which of those three things happened, carrying a JobHandle you can await later.

Jobs

A job is an async closure paired with an idempotency key and a priority. It returns Result<String, Box<dyn Error + Send + Sync>>. The String is cached against the key, so a later push of the same key gets that output back without re-executing.

Sharing a queue

Queue is cheap to clone and every clone refers to the same queue, so share it across tasks with queue.clone() rather than wrapping it in an Arc.

Both Queue::push and Queue::push_and_wait are cancel safe. Dropping the future before it completes, from a timeout or a select! branch, leaves no claimed key behind: either the job was dispatched and runs to completion, or the key is released and can be pushed again.

How it works

Each Priority gets a bounded channel and its own semaphore, sized by the LaneConfig you give it. QueueBuilder::start spawns one worker task per lane, which pulls jobs off that lane’s channel. Before dispatching a job the worker acquires a permit from the lane’s semaphore, so a saturated lane applies backpressure at the channel rather than piling up unbounded in-flight work.

Each job then runs in its own spawned task, holding that permit until it finishes. That isolation is what keeps a panicking job from killing the lane, and a drop guard makes sure every job lands in a terminal state even if it panics partway through.

Dedup state lives in a single map keyed by idempotency key. Queue::push checks and claims a key under one lock, so two concurrent pushes with the same key cannot both get through.

Caveats

A job that pushes to its own queue and waits on the result can deadlock, if every permit in that lane is held by jobs doing the same thing. There is no mechanism to release a permit while waiting.

Keys and their cached outputs are retained for the life of the process, so a producer with unbounded distinct keys grows memory. Dropping the queue abandons in-flight work.

agentq deliberately does not persist anything and does not coordinate across processes. Everything lives in your binary; if it dies, queued work dies with it. If you need durability or a shared queue across machines, you want a different tool.

Upcoming

• Capped retries with exponential backoff. A failed job is currently recorded as failed and not re-attempted; re-pushing is the caller’s decision.
• Time-windowed keys, so cached outputs expire rather than accumulating.
• Graceful shutdown, draining in-flight work before exit.
• Arbitration between lanes, so a High job can preempt a Low one rather than the lanes merely being independent.

Structs

Job

JobHandle

JobLost

LaneConfig

Queue

QueueBuilder

Enums

Accepted

Outcome

Priority

PushError

State

WaitError

Type Aliases

Func

JobResult

Key