Dispatch validated background jobs

Tenchi validates the message boundary between a producer and consumer without becoming a queue. Your infrastructure still owns persistence, claiming, acknowledgement, retries, backoff, concurrency, and dead letters.

Declare the message

Keep the stable name and payload type with the feature that owns the event:

# app/features/projects/jobs.py
from tenchi.jobs import job

from .schemas import MemberAdded


member_added_job = job(
    "projects.member_added",
    request=MemberAdded,
    result=None,
    description="Notify a user after project membership is committed.",
)

Changing a job name or payload can strand messages already stored in a queue. Treat both as durable wire contracts. Introduce a new name when a consumer cannot safely read both payload versions.

Validate before enqueueing

Build a JobMessage before handing data to an outbox or queue port:

from tenchi.jobs import job_message


message = job_message(
    member_added_job,
    MemberAdded(
        project_id=project.id,
        project_name=project.name,
        user_id=user_id,
    ),
)
await context.outbox.enqueue(
    job=message.name,
    payload_json=message.payload_json,
)

job_message() validates Python input, emits compact JSON bytes, and confirms that the bytes satisfy the schema published in jobs.json and can be read back strictly by the declared consumer type. A useful queue port therefore stays transport-shaped:

from typing import Protocol


class Outbox(Protocol):
    async def enqueue(self, *, job: str, payload_json: bytes) -> None: ...

If a custom serializer emits a different wire shape, job_message() raises JobBindingError before enqueueing it. Keep custom serialization aligned with the JSON Schema generated from the request annotation.

The job payload carries application facts. Delivery metadata such as message ids, trace ids, attempt counts, scheduled time, and queue partition belongs in your infrastructure envelope, not in every business payload.

Bind the consumer

Bind declarations to plain async use cases at the composition root:

# app/server/jobs.py
from app.features.projects.jobs import member_added_job
from app.features.projects.use_cases.notify_member_added import (
    notify_member_added,
)
from tenchi.jobs import create_job_dispatcher, job_group, job_handler


jobs = job_group(
    job_handler(member_added_job, notify_member_added),
)
dispatcher = create_job_dispatcher(jobs=jobs)

The handler must accept request and context. Its request and return annotations must exactly match the job declaration, so bad wiring fails when the application imports:

async def notify_member_added(
    request: MemberAdded,
    context: AppContext,
) -> None:
    await context.notifications.record(
        user_id=request.user_id,
        message=f"You joined {request.project_name}",
    )

job_group() rejects duplicate names. Put every registered group in app/server/jobs.py; tenchi map then shows job nodes and their handler bindings.

Dispatch one delivery

After your worker claims a message and creates its unit-of-work context, dispatch the raw stored JSON:

await dispatcher.dispatch(
    entry.job,
    payload_json=entry.payload_json,
    context=context,
)

The dispatcher:

1. rejects unknown names before opening the supplied context;
2. validates JSON before the handler runs;
3. invokes the handler through the shared use-case observer boundary;
4. validates the result before the context exits successfully.

If you pass an async context manager or factory, result validation happens before its successful exit, so an invalid result can still roll back the unit of work. Passing a ready context leaves commit and rollback with the caller.

JobDispatcher does not choose an execution timeout. The worker should apply a bounded delivery deadline, while a use case that needs the same limit in every entrypoint should enforce an entrypoint-neutral application deadline. Cancellation must still reach the dispatcher so the context can roll back before the worker releases or retries the claim.

Observe handlers

Pass use_case_observers= to create_job_dispatcher(). Each UseCaseOutcome.entrypoint is "job" and contains only the use-case identity, status, duration, UTC completion time, and stable application error code. Queue latency, attempt number, message id, and dead-letter state remain worker telemetry because the dispatcher never owns them.

Protect stored messages with a snapshot

Write the canonical manifest after registering a job:

uv run tenchi jobs --write jobs.json

The manifest contains stable names, descriptions, and input JSON Schemas. It never contains queued payloads or handler results. Commit jobs.json, then check exact drift locally and in CI (tenchi check runs this step once app/server/jobs.py exists, and jobs = true under [verify] in tenchi.toml, declared in the same change, makes tenchi verify enforce the historical comparison):

uv run tenchi jobs --check jobs.json

Before accepting a changed snapshot, compare it with the current file or a Git baseline:

uv run tenchi jobs --diff jobs.json
uv run tenchi jobs --diff-ref origin/main --snapshot jobs.json

When adopting the manifest for the first time and the selected Git ref truly predates jobs.json, add --allow-missing-baseline to that one --diff-ref comparison. The report records the missing baseline as metadata. Future comparisons fail if the historical file is absent. tenchi verify needs no override when app/server/jobs.py itself did not exist at the baseline.

Removing a job or narrowing the payloads its consumer accepts is breaking. Adding a job or widening accepted payloads is additive; description-only changes are metadata. Changes the analyzer cannot prove safe require review. If a breaking payload change is intentional, declare a new job name so workers can continue consuming messages stored under the old contract during rollout.

Compatibility is directional: it proves that the new consumer accepts messages valid under the historical manifest. Deploy that consumer before producers can enqueue the new shape. If rollback must restore an older consumer after new messages exist, keep the emitted shape acceptable to both versions or use a new job name.

Continue with Retries and background work to connect typed producer messages and registered dispatchers to a transactional outbox, rollback rules, dead-lettering, and worker-owned retries.