CookbookWorkflows
Task Worker Pool
Run one worker handle per pending task and show progress from a task ledger.
This recipe uses a task ledger as the visible work record and your own code as the runner. The ledger records state; it does not provide queueing, retries, or scheduling.
Setup
import { prompt, generate } from "@use-crux/ai";
import { tasks } from "@use-crux/core/tasks";
const work = await tasks({ title: "Batch enrichment" });
await work.add({ id: "account-1", label: "Enrich account 1" });
await work.add({ id: "account-2", label: "Enrich account 2" });
await work.add({ id: "account-3", label: "Enrich account 3" });Drain Pending Tasks
async function drainPending(taskListId: string) {
const work = tasks.ref(taskListId);
const pending = (await work.list()).filter(
(item) => item.status === "pending",
);
await Promise.all(
pending.map((item) => {
const worker = work.worker(item.id, {
guidelines: "Return a short JSON-safe result.",
});
return generate(
prompt({
use: [worker.asContext()],
tools: worker.asTools(),
system: "Complete the assigned enrichment task.",
}),
{ model, input: {} },
);
}),
);
}Add Backpressure
Use your own runner when you need concurrency limits:
async function drainInBatches(taskListId: string, batchSize = 3) {
const work = tasks.ref(taskListId);
const pending = (await work.list()).filter(
(item) => item.status === "pending",
);
for (let index = 0; index < pending.length; index += batchSize) {
const batch = pending.slice(index, index + batchSize);
await Promise.all(batch.map((item) => runOneTask(work.id, item.id)));
}
}
async function runOneTask(taskListId: string, taskId: string) {
const work = tasks.ref(taskListId);
const worker = work.worker(taskId);
return generate(
prompt({
use: [worker.asContext()],
tools: worker.asTools(),
system: "Complete this task.",
}),
{ model, input: {} },
);
}For retries, schedules, and cross-process locks, put the IDs into a durable runner payload and rebind with tasks.ref(taskListId) inside the job.
UI
Subscribe with useTasks(taskListId) and render the core status vocabulary directly.
function BatchProgress({ taskListId }: { taskListId: string }) {
const state = useTasks(taskListId);
if (state.status === "loading") return <p>Loading...</p>;
return (
<ul>
{state.data.map((item) => (
<li key={item.id}>
<span>{item.label}</span>
<span>{item.status}</span>
{item.progress ? <small>{item.progress}</small> : null}
</li>
))}
</ul>
);
}