Background jobs

import { setTimeout as delay } from "node:timers/promises";
import { Agent, type ModelProvider } from "@agentium/core";
import { AgentQueue, AgentWorker } from "@agentium/queue";

// A deterministic provider keeps this infrastructure example free of API keys.
const model: ModelProvider = {
  providerId: "example",
  modelId: "queue-demo",
  async generate() {
    return {
      message: { role: "assistant", content: "Background job completed." },
      finishReason: "stop", raw: null,
      usage: { promptTokens: 0, completionTokens: 0, totalTokens: 0 },
    };
  },
  async *stream() {
    yield { type: "text", text: "Background job completed." };
    yield { type: "finish", finishReason: "stop" };
  },
};
const agent = new Agent({ name: "worker", model, register: false });
const connection = process.env.REDIS_URL ?? "redis://127.0.0.1:6379";
const queueName = process.env.AGENTIUM_QUEUE_NAME ?? "agentium-quickstart";
const queue = new AgentQueue({ connection, queueName });
// Construction starts consumption; start() is not required.
const worker = new AgentWorker({
  connection, queueName, agentRegistry: { worker: agent }, concurrency: 1,
});

try {
  const { jobId } = await queue.enqueueAgentRun({
    agentName: "worker", input: "Run the background-job example.",
  });
  const deadline = Date.now() + 30_000;
  while (true) {
    const status = await queue.getJobStatus(jobId);
    if (status.state === "completed") {
      console.log(status.result?.text);
      await queue.cancelJob(jobId); // Remove this completed demo record.
      break;
    }
    if (status.state === "failed") throw new Error(status.error ?? "Job failed");
    if (Date.now() >= deadline) throw new Error("Timed out waiting for the job");
    await delay(100);
  }
} finally {
  // Drain the worker before closing the resources its jobs use.
  let drained = false;
  try { await worker.stop(); drained = true; }
  finally {
    await queue.close();
    if (drained) await agent.close();
  }
}