From f5129a56567ad69c60215e02fb925281473f9548 Mon Sep 17 00:00:00 2001
From: Alex <a@tushynski.me>
Date: Wed, 5 Aug 2026 11:24:59 +0100
Subject: [PATCH 1/5] fix(workflows): dispatch agent nodes by provider, not
 display label
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

/api/models reports `display_provider` when a catalog YAML sets one
(`foundry`, `azure_foundry`, `cloudflare`), and the builder persists that
string as a node's `llm_name`. The engine passed it straight to
LLMCreator, which only knows the names in PROVIDERS_BY_NAME and raises
`No LLM class found for type <label>`.

That fails the node before any LLM call, so the turn ends in ~60ms with
an empty answer and no tokens generated. It hits the *default* path: the
platform-default model's label is stamped into every newly dragged agent
node and validateWorkflow requires an agent node, so a new user's
untouched workflow could not produce a token regardless of what they
typed. Ordinary chat was unaffected because it resolves the provider from
the model registry and never reads the stored name.

Adds `resolve_dispatch_provider`, which prefers a stored name that is a
real dispatch provider, then the registry lookup, then the parent agent.
Nodes already saved with a label are repaired at run time, so no
migration is needed. The api_key now resolves from the normalized name
too — `get_api_key_for_provider` falls back to settings.API_KEY for names
it does not recognize, which would have sent the deployment key to
whatever endpoint the label happened to select.
---
 .../agents/workflows/workflow_engine.py       | 23 +++--
 application/core/model_utils.py               | 46 +++++++++
 tests/agents/test_workflow_engine.py          | 93 +++++++++++++++++++
 3 files changed, 155 insertions(+), 7 deletions(-)

diff --git a/application/agents/workflows/workflow_engine.py b/application/agents/workflows/workflow_engine.py
index e08cf444..0c4f39a3 100644
--- a/application/agents/workflows/workflow_engine.py
+++ b/application/agents/workflows/workflow_engine.py
@@ -291,7 +291,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 +314,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;
diff --git a/application/core/model_utils.py b/application/core/model_utils.py
index fbff0409..d6ec8f6c 100644
--- a/application/core/model_utils.py
+++ b/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]]:
diff --git a/tests/agents/test_workflow_engine.py b/tests/agents/test_workflow_engine.py
index e8736b07..6f375f8a 100644
--- a/tests/agents/test_workflow_engine.py
+++ b/tests/agents/test_workflow_engine.py
@@ -507,3 +507,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"

From bda11242e1dfeeca4ef4d8a4e73b5ccd153407dc Mon Sep 17 00:00:00 2001
From: Alex <a@tushynski.me>
Date: Wed, 5 Aug 2026 11:25:11 +0100
Subject: [PATCH 2/5] fix(stream): persist a turn that errored without an
 answer as failed
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

WorkflowEngine reports node failures by *yielding* `{"type": "error"}`
rather than raising, so complete_stream's generator returns normally and
the except handler never runs. The turn was finalized `status="complete"`
with an empty response.

Live, the client renders an error bubble with a Retry button. On reload
it does not: mapServerQueryToClient only surfaces `metadata.error` for
`failed` rows, so history showed a blank message with no error and no way
to retry. A user hitting this re-sends the same prompt into new
conversations, which is exactly what the 2026-08-01 report shows — nine
blank first messages in seven hours.

Tracks a `stream_error` flag alongside the existing `paused` machinery
and finalizes `failed` when the turn produced no answer, recording the
user-facing message in `metadata.error`. An error arriving *after* output
keeps `complete` so partial text is not discarded; structured answers
count as output too, since they live in `structured_chunks` rather than
`response_full`. The flag is recorded before the pause branches so those
paths cannot lose it.

save_conversation grew a `status` parameter (default `complete`) for the
non-WAL branch, which took no status and so landed on the column default
— the same blank-complete row on a path the WAL fix did not cover.
Title generation now also runs for failed turns: _maybe_generate_title
only regenerates while the name is still the question-prefix fallback, so
skipping it would strand a conversation whose first turn failed with the
raw prompt as its name forever.

logging.py counts a yielded error toward `activity_finished.status`.
These failures previously logged `status=ok` with `answer_length=0`,
which is why user-visible blank answers never appeared in error metrics.
---
 application/api/answer/routes/base.py         |  33 +++-
 .../answer/services/conversation_service.py   |  14 +-
 application/logging.py                        |  19 +-
 .../storage/db/repositories/conversations.py  |   7 +-
 tests/api/answer/test_base_routes.py          | 164 ++++++++++++++++++
 tests/test_logging.py                         |  29 ++++
 6 files changed, 261 insertions(+), 5 deletions(-)

diff --git a/application/api/answer/routes/base.py b/application/api/answer/routes/base.py
index 380b6f13..df68bae1 100644
--- a/application/api/answer/routes/base.py
+++ b/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)
diff --git a/application/api/answer/services/conversation_service.py b/application/api/answer/services/conversation_service.py
index 97f6d54b..d89b503a 100644
--- a/application/api/answer/services/conversation_service.py
+++ b/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,
diff --git a/application/logging.py b/application/logging.py
index 5c03f2a2..a244dde1 100644
--- a/application/logging.py
+++ b/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
diff --git a/application/storage/db/repositories/conversations.py b/application/storage/db/repositories/conversations.py
index 97f8c56f..a9e46825 100644
--- a/application/storage/db/repositories/conversations.py
+++ b/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
diff --git a/tests/api/answer/test_base_routes.py b/tests/api/answer/test_base_routes.py
index 7b6fdaaa..dac6b2c4 100644
--- a/tests/api/answer/test_base_routes.py
+++ b/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,
     ):
diff --git a/tests/test_logging.py b/tests/test_logging.py
index 8bad293b..d10ba27d 100644
--- a/tests/test_logging.py
+++ b/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

From a51dd10cc7b67f8381a4ad58d9940709bc2db1ef Mon Sep 17 00:00:00 2001
From: Alex <a@tushynski.me>
Date: Wed, 5 Aug 2026 11:25:28 +0100
Subject: [PATCH 3/5] fix(openai): stop telling users their readable PDF could
 not be processed
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

_upload_file_to_openai short-circuited on `"openai_file_id" in attachment`.
Attachment dicts are built by row_to_dict from `SELECT *`, so that key is
always present and is NULL until an upload caches one — every fresh
attachment read as a cache hit. The method returned None without
uploading and, critically, without raising, so the except branch that
inlines the extracted text never ran. The part went out as
`{"file_id": None}`, which _resolve_file_part degraded to
`[File 'upload.pdf' could not be processed]`.

The model then relayed that to the user while thousands of tokens of
extracted text sat unused on the attachment row. Reproduced live against
the Foundry deployment: the model answers "I couldn't access the attached
PDF because it could not be processed; please re-upload it or paste the
text". The same bug was fixed in google_ai.py by 8e9f661e ("Truthy check,
not membership") which never touched this file; d29a7aaf later turned the
resulting hard 400 into this silent wrong answer.

Truthy check restored, a falsy file_id now raises so the fallback runs,
and the fallback tests `content` truthily — it too was a membership test,
so an empty extraction sent a bare "File content:" header that reads as a
successfully-read document. With no content at all we emit a note naming
the user's own file instead of the `upload.pdf` placeholder.

The file part deliberately carries `file_id` only: sending `filename`
alongside it is rejected with
`400 Unknown parameter: 'messages[0].content[1].file.filename'`.

Two consequences of making this path live for the first time are handled
here. `attachments.openai_file_id` is a single global column, so ids are
now stamped with an endpoint fingerprint — an id minted against one
deployment replayed against another answers "No such File object" on
every retry, permanently, since the row keeps the bad id. A foreign or
legacy unscoped value is treated as a miss and re-uploaded, so it
self-heals. And inlined text is bounded against the model's context
window, because _enforce_context_window runs before attachments are
merged into the messages and nothing downstream would trim a 200-page
extraction.
---
 application/llm/openai.py | 118 ++++++++++++++++++++++++--
 tests/llm/test_openai.py  | 172 ++++++++++++++++++++++++++++++++++++--
 2 files changed, 278 insertions(+), 12 deletions(-)

diff --git a/application/llm/openai.py b/application/llm/openai.py
index 0a0a27b2..466ea0b1 100644
--- a/application/llm/openai.py
+++ b/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,75 @@ 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:
+            if num_tokens_from_string(content) <= budget:
+                return content
+            chars_per_token = len(content) / max(num_tokens_from_string(content), 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 +1805,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 +1848,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(
diff --git a/tests/llm/test_openai.py b/tests/llm/test_openai.py
index 024ebe04..a58031ef 100644
--- a/tests/llm/test_openai.py
+++ b/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"
 

From 23e249fca1f79435a89799ee944c3d6e6ad0a87b Mon Sep 17 00:00:00 2001
From: Alex <a@tushynski.me>
Date: Wed, 5 Aug 2026 11:26:23 +0100
Subject: [PATCH 4/5] fix(workflows): validate CEL at save time and document
 which fields use it
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Workflows ship two expression languages in adjacent fields. Agent
`prompt_template` and end `output_template` are Jinja2, so `{{query}}` is
correct there. State and condition `expression` fields are bare CEL,
where the same string is a parse error.

Nothing told users this. docs/content/Agents/nodes.mdx documented the
pre-CEL Set State node with `{{variable_name}}` examples and Set /
Increment / Append operations that no longer exist — "CEL" appeared
nowhere in the docs. The builder's own placeholder was `input.foo + 1`,
referencing an `input` namespace that is not in workflow state, while the
panel beside it advertised `{{ agent.variable }}`.

There was also no validation anywhere: neither validateWorkflow nor
validate_workflow_structure looked at state nodes at all, and conditions
were only checked for a non-empty string. A workflow containing
`{{query}}` saved and published clean, then aborted on the first message
with a bare caret dump that sanitize_api_error collapsed into "An error
occurred … please try again later" — advice that sends the user round the
same loop, which is what the 2026-08-01 report shows.

Adds validate_cel_expression (compile-only, since state is built at run
time and unresolved names are not knowable when saving) and wires it into
the workflow save path for both node types, plus the half-configured
state operations the engine silently skips. `{{ }}` gets a targeted hint,
raised only after compilation has already failed so valid CEL that
contains braces — `x == "{{y}}"`, or a nested map literal — is not
rejected.

celpy errors are now summarized: undeclared-reference messages embed a
repr of the entire activation, thousands of characters including the
user's own query, and others interpolate the offending state value.
Quoted fragments that are not bare identifiers are redacted, since
config errors now bypass sanitize_api_error to reach the user, and a
shared agent's runner is not its owner.

Docs rewritten with a table of which field takes which syntax, CEL
examples for Set State, and the Condition node section that basics.mdx
has been linking to all along.
---
 application/agents/workflows/cel_evaluator.py | 93 ++++++++++++++++++-
 .../agents/workflows/workflow_engine.py       | 20 +++-
 application/api/user/workflows/routes.py      | 61 ++++++++++++
 docs/content/Agents/nodes.mdx                 | 91 ++++++++++++++----
 .../src/agents/workflow/WorkflowBuilder.tsx   | 13 ++-
 tests/agents/test_cel_evaluator.py            | 83 +++++++++++++++++
 tests/agents/test_workflow_engine.py          | 90 ++++++++++++++++++
 7 files changed, 426 insertions(+), 25 deletions(-)

diff --git a/application/agents/workflows/cel_evaluator.py b/application/agents/workflows/cel_evaluator.py
index 5f8bab37..2a7a1748 100644
--- a/application/agents/workflows/cel_evaluator.py
+++ b/application/agents/workflows/cel_evaluator.py
@@ -1,3 +1,4 @@
+import re
 from typing import Any, Dict
 
 import celpy
@@ -32,19 +33,105 @@ 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()
+    # Quoted fragments in celpy messages are either the offending identifier
+    # (safe, and the useful part) or a state value (not ours to echo). Keep
+    # bare identifiers, redact anything else.
+    def _redact(match: "re.Match[str]") -> str:
+        inner = match.group(1)
+        return f"'{inner}'" if inner.isidentifier() else "'…'"
+
+    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)
 
 
diff --git a/application/agents/workflows/workflow_engine.py b/application/agents/workflows/workflow_engine.py
index 0c4f39a3..b4b5c145 100644
--- a/application/agents/workflows/workflow_engine.py
+++ b/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)
diff --git a/application/api/user/workflows/routes.py b/application/api/user/workflows/routes.py
index 47a9798a..c5e92832 100644
--- a/application/api/user/workflows/routes.py
+++ b/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"))
diff --git a/docs/content/Agents/nodes.mdx b/docs/content/Agents/nodes.mdx
index 76656742..d0b2eb12 100644
--- a/docs/content/Agents/nodes.mdx
+++ b/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.
 
 ---
 
diff --git a/frontend/src/agents/workflow/WorkflowBuilder.tsx b/frontend/src/agents/workflow/WorkflowBuilder.tsx
index a2dca188..1dc81940 100644
--- a/frontend/src/agents/workflow/WorkflowBuilder.tsx
+++ b/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"
diff --git a/tests/agents/test_cel_evaluator.py b/tests/agents/test_cel_evaluator.py
index 70209b56..d42c274b 100644
--- a/tests/agents/test_cel_evaluator.py
+++ b/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,85 @@ 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
+    def test_error_does_not_echo_state_values(self):
+        """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": "SUPER-SECRET-VALUE"})
+        assert "SUPER-SECRET-VALUE" not in str(exc.value)
+
+    @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")
diff --git a/tests/agents/test_workflow_engine.py b/tests/agents/test_workflow_engine.py
index 6f375f8a..6f77cd46 100644
--- a/tests/agents/test_workflow_engine.py
+++ b/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": {}},

From e1962480580b4b256242f239233c242a314cfa8d Mon Sep 17 00:00:00 2001
From: Alex <a@tushynski.me>
Date: Wed, 5 Aug 2026 11:57:27 +0100
Subject: [PATCH 5/5] fix(workflows): redact CEL error values by default;
 tokenize attachments once
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Two follow-ups from PR review.

_fit_attachment_text tokenized the extraction twice on the oversized
path — once for the budget check and once for chars_per_token. That is a
full BPE pass over the whole document (~12ms per 250k chars) on the hot
path of every attachment turn. Compute it once.

The CEL error summarizer kept quoted fragments that looked like
identifiers, on the theory that those are variable names rather than
data. They are not distinguishable: celpy interpolates state values into
messages (`StringType('SECRET')`), and a one-word user query such as
`invoice` is a valid identifier, so it survived redaction and reached
both the log line and — since config errors now bypass
sanitize_api_error — the user. The earlier test passed only because its
sample value contained hyphens.

Quoted fragments are now redacted by default, allowing only the two
positions known to hold names rather than data: the variable in
"undeclared reference to 'x'" and the class in "<class 'ValueError'>".
Tests cover identifier-shaped values, that the exception class survives,
the truncation backstop, and the empty-expression guard.
---
 application/agents/workflows/cel_evaluator.py | 15 ++++++--
 application/llm/openai.py                     |  7 +++-
 tests/agents/test_cel_evaluator.py            | 38 +++++++++++++++++--
 3 files changed, 51 insertions(+), 9 deletions(-)

diff --git a/application/agents/workflows/cel_evaluator.py b/application/agents/workflows/cel_evaluator.py
index 2a7a1748..121b5a28 100644
--- a/application/agents/workflows/cel_evaluator.py
+++ b/application/agents/workflows/cel_evaluator.py
@@ -52,12 +52,19 @@ def _summarize_cel_error(exc: Exception) -> str:
     if marker in text:
         text = text.split(marker, 1)[0]
     text = text.strip().lstrip("(").strip()
-    # Quoted fragments in celpy messages are either the offending identifier
-    # (safe, and the useful part) or a state value (not ours to echo). Keep
-    # bare identifiers, redact anything else.
+
+    # 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)
-        return f"'{inner}'" if inner.isidentifier() else "'…'"
+        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()
diff --git a/application/llm/openai.py b/application/llm/openai.py
index 466ea0b1..3bd65446 100644
--- a/application/llm/openai.py
+++ b/application/llm/openai.py
@@ -1742,9 +1742,12 @@ class OpenAILLM(BaseLLM):
         # Half the window: the prompt, history and the answer share the rest.
         budget = int(limit * 0.5) if limit else 24000
         try:
-            if num_tokens_from_string(content) <= budget:
+            # 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(num_tokens_from_string(content), 1)
+            chars_per_token = len(content) / max(token_count, 1)
             keep = max(int(budget * chars_per_token * 0.95), 0)
         except Exception:
             keep = budget * 4
diff --git a/tests/agents/test_cel_evaluator.py b/tests/agents/test_cel_evaluator.py
index d42c274b..6592bc6a 100644
--- a/tests/agents/test_cel_evaluator.py
+++ b/tests/agents/test_cel_evaluator.py
@@ -204,12 +204,44 @@ class TestCelErrorMessages:
         assert "NameContainer" not in message
 
     @pytest.mark.unit
-    def test_error_does_not_echo_state_values(self):
+    @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": "SUPER-SECRET-VALUE"})
-        assert "SUPER-SECRET-VALUE" not in str(exc.value)
+            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(