Fix zombie jobs and unprotected JSON parse

Bug #1: _on_task_done now receives job_id via functools.partial and
calls _mark_failed when the background task raises or is cancelled,
preventing jobs from being stuck in RUNNING state forever.

Bug #5: _parse_response wraps json.loads in try/except JSONDecodeError
so malformed json_content strings fall back gracefully instead of crashing.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

document-parser/infra/serve_converter.py

@@ -138,7 +138,11 @@ def _parse_response(data: dict) -> ConversionResult:
     # json_content contains the full DoclingDocument with pages, elements, bboxes
     json_content = document.get("json_content")
     if isinstance(json_content, str):
-        json_content = json.loads(json_content)
+        try:
+            json_content = json.loads(json_content)
+        except json.JSONDecodeError:
+            logger.warning("Failed to parse json_content as JSON, ignoring structured data")
+            json_content = None
 
     pages: list[PageDetail] = []
     if json_content:

document-parser/services/analysis_service.py

@@ -7,6 +7,7 @@ from the conversion implementation (local Docling lib vs remote Docling Serve).
 from __future__ import annotations
 
 import asyncio
+import functools
 import json
 import logging
 from dataclasses import asdict
@@ -42,7 +43,7 @@ class AnalysisService:
         task = asyncio.create_task(
             self._run_analysis(job.id, doc.storage_path, doc.filename, pipeline_options)
         )
-        task.add_done_callback(_on_task_done)
+        task.add_done_callback(functools.partial(_on_task_done, job_id=job.id))
 
         return job
 
@@ -99,14 +100,16 @@ class AnalysisService:
             await _mark_failed(job_id, str(e))
 
 
-def _on_task_done(task: asyncio.Task) -> None:
-    """Log unhandled exceptions from background analysis tasks."""
+def _on_task_done(task: asyncio.Task, *, job_id: str) -> None:
+    """Log unhandled exceptions from background analysis tasks and mark job as FAILED."""
     if task.cancelled():
-        logger.warning("Analysis task was cancelled")
+        logger.warning("Analysis task was cancelled: %s", job_id)
+        asyncio.ensure_future(_mark_failed(job_id, "Task was cancelled"))
         return
     exc = task.exception()
     if exc:
-        logger.error("Unhandled exception in analysis task: %s", exc, exc_info=True)
+        logger.error("Unhandled exception in analysis task %s: %s", job_id, exc, exc_info=True)
+        asyncio.ensure_future(_mark_failed(job_id, str(exc)))
 
 
 async def _mark_failed(job_id: str, error: str) -> None:

document-parser/tests/test_analysis_service.py

@@ -0,0 +1,71 @@
+"""Tests for AnalysisService — focus on _on_task_done callback (bug #1 fix)."""
+
+from __future__ import annotations
+
+import asyncio
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from services.analysis_service import _on_task_done
+
+
+class TestOnTaskDone:
+    """Bug #1: _on_task_done must call _mark_failed when the task raises."""
+
+    @pytest.mark.asyncio
+    async def test_exception_marks_job_failed(self):
+        """When a background task raises, the job should be marked FAILED."""
+        job_id = "job-123"
+
+        # Create a task that raises
+        async def failing_task():
+            raise RuntimeError("boom")
+
+        task = asyncio.create_task(failing_task())
+        await asyncio.sleep(0)  # let the task fail
+
+        with patch("services.analysis_service._mark_failed", new_callable=AsyncMock) as mock_mark:
+            _on_task_done(task, job_id=job_id)
+            # ensure_future schedules it; give the event loop a tick
+            await asyncio.sleep(0)
+
+        mock_mark.assert_called_once_with(job_id, "boom")
+
+    @pytest.mark.asyncio
+    async def test_cancelled_task_marks_job_failed(self):
+        """When a background task is cancelled, the job should be marked FAILED."""
+        job_id = "job-456"
+
+        async def slow_task():
+            await asyncio.sleep(999)
+
+        task = asyncio.create_task(slow_task())
+        task.cancel()
+        try:
+            await task
+        except asyncio.CancelledError:
+            pass
+
+        with patch("services.analysis_service._mark_failed", new_callable=AsyncMock) as mock_mark:
+            _on_task_done(task, job_id=job_id)
+            await asyncio.sleep(0)
+
+        mock_mark.assert_called_once_with(job_id, "Task was cancelled")
+
+    @pytest.mark.asyncio
+    async def test_successful_task_does_not_mark_failed(self):
+        """When a background task succeeds, _mark_failed should not be called."""
+        job_id = "job-789"
+
+        async def ok_task():
+            return "done"
+
+        task = asyncio.create_task(ok_task())
+        await task
+
+        with patch("services.analysis_service._mark_failed", new_callable=AsyncMock) as mock_mark:
+            _on_task_done(task, job_id=job_id)
+            await asyncio.sleep(0)
+
+        mock_mark.assert_not_called()

document-parser/tests/test_serve_converter.py

@@ -147,6 +147,21 @@ class TestParseResponse:
         result = _parse_response(data)
         assert result.page_count == 1
 
+    def test_json_content_malformed_string_falls_back(self):
+        """Bug #5: malformed JSON string in json_content must not crash."""
+        data = {
+            "document": {
+                "md_content": "# Hello",
+                "html_content": "<h1>Hello</h1>",
+                "json_content": "NOT VALID JSON {{{",
+            }
+        }
+        result = _parse_response(data)
+        assert isinstance(result, ConversionResult)
+        assert result.content_markdown == "# Hello"
+        assert result.pages == []
+        assert result.page_count == 1
+
     def test_tables_and_pictures(self):
         data = {
             "document": {