diff --git a/application/agents/workflows/cel_evaluator.py b/application/agents/workflows/cel_evaluator.py index 5f8bab37..121b5a28 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,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 "". + 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(" _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 e08cf444..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) @@ -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; 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/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/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 +