[Medium — correctness / never-raises] row["status"] uses bracket access (KeyError-raising), while result/error below use .get(). This access is not inside the malformed-result try (that only guards from_dict), so a result row missing status would raise straight out of dispatch() and violate the documented never-raises contract.

if (
    row.get("status") == PgTaskStatus.COMPLETED.value
    and row.get("result") is not None
):

Producer (result_backend.get_result) always sets status today, so this is latent — but the contract should not depend on the producer.

Fixed in 6e3e5c6 — row.get("status") now (result/error already used .get), so a status-less row can't raise out of dispatch(). Never-raises no longer depends on the producer writing every key.

[Medium — silent config masking] Guarding the parse so a bad EXECUTOR_RESULT_TIMEOUT can't break the never-raises contract is right, but the swallow is silent: an operator who fat-fingers EXECUTOR_RESULT_TIMEOUT=3O00 gets the 3600s default with no signal, and every RPC may wait up to an hour instead of the intended value. Log the fallback:

except (TypeError, ValueError):
    logger.warning(
        "Invalid %s=%r; falling back to %ss",
        _DEFAULT_TIMEOUT_ENV, os.environ.get(_DEFAULT_TIMEOUT_ENV), _DEFAULT_TIMEOUT,
    )
    timeout = _DEFAULT_TIMEOUT

Fixed in 6e3e5c6 — the EXECUTOR_RESULT_TIMEOUT parse fallback now logs a WARNING (with the bad value) before defaulting to 3600s.

[Medium — substitutability / type design] PgExecutionDispatcher.dispatch(context, timeout) omits the headers param that both the SDK ExecutionDispatcher.dispatch and RoutingExecutionDispatcher.dispatch accept. Today only RoutingExecutionDispatcher is handed out (and it drops headers on the PG path intentionally), so this is latent — but anyone calling a PgExecutionDispatcher directly (the tests do) gets a non-substitutable object, and nothing static catches it. Minimum fix: add headers: dict[str, Any] | None = None (accept-and-ignore) so the dispatch shapes match. Better: define an ExecutorDispatcher Protocol in unstract.core and annotate the SDK base + both PG classes against it — structural, so it preserves the "no SDK edit" constraint while letting mypy verify the duck-type.

Fixed in 6e3e5c6 — added headers: dict | None = None (accept-and-ignore) to PgExecutionDispatcher.dispatch so the shape matches the SDK/Routing dispatch. The shared ExecutorDispatcher Protocol in unstract.core is the broader TODO(shared) (same place the de-dup would land).

[Low] The logger.exception captures the traceback (good), but the caller-facing ExecutionResult.failure only says Malformed executor result for reply_key {reply_key} — the actual parse cause isn't surfaced to a UI reading result.error. For parity with the enqueue/wait paths (which include type(exc).__name__), consider error=f"Malformed executor result ({type(exc).__name__}) for reply_key {reply_key}".

Fixed in 6e3e5c6 — the malformed-result failure now includes type(exc).__name__, matching the enqueue/wait paths.

[Low — doc/concurrency note] The docstring ("structure_tool dispatches are sequential", "one per in-flight RPC") is true within one task, but the file_processing worker runs --pool=prefork with WORKER_FILE_PROCESSING_CONCURRENCY (default 4), so up to 4 processes can each hold a PgResultBackend connection open for the full RPC (up to EXECUTOR_RESULT_TIMEOUT, default 3600s). Bounded by concurrency and not a leak (context manager closes on exit), but it differs from the backend twin, which calls close_old_connections() between polls. Worth a one-line note that the pin is bounded by worker concurrency, or release the connection between polls if file_processing concurrency is ever raised.

Fixed in 6e3e5c6 — added a docstring note that the pin is bounded by file_processing prefork concurrency (default 4), and to release between polls like the backend twin if that concurrency is raised materially. Kept connection-per-wait for now (simpler; PgResultBackend owns + closes it).

[Medium — test gap] _enqueue is mocked in all 11 dispatch tests, so nothing verifies the actual PG-transport wiring: that it builds to_payload(_EXECUTE_TASK, args=[context.to_dict()], queue=queue, reply_key=reply_key) and passes org_id=org to client.send. Since the queue name, the args shape, and org_id are the only routing/identity info on the PG path (Celery headers are dropped), a bug here misroutes or breaks org-fairness with no test catching it. Suggest a test that patches PgQueueClient and asserts client.send is called with the queue, org_id, and a payload carrying task_name/args/reply_key.

Fixed in 6e3e5c6 — added TestPgExecutionDispatcherEnqueueWiring that patches PgQueueClient and asserts client.send gets the queue (celery_executor_legacy), org_id, and a payload carrying task_name/args/reply_key (to_payload runs for real).

[Medium — test gap] The routing tests assert only which sub-dispatcher is called, never the arguments. Two PR claims are unverified: (a) gate-OFF forwards headers=/timeout= through to celery.dispatch unchanged (the zero-regression claim is incomplete without this), and (b) gate-ON calls pg.dispatch without headers (the intentional drop documented just above). Suggest asserting celery.dispatch.assert_called_once_with(..., timeout=9, headers={...}) in the gate-off test and "headers" not in pg.dispatch.call_args.kwargs in the gate-on test.

Fixed in 6e3e5c6 — routing tests now assert arguments: gate-off forwards timeout= and headers= to celery.dispatch unchanged; gate-on calls pg.dispatch with timeout= and "headers" not in kwargs (the intentional drop).

[Medium — test gap] This new reply_key branch is the producer side of the entire RPC contract, but TestToPayload (in test_dispatch_pg.py) predates it and never passes reply_key. A regression that drops or mis-keys the field makes the consumer (payload.get("reply_key")) silently treat the message as fire-and-forget, and the caller blocks until timeout. Add two cases: to_payload("t", reply_key="rk")["reply_key"] == "rk", and "reply_key" not in to_payload("t") (the byte-identical fire-and-forget guarantee the docstring claims).

Fixed in 6e3e5c6 — added two TestToPayload cases: reply_key set when passed, and absent when not (the byte-identical fire-and-forget guarantee).

[Low — test gap] This is the actual call-site swap the PR makes (direct ExecutionDispatcher → get_executor_dispatcher(celery_app=app) returning a RoutingExecutionDispatcher), but no test verifies it. test_structure_tool_task.py only covers _fairness_headers in isolation. If get_executor_dispatcher were mis-imported or returned the wrong type, nothing fails. Suggest patching structure_tool_task.get_executor_dispatcher and asserting _execute_structure_tool_impl calls it with celery_app=app and uses the result as the dispatcher.

Fixed in 6e3e5c6 — added TestDispatcherFactory asserting _execute_structure_tool_impl builds its dispatcher via get_executor_dispatcher(celery_app=app).

[Low — comment clarity] Verified: this import is necessary and not redundant — the PG consumer bootstraps via root worker.py → WorkerBuilder → importlib.import_module("executor.tasks") (WorkerType.to_import_path() returns executor.tasks), which does not import executor/worker.py where import executor.executors historically lived. So the fix is correct. Only the wording can mislead: pg_queue_consumer/__main__.py literally does import worker (the root worker), so "loads tasks.py but not worker.py" reads as contradictory. Suggest clarifying it means executor/worker.py (the Celery entrypoint), e.g. "...loads executor/tasks.py (via the root-worker bootstrap) but not executor/worker.py...".

Fixed in 6e3e5c6 — reworded the comment to name executor/worker.py (the Celery entrypoint the PG-consumer bootstrap skips) instead of the ambiguous worker.py. Thanks for confirming the fix is correct.