Method generate

src/harness/agentic_loop/plan_generator.py:45–105

Generate a YAML plan for the given task. Args: task_id: Identifier used for the output filename (e.g. "task_001"). task_description: What the task should accomplish. context_summary: Optional summary of relevant context from the orchestrat

(
        self,
        task_id: str,
        task_description: str,
        context_summary: Optional[str] = None,
        hints: Optional[str] = None,
        recommended_tools: Optional[list[str]] = None,
    )

        self.plans_dir.mkdir(parents=True, exist_ok=True)

    def generate(
        self,
        task_id: str,
        task_description: str,
        context_summary: Optional[str] = None,
        hints: Optional[str] = None,
        recommended_tools: Optional[list[str]] = None,
    ) -> Path:
        """Generate a YAML plan for the given task.

        Args:
            task_id: Identifier used for the output filename (e.g. "task_001").
            task_description: What the task should accomplish.
            context_summary: Optional summary of relevant context from the
                orchestrator's conversation so far.
            hints: Optional hints for the plan generator (e.g. preferred
                structure, tool preferences).
            recommended_tools: Optional list of MCP tool names pre-selected
                by the orchestrator. When provided, the plan generator is
                instructed to use ONLY these tools. The orchestrator (more
                capable model with full tool schemas) makes the tool selection
                decision; the plan generator focuses on workflow structure.

        Returns:
            Path to the generated YAML file.

        Raises:
            ValueError: If YAML generation fails after retry.
        """
        user_message = self._build_user_message(
            task_description,
            context_summary,
            hints,
            recommended_tools,
        )

        # First attempt
        raw_response = self._call_llm(user_message)
        yaml_str = self._extract_yaml(raw_response)

        parse_error = self._try_parse(yaml_str)
        if parse_error:
            # Retry with error feedback
            retry_message = (
                f"The YAML you produced has a syntax error:\n{parse_error}\n\n"
                f"Original YAML:\n```yaml\n{yaml_str}\n```\n\n"
                "Please fix the error and return the corrected YAML "
                "inside a ```yaml fenced code block."
            )
            raw_response = self._call_llm(retry_message)
            yaml_str = self._extract_yaml(raw_response)
            parse_error = self._try_parse(yaml_str)
            if parse_error:
                raise ValueError(
                    f"Plan generation failed after retry. YAML error: {parse_error}"
                )

        # Write to disk