aetos/docs

Files

Vishesh 'ironeagle' Bangotra 406affcfcf feat: renumber lib reference into 4-pillar anatomy

Renumber the 15 reference docs into pillar groups (recipe -> anatomy):
- 01 Build Recipe (entry), 02 Composition Root
- domains: 10 Layering Rules
- contract: 20 Spec-First Routing, 21 CRUD Template, 22 Auth Integration
- interfaces/infra: 30 Infra Adapters
- UI: 40 UI: Auto-Rendered Admin & Custom Pages (new; @aetoskia packages,
  client template, x- catalog via react-openapi MCP module)
- pipelines: 50 Pipeline Scaffold, 51 Worker & Progress, 52 Human-in-the-Loop
- delivery/meta: 90 Drone Pipelines, 91 Versioning & Changelog,
  92 Genericity Matrix, 93 Library Registry, 94 Repository Inventory

MCP module names are unchanged (NN_ prefix stripped); all cross-references
(wiki, templates, hooks, registry) updated; lib index rewritten pillar-first;
build recipe bootstraps the client template. Bundle + HTML sites regenerated.

2026-10-19 17:42:55 +05:30

3.0 KiB

Raw Blame History

Worker & Progress

Summary

Long-running work runs in a background worker. Jobs are durable documents that move through an explicit status machine. Progress is streamed to clients.

The status machine

pending -> processing -> completed
                     \-> failed    (retried up to N times)
                     \-> paused    (waiting for human input)

pending — queued, eligible for pickup.
processing — claimed by a worker.
paused — blocked on a human-in-the-loop question.
failed — error; retry_count increments until a max, then skipped.
completed — terminal success.

The poll loop

async def poll(resources):
    backoff = MIN_POLL
    while True:
        jobs = await job_svc.search(filter={"status": "pending"}, limit=100)
        backoff = MIN_POLL if jobs else min(backoff * 2, MAX_POLL)
        for job in jobs:
            if (job.retry_count or 0) >= MAX_RETRIES:
                continue
            await job_svc.update_status(job.id, "processing")
            try:
                await process(job, resources)
            except PauseForResolution:
                await job_svc.update_status(job.id, "paused")
            except Exception as e:
                await job_svc.update_status(
                    job.id, "failed", error=str(e),
                    retry_count=(job.retry_count or 0) + 1,
                )
        await asyncio.sleep(backoff)

Tunables live in environment variables: WORKER_POLL_MIN, WORKER_POLL_MAX, WORKER_MAX_RETRIES.

Progress events

A worker publishes structured progress events to a bus (a port). The HTTP edge exposes a streaming endpoint that replays events for a given job id.

class ProgressEvent(BaseModel):
    job_id: str
    step: str
    status: str            # started | progress | completed | skipped | paused
    message: dict | None = None

The bus is an in-memory adapter by default; because it is a port, it can be swapped for a durable broker. Emit from the orchestrator at each stage:

await self.bus.publish(ProgressEvent(job_id=jid, step="normalize",
                                     status="completed", message={"count": n}))

Streaming endpoint

async def stream_job_events(request, id):
    bus = request.app.state.db_resources.get("progress_bus")
    if bus is None:
        return {"events": []}

    async def gen():
        sent = 0
        while True:
            events = bus.get_events(id)
            for e in events[sent:]:
                yield f"data: {e.model_dump_json(exclude_none=True)}\n\n"
                sent += 1
            await asyncio.sleep(0.5)

    return StreamingResponse(gen(), media_type="text/event-stream")

Known limitation. The default in-memory bus is per-process. If the worker and the API run as separate processes, events published by the worker are not visible to the API. For production, back the bus with a shared broker. The port boundary makes this a one-adapter change.

3.0 KiB Raw Blame History