Merge pull request #2637 from arc53/fix/docling-parse-error-and-schedule-status

Fix/docling parse error and schedule status

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,
     }

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)

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:

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,
 )

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."""
 

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'):

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):

application/requirements.txt

@@ -12,7 +12,7 @@ cryptography==46.0.7
 dataclasses-json==0.6.7
 daytona==0.190.1
 defusedxml==0.7.1
-docling>=2.16.0
+docling==2.84.0
 rapidocr>=1.4.0
 onnxruntime>=1.19.0
 docx2txt==0.9
@@ -112,4 +112,4 @@ yarl==1.23.0
 markdownify==1.2.2
 tldextract==5.3.1
 websockets==16.0
-websocket-client==1.9.0
+websocket-client==1.9.0

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;

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"}]

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"

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,
     ):

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
 

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

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: