fix(api): reserve the app when starting an annotation reply job

[alanhuangyoo]

Closes #40595

Problem

AppAnnotationService.enable_app_annotation reads the app-level key that is supposed to serialize annotation reply jobs, but nothing ever writes it:

enable_app_annotation_key = f"enable_app_annotation_{app_id}"
cache_result = redis_client.get(enable_app_annotation_key)      # read
if cache_result is not None:
    return {"job_id": cache_result, "job_status": "processing"}

job_id = str(uuid.uuid4())
redis_client.setnx(f"enable_app_annotation_job_{job_id}", "waiting")   # only the per-job key
enable_annotation_reply_task.delay(...)                                # app key never written

The guard can therefore never fire. Every call mints a new job id and enqueues another task, and those tasks concurrently rebuild or delete the same annotation vector data and mutate the same AppAnnotationSetting. The worker's finally block then deletes a key that was never acquired. disable_app_annotation has the same shape.

Two smaller defects come from the same code:

• The per-job status key is written with setnx and no expiry, so it survives forever whenever the task does not reach a terminal path.
• The workers return early when the app or the annotation setting is missing, before any status is written, so a caller polling that job id sees waiting indefinitely.

Change

• Reserve the app with SET key job_id NX EX before enqueueing. A second request for the same action gets the in-flight job id back with processing instead of starting a competing task.
• Give the per-job status key the same TTL the workers already use for terminal statuses (600s), so nothing outlives the job it describes.
• Release the reservation if .delay() raises, so a broker failure does not lock the app for the full TTL.
• Record a terminal error status (plus the existing *_error_* key) on the worker early-return paths and drop the reservation there.

_running_job_id() decodes the reserved value because the client is configured with decode_responses=False; the previous code would have returned raw bytes as the job id, which was masked by the guard never firing.

Enable and disable still use separate reservations. The issue also proposes making them mutually exclusive with each other, which changes behaviour beyond the broken guard, so I left it out of this PR — happy to follow up if you want it.

Tests

TestAppAnnotationServiceEnableDisable previously mocked the whole redis client and asserted the exact setnx call, so it described the implementation rather than the behaviour and could not observe the missing reservation. It is rewritten against a small fake that implements SET NX EX semantics, and covers: the reservation is written with a TTL, a second request reuses the in-flight job without enqueueing, and a failed enqueue releases the reservation.

Against the current main sources, all five fail:

FAILED ...::test_enable_enqueues_and_reserves_the_app
FAILED ...::test_enable_reuses_the_in_flight_job
FAILED ...::test_enable_releases_the_reservation_when_enqueue_fails
FAILED ...::test_disable_enqueues_and_reserves_the_app
FAILED ...::test_disable_reuses_the_in_flight_job

with, for the reuse cases:

E  AssertionError: {'job_status': 'waiting'} != {'job_status': 'processing'}
E                  {'job_id': 'eb28b41d-...'} != {'job_id': '1e812dc9-...'}

With this change:

api/tests/unit_tests/services/test_annotation_service.py   ->  47 passed
api/tests/unit_tests/controllers -k annotation             ->  74 passed

ruff check and ruff format --check are clean on the four touched files.