fix(core,rag,models,ui): stop word-count sub-agent fan-out; wire decomposition client; per-page PDF citations

Author: ctrl-gaurav

effgen/cli/_main.py

@@ -881,13 +881,18 @@ def run_agent(self, args):
 
                 agent = Agent(agent_config, session_id=getattr(args, 'session_id', None))
 
-            # Determine execution mode
-            mode = AgentMode.AUTO
-            if args.mode:
-                if args.mode == "single":
-                    mode = AgentMode.SINGLE
-                elif args.mode == "sub_agents":
-                    mode = AgentMode.SUB_AGENTS
+            # Determine execution mode. Default to the agent's own
+            # config.mode (SINGLE unless set otherwise) instead of forcing
+            # AUTO, so a plain `effgen run "<task>"` doesn't switch to
+            # sub-agent decomposition on its own — pass --mode auto to
+            # opt in explicitly for a call.
+            mode = None
+            if args.mode == "single":
+                mode = AgentMode.SINGLE
+            elif args.mode == "sub_agents":
+                mode = AgentMode.SUB_AGENTS
+            elif args.mode == "auto":
+                mode = AgentMode.AUTO
 
             # Run task
             self.print(f"\n[bold]Task:[/bold] {args.task}" if self.console else f"\nTask: {args.task}")

effgen/core/agent.py

@@ -178,12 +178,25 @@ class AgentConfig:
         temperature: Generation temperature
         top_p: Nucleus-sampling threshold; overridden per call by run(top_p=...)
         top_k: Top-k sampling cutoff (providers that don't support it ignore it)
-        seed: Sampling seed; a fixed seed plus temperature=0 reproduces a
-            generation exactly, on providers that support it
+        seed: Sampling seed. A fixed seed plus temperature=0 reproduces a
+            generation exactly on Gemini, Groq, and local engines
+            (transformers/vllm/gguf/mlx). OpenAI's chat models accept
+            ``seed`` and typically reproduce output, but the same
+            seed+temperature=0 request can still return a different
+            completion — OpenAI documents this as best-effort determinism,
+            not a guarantee, especially for reasoning-tier models. Treat an
+            OpenAI ``seed`` as "usually reproducible," not "always."
         presence_penalty: Penalizes tokens already present anywhere in the text
         frequency_penalty: Penalizes tokens proportionally to how often they
             already appeared (the standard anti-repetition knob for long text)
         repetition_penalty: Multiplicative repeat penalty used by local/HF engines
+        mode: Default execution mode for run()/run_async() when the call
+            site doesn't pass its own ``mode=``. Defaults to
+            ``AgentMode.SINGLE`` so a plain ``Agent(config).run(task)`` never
+            switches to sub-agent decomposition on its own; set to
+            ``AgentMode.AUTO`` to have the router decide per call, or pass
+            ``mode=`` on an individual ``run()`` call to override this
+            default just for that call.
         enable_sub_agents: Enable sub-agent spawning
         enable_memory: Enable memory systems
         enable_streaming: Enable response streaming
@@ -223,6 +236,7 @@ class AgentConfig:
     presence_penalty: float = 0.0
     frequency_penalty: float = 0.0
     repetition_penalty: float = 1.0
+    mode: AgentMode = AgentMode.SINGLE
     enable_sub_agents: bool = True
     enable_memory: bool = True
     enable_streaming: bool = False
@@ -472,6 +486,19 @@ def to_dict(self) -> dict[str, Any]:
 from .agent_react import AgentReActMixin  # noqa: E402
 
 
+class _AgentDecompositionClient:
+    """Adapts an :class:`Agent` to the ``generate(prompt, **kwargs) -> str``
+    interface :class:`~effgen.core.decomposition_engine.DecompositionEngine`
+    expects for LLM-assisted task decomposition, by routing through the
+    agent's own model-generation path (:meth:`Agent._generate`)."""
+
+    def __init__(self, agent: "Agent") -> None:
+        self._agent = agent
+
+    def generate(self, prompt: str, **kwargs: Any) -> str:
+        return self._agent._generate(prompt, **kwargs).get("text", "")
+
+
 class Agent(AgentGenerationMixin, AgentReActMixin, AgentRuntimeMixin):
     """
     Main Agent implementation with ReAct loop and sub-agent support.
@@ -749,7 +776,7 @@ def __init__(self, config: AgentConfig | None = None, session_id: str | None = N
         if config.enable_sub_agents:
             self.router = SubAgentRouter(
                 config=config.router_config,
-                llm_client=self  # Pass self as LLM client
+                llm_client=_AgentDecompositionClient(self)
             )
             self.sub_agent_manager = SubAgentManager(
                 parent_agent=self,
@@ -1057,7 +1084,7 @@ def _auto_detect_verbose(self) -> bool:
 
     def run(self,
             task: "str | Message | list[Any]",
-            mode: AgentMode = AgentMode.AUTO,
+            mode: AgentMode | None = None,
             context: dict[str, Any] | None = None,
             output_schema: dict[str, Any] | None = None,
             output_model: Any = None,
@@ -1071,7 +1098,11 @@ def run(self,
                 ``Message``, or a ``list[ContentPart]``; text is extracted and
                 any image/audio/video parts are routed through the multimodal
                 path.
-            mode: Execution mode (single, sub_agents, auto)
+            mode: Execution mode (single, sub_agents, auto). Defaults to
+                ``self.config.mode`` (``AgentMode.SINGLE`` unless the config
+                sets otherwise) when omitted — pass ``mode=AgentMode.AUTO``
+                to let the router decide based on task complexity for this
+                call only.
             context: Optional context
             output_schema: A JSON-Schema ``dict`` **or** a Pydantic
                 ``BaseModel`` subclass — when provided, the final output is
@@ -1095,6 +1126,8 @@ def run(self,
         Returns:
             AgentResponse with results
         """
+        if mode is None:
+            mode = self.config.mode
         unrecognized = {k for k in kwargs if k not in _RUN_KWARGS and not k.startswith("_")}
         if unrecognized:
             bad = sorted(unrecognized)[0]
@@ -1748,7 +1781,7 @@ def synthesize(self, synthesis_data: dict[str, Any]) -> str:
 
     async def run_async(self,
                        task: "str | Message | list[Any]",
-                       mode: AgentMode = AgentMode.AUTO,
+                       mode: AgentMode | None = None,
                        context: dict[str, Any] | None = None,
                        output_schema: dict[str, Any] | None = None,
                        output_model: Any = None,
@@ -1763,7 +1796,8 @@ async def run_async(self,
 
         Args:
             task: Task description (str, Message, or list[ContentPart])
-            mode: Execution mode
+            mode: Execution mode. Defaults to ``self.config.mode`` when
+                omitted (see ``run()``).
             context: Optional context
             output_schema: A JSON-Schema ``dict`` or a Pydantic ``BaseModel``
                 subclass (see ``run``).
@@ -1911,7 +1945,7 @@ def _stream_direct(self, task: str, on_answer: Callable[[str], None] | None = No
 
     def stream(self,
                task: "str | Message | list[Any]",
-               mode: AgentMode = AgentMode.AUTO,
+               mode: AgentMode | None = None,
                context: dict[str, Any] | None = None,
                on_thought: Callable[[str], None] | None = None,
                on_tool_call: Callable[[str, str], None] | None = None,

effgen/core/agent_react.py

@@ -1171,6 +1171,16 @@ def _attach_citations(self, response: "AgentResponse") -> None:
         ``web_search_call`` action sources). Those widen ``response.sources``
         so a search that ran is never left unsourced, but they never
         manufacture a ``Citation`` entry the model did not actually make.
+
+        Locally-mined URL sources (``_collect_citations``, e.g. a plain
+        ``web_search`` tool) carry no such explicit marker; a search can
+        return results the model never draws on. For those, "cited" is
+        determined by whether the model's final answer text actually
+        contains the URL — mirroring the native-provider distinction above.
+        Non-URL sources (e.g. a RAG file path) use inline ``[N]`` bracket
+        markers this generic mining path can't reliably map back to one
+        item across multiple tool calls, so they keep the previous default
+        (cited=True) rather than risk dropping a real citation.
         """
         raw = list(getattr(self, "_collected_citations", None) or [])
         # Fold provider-native grounding chunks ({url, title}) into the same
@@ -1196,12 +1206,20 @@ def _attach_citations(self, response: "AgentResponse") -> None:
 
         from ..rag.attribution import Citation
 
+        answer_text = response.output or ""
         seen: set[tuple[str, str, str]] = set()
         citations: list[Citation] = []
         sources: list[str] = []
         seen_sources: set[str] = set()
         for entry in raw:
-            if entry.get("cited", True):
+            is_cited = entry.get("cited")
+            if is_cited is None:
+                source = entry["source"]
+                if source.startswith(("http://", "https://")):
+                    is_cited = source in answer_text
+                else:
+                    is_cited = True
+            if is_cited:
                 key = (entry["source"], entry["chunk_id"], entry["quote"][:80])
                 if key not in seen:
                     seen.add(key)

effgen/core/complexity_analyzer.py

@@ -212,8 +212,17 @@ def _count_requirements(self, task: str) -> int:
         # Count questions
         num_questions = task.count("?")
 
-        # Count "and" clauses (heuristic for multiple requirements)
-        num_and_clauses = task.lower().count(" and ")
+        # Count "and" clauses (heuristic for multiple requirements), scaled
+        # by task length. A short task with several "and"s describes several
+        # sequential asks ("fetch the data and convert it and email me the
+        # result"); a long passage — e.g. one instruction plus a pasted
+        # reference document such as a paper abstract — accumulates "and"s in
+        # ordinary prose without describing separate requirements. Density,
+        # not the raw count, is the signal, so the contribution tapers off
+        # for longer text instead of scaling unbounded with length.
+        words = task.split()
+        raw_and_clauses = task.lower().count(" and ")
+        num_and_clauses = round(raw_and_clauses * min(1.0, 30 / max(len(words), 1)))
 
         # Count numbered items
         num_numbered = len(re.findall(r'\d+[\.)]\s', task))

effgen/core/decomposition_engine.py

@@ -79,6 +79,7 @@ class DecompositionEngine:
 3. Subtasks should not depend on each other
 4. Cover all aspects of the original task
 5. Be specific about what each subtask should accomplish
+6. required_specialization must be exactly one of: general, research, coding, analysis, synthesis, data, creative
 
 Format your response as JSON:
 {{
@@ -88,7 +89,7 @@ class DecompositionEngine:
             "description": "Detailed description of what to do",
             "expected_output": "What this subtask should produce",
             "estimated_complexity": 5.0,
-            "required_specialization": "research|coding|analysis|synthesis",
+            "required_specialization": "research",
             "priority": "high|medium|low"
         }}
     ],
@@ -107,6 +108,7 @@ class DecompositionEngine:
 3. Show dependencies clearly
 4. Cover all aspects of the original task
 5. Be specific about inputs and outputs
+6. required_specialization must be exactly one of: general, research, coding, analysis, synthesis, data, creative
 
 Format your response as JSON:
 {{
@@ -117,7 +119,7 @@ class DecompositionEngine:
             "depends_on": [],
             "expected_output": "What this subtask should produce",
             "estimated_complexity": 5.0,
-            "required_specialization": "research|coding|analysis|synthesis",
+            "required_specialization": "research",
             "priority": "high|medium|low"
         }}
     ],
@@ -136,6 +138,7 @@ class DecompositionEngine:
 3. Optimize for both speed and dependencies
 4. Cover all aspects of the original task
 5. Be specific about what each subtask should accomplish
+6. required_specialization must be exactly one of: general, research, coding, analysis, synthesis, data, creative
 
 Format your response as JSON:
 {{
@@ -146,7 +149,7 @@ class DecompositionEngine:
             "depends_on": ["st_0"],
             "expected_output": "What this subtask should produce",
             "estimated_complexity": 5.0,
-            "required_specialization": "research|coding|analysis|synthesis",
+            "required_specialization": "research",
             "priority": "high|medium|low"
         }}
     ],
@@ -332,13 +335,28 @@ def _parse_decomposition(self, response: str) -> list[SubTask]:
             }
             priority = priority_map.get(priority_str, TaskPriority.MEDIUM)
 
+            # Normalize the model's free-text specialization to one of the
+            # known values (see SubAgentSpecialization) before it reaches
+            # sub-agent dispatch. A model does not always reproduce the
+            # prompt's example format exactly (e.g. echoing the pipe-joined
+            # placeholder list itself, or inventing a value like
+            # "data_processing"); default to "general" instead of letting an
+            # unrecognized value fail the whole run downstream.
+            valid_specializations = {
+                "general", "research", "coding", "analysis", "synthesis", "data", "creative",
+            }
+            raw_specialization = subtask_data.get("required_specialization")
+            specialization = (raw_specialization or "").strip().lower()
+            if specialization not in valid_specializations:
+                specialization = "general"
+
             # Create SubTask
             subtask = SubTask(
                 id=subtask_data.get("id", f"st_{i+1}"),
                 description=subtask_data["description"],
                 expected_output=subtask_data["expected_output"],
                 estimated_complexity=subtask_data.get("estimated_complexity", 5.0),
-                required_specialization=subtask_data.get("required_specialization"),
+                required_specialization=specialization,
                 depends_on=subtask_data.get("depends_on", []),
                 metadata={
                     "reasoning": data.get("reasoning", ""),

effgen/core/router.py

@@ -168,11 +168,19 @@ def should_use_sub_agents(self, task: str, context: dict[str, Any] | None = None
             strategy = RoutingStrategy.SINGLE_AGENT
 
         # Generate reasoning
-        reasoning = (
-            f"Complexity score: {complexity_score.overall:.2f}. "
-            f"{'Sub-agents recommended' if use_sub_agents else 'Single agent sufficient'} "
-            f"based on complexity threshold of {self.config['complexity_threshold']}."
-        )
+        num_requirements = complexity_score.breakdown.get("num_requirements", 1)
+        if use_sub_agents:
+            reasoning = (
+                f"Complexity score: {complexity_score.overall:.2f}. Sub-agents "
+                f"recommended: an explicit request, a multi-step keyword phrase, "
+                f"or {num_requirements} distinct requirements indicate a compound task."
+            )
+        else:
+            reasoning = (
+                f"Complexity score: {complexity_score.overall:.2f}. Single agent "
+                f"sufficient: no explicit request, multi-step keyword phrase, or "
+                f"multiple distinct requirements — the task is a single unit of work."
+            )
 
         return RoutingDecision(
             use_sub_agents=use_sub_agents,
@@ -288,12 +296,19 @@ def _should_use_sub_agents(self, complexity_score: ComplexityScore, task: str) -
         """
         Decision criteria for using sub-agents.
 
-        Factors considered:
+        Factors considered, in order:
         1. User-explicit request (always honored)
-        2. Complexity score vs threshold
-        3. Presence of keyword triggers
-        4. Task length and structure
-        5. Number of requirements
+        2. Presence of keyword triggers (an explicit multi-step phrase)
+        3. Number of distinct requirements (structural compoundness)
+
+        The raw complexity ``overall`` score is not used as an independent
+        trigger: it is weighted by task length and domain/tool vocabulary
+        breadth as much as by actual task structure, so a single dense
+        instruction — or a short instruction plus a long pasted reference
+        document (a paper abstract, a log excerpt) — can score as "complex"
+        without describing more than one thing to do. Decomposition is
+        warranted by evidence the task is genuinely *compound*, not merely
+        long or information-dense.
 
         Args:
             complexity_score: Calculated complexity score
@@ -307,31 +322,19 @@ def _should_use_sub_agents(self, complexity_score: ComplexityScore, task: str) -
             logger.debug("User explicitly requested sub-agents — honoring request.")
             return True
 
-        # Complexity threshold check
-        threshold = self.config["complexity_threshold"]
-        if complexity_score.overall < threshold:
-            # Check if triggers present that override threshold
-            if not self._check_keyword_triggers(task):
-                return False
-
-        # Keyword trigger check
+        # Keyword trigger check — an explicit multi-step phrase ("research
+        # and analyze", "compare multiple", "comprehensive", ...) is a
+        # reliable compound-task signal regardless of the raw score.
         if self._check_keyword_triggers(task):
             return True
 
-        # Multiple requirements check
+        # Multiple requirements check — several questions, numbered/bulleted
+        # items, semicolon-separated asks, or "and"-joined requests are a
+        # structural signal the task actually has multiple parts.
         num_requirements = complexity_score.breakdown.get("num_requirements", 1)
         if num_requirements >= self.config["min_subtasks"]:
             return True
 
-        # Task length check (very long tasks benefit from decomposition)
-        word_count = complexity_score.breakdown.get("word_count", 0)
-        if word_count > 100:
-            return True
-
-        # High complexity check (override threshold with margin)
-        if complexity_score.overall >= threshold * 0.9:
-            return True
-
         return False
 
     def _check_keyword_triggers(self, task: str) -> bool: