Critical — on_error callback dropped on the early-drop branches. This new block correctly self-chains on_error when the run raises, but the three early-drop branches above — malformed payload, poison/max_attempts exceeded, and unknown/unregistered task — only call _fail_reply(reply_key, ...) and then delete(). For a dispatch_with_callback message (no reply_key, has on_error) those branches delete the row and never chain on_error. A poison executor task is exactly the realistic failure mode, and it's the case where the user most needs the error surfaced; instead the HTTP-202 caller waits on a WebSocket error event that never arrives (run appears to hang, no terminal state). This is asymmetric with the reply_key path, which does fail those branches.

Fix: in each early-drop branch, in addition to _fail_reply, chain the error continuation when present, e.g. a helper _fail_dispatch(payload, error=...) that does _fail_reply(reply_key, ...) and if on_error: self._chain_continuation(on_error, prepend=payload.get('task_id') or '', payload=payload) before the delete(). Add tests for the unknown-task / poison / malformed cases with on_error set.

Important — PG on_error loses the real executor error text. prepend=payload.get('task_id') hands the callback (e.g. ide_index_error) the dispatch task_id as failed_task_id, but that callback resolves the actual message via AsyncResult(failed_task_id, app=app).result. On the PG path the executor ran eagerly via task.apply and never wrote to a Celery result backend under that id, so AsyncResult(...).result is empty and the callback degrades to its generic default error string. Parity is structural (right positional shape) but not behavioural — the real error is dropped.

Fix: the consumer has the exception in hand at the failure branch — carry the real error text into the on_error continuation (e.g. prepend f"{type(exc).__name__}: {exc}", or pass it via the continuation kwargs), or store the error under task_id in a backend the callback can read.

Important — success callback silently dropped on non-JSON-safe results. prepend=eager.result is enqueued as args=[eager.result]; client.send serialises with plain json.dumps(message) (no default=str). An executor result dict containing a UUID/datetime — exactly the kind of payload these results carry — makes json.dumps raise TypeError, which _chain_continuation's broad except swallows ("FAILED to self-chain"), so the success callback and its user-facing WebSocket event are lost. The Celery path serialises via its configured serializer and has no such gap.

Fix: JSON-coerce the prepended result before enqueuing (mirror backend/pg_queue/producer.py:_json_safe), or give client.send's json.dumps a default=str.

Suggestion — a returned ExecutionResult(success=False) chains on_success. The success/error split keys purely on whether task.apply(...) raised, but execute_extraction can return a failed ExecutionResult (success=False, via ExecutionResult.failure(...)) without raising. That takes this elif on_success branch and chains the success continuation with a failure payload; on_error never fires. This matches Celery semantics (link_error also only fires on raised exceptions), so it's parity rather than a regression — but given the user-facing impact, either inspect eager.result.success is False and route to on_error, or add a comment documenting that returned-failure deliberately follows the success path, so a future reader doesn't mistake it for the drop-branch bug above.

Suggestion — restore the typed contract at the consumer boundary. _chain_continuation(self, spec: dict, *, prepend, payload: dict) re-widens to bare dict, discarding the new ContinuationSpec / TaskPayload types the PR added precisely for these shapes. A key rename across the producer↔consumer boundary (e.g. task_name → name) wouldn't be caught by the type checker. Annotate spec: ContinuationSpec, payload: TaskPayload (imports already available in this tree).

Important — on_success/on_error bypass _json_safe. args/kwargs go through _json_safe(...) above (lines 96-97), but the continuation specs (each with a nested kwargs dict) are written into the JSONField verbatim. If a callback's kwargs carry a UUID/datetime, PgQueueMessage.objects.create raises at insert time — and unlike the worker path this is caller-visible at dispatch. Fix: message["on_success"] = _json_safe(on_success) (and likewise on_error).

Important — _signature_to_spec validates queue but not task, and silently drops positional args. Two gaps with the same fail-fast philosophy already applied to queue:

1. task_name=sig.task is read with no guard. If sig.task is empty/None (a malformed signature), the continuation enqueues with a falsy task_name and is dropped far downstream as "malformed payload: missing task_name" — a lost callback with a misleading log. Add if not getattr(sig, 'task', None): raise ValueError(...).
2. ContinuationSpec has no args, and only sig.kwargs is read. A Celery Signature may carry positional args; on the PG branch they're silently dropped, so the callback runs with a different arg list than on Celery — violating the "call sites unchanged" contract. Today's call sites are kwargs-only (latent), but make it loud: if getattr(sig, 'args', None): raise ValueError("callback signature has positional args; PG self-chaining supports kwargs only").

Suggestion — _DispatchHandle and _signature_to_spec are duplicated byte-for-byte with backend/pg_queue/executor_rpc.py. Both files now carry identical copies of this class and the _signature_to_spec helper (and near-identical dispatch_async/dispatch_with_callback bodies). The new translation logic — especially the correctness-bearing "fail fast on missing queue" rule — can now drift silently between the two copies. ContinuationSpec/PgTaskStatus were hoisted into unstract.core for exactly this reason; _DispatchHandle and _signature_to_spec (no Django/worker deps) belong there too. Import both copies from one shared definition.

Suggestion — "before kwargs" is slightly misleading. _chain_continuation passes the chained value as a separate positional args=[prepend] list and kwargs as a distinct mapping — they aren't concatenated into one positional sequence, so "before kwargs" reads as if kwargs were positional. Reword to e.g.: "… passes the chained argument as the callback's first positional arg, alongside the spec's kwargs — mirroring how Celery's link prepends the parent task's return value."

Suggestion — "Dark" is jargon for a compose file. Clear to the authors (idle / no traffic) but opaque to an operator scanning services. Suggest: "Idle (receives no work) until dispatch_with_callback routes to PG, i.e. the gating flag is on". Also worth confirming: the sibling worker-pg-callback/worker-pg-scheduler declare depends_on: rabbitmq; this service doesn't. Likely fine since ide_callback tasks only do API writes + ws emits, but verify none dispatch onward to a Celery broker.