fix(integration): run general preset on Gemini, honor workflow edges, classify 413

Author: ctrl-gaurav

effgen/cli/_main.py

@@ -2122,16 +2122,22 @@ def _models_info(self, args):
             for k, v in rows.items():
                 table.add_row(k, str(v))
             self.console.print(table)
-            self.console.print(f"\n[dim]Use: [cyan]effgen run --provider {rec.provider} "
-                               f"-m {rec.id} \"...\"[/cyan][/dim]")
         else:
             for k, v in rows.items():
                 print(f"  {k}: {v}")
 
-        # If the same id is also downloaded locally, surface the local copy and a
-        # local invocation alongside the cloud row (don't pretend it's cloud-only).
+        cloud_hint = f"effgen run --provider {rec.provider} -m {rec.id} \"...\""
+        # If the same id is also downloaded locally, lead with the local engine
+        # path (the local block prints its own "Run locally" hint) and show the
+        # cloud invocation as an alternative — don't present it as cloud-only.
         if local_entry is not None:
             self._render_local_model_info(self._local_model_payload(local_entry))
+            if self.console:
+                self.console.print(f"\n[dim]Cloud alternative: [cyan]{cloud_hint}[/cyan][/dim]")
+            else:
+                print(f"\n  Cloud alternative: {cloud_hint}")
+        elif self.console:
+            self.console.print(f"\n[dim]Use: [cyan]{cloud_hint}[/cyan][/dim]")
         return 0
 
     def _models_load(self, args):
@@ -3761,7 +3767,18 @@ def _handle_eval_command(args, cli) -> int:
                 cli.print(f"  {name:16s} — {desc}")
             return 0
 
-        suite = _resolve_eval_suite(suite_name, difficulty=difficulty, max_cases=max_cases)
+        # A bad data file (unknown field, empty, missing) is a user error, not a
+        # crash — report it and exit 2 (matching `compare`) instead of a
+        # traceback.
+        try:
+            suite = _resolve_eval_suite(suite_name, difficulty=difficulty, max_cases=max_cases)
+        except (ValueError, FileNotFoundError) as exc:
+            cli.print(
+                f"Could not load suite '{suite_name}' ({exc}). "
+                f"Use a built-in suite ({', '.join(list_suites())}) "
+                "or a path to a .jsonl/.json file of test cases."
+            )
+            return 2
 
         # Report any narrowing applied
         if difficulty:

effgen/core/workflow.py

@@ -548,7 +548,10 @@ def from_yaml(cls, path: str, agent_factory: Callable[[dict[str, Any]], Any] | N
         """
         Load a workflow from a YAML file.
 
-        Expected format::
+        Dependencies can be declared per-node with ``depends_on`` or in a
+        top-level ``edges`` list; both build the same graph. Each edge is either
+        a ``[source, target]`` pair or a mapping with ``source``/``target``
+        (aliases ``from``/``to``) and an optional ``key``::
 
             workflow:
               name: my_pipeline
@@ -560,6 +563,10 @@ def from_yaml(cls, path: str, agent_factory: Callable[[dict[str, Any]], Any] | N
                   agent: summary_agent
                   depends_on: [search]
 
+            # equivalent wiring via a top-level edges block:
+            #   edges:
+            #     - [search, summarize]
+
         Args:
             path: Path to the YAML file
             agent_factory: Optional callable that receives a node dict and
@@ -568,6 +575,10 @@ def from_yaml(cls, path: str, agent_factory: Callable[[dict[str, Any]], Any] | N
 
         Returns:
             A validated WorkflowDAG
+
+        An unrecognized top-level key is reported with a warning rather than
+        dropped silently, so a mis-keyed file (e.g. ``edge:`` instead of
+        ``edges:``) does not validate as a workflow with all its declared wiring.
         """
         import yaml  # pyyaml is an existing dependency
 
@@ -577,6 +588,17 @@ def from_yaml(cls, path: str, agent_factory: Callable[[dict[str, Any]], Any] | N
         wf_data = data.get("workflow", data)
         name = wf_data.get("name", "workflow")
 
+        # Surface unknown top-level keys instead of dropping the wiring silently.
+        _known_top = {"name", "nodes", "edges", "description", "metadata"}
+        unknown = [k for k in wf_data if k not in _known_top]
+        if unknown:
+            logger.warning(
+                "Workflow '%s': ignoring unrecognized top-level key(s) %s "
+                "(recognized: %s). Declare dependencies with per-node "
+                "'depends_on' or a top-level 'edges' list.",
+                name, sorted(unknown), sorted(_known_top),
+            )
+
         dag = cls(name=name)
 
         node_defs = wf_data.get("nodes", [])
@@ -589,19 +611,54 @@ def from_yaml(cls, path: str, agent_factory: Callable[[dict[str, Any]], Any] | N
                 id=nd["id"],
                 agent=agent,
                 tools=nd.get("tools", []),
+                input_keys=nd.get("input_keys", []),
                 output_key=nd.get("output_key", nd["id"]),
                 metadata={k: v for k, v in nd.items()
-                          if k not in ("id", "tools", "output_key", "depends_on", "agent")},
+                          if k not in ("id", "tools", "input_keys",
+                                       "output_key", "depends_on", "agent")},
             )
             dag.add_node(node)
 
-        # Create edges from depends_on
+        # Create edges from per-node depends_on ...
         for nd in node_defs:
             for dep in nd.get("depends_on", []):
                 dag.connect(dep, nd["id"])
 
+        # ... and from a top-level edges list (same graph; both may be present).
+        for edge in wf_data.get("edges", []):
+            src, tgt, key = cls._parse_yaml_edge(edge)
+            dag.connect(src, tgt, key=key)
+
         return dag
 
+    @staticmethod
+    def _parse_yaml_edge(edge: Any) -> tuple[str, str, str | None]:
+        """Parse one entry of a YAML ``edges`` list into ``(source, target, key)``.
+
+        Accepts a ``[source, target]`` pair or a mapping with ``source``/``target``
+        (aliases ``from``/``to``) and an optional ``key``.
+        """
+        if isinstance(edge, list | tuple):
+            if len(edge) < 2:
+                raise ValueError(
+                    f"Workflow edge {edge!r} must be [source, target]."
+                )
+            return str(edge[0]), str(edge[1]), (str(edge[2]) if len(edge) > 2 else None)
+        if isinstance(edge, dict):
+            src = edge.get("source", edge.get("from"))
+            tgt = edge.get("target", edge.get("to"))
+            if not src or not tgt:
+                raise ValueError(
+                    f"Workflow edge {edge!r} needs 'source'/'target' "
+                    "(aliases 'from'/'to')."
+                )
+            key = edge.get("key")
+            return str(src), str(tgt), (str(key) if key is not None else None)
+        raise ValueError(
+            f"Unsupported workflow edge {edge!r}: use [source, target] or "
+            "{source, target}."
+        )
+
     # -- Introspection --
 
     def get_node(self, node_id: str) -> WorkflowNode | None:

effgen/eval/evaluator.py

@@ -41,7 +41,8 @@ class TestCase:
 
     Attributes:
         query: The input query for the agent. ``input=`` is accepted as an
-            ergonomic alias when constructing a TestCase.
+            ergonomic alias when constructing a TestCase; :meth:`from_dict`
+            additionally reads ``prompt`` / ``question``.
         expected_output: Expected output text (used for exact/contains/regex).
             ``expected=`` is accepted as an ergonomic alias.
         expected_tools: Tool names the agent should invoke.
@@ -70,17 +71,35 @@ def __post_init__(self) -> None:
             self.difficulty = Difficulty(self.difficulty)
         if not self.query:
             raise ValueError(
-                "TestCase requires a non-empty 'query' (or 'input=' alias)."
+                "TestCase requires a non-empty 'query' "
+                "(aliases: input/prompt/question)."
             )
 
+    # Field names accepted for the query text, in priority order.
+    _QUERY_KEYS = ("query", "input", "prompt", "question")
+
     @classmethod
     def from_dict(cls, data: dict[str, Any]) -> TestCase:
-        """Create a TestCase from a dict (e.g. parsed JSONL line)."""
+        """Create a TestCase from a dict (e.g. parsed JSONL line).
+
+        The query text is read from ``query`` or any of the aliases
+        ``input`` / ``prompt`` / ``question``. A dict with none of these keys
+        raises a ``ValueError`` naming the field it needs and the keys it saw.
+        """
+        query = next(
+            (data[k] for k in cls._QUERY_KEYS if data.get(k)), None
+        )
+        if query is None:
+            raise ValueError(
+                "each test case needs a 'query' field "
+                "(aliases: input/prompt/question); got keys: "
+                f"{sorted(data.keys())}"
+            )
         difficulty = data.get("difficulty", "medium")
         if isinstance(difficulty, str):
             difficulty = Difficulty(difficulty)
         return cls(
-            query=data["query"],
+            query=query,
             expected_output=data.get("expected", data.get("expected_output", "")),
             expected_tools=data.get("tools", data.get("expected_tools", [])),
             tags=data.get("tags", []),

effgen/models/errors.py

@@ -521,7 +521,9 @@ def classify_provider_error(exc: Exception) -> ErrorClass:
             return _NOT_FOUND
         if status == 429:
             return _RATE_LIMITED
-        if status in (400, 422):
+        # 413 payload-too-large is a property of the request, not a transient
+        # rate limit — retrying the same oversized request will not succeed.
+        if status in (400, 413, 422):
             return _INVALID
         if status == 408 or status >= 500:
             return _TRANSIENT

effgen/models/gemini_adapter.py

@@ -314,6 +314,15 @@ def _sanitize_schema(cls, schema: Any) -> Any:
                     cleaned[k] = list(v) if isinstance(v, list | tuple) else v
                 else:
                     cleaned[k] = v
+            # Gemini rejects an ``array`` property that omits ``items``. Default a
+            # missing element schema to a string so tools that declare a plain
+            # ``type: array`` parameter are accepted (OpenAI/Groq tolerate the
+            # omission). Covers every current and future tool without per-tool
+            # patches.
+            type_val = cleaned.get("type")
+            if isinstance(type_val, str) and type_val.lower() == "array" \
+                    and "items" not in cleaned:
+                cleaned["items"] = {"type": "string"}
             return cleaned
         if isinstance(schema, list):
             return [cls._sanitize_schema(v) for v in schema]

effgen/models/groq_adapter.py

@@ -49,6 +49,24 @@
 _GROQ_MODEL_TYPE_VALUE = "groq"
 
 
+def _redact_groq_org(message: str) -> str:
+    """Remove the caller's organization id from a Groq error body before it is
+    surfaced (it is an account identifier, not useful for debugging)."""
+    return re.sub(r"organization `org_[^`]+`", "organization `***`", message)
+
+
+def _is_request_too_large(message: str, message_lower: str) -> bool:
+    """True when a Groq error is a 413 payload-too-large (a single oversized
+    request), not a 429 rate limit. Groq returns 413 with a ``rate_limit_exceeded``
+    code for a request over the per-minute token limit, so status + wording are
+    checked rather than the misleading code."""
+    return (
+        "413" in message
+        or "request too large" in message_lower
+        or "reduce your message size" in message_lower
+    )
+
+
 def _parse_failed_generation_tool_call(message: str) -> dict[str, Any] | None:
     """Extract a tool call from Groq's ``tool_use_failed`` failed_generation text."""
     match = re.search(
@@ -441,6 +459,22 @@ def _do_generate(
                         message=msg,
                     ) from exc
 
+                # A 413 payload-too-large is a permanent property of this
+                # request, not a transient rate limit — fail fast with a
+                # fix-oriented hint instead of routing it through retry/failover.
+                if _is_request_too_large(msg, msg_lower):
+                    from effgen.models.errors import InvalidRequestError as _IRE
+                    raise _IRE(
+                        provider="groq",
+                        model_name=self.model_name,
+                        message=(
+                            f"request too large for {self.model_name}: "
+                            f"{_redact_groq_org(msg)} — reduce the request "
+                            "(fewer/smaller tools or shorter input) or use a "
+                            "larger-context model."
+                        ),
+                    ) from exc
+
                 is_rate = "429" in msg or "rate_limit" in msg_lower or "rate limit" in msg_lower
                 is_server = "500" in msg or "503" in msg or "internal" in msg_lower
                 is_timeout = "timeout" in msg_lower
@@ -733,6 +767,18 @@ def generate_stream(
                     model_name=self.model_name,
                     message=msg,
                 ) from exc
+            if _is_request_too_large(msg, msg_lower):
+                from effgen.models.errors import InvalidRequestError as _IRE
+                raise _IRE(
+                    provider="groq",
+                    model_name=self.model_name,
+                    message=(
+                        f"request too large for {self.model_name}: "
+                        f"{_redact_groq_org(msg)} — reduce the request "
+                        "(fewer/smaller tools or shorter input) or use a "
+                        "larger-context model."
+                    ),
+                ) from exc
             is_rate = "429" in msg or "rate_limit" in msg_lower or "rate limit" in msg_lower
             if is_rate:
                 from effgen.models._rate_limit import RateLimitExceeded as _RLE

effgen/tools/protocols/mcp_official/client.py

@@ -352,6 +352,12 @@ def get_effgen_tools(self, prefix: str = "mcp_") -> list[Any]:
         This is a **synchronous** accessor (it reads the catalog cached at
         :meth:`connect`); call it directly, do not ``await`` it.
 
+        Each returned tool is named ``{prefix}{server_tool_name}`` — with the
+        default ``prefix="mcp_"``, a server tool ``calculator`` is exposed to the
+        agent as ``mcp_calculator``. Look tools up by that namespaced name; to
+        reach the original server name pass ``prefix=""`` here, or use
+        :meth:`get_tool` (which is keyed by the unprefixed name).
+
         .. note::
             An ``Agent`` that holds these tools must be driven with
             ``await agent.run_async(...)``, not the synchronous ``Agent.run()``:
@@ -361,10 +367,12 @@ def get_effgen_tools(self, prefix: str = "mcp_") -> list[Any]:
 
         Args:
             prefix: Name prefix for the wrapped tools (avoids collisions with
-                effGen's builtin tool names).
+                effGen's builtin tool names). Defaults to ``"mcp_"``; pass
+                ``""`` to keep the original server tool names.
 
         Returns:
-            List of effGen ``BaseTool`` instances.
+            List of effGen ``BaseTool`` instances, each named
+            ``{prefix}{server_tool_name}``.
         """
         return [
             _MCPOfficialToolBridge(tool, self, prefix=prefix)

tests/integration/test_gemini_tools_live.py

@@ -170,3 +170,21 @@ def test_parallel_function_calls_live():
         print("Parallel function calls response:", response.output[:300])
     finally:
         model.unload()
+
+
+@SKIP
+def test_general_preset_array_param_tools_accepted_live():
+    """The general preset attaches tools whose parameters include a plain
+    ``type: array`` field (map markers, email attachments). Gemini rejects an
+    array property without ``items``; the schema sanitizer defaults it so the
+    request is accepted and the preset runs instead of failing with a 400."""
+    from effgen.presets import create_agent
+
+    agent = create_agent("general", f"gemini:{MODEL}")
+    response = _run_agent_or_skip_transient(agent, "Say hello in one word.")
+    _skip_if_gemini_transient(response.output)
+    # The prior failure surfaced as a 400 "properties[markers].items: missing
+    # field" before generation — assert the request was accepted and answered.
+    assert response.success, f"general preset failed on Gemini: {response.metadata.get('error')}"
+    assert response.output.strip(), "Expected a non-empty answer"
+    print("general preset on Gemini:", response.output[:120])

tests/models/test_gemini_adapter.py

@@ -44,6 +44,31 @@ def test_sanitize_schema_strips_unsupported_fields():
     }
 
 
+def test_sanitize_schema_defaults_missing_array_items():
+    """Gemini rejects an ``array`` property without ``items``; the sanitizer
+    defaults the element schema so tools with a plain ``type: array`` param are
+    accepted (this is what the general preset's maps/email tools declare)."""
+    raw = {
+        "type": "object",
+        "properties": {
+            "markers": {"type": "array", "description": "map markers"},
+            "attachments": {"type": "array"},
+            "typed": {"type": "array", "items": {"type": "integer"}},
+        },
+        "required": ["markers"],
+    }
+    cleaned = GeminiAdapter._sanitize_schema(raw)
+    assert cleaned["properties"]["markers"]["items"] == {"type": "string"}
+    assert cleaned["properties"]["attachments"]["items"] == {"type": "string"}
+    # An explicit items schema is preserved, not overwritten.
+    assert cleaned["properties"]["typed"]["items"] == {"type": "integer"}
+
+
+def test_sanitize_schema_top_level_array_gets_items():
+    cleaned = GeminiAdapter._sanitize_schema({"type": "array"})
+    assert cleaned == {"type": "array", "items": {"type": "string"}}
+
+
 def test_sanitize_schema_preserves_property_names():
     """`properties` keys are user-defined names, not schema fields — keep them."""
     raw = {