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.

Longer guides, covering idempotency, lane configuration, error handling and the internals, live at https://agentq.hiesenbug.dev.

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.

Workflows

A workflow is an ordered list of StepDef entries under a stable id. Each step carries a name for logging and UI, a RetryPolicy that caps attempts and chooses backoff, and an optional per-step timeout.

While a workflow runs, every step is in some StepStatus: waiting to start, leased to a worker, completed with output, failed after an attempt, or blocked on an external condition. The engine records Event values along the way so you can reconstruct what happened; workers learn whether they claimed a step through ClaimResult.

WorkflowEngine drives a Workflow’s steps through the same Queue V1 jobs use — it is the only thing that calls Queue::push for a workflow step. A failed step is retried per its RetryPolicy by the engine itself, not by Queue; Queue stays exactly as retry-agnostic as it is for plain jobs. A step blocked on something outside the process (approval, a webhook, a human) ends up StepStatus::Waiting, and WorkflowEngine::resume unblocks it with whatever input arrived.

Where a step’s status and event history live is up to a DurableStore implementation, not the engine. SqliteStore ships as the one concrete implementation, behind the sqlite Cargo feature, storing an append-only event log alongside a queryable projection of each step’s current status — so a consumer who only wants V1’s Queue never pulls in rusqlite. If you need a different backend, implement DurableStore yourself; the engine doesn’t know or care what’s behind the trait.

A worker that dies mid-step leaves its lease to expire; recover finds those expired leases via DurableStore::expired_leases and re-admits the steps, recording Event::WorkerRecovered.

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.

Queue itself 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 a job’s state to survive a restart, use a Workflow with a DurableStore (see above) rather than the raw Queue — but agentq still does not coordinate across machines; if you need that, 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

Execution

Job

JobHandle

JobLost

LaneConfig

NonRetryable

Marker error returned from a step body to fail the workflow without retrying.

Queue

QueueBuilder

RetryPolicy

SqliteStore

SQLite-backed DurableStore with an append-only event log and a steps projection.

StepDef

WaitForInput

Marker error returned from a step body to enter StepStatus::Waiting.

Workflow

WorkflowEngine

Drives workflow steps through the V1 Queue and records V2 Events.

Enums

Accepted

Backoff

ClaimResult

EngineError

Event

Outcome

Priority

PushError

State

StepStatus

StoreError

WaitError

Traits

DurableStore

Functions

recover

Re-admits steps whose worker lease expired while the worker was unavailable.

Type Aliases

Func

JobResult

Key

StepFunc

A step’s body. Fn, not FnOnce like Job’s Func — the engine may call this again across retry attempts, so it can’t be consumed on first run.