Queued jobs

The queued-jobs channel takes whole units of work off a Choria asyncjobs work queue, runs the agent loop against each one, and stores the answer back on the task. The submitter holds no connection to the worker: it enqueues a task, and reads the answer off the task record once a worker has written it.

Creating the storage

A worker requires its storage to exist. Create it with ajc, version 0.4.0 or newer.

The task store holds every job and the answer written back to it:

$ ajc tasks initialize

The work queue holds the jobs waiting to be taken:

$ ajc queue add FISK_AI --run-time 10m --tries 3 --concurrent 10

Run time, retry cap and concurrency are properties of the queue, not of the agent configuration. The worker reads them from the consumer at startup and prints them on the banner. The run time must be longer than a job takes, or the queue redelivers work that is still running.

A worker started before either exists fails:

fisk: error: building the jobs endpoint: connecting to queue "FISK_AI": storage not ready: stream CHORIA_AJ_TASKS does not exist, create it with 'ajc tasks initialize'

Submitting work

A caller enqueues a task with the queue engine’s own client. The task must name the configured queue and task type, and its payload is a v1 prompt request:

A queue has nobody waiting on it, so the three other kinds of request are refused here: they act on a conversation somebody is watching.

The request holds the prompt and the framing every v1 message needs:

{
  "protocol": "io.choria.fisk-ai.v1.request.prompt",
  "id": "2fT8kMxQ1aBcDeFgHiJkLmNoPqR",
  "request": "2fT8kMxQ1aBcDeFgHiJkLmNoPqR",
  "conversation": "2fT8kMxQ1aBcDeFgHiJkLmNoPqR",
  "sequence": 0,
  "time": "2026-08-15T11:30:00Z",
  "sender": {
    "name": "ops-console"
  },
  "prompt": "Which streams have no consumers?"
}

On a request, id, request and conversation all hold the same value. sender.name is limited to letters, digits, - and _.

$ ajc tasks add fisk-ai:run --payload-file request.json --queue FISK_AI
Enqueued task 3Hwxl119ZbwKHCKPzWlslXZYnB0

The submitter supplies the task id, or the engine mints one. The worker hashes it with the serving identity to get the session the run journals under, so a job creates a session or resumes one an earlier delivery of the same task made, and reaches nothing else on the worker. Every id the queue accepts works, a leading dash and a colon included.

Optional fields narrow what one job may do:

A budget may only lower what the configuration allows. A value above the configured limit is ignored.

The worker refuses a payload it cannot run and does not retry it, recording the reason in the task’s LastErr. This covers:

• an oversized payload
• a payload that is not a valid v1 request
• a payload that is not an io.choria.fisk-ai.v1.request.prompt, or whose prompt is empty

Reading the answer

The answer is stored on the task itself as a v1 result message:

$ ajc tasks view 3Hwxl119ZbwKHCKPzWlslXZYnB0 --json

{
  "protocol": "io.choria.fisk-ai.v1.result",
  "request": "2fT8kMxQ1aBcDeFgHiJkLmNoPqR",
  "sender": {
    "name": "worker"
  },
  "recipient": {
    "name": "ops-console"
  },
  "stop_reason": "end_turn",
  "text": "ORDERS and PAYMENTS have no consumers.",
  "usage": {
    "input_tokens": 1608,
    "cache_read_tokens": 663,
    "output_tokens": 51,
    "llm_calls": 2,
    "tool_calls": 1
  }
}

The request field echoes the id the caller submitted, and recipient names the caller that asked. input_tokens counts every input token the job consumed. cache_read_tokens and cache_create_tokens are subsets of that total, not additions to it.

A failed run is still a completed job. The worker stores a v1 error message with a stop_reason and acknowledges the task. It is not retried: a model refusal or an exhausted budget fails the same way on redelivery.

Redelivery

The worker journals every run under the session its task id derives. When a worker dies mid-job, the redelivery derives the same session and resumes that journal instead of starting again.

A job whose session already completed is answered from the journal, without running the agent or calling the model. This is the case when a worker finished a job and died before acknowledging the task.

Configuration

Every field under expose.agent.jobs has a default, so an empty block is valid.

expose:
  agent:
    jobs:
      # The work queue to consume. It must already exist.
      queue: FISK_AI
      # The task type this worker handles. Tasks of another type on the
      # same queue are left alone.
      task_type: fisk-ai:run
      # How many jobs this process runs at once. The --workers flag
      # overrides it.
      workers: 1
      # The NATS context the queue is reached over, defaulting to the
      # top-level nats_context. It is dialed separately, so the queue may
      # live on a different cluster from the session store.
      nats_context: production
      # Bounds a task payload in bytes before anything decodes it.
      max_payload: 524288

A worker only claims tasks of its configured task_type. Submit a different type and the task stays in the queue until it expires, with no error logged at either end.

Safety

Publish permission on the queue is the only access control. Anyone who can enqueue a task of the configured type runs the full agent loop with prompt text of their choosing, against every tool the configuration allows. Restrict publish permission on the queue’s subjects the way any other NATS resource is restricted.

The rest of what applies to any served run is covered in Serving.