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.

Fixed (42862651b). Added a shared _fail_dispatch(payload, error=...) used by the run-raised path and all three early-drop branches (malformed / poison / unknown task), so a dispatch_with_callback failure always self-chains on_error. Tests cover all three drop branches 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.

Fixed (42862651b). The consumer now carries the real error text into callback_kwargs['error'], and _get_task_error prefers it over the (empty-on-PG) AsyncResult lookup. Test asserts RuntimeError: boom reaches the on_error continuation.

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.

Fixed (42862651b). Added _json_safe (json round-trip with default=str) applied to the prepended value in _chain_continuation. Test uses a task returning a datetime and asserts the chained arg is coerced + plain-json serialisable.

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.

Documented (42862651b). Added a comment on the elif on_success branch: "success" == did-not-raise (Celery link parity); a returned failed ExecutionResult deliberately follows on_success — distinct from 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).

Fixed (42862651b). _chain_continuation(spec: ContinuationSpec, *, payload: TaskPayload, ...) and _continuation_org(payload: TaskPayload) now carry the real types.

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).

Fixed (42862651b). message['on_success'] = _json_safe(on_success) (and on_error). Test passes a UUID in callback_kwargs and asserts it's coerced.

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").

Fixed (42862651b, in the now-shared unstract.core.execution_dispatch). Guards both: raises on a falsy task, and raises on positional args (kwargs-only callbacks). Tests for each.

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.

Done — extracted DispatchHandle + signature_to_continuation into unstract.core.execution_dispatch (commit 0f8b57151); both mirrors import them and the duplicated TestSignatureToSpec is gone in favour of one shared test. (Full mirror retirement remains UN-3607.)

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."

Fixed (42862651b). Reworded to '… passes the chained value as the callback's first positional arg, alongside the spec's kwargs as a distinct mapping …'.

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.

Fixed (42862651b). Reworded 'Dark' → 'Idle (receives no work) until …'. Confirmed no broker dep needed: grepped ide_callback/tasks.py — no .delay/.apply_async/send_task/signature/chord; these callbacks only do API writes + ws emits. Added that note to the compose comment.

Correctness — an on_success-only callback dispatch that raises falls through to vt-redelivery and re-runs the executor. The guard here is if reply_key or on_error:, but dispatch_with_callback(on_success=..., on_error=None) is a permitted shape: both kwargs default to None and signature_to_continuation(None) returns None, so an on_success-only callback payload is fully reachable. When such a dispatch's executor raises, this condition is false and control drops into the fire-and-forget branch below, leaving the row for vt-expiry redelivery — exactly the costly re-run (LLM double-spend) the comment on lines 252-258 says this path must avoid — and the caller never receives a terminal event.

Broaden the guard to if reply_key or on_success or on_error: so any reply/callback dispatch acks-on-failure rather than redelivering, or enforce in dispatch_with_callback that on_error is required whenever on_success is set.

Fixed (ec16fefda). Broadened the failure guard to if reply_key or on_success or on_error: — an on_success-only callback dispatch that raises now ACKs instead of vt-redelivering (no executor re-run). Test: on_success-only + raise → delete called, not redelivered.

Silent failure — a failed on_success enqueue leaves the caller with no terminal event. On the success path _chain_continuation is best-effort and never raises: if the enqueue fails (transport/DB error, connection drop, a bad spec["queue"]), it logs and the executor message is acked with no fallback. Since the on_success callback (ide_index_complete / ide_prompt_complete / extraction_complete) is what emits the WebSocket event the prompt-studio UI blocks on after its 202, a dropped success continuation leaves the user with neither a success nor an error event — the session hangs indefinitely, after the LLM spend already occurred.

This is distinct from the now-fixed non-JSON-safe-result drop: _json_safe removes one cause of the enqueue raising, but a transport/DB failure still triggers the same silent drop with no recovery channel. Consider having _chain_continuation return whether it enqueued, and on a success-path failure fall back to _fail_dispatch(payload, error="result delivery failed; see worker logs") so the session still terminates. At minimum, carry run_id/task_id/org_id into the failure log so the stranded session is correlatable later.

Fixed (ec16fefda). _chain_continuation now returns whether it enqueued; on a success-path enqueue failure we fall back to _fail_dispatch(payload, error='result delivery failed; see worker logs') so the caller gets a terminal on_error event instead of hanging. The failure log now carries run_id/task_id/org for correlation. Test asserts the on_error fallback fires when the on_success send fails.