…rier wrapper + lift chord call sites + plumb fairness

First sub-task of Phase 6 (UN-3504 — Distributed Barrier). Ships the
abstraction layer + chord-call-site lift + fairness plumbing with
**zero runtime behaviour change** — every operation still routes through
``celery.chord(...)`` under the hood. The risky behaviour-change PR
(``RedisDecrBarrier`` replacing chord_unlock polling) is deliberately
deferred to Phase 6b behind a feature flag.

## What this adds

* ``workers/queue_backend/barrier.py``
  * ``Barrier`` Protocol — minimum contract every barrier substrate
    satisfies (``enqueue(header, callback, fairness=None)`` shape).
  * ``BarrierHandle`` Protocol — minimum return-value contract
    (``id: str``); celery's ``AsyncResult`` satisfies this; future
    substrate handles must too so chord-id logging stays working.
  * ``CeleryChordBarrier`` impl — wraps ``chord(header)(body)`` 1:1.
    Optional ``FairnessKey`` is stamped as a message header on every
    enqueued task AND the callback signature.

* Both production chord call sites lifted behind the barrier:
  * ``shared/workflow/execution/orchestration_utils.py:67`` —
    ``WorkflowOrchestrationUtils.create_chord_execution`` now
    delegates to ``CeleryChordBarrier``. Method signature unchanged
    for callers; adds optional ``fairness=`` kwarg.
  * ``api-deployment/tasks.py:674`` — inline ``chord(...)`` replaced
    with a call to ``WorkflowOrchestrationUtils.create_chord_execution``
    with ``fairness=FairnessKey(workload_type=WorkloadType.API)``.
  * ``general/tasks.py:941`` — existing chord call site passes
    ``fairness=FairnessKey(workload_type=WorkloadType.NON_API)``.

* Inventory canary in ``tests/test_chord_sites_characterisation.py``
  tightened: was "exactly 2 chord call sites in workers/", now
  "exactly 1 chord call site, inside ``queue_backend/barrier.py``".
  A future PR that adds a direct ``chord(...)`` call outside the
  barrier abstraction fails this test loudly.

## What this does NOT change (zero-behaviour-change posture)

* The ``chord(header)(body)`` Celery primitive itself — same two-step
  call, same chord_unlock polling, same result backend interaction,
  same retry semantics, same callback firing.
* Task execution — same workers consume the same queues.
* Result aggregation — chord still collects all header return values
  into a list and passes them to the body.
* Failure handling — same chord error semantics, same link_error path.
* Mixed-version rolling deploys — both directions safe:
  * old worker enqueues chord (no fairness header) → new callback
    receives → callback doesn't read fairness header → works
  * new worker enqueues chord (with fairness header) → old callback
    receives → celery passes unknown headers through → works

## Why plumb fairness here

Phase 5.1 added fairness to bare ``dispatch()`` call sites only (2 of
them today). The two ``chord(...)`` call sites bypass ``dispatch()``
entirely — at runtime, MOST workflow-execution tasks fan out through
chord (process_file_batch × N as header tasks), so Phase 5.1 only
covered ~5% of the workflow-execution wire traffic. This PR closes
the remaining ~95%.

Doing it at the barrier lift point (rather than retroinstrumenting the
raw chord call) means future Phase 6b can swap in a ``RedisDecrBarrier``
or ``PgBarrier`` without touching call sites a second time.

## Tests

New ``tests/test_barrier.py`` (12 tests):
* Barrier Protocol shape — ``CeleryChordBarrier`` satisfies the
  ``Barrier`` Protocol; AsyncResult-shaped handles satisfy
  ``BarrierHandle``.
* Wire equivalence with the pre-uplift chord call — given the same
  args, ``CeleryChordBarrier.enqueue(...)`` produces the same
  ``chord(header)(body)`` invocation, same ``app.signature(...)``
  args, same return value. Locks the byte-for-byte equivalence claim.
* Fairness header plumbing — when ``fairness=`` is passed, the
  fairness slot is attached to every enqueued task and the callback.
  When ``fairness=None``, no ``headers=`` kwarg is added (preserves
  byte-for-byte pre-uplift wire).
* Mixin wrapper invariants preserved — ``WorkflowOrchestrationMixin
  .create_chord`` still extracts ``self.app`` and raises on missing
  app context.

Updated ``tests/test_chord_sites_characterisation.py``:
* Existing 6 characterisation tests still pass — patch targets
  updated to ``queue_backend.barrier.chord``.
* Inventory canary tightened: 2 sites → 1 site (inside barrier.py).
* ``from celery import chord`` canary: 2 files → 1 file.

Updated ``tests/test_queue_backend_seam.py::test_all_exports`` for the
new public surface (``Barrier``, ``BarrierHandle``,
``CeleryChordBarrier``).

## Validation

* Touched test files: 21 / 21 passing
* Full workers suite: 6 failed (pre-existing baseline, unchanged) /
  659 passed (was 627 → +32 new tests)
* Five deterministic-order full-suite runs: exactly 6/659 each time
  — zero flakiness from this change

## What's deferred

* **Phase 6b (UN-3523):** ``RedisDecrBarrier`` impl + ``WORKER_BARRIER_BACKEND``
  env flag (default ``chord`` — opt-in). New code path gated.
* **Phase 6c (UN-3524):** production rollout / verification (config-only,
  no code changes).

## Risk

* Wire shape change: each chord member signature + callback signature
  now carries a fairness header (~80 bytes / task). Additive — celery
  passes unknown headers through; no consumer reads the slot yet.
* All other behaviour identical to direct ``chord(...)`` calls.
* Mixed-version rolling deploy safe (verified above).
* Rollback: a single revert commit removes the abstraction; both call
  sites are one-line delegations.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…loud

SonarCloud flagged 7.9% duplicated lines on workers/tests/test_barrier.py
(Phase 6a barrier characterisation suite). Three sources:

1. ``TestWorkflowOrchestrationMixinCreateChord`` class was defined in
   both ``test_barrier.py`` AND ``test_chord_sites_characterisation.py``
   (same 2 tests, byte-for-byte). The mixin lives on the chord-call-site
   characterisation surface, so removed the dup from ``test_barrier.py``
   — coverage is preserved by the copy in
   ``test_chord_sites_characterisation.py``.

2. ``_make_app`` helper was defined twice (one per test class) with the
   same body. Extracted to a module-level ``app`` pytest fixture used
   by every test that drives ``CeleryChordBarrier.enqueue(...)``.

3. ``with patch("queue_backend.barrier.chord") as mock_chord:`` appeared
   in 8 separate tests with identical patch target. Extracted to a
   ``mock_chord`` pytest fixture that patches once and yields the mock;
   each test configures behaviour as needed (``return_value`` /
   ``side_effect``).

LOC delta: -194 / +103 = 91 LOC net removed.

Validation:
* ``tests/test_barrier.py``: 10/10 passing
* ``tests/test_chord_sites_characterisation.py``: 9/9 passing
* Full workers suite: 6 pre-existing failures / 657 passed (was 659 →
  -2 from removing the duplicate ``TestWorkflowOrchestrationMixinCreateChord``
  methods; identical coverage retained in the characterisation suite)
* Five deterministic-order full-suite runs: exactly 6/657 each time —
  zero flakiness from this refactor

No production code changed; this is test-file dedup only.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…ngs (9 fixes)

Comprehensive response to the multi-source review pass on PR #2024.
All fixes preserve zero behaviour change vs pre-PR-2024 main —
including the previously-theoretical zero-files API regression.

## Production code fixes

* **G1 — Empty-header guard moved before signature construction**
  (`barrier.py`). Zero-task runs now skip callback signature build
  + fairness serialisation entirely; non-empty path unchanged.
  Constrains the ``try`` block's exception surface to the
  branch where work actually happens.

* **G2 — Removed redundant inline import of
  ``WorkflowOrchestrationUtils``** in ``api-deployment/tasks.py``.
  Already imported at module level (line 25); the deferred import
  inside the ``try`` block was a vestige of the pre-Barrier
  ``from celery import chord`` pattern.

* **T1 — Zero-files API path now matches pre-Barrier behaviour
  byte-for-byte**. Before the Barrier uplift, the inline
  ``chord(empty)(callback)`` call returned a truthy ``AsyncResult``
  and Celery fired the callback immediately with ``[]``. Post-uplift
  the Barrier returns ``None`` for empty headers, and the existing
  ``if not result: raise`` mapped zero-files runs to
  ``ExecutionStatus.ERROR``.

  New defensive branch in ``api-deployment/tasks.py`` explicitly
  dispatches ``process_batch_callback_api`` with ``args=[[]]``
  (body=[] matches Celery's chord-empty semantic) via
  ``queue_backend.dispatch`` — same seam every other dispatch uses,
  picks up fairness routing the same way. Unreachable in practice
  (upstream requires non-empty ``created_files``) but closes the
  theoretical regression entirely.

* **T3 — Header signatures stamped via
  ``Signature.clone().set(headers=...)`` instead of in-place
  ``.set(...)``**. Removes cross-tenant header-leakage foot-gun
  where a future retry path / signature cache could re-use a
  header_tasks list with a different ``FairnessKey``. When
  ``fairness=None`` the clone is skipped — the ``chord(...)`` call
  receives the caller's signatures unchanged, preserving byte-for-
  byte pre-Barrier wire.

* **T7 — ``WorkflowOrchestrationMixin.create_chord`` forwards
  ``fairness=`` kwarg** to ``create_chord_execution``. Backward-
  compat: default is ``None`` so existing callers don't break.

## Docstring + comment fixes

* **T2 — Docstring qualified.** "Byte-identical wire output" is
  true only when ``fairness=None``. Both production call sites
  pass a ``FairnessKey`` so post-uplift wire has the additive
  ``x-fairness-key`` AMQP header; qualified the claim and the
  "Mixed-version rolling deploys" claim two lines down.

## Test improvements

* **T4 — ``_BARRIER`` singleton routing pinned** by new
  ``TestOrchestrationUtilsRoutesThroughSingleton``. Patches
  ``orchestration_utils._BARRIER`` and asserts
  ``enqueue.assert_called_once_with(...)``. Locks the singleton
  as the single dispatch point ahead of Phase 6b's factory swap.

* **T5 — Inventory canary regex tightened**
  (``test_chord_sites_characterisation.py``). Was
  ``(?:^|[\s=(])chord\(`` which could match docstring prose. Now
  requires assignment form ``=\s*chord\(``. Comment lines
  (lstrip + startswith "#") explicitly skipped.

* **T6 — Call-site fairness contracts pinned** by new
  ``TestCallSiteFairnessContracts``. ``api-deployment/tasks.py``
  → ``WorkloadType.API`` + ``org_id=str(schema_name)``;
  ``general/tasks.py`` → ``WorkloadType.NON_API`` +
  ``org_id=organization_id``. A swap goes undetected by the
  isolated barrier tests.

* **T8 — Zero-files contract pinned** by new
  ``TestApiDeploymentZeroFilesContract``. Asserts the defensive
  branch dispatches via the seam (``dispatch(...)``) with body=[],
  not raw ``send_task``.

## CodeRabbit / SonarCloud already addressed

* CodeRabbit's ``logger.exception`` suggestion + SonarCloud's
  ``python:S8572`` finding both addressed in ``13c10d9fe``.

## Validation

* Touched test modules: 23 / 23 passing (was 19 → +4 new tests)
* Full workers suite: 6 failed (pre-existing baseline, unchanged)
  / 661 passed (was 657 → +4 new tests)
* Five deterministic-order runs: exactly 6/661 each time — zero
  flakiness
* The dispatch-sites canary
  (``test_dispatch_sites_characterisation``) caught my initial
  T1 fix using ``app.send_task`` directly; refactored to
  ``queue_backend.dispatch`` so the seam stays enforced.

## Production regression risk

After this commit + parent PR, vs. pre-PR-2024 main:
* Wire size: +~80 bytes per chord member task (additive fairness
  header). Negligible at 130K scale.
* All other paths byte-identical: API chord (non-empty + zero-
  files), general/ETL chord, chord_unlock polling, result backend
  + callback firing, backend Django chord sites, mixed-version
  rolling deploy.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>