Server Agent

A stateless schema-backed worker: wake, claim, commit, go idle.

A server agent is backend code — a cron job, a queue worker, an AI record — that reads and writes your app’s records outside the browser. The hard part is doing it without racing whatever else is working: if two workers pick up the same record at once, one write clobbers the other. This is what claim() is for.

Agents hold no socket, so pass transport: 'http' and import the same schema the rest of the app uses. Below, a worker finishes a record by claiming it, writing the result, and releasing it automatically when the claim goes out of scope.

claim({ id }) takes the record for your worker and returns a disposable handle: the fresh post-lease row is on claim.data, and holding the handle with await using releases the claim on scope exit (or call claim.release()). Claims don’t lock. If another writer holds the row, claim waits for them, re-reads the fresh row, then hands it to you — so two writers serialize instead of clobbering.

import Ablo from '@abloatai/ablo';
import { defineSchema, model, z } from '@abloatai/ablo/schema';

const schema = defineSchema({
  records: model({
    title: z.string(),
    status: z.enum(['todo', 'doing', 'done']),
    summary: z.string().optional(),
  }),
});

const control = Ablo({
  schema,
  apiKey: process.env.ABLO_API_KEY,
});

async function clientForWorker(workerId: string) {
  const { token } = await control.sessions.create({
    agent: { id: workerId },
    can: { records: ['read', 'update'] },
  });
  return Ablo({ schema, apiKey: token, transport: 'http' });
}

export async function completeTask(recordId: string, workerId: string) {
  // Participant identity comes from this worker-specific session. Two clients
  // made directly from the same root key are re-entrant, not contenders.
  const ablo = await clientForWorker(workerId);
  await ablo.ready();

  const record = await ablo.records.get({ id: recordId });
  if (!record) return { status: 'not_found' };

  const acquired = await ablo.records.claim({
    id: recordId,
    contention: {
      mode: 'skip',
      onStatus(event) {
        if (event.type === 'skipped') {
          console.info('record already owned', event.error.code);
        }
      },
    },
    description: 'completing',
  });
  if (!acquired) return { status: 'already_claimed' };

  await using claim = acquired;
  const updated = await ablo.records.update({
    id: claim.data.id,
    data: { status: 'done' },
  });

  return { status: 'done', record: updated };
  // claim auto-releases as the function returns
}

get({ id }) is an async server read — it hits the server and returns the row (or undefined, which the early not_found guard handles). The update runs while the claim is held; awaiting it resolves only once your database has confirmed the row landed.

The two options on the claim:

• queue: false — skip this record if another claim is already in progress, rather than queueing behind it. Fail-fast dedup: if someone else has this job, skip it. It resolves null; it does not throw. (The default queues.)
• description: 'completing' — a readable label for what your worker is doing, visible to anyone reading claim.state({ id }).

Atomic batches

When several rows must change together, submit one atomic commit through the same schema-backed client:

await ablo.commits.create({
  operations: [
    { action: 'update', model: 'records', id: 'record_123', data: { status: 'done' } },
  ],
  wait: 'confirmed',
});

Because the worker uses the same schema and claim() as everything else, its writes reach every connected client in real time and never collide with work already in progress.