Refuse to retry a job a worker is still running

CHANGELOG.md

@@ -6,6 +6,9 @@ a decision with visible consequences.
 
 ## Unreleased
 
+- Retrying a job that a worker is still running (its lease has not expired) is refused with a 409.
+  It used to be queued again and cleared the lease, so a second worker could process the same
+  evidence while the first was still on it.
 - A job whose stored evidence payload cannot be read - not JSON, or JSON that is not an object -
   used to raise inside the worker loop, leaving the job `processing` with its lease; when the
   lease ran out the next poll claimed it again and failed the same way, forever, one wasted claim

services/api/app/routes/runtime.py

@@ -30,6 +30,7 @@
 from app.services.memory_truth import mark_unsupported
 from app.services.ollama_service import answer_with_context, ollama_health
 from app.services.pi_inference import CHOSEN_IN, last_answer, model_choice, pi_configured
+from app.services.processing_worker import requeue
 from app.services.qdrant_store import INDEX_UNREACHABLE, search_memory_embeddings, semantic_status
 from app.services.scoring import memory_rank_bonus
 from fastapi import APIRouter, Depends, Header, HTTPException, Request
@@ -370,12 +371,8 @@ def retry_job(job_id: str, agent_id: str = Depends(get_agent_id), _: str = Depen
         evidence = db.get(EvidenceObject, job.evidence_id)
         if not evidence:
             raise HTTPException(404, "Evidence no longer exists")
-        job.status = "pending"
-        job.stage = "retry_queued"
-        job.error = ""
-        job.lease_owner = None
-        job.lease_expires_at = None
-        db.commit()
+        if not requeue(db, job.id):
+            raise HTTPException(409, "Processing job is running; retry after it ends")
         return {"status": "queued", "job_id": job.id}
     finally:
         db.close()

services/api/app/services/processing_worker.py

@@ -67,6 +67,30 @@ def claim_next_job(db, worker_id: str, *, now: datetime | None = None,
         db.rollback()
 
 
+def requeue(db, job_id: str, *, now: datetime | None = None) -> bool:
+    """Queue a job again unless a worker holds a live lease on it; True when queued."""
+    now = now or datetime.now(UTC)
+    not_running = or_(
+        ProcessingJob.status != "processing",
+        ProcessingJob.lease_expires_at.is_(None),
+        ProcessingJob.lease_expires_at <= now,
+    )
+    result = db.execute(
+        update(ProcessingJob)
+        .where(ProcessingJob.id == job_id, not_running)
+        .values(
+            status="pending",
+            stage="retry_queued",
+            error="",
+            lease_owner=None,
+            lease_expires_at=None,
+        )
+        .execution_options(synchronize_session=False)
+    )
+    db.commit()
+    return result.rowcount == 1
+
+
 def _fail(db, job: ProcessingJob, error: str, evidence: EvidenceObject | None = None) -> None:
     fence_job_lease(db, job.id, job.lease_owner)
     job.status = "failed"

services/api/tests/test_processing_worker.py

@@ -8,7 +8,7 @@
 from app.models.processing_job import ProcessingJob
 from app.services import processing_worker, runtime_pipeline
 from app.services.processing_lease import LeaseHeartbeat, LeaseLostError
-from app.services.processing_worker import claim_next_job
+from app.services.processing_worker import claim_next_job, requeue
 from sqlalchemy import create_engine, select
 from sqlalchemy.orm import sessionmaker
 
@@ -316,6 +316,106 @@ def lose_lease_and_fail(*_args):
         engine.dispose()
 
 
+def test_retry_refuses_a_job_with_a_live_lease(tmp_path):
+    engine, sessions = _database(tmp_path)
+    try:
+        started_at = datetime.now(UTC)
+        with sessions() as db:
+            evidence = EvidenceObject(
+                agent_id="owner", source_id="source-1", source_key="listener",
+                source_type="listener", summary="A bounded event",
+                normalized_payload_json='{"content":"A bounded event"}',
+            )
+            db.add(evidence)
+            db.flush()
+            job = ProcessingJob(agent_id="owner", evidence_id=evidence.id, stage="queued")
+            db.add(job)
+            db.commit()
+
+        with sessions() as db:
+            claimed = claim_next_job(db, "worker-a", now=started_at, lease_seconds=60)
+            assert claimed is not None
+
+        with sessions() as db:
+            result = requeue(db, claimed.id, now=started_at + timedelta(seconds=10))
+            assert result is False
+            job = db.get(ProcessingJob, claimed.id)
+            assert job.status == "processing"
+            assert job.lease_owner == "worker-a"
+            # Compare with timezone-aware datetime
+            expected_expiry = started_at + timedelta(seconds=60)
+            assert job.lease_expires_at.replace(tzinfo=UTC) == expected_expiry
+    finally:
+        engine.dispose()
+
+
+def test_retry_requeues_a_failed_job_and_an_expired_lease(tmp_path):
+    engine, sessions = _database(tmp_path)
+    try:
+        # Test (a): failed job with error
+        with sessions() as db:
+            evidence = EvidenceObject(
+                agent_id="owner", source_id="source-1", source_key="listener",
+                source_type="listener", summary="A bounded event",
+                normalized_payload_json='{"content":"A bounded event"}',
+            )
+            db.add(evidence)
+            db.flush()
+            failed_job = ProcessingJob(
+                agent_id="owner", evidence_id=evidence.id, status="failed",
+                stage="analysis", error="some error"
+            )
+            db.add(failed_job)
+            db.commit()
+
+        with sessions() as db:
+            result = requeue(db, failed_job.id)
+            assert result is True
+            job = db.get(ProcessingJob, failed_job.id)
+            assert job.status == "pending"
+            assert job.stage == "retry_queued"
+            assert job.error == ""
+            assert job.lease_owner is None
+            assert job.lease_expires_at is None
+
+        # Test (b): job with expired lease
+        started_at = datetime.now(UTC)
+        with sessions() as db:
+            evidence = EvidenceObject(
+                agent_id="owner", source_id="source-2", source_key="listener",
+                source_type="listener", summary="Another event",
+                normalized_payload_json='{"content":"Another event"}',
+            )
+            db.add(evidence)
+            db.flush()
+            claimed_job = ProcessingJob(agent_id="owner", evidence_id=evidence.id, stage="queued")
+            db.add(claimed_job)
+            db.commit()
+
+        with sessions() as db:
+            claimed = claim_next_job(db, "worker-a", now=started_at, lease_seconds=60)
+            assert claimed is not None
+
+        with sessions() as db:
+            result = requeue(db, claimed.id, now=started_at + timedelta(seconds=61))
+            assert result is True
+            job = db.get(ProcessingJob, claimed.id)
+            assert job.lease_owner is None
+            assert job.lease_expires_at is None
+    finally:
+        engine.dispose()
+
+
+def test_requeue_of_a_missing_job_is_false(tmp_path):
+    engine, sessions = _database(tmp_path)
+    try:
+        with sessions() as db:
+            result = requeue(db, "no-such-job")
+            assert result is False
+    finally:
+        engine.dispose()
+
+
 @pytest.mark.parametrize("payload", ["not json", "[1, 2]"], ids=["not-json", "json-list"])
 def test_an_unreadable_payload_fails_the_job_once(tmp_path, payload):
     engine, sessions = _database(tmp_path)