fix: report invalid office uploads

apps/api/app/services/jobs/result_projection.py

@@ -15,6 +15,7 @@
 JobStatusValue = Literal[
     "pending", "waiting-file", "running", "converting", "done", "failed"
 ]
+_CHARGED_BILLING_STATUS: str = "charged"
 
 
 def build_error_response(
@@ -112,6 +113,17 @@ def _resolve_duration_seconds(job: Any) -> float | None:
     return None
 
 
+def _resolve_credits_spent(job: Any) -> float:
+    if to_job_status_value(job.status) == "failed":
+        return 0.0
+
+    billing_status = getattr(job, "billing_status", None)
+    if billing_status != _CHARGED_BILLING_STATUS:
+        return 0.0
+
+    return MicroDollar(getattr(job, "credits_charged", 0) or 0).to_credit()
+
+
 async def _resolve_result_delivery(
     job: Any,
 ) -> tuple[dict[str, Any] | None, str | None, datetime]:
@@ -162,9 +174,5 @@ async def build_job_result_response(
         model=parsing_params.get("model"),
         ocr_enabled=parsing_params.get("ocr_enabled"),
         duration_seconds=_resolve_duration_seconds(job),
-        credits_spent=(
-            MicroDollar(job.credits_charged).to_credit()
-            if hasattr(job, "credits_charged")
-            else 0
-        ),
+        credits_spent=_resolve_credits_spent(job),
     )

apps/api/tests/contract/test_job_read_contract.py

@@ -132,6 +132,46 @@ async def test_should_return_job_details_for_an_existing_waiting_file_job(
     assert response_json["credits_spent"] == 0.0
 
 
+@pytest.mark.asyncio
+async def test_should_report_zero_credits_spent_for_refunded_failed_job(
+    developer_api_client_factory: Callable[
+        [], AbstractAsyncContextManager[AsyncClient]
+    ],
+) -> None:
+    async with developer_api_client_factory() as api_client:
+        created_job = await _create_waiting_file_job(api_client)
+        job_id = cast(str, created_job["job_id"])
+
+        await ContractDatabase.execute(
+            """
+            UPDATE jobs
+            SET
+                status = 'failed',
+                error_code = 'INVALID_ARGUMENT',
+                error_message = 'Invalid file: the uploaded .docx file is not a valid Word document. Please check the file and upload again.',
+                credits_charged = 15000,
+                billing_status = 'refunded'
+            WHERE job_id = :job_id
+            """,
+            {"job_id": job_id},
+        )
+
+        response = await api_client.get(f"/api/v1/jobs/{job_id}")
+
+    assert response.status_code == 200
+
+    response_json = cast(dict[str, object], response.json())
+    error = cast(dict[str, object], response_json["error"])
+
+    assert response_json["status"] == "failed"
+    assert error["code"] == "INVALID_ARGUMENT"
+    assert (
+        error["message"]
+        == "Invalid file: the uploaded .docx file is not a valid Word document. Please check the file and upload again."
+    )
+    assert response_json["credits_spent"] == 0.0
+
+
 @pytest.mark.asyncio
 async def test_should_return_not_found_when_requesting_an_unknown_job(
     developer_api_client_factory: Callable[

apps/worker/app/services/document_parser/orchestration/office_container_validator.py

@@ -0,0 +1,98 @@
+from __future__ import annotations
+
+import zipfile
+from dataclasses import dataclass
+
+from app.services.document_parser.orchestration.format_router import DocumentFormat
+
+from shared.core.exceptions.domain_exceptions import ValidationException
+
+
+@dataclass(frozen=True)
+class _OfficeContainerRequirement:
+    extension: str
+    document_label: str
+    required_member: str
+
+    @property
+    def user_message(self) -> str:
+        return (
+            f"Invalid file: the uploaded {self.extension} file is not a valid "
+            f"{self.document_label}. Please check the file and upload again."
+        )
+
+    @property
+    def violation_description(self) -> str:
+        return (
+            f"Expected a valid {self.extension[1:].upper()} ZIP package containing "
+            f"{self.required_member}"
+        )
+
+
+_CONTENT_TYPES_MEMBER: str = "[Content_Types].xml"
+_OFFICE_CONTAINER_REQUIREMENTS: dict[
+    DocumentFormat,
+    _OfficeContainerRequirement,
+] = {
+    DocumentFormat.DOCX: _OfficeContainerRequirement(
+        extension=".docx",
+        document_label="Word document",
+        required_member="word/document.xml",
+    ),
+    DocumentFormat.XLSX: _OfficeContainerRequirement(
+        extension=".xlsx",
+        document_label="Excel workbook",
+        required_member="xl/workbook.xml",
+    ),
+    DocumentFormat.PPTX: _OfficeContainerRequirement(
+        extension=".pptx",
+        document_label="PowerPoint presentation",
+        required_member="ppt/presentation.xml",
+    ),
+}
+
+
+def validate_office_container(
+    file_path: str,
+    document_format: DocumentFormat,
+) -> None:
+    """Validate OOXML containers before dispatching to archive-based parsers."""
+    requirement = _OFFICE_CONTAINER_REQUIREMENTS.get(document_format)
+    if requirement is None:
+        return
+
+    if not zipfile.is_zipfile(file_path):
+        _raise_invalid_office_file(requirement)
+
+    try:
+        with zipfile.ZipFile(file_path, "r") as archive:
+            member_names = set(archive.namelist())
+    except zipfile.BadZipFile as exc:
+        raise _build_invalid_office_file_exception(requirement) from exc
+
+    if (
+        _CONTENT_TYPES_MEMBER not in member_names
+        or requirement.required_member not in member_names
+    ):
+        _raise_invalid_office_file(requirement)
+
+
+def _raise_invalid_office_file(requirement: _OfficeContainerRequirement) -> None:
+    raise _build_invalid_office_file_exception(requirement)
+
+
+def _build_invalid_office_file_exception(
+    requirement: _OfficeContainerRequirement,
+) -> ValidationException:
+    return ValidationException(
+        user_message=requirement.user_message,
+        violations=[
+            {
+                "field": "file",
+                "description": requirement.violation_description,
+            }
+        ],
+    )
+
+
+__all__ = ["validate_office_container"]

apps/worker/app/services/document_parser/orchestration/route_parse.py

@@ -2,12 +2,16 @@
     get_document_parse_adapter,
     resolve_document_format,
 )
+from app.services.document_parser.orchestration.office_container_validator import (
+    validate_office_container,
+)
 from app.services.document_parser.orchestration.parse_output import ParseOutput
 from app.services.document_parser.orchestration.parse_session import ParseSession
 
 
 def route_document_parse(session: ParseSession) -> ParseOutput:
     """Route a parser session to the correct adapter and return its output."""
     document_format = resolve_document_format(session.file_full_path)
+    validate_office_container(session.file_full_path, document_format)
     adapter = get_document_parse_adapter(document_format)
     return adapter.parse(session)

apps/worker/tests/contract/test_parse_task_contract.py

@@ -483,6 +483,58 @@ def test_parse_task_should_refund_charged_job_when_uploaded_file_cannot_be_parse
     ]
 
 
+def test_parse_task_should_report_invalid_docx_as_client_file_error(
+    worker_contract_environment: None,
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    contract = WorkerParseContract.create()
+    contract.use_workspace_root(monkeypatch, tmp_path)
+    contract.use_billing(monkeypatch, is_enabled=True)
+
+    invalid_docx_path = tmp_path / "invalid.docx"
+    invalid_docx_path.write_bytes(b"this is not a docx package")
+    job = contract.create_file_job(
+        source_file_name="contract-invalid.docx",
+        job_id_prefix="job_invalid_docx",
+    )
+    contract.upload_source_file(
+        local_file_path=invalid_docx_path,
+        s3_key=job["s3_key"],
+    )
+
+    celery_result = contract.enqueue_parse_task(
+        job_id=job["job_id"],
+        user_id=job["user_id"],
+    )
+
+    assert celery_result.failed()
+    assert contract.find_task_workspaces(tmp_path, job["job_id"]) == []
+
+    job_row = contract.observe_job_status(job["job_id"])
+    assert job_row["status"] == "failed"
+    assert job_row["billing_status"] == "refunded"
+    assert job_row["credits_charged"] == int(contract.settings.MICRO_DOLLARS_PER_PAGE)
+    assert job_row["error_code"] == "INVALID_ARGUMENT"
+    assert (
+        job_row["error_message"]
+        == "Invalid file: the uploaded .docx file is not a valid Word document. Please check the file and upload again."
+    )
+    assert contract.count_job_results(job["job_id"]) == 0
+
+    metadata = contract.get_job_metadata(job["job_id"])
+    assert metadata["error_details"] == {
+        "violations": [
+            {
+                "field": "file",
+                "description": (
+                    "Expected a valid DOCX ZIP package containing word/document.xml"
+                ),
+            }
+        ]
+    }
+
+
 def test_should_reject_pdf_when_page_count_exceeds_configured_limit(
     worker_contract_environment: None,
     monkeypatch: pytest.MonkeyPatch,

packages/shared-python/shared/core/state_machine/service_sync.py

@@ -36,7 +36,10 @@
 )
 from shared.models.database.job import Job
 from shared.models.database.job_state_audit_log import JobStateAuditLog
-from shared.services.redis.redis_sync_service import SyncRedisServiceFactory
+from shared.services.redis.redis_sync_service import (
+    SyncJobMetadataService,
+    SyncRedisServiceFactory,
+)
 from shared.services.redis.key_builder import RedisKeyType, redis_key_builder
 
 
@@ -412,6 +415,7 @@ def _update_job_error(
                 "error_message": error_message,
                 "error_code": error_code,
             }
+            metadata_updates: Dict[str, Any] = {}
 
             if error_details:
                 import json as _json
@@ -425,8 +429,14 @@ def _update_job_error(
                     pg_array(["error_details"]),
                     cast(literal(_json.dumps(error_details)), JSONB),
                 )
+                metadata_updates["error_details"] = error_details
 
             db.execute(update(Job).where(Job.job_id == job_id).values(**update_values))
+            if metadata_updates:
+                SyncJobMetadataService(self.redis).update_metadata(
+                    job_id,
+                    metadata_updates,
+                )
         except Exception as e:
             logger.error(f"Failed to update Job {job_id} error info: {e}")