Fail a job with an unreadable evidence payload once instead of forever

CHANGELOG.md

@@ -6,6 +6,12 @@ a decision with visible consequences.
 
 ## Unreleased
 
+- 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
+  per lease. The payload is now parsed in `run_job`, which fails the job once ("evidence payload
+  is not readable") and quarantines the evidence, the same outcome `process_evidence` already
+  gives a job whose processing raises.
 - A malformed time in a request field (`occurred_at`, `session_start`/`session_end`,
   `occurred_start`/`occurred_end`, `review_by`) answers 422 naming the bad field instead of a
   500. Each route module had its own parse helper that let the `ValueError` from

services/api/app/services/processing_worker.py

@@ -67,29 +67,44 @@ def claim_next_job(db, worker_id: str, *, now: datetime | None = None,
         db.rollback()
 
 
+def _fail(db, job: ProcessingJob, error: str, evidence: EvidenceObject | None = None) -> None:
+    fence_job_lease(db, job.id, job.lease_owner)
+    job.status = "failed"
+    job.error = error
+    job.lease_owner = None
+    job.lease_expires_at = None
+    if evidence:
+        evidence.processing_state = "quarantined"
+    db.commit()
+
+
+def run_job(db, job: ProcessingJob) -> None:
+    """Process one claimed job, failing it at once if its evidence is unreadable."""
+    evidence = db.get(EvidenceObject, job.evidence_id)
+    if not evidence:
+        _fail(db, job, "evidence no longer exists")
+        return
+    try:
+        content = json.loads(evidence.normalized_payload_json or "{}").get("content") or evidence.summary
+    except (ValueError, AttributeError):
+        _fail(db, job, "evidence payload is not readable", evidence)
+        return
+    with LeaseHeartbeat(
+        SessionLocal,
+        job.id,
+        job.lease_owner,
+        lease_seconds=PROCESSING_LEASE_SECONDS,
+    ) as lease:
+        process_evidence(db, evidence, content, job, lease=lease)
+
+
 def _run() -> None:
     while not _stop.is_set():
         db = SessionLocal()
         try:
             job = claim_next_job(db, _worker_id)
             if job:
-                evidence = db.get(EvidenceObject, job.evidence_id)
-                if not evidence:
-                    fence_job_lease(db, job.id, job.lease_owner)
-                    job.status = "failed"
-                    job.error = "evidence no longer exists"
-                    job.lease_owner = None
-                    job.lease_expires_at = None
-                    db.commit()
-                else:
-                    content = json.loads(evidence.normalized_payload_json or "{}").get("content") or evidence.summary
-                    with LeaseHeartbeat(
-                        SessionLocal,
-                        job.id,
-                        job.lease_owner,
-                        lease_seconds=PROCESSING_LEASE_SECONDS,
-                    ) as lease:
-                        process_evidence(db, evidence, content, job, lease=lease)
+                run_job(db, job)
         except Exception:
             db.rollback()
         finally:

services/api/tests/test_processing_worker.py

@@ -6,7 +6,7 @@
 from app.core.db import Base
 from app.models.evidence_object import EvidenceObject
 from app.models.processing_job import ProcessingJob
-from app.services import runtime_pipeline
+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 sqlalchemy import create_engine, select
@@ -314,3 +314,39 @@ def lose_lease_and_fail(*_args):
             assert evidence.processing_state != "quarantined"
     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)
+    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=payload,
+            )
+            db.add(evidence)
+            db.flush()
+            job = ProcessingJob(agent_id="owner", evidence_id=evidence.id, stage="queued")
+            db.add(job)
+            db.commit()
+            evidence_id = evidence.id
+
+        with sessions() as db:
+            claimed = claim_next_job(db, "worker-a", now=started_at, lease_seconds=60)
+            processing_worker.run_job(db, claimed)
+
+        with sessions() as db:
+            job = db.get(ProcessingJob, claimed.id)
+            evidence = db.get(EvidenceObject, evidence_id)
+            assert job.status == "failed"
+            assert job.error == "evidence payload is not readable"
+            assert job.lease_owner is None
+            assert evidence.processing_state == "quarantined"
+            assert claim_next_job(
+                db, "worker-b", now=started_at + timedelta(seconds=61), lease_seconds=60
+            ) is None
+    finally:
+        engine.dispose()