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: