Indexing, export, and retention should not hold an HTTP request or WebSocket run open. Moving work to a queue adds another state machine that the API and browser must represent honestly. This chapter builds the checkpoint and retry contract locally, then defines how job ids, progress, failure, and manual retry cross the service boundary.
A Job has an id, kind, payload, checkpoint, attempt count, status, error, and checkpoint history. Enqueue is idempotent by job id. The course’s local JobQueue keeps these transitions explicit and deterministic; a deployed adapter must persist them in SQLite/PostgreSQL or a queue backend before exposing them as recoverable browser state.
The frontend should receive a job id immediately, then read or subscribe to status. It must distinguish queued, running, retrying, complete, and dead rather than showing one indefinite spinner.
from autocode.jobs import JobQueue, JobStatus
queue = JobQueue(max_attempts=3)
job = queue.enqueue("reindex:s-1", "reindex", {"session_id": "s-1"})
assert job.status == JobStatus.QUEUED
result = queue.run(job.job_id, lambda current: 4)
assert result.status == JobStatus.COMPLETE
assert result.checkpoint == 4
assert result.history == [4]
print("completed checkpoint:", result.checkpoint)completed checkpoint: 4The worker returns a checkpoint only after its side effect is safe to repeat or has been committed with the checkpoint. If a worker performs an external write and then crashes before recording progress, the external operation needs its own idempotency key. A checkpoint number by itself cannot undo a duplicate side effect.
A retryable error leaves the job in retry state with its prior checkpoint. A second attempt can resume from the same point. Exhausted attempts land in a dead-letter list with the last actionable error.
queue = JobQueue(max_attempts=2)
queue.enqueue("export:1", "export", {})
def crash_once(current):
if current.attempts == 1:
raise RuntimeError("worker stopped after checkpoint 2")
return 5
queue.jobs["export:1"].checkpoint = 2
assert queue.run("export:1", crash_once).status == JobStatus.RETRY
finished = queue.run("export:1", crash_once)
assert finished.status == JobStatus.COMPLETE
assert finished.checkpoint == 5
queue.enqueue("poison", "reindex", {})
for _ in range(2):
queue.run("poison", lambda current: (_ for _ in ()).throw(ValueError("invalid payload")))
assert queue.jobs["poison"].status == JobStatus.DEAD
print("dead letters:", [item.job_id for item in queue.dead_letters])dead letters: ['poison']The experiment should inject a crash at every checkpoint boundary, restart the worker, and compare output artifact ids and final checkpoint. The acceptance condition is exactly-once effect where the external effect is idempotent, not the stronger and usually unavailable claim that a distributed queue delivers exactly once.
A reindex job should expose file-save time, enqueue time, start time, checkpoint time, completion time, and resulting index freshness. The REST shape can return a current job resource, while WebSocket events notify the selected session of important transitions. The browser treats those events as projections of persisted job state, not the only record.
Sweep worker concurrency against interactive stream latency. Background throughput that delays agent feedback or SQLite writes beyond the product budget is a regression, even when the queue drains faster.
from autocode.jobs import JobQueue
queue = JobQueue()
for i in range(3):
queue.enqueue(f"index:{i}", "reindex", {"document": i})
completed = [queue.run(job_id, lambda job: job.checkpoint + 1).status for job_id in queue.jobs]
assert all(status == "complete" for status in completed)
print("jobs completed:", len(completed))jobs completed: 3The browser queue view should show queued, running, retrying, complete, and dead-letter counts, plus a manual retry action that is authenticated and auditable. The current in-process queue teaches the transition contract; replacing it with a worker process requires a durable store and an integration test that restarts both API and worker at every checkpoint boundary.
Add a job resource shape for an idempotent export worker that writes a content-addressed artifact and records the digest as its checkpoint. Simulate a crash after object write but before status update, then specify the API and browser states during retry and dead-letter handling.
An export writes an artifact and crashes before recording its checkpoint. Explain how retry avoids duplicate objects and what the jobs API and browser show before, during, and after recovery.