Merge pull request #2635 from arc53/fix/workflow-errors-and-attachment-degrade

Fix/workflow errors and attachment degrade

application/agents/workflows/cel_evaluator.py

@@ -1,3 +1,4 @@
+import re
 from typing import Any, Dict
 
 import celpy
@@ -32,19 +33,112 @@ def build_activation(state: Dict[str, Any]) -> Dict[str, Any]:
     return {k: _convert_value(v) for k, v in state.items()}
 
 
+_MAX_ERROR_CHARS = 200
+
+
+def _summarize_cel_error(exc: Exception) -> str:
+    """Reduce a celpy error to a short, value-free description.
+
+    Two problems with the raw text. ``undeclared reference`` errors embed a
+    repr of the *entire* activation — every binding, thousands of characters,
+    including the user's own query. Others interpolate the offending **value**
+    (``invalid literal for int() with base 10: StringType('…')``). Both get
+    logged and, now that config errors skip ``sanitize_api_error``, shown. So
+    strip the activation dump, drop anything that looks like a quoted value,
+    and cap the result.
+    """
+    text = " ".join(str(exc).split())
+    marker = " (in activation"
+    if marker in text:
+        text = text.split(marker, 1)[0]
+    text = text.strip().lstrip("(").strip()
+
+    # Redact quoted fragments by DEFAULT. celpy interpolates state values into
+    # messages (``StringType('SECRET')``), and a value that happens to look
+    # like an identifier is indistinguishable from a variable name — so allow
+    # only the two positions that are known to hold names, not data: the
+    # offending variable in "undeclared reference to 'x'", and the exception
+    # class in "<class 'ValueError'>".
+    def _redact(match: "re.Match[str]") -> str:
+        inner = match.group(1)
+        prefix = text[max(0, match.start() - 14):match.start()]
+        if prefix.endswith("reference to ") or prefix.endswith("<class "):
+            return f"'{inner}'"
+        return "'…'"
+
+    text = re.sub(r"'([^']*)'", _redact, text)
+    text = text.strip().strip('"').strip()
+    if len(text) > _MAX_ERROR_CHARS:
+        text = text[: _MAX_ERROR_CHARS - 1].rstrip() + "…"
+    return text
+
+
+def _looks_like_template_syntax(expression: str) -> bool:
+    """Whether the source contains a ``{{ … }}`` substitution.
+
+    Only consulted **after** compilation has already failed: ``{{`` is legal
+    CEL inside a string literal (``x == "{{y}}"``) and at the head of a nested
+    map literal (``{{"a": 1}: 2}``), both of which compile fine and must not be
+    rejected.
+    """
+    return "{{" in expression and "}}" in expression
+
+
+def _template_syntax_error(expression: str) -> "CelEvaluationError":
+    inner = expression.split("{{", 1)[1].split("}}", 1)[0].strip()
+    suggestion = inner or "query"
+    return CelEvaluationError(
+        "This field takes a CEL expression, not {{ }} template syntax. "
+        f"Write {suggestion} instead of {{{{{suggestion}}}}}."
+    )
+
+
+def _compile(expression: str):
+    """Compile, translating failures into actionable messages."""
+    try:
+        return celpy.Environment().compile(expression)
+    except Exception as exc:
+        # Agent ``prompt_template`` and end ``output_template`` fields *are*
+        # Jinja2, so ``{{query}}`` is correct one panel over. Here it fails to
+        # parse with a bare caret, which reads as "the product is broken"
+        # rather than "wrong syntax for this field".
+        if _looks_like_template_syntax(expression):
+            raise _template_syntax_error(expression) from exc
+        raise CelEvaluationError(f"CEL error: {_summarize_cel_error(exc)}") from exc
+
+
+def validate_cel_expression(expression: str) -> None:
+    """Compile-check an expression without running it.
+
+    Used at save time. Only syntax is knowable then — workflow state is built
+    during a run, so unresolved names are *not* an error here.
+
+    Args:
+        expression: The CEL source to check.
+
+    Raises:
+        CelEvaluationError: If the expression is empty or does not compile.
+    """
+    if not expression or not expression.strip():
+        raise CelEvaluationError("Empty expression")
+    _compile(expression)
+
+
 def evaluate_cel(expression: str, state: Dict[str, Any]) -> Any:
     if not expression or not expression.strip():
         raise CelEvaluationError("Empty expression")
+    ast = _compile(expression)
     try:
         env = celpy.Environment()
-        ast = env.compile(expression)
         program = env.program(ast)
         activation = build_activation(state)
         result = program.evaluate(activation)
     except celpy.CELEvalError as exc:
-        raise CelEvaluationError(f"CEL evaluation error: {exc}") from exc
+        raise CelEvaluationError(
+            f"CEL evaluation error: {_summarize_cel_error(exc)}"
+        ) from exc
     except Exception as exc:
-        raise CelEvaluationError(f"CEL error: {exc}") from exc
+        raise CelEvaluationError(f"CEL error: {_summarize_cel_error(exc)}") from exc
     return cel_to_python(result)
 
 

application/agents/workflows/workflow_engine.py

@@ -165,7 +165,19 @@ class WorkflowEngine:
                 self._finalize_log_entry(log_entry, pre_state)
                 self.execution_log.append(log_entry)
 
-                user_friendly_error = sanitize_api_error(e)
+                # A misconfigured expression is the builder's own typo, and the
+                # message says exactly how to fix it. Routing it through
+                # sanitize_api_error would match the ``{`` in ``{{query}}`` and
+                # replace the whole thing with "An error occurred … try again
+                # later" — advice that sends the user round the same loop.
+                # Everything else stays sanitized: node internals and upstream
+                # provider errors are not for end users.
+                is_config_error = isinstance(e, CelEvaluationError)
+                user_friendly_error = (
+                    f"{node.title or node.type.value}: {e}"
+                    if is_config_error
+                    else sanitize_api_error(e)
+                )
                 yield {
                     "type": "workflow_step",
                     "node_id": node.id,
@@ -175,7 +187,11 @@ class WorkflowEngine:
                     "state_delta": log_entry["state_delta"],
                     "error": user_friendly_error,
                 }
-                yield {"type": "error", "error": user_friendly_error}
+                yield {
+                    "type": "error",
+                    "error": user_friendly_error,
+                    "user_facing": is_config_error,
+                }
                 break
             self.execution_log.append(log_entry)
             pre_state = dict(self.state)
@@ -291,7 +307,7 @@ class WorkflowEngine:
         from application.core.model_utils import (
             get_api_key_for_provider,
             get_model_capabilities,
-            get_provider_from_model_id,
+            resolve_dispatch_provider,
         )
 
         node_config = AgentNodeConfig(**node.config.get("config", node.config))
@@ -314,13 +330,22 @@ class WorkflowEngine:
             if isinstance(self.agent.decoded_token, dict)
             else None
         )
-        node_llm_name = (
-            node_config.llm_name
-            or get_provider_from_model_id(
-                node_model_id or "", user_id=node_user_id
-            )
-            or self.agent.llm_name
+        # ``node_config.llm_name`` is whatever the builder stored, which for
+        # catalogs with a ``display_provider`` is a presentation label
+        # ("foundry", "azure_foundry", "cloudflare") that no LLM class is
+        # registered under. Trusting it verbatim fails the node before any LLM
+        # call and hands the user a blank answer, so resolve it to a real
+        # dispatch provider first — this also repairs nodes already saved with
+        # a label, without a migration.
+        node_llm_name = resolve_dispatch_provider(
+            node_config.llm_name,
+            node_model_id,
+            user_id=node_user_id,
+            fallback=self.agent.llm_name,
         )
+        # Resolve the key from the normalized name: get_api_key_for_provider
+        # silently returns settings.API_KEY for names it does not know, so a
+        # label here would leak the deployment key to the wrong endpoint.
         node_api_key = get_api_key_for_provider(node_llm_name) or self.agent.api_key
 
         # Structured output gates on the model's registry capability flags;

application/api/answer/routes/base.py

@@ -246,6 +246,14 @@ class BaseAnswerResource:
         structured_chunks = []
         query_metadata: Dict[str, Any] = {}
         paused = False
+        # Set when the agent *yields* a terminal ``error`` event instead of
+        # raising. Workflow node failures take that route (the engine catches
+        # the node exception and reports it as an event), so the generator
+        # returns normally and the ``except`` handler below never runs. Without
+        # this flag the turn was finalized ``complete`` with an empty response:
+        # the live client showed an error bubble, but on reload history mapped
+        # the row to a blank answer with no error text and no retry.
+        stream_error: Optional[str] = None
         # A ``tool_calls_pending`` event is held back and only flushed after
         # continuation state is committed (or the stateless finalize path is
         # reached): the v1 translator turns it into ``finish_reason:"tool_calls"``,
@@ -568,6 +576,7 @@ class BaseAnswerResource:
                         error_text = line.get("error", "An error occurred")
                         if not line.get("user_facing"):
                             error_text = sanitize_api_error(error_text)
+                        stream_error = error_text
                         yield _emit({"type": "error", "error": error_text})
                     elif line.get("type") == "notice":
                         # Non-fatal, non-terminal notice (e.g. some workflow input
@@ -595,6 +604,14 @@ class BaseAnswerResource:
                     }
                 )
 
+            # Record a yielded error before any early return so the pause /
+            # stateless-tool-round paths persist it too. No producer currently
+            # emits a non-terminal error and then pauses, but leaving the only
+            # write below the pause blocks would make that combination lose the
+            # error silently — the exact shape of the bug being fixed here.
+            if stream_error:
+                query_metadata.setdefault("error", stream_error)
+
             # ---- Paused: save continuation state and end stream early ----
             if paused:
                 continuation = getattr(agent, "_pending_continuation", None)
@@ -848,6 +865,19 @@ class BaseAnswerResource:
                 )
                 llm._token_usage_source = "title"
 
+            # The error was recorded above so the failure stays greppable, but
+            # it only *fails* the turn when nothing was produced. An error
+            # arriving after partial output (e.g. a later workflow node) must
+            # stay ``complete``, since the client only renders ``response`` for
+            # complete rows — failing it would discard text the user already
+            # saw. ``structured_chunks`` counts as output for the same reason:
+            # a structured answer lives there, not in ``response_full``.
+            errored_empty = (
+                bool(stream_error)
+                and not response_full.strip()
+                and not structured_chunks
+            )
+
             if should_persist:
                 if reserved_message_id is not None:
                     self.conversation_service.finalize_message(
@@ -858,7 +888,7 @@ class BaseAnswerResource:
                         tool_calls=tool_calls,
                         model_id=model_id or self.default_model_id,
                         metadata=query_metadata if query_metadata else None,
-                        status="complete",
+                        status="failed" if errored_empty else "complete",
                         title_inputs={
                             "llm": llm,
                             "question": question,
@@ -889,6 +919,7 @@ class BaseAnswerResource:
                         attachment_ids=attachment_ids,
                         metadata=query_metadata if query_metadata else None,
                         visibility=visibility,
+                        status="failed" if errored_empty else "complete",
                     )
                 # Persist compression metadata/summary if it exists and wasn't saved mid-execution
                 compression_meta = getattr(agent, "compression_metadata", None)

application/api/answer/services/conversation_service.py

@@ -100,9 +100,15 @@ class ConversationService:
         attachment_ids: Optional[List[str]] = None,
         metadata: Optional[Dict[str, Any]] = None,
         visibility: str = "hidden",
+        status: str = "complete",
     ) -> str:
         """Save or update a conversation in Postgres.
 
+        ``status`` lets a caller record a turn that failed without producing
+        an answer. It defaults to ``complete``, matching the column default
+        this path relied on before; passing ``failed`` is what stops a blank
+        errored turn from rendering as an empty bubble with no retry.
+
         Returns the string conversation id (PG UUID as string, or the
         caller-provided id if it was already a UUID).
         """
@@ -127,6 +133,7 @@ class ConversationService:
             "attachments": attachment_ids,
             "model_id": model_id,
             "timestamp": current_time,
+            "status": status,
         }
         if metadata:
             message_payload["metadata"] = metadata
@@ -410,7 +417,12 @@ class ConversationService:
             repo.confirm_executed_tool_calls(message_id)
 
         # Outside the txn — title-gen is a multi-second LLM round trip.
-        if title_inputs and status == "complete":
+        # ``failed`` counts too: the conversation is still listed, and
+        # ``_maybe_generate_title`` only regenerates while the name is still
+        # the question-prefix fallback. Skipping it here would strand a
+        # conversation whose first turn failed with the raw prompt as its
+        # name forever, because by turn two the fallback no longer matches.
+        if title_inputs and status in ("complete", "failed"):
             if async_title_generation:
                 threading.Thread(
                     target=self._generate_title_safely,

application/api/user/workflows/routes.py

@@ -5,6 +5,10 @@ from typing import Any, Dict, List, Optional, Set
 from flask import current_app, request
 from flask_restx import Namespace, Resource
 
+from application.agents.workflows.cel_evaluator import (
+    CelEvaluationError,
+    validate_cel_expression,
+)
 from application.storage.db.base_repository import looks_like_uuid
 from application.storage.db.repositories.workflow_edges import WorkflowEdgesRepository
 from application.storage.db.repositories.workflow_nodes import WorkflowNodesRepository
@@ -323,6 +327,14 @@ def validate_workflow_structure(
                 errors.append(
                     f"Condition node '{cnode_title}' case '{handle}' has an outgoing edge but no expression"
                 )
+            if has_expression:
+                try:
+                    validate_cel_expression(raw_expression)
+                except CelEvaluationError as exc:
+                    errors.append(
+                        f"Condition node '{cnode_title}' case '{handle}' "
+                        f"has an invalid expression: {exc}"
+                    )
 
         for handle, handle_edges in outgoing_by_handle.items():
             if not handle:
@@ -335,6 +347,55 @@ def validate_workflow_structure(
                         f"must eventually reach an end node"
                     )
 
+    # Set State nodes were validated nowhere. A node whose CEL does not compile
+    # saves and publishes clean, then aborts the run on first execution — the
+    # user only finds out as a failed answer, with no pointer to the node.
+    state_nodes = [n for n in nodes if n.get("type") == "state"]
+    for snode in state_nodes:
+        snode_title = snode.get("title", snode.get("id"))
+        node_data = snode.get("data", {}) or {}
+        # The builder writes state config under ``data.config`` but other
+        # payloads keep it flat on ``data``; the engine reads
+        # ``node.config.get("config", node.config)`` for exactly this reason,
+        # so accept both rather than silently validating nothing.
+        nested = node_data.get("config")
+        source = nested if isinstance(nested, dict) else node_data
+        operations = source.get("operations", [])
+        if not isinstance(operations, list):
+            continue
+        for index, operation in enumerate(operations):
+            if not isinstance(operation, dict):
+                continue
+            raw_expression = operation.get("expression", "")
+            target_variable = operation.get("target_variable", "")
+            has_expression = isinstance(raw_expression, str) and bool(
+                raw_expression.strip()
+            )
+            has_target = isinstance(target_variable, str) and bool(
+                target_variable.strip()
+            )
+            # The engine silently skips an operation missing either half
+            # (workflow_engine._execute_state_node), so downstream nodes read a
+            # variable that never gets set. Surface it at save time instead.
+            if has_expression and not has_target:
+                errors.append(
+                    f"Set State node '{snode_title}' operation {index + 1} "
+                    f"has an expression but no target variable"
+                )
+            if has_target and not has_expression:
+                errors.append(
+                    f"Set State node '{snode_title}' operation {index + 1} "
+                    f"has a target variable but no expression"
+                )
+            if has_expression:
+                try:
+                    validate_cel_expression(raw_expression)
+                except CelEvaluationError as exc:
+                    errors.append(
+                        f"Set State node '{snode_title}' operation {index + 1} "
+                        f"has an invalid expression: {exc}"
+                    )
+
     agent_nodes = [n for n in nodes if n.get("type") == "agent"]
     for agent_node in agent_nodes:
         agent_title = agent_node.get("title", agent_node.get("id", "unknown"))

application/core/model_utils.py

@@ -20,6 +20,52 @@ def get_api_key_for_provider(provider: str) -> Optional[str]:
     return settings.API_KEY
 
 
+def resolve_dispatch_provider(
+    stored_llm_name: Optional[str],
+    model_id: Optional[str] = None,
+    user_id: Optional[str] = None,
+    fallback: Optional[str] = None,
+) -> Optional[str]:
+    """Resolve a stored provider string to one ``LLMCreator`` can dispatch.
+
+    ``/api/models`` reports ``display_provider`` when a catalog YAML sets one
+    (``foundry``, ``azure_foundry``, ``cloudflare``, …), and clients persist
+    that label as ``llm_name`` on agents and workflow nodes. Those labels are
+    presentation-only: they are absent from ``PROVIDERS_BY_NAME``, so passing
+    one to ``LLMCreator.create_llm`` raises ``No LLM class found for type
+    <label>``. Normalizing here keeps already-stored records working without a
+    migration.
+
+    Resolution order: a stored name that is a real dispatch provider wins;
+    otherwise the model registry decides; otherwise ``fallback``.
+
+    Args:
+        stored_llm_name: Provider string persisted on the record, possibly a
+            display label.
+        model_id: Model whose registry entry knows the true provider.
+        user_id: BYOM-resolution scope for per-user model records.
+        fallback: Used when neither the stored name nor the registry resolves.
+
+    Returns:
+        A dispatchable provider name, or ``fallback`` when nothing resolves.
+    """
+    from application.llm.providers import PROVIDERS_BY_NAME
+
+    # Return the *canonical* lowercase name. ``LLMCreator`` lowercases before
+    # its own lookup, but ``get_api_key_for_provider`` matches exactly — so a
+    # stored "OpenAI" would pass this guard and then silently fall through to
+    # ``settings.API_KEY``, which is the key leak this function exists to stop.
+    if stored_llm_name and stored_llm_name.lower() in PROVIDERS_BY_NAME:
+        return stored_llm_name.lower()
+    if model_id:
+        resolved = get_provider_from_model_id(model_id, user_id=user_id)
+        if resolved:
+            return resolved
+    if fallback and fallback.lower() in PROVIDERS_BY_NAME:
+        return fallback.lower()
+    return fallback
+
+
 def get_all_available_models(
     user_id: Optional[str] = None,
 ) -> Dict[str, Dict[str, Any]]:

application/llm/openai.py

@@ -3,6 +3,7 @@ import hashlib
 import io
 import json
 import logging
+import os.path
 
 from openai import OpenAI
 
@@ -1654,16 +1655,49 @@ class OpenAILLM(BaseLLM):
                 logging.info(f"Attempting to upload PDF to OpenAI: {attachment.get('path', 'unknown')}")
                 try:
                     file_id = self._upload_file_to_openai(attachment)
+                    if not file_id:
+                        # Never emit ``{"file_id": None}``: the part carries no
+                        # document, providers 4xx or silently ignore it, and
+                        # downstream cleaning degrades it to a note claiming the
+                        # file "could not be processed" — while the parsed text
+                        # sits unused on the attachment row.
+                        raise ValueError(
+                            "no file_id returned for PDF upload "
+                            f"{attachment.get('path', 'unknown')!r}"
+                        )
+                    # ``file_id`` only. Sending ``filename`` alongside it is
+                    # rejected: 400 "Unknown parameter: …file.filename".
                     prepared_messages[user_message_index]["content"].append(
                         {"type": "file", "file": {"file_id": file_id}}
                     )
                 except Exception as e:
                     logging.error(f"Error uploading PDF to OpenAI: {e}", exc_info=True)
-                    if "content" in attachment:
+                    # Truthy, not membership — ``content`` is always a key on a
+                    # PG-backed attachment and is "" when extraction produced
+                    # nothing, which would otherwise send an empty "File
+                    # content:" block and read as a successfully-read document.
+                    if attachment.get("content"):
                         prepared_messages[user_message_index]["content"].append(
                             {
                                 "type": "text",
-                                "text": f"File content:\n\n{attachment['content']}",
+                                "text": (
+                                    "File content:\n\n"
+                                    f"{self._fit_attachment_text(attachment['content'])}"
+                                ),
+                            }
+                        )
+                    else:
+                        # Nothing to fall back on: say so, naming the user's own
+                        # file so the answer can't invent a generic excuse.
+                        filename = (
+                            attachment.get("filename")
+                            or os.path.basename(attachment.get("path") or "")
+                            or "the attached file"
+                        )
+                        prepared_messages[user_message_index]["content"].append(
+                            {
+                                "type": "text",
+                                "text": f"[File '{filename}' could not be processed]",
                             }
                         )
             else:
@@ -1689,6 +1723,78 @@ class OpenAILLM(BaseLLM):
         except FileNotFoundError:
             raise FileNotFoundError(f"File not found: {file_path}")
 
+    def _fit_attachment_text(self, content: str) -> str:
+        """Bound inlined attachment text to a share of the model's window.
+
+        ``_enforce_context_window`` runs in ``_llm_gen``, *before* attachments
+        are merged into the messages here, so nothing downstream trims this.
+        A 200-page PDF on an endpoint without a Files API would otherwise be
+        appended whole and get the whole request rejected for length — a worse
+        outcome than the degrade note it replaces.
+        """
+        from application.core.model_utils import get_token_limit
+        from application.utils import num_tokens_from_string
+
+        try:
+            limit = get_token_limit(self.model_id) if self.model_id else 0
+        except Exception:
+            limit = 0
+        # Half the window: the prompt, history and the answer share the rest.
+        budget = int(limit * 0.5) if limit else 24000
+        try:
+            # Tokenize once: this is a full BPE pass over the whole extraction
+            # (~12ms per 250k chars), on the hot path of every attachment turn.
+            token_count = num_tokens_from_string(content)
+            if token_count <= budget:
+                return content
+            chars_per_token = len(content) / max(token_count, 1)
+            keep = max(int(budget * chars_per_token * 0.95), 0)
+        except Exception:
+            keep = budget * 4
+        if keep <= 0 or keep >= len(content):
+            return content
+        return (
+            f"{content[:keep]}\n\n[Content truncated: the file is larger than "
+            "this model's context window.]"
+        )
+
+    def _endpoint_scope(self) -> str:
+        """Short fingerprint of ``(provider, base_url, api_key)``.
+
+        A Files-API ``file_id`` is only meaningful to the endpoint and
+        credential it was uploaded to — the same reasoning the Redis inline
+        cache already applies in ``_inline_file_id_cache_key``.
+        """
+        creds = "\0".join(
+            (
+                self.provider_name or "",
+                self._effective_base_url or "",
+                self.api_key or "",
+            )
+        )
+        return hashlib.sha256(creds.encode("utf-8")).hexdigest()[:16]
+
+    def _stamp_file_id(self, file_id: str) -> str:
+        """Tag a file_id with the endpoint it belongs to before persisting."""
+        return f"{self._endpoint_scope()}:{file_id}"
+
+    def _scoped_file_id(self, stored):
+        """Return a persisted file_id only if this endpoint can resolve it.
+
+        ``attachments.openai_file_id`` is a single global column, so without a
+        scope an id minted against one deployment would be replayed against
+        another, which answers ``No such File object`` on every retry —
+        permanently, since the row keeps the bad id. A miss re-uploads and
+        overwrites, so this self-heals both cross-endpoint reuse and ids that
+        aged out of provider retention. Legacy unscoped values are ignored.
+        """
+        if not stored or not isinstance(stored, str):
+            return None
+        scope, _, file_id = stored.partition(":")
+        if not file_id or scope != self._endpoint_scope():
+            return None
+        return file_id
+
     def _upload_file_to_openai(self, attachment):
         """
         Upload a file to OpenAI and return the file_id.
@@ -1702,8 +1808,13 @@ class OpenAILLM(BaseLLM):
         Returns:
             str: OpenAI file_id for the uploaded file.
         """
-        if "openai_file_id" in attachment:
-            return attachment["openai_file_id"]
+        # Truthy check, not membership: attachment dicts are built from
+        # ``SELECT *``, so ``openai_file_id`` is always a key and is NULL
+        # until an upload caches one. Testing membership treated every fresh
+        # attachment as a cache hit and returned None without uploading.
+        cached = self._scoped_file_id(attachment.get("openai_file_id"))
+        if cached:
+            return cached
         file_path = attachment.get("path")
 
         if not self.storage.file_exists(file_path):
@@ -1740,7 +1851,7 @@ class OpenAILLM(BaseLLM):
                         AttachmentsRepository(conn).update_any(
                             str(attachment_id),
                             user_id,
-                            {"openai_file_id": file_id},
+                            {"openai_file_id": self._stamp_file_id(file_id)},
                         )
                 except Exception as cache_err:
                     logging.warning(

application/logging.py

@@ -34,6 +34,12 @@ class LogContext:
         self.thought_length = 0
         self.source_count = 0
         self.tool_call_count = 0
+        # Terminal ``error`` events are *yielded*, not raised (workflow node
+        # failures, agent-reported errors), so they never reach the decorator's
+        # ``except``. Recording one here keeps ``activity_finished`` from
+        # reporting ``status="ok"`` on a turn the user saw fail — that gap is
+        # why blank-answer incidents did not show up in error dashboards.
+        self.stream_error: str | None = None
 
 
 def build_stack_data(
@@ -174,8 +180,12 @@ def _emit_activity_finished(
             "user_id": context.user,
             "endpoint": context.endpoint,
             "duration_ms": duration_ms,
-            "status": "error" if error is not None else "ok",
-            "error_class": type(error).__name__ if error is not None else None,
+            "status": "error" if (error is not None or context.stream_error) else "ok",
+            "error_class": (
+                type(error).__name__
+                if error is not None
+                else ("StreamError" if context.stream_error else None)
+            ),
             "answer_length": context.answer_length,
             "thought_length": context.thought_length,
             "source_count": context.source_count,
@@ -192,6 +202,11 @@ def _accumulate_response_summary(item: Any, context: "LogContext") -> None:
     """
     if not isinstance(item, dict):
         return
+    if item.get("type") == "error":
+        # Fall back to a sentinel: an error event carrying no message would
+        # otherwise store "" and read as falsy, reporting the activity "ok".
+        context.stream_error = str(item.get("error") or "")[:200] or "unspecified"
+        return
     if "answer" in item:
         context.answer_length += len(str(item["answer"]))
         return

application/storage/db/repositories/conversations.py

@@ -720,6 +720,11 @@ class ConversationsRepository:
             "model_id": message.get("model_id"),
             "message_metadata": message.get("metadata") or {},
         }
+        # Callers that know the turn failed (e.g. an agent that yielded a
+        # terminal error) must be able to say so; without this the column
+        # default silently made every appended row "complete".
+        if message.get("status") is not None:
+            values["status"] = message["status"]
         if message.get("timestamp") is not None:
             values["timestamp"] = message["timestamp"]
 
@@ -758,7 +763,7 @@ class ConversationsRepository:
         """
         allowed = {
             "prompt", "response", "thought", "sources", "tool_calls",
-            "attachments", "model_id", "metadata", "timestamp",
+            "attachments", "model_id", "metadata", "timestamp", "status",
             # Feedback can be re-set in rare continuation flows; without
             # it in the whitelist an upstream re-append that happens to
             # carry feedback would silently lose it. Mirrors

docs/content/Agents/nodes.mdx

@@ -5,9 +5,29 @@ DocsGPT workflows are composed of **Nodes** that are connected to form a process
 ## The Shared State
 
 Every workflow run maintains a state object (a JSON-like dictionary).
-- **Initial State**: Contains the user's input query (`{{query}}`) and chat history (`{{chat_history}}`).
-- **Accessing Variables**: You can access any variable in the state using the double-curly braces syntax: `{{variable_name}}`.
+- **Initial State**: Contains the user's input query (`query`) and chat history (`chat_history`).
 - **Modifying State**: Nodes read from this state and write their outputs back to it.
+- **Node Outputs**: Each node writes its result to `node_<node_id>_output`.
+
+### Two ways to reference state
+
+Which syntax a field takes depends on the field, so check this table before
+typing `{{ }}` anywhere:
+
+| Field | Node | Syntax |
+| --- | --- | --- |
+| Prompt Template | AI Agent | Template — `{{variable_name}}` |
+| Output Template | End | Template — `{{variable_name}}` |
+| Expression | Set State | [CEL](https://cel.dev/) — bare `variable_name` |
+| Expression | Condition | [CEL](https://cel.dev/) — bare `variable_name` |
+
+**Template fields** substitute `{{variable_name}}` into surrounding text, so
+`Analysis: {{analysis_result}}` produces a string.
+
+**Expression fields** are evaluated as [Common Expression
+Language](https://cel.dev/), where a bare name *is* the variable. Write
+`query`, not `{{query}}` — the braces are a syntax error there, and saving a
+workflow that contains one is rejected with the correction.
 
 ---
 
@@ -53,26 +73,65 @@ The **Set State Node** allows you to manipulate variables within the shared stat
 
 ### Operations
 
-You can define multiple operations in a single node. Each operation targets a specific **Key** (variable name).
+You can define multiple operations in a single node. Each operation has two
+parts:
 
-1.  **Set**: Assigns a specific value to a variable.
-    - *Value*: Can be a static string or a template using variables.
-    - *Example*: Set `current_step` to `1`.
-    - *Example*: Set `formatted_response` to `Analysis: {{analysis_result}}`.
+- **Target Variable**: the state key to write to.
+- **Expression**: a [CEL](https://cel.dev/) expression evaluated against the
+  current state. Its result becomes the variable's value.
 
-2.  **Increment**: Increases the value of a numeric variable.
-    - *Value*: The amount to add (default is 1).
-    - *Example*: Increment `retry_count` by `1`.
+Reference state variables by bare name. `{{ }}` is template syntax and does
+**not** work here — an operation must supply both parts, or it is rejected
+when you save.
 
-3.  **Append**: Adds a value to a list variable.
-    - *Value*: The item to add to the list.
-    - *Example*: Append `{{last_result}}` to `history_list`.
+| Goal | Target Variable | Expression |
+| --- | --- | --- |
+| Initialize a counter | `retry_count` | `0` |
+| Increment a counter | `retry_count` | `retry_count + 1` |
+| Copy a node's output to a stable name | `context` | `node_search_1_output` |
+| Build a string | `formatted_response` | `"Analysis: " + analysis_result` |
+| Append to a list | `history_list` | `history_list + [last_result]` |
+| Derive a boolean | `needs_review` | `size(context) < 100` |
+
+CEL supports arithmetic, string concatenation with `+`, comparisons,
+`&&`/`||`, ternaries (`cond ? a : b`), and built-ins such as `size()`,
+`startsWith()`, and `contains()`.
 
 ### Usage Examples
 
-- **Loop Counters**: Use a *Set State* node to initialize a counter (`i = 0`) before a loop, and another to increment it inside the loop.
-- **Accumulators**: Use *Append* to collect results from multiple parallel branches into a single list.
-- **Renaming**: Copy the output of a previous node to a more generic name (e.g., set `context` to `{{search_results}}`) so subsequent nodes can use a standard variable name.
+- **Loop Counters**: Initialize `retry_count` to `0` before a loop, then set it
+  to `retry_count + 1` inside the loop.
+- **Accumulators**: Collect results across branches with
+  `history_list + [last_result]`.
+- **Renaming**: Copy a previous node's output to a generic name (target
+  `context`, expression `node_search_1_output`) so later nodes can use one
+  standard variable.
+
+---
+
+## Condition Node
+
+The **Condition Node** branches the workflow. Each case pairs a
+[CEL](https://cel.dev/) **Expression** with an outgoing branch; the first case
+whose expression is true wins, and execution follows that branch.
+
+- Expressions use the same syntax as the Set State node: reference state by
+  bare name, not `{{ }}`.
+- Every condition node needs an **else** branch, which is taken when no case
+  matches.
+- Each case with an expression must have an outgoing edge, and every branch
+  must eventually reach an end node.
+
+| Goal | Expression |
+| --- | --- |
+| Route on a previous answer | `node_classify_1_output == "refund"` |
+| Guard on retrieved context | `size(context) > 0` |
+| Combine checks | `needs_review && retry_count < 3` |
+| Match text | `query.contains("invoice")` |
+
+A case whose expression fails to evaluate at run time is skipped and the next
+case is tried, so a workflow that always lands on **else** usually means an
+expression is referencing a variable that no earlier node writes.
 
 ---
 

frontend/src/agents/workflow/WorkflowBuilder.tsx

@@ -2473,11 +2473,13 @@ function WorkflowBuilderInner() {
                                         }}
                                         className="border-border focus-visible:ring-ring/50 focus-visible:border-ring bg-card dark:bg-accent mb-1 w-full rounded-xl border px-3 py-2 text-sm transition-all outline-none focus-visible:ring-2 dark:text-white"
                                         rows={2}
-                                        placeholder="input.foo + 1"
+                                        placeholder="query"
                                       />
                                       <p className="text-muted-foreground mb-3 text-xs">
                                         Use Common Expression Language to create
-                                        a custom expression.{' '}
+                                        a custom expression. Reference state by
+                                        bare name (<code>query</code>), not{' '}
+                                        <code>{'{{query}}'}</code>.{' '}
                                         <a
                                           href="https://cel.dev/"
                                           target="_blank"
@@ -2818,11 +2820,14 @@ function WorkflowBuilderInner() {
                                             }}
                                             className="border-border focus-visible:ring-ring/50 focus-visible:border-ring bg-card dark:bg-accent w-full rounded-xl border px-3 py-2 text-sm transition-all outline-none focus-visible:ring-2 dark:text-white"
                                             rows={2}
-                                            placeholder="Enter condition, e.g. input == 5"
+                                            placeholder='Enter condition, e.g. query.contains("refund")'
                                           />
                                           <p className="text-muted-foreground mt-1 text-xs">
                                             Use Common Expression Language to
-                                            create a custom expression.{' '}
+                                            create a custom expression.
+                                            Reference state by bare name (
+                                            <code>query</code>), not{' '}
+                                            <code>{'{{query}}'}</code>.{' '}
                                             <a
                                               href="https://cel.dev/"
                                               target="_blank"

tests/agents/test_cel_evaluator.py

@@ -8,6 +8,7 @@ from application.agents.workflows.cel_evaluator import (
     build_activation,
     cel_to_python,
     evaluate_cel,
+    validate_cel_expression,
 )
 import celpy.celtypes
 
@@ -167,3 +168,117 @@ class TestCelToPython:
     def test_unknown_type_passthrough(self):
         result = cel_to_python("raw_value")
         assert result == "raw_value"
+
+
+class TestCelErrorMessages:
+    """State/condition expressions are authored by users in the builder, so
+    their errors have to read like guidance, not like a parser dump."""
+
+    @pytest.mark.unit
+    def test_template_syntax_gets_a_targeted_hint(self):
+        """``{{query}}`` is valid in agent/end templates but not here.
+
+        The docs used to document ``{{variable}}`` for Set State nodes, so
+        users type it, get a bare caret dump, and have no way to learn that
+        this one field is CEL.
+        """
+        with pytest.raises(CelEvaluationError) as exc:
+            evaluate_cel("{{query}}", {"query": "hi"})
+        message = str(exc.value)
+        # The pre-fix message was "CEL error: {{query}}\n       ^\n", which
+        # already contained "CEL", "query" and "{{" — so assert on the
+        # guidance itself, not on incidental substrings.
+        assert "not {{ }} template syntax" in message
+        assert "Write query instead of {{query}}" in message
+
+    @pytest.mark.unit
+    def test_undeclared_reference_error_is_short(self):
+        """celpy embeds the whole activation — thousands of chars, and it
+        contains the user's own query text, which then lands in logs."""
+        with pytest.raises(CelEvaluationError) as exc:
+            evaluate_cel("customer_email", {"query": "a very secret question"})
+        message = str(exc.value)
+        assert len(message) <= 240, f"error is {len(message)} chars"
+        assert "customer_email" in message
+        assert "a very secret question" not in message
+        assert "NameContainer" not in message
+
+    @pytest.mark.unit
+    @pytest.mark.parametrize(
+        "secret",
+        [
+            "SUPER-SECRET-VALUE",
+            # An identifier-shaped value is indistinguishable from a variable
+            # name, so it must be redacted too — this is the realistic case
+            # (a one-word user query) and the earlier redaction let it through.
+            "invoice",
+            "SECRET",
+        ],
+    )
+    def test_error_does_not_echo_state_values(self, secret):
+        """Config errors skip sanitize_api_error now, so they must not carry
+        state contents — a shared agent's runner is not its owner."""
+        with pytest.raises(CelEvaluationError) as exc:
+            evaluate_cel("int(query)", {"query": secret})
+        assert secret not in str(exc.value)
+
+    @pytest.mark.unit
+    def test_error_keeps_the_exception_class(self):
+        """Redaction must not swallow the one safe diagnostic."""
+        with pytest.raises(CelEvaluationError) as exc:
+            evaluate_cel("int(query)", {"query": "nope"})
+        assert "ValueError" in str(exc.value)
+
+    @pytest.mark.unit
+    def test_long_error_is_truncated(self):
+        """Backstop for celpy messages that survive activation-stripping."""
+        from application.agents.workflows.cel_evaluator import _summarize_cel_error
+
+        summary = _summarize_cel_error(Exception("word " * 200))
+        assert len(summary) <= 200
+        assert summary.endswith("…")
+
+    @pytest.mark.unit
+    def test_validate_rejects_empty_expression(self):
+        with pytest.raises(CelEvaluationError, match="Empty expression"):
+            validate_cel_expression("   ")
+
+    @pytest.mark.unit
+    @pytest.mark.parametrize(
+        "expression",
+        [
+            'query == "{{y}}"',  # braces inside a string literal
+            '{{"a": 1}: 2}',     # map literal used as a map key
+        ],
+    )
+    def test_valid_cel_containing_braces_is_not_rejected(self, expression):
+        """The hint must never fire on expressions that actually compile."""
+        validate_cel_expression(expression)
+
+    @pytest.mark.unit
+    def test_valid_expression_still_evaluates(self):
+        assert evaluate_cel('query + "!"', {"query": "hi"}) == "hi!"
+
+
+class TestValidateCelExpression:
+    """Save-time gate: a workflow that cannot run should not save clean."""
+
+    @pytest.mark.unit
+    def test_accepts_valid_expression(self):
+        validate_cel_expression("query + \"!\"")
+
+    @pytest.mark.unit
+    def test_rejects_template_syntax(self):
+        with pytest.raises(CelEvaluationError, match="CEL"):
+            validate_cel_expression("{{query}}")
+
+    @pytest.mark.unit
+    def test_rejects_syntax_error(self):
+        with pytest.raises(CelEvaluationError):
+            validate_cel_expression("query +")
+
+    @pytest.mark.unit
+    def test_accepts_reference_unknown_at_save_time(self):
+        """Only syntax is knowable when saving — state is built at runtime,
+        so an unresolved name must not block saving a valid workflow."""
+        validate_cel_expression("node_abc_output + customer_email")

tests/agents/test_workflow_engine.py

@@ -146,6 +146,96 @@ def test_execute_agent_node_falls_back_to_text_when_schema_not_configured(monkey
     assert engine.state["result"] == "plain text answer"
 
 
+def _state_workflow(operations, nested=True):
+    """Start → state → end, with state config in either accepted shape."""
+    data = {"config": {"operations": operations}} if nested else {
+        "operations": operations
+    }
+    nodes = [
+        {"id": "start", "type": "start", "title": "Start", "data": {}},
+        {"id": "state", "type": "state", "title": "Build reply", "data": data},
+        {"id": "end", "type": "end", "title": "End", "data": {}},
+    ]
+    edges = [
+        {"id": "edge_1", "source": "start", "target": "state"},
+        {"id": "edge_2", "source": "state", "target": "end"},
+    ]
+    return nodes, edges
+
+
+@pytest.mark.parametrize("nested", [True, False])
+def test_validate_workflow_structure_rejects_template_syntax_in_state_node(nested):
+    """``{{query}}`` compiles nowhere but used to save and publish clean.
+
+    It then aborted the run on the first message with a bare caret dump that
+    ``sanitize_api_error`` collapsed into "try again later" — the 2026-08-01
+    report, where a new user retried eight times over seven hours.
+    """
+    nodes, edges = _state_workflow(
+        [{"expression": "{{query}}", "target_variable": "reply"}], nested=nested
+    )
+
+    errors = validate_workflow_structure(nodes, edges)
+
+    assert any(
+        "Set State node 'Build reply'" in err and "invalid expression" in err
+        for err in errors
+    ), errors
+    # The message must teach the correction, not just report a failure.
+    assert any("CEL" in err for err in errors), errors
+
+
+def test_validate_workflow_structure_accepts_valid_state_expression():
+    nodes, edges = _state_workflow(
+        [{"expression": 'query + "!"', "target_variable": "reply"}]
+    )
+    assert validate_workflow_structure(nodes, edges) == []
+
+
+def test_validate_workflow_structure_accepts_runtime_only_state_reference():
+    """Names resolve from run state, so they cannot be checked at save time."""
+    nodes, edges = _state_workflow(
+        [{"expression": "node_agent_1_output", "target_variable": "reply"}]
+    )
+    assert validate_workflow_structure(nodes, edges) == []
+
+
+def test_validate_workflow_structure_rejects_half_configured_state_operation():
+    """The engine skips these silently, so downstream reads an unset var."""
+    nodes, edges = _state_workflow(
+        [{"expression": "query", "target_variable": ""}]
+    )
+    errors = validate_workflow_structure(nodes, edges)
+    assert any("no target variable" in err for err in errors), errors
+
+
+def test_validate_workflow_structure_rejects_invalid_condition_expression():
+    nodes = [
+        {"id": "start", "type": "start", "title": "Start", "data": {}},
+        {
+            "id": "cond",
+            "type": "condition",
+            "title": "Route",
+            "data": {
+                "cases": [
+                    {"expression": "{{query}}", "sourceHandle": "case_0"},
+                    {"expression": "true", "sourceHandle": "else"},
+                ]
+            },
+        },
+        {"id": "end", "type": "end", "title": "End", "data": {}},
+    ]
+    edges = [
+        {"id": "e1", "source": "start", "target": "cond"},
+        {"id": "e2", "source": "cond", "target": "end", "sourceHandle": "case_0"},
+        {"id": "e3", "source": "cond", "target": "end", "sourceHandle": "else"},
+    ]
+
+    errors = validate_workflow_structure(nodes, edges)
+
+    assert any("invalid expression" in err for err in errors), errors
+
+
 def test_validate_workflow_structure_rejects_invalid_agent_json_schema():
     nodes = [
         {"id": "start", "type": "start", "title": "Start", "data": {}},
@@ -507,3 +597,96 @@ class TestWorkflowEngineAdditionalCoverage:
         # A non-dict schema triggers JsonSchemaValidationError
         with pytest.raises(ValueError, match="Invalid JSON schema"):
             engine._normalize_node_json_schema("not_a_dict", "TestNode")
+
+
+class TestAgentNodeProviderResolution:
+    """``llm_name`` stored on a node is a *display* label, not a dispatch name.
+
+    ``/api/models`` reports ``display_provider`` (e.g. ``foundry``,
+    ``azure_foundry``, ``cloudflare``) and the builder stores that string on
+    the node. Handing it to ``LLMCreator`` raises ``No LLM class found for
+    type <label>``, which fails the node before any LLM call and returns a
+    blank answer to the user. The engine must resolve the real dispatch
+    provider from the model registry instead.
+    """
+
+    @staticmethod
+    def _run(monkeypatch, node, *, registry_provider="openai_compatible"):
+        """Execute one agent node, returning the kwargs the factory saw."""
+        engine = create_engine()
+        engine.state["query"] = "test"
+        captured = {}
+
+        def _capture(**kwargs):
+            captured.update(kwargs)
+            return StubNodeAgent([{"answer": "ok"}])
+
+        monkeypatch.setattr(
+            WorkflowNodeAgentFactory, "create", staticmethod(_capture)
+        )
+        monkeypatch.setattr(
+            "application.core.model_utils.get_api_key_for_provider",
+            lambda name: f"key-for-{name}",
+        )
+        monkeypatch.setattr(
+            "application.core.model_utils.get_provider_from_model_id",
+            lambda _, **_kwargs: registry_provider,
+        )
+        monkeypatch.setattr(
+            "application.core.model_utils.get_model_capabilities",
+            lambda _, **_kwargs: None,
+        )
+        list(engine._execute_agent_node(node))
+        return captured
+
+    @pytest.mark.parametrize(
+        "display_label", ["azure_foundry", "cloudflare", "foundry"]
+    )
+    def test_display_provider_label_resolves_to_dispatch_provider(
+        self, monkeypatch, display_label
+    ):
+        """A display label must not reach LLMCreator (err#33)."""
+        node = create_agent_node(node_id="n1")
+        node.config["model_id"] = "Kimi-K2.6"
+        node.config["llm_name"] = display_label
+
+        captured = self._run(monkeypatch, node)
+
+        assert captured["llm_name"] == "openai_compatible"
+        # The api_key must follow the *normalized* name: resolving against the
+        # display label silently falls through to settings.API_KEY.
+        assert captured["api_key"] == "key-for-openai_compatible"
+
+    def test_real_provider_name_is_preserved(self, monkeypatch):
+        """A node storing a genuine dispatch name keeps it."""
+        node = create_agent_node(node_id="n2")
+        node.config["model_id"] = "gpt-4o"
+        node.config["llm_name"] = "openai"
+
+        captured = self._run(monkeypatch, node, registry_provider="openai")
+
+        assert captured["llm_name"] == "openai"
+
+    def test_unresolvable_label_falls_back_to_parent_agent(self, monkeypatch):
+        """No registry hit: inherit the parent agent rather than dispatching junk."""
+        node = create_agent_node(node_id="n3")
+        node.config["model_id"] = "mystery-model"
+        node.config["llm_name"] = "some_unknown_label"
+
+        captured = self._run(monkeypatch, node, registry_provider=None)
+
+        # create_engine()'s parent agent is llm_name="openai"
+        assert captured["llm_name"] == "openai"
+
+    def test_retriever_config_uses_normalized_provider(self, monkeypatch):
+        """The agentic retriever path builds its own kwargs — normalize there too."""
+        node = create_agent_node(node_id="n4")
+        node.config["model_id"] = "Kimi-K2.6"
+        node.config["llm_name"] = "azure_foundry"
+        node.config["agent_type"] = "agentic"
+        node.config["sources"] = ["src-1"]
+
+        captured = self._run(monkeypatch, node)
+
+        assert captured["retriever_config"]["llm_name"] == "openai_compatible"
+        assert captured["retriever_config"]["api_key"] == "key-for-openai_compatible"

tests/api/answer/test_base_routes.py

@@ -909,6 +909,15 @@ def _patch_db_session(conn):
         # uncommitted writes from this transaction.
         "application.api.answer.routes.base.db_readonly",
         _yield,
+    ), patch(
+        # The terminal ``stream_answer`` user_logs write opens its own
+        # ``db_session``. Left unpatched it is a *second* connection that
+        # blocks on the uncommitted ``users`` row this transaction just
+        # inserted (via the ``ensure_user_exists`` trigger) until the
+        # statement timeout fires — ~30s per test, swallowed by the
+        # caller's except, so it only ever showed up as slowness.
+        "application.api.answer.routes.base.db_session",
+        _yield,
     ):
         yield
 
@@ -972,6 +981,161 @@ class TestCompleteStreamWalAcceptance:
             assert "RuntimeError" in msgs[0]["metadata"]["error"]
             assert "LLM upstream failed" in msgs[0]["metadata"]["error"]
 
+    def test_workflow_node_error_persists_as_failed_not_blank_complete(
+        self, pg_conn, flask_app,
+    ):
+        """A workflow node failure must not land as a blank ``complete`` row.
+
+        The engine reports node failures by *yielding* ``{"type": "error"}``
+        rather than raising, so the generator returns normally and the turn
+        used to be finalized ``complete`` with an empty response. Live, the
+        client shows an error bubble; on reload the row mapped to an empty
+        answer with no error and no retry affordance — the user saw a blank
+        message and re-sent the prompt. Reproduces the 2026-08-01 report.
+        """
+        from application.api.answer.routes.base import BaseAnswerResource
+        from application.storage.db.repositories.conversations import (
+            ConversationsRepository,
+        )
+
+        with flask_app.app_context():
+            resource = BaseAnswerResource()
+
+            mock_agent = MagicMock()
+            mock_agent.gen.return_value = iter(
+                [{"type": "error", "error": "No LLM class found for type foundry"}]
+            )
+
+            with _patch_db_session(pg_conn):
+                stream = list(
+                    resource.complete_stream(
+                        question="hello",
+                        agent=mock_agent,
+                        conversation_id=None,
+                        user_api_key=None,
+                        decoded_token={"sub": "u-wf-error"},
+                        should_persist=True,
+                        model_id="gpt-4",
+                    )
+                )
+            assert len([s for s in stream if '"type": "error"' in s]) == 1
+
+            from sqlalchemy import text as sql_text
+            convs = pg_conn.execute(
+                sql_text("SELECT id FROM conversations WHERE user_id = :u"),
+                {"u": "u-wf-error"},
+            ).fetchall()
+            assert len(convs) == 1
+            msgs = ConversationsRepository(pg_conn).get_messages(str(convs[0][0]))
+            assert len(msgs) == 1
+            assert msgs[0]["prompt"] == "hello"
+            assert msgs[0]["status"] == "failed", (
+                "a turn that produced no answer and emitted an error must be "
+                "failed, so history renders the error and a retry button"
+            )
+            assert "foundry" in msgs[0]["metadata"]["error"]
+
+    def test_error_after_partial_answer_keeps_the_answer(
+        self, pg_conn, flask_app,
+    ):
+        """A late error must not discard text the user already received.
+
+        Only the *blank* turn is a failure. If a workflow produced output and
+        then a downstream node failed, the row stays ``complete`` so the
+        partial answer still renders; the error was already surfaced live.
+        """
+        from application.api.answer.routes.base import BaseAnswerResource
+        from application.storage.db.repositories.conversations import (
+            ConversationsRepository,
+        )
+
+        with flask_app.app_context():
+            resource = BaseAnswerResource()
+
+            mock_agent = MagicMock()
+            mock_agent.gen.return_value = iter(
+                [
+                    {"answer": "partial result"},
+                    {"type": "error", "error": "node 2 blew up"},
+                ]
+            )
+
+            with _patch_db_session(pg_conn):
+                list(
+                    resource.complete_stream(
+                        question="run the workflow",
+                        agent=mock_agent,
+                        conversation_id=None,
+                        user_api_key=None,
+                        decoded_token={"sub": "u-wf-partial"},
+                        should_persist=True,
+                        model_id="gpt-4",
+                    )
+                )
+
+            from sqlalchemy import text as sql_text
+            convs = pg_conn.execute(
+                sql_text("SELECT id FROM conversations WHERE user_id = :u"),
+                {"u": "u-wf-partial"},
+            ).fetchall()
+            msgs = ConversationsRepository(pg_conn).get_messages(str(convs[0][0]))
+            assert msgs[0]["status"] == "complete"
+            assert msgs[0]["response"] == "partial result"
+            # Still recorded, so the failure is greppable in review queries.
+            assert "node 2 blew up" in msgs[0]["metadata"]["error"]
+
+    def test_workflow_error_fails_the_row_on_the_non_wal_path_too(
+        self, pg_conn, flask_app,
+    ):
+        """The same guarantee when no placeholder row was reserved.
+
+        ``save_conversation`` has its own insert path (used when the WAL
+        reservation failed, or on a continuation carrying no
+        ``reserved_message_id``). It took no status, so the row landed on the
+        column default ``complete`` — leaving exactly the blank bubble this
+        changeset removes on the other branch.
+        """
+        from application.api.answer.routes.base import BaseAnswerResource
+        from application.storage.db.repositories.conversations import (
+            ConversationsRepository,
+        )
+
+        with flask_app.app_context():
+            resource = BaseAnswerResource()
+
+            mock_agent = MagicMock()
+            mock_agent.gen.return_value = iter(
+                [{"type": "error", "error": "CEL error in node 'Build reply'"}]
+            )
+
+            with _patch_db_session(pg_conn), patch.object(
+                resource.conversation_service,
+                "save_user_question",
+                side_effect=RuntimeError("WAL reservation unavailable"),
+            ):
+                list(
+                    resource.complete_stream(
+                        question="hello",
+                        agent=mock_agent,
+                        conversation_id=None,
+                        user_api_key=None,
+                        decoded_token={"sub": "u-wf-nonwal"},
+                        should_persist=True,
+                        model_id="gpt-4",
+                    )
+                )
+
+            from sqlalchemy import text as sql_text
+            convs = pg_conn.execute(
+                sql_text("SELECT id FROM conversations WHERE user_id = :u"),
+                {"u": "u-wf-nonwal"},
+            ).fetchall()
+            assert len(convs) == 1
+            msgs = ConversationsRepository(pg_conn).get_messages(str(convs[0][0]))
+            assert len(msgs) == 1
+            assert msgs[0]["status"] == "failed"
+            assert "CEL error" in msgs[0]["metadata"]["error"]
+
     def test_tool_approval_event_only_fires_when_state_saved(
         self, pg_conn, flask_app,
     ):

tests/llm/test_openai.py

@@ -89,7 +89,11 @@ class FakeChatCompletions:
 
 
 class FakeFiles:
+    def __init__(self):
+        self.created = []
+
     def create(self, file=None, purpose=None):
+        self.created.append(purpose)
         return types.SimpleNamespace(id="file_id_uploaded")
 
 
@@ -729,6 +733,149 @@ class TestPrepareMessagesWithAttachments:
         assert isinstance(user_msg["content"], list)
 
 
+@pytest.mark.unit
+class TestPdfAttachmentNullFileIdCache:
+    """A NULL ``openai_file_id`` column must not read as a cache hit.
+
+    Attachment dicts come from ``SELECT *`` (``row_to_dict``), so every row
+    carries an ``openai_file_id`` key and it is NULL until an upload caches
+    one. Testing for key *membership* therefore treated every fresh
+    attachment as cached, returned ``None`` without uploading, and — because
+    nothing raised — skipped the fallback that inlines the extracted text.
+    The user was told their perfectly good PDF "could not be processed".
+    """
+
+    @pytest.fixture(autouse=True)
+    def _real_pdf_on_disk(self, tmp_path):
+        """The upload path opens the file, so it has to exist."""
+        pdf = tmp_path / "contract.pdf"
+        pdf.write_bytes(b"%PDF-1.4\ntrailer<</Root 1 0 R>>\n")
+        self._pdf_path = str(pdf)
+
+    def _pg_shaped_attachment(self, **overrides):
+        """What ``_attachment_to_dict`` actually produces for a fresh row."""
+        attachment = {
+            "id": "att-1",
+            "filename": "contract.pdf",
+            "path": self._pdf_path,
+            "upload_path": self._pdf_path,
+            "mime_type": "application/pdf",
+            "size": 120_000,
+            "content": "REAL EXTRACTED TEXT",
+            "token_count": 6748,
+            "openai_file_id": None,
+            "google_file_uri": None,
+        }
+        attachment.update(overrides)
+        return attachment
+
+    def test_null_cached_id_still_uploads(self, llm):
+        file_id = llm._upload_file_to_openai(self._pg_shaped_attachment())
+        assert file_id, "a NULL cached id must not short-circuit the upload"
+        assert llm.client.files.created, "files.create was never called"
+
+    def test_populated_cached_id_short_circuits(self, llm):
+        stamped = llm._stamp_file_id("file-cached")
+        attachment = self._pg_shaped_attachment(openai_file_id=stamped)
+        assert llm._upload_file_to_openai(attachment) == "file-cached"
+        assert not llm.client.files.created, "should not re-upload a cached file"
+
+    def test_cached_id_from_another_endpoint_is_ignored(self, llm):
+        """A file_id only resolves at the endpoint that minted it.
+
+        ``attachments.openai_file_id`` is one global column, so an id cached
+        against deployment A would otherwise be replayed against deployment B,
+        which answers "No such File object" on every retry — permanently,
+        since the row keeps the bad id.
+        """
+        attachment = self._pg_shaped_attachment(
+            openai_file_id="0123456789abcdef:file-from-elsewhere"
+        )
+        file_id = llm._upload_file_to_openai(attachment)
+        assert file_id == "file_id_uploaded"
+        assert llm.client.files.created, "a foreign-scope id must re-upload"
+
+    def test_legacy_unscoped_cached_id_is_ignored(self, llm):
+        """Rows written before scoping carry a bare id; re-upload and restamp."""
+        attachment = self._pg_shaped_attachment(openai_file_id="file-legacy")
+        assert llm._upload_file_to_openai(attachment) == "file_id_uploaded"
+        assert llm.client.files.created
+
+    def test_oversized_content_is_truncated_before_inlining(self, llm):
+        """Attachments merge in *after* the context-window check, so nothing
+        downstream trims this — a big PDF would blow the whole request."""
+        llm._upload_file_to_openai = lambda _attachment: None
+        llm.model_id = None  # exercise the no-registry default budget
+        huge = "word " * 200_000
+        attachment = self._pg_shaped_attachment(content=huge)
+
+        result = llm.prepare_messages_with_attachments(
+            [{"role": "user", "content": "summarize"}], [attachment]
+        )
+        text = next(
+            p["text"]
+            for p in result[-1]["content"]
+            if p.get("type") == "text" and p["text"].startswith("File content:")
+        )
+        assert len(text) < len(huge)
+        assert "Content truncated" in text
+
+    def test_extracted_text_reaches_the_model_when_upload_yields_no_id(self, llm):
+        """The production failure: no id, no exception, no text, wrong answer."""
+        llm._upload_file_to_openai = lambda _attachment: None
+
+        msgs = [{"role": "user", "content": "summarize the attached pdf"}]
+        result = llm.prepare_messages_with_attachments(
+            msgs, [self._pg_shaped_attachment()]
+        )
+
+        user_msg = next(m for m in result if m["role"] == "user")
+        assert not any(
+            p.get("type") == "file" and not p.get("file", {}).get("file_id")
+            for p in user_msg["content"]
+        ), "must not emit a file part with an empty file_id"
+        assert any(
+            "REAL EXTRACTED TEXT" in p.get("text", "")
+            for p in user_msg["content"]
+            if p.get("type") == "text"
+        ), "extracted content must be inlined rather than dropped"
+
+    def test_degrade_note_names_the_real_file(self, llm):
+        """Fall back to the note only with no text, and never say 'upload.pdf'."""
+        llm._upload_file_to_openai = lambda _attachment: None
+        attachment = self._pg_shaped_attachment(content="")
+
+        msgs = [{"role": "user", "content": "summarize"}]
+        result = llm.prepare_messages_with_attachments(msgs, [attachment])
+
+        user_msg = next(m for m in result if m["role"] == "user")
+        notes = [
+            p["text"]
+            for p in user_msg["content"]
+            if p.get("type") == "text" and "could not be processed" in p.get("text", "")
+        ]
+        assert len(notes) == 1
+        assert "contract.pdf" in notes[0]
+        assert "upload.pdf" not in notes[0]
+
+    def test_file_part_carries_only_file_id(self, llm):
+        """Never send ``filename`` next to ``file_id``.
+
+        Verified against the Azure Foundry deployment: a file part carrying
+        both is rejected outright —
+        ``400 Unknown parameter: 'messages[0].content[1].file.filename'``.
+        The upload itself and a ``file_id``-only part both return 200 and the
+        model reads the PDF correctly, so the reference is all we may send.
+        """
+        msgs = [{"role": "user", "content": "summarize"}]
+        result = llm.prepare_messages_with_attachments(
+            msgs, [self._pg_shaped_attachment()]
+        )
+        user_msg = next(m for m in result if m["role"] == "user")
+        file_part = next(p for p in user_msg["content"] if p.get("type") == "file")
+        assert file_part["file"] == {"file_id": "file_id_uploaded"}
+
+
 # _get_base64_image
 
 
@@ -1048,8 +1195,15 @@ class TestPrepareMessagesWithAttachmentsAdditional:
 class TestUploadFileToOpenai:
 
     def test_cached_file_id_returned(self, llm):
-        """Cover line 469: cached openai_file_id."""
-        result = llm._upload_file_to_openai({"openai_file_id": "cached_id"})
+        """A cached openai_file_id short-circuits the upload.
+
+        The stored value is endpoint-scoped: a bare id is a legacy row and is
+        deliberately treated as a miss (see
+        ``TestPdfAttachmentNullFileIdCache``).
+        """
+        result = llm._upload_file_to_openai(
+            {"openai_file_id": llm._stamp_file_id("cached_id")}
+        )
         assert result == "cached_id"
 
     def test_file_not_found_raises(self, llm):
@@ -1156,8 +1310,10 @@ class TestOpenAILLMConstructor:
 class TestUploadFileToOpenai2:
 
     def test_returns_cached_file_id(self, llm):
-        """Cover line 491-492: returns cached openai_file_id."""
-        result = llm._upload_file_to_openai({"openai_file_id": "file-123"})
+        """Returns the cached openai_file_id when it belongs to this endpoint."""
+        result = llm._upload_file_to_openai(
+            {"openai_file_id": llm._stamp_file_id("file-123")}
+        )
         assert result == "file-123"
 
     def test_file_not_found_raises(self, llm):
@@ -1386,7 +1542,9 @@ class TestUploadFileToOpenaiLine469:
     """Cover line 469: cached openai_file_id returned early."""
 
     def test_cached_id_returned_immediately(self, llm):
-        result = llm._upload_file_to_openai({"openai_file_id": "file-cached-123"})
+        result = llm._upload_file_to_openai(
+            {"openai_file_id": llm._stamp_file_id("file-cached-123")}
+        )
         assert result == "file-cached-123"
 
 
@@ -1529,9 +1687,9 @@ class TestUploadFileToOpenAIError:
             llm._upload_file_to_openai({"path": "/doc.pdf"})
 
     def test_upload_cached_file_id(self, llm):
-        """Cover line 491-492: already has openai_file_id."""
+        """Already has an endpoint-scoped openai_file_id."""
         result = llm._upload_file_to_openai(
-            {"path": "/doc.pdf", "openai_file_id": "file-cached"}
+            {"path": "/doc.pdf", "openai_file_id": llm._stamp_file_id("file-cached")}
         )
         assert result == "file-cached"
 

tests/test_logging.py

@@ -257,6 +257,35 @@ class TestLogActivity:
         assert finished.status == "error"
         assert finished.error_class == "ValueError"
 
+    def test_log_activity_records_error_status_on_yielded_error(self, caplog):
+        # Workflow node failures are reported as a yielded ``type: error``
+        # event rather than a raised exception, so the decorator's ``except``
+        # never runs. Those turns used to log ``status="ok"`` with
+        # ``answer_length=0``, which kept real user-visible failures out of
+        # every error dashboard.
+        import logging as _logging
+
+        from application.logging import log_activity
+
+        class FakeAgent:
+            endpoint = "stream"
+            user = "user1"
+            user_api_key = ""
+            query = "hello"
+
+        @log_activity()
+        def erroring(agent, log_context=None):
+            yield {"type": "error", "error": "No LLM class found for type foundry"}
+
+        with patch("application.logging._log_activity_to_db"), \
+                caplog.at_level(_logging.INFO, logger="root"):
+            list(erroring(FakeAgent()))
+
+        finished = next(r for r in caplog.records if r.message == "activity_finished")
+        assert finished.status == "error"
+        assert finished.error_class == "StreamError"
+        assert finished.answer_length == 0
+
     def test_log_activity_emits_response_summary_aggregates(self, caplog):
         # Replaces the ``agent_response`` event that ``run_agent_logic``
         # used to emit only on the Celery webhook path: every Flask