workers/shared/infrastructure/config/builder.py (1)

99-106: QoS override logic is correct but duplicated with celery_config.py.

The monkey-patch correctly forces per-consumer prefetch semantics for quorum queue compatibility. The unused lambda arguments (self, conn) are intentional since the override unconditionally returns True.

However, this logic is duplicated in backend/backend/celery_config.py (lines 84-90). Consider extracting a shared utility to avoid drift:

# shared/utils/rabbitmq_ha.py
def patch_qos_for_quorum_queues():
    if os.environ.get("RABBITMQ_HA_ENABLED", "").lower() == "true":
        from kombu.transport import pyamqp
        pyamqp.Transport.qos_semantics_matches_spec = lambda self, conn: True

This is acceptable for now since backend and workers may be deployed separately, but the duplication increases maintenance burden.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@workers/shared/infrastructure/config/builder.py` around lines 99 - 106,
Extract the duplicated RabbitMQ HA monkey-patch into a shared helper (e.g.
patch_qos_for_quorum_queues) and call it from both
workers/shared/infrastructure/config/builder.py and
backend/backend/celery_config.py; specifically move the logic that sets
pyamqp.Transport.qos_semantics_matches_spec = lambda self, conn: True behind a
single function (ensuring it still checks
os.environ.get("RABBITMQ_HA_ENABLED").lower() == "true") and replace the inline
monkey-patch in both places with an import and invocation of that helper.

backend/backend/celery_config.py (1)

69-82: Hardcoded queue list may drift from actual queue usage.

The _BACKEND_QUEUES list is manually maintained. If new queues are added to backend task routing elsewhere, they must also be added here for HA compatibility. Consider:

1. Adding a comment noting where this list should be kept in sync with
2. Or, deriving the list from a single source of truth (e.g., a shared constant or Django settings)

The commit history shows dashboard_metric_events was added in a follow-up fix, indicating this list can be easily missed during updates.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@backend/backend/celery_config.py` around lines 69 - 82, The hardcoded
_BACKEND_QUEUES list used when is_rabbitmq_ha_enabled() runs can drift from
actual task routing; replace or centralize it by deriving the queue names from a
single source of truth and update CeleryConfig.task_queues = [make_queue(q) for
q in _BACKEND_QUEUES] accordingly: either import the canonical queue list from
existing routing/constants in your project (e.g., a shared QUEUES constant or
Django settings) and use that instead of the inline _BACKEND_QUEUES, or add a
clear comment pointing to the canonical location and a test to assert parity
between the canonical list and _BACKEND_QUEUES (reference symbols:
_BACKEND_QUEUES, is_rabbitmq_ha_enabled, CeleryConfig.task_queues, make_queue).

backend/backend/celery_config.py (1)

81-87: Duplicated QoS override logic with builder.py.

This identical monkey-patch exists in workers/shared/infrastructure/config/builder.py:99-107. While double-patching the same value is harmless, maintaining duplicate logic increases risk of divergence. Consider extracting to a shared utility (e.g., a function in a common module) that both locations can call.

The static analysis warnings about unused lambda arguments (self, conn) are false positives—these parameters are required by the method signature being overridden.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@backend/backend/celery_config.py` around lines 81 - 87, The QoS monkey-patch
(pyamqp.Transport.qos_semantics_matches_spec = lambda self, conn: True) is
duplicated in builder.py; extract it to a single shared helper (e.g., a function
like enable_quorum_qos_compatibility() in a common utilities module) and replace
both occurrences to call that helper from celery_config.py and
workers/shared/infrastructure/config/builder.py; ensure the helper performs the
same override using the full method signature (accepting self and conn) to avoid
static analysis false positives and keep behavior identical.

workers/shared/models/worker_models.py (1)

505-525: Consider consolidating quorum queue utilities.

This method duplicates the environment check and quorum queue creation logic found in backend/backend/celery_config.py (is_rabbitmq_ha_enabled() and make_queue()). If a shared utilities module is feasible between backend and workers, consolidating this logic would reduce maintenance burden and ensure consistent behavior.

The implementation is correct—the lazy import (line 519) and sorted queue list (line 524) are good practices.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@workers/shared/models/worker_models.py` around lines 505 - 525, This
duplicates the RabbitMQ HA env check and quorum queue creation logic; extract
the logic into a shared helper (e.g., is_rabbitmq_ha_enabled() and
make_queue(queue_name) used by backend) and call it from
_add_quorum_queue_configuration: replace the inline os.environ check with
is_rabbitmq_ha_enabled(), import and use make_queue(queue_name) (or a shared
wrapper that returns a kombu.Queue with {"x-queue-type":"quorum"}) for each name
from self.queue_config.all_queues(), preserving the lazy import behavior and
sorted order.

workers/shared/worker_builder.py (2)

88-88: Use underscore prefix for intentionally unused lambda parameters.

The lambda arguments are unused because this is a method signature override. Using underscore prefix makes the intent explicit and silences linters (ARG005).

✏️ Suggested fix

-            pyamqp.Transport.qos_semantics_matches_spec = lambda self, conn: True
+            pyamqp.Transport.qos_semantics_matches_spec = lambda _self, _conn: True

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@workers/shared/worker_builder.py` at line 88, The override of
pyamqp.Transport.qos_semantics_matches_spec uses an unused lambda parameter
named conn; change the lambda to use an underscore-prefixed parameter (e.g.,
_conn) to signal intentional non-use and satisfy linters (ARG005) when
overriding qos_semantics_matches_spec on the pyamqp.Transport class.

---

75-88: Duplicated HA configuration logic across modules.

This block is identical to lines 99-112 in workers/shared/infrastructure/config/builder.py. Both files implement the same quorum queue and QoS override logic. Additionally, backend/backend/celery_config.py already defines is_rabbitmq_ha_enabled() and make_queue() helpers that should be reused.

Consider extracting a shared helper to avoid drift between these implementations:

♻️ Suggested approach: create shared helper

Create a shared utility in the workers package:

# workers/shared/utils/rabbitmq_ha.py
import os
from kombu import Queue

def is_rabbitmq_ha_enabled() -> bool:
    return os.environ.get("RABBITMQ_HA_ENABLED", "").lower() == "true"

def configure_quorum_queues(app, queue_names: set[str]) -> None:
    """Configure quorum queues and QoS semantics for HA mode."""
    if not is_rabbitmq_ha_enabled():
        return
    
    from kombu.transport import pyamqp
    
    quorum_args = {"x-queue-type": "quorum"}
    app.conf.task_queues = [
        Queue(q, queue_arguments=quorum_args)
        for q in sorted(queue_names)
    ]
    pyamqp.Transport.qos_semantics_matches_spec = lambda _self, _conn: True

Then call configure_quorum_queues(app, worker_celery_config.queue_config.all_queues()) in both builder modules.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@workers/shared/worker_builder.py` around lines 75 - 88, Extract the
duplicated RabbitMQ HA logic into a shared helper (e.g., is_rabbitmq_ha_enabled
and configure_quorum_queues) and replace the inline block in worker_builder.py
that declares quorum queues and overrides QoS with a call to that helper;
specifically, move the quorum_args/Queue creation and the
pyamqp.Transport.qos_semantics_matches_spec override into
configure_quorum_queues(app, queue_names) and call it with
worker_celery_config.queue_config.all_queues(), or reuse the existing
is_rabbitmq_ha_enabled()/make_queue() helpers from backend.backend.celery_config
if preferred, ensuring the original behavior (sorting queue names, setting
app.conf.task_queues, and setting qos_semantics_matches_spec) is preserved.