UN-3508 [FEAT] Plumb fairness through ExecutionDispatcher

[muhammad-ali-e]

What

• Add optional headers: dict | None kwarg to all three dispatch methods on unstract.sdk1.execution.dispatcher.ExecutionDispatcher (dispatch, dispatch_async, dispatch_with_callback). When non-None, forwarded as Celery message headers; when None, the kwarg is omitted from send_task so existing callers see the identical pre-change call shape.
• New FairnessKey.as_header() helper in workers/queue_backend/fairness.py returns the wire-ready {"x-fairness-key": ...} dict, so producers don't need to import the slot-name constant.
• Wire fairness from the three dispatcher.dispatch(...) sites in workers/file_processing/structure_tool_task.py (lines 468, 507, 720). Each passes headers=_fairness_headers(organization_id). A small in-file helper defaults workload_type to NON_API; propagating the real type from upstream is Phase 6 work.
• New workers/tests/test_executor_dispatch.py (7 tests):
  • Header forwarding through all three dispatcher methods.
  • "Omit when None" — call shape preservation for callers that don't opt in (sdk1 round-trip tests still green unchanged).
  • FairnessKey.as_header() shape.
  • AST inventory canary: no production code calls *.send_task("execute_extraction", ...) outside ExecutionDispatcher.

Why

UN-3501 (#2003) plumbed fairness on bare dispatch(...) call sites — but execute_extraction, the most workflow-execution-y dispatch in the codebase, bypasses queue_backend.dispatch() entirely (it uses sdk1's ExecutionDispatcher). The Phase 5.1 canary audits bare-name dispatch() and missed it. This PR brings the third dispatch path under the same fairness umbrella.

Under Celery today the header sits inertly — no consumer reads it. Under the future PG Queue substrate, execute_extraction tasks land in the staging queue and the fairness scheduler sorts on the same three fields. Producer-side plumbing now means the consumer can land without backfilling in-flight payloads.

How

• ExecutionDispatcher only attaches headers= to send_task when the caller provides them. That preserves the pre-change mock.assert_called_with(...) shape used by sdk1's existing 78 dispatcher tests — they all stay green without modification.
• FairnessKey.as_header() shifts the wire-shape knowledge into the seam module, so producers compose a FairnessKey(...) and call .as_header(). The constant FAIRNESS_HEADER_NAME stays internal to queue_backend.fairness — keeps the "no consumer reads it yet" canary green at the producer level.
• The _fairness_headers helper in structure_tool_task.py is intentionally one function: same workload_type for all three sites today; Phase 6 (chord lift) is the right place to thread the real workload type from the workflow's class.

Can this PR break any existing features. If yes, please list possible items. If no, please explain why. (PS: Admins do not merge the PR without this section filled)

No.

• headers is optional and defaults to None on all three dispatcher methods. Callers that don't opt in see the identical send_task call shape.
• All 78 existing sdk1 dispatcher tests pass unchanged.
• Worker seam suite extended (53 → 60 tests, all green).
• Producer-side only: no consumer in workers/ reads x-fairness-key yet (canary in test_fairness_key.py still asserts this).
• No queue routing, task name, or args/kwargs change.

Database Migrations

• None.

Env Config

• None.

Relevant Docs

• N/A. Module-level docstrings in queue_backend/fairness.py and the dispatcher cover the header semantics.

Related Issues or PRs

• Ticket: UN-3508 (sub-task) under UN-3500 (Story: PG Queue Phase 5 — Payload typing) under UN-3445 (Epic).
• Builds on: UN-3501 (UN-3501 [FEAT] Plumb fairness key on every dispatch() call site #2003).

Dependencies Versions

• None.

Notes on Testing

```
cd workers
.venv/bin/python -m pytest \
tests/test_executor_dispatch.py \
tests/test_fairness_key.py \
tests/test_queue_backend_seam.py \
tests/test_dispatch_sites_characterisation.py

60 passed in ~6s

cd ../unstract/sdk1
uv run pytest tests/test_execution.py

80 passed

```

Verification:

```

Every execute_extraction dispatch goes through ExecutionDispatcher:

grep -rn "send_task.execute_extraction" workers/ --include=".py" \
| grep -v "ExecutionDispatcher\|tests/\|sdk1/"

(empty — only sdk1's ExecutionDispatcher.dispatch* methods)

```

Screenshots

N/A (no UI surface).

Checklist

I have read and understood the Contribution Guidelines.

[coderabbitai]

Important

Review skipped

Draft detected.

Please check the settings in the CodeRabbit UI or the .coderabbit.yaml file in this repository. To trigger a single review, invoke the @coderabbitai review command.

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: d764b7ff-8bb1-48e4-a93e-79863da7fce7

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

Use the checkbox below for a quick retry:

• 🔍 Trigger review

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch UN-3508-executor-fairness

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[sonarqubecloud]

Quality Gate failed

Failed conditions
5.7% Duplication on New Code (required ≤ 3%)

See analysis details on SonarQube Cloud