Critical — unguarded success-path store re-runs the executor and blocks the caller.

_store_reply(...) here runs before the ack (delete) at line 251 and is not wrapped in try/except. If store_result raises (DB write error, serialization failure, dropped connection), the exception propagates out of _handle and is caught only by the generic handler in poll_once, logged as a generic "poll failed" with no reply_key/task_name. Consequences:

• the message is never deleted → vt expires → the executor re-runs the full extraction (real LLM spend) up to max_attempts, then is dropped as poison;
• the waiting caller never gets a row and blocks to its full timeout (default 3600s);
• the failure is mislabeled, contradicting the line 219-225 promise that the caller "gets a definitive result instead of blocking to its timeout" (that only holds for the task-raised branch, not a store failure).

Also latent: store_result classifies any non-dict / None result as STATUS_FAILED, so a request-reply task that legitimately returns None/empty would be mis-stored as failed (not reachable today since execute_extraction always returns a dict, but fragile for future tasks).

Fix: wrap the store with try/except logging reply_key/task_name/msg_id at error severity, and decide explicitly whether a store failure should redeliver. This branch is also entirely untested — test_pg_queue_consumer.py was not updated; add ok+reply_key (store result then ack) and boom+reply_key (store error then ack) cases.

Fixed in dce4e1c — the success-path store is now wrapped: on failure it logs at ERROR (task/msg_id/reply_key) and acks anyway to avoid re-running the executor (LLM spend); the caller degrades to a timeout rather than a poison-loop. The task-raised branch and the drop branches go through a new guarded _fail_reply. Added consumer tests: success→store+ack, raise→store-error+ack, store-failure→still-acks.

Medium — pre-execution drop branches lose the reply, blocking the caller to full timeout.

reply_key is read here, but the earlier drop branches above (missing task_name, poison read_ct > max_attempts, unknown task) delete the row and return without storing any reply. For a request-reply message that hits one of those paths, the blocking caller waits the full timeout (default 3600s) for a row that will never appear. For the unknown-task and poison cases especially, a definitive failure reply could be stored so the caller fails fast.

Fix: in those drop branches, if the payload carries a reply_key, call _store_reply(reply_key, error=...) before deleting so the caller gets an immediate descriptive failure.

Fixed in dce4e1c — reply_key is read up front and the drop branches (missing task_name, poison read_ct, unknown task) now call _fail_reply(reply_key, …) before deleting, so a request-reply caller gets an immediate descriptive failure instead of blocking to its timeout. Tested (unknown-task → store error + ack).

Medium — magic status string + unguarded from_dict can break the never-raises contract.

Two issues on this branch:

1. row.status == "completed" is a bare literal here, while the writer side names it STATUS_COMPLETED in result_backend.py. The two live in different deployables (backend Django ORM reader vs workers psycopg2 writer) with no shared import, so the contract is matched by eye across a process boundary — rename the writer constant and this silently treats every result as a failure. Consider promoting the status vocabulary to a StrEnum in unstract.core (which both trees already depend on).
2. ExecutionResult.from_dict(row.result) hard-indexes data["success"]. A malformed/partial completed row raises KeyError, which is not caught — violating this class's documented "never raises" contract and surfacing as a 500 to prompt-studio.

Fix: branch on the shared enum, and wrap from_dict in try/except converting a decode error to ExecutionResult.failure(...).

Fixed in dce4e1c — both: (1) added a shared PgTaskStatus(str, Enum) in unstract.core.data_models; the reader branches on PgTaskStatus.COMPLETED.value and the writer re-exports STATUS_COMPLETED/FAILED from it → one source of truth across the process boundary. (2) from_dict is wrapped in try/except so a malformed completed row becomes ExecutionResult.failure(...), preserving the never-raises contract.

High (test gap) — PgExecutionDispatcher.dispatch is entirely untested.

test_executor_rpc.py mocks PgExecutionDispatcher out, so none of this method's four outcome branches ever run, yet its contract ("never raises; failure/timeout → ExecutionResult.failure") is the load-bearing promise prompt-studio depends on.

Add DB-free tests (mock enqueue_task and _wait_for_result): enqueue raises → returns failure and does not re-raise; _wait_for_result returns None → timeout failure with the within {timeout}s message; completed row → ExecutionResult.from_dict(row.result); failed row → failure with row.error (and the empty-error → "executor task failed" fallback); completed-but-result is None falls through to failure; timeout=None reads EXECUTOR_RESULT_TIMEOUT else 3600. Also cover _wait_for_result backoff + the min(delay, remaining) deadline clamp.

Fixed in dce4e1c — added TestPgExecutionDispatcherDispatch (8 DB-free cases, mocking enqueue_task + _wait_for_result): enqueue-raises→failure (no re-raise); timeout→failure with the within {timeout}s message; completed→from_dict; failed→row.error; empty-error→fallback; completed-but-result-None→failure; malformed-completed→failure-not-raise; timeout=None→EXECUTOR_RESULT_TIMEOUT. The _wait_for_result backoff/clamp is the same logic as PgResultBackend.wait_for_result, which has real-PG timeout + late-write tests.

Medium — timeout is silent and indistinguishable from a real failure.

On timeout the caller gets ExecutionResult.failure("TimeoutError: ..."), but the executor task may still be running (and spending) and the row may land just after the deadline. Neither the timeout branch nor the task-failed branch (line 162) logs anything, so a 3600s blocking RPC can give up with no operator-visible event.

Fix: log the timeout and the task-failed branches at WARNING/ERROR with reply_key/run_id/elapsed so a caller-side timeout can be correlated with the still-running/failed pg_task_result row.

Fixed in dce4e1c — the timeout branch and the executor-failure branch now log at WARNING with reply_key/run_id, so a caller-side timeout/failure is operator-visible and correlatable with the pg_task_result row.

Medium (comment accuracy) — describes a DELETE that doesn't exist.

This index comment says it "Drives the reaper's retention sweep (DELETE ... WHERE expires_at <= now())", but no such sweep consumes pg_task_result.expires_at — the index currently backs no query. Reword to e.g. "Index for a future retention sweep; no sweeper reads it yet." (Same issue at line 318: "NULL = no expiry recorded yet" describes a state no code produces — the only writer always sets expires_at.)

Fixed in dce4e1c — reworded the index + expires_at comments to "index for a future retention sweep; no sweeper reads it yet".

Medium — stale flag name contradicts the code and the PR's own goal.

This docstring says "when pg_executor_enabled is on", but no such flag exists — the dispatcher reads PG_QUEUE_FLAG_KEY = "pg_queue_enabled" (executor_rpc.py:62), which is exactly the unification this PR makes. An operator following this comment would look for a flag that isn't there.

Fix: change pg_executor_enabled → pg_queue_enabled. Also: _get_dispatcher (line 290) dropped its return annotation; the prior signature was -> ExecutionDispatcher. Consider annotating -> "RoutingExecutionDispatcher" (under TYPE_CHECKING) for consistency with the typed surroundings.

Fixed in dce4e1c — pg_executor_enabled → pg_queue_enabled in the docstring. (Left _get_dispatcher un-annotated to avoid a forward-ref import dance; the routing dispatcher is duck-typed against the SDK surface.)

Medium — stale flag name. "Dark until the backend pg_executor_enabled gate is flipped" references a non-existent flag. The worker-pg-executor service is dark until the unified pg_queue_enabled flag (read by resolve_executor_transport) is flipped. Change pg_executor_enabled → pg_queue_enabled.

Fixed in dce4e1c — pg_executor_enabled → pg_queue_enabled in the compose comment.

Low — reply_key: NotRequired[str | None] has three representations for two meanings.

The key can be absent, present-and-None, or present-and-string, but only two are meaningful (fire-and-forget vs request-reply). The producer only omits it or sets a string, and the consumer reads .get("reply_key") which collapses absent and None. The | None arm is dead-but-legal and reintroduces the "two empty states" problem the PgTaskResult.error column docstring explicitly set out to avoid.

Fix: tighten to reply_key: NotRequired[str] — "request-reply" is then exactly "key present."

Fixed in dce4e1c — tightened to reply_key: NotRequired[str] so "request-reply" is exactly "key present" (no dead | None arm).

Low — retention is decoupled from the caller's actual timeout.

DEFAULT_RETENTION_SECONDS = 3600 equals the executor caller-timeout default, but a caller can raise EXECUTOR_RESULT_TIMEOUT above 3600 (PgExecutionDispatcher.dispatch honours it). The docstring's promise that a result "always outlives any caller still waiting on it" then breaks: if the retention sweep is added (see models.py:304), a result could be deleted while a caller is still polling → spurious timeout. These two constants live in separate packages with no shared source.

Fix: when the sweep lands, derive retention from max(EXECUTOR_RESULT_TIMEOUT, DEFAULT_RETENTION_SECONDS) plus a margin (or pass the caller timeout through the payload), and document the retention >= caller_timeout invariant.

Acknowledged — deferred with the sweep (models.py:304). With no sweep yet, the decoupling is currently moot (nothing deletes a row out from under a caller). When the sweep lands I'll derive retention from max(EXECUTOR_RESULT_TIMEOUT, default) + margin and document the retention >= caller_timeout invariant. Tracked in the same follow-up.

The return type annotation was dropped when the dispatcher was changed from ExecutionDispatcher to RoutingExecutionDispatcher. Adding it restores static-analysis coverage and makes the duck-typed interface explicit to readers.

This is a comment left during a code review.
Path: backend/prompt_studio/prompt_studio_core_v2/prompt_studio_helper.py
Line: 289-291

Comment:
The return type annotation was dropped when the dispatcher was changed from `ExecutionDispatcher` to `RoutingExecutionDispatcher`. Adding it restores static-analysis coverage and makes the duck-typed interface explicit to readers.

```suggestion
    @staticmethod
    def _get_dispatcher() -> "RoutingExecutionDispatcher":
        """Executor dispatcher for the executor worker.
```

How can I resolve this? If you propose a fix, please make it concise.

Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!