fix: increasing concurrency

[TEJASNARAYANS]

What Changed?

Briefly describe what this PR changes.

Why?

Explain the problem this solves and why this approach was chosen.

How to Test?

List clear steps for reviewers to verify the change.

Release Label

Select one semantic version bump intent for this PR:

• major - breaking change, next release bumps major version
• minor - backward-compatible feature, next release bumps minor version
• fix - backward-compatible bug fix, next release bumps patch version
• No label (defaults to patch release)

If you do not have permission to apply labels, mention the intended release label here and a maintainer will set it.

Checklist

• I have read the CONTRIBUTING.md guide.
• My code follows the project's style guidelines.
• I have added tests that prove my fix is effective or my feature works.
• I have updated documentation where needed.

[greptile-apps]

Greptile Summary

This PR extends the fair round-robin dispatch system to cover recording-fetch imports and standalone diarization (previously only evaluations had fair dispatch), splits evaluation tasks into a separate audio-metrics phase, adds a per-job (EVAL_JOB_INFLIGHT_LIMIT) concurrency cap, and introduces graceful handling for rows deleted mid-flight (StaleDataError, DELETING status).

• New dispatchers: fair_import_dispatch and fair_diarization_dispatch give imports and standalone diarization the same workspace-interleaved scheduling that evaluations already had; Redis pipeline used for batch diarization param storage with explicit failure handling when Redis is unavailable.
• Audio/LLM task split: evaluate_call_import_row_audio_task now runs on a dedicated audio-metrics queue, keeping torch/Praat off the lightweight worker-imports pool; the two phases chain via evaluate_call_import_row_task when needed.
• Delete safety: StaleDataError is caught throughout process_call_import_row, and _rollup_parent_status skips status mutation when the parent import is in DELETING state; a new migration adds the deleting enum value.

Confidence Score: 4/5

Safe to merge with one fix: the new failed-diarisation skip in eval_dispatch.py will permanently stall eval rows whose source rows failed transcription with no stored transcript.

The new fair-dispatch infrastructure for imports and diarization is well-structured, the audio/LLM task split is clean, and the delete-safety changes are thorough. The one issue is in eval_dispatch.py: the added guard if dia_status == 'failed' and not transcribe_overwrite: return 'skip' causes eval rows to be permanently stuck in 'pending' whenever the source row's transcription failed and left no stored transcript. Previously the dispatcher would automatically reset the status to 'pending' and re-enqueue transcription; now it skips indefinitely with no user-visible error on the eval row.

app/workers/concurrency/eval_dispatch.py — the new dia_status == 'failed' branch at lines 150-151

Important Files Changed

Filename	Overview
app/workers/concurrency/eval_dispatch.py	New audio-phase routing, per-job inflight limit, and queue changes look correct, but the new dia_status == "failed" early-return silently stalls eval rows whose source rows failed transcription with no stored transcript.
app/workers/concurrency/fair_dispatch.py	Adds per-workspace per-evaluation round-robin cursor, refactors _dispatch_batch_for_workspace to interleave rows across evaluations — logic is sound.
app/workers/concurrency/fair_diarization_dispatch.py	New standalone diarization fair dispatcher. Missing-params rows are now explicitly marked failed rather than left stuck in pending.
app/workers/concurrency/fair_import_dispatch.py	New fair-dispatch dispatcher for import row recording fetch — mirrors fair_dispatch structure, DELETING guard is correctly applied at workspace, call_import, and row query levels.
app/workers/concurrency/limits.py	Adds per-job (per-evaluation) inflight limit via a new Lua key (KEYS[5]). When evaluation_id is absent, job_key falls back to task_key to avoid a nil-key call; the release script correctly conditions decr_min_zero on whether evaluation_id was stored.
app/workers/tasks/evaluate_call_import_row_audio.py	New audio-phase task split off from the main eval task. The apply_async failure path now correctly sets eval_row.status = "failed" and returns before setting chain_llm = True, so the finally block releases the slot as expected.
app/workers/tasks/process_call_import_row.py	Adds StaleDataError handling throughout so rows deleted mid-flight don't crash the worker, DELETING status guard at task start, and slot release via fair_import_dispatch in the finally block.
app/workers/tasks/transcribe_call_import_row.py	Adds empty-transcript guard, top-level except block to mark rows failed on unexpected crash, and slot release via fair_diarization_dispatch for standalone diarization runs.
app/services/call_imports/bulk_ops.py	New bulk ops service: handles batch diarization (with Redis pipeline and failure marking), evaluation materialization, row/import deletion. Rows failing Redis param storage are now correctly marked failed rather than left pending.

_{Reviews (7): Last reviewed commit: "fix: updating test cases" | Re-trigger Greptile}

[greptile-apps]

Failed-diarisation rows permanently stuck in pending

When source_row.diarised_transcript_status == "failed" and source_row.diarised_transcript is empty (no transcript was written before the failure), _needs_transcribe_for_eval returns True because needs_diarisation = not existing_dia is True. The new guard then returns "skip" immediately. Because the eval dispatcher is the only thing that would re-enqueue transcription for an eval-triggered source row — and the standalone diarization dispatcher only picks up rows with diarised_transcript_status == "pending" — these eval rows remain in "pending" forever. There is no periodic sweep and no error surfaced; the evaluation stalls in "running" without any user-visible reason.

The previous code had no "failed" check in this block, so it would reset diarised_transcript_status to "pending" and re-enqueue transcription automatically. Recovery now requires the user to explicitly re-trigger the evaluation with transcribe_overwrite=True, which is not documented anywhere and not exposed in the UI by default.