From 0ecb4219548b7fd045c60304b24a7623c069a699 Mon Sep 17 00:00:00 2001
From: Alex
Date: Thu, 6 Aug 2026 12:29:51 +0100
Subject: [PATCH 1/2] fix: error type fixes and docling parsing improvements
---
application/agents/headless_runner.py | 100 ++++++++--
.../versions/0027_schedule_run_error_types.py | 61 ++++++
application/api/user/scheduler_worker.py | 32 +++-
application/api/user/tasks.py | 18 +-
application/parser/file/base_parser.py | 16 ++
application/parser/file/bulk.py | 76 ++++++--
application/parser/file/docling_parser.py | 13 +-
application/requirements.txt | 7 +-
frontend/src/agents/types/schedule.ts | 5 +-
tests/agents/test_headless_runner_errors.py | 99 ++++++++++
tests/agents/test_headless_runner_workflow.py | 179 ++++++++++++++++++
tests/api/user/test_scheduler_worker.py | 165 ++++++++++++++++
tests/parser/file/test_bulk.py | 114 +++++++++++
tests/parser/file/test_docling_parser.py | 44 ++++-
tests/worker/test_attachment_worker.py | 68 +++++++
15 files changed, 938 insertions(+), 59 deletions(-)
create mode 100644 application/alembic/versions/0027_schedule_run_error_types.py
create mode 100644 tests/agents/test_headless_runner_errors.py
create mode 100644 tests/agents/test_headless_runner_workflow.py
diff --git a/application/agents/headless_runner.py b/application/agents/headless_runner.py
index 017f3df2..4c51f731 100644
--- a/application/agents/headless_runner.py
+++ b/application/agents/headless_runner.py
@@ -29,6 +29,34 @@ def _resolve_agent_id(agent_config: Dict[str, Any]) -> Optional[str]:
return str(raw) if raw else None
+def _workflow_kwargs(agent_config: Dict[str, Any], owner: str) -> Dict[str, Any]:
+ """Bind a workflow agent to its graph, mirroring ``StreamProcessor``.
+
+ A ``WorkflowAgent`` built without one of these loads no graph and its
+ entire run is a single "Failed to load workflow configuration." error, so
+ a scheduled or webhook-fired workflow agent never does anything. The PG
+ ``agents`` row stores a UUID under ``workflow_id``; the legacy Mongo shape
+ used ``workflow``, which also carried an embedded graph.
+ """
+ kwargs: Dict[str, Any] = {"workflow_owner": owner}
+ embedded = agent_config.get("workflow")
+ if isinstance(embedded, dict):
+ kwargs["workflow"] = embedded
+ saved_id = agent_config.get("workflow_id")
+ if saved_id:
+ kwargs["workflow_id"] = str(saved_id)
+ return kwargs
+ wf_ref = agent_config.get("workflow_id") or embedded
+ if wf_ref:
+ kwargs["workflow_id"] = str(wf_ref)
+ else:
+ logger.warning(
+ "Workflow agent %s has no workflow reference; the run will load no graph.",
+ _resolve_agent_id(agent_config),
+ )
+ return kwargs
+
+
def run_agent_headless(
agent_config: Dict[str, Any],
query: str,
@@ -133,22 +161,24 @@ def run_agent_headless(
if conversation_id:
tool_executor.conversation_id = str(conversation_id)
- agent = AgentCreator.create_agent(
- agent_type,
- endpoint=endpoint,
- llm_name=provider or settings.LLM_PROVIDER,
- model_id=model_id,
- api_key=system_api_key,
- agent_id=agent_id,
- user_api_key=user_api_key,
- prompt=prompt,
- chat_history=chat_history or [],
- retrieved_docs=retrieved_docs,
- decoded_token=decoded_token,
- attachments=[],
- json_schema=json_schema,
- tool_executor=tool_executor,
- )
+ agent_kwargs: Dict[str, Any] = {
+ "endpoint": endpoint,
+ "llm_name": provider or settings.LLM_PROVIDER,
+ "model_id": model_id,
+ "api_key": system_api_key,
+ "agent_id": agent_id,
+ "user_api_key": user_api_key,
+ "prompt": prompt,
+ "chat_history": chat_history or [],
+ "retrieved_docs": retrieved_docs,
+ "decoded_token": decoded_token,
+ "attachments": [],
+ "json_schema": json_schema,
+ "tool_executor": tool_executor,
+ }
+ if agent_type == "workflow":
+ agent_kwargs.update(_workflow_kwargs(agent_config, owner))
+ agent = AgentCreator.create_agent(agent_type, **agent_kwargs)
if conversation_id:
agent.conversation_id = str(conversation_id)
@@ -156,9 +186,28 @@ def run_agent_headless(
thought = ""
sources_log: List[Dict[str, Any]] = []
tool_calls: List[Dict[str, Any]] = []
+ stream_error: Optional[str] = None
+ steps_completed = 0
for event in agent.gen(query=query):
if not isinstance(event, dict):
continue
+ # ``Agent.gen`` reports a failed stream with an error event rather than
+ # by raising. Dropping it here (as this loop used to) makes a broken run
+ # indistinguishable from one that simply had nothing to say, and the
+ # caller records it as a success. Mirrors the sentinel in
+ # ``application/logging.py`` so an error carrying no message is still
+ # truthy instead of reading as "ok".
+ if event.get("type") == "error":
+ stream_error = str(event.get("error") or "")[:500] or "unspecified"
+ continue
+ # A workflow's work is its nodes: its tool calls stay in the engine's
+ # execution log and its node agents own their LLMs, so neither
+ # ``tool_calls`` nor the token tally below sees them. Counting
+ # completed steps is the only evidence a quiet workflow ran at all.
+ if event.get("type") == "workflow_step":
+ if event.get("status") == "completed":
+ steps_completed += 1
+ continue
if "answer" in event:
answer_full += str(event["answer"])
elif "sources" in event:
@@ -169,7 +218,22 @@ def run_agent_headless(
thought += str(event["thought"])
denied = list(getattr(tool_executor, "headless_denials", []))
- error_type = "tool_not_allowed" if denied and not answer_full.strip() else None
+ error: Optional[str] = None
+ if denied and not answer_full.strip():
+ error_type = "tool_not_allowed"
+ blocked = ", ".join(
+ str(d.get("tool_name") or d.get("action_name") or "?") for d in denied
+ )
+ error = f"headless allowlist blocked required tool: {blocked}"[:500]
+ elif stream_error:
+ error_type = "stream_error"
+ error = stream_error
+ else:
+ error_type = None
+ if stream_error:
+ logger.warning(
+ "Headless run for agent %s failed mid-stream: %s", agent_id, stream_error
+ )
# Use the LLM accumulator (gen_token_usage / stream_token_usage decorators);
# current_token_count is a context-size sentinel, not a usage tally.
@@ -186,5 +250,7 @@ def run_agent_headless(
"generated_tokens": generated_tokens,
"denied": denied,
"error_type": error_type,
+ "error": error,
+ "steps_completed": steps_completed,
"model_id": model_id,
}
diff --git a/application/alembic/versions/0027_schedule_run_error_types.py b/application/alembic/versions/0027_schedule_run_error_types.py
new file mode 100644
index 00000000..22c1c816
--- /dev/null
+++ b/application/alembic/versions/0027_schedule_run_error_types.py
@@ -0,0 +1,61 @@
+"""0027 schedule_runs.error_type — allow 'stream_error' and 'empty_output'.
+
+``schedule_runs.status`` could only ever be written ``success`` for a run whose
+stream failed: the headless runner dropped the agent's error event, so the
+worker saw an ordinary return with an empty answer and no error type. With that
+fixed the worker now classifies two further outcomes, and both need to be
+accepted by the CHECK constraint added in 0010:
+
+* ``stream_error`` — the agent reported a mid-stream failure (provider error,
+ failed fallback) rather than raising.
+* ``empty_output`` — backstop: the run produced no answer, no tool call and no
+ completion tokens, so it did no work even though nothing raised.
+
+Constraint-only change; no data is rewritten and existing rows stay valid.
+Idempotent both ways.
+
+Revision ID: 0027_schedule_run_error_types
+Revises: 0026_stack_logs_agent_id
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+
+
+revision: str = "0027_schedule_run_error_types"
+down_revision: Union[str, None] = "0026_stack_logs_agent_id"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+_OLD = (
+ "'auth_expired', 'tool_not_allowed', 'budget_exceeded', "
+ "'timeout', 'agent_error', 'internal', 'missed', 'overlap'"
+)
+_NEW = _OLD + ", 'stream_error', 'empty_output'"
+
+
+def _recreate(values: str) -> None:
+ op.execute(
+ "ALTER TABLE schedule_runs "
+ "DROP CONSTRAINT IF EXISTS schedule_runs_error_type_chk;"
+ )
+ op.execute(
+ "ALTER TABLE schedule_runs ADD CONSTRAINT schedule_runs_error_type_chk "
+ f"CHECK (error_type IS NULL OR error_type IN ({values}));"
+ )
+
+
+def upgrade() -> None:
+ _recreate(_NEW)
+
+
+def downgrade() -> None:
+ # Rows written by the new classifier would violate the narrower constraint,
+ # so retire those values before restoring it.
+ op.execute(
+ "UPDATE schedule_runs SET error_type = 'internal' "
+ "WHERE error_type IN ('stream_error', 'empty_output');"
+ )
+ _recreate(_OLD)
diff --git a/application/api/user/scheduler_worker.py b/application/api/user/scheduler_worker.py
index 85d4a30e..17ffbc3e 100644
--- a/application/api/user/scheduler_worker.py
+++ b/application/api/user/scheduler_worker.py
@@ -290,14 +290,14 @@ def execute_scheduled_run_body(run_id: str, celery_task_id: Optional[str]) -> Di
finished = datetime.now(timezone.utc)
- # Headless denial with no usable output → tool_not_allowed.
- if (
- error_type is None
- and (outcome.get("denied") or [])
- and not (outcome.get("answer") or "").strip()
- ):
- error_type = "tool_not_allowed"
- error_text = "headless allowlist blocked required tool"
+ # Honour the outcome the runner reported: it classifies headless denials
+ # (``tool_not_allowed``) and mid-stream failures (``stream_error``) alike,
+ # and owns the message. Re-deriving the denial case here only duplicated
+ # the runner's rule while leaving a failed stream — which returns normally,
+ # with an empty answer — to fall through to "success".
+ if error_type is None and outcome.get("error_type"):
+ error_type = str(outcome["error_type"])
+ error_text = str(outcome.get("error") or "") or error_type
prompt_tokens = int(outcome.get("prompt_tokens", 0) or 0)
generated_tokens = int(outcome.get("generated_tokens", 0) or 0)
@@ -313,6 +313,22 @@ def execute_scheduled_run_body(run_id: str, celery_task_id: Optional[str]) -> Di
f"{schedule['token_budget']}"
)
+ # Backstop for silent failures that raise nothing and emit no error event:
+ # a run that produced no answer, did no work and burned no completion
+ # tokens did not do its job, whatever the absence of an exception suggests.
+ # "Work" is deliberately broad — tool calls for a chat agent, completed
+ # nodes for a workflow — so a schedule whose whole purpose is a side effect
+ # (post a message, file a ticket) is never flagged for staying quiet.
+ if (
+ error_type is None
+ and not (outcome.get("answer") or "").strip()
+ and not (outcome.get("tool_calls") or [])
+ and not int(outcome.get("steps_completed") or 0)
+ and generated_tokens == 0
+ ):
+ error_type = "empty_output"
+ error_text = "run produced no answer, tool calls or completion tokens"
+
answer = outcome.get("answer") or ""
truncated = False
if len(answer) > _OUTPUT_CAP_CHARS:
diff --git a/application/api/user/tasks.py b/application/api/user/tasks.py
index d1fc60d2..0f9d812f 100644
--- a/application/api/user/tasks.py
+++ b/application/api/user/tasks.py
@@ -5,6 +5,7 @@ from sqlalchemy.exc import DataError
from application.api.user.idempotency import with_idempotency
from application.celery_init import celery
+from application.parser.file.base_parser import DocumentParseError
from application.worker import (
AttachmentRejectedError,
agent_webhook_worker,
@@ -69,7 +70,10 @@ def _emit_ingest_poison_event(task_name, bound):
)
-@celery.task(**DURABLE_TASK)
+# ``dont_autoretry_for``: a file that cannot be converted to text fails
+# identically on every attempt, so retrying only multiplies the log noise
+# before the same failure — go straight to the poison/failure path.
+@celery.task(**DURABLE_TASK, dont_autoretry_for=(DocumentParseError,))
@with_idempotency(task_name="ingest", on_poison=_emit_ingest_poison_event)
def ingest(
self,
@@ -237,10 +241,14 @@ def _emit_attachment_poison_event(task_name, bound):
# ``dont_autoretry_for``: a DataError (poison payload, e.g. NUL bytes or an
-# over-long value) or an AttachmentRejectedError (zip bomb) is deterministic —
-# retrying re-fails identically and multiplies log noise, so it goes straight
-# to the failure path.
-@celery.task(**DURABLE_TASK, dont_autoretry_for=(DataError, AttachmentRejectedError))
+# over-long value), an AttachmentRejectedError (zip bomb) or a
+# DocumentParseError (the file cannot be converted to text at all) is
+# deterministic — retrying re-fails identically and multiplies log noise, so it
+# goes straight to the failure path.
+@celery.task(
+ **DURABLE_TASK,
+ dont_autoretry_for=(DataError, AttachmentRejectedError, DocumentParseError),
+)
@with_idempotency(
task_name="store_attachment", on_poison=_emit_attachment_poison_event,
)
diff --git a/application/parser/file/base_parser.py b/application/parser/file/base_parser.py
index ff7a095d..ad55eb7f 100644
--- a/application/parser/file/base_parser.py
+++ b/application/parser/file/base_parser.py
@@ -5,6 +5,22 @@ from pathlib import Path
from typing import Dict, List, Optional, Union
+class DocumentParseError(Exception):
+ """A file could not be converted to text.
+
+ Raised instead of returning the failure message as the document body: a
+ parser that hands back its own traceback produces an attachment/source
+ whose "content" is an error string, which then reaches the LLM as if it
+ were the document. Callers should treat this as a failed upload and tell
+ the user, rather than storing anything.
+
+ Deterministic for a given input, so it is marked non-retryable on the
+ Celery tasks that parse (alongside ``DataError`` and
+ ``AttachmentRejectedError``) — a poison file fails once instead of
+ re-failing identically on every retry.
+ """
+
+
class BaseParser:
"""Base class for all parsers."""
diff --git a/application/parser/file/bulk.py b/application/parser/file/bulk.py
index 516de94b..2ce637fc 100644
--- a/application/parser/file/bulk.py
+++ b/application/parser/file/bulk.py
@@ -1,10 +1,10 @@
"""Simple reader that reads files of different formats from a directory."""
import logging
from pathlib import Path
-from typing import Callable, Dict, List, Optional, Union
+from typing import Callable, Dict, List, Optional, Tuple, Union
from application.parser.file.base import BaseReader
-from application.parser.file.base_parser import BaseParser
+from application.parser.file.base_parser import BaseParser, DocumentParseError
from application.parser.file.docs_parser import DocxParser, PDFParser
from application.parser.file.epub_parser import EpubParser
from application.parser.file.html_parser import HTMLParser
@@ -177,6 +177,8 @@ class SimpleDirectoryReader(BaseReader):
self.file_extractor = file_extractor or DEFAULT_FILE_EXTRACTOR
self.file_metadata = file_metadata
+ # (path, message) per file skipped by ``load_data`` as unparseable.
+ self.failed_files: List[Tuple[Path, str]] = []
def _add_files(self, input_dir: Path) -> List[Path]:
"""Add files."""
@@ -230,27 +232,51 @@ class SimpleDirectoryReader(BaseReader):
Returns:
List[Document]: A list of documents.
+
+ Raises:
+ DocumentParseError: if no input file could be parsed. Individual
+ unreadable files are skipped and recorded in ``failed_files``
+ so one corrupt document does not cost the caller the rest of
+ the batch; a single-file read (the attachment path) still
+ raises, since skipping there would only defer the failure to
+ an empty result.
"""
data: Union[str, List[str]] = ""
data_list: List[str] = []
metadata_list = []
self.file_token_counts = {}
+ self.failed_files = []
total_files = len(self.input_files)
+
+ def report_progress(files_done: int) -> None:
+ if progress_callback is None:
+ return
+ try:
+ progress_callback(files_done, total_files)
+ except Exception:
+ logging.warning("load_data progress callback failed", exc_info=True)
+
for file_index, input_file in enumerate(self.input_files):
suffix_lower = input_file.suffix.lower()
parser_metadata = {}
- if suffix_lower in self.file_extractor:
- parser = self.file_extractor[suffix_lower]
- if not parser.parser_config_set:
- parser.init_parser()
- data = parser.parse_file(input_file, errors=self.errors)
- parser_metadata = parser.get_file_metadata(input_file)
- else:
- # do standard read
- with open(input_file, "r", errors=self.errors) as f:
- data = f.read()
-
+ try:
+ if suffix_lower in self.file_extractor:
+ parser = self.file_extractor[suffix_lower]
+ if not parser.parser_config_set:
+ parser.init_parser()
+ data = parser.parse_file(input_file, errors=self.errors)
+ parser_metadata = parser.get_file_metadata(input_file)
+ else:
+ # do standard read
+ with open(input_file, "r", errors=self.errors) as f:
+ data = f.read()
+ except DocumentParseError as e:
+ logging.warning(f"Skipping unreadable file {input_file.name}: {e}")
+ self.failed_files.append((input_file, str(e)))
+ report_progress(file_index + 1)
+ continue
+
# Calculate token count for this file
if isinstance(data, List):
file_tokens = sum(num_tokens_from_string(str(d)) for d in data)
@@ -288,13 +314,23 @@ class SimpleDirectoryReader(BaseReader):
data_list.append(str(data))
metadata_list.append(base_metadata)
- if progress_callback is not None:
- try:
- progress_callback(file_index + 1, total_files)
- except Exception:
- logging.warning(
- "load_data progress callback failed", exc_info=True
- )
+ report_progress(file_index + 1)
+
+ # Every file failed: there is nothing to ingest, so this is a failed
+ # read rather than an empty one. Callers (the attachment worker, the
+ # ingest tasks) treat it as terminal and tell the user.
+ if self.failed_files and not data_list:
+ if len(self.failed_files) == 1:
+ # Single-file read (the attachment path): the parser's own
+ # message reaches the user verbatim, so don't wrap it in
+ # "None of the 1 file(s)…".
+ raise DocumentParseError(self.failed_files[0][1])
+ names = ", ".join(p.name for p, _ in self.failed_files[:5])
+ raise DocumentParseError(
+ f"None of the {len(self.failed_files)} files could be parsed: "
+ f"{names}{'…' if len(self.failed_files) > 5 else ''} "
+ f"({self.failed_files[0][1]})"
+ )
# Build directory structure if input_dir is provided
if hasattr(self, 'input_dir'):
diff --git a/application/parser/file/docling_parser.py b/application/parser/file/docling_parser.py
index 50f5ef98..52eaf170 100644
--- a/application/parser/file/docling_parser.py
+++ b/application/parser/file/docling_parser.py
@@ -14,7 +14,7 @@ import zipfile
from pathlib import Path
from typing import Dict, List, Optional, Union
-from application.parser.file.base_parser import BaseParser
+from application.parser.file.base_parser import BaseParser, DocumentParseError
from application.utils import truncate_to_line_boundary
logger = logging.getLogger(__name__)
@@ -346,9 +346,14 @@ class DoclingParser(BaseParser):
except Exception as e:
logger.error(f"Error parsing file with docling: {e}", exc_info=True)
- if errors == "ignore":
- return f"[Error parsing file with docling: {str(e)}]"
- raise
+ # ``errors`` governs *decoding* leniency, not whether a total
+ # conversion failure may be substituted for the document. Returning
+ # the message here (the old ``errors == "ignore"`` branch) made the
+ # caller store the traceback as the file's text and report the
+ # upload successful — the model then read the error as the document.
+ raise DocumentParseError(
+ f"Failed to parse {Path(file).name} with docling: {e}"
+ ) from e
class DoclingPDFParser(DoclingParser):
diff --git a/application/requirements.txt b/application/requirements.txt
index da5a94ef..afae15ed 100644
--- a/application/requirements.txt
+++ b/application/requirements.txt
@@ -12,7 +12,12 @@ cryptography==46.0.7
dataclasses-json==0.6.7
daytona==0.190.1
defusedxml==0.7.1
-docling>=2.16.0
+# Pinned, not floating: docling 2.118.0 torch.compile()s its PDF layout stage,
+# which needs a C++ compiler at runtime. The runtime image has none (gcc/g++ are
+# installed in the builder stage and discarded), so a floating ">=" pulled in a
+# release that broke every PDF parse in prod on 2026-08-04. 2.84.0 is verified
+# to parse with no compiler present. Re-verify that before bumping.
+docling==2.84.0
rapidocr>=1.4.0
onnxruntime>=1.19.0
docx2txt==0.9
diff --git a/frontend/src/agents/types/schedule.ts b/frontend/src/agents/types/schedule.ts
index ff3977fa..2cbf3e83 100644
--- a/frontend/src/agents/types/schedule.ts
+++ b/frontend/src/agents/types/schedule.ts
@@ -10,6 +10,7 @@ export type ScheduleRunStatus =
| 'skipped'
| 'timeout';
+// Mirrors the schedule_runs.error_type CHECK constraint (migrations 0010, 0027).
export type ScheduleRunErrorType =
| 'auth_expired'
| 'tool_not_allowed'
@@ -18,7 +19,9 @@ export type ScheduleRunErrorType =
| 'agent_error'
| 'internal'
| 'missed'
- | 'overlap';
+ | 'overlap'
+ | 'stream_error'
+ | 'empty_output';
export type Schedule = {
id: string;
diff --git a/tests/agents/test_headless_runner_errors.py b/tests/agents/test_headless_runner_errors.py
new file mode 100644
index 00000000..f69389af
--- /dev/null
+++ b/tests/agents/test_headless_runner_errors.py
@@ -0,0 +1,99 @@
+"""``run_agent_headless`` must not swallow the agent's error event.
+
+Regression (prod 2026-08-06): ``Agent.gen`` signals a failed stream by
+yielding ``{"type": "error", "error": ...}``. The activity logger consumes
+that event (it is what produces ``activity_finished status=error
+error_class=StreamError``), but the headless runner's event loop only looked
+at the ``answer``/``sources``/``tool_calls``/``thought`` keys and dropped it.
+A failed stream therefore returned normally with an empty answer and
+``error_type=None``, and the scheduler recorded the run as ``success``.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+
+def _run(events, monkeypatch):
+ """Drive ``run_agent_headless`` over a canned agent event stream."""
+ from application.agents import headless_runner as hr
+
+ agent = MagicMock(name="agent")
+ agent.gen.return_value = iter(events)
+ agent.llm.token_usage = {"prompt_tokens": 7, "generated_tokens": 0}
+
+ retriever = MagicMock(name="retriever")
+ retriever.search.return_value = []
+
+ tool_executor = MagicMock(name="tool_executor")
+ tool_executor.headless_denials = []
+
+ monkeypatch.setattr(hr, "get_prompt", lambda _pid: "system prompt")
+ monkeypatch.setattr(
+ hr.RetrieverCreator, "create_retriever",
+ classmethod(lambda cls, *a, **kw: retriever),
+ )
+ monkeypatch.setattr(hr, "ToolExecutor", lambda *a, **kw: tool_executor)
+ monkeypatch.setattr(
+ hr.AgentCreator, "create_agent",
+ classmethod(lambda cls, *a, **kw: agent),
+ )
+
+ with patch("application.core.model_utils.validate_model_id", return_value=True), \
+ patch("application.core.model_utils.get_default_model_id", return_value="m"), \
+ patch(
+ "application.core.model_utils.get_provider_from_model_id",
+ return_value="openai",
+ ), \
+ patch("application.core.model_utils.get_api_key_for_provider", return_value="k"), \
+ patch("application.utils.calculate_doc_token_budget", return_value=1000):
+ return hr.run_agent_headless(
+ {"user_id": "u1", "id": "agent-1", "default_model_id": "m"},
+ "do the thing",
+ )
+
+
+@pytest.mark.unit
+class TestHeadlessRunnerStreamError:
+ def test_error_event_is_surfaced_as_error_type(self, monkeypatch):
+ outcome = _run(
+ [{"type": "error", "error": "Fallback LLM also failed mid-stream"}],
+ monkeypatch,
+ )
+
+ assert outcome["error_type"] == "stream_error"
+ assert "Fallback LLM also failed" in (outcome.get("error") or "")
+ assert outcome["answer"] == ""
+
+ def test_error_event_with_no_message_still_flags_failure(self, monkeypatch):
+ """An empty message must not read as falsy and report success."""
+ outcome = _run([{"type": "error"}], monkeypatch)
+
+ assert outcome["error_type"] == "stream_error"
+ assert outcome.get("error")
+
+ def test_partial_answer_before_error_still_fails(self, monkeypatch):
+ """Text already streamed does not make a broken run a success."""
+ outcome = _run(
+ [
+ {"answer": "here is half an ans"},
+ {"type": "error", "error": "peer closed"},
+ ],
+ monkeypatch,
+ )
+
+ assert outcome["answer"] == "here is half an ans"
+ assert outcome["error_type"] == "stream_error"
+
+ def test_clean_run_reports_no_error(self, monkeypatch):
+ outcome = _run(
+ [{"answer": "all good"}, {"sources": [{"title": "s"}]}],
+ monkeypatch,
+ )
+
+ assert outcome["error_type"] is None
+ assert outcome.get("error") is None
+ assert outcome["answer"] == "all good"
+ assert outcome["sources"] == [{"title": "s"}]
diff --git a/tests/agents/test_headless_runner_workflow.py b/tests/agents/test_headless_runner_workflow.py
new file mode 100644
index 00000000..e792a5ab
--- /dev/null
+++ b/tests/agents/test_headless_runner_workflow.py
@@ -0,0 +1,179 @@
+"""``run_agent_headless`` must bind a workflow agent to its workflow.
+
+Regression (prod 2026-08-06): the streaming path passes ``workflow_id`` /
+``workflow_owner`` into ``AgentCreator.create_agent`` (see
+``StreamProcessor``), but the headless path never did. A schedule or webhook
+bound to a workflow agent therefore built ``WorkflowAgent(workflow_id=None)``,
+``_load_workflow_graph()`` returned ``None``, and the agent's whole run was a
+single ``{"type": "error", "error": "Failed to load workflow configuration."}``
+— which the runner then dropped, so the schedule recorded ``success`` with no
+output for as long as the schedule existed.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+
+def _run(agent_config, monkeypatch, events=None):
+ """Drive ``run_agent_headless`` and capture the agent-factory kwargs."""
+ from application.agents import headless_runner as hr
+
+ agent = MagicMock(name="agent")
+ agent.gen.return_value = iter(events if events is not None else [{"answer": "ok"}])
+ agent.llm.token_usage = {"prompt_tokens": 1, "generated_tokens": 1}
+
+ captured = {}
+
+ def _create_agent(cls, agent_type, **kwargs):
+ captured["agent_type"] = agent_type
+ captured.update(kwargs)
+ return agent
+
+ retriever = MagicMock(name="retriever")
+ retriever.search.return_value = []
+
+ tool_executor = MagicMock(name="tool_executor")
+ tool_executor.headless_denials = []
+
+ monkeypatch.setattr(hr, "get_prompt", lambda _pid: "system prompt")
+ monkeypatch.setattr(
+ hr.RetrieverCreator, "create_retriever",
+ classmethod(lambda cls, *a, **kw: retriever),
+ )
+ monkeypatch.setattr(hr, "ToolExecutor", lambda *a, **kw: tool_executor)
+ monkeypatch.setattr(
+ hr.AgentCreator, "create_agent", classmethod(_create_agent),
+ )
+
+ with patch("application.core.model_utils.validate_model_id", return_value=True), \
+ patch("application.core.model_utils.get_default_model_id", return_value="m"), \
+ patch(
+ "application.core.model_utils.get_provider_from_model_id",
+ return_value="openai",
+ ), \
+ patch("application.core.model_utils.get_api_key_for_provider", return_value="k"), \
+ patch("application.utils.calculate_doc_token_budget", return_value=1000):
+ outcome = hr.run_agent_headless(agent_config, "do the thing")
+ return captured, outcome
+
+
+@pytest.mark.unit
+class TestHeadlessWorkflowBinding:
+ def test_workflow_agent_receives_workflow_id_and_owner(self, monkeypatch):
+ """The PG ``agents.workflow_id`` column must reach the agent."""
+ captured, _ = _run(
+ {
+ "user_id": "u1",
+ "id": "agent-1",
+ "agent_type": "workflow",
+ "workflow_id": "0f0c4b9e-1111-2222-3333-444455556666",
+ "default_model_id": "m",
+ },
+ monkeypatch,
+ )
+
+ assert captured["agent_type"] == "workflow"
+ assert captured["workflow_id"] == "0f0c4b9e-1111-2222-3333-444455556666"
+ # Without an owner the repository lookup is unscoped and returns nothing.
+ assert captured["workflow_owner"] == "u1"
+
+ def test_legacy_workflow_key_also_binds(self, monkeypatch):
+ """Legacy Mongo shape stored the reference under ``workflow``."""
+ captured, _ = _run(
+ {
+ "user_id": "u1",
+ "id": "agent-1",
+ "agent_type": "workflow",
+ "workflow": "wf-legacy-ref",
+ "default_model_id": "m",
+ },
+ monkeypatch,
+ )
+
+ assert captured["workflow_id"] == "wf-legacy-ref"
+ assert captured["workflow_owner"] == "u1"
+
+ def test_embedded_graph_is_passed_as_workflow(self, monkeypatch):
+ """An inline graph goes to ``workflow=``, not stringified into an id."""
+ graph = {"name": "inline", "nodes": [], "edges": []}
+ captured, _ = _run(
+ {
+ "user_id": "u1",
+ "id": "agent-1",
+ "agent_type": "workflow",
+ "workflow": graph,
+ "default_model_id": "m",
+ },
+ monkeypatch,
+ )
+
+ assert captured["workflow"] == graph
+ assert "workflow_id" not in captured
+ assert captured["workflow_owner"] == "u1"
+
+ def test_non_workflow_agent_gets_no_workflow_kwargs(self, monkeypatch):
+ """``ClassicAgent`` has no such kwargs; passing them would TypeError."""
+ captured, _ = _run(
+ {
+ "user_id": "u1",
+ "id": "agent-1",
+ "agent_type": "classic",
+ "workflow_id": "should-be-ignored",
+ "default_model_id": "m",
+ },
+ monkeypatch,
+ )
+
+ assert "workflow_id" not in captured
+ assert "workflow" not in captured
+ assert "workflow_owner" not in captured
+
+ def test_workflow_steps_are_counted_as_work(self, monkeypatch):
+ """Completed nodes are the workflow's proof of work.
+
+ A workflow's tool calls never reach the caller as ``tool_calls``
+ events (the engine keeps them in its execution log) and its node
+ agents carry their own LLMs, so the runner's token tally is 0. Without
+ a step count, a side-effecting workflow looks identical to one that
+ did nothing at all.
+ """
+ _, outcome = _run(
+ {
+ "user_id": "u1",
+ "id": "agent-1",
+ "agent_type": "workflow",
+ "workflow_id": "wf-1",
+ "default_model_id": "m",
+ },
+ monkeypatch,
+ events=[
+ {"type": "workflow_step", "node_id": "n1", "status": "running"},
+ {"type": "workflow_step", "node_id": "n1", "status": "completed"},
+ {"type": "workflow_step", "node_id": "n2", "status": "completed"},
+ ],
+ )
+
+ assert outcome["steps_completed"] == 2
+ assert outcome["error_type"] is None
+
+ def test_failed_steps_are_not_counted(self, monkeypatch):
+ _, outcome = _run(
+ {
+ "user_id": "u1",
+ "id": "agent-1",
+ "agent_type": "workflow",
+ "workflow_id": "wf-1",
+ "default_model_id": "m",
+ },
+ monkeypatch,
+ events=[
+ {"type": "workflow_step", "node_id": "n1", "status": "failed"},
+ {"type": "error", "error": "Node http_1: connection refused"},
+ ],
+ )
+
+ assert outcome["steps_completed"] == 0
+ assert outcome["error_type"] == "stream_error"
diff --git a/tests/api/user/test_scheduler_worker.py b/tests/api/user/test_scheduler_worker.py
index c02346a8..0d3be23b 100644
--- a/tests/api/user/test_scheduler_worker.py
+++ b/tests/api/user/test_scheduler_worker.py
@@ -173,6 +173,171 @@ class TestExecuteScheduledRunBody:
assert row["status"] == "failed"
assert row["error_type"] == "tool_not_allowed"
+ def test_stream_error_from_runner_marks_failed(
+ self, pg_engine, patched_engine, stub_events,
+ ):
+ """The runner's ``error_type`` must be honoured, not re-derived.
+
+ Regression: the worker only ever inferred ``tool_not_allowed`` itself
+ and never read ``outcome["error_type"]``, so a stream that failed
+ mid-flight was recorded ``success``.
+ """
+ with pg_engine.begin() as conn:
+ schedule, run, _ = _make_pending_run(conn)
+ with patch(
+ "application.api.user.scheduler_worker.run_agent_headless",
+ return_value={
+ "answer": "",
+ "tool_calls": [],
+ "sources": [],
+ "thought": "",
+ "prompt_tokens": 9417,
+ "generated_tokens": 0,
+ "denied": [],
+ "error_type": "stream_error",
+ "error": "Fallback LLM also failed mid-stream; giving up",
+ "model_id": "fake",
+ },
+ ):
+ result = execute_scheduled_run_body(str(run["id"]), "celery-se")
+ assert result["status"] == "failed"
+ with pg_engine.connect() as conn:
+ row = ScheduleRunsRepository(conn).get_internal(str(run["id"]))
+ sched = SchedulesRepository(conn).get_internal(str(schedule["id"]))
+ assert row["status"] == "failed"
+ assert row["error_type"] == "stream_error"
+ assert "Fallback LLM also failed" in (row["error"] or "")
+ assert sched["consecutive_failure_count"] == 1
+ assert "schedule.run.failed" in {e[0] for e in stub_events}
+
+ def test_empty_output_run_marks_failed(
+ self, pg_engine, patched_engine, stub_events,
+ ):
+ """A run that produced nothing at all is not a success.
+
+ This is the exact prod shape: seven consecutive daily runs recorded
+ ``success`` with NULL output and 0/0 tokens, so nothing surfaced that
+ the user's scheduled agent had been dead since it was created. It is
+ the backstop for silent paths that emit no error event either.
+ """
+ with pg_engine.begin() as conn:
+ schedule, run, _ = _make_pending_run(conn)
+ with patch(
+ "application.api.user.scheduler_worker.run_agent_headless",
+ return_value={
+ "answer": "",
+ "tool_calls": [],
+ "sources": [],
+ "thought": "",
+ "prompt_tokens": 0,
+ "generated_tokens": 0,
+ "denied": [],
+ "error_type": None,
+ "model_id": "fake",
+ },
+ ):
+ result = execute_scheduled_run_body(str(run["id"]), "celery-eo")
+ assert result["status"] == "failed"
+ with pg_engine.connect() as conn:
+ row = ScheduleRunsRepository(conn).get_internal(str(run["id"]))
+ assert row["status"] == "failed"
+ assert row["error_type"] == "empty_output"
+
+ def test_tool_call_only_run_is_not_empty_output(
+ self, pg_engine, patched_engine, stub_events,
+ ):
+ """A run whose work was tool side effects still counts as a success.
+
+ Guards the backstop against over-reach: "no prose answer" is normal
+ for a schedule whose whole job is to call a tool (post to Slack, file
+ a ticket), so tokens or tool calls are enough to call it a success.
+ """
+ with pg_engine.begin() as conn:
+ schedule, run, _ = _make_pending_run(conn)
+ with patch(
+ "application.api.user.scheduler_worker.run_agent_headless",
+ return_value={
+ "answer": "",
+ "tool_calls": [{"tool_name": "telegram_send", "result": "ok"}],
+ "sources": [],
+ "thought": "",
+ "prompt_tokens": 120,
+ "generated_tokens": 8,
+ "denied": [],
+ "error_type": None,
+ "model_id": "fake",
+ },
+ ):
+ result = execute_scheduled_run_body(str(run["id"]), "celery-tc")
+ assert result["status"] == "success"
+ with pg_engine.connect() as conn:
+ row = ScheduleRunsRepository(conn).get_internal(str(run["id"]))
+ assert row["status"] == "success"
+ assert row["error_type"] is None
+
+ def test_workflow_run_with_completed_steps_is_not_empty_output(
+ self, pg_engine, patched_engine, stub_events,
+ ):
+ """A workflow that ran its nodes did work, however quiet the stream.
+
+ Workflow tool calls never surface as ``tool_calls`` events (the engine
+ keeps them in its execution log) and node agents own their LLMs, so
+ the runner reports 0 generated tokens. A workflow whose nodes don't
+ stream and whose end node has no output template therefore matches the
+ empty-output shape exactly — ``steps_completed`` is what tells them
+ apart. Without it, every such schedule fails and then autopauses.
+ """
+ with pg_engine.begin() as conn:
+ schedule, run, _ = _make_pending_run(conn)
+ with patch(
+ "application.api.user.scheduler_worker.run_agent_headless",
+ return_value={
+ "answer": "",
+ "tool_calls": [],
+ "sources": [],
+ "thought": "",
+ "prompt_tokens": 0,
+ "generated_tokens": 0,
+ "denied": [],
+ "error_type": None,
+ "steps_completed": 3,
+ "model_id": "fake",
+ },
+ ):
+ result = execute_scheduled_run_body(str(run["id"]), "celery-wf")
+ assert result["status"] == "success"
+ with pg_engine.connect() as conn:
+ row = ScheduleRunsRepository(conn).get_internal(str(run["id"]))
+ assert row["status"] == "success"
+ assert row["error_type"] is None
+
+ def test_workflow_run_with_no_completed_steps_is_empty_output(
+ self, pg_engine, patched_engine, stub_events,
+ ):
+ """A workflow that completed no node still did nothing."""
+ with pg_engine.begin() as conn:
+ schedule, run, _ = _make_pending_run(conn)
+ with patch(
+ "application.api.user.scheduler_worker.run_agent_headless",
+ return_value={
+ "answer": "",
+ "tool_calls": [],
+ "sources": [],
+ "thought": "",
+ "prompt_tokens": 0,
+ "generated_tokens": 0,
+ "denied": [],
+ "error_type": None,
+ "steps_completed": 0,
+ "model_id": "fake",
+ },
+ ):
+ result = execute_scheduled_run_body(str(run["id"]), "celery-wf0")
+ assert result["status"] == "failed"
+ with pg_engine.connect() as conn:
+ row = ScheduleRunsRepository(conn).get_internal(str(run["id"]))
+ assert row["error_type"] == "empty_output"
+
def test_one_time_loads_chat_history(
self, pg_engine, patched_engine, stub_events,
):
diff --git a/tests/parser/file/test_bulk.py b/tests/parser/file/test_bulk.py
index 9e210e7f..5a81d2d4 100644
--- a/tests/parser/file/test_bulk.py
+++ b/tests/parser/file/test_bulk.py
@@ -4,6 +4,7 @@ Covers: SimpleDirectoryReader (init, file discovery, load_data, directory
structure building), get_default_file_extractor.
"""
+from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
@@ -158,6 +159,119 @@ class TestSimpleDirectoryReaderLoadData:
for doc in docs:
assert isinstance(doc, Document)
+ def test_unparseable_file_is_skipped_not_fatal(self, tmp_path):
+ """One bad file must not cost the user the other 99.
+
+ Parsers raise ``DocumentParseError`` rather than returning their own
+ traceback as the document text, so an unguarded loop turns a single
+ corrupt PDF into a failed ingest for the whole zip/folder/sync.
+ """
+ from application.parser.file.base_parser import DocumentParseError
+ from application.parser.file.bulk import SimpleDirectoryReader
+
+ (tmp_path / "good1.md").write_text("first")
+ (tmp_path / "bad.md").write_text("corrupt")
+ (tmp_path / "good2.md").write_text("second")
+
+ def _parse(path, **kwargs):
+ if path.name == "bad.md":
+ raise DocumentParseError("Failed to parse bad.md with docling: boom")
+ return f"parsed {path.name}"
+
+ mock_parser = MagicMock()
+ mock_parser.parser_config_set = True
+ mock_parser.parse_file.side_effect = _parse
+ mock_parser.get_file_metadata.return_value = {}
+
+ reader = SimpleDirectoryReader(
+ input_dir=str(tmp_path),
+ file_extractor={".md": mock_parser},
+ recursive=False,
+ exclude_hidden=True,
+ )
+ docs = reader.load_data()
+
+ texts = sorted(doc.text for doc in docs)
+ assert texts == ["parsed good1.md", "parsed good2.md"]
+ # The skip is recorded, not silent: callers can surface it.
+ assert [Path(p).name for p, _ in reader.failed_files] == ["bad.md"]
+ assert "boom" in reader.failed_files[0][1]
+
+ def test_single_unparseable_file_still_raises(self, tmp_path):
+ """The attachment path parses exactly one file and must fail loudly.
+
+ Skipping there would hand ``load_data()[0]`` an empty list and turn a
+ clear "this PDF could not be read" into an IndexError.
+ """
+ from application.parser.file.base_parser import DocumentParseError
+ from application.parser.file.bulk import SimpleDirectoryReader
+
+ (tmp_path / "bad.pdf").write_text("not really a pdf")
+
+ mock_parser = MagicMock()
+ mock_parser.parser_config_set = True
+ mock_parser.parse_file.side_effect = DocumentParseError(
+ "Failed to parse bad.pdf with docling: InvalidCxxCompiler"
+ )
+
+ reader = SimpleDirectoryReader(
+ input_files=[str(tmp_path / "bad.pdf")],
+ file_extractor={".pdf": mock_parser},
+ )
+ # The parser's own message reaches the user verbatim, unwrapped.
+ with pytest.raises(DocumentParseError, match="InvalidCxxCompiler"):
+ reader.load_data()
+
+ def test_all_files_unparseable_raises(self, tmp_path):
+ """Nothing parsed is a failed ingest, not an empty success."""
+ from application.parser.file.base_parser import DocumentParseError
+ from application.parser.file.bulk import SimpleDirectoryReader
+
+ (tmp_path / "a.md").write_text("x")
+ (tmp_path / "b.md").write_text("y")
+
+ mock_parser = MagicMock()
+ mock_parser.parser_config_set = True
+ mock_parser.parse_file.side_effect = DocumentParseError("nope")
+
+ reader = SimpleDirectoryReader(
+ input_dir=str(tmp_path),
+ file_extractor={".md": mock_parser},
+ recursive=False,
+ exclude_hidden=True,
+ )
+ with pytest.raises(DocumentParseError) as excinfo:
+ reader.load_data()
+ assert "None of the 2 files" in str(excinfo.value)
+
+ def test_skipped_file_still_advances_progress(self, tmp_path):
+ """Progress must not stall on a skipped file."""
+ from application.parser.file.base_parser import DocumentParseError
+ from application.parser.file.bulk import SimpleDirectoryReader
+
+ (tmp_path / "good.md").write_text("ok")
+ (tmp_path / "bad.md").write_text("bad")
+
+ def _parse(path, **kwargs):
+ if path.name == "bad.md":
+ raise DocumentParseError("boom")
+ return "ok"
+
+ mock_parser = MagicMock()
+ mock_parser.parser_config_set = True
+ mock_parser.parse_file.side_effect = _parse
+ mock_parser.get_file_metadata.return_value = {}
+
+ reader = SimpleDirectoryReader(
+ input_dir=str(tmp_path),
+ file_extractor={".md": mock_parser},
+ recursive=False,
+ exclude_hidden=True,
+ )
+ calls = []
+ reader.load_data(progress_callback=lambda done, total: calls.append((done, total)))
+ assert [c[0] for c in calls] == [1, 2]
+
def test_load_data_progress_callback_fires_per_file(self, temp_dir):
from application.parser.file.bulk import SimpleDirectoryReader
diff --git a/tests/parser/file/test_docling_parser.py b/tests/parser/file/test_docling_parser.py
index 13b3a8b2..e4e8beee 100644
--- a/tests/parser/file/test_docling_parser.py
+++ b/tests/parser/file/test_docling_parser.py
@@ -270,7 +270,16 @@ class TestDoclingParserParseFile:
result = parser.parse_file(Path("test.pdf"))
assert "content" in result
- def test_parse_file_error_ignore(self):
+ def test_parse_file_error_ignore_raises_instead_of_returning_error_text(self):
+ """A failed conversion must never become the document's text.
+
+ Regression: ``errors="ignore"`` used to return
+ ``"[Error parsing file with docling: ...]"``, which the attachment
+ worker then stored as ``attachments.content`` and handed to the LLM
+ as if it were the PDF. ``errors`` controls *decoding* leniency, not
+ "substitute the traceback for the document".
+ """
+ from application.parser.file.base_parser import DocumentParseError
from application.parser.file.docling_parser import DoclingParser
parser = DoclingParser()
@@ -278,8 +287,37 @@ class TestDoclingParserParseFile:
mock_converter.convert.side_effect = Exception("Parse failed")
parser._converter = mock_converter
- result = parser.parse_file(Path("bad.pdf"), errors="ignore")
- assert "Error" in result
+ with pytest.raises(DocumentParseError) as excinfo:
+ parser.parse_file(Path("bad.pdf"), errors="ignore")
+
+ # The message identifies the file and preserves the cause for triage…
+ assert "bad.pdf" in str(excinfo.value)
+ assert "Parse failed" in str(excinfo.value)
+ # …and the original exception is chained, not swallowed.
+ assert isinstance(excinfo.value.__cause__, Exception)
+ assert "Parse failed" in str(excinfo.value.__cause__)
+
+ def test_parse_file_error_ignore_never_returns_a_string(self):
+ """Belt-and-braces: no code path may hand back error text as content."""
+ from application.parser.file.base_parser import DocumentParseError
+ from application.parser.file.docling_parser import DoclingParser
+
+ parser = DoclingParser()
+ mock_converter = MagicMock()
+ mock_converter.convert.side_effect = RuntimeError(
+ "Conversion failed for: x.pdf with status: failure. Errors: "
+ "InvalidCxxCompiler: No working C++ compiler found"
+ )
+ parser._converter = mock_converter
+
+ try:
+ result = parser.parse_file(Path("x.pdf"), errors="ignore")
+ except DocumentParseError:
+ return # expected
+ pytest.fail(
+ f"parse_file returned {result!r} instead of raising; error text "
+ "must not be usable as document content"
+ )
def test_parse_file_error_raise(self):
from application.parser.file.docling_parser import DoclingParser
diff --git a/tests/worker/test_attachment_worker.py b/tests/worker/test_attachment_worker.py
index c6a63aca..1f1e8624 100644
--- a/tests/worker/test_attachment_worker.py
+++ b/tests/worker/test_attachment_worker.py
@@ -67,6 +67,74 @@ class TestAttachmentWorker:
assert row["content"] == "hello world"
assert row["user_id"] == "user1"
+ def test_parse_failure_stores_nothing_and_tells_the_user(
+ self, pg_conn, patch_worker_db, task_self, monkeypatch
+ ):
+ """A parse failure must fail loudly, not store the error as content.
+
+ Regression (prod 2026-08-05): docling's PDF layout stage died, the
+ parser returned its own traceback as the "document", and the worker
+ stored it and published ``attachment.completed`` — so the upload
+ looked fine and the model was handed an error message as the PDF.
+ """
+ from application import worker
+ from application.parser.file.base_parser import DocumentParseError
+
+ published: list[tuple[str, dict]] = []
+ monkeypatch.setattr(
+ worker,
+ "publish_user_event",
+ lambda user, event, payload, **kw: published.append((event, payload)),
+ )
+
+ fake_storage = MagicMock(name="storage")
+ fake_storage.process_file.side_effect = DocumentParseError(
+ "Failed to parse scan.pdf with docling: Conversion failed for: "
+ "scan.pdf with status: failure. Errors: InvalidCxxCompiler"
+ )
+ monkeypatch.setattr(
+ worker.StorageCreator, "get_storage", lambda: fake_storage
+ )
+ monkeypatch.setattr(
+ worker, "get_default_file_extractor", lambda ocr_enabled=False: {}
+ )
+
+ file_info = {
+ "filename": "scan.pdf",
+ "attachment_id": "507f1f77bcf86cd799439012",
+ "path": "uploads/user1/scan.pdf",
+ "metadata": {"source": "chat"},
+ }
+
+ with pytest.raises(DocumentParseError):
+ worker.attachment_worker(task_self, file_info, "user1")
+
+ # Nothing may be persisted — an attachment whose text is a traceback
+ # is worse than no attachment, because the model will read it.
+ row = AttachmentsRepository(pg_conn).get_by_legacy_id(
+ file_info["attachment_id"], "user1"
+ )
+ assert row is None, "a failed parse must not insert an attachment row"
+
+ # The user is told it failed, and never told it completed.
+ events = [event for event, _ in published]
+ assert "attachment.failed" in events
+ assert "attachment.completed" not in events
+
+ @pytest.mark.parametrize("task_name", ["store_attachment", "ingest"])
+ def test_parse_failure_is_not_retried(self, task_name):
+ """A parse failure is deterministic; retrying only multiplies noise.
+
+ Both parsing entry points matter: now that the parser raises instead of
+ returning its traceback as content, an unguarded task would turn one
+ unreadable upload into a retry loop of identical failures.
+ """
+ from application.api.user import tasks as user_tasks
+ from application.parser.file.base_parser import DocumentParseError
+
+ task = getattr(user_tasks, task_name)
+ assert DocumentParseError in task.dont_autoretry_for
+
@pytest.mark.unit
class TestBoundedAttachmentCopy:
From 73e130591c8783a2b8d0a251ab4248afe26c1c07 Mon Sep 17 00:00:00 2001
From: Alex
Date: Thu, 6 Aug 2026 12:30:01 +0100
Subject: [PATCH 2/2] fix: remove comment
---
application/requirements.txt | 7 +------
1 file changed, 1 insertion(+), 6 deletions(-)
diff --git a/application/requirements.txt b/application/requirements.txt
index afae15ed..6773338e 100644
--- a/application/requirements.txt
+++ b/application/requirements.txt
@@ -12,11 +12,6 @@ cryptography==46.0.7
dataclasses-json==0.6.7
daytona==0.190.1
defusedxml==0.7.1
-# Pinned, not floating: docling 2.118.0 torch.compile()s its PDF layout stage,
-# which needs a C++ compiler at runtime. The runtime image has none (gcc/g++ are
-# installed in the builder stage and discarded), so a floating ">=" pulled in a
-# release that broke every PDF parse in prod on 2026-08-04. 2.84.0 is verified
-# to parse with no compiler present. Re-verify that before bumping.
docling==2.84.0
rapidocr>=1.4.0
onnxruntime>=1.19.0
@@ -117,4 +112,4 @@ yarl==1.23.0
markdownify==1.2.2
tldextract==5.3.1
websockets==16.0
-websocket-client==1.9.0
\ No newline at end of file
+websocket-client==1.9.0