Multimodal GPU workers

The Control Plane runs a multimodal operation inline when an adapter is available in its own process. Otherwise the job stays in eai.multimodal_jobs with status queued ("no adapter available, queued for GPU workers") and a GPU worker, python -m eai.media_worker, processes it.

Running

python -m eai.media_worker --kind stt [--kind diarization] [--stream EAI_MULTIMODAL] [--consumer stt-workers]
python -m eai.media_worker --kind tts --once          # at most one job, then exit
python -m eai.media_worker --kind stt --max-jobs 10   # exit after 10 jobs

--kind	Adapter capability	Local adapters (workers extra)
stt	stt	faster-whisper, or EAI_STT_URL
diarization	diarization	pyannote.audio (EAI_HF_TOKEN)
tts	tts	piper (piper-tts), or EAI_TTS_URL
image_generation, image_edit	image_generation	OpenAI compatible endpoint
vision_understanding	vision	OpenAI compatible endpoint, or local metadata and OCR
video_generation	video_generation	video generation endpoint

At start the worker refuses to run (exit code 2, explicit message) when no adapter is available for one of the requested --kind. The deploy/docker/worker.Dockerfile image installs the workers extra of services/control-plane/pyproject.toml (faster-whisper, pyannote.audio, piper-tts, silero-vad, torch, diffusers; the nats-py NATS client is a core dependency); a failing install fails the image build.

Processing a job

1. Claim: one statement in the system context, FOR UPDATE SKIP LOCKED, the oldest queued job of the requested operations becomes running, adapter = worker:<hostname>, attempts + 1. Two workers never claim the same job.
2. Run in a transaction scoped to the job tenant, with the system principal of that tenant: RLS, the FinOps usage event and the audit belong to the job tenant.
3. The adapter payload is rebuilt by adapter_payload() (eai.domains.multimodal.service) from the job input and the decrypted media. The inline path uses the same function, so the two runs cannot drift. The job input never holds a secret; a base64 media is stored encrypted when the job is queued.
4. Post processing (DLP on the extracted text, storage of the produced media) and finish_job are those of the inline path.

Errors: an AdapterError fails the job with a message cut at 2000 characters; any other exception fails it with the generic code worker_error, and the log keeps only the exception type and location (never the content, which may hold personal data).

Shutdown: SIGTERM or SIGINT stops claiming; the job in progress finishes, a claimed job not started yet is released (queued), then the process exits with code 0.

Lease and requeue

While a job runs the worker refreshes its updated_at (the lease) every EAI_MEDIA_WORKER_HEARTBEAT_SECONDS. When a worker dies between claim and finish, the periodic job media_requeue of python -m eai.worker requeues the running jobs claimed by a worker (adapter starting with worker:) whose updated_at is older than EAI_MEDIA_JOB_LEASE_SECONDS, and fails them (lease_expired_max_attempts) once attempts reaches EAI_MEDIA_JOB_MAX_ATTEMPTS. Every requeue is audited (multimodal.job.requeue).

NATS JetStream and KEDA

With EAI_NATS_URL, queueing a job also writes {job_id, tenant_id, operation} on subject multimodal.jobs.<operation> in the transactional outbox (eai.event_outbox): the event commits with the job. The outbox job of python -m eai.worker relays it to JetStream, which prefixes the tenant: the real subject is eai.<tenant_id>.multimodal.jobs.<operation>.

Stream to create (once, by the platform team):

nats stream add EAI_MULTIMODAL --subjects 'eai.*.multimodal.jobs.>' --storage file --retention work \
  --replicas 3 --discard old --max-age 7d

The worker creates or updates its durable --consumer on --stream, filtered on eai.*.multimodal.jobs.<kind> for each --kind (explicit ack, ack_wait equal to the lease). Each message is a wake-up: the worker claims and runs the oldest job, then acks the message after processing, so KEDA (the nats-jetstream ScaledObject of the chart, workers.items[].lagThreshold) sees a lag that counts the jobs not processed yet. The NATS permissions of deploy/data/nats/values.yaml are aligned: user worker publishes eai.*.multimodal.jobs.>, user multimodal subscribes to it.

Without NATS (or without the nats-py client) the worker polls the database every EAI_MEDIA_WORKER_POLL_SECONDS.

Consumer side deduplication (inbox)

Each received message goes through the consumer inbox (eai.inbox, table eai.inbox_events, migration V166) before its processing. It is recorded under the unique key (tenant, --consumer, message id); the id is the Nats-Msg-Id header when set, else <stream>:<stream sequence>, identical on every redelivery. The tenant comes from the subject eai.<tenant_id>.<subject> or from the tenant_id field of the message, so the row is under the tenant RLS.

• A message already processed (redelivery after a lost ack, duplicate publication) is acked without a new job claim.
• A failed processing marks the row failed (short error code, never the content), the message is naked (nak) and the next delivery replays it (attempts + 1).
• The message content is never copied into the table, only its identifiers.
• The inbox_purge job of python -m eai.worker deletes the processed rows older than EAI_INBOX_RETENTION_DAYS; received and failed rows stay for the replay and the investigation.

Deployment

• Helm: workers.enabled: true and one entry per operation in workers.items (name, kind, consumer, gpus, minReplicas, maxReplicas, lagThreshold). The pods read the Control Plane ConfigMap and Secret (database, encryption keys, object storage, NATS) and keep the hardening: non root, read-only root filesystem, drop: [ALL], seccomp RuntimeDefault, no service account token. The multimodal-workers network policy opens no ingress; egress only to PostgreSQL, NATS, object storage (encrypted media), OpenBao, OTLP and cluster DNS.
• Compose: service media-worker of the gpu profile (docker compose --profile gpu up media-worker).

Settings

Variable	Default	Purpose
EAI_MEDIA_WORKER_POLL_SECONDS	2	database polling interval without NATS
EAI_MEDIA_WORKER_HEARTBEAT_SECONDS	60	lease refresh while a job runs
EAI_MEDIA_JOB_LEASE_SECONDS	900	updated_at age after which a running job is requeued
EAI_MEDIA_JOB_MAX_ATTEMPTS	3	claims before the job is failed for good
EAI_WORKER_MEDIA_REQUEUE_SECONDS	60	interval of the media_requeue job of python -m eai.worker
EAI_INBOX_RETENTION_DAYS	7	retention of the processed rows of eai.inbox_events
EAI_WORKER_INBOX_PURGE_SECONDS	3600	interval of the inbox_purge job of python -m eai.worker
EAI_NATS_URL	(empty)	enables the JetStream wake-ups