fix(workflow): raise ToolExecutionError on expert-analysis failure

[sanjibani]

Summary

tools/workflow/workflow_mixin.py._call_expert_analysis caught except Exception as e: and returned {"error": str(e), "status": "analysis_error"} as a dict. FastMCP wraps that as success content with isError=false, so MCP clients treat the failure as data and the LLM often proceeds as if the call had succeeded.

This is the same isError-compliance gap flagged in the recent MCP security audit (Dayna Blackwell, Tool Poisoning, Rug Pulls, and Prompt Injections — oh my!).

Changes

The swallowed-error return is replaced with raise ToolExecutionError(json.dumps({...})) so FastMCP sets isError=true on the wire while preserving the formatted JSON payload in content for the LLM. ToolExecutionError is already used elsewhere in this codebase (tools/simple/base.py) for the same purpose, so this just aligns the workflow mixin with the established pattern.

Note: This is the single site in pal-mcp-server that hadn't been migrated yet; the project's main execute method already raises ToolExecutionError correctly. This PR closes the remaining gap.

Tests

tests/test_workflow_iserror_compliance.py asserts that a failing expert-analysis call raises ToolExecutionError with a JSON payload containing the original error message and status='analysis_error'.

Reference: https://composio.dev/blog/mcp-security-vulnerabilities

[gemini-code-assist]

Code Review

This pull request modifies _call_expert_analysis in tools/workflow/workflow_mixin.py to raise a ToolExecutionError instead of returning an error dictionary when an exception occurs, ensuring that FastMCP correctly sets isError=true on the wire. A regression test is also added to verify this behavior. However, the review points out a critical issue in the test implementation: instantiating BaseWorkflowMixin using __new__ bypasses initialization and missing methods, causing an AttributeError to be raised and caught instead of the intended upstream model timeout. A concrete dummy subclass should be used in the test to properly mock the dependencies.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

[gemini-code-assist]

The test currently uses BaseWorkflowMixin.__new__(BaseWorkflowMixin) to instantiate the mixin directly. This bypasses __init__, leaving self.consolidated_findings uninitialized. When _call_expert_analysis is executed, it calls self.prepare_expert_analysis_context(self.consolidated_findings), which immediately raises an AttributeError.

Furthermore, _call_expert_analysis relies on several methods that are not defined in BaseWorkflowMixin but are expected to be mixed in from BaseTool (such as _augment_system_prompt_with_capabilities and validate_and_correct_temperature). Calling these also raises AttributeError.

Because of these errors, the code never actually reaches the mocked provider's generate_content call. Instead, it jumps straight to the except Exception as e: block, which catches the AttributeError and raises ToolExecutionError. Consequently, the assertion assert "upstream model timeout" in str(data.get("error", "")) will fail because the error payload contains the AttributeError message instead of the expected "upstream model timeout".

To fix this, we should define a concrete dummy subclass of BaseWorkflowMixin that implements the abstract methods and stubs the mixed-in methods, and then instantiate it normally so that __init__ is properly called.

        from tools.workflow.workflow_mixin import BaseWorkflowMixin

        class DummyWorkflowTool(BaseWorkflowMixin):
            def get_name(self) -> str:
                return "test_tool"
            def get_workflow_request_model(self):
                pass
            def get_system_prompt(self) -> str:
                return "system prompt"
            def get_language_instruction(self) -> str:
                return "language instruction"
            def get_default_temperature(self) -> float:
                return 0.7
            def get_model_provider(self, model_name: str):
                pass
            def _resolve_model_context(self, arguments: dict, request):
                pass
            def _prepare_file_content_for_prompt(self, *args, **kwargs):
                return "", []
            def get_work_steps(self, request):
                return []
            def get_required_actions(self, *args, **kwargs):
                return []
            def _augment_system_prompt_with_capabilities(self, system_prompt, capabilities):
                return system_prompt
            def validate_and_correct_temperature(self, temp, context):
                return temp, []

        mixin = DummyWorkflowTool()

        # Force the model call to raise
        async def _boom(*args, **kwargs):
            raise RuntimeError("upstream model timeout")

        # Inject the failing provider into the mixin's model_context
        class _FakeModelContext:
            def __init__(self):
                self.provider = type(
                    "_P", (), {"generate_content": _boom}
                )()
                self.capabilities = None

        mixin._model_context = _FakeModelContext()

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 5bdd4bc727

ℹ️ About Codex in GitHub

Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".

[chatgpt-codex-connector]

Instantiate a concrete workflow subclass

This test cannot reach the expert-analysis path because BaseWorkflowMixin is an abstract class; calling BaseWorkflowMixin.__new__(BaseWorkflowMixin) raises TypeError: Can't instantiate abstract class ... before the fake provider is installed. Use a minimal concrete subclass (or an existing workflow tool) so the regression test actually exercises _call_expert_analysis.

Useful? React with 👍 / 👎.

[chatgpt-codex-connector]

Make the fake provider call synchronous

After the abstract-instantiation issue is fixed, this stub still will not raise RuntimeError("upstream model timeout") in _call_expert_analysis: generate_content is a synchronous provider API and the workflow immediately dereferences model_response.content, so an async def here returns a coroutine and the test fails on an AttributeError/unawaited coroutine instead of preserving the intended timeout message. Define _boom as a regular function that raises.

Useful? React with 👍 / 👎.