High — lost coverage: on_error translation + default task_id are no longer tested.

This is the only dispatch_with_callback test and it passes only on_success plus an explicit task_id="tid-7". The old mirror suites covered two behaviours that are now untested:

1. on_error → signature_to_continuation → forwarded as on_error= (shared module executor_rpc.py:306,312). A regression dropping on_error would pass CI — and on-error continuations are how PG-path failures notify the IDE/callback.
2. auto-generated task_id when omitted (task_id or str(uuid.uuid4()), shared module :302).

Fix: also pass on_error=<second MagicMock signature> and assert kw["on_error"]["task_name"]; add a test_dispatch_with_callback_defaults_task_id that calls with no task_id and asserts handle.id is truthy and equals t.enqueue_calls[0]["task_id"].

Fixed (ae5f99749). test_dispatch_with_callback_translates_both_signatures now passes on_error and asserts kw['on_error']['task_name']; added test_dispatch_with_callback_defaults_task_id (no task_id → uuid on handle == payload task_id).

High (highest-leverage type fix) — declare conformance to this Protocol on both transports.

QueueTransport is a structural Protocol, but neither DjangoQueueTransport (backend/pg_queue/executor_rpc.py:62) nor PgClientQueueTransport (workers/queue_backend/pg_queue/executor_rpc.py:62) inherits from it. Conformance is therefore only checked at the PgExecutionDispatcher(...) construction site — a signature drift on one side (e.g. renaming org_id) wouldn't surface until runtime, on that side only.

Fix: have both transport classes explicitly subclass QueueTransport. Protocols support explicit inheritance at zero runtime cost, and mypy/pyright will then check each implementation against the seam independently. Cheapest way to keep the seam well-typed across both trees.

Fixed (ae5f99749). Both DjangoQueueTransport and PgClientQueueTransport now explicitly inherit QueueTransport, so mypy/pyright check each implementation against the seam independently (zero runtime cost — not @runtime_checkable, just static conformance).

Medium — on an enqueue failure here, the carried on_error continuation never fires.

Unlike dispatch (guarded, never raises), dispatch_with_callback calls self._transport.enqueue(...) unguarded, so a transport error (DB down, psycopg2.OperationalError, JSON-serialization error in to_payload) propagates raw. This matches the SDK parity contract, so it's not a regression — but note the hazard: when continuations are carried in the payload and the enqueue itself fails, the executor never runs and the on_error continuation is never fired. Cleanup only happens because call sites wrap dispatch in their own try/except (e.g. prompt_studio/.../views.py:461).

Fix (no behaviour change): log-before-raise so the failure is observable in Sentry even when a caller swallows it, and add a one-line note to the docstring stating callers MUST treat a raised enqueue as the on_error path because the carried on_error will not fire:

try:
    self._transport.enqueue(...)
except Exception:
    logger.exception("PG executor dispatch_with_callback: enqueue failed (executor=%s run_id=%s)", context.executor_name, context.run_id)
    raise

Fixed (ae5f99749, no behaviour change). dispatch_async + dispatch_with_callback now log-before-raise on enqueue failure (observable in Sentry even if a caller swallows it), and the docstring states a raised enqueue IS the failure signal since on_error can't fire (the prompt-studio views already handle this in their try/except).

Medium — a COMPLETED row with result is None is silently reported as a generic task failure.

The success branch requires status == COMPLETED.value AND result is not None; anything else falls through to return ExecutionResult.failure(error=row.error or "executor task failed") at line 265. So a producer bug (writes COMPLETED but no result, or a typo status) is indistinguishable from a genuine executor failure — there's no diagnostic that the row was malformed. ExecResultRow.status is also str | None (line 78), so the enum invariant is never enforced at the read boundary.

Fix: add an explicit branch mirroring the malformed-parse handling at lines 242-257 — if status == COMPLETED.value and row.result is None, log + return a "malformed completed row (missing result)" failure. Optionally validate status against PgTaskStatus and warn on an unknown value. Keeps the fail-safe direction while making the producer-contract violation observable.

Fixed (ae5f99749). A COMPLETED row with result is None now returns a distinct Executor reported completion with no result failure (+ a warning log), not the generic executor task failed.

Medium — on_success/on_error typed Any | None discards the real Signature contract.

signature_to_continuation reads .task, .options["queue"], .args, .kwargs and raises ValueError if .task/queue are missing — so the real precondition is "a Celery-Signature-like object." Typing these Any is actually weaker than the SDK method this mirrors (sdk1/.../dispatcher.py types them Signature | None), and lets any value through to a runtime ValueError.

Fix: define a small structural Protocol (e.g. class CallbackSignature(Protocol): task: str; options: dict; args: tuple; kwargs: dict) next to signature_to_continuation in unstract.core.execution_dispatch, and type both that helper and these params as CallbackSignature | None. Documents and checks the precondition at the seam.

Fixed (ae5f99749). Added a CallbackSignature(Protocol) (task/options/args/kwargs) in unstract.core.execution_dispatch and typed both signature_to_continuation and the dispatch_with_callback params with it — celery-free, a real Signature conforms structurally.

Low — confirmed ruff I001 failure (CI enforces I). The first-party pg_queue.* imports (lines 22-23) precede the third-party unstract.* imports (lines 24-31); isort wants the local group last. Verified failing via ruff check.

Fix: run ruff check --fix (or move the two from pg_queue... lines below the unstract.workflow_execution import block).

Addressed (ae5f99749). Note: uvx ruff (latest) and the repo's pinned pre-commit ruff disagree on first-party classification for this file — pre-commit (the CI authority) reformatted the import order on commit, so it now matches what CI enforces.

Low — confirmed ruff D209 failure (CI enforces D). A multi-line docstring must have its closing """ on its own line. Verified failing via ruff check.

Fix: move the closing quotes to a new line (or ruff check --fix):

    """Records ``enqueue`` calls and returns a configured result for the poll —
    so the shared dispatcher's logic is tested without any DB.
    """

Fixed (ae5f99749). Moved the _FakeTransport docstring closing quotes to their own line (also fixed the same on _CeleryDispatcher).

Low — the master-gate env parse fails silently on a fat-fingered value.

os.environ.get(_MASTER_GATE_ENV, "false").lower() == "true" resolves anything other than exactly "true" ("1", "yes", "on", "True " with whitespace) to False, leaving the whole PG transport off with no log line. An operator rolling out with PG_QUEUE_TRANSPORT_ENABLED=1 would see it silently do nothing. This is parity with the original (not a regression), but it's the one gate input that fails quietly — contrast the carefully-logged _resolve_timeout and the FLIPT_SERVICE_AVAILABLE warning.

Fix: warn when the var is set but doesn't parse to a recognized token, e.g. if os.environ.get(_MASTER_GATE_ENV) is non-empty and not in {"true","false"}.

Fixed (ae5f99749). The workers master-gate now warns when PG_QUEUE_TRANSPORT_ENABLED is set to anything other than 'true'/'false' (e.g. '1'/'yes'/'on') — no longer a silent no-op.

Low — celery: Any leaves the three duck-typed delegations unchecked.

The class is "duck-typed against the SDK ExecutionDispatcher" but holds it as Any, so the assumption that celery exposes dispatch/dispatch_async/dispatch_with_callback is unverified. Both factories already pass a real ExecutionDispatcher (backend/pg_queue/executor_rpc.py:119, workers/queue_backend/pg_queue/executor_rpc.py:113), and this module already imports it is available via unstract.sdk1.execution.dispatcher.

Fix: type celery: ExecutionDispatcher (import it here) — turns three duck-typed delegations into checked ones. Optionally type the routing dispatch_with_callback return (line 379) as a HasId/DispatchHandle-shaped protocol instead of Any.

Fixed (ae5f99749). Added a _CeleryDispatcher(Protocol) with the three delegated methods; the router now holds celery: _CeleryDispatcher instead of Any.

Low — the documented dispatch_async "enqueue failures propagate" contract is asserted nowhere.

The shared dispatch_async docstring (executor_rpc.py:272-273) deliberately lets enqueue errors propagate — an intentional asymmetry with the never-raises dispatch. This test covers only the happy path, so a future "consistency" refactor could wrap the enqueue in try/except and silently break the contract (callers rely on the raise to know the task was never enqueued).

Fix: add test_dispatch_async_propagates_enqueue_failure using _FakeTransport(enqueue_raises=RuntimeError(...)) and pytest.raises(RuntimeError) around PgExecutionDispatcher(t).dispatch_async(_ctx()).

Fixed (ae5f99749). Added test_dispatch_async_propagates_enqueue_failure (pytest.raises) — pins the intentional asymmetry with the never-raises dispatch.