> For the complete documentation index, see [llms.txt](https://fleet.hackmap.win/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://fleet.hackmap.win/guides/batch-automation.md). # Writing a batch automation Use this shape when work arrives as discrete tasks. Each task has a typed payload; you do something with it; you emit a result. The framework manages the queue, the dispatch, and the schema validation. ## Skeleton ```python from pydantic import BaseModel from fleet.core import BaseConfig, BatchAutomation, register class MyConfig(BaseConfig): api_key: str class ScrapeTask(BaseModel): url: str priority: int = 0 class ScrapeResult(BaseModel): url: str status: int body_len: int @register("scraper") class Scraper(BatchAutomation[MyConfig]): Config = MyConfig TaskPayload = ScrapeTask # required for BatchAutomation Output = ScrapeResult # optional — the type your stream emits async def run_one(self, task, ctx): # task.payload is a ScrapeTask instance, already validated resp = await fetch(task.payload.url, api_key=ctx.config.api_key) await ctx.emit(ScrapeResult( url=task.payload.url, status=resp.status, body_len=len(resp.text), )) ``` The framework calls `run_one` once per task. You don't loop. You don't pull from the queue. The framework does that. `TaskPayload` is required on every `BatchAutomation`. The registry rejects the class at import time if it's missing, with a `TypeError`. This is on purpose: a batch automation without a declared task schema can't be safely validated at the master, and silent type drift is exactly what we're trying to avoid. ## Pushing tasks Two ways. The first is from another automation — typed, validated client-side and server-side. Declare a `Queue[T]` reference (typically in a shared contracts package): ```python # in cf_contracts/__init__.py from fleet.core import Queue SCRAPER = Queue[ScrapeTask]("scraper", payload=ScrapeTask) ``` Submit from anywhere a `Context` is available: ```python await ctx.submit(SCRAPER, ScrapeTask(url="https://...", priority=5)) ``` The second is from outside the fleet — a UI, a scheduled job, a one-off CLI. Use the REST endpoint: ```bash curl -X POST http://master/api/v1/automations/scraper/tasks \ -H "Authorization: Bearer $ADMIN_TOKEN" \ -H "Content-Type: application/json" \ -d '{"payload": {"url": "https://...", "priority": 5}}' ``` The master validates the `payload` JSON against the receiver's `TaskPayload` schema and returns a `task_id`. Bad payload → 422 with the Pydantic error pointing at the offending field. ## Concurrency The number of `run_one` calls happening in parallel is `slots`. Each slot is a Python task running an internal loop that BLPOPs the receiver's queue, validates the task, calls `run_one`, and loops. BLPOP is atomic across slots and across workers. If you have two workers of type `scraper`, each with 5 slots, you have 10 concurrent consumers all racing for tasks. Redis hands each task to exactly one of them. Scaling concurrency is just bumping `slots` or adding workers. The framework will not duplicate-deliver a task. ## Task identity and in-flight tracking `task.id` is the framework-assigned identifier (a uuid hex by default; set explicitly via `submit(..., task_id="...")` or the REST endpoint's `task_id` field). The framework doesn't track which slot is processing which task. If you want a record of in-flight tasks, use a `ctx.kv` or `ctx.counter`: ```python async def run_one(self, task, ctx): inflight = ctx.counter("scraper:inflight") await inflight.incr() try: await do_work(task.payload) finally: await inflight.decr() ``` For "is task X being worked on right now", a `ctx.kv("scraper-inflight").set(task.id, ctx.worker_id, ttl_seconds=300)` paired with `kv.delete(task.id)` in `finally` is the simple shape. The framework doesn't ship this as a built-in because the right TTL and recovery semantics are per-plugin. ## Failure `run_one` exceptions don't lose the task. The slot runner reserves each task with a visibility timeout (`Config.visibility_seconds`, default 300s) and `nack`s on exception with exponential backoff. After `max_attempts` (default 3 from `Config.max_attempts`) the task is moved to the dead-letter queue. The master also sweeps visibility-expired reservations every 10s and re-queues them, so a crashed slot that never gets to nack still recovers. What you write: ```python async def run_one(self, task, ctx): await do_work(task.payload) # just raise; the framework handles retries ``` Per-task overrides at submit time: | field | default | use | | ----------------- | ------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------ | | `max_attempts` | `Config.max_attempts` (3) | bump for flaky external calls; drop to 1 for "no retries please" | | `priority` | 0 | `>0` routes to high tier (drained first), `<0` to low | | `ttl_seconds` | none | task is silently acked + dropped on reserve if older than this. Use for "act now or never" | | `idempotency_key` | none | a re-submit with the same `(automation_type, key)` inside `FLEET_IDEMPOTENCY_TTL_SECONDS` (default 24h) returns the original `task_id`, no duplicate enqueue | Inspecting failures: `GET /api/v1/automations/{type}/dead?n=50` peeks the DLQ; `POST /api/v1/automations/{type}/dead/replay` re-enqueues with `attempt=0`; `DELETE /api/v1/automations/{type}/dead` drains it. The dashboard's DLQ inspector card calls these. See features 2.1+2.2 (retries+DLQ), 2.4 (priority), 2.5 (TTL), 2.6 (idempotency). ## Idempotent submission Use `idempotency_key` when the submit caller can be re-entered safely but shouldn't enqueue twice: ```bash # Re-running this cron job at 09:00 every day is fine — the second call # returns the original task_id with queued=false, duplicate=true. curl -X POST $MASTER/api/v1/automations/cf-cookie-harvester/tasks \ -H "Authorization: Bearer $TOKEN" \ -H "Content-Type: application/json" \ -d '{ "payload": {"site": "example.com"}, "idempotency_key": "harvest-example.com-2026-05-15-09" }' ``` The implementation is `Backend.kv_setnx` (Redis `SET NX EX`, sqlite `INSERT … ON CONFLICT`, in-memory check-then-set) so two simultaneous submits with the same key still result in exactly one enqueue. ## When to use a batch automation vs a continuous one Batch if any of these are true: * The unit of work has an identity (a task ID). * You want to enqueue work from outside the worker. * You want explicit concurrency control via `slots`. * You'll need retries or dead-lettering. Continuous if all of these are true: * The work is a steady stream from a source the slot itself owns. * There's no natural "task" — just a loop. * You don't need to enqueue from outside. If you're not sure, lean batch. It costs you almost nothing and gives you flexibility. --- # Agent Instructions This documentation is published with GitBook. GitBook is the documentation platform designed so that both humans and AI agents can read, navigate, and reason over technical content effectively. Learn more at gitbook.com. ## Querying This Documentation If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question. Perform an HTTP GET request on the current page URL with the `ask` query parameter, and the optional `goal` query parameter: ``` GET https://fleet.hackmap.win/guides/batch-automation.md?ask=&goal= ``` `ask` is the immediate question: it should be specific, self-contained, and written in natural language. `goal` is optional and describes the broader end goal you are ultimately trying to accomplish on behalf of the user. GitBook uses it to tailor the answer towards what is most useful for that goal. The response will contain a direct answer to the question and relevant excerpts and sources from the documentation. Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.