many design improvements

9 files changed, 145 insertions, 57 deletions

diff --git a/continuedev/src/continuedev/core/autopilot.py b/continuedev/src/continuedev/core/autopilot.py
index db06c975..d55e521b 100644
--- a/continuedev/src/continuedev/core/autopilot.py
+++ b/continuedev/src/continuedev/core/autopilot.py
@@ -1,13 +1,13 @@
 import traceback
 import time
-from typing import Callable, Coroutine, List
+from typing import Any, Callable, Coroutine, Dict, List
 from ..models.filesystem_edit import FileEditWithFullContents
 from ..libs.llm import LLM
 from .observation import Observation, InternalErrorObservation
 from ..server.ide_protocol import AbstractIdeProtocolServer
 from ..libs.util.queue import AsyncSubscriptionQueue
 from ..models.main import ContinueBaseModel
-from .main import Policy, History, FullState, Step, HistoryNode
+from .main import Context, ContinueCustomException, Policy, History, FullState, Step, HistoryNode
 from ..steps.core.core import ReversibleStep, ManualEditStep, UserInputStep
 from ..libs.util.telemetry import capture_event
 from .sdk import ContinueSDK
@@ -18,6 +18,7 @@ class Autopilot(ContinueBaseModel):
     policy: Policy
     ide: AbstractIdeProtocolServer
     history: History = History.from_empty()
+    context: Context = Context()
     _on_update_callbacks: List[Callable[[FullState], None]] = []
 
     _active: bool = False
@@ -26,6 +27,10 @@ class Autopilot(ContinueBaseModel):
 
     _user_input_queue = AsyncSubscriptionQueue()
 
+    @property
+    def continue_sdk(self) -> ContinueSDK:
+        return ContinueSDK(self)
+
     class Config:
         arbitrary_types_allowed = True
 
@@ -60,7 +65,7 @@ class Autopilot(ContinueBaseModel):
                 current_step = self.history.get_current().step
                 self.history.step_back()
                 if issubclass(current_step.__class__, ReversibleStep):
-                    await current_step.reverse(ContinueSDK(self))
+                    await current_step.reverse(self.continue_sdk)
 
                 await self.update_subscribers()
         except Exception as e:
@@ -105,7 +110,16 @@ class Autopilot(ContinueBaseModel):
         self._step_depth += 1
 
         try:
-            observation = await step(ContinueSDK(self))
+            observation = await step(self.continue_sdk)
+        except ContinueCustomException as e:
+            # Attach an InternalErrorObservation to the step and unhide it.
+            error_string = e.message
+            print(
+                f"\n{error_string}\n{e}")
+
+            observation = InternalErrorObservation(
+                error=error_string)
+            step.hide = False
         except Exception as e:
             # Attach an InternalErrorObservation to the step and unhide it.
             error_string = '\n\n'.join(
@@ -125,11 +139,13 @@ class Autopilot(ContinueBaseModel):
         await self.update_subscribers()
 
         # Update its description
-        async def update_description():
-            step._set_description(await step.describe(ContinueSDK(self).models))
-            # Update subscribers with new description
-            await self.update_subscribers()
-        asyncio.create_task(update_description())
+        if step.description is None:
+            async def update_description():
+                step.description = await step.describe(self.continue_sdk.models)
+                # Update subscribers with new description
+                await self.update_subscribers()
+
+            asyncio.create_task(update_description())
 
         return observation
 
diff --git a/continuedev/src/continuedev/core/main.py b/continuedev/src/continuedev/core/main.py
index b2b97bae..17b30e96 100644
--- a/continuedev/src/continuedev/core/main.py
+++ b/continuedev/src/continuedev/core/main.py
@@ -105,7 +105,7 @@ class Policy(ContinueBaseModel):
 class Step(ContinueBaseModel):
     name: str = None
     hide: bool = False
-    _description: Union[str, None] = None
+    description: Union[str, None] = None
 
     system_message: Union[str, None] = None
 
@@ -113,17 +113,14 @@ class Step(ContinueBaseModel):
         copy_on_model_validation = False
 
     async def describe(self, models: Models) -> Coroutine[str, None, None]:
-        if self._description is not None:
-            return self._description
+        if self.description is not None:
+            return self.description
         return "Running step: " + self.name
 
-    def _set_description(self, description: str):
-        self._description = description
-
     def dict(self, *args, **kwargs):
         d = super().dict(*args, **kwargs)
-        if self._description is not None:
-            d["description"] = self._description
+        if self.description is not None:
+            d["description"] = self.description
         else:
             d["description"] = "`Description loading...`"
         return d
@@ -173,4 +170,21 @@ class Validator(Step):
         raise NotImplementedError
 
 
+class Context:
+    key_value: Dict[str, str] = {}
+
+    def set(self, key: str, value: str):
+        self.key_value[key] = value
+
+    def get(self, key: str) -> str:
+        return self.key_value[key]
+
+
+class ContinueCustomException(Exception):
+    message: str
+
+    def __init__(self, message: str):
+        self.message = message
+
+
 HistoryNode.update_forward_refs()
diff --git a/continuedev/src/continuedev/core/policy.py b/continuedev/src/continuedev/core/policy.py
index e65b6c9d..4934497d 100644
--- a/continuedev/src/continuedev/core/policy.py
+++ b/continuedev/src/continuedev/core/policy.py
@@ -17,9 +17,10 @@ class DemoPolicy(Policy):
     def next(self, history: History) -> Step:
         # At the very start, run initial Steps spcecified in the config
         if history.get_current() is None:
-            return (MessageStep(message="Welcome to Continue!") >>
-                    SetupContinueWorkspaceStep() >>
-                    CreateCodebaseIndexChroma() >>
+            return (
+                    MessageStep(name="Welcome to Continue!", message="") >>
+                    # SetupContinueWorkspaceStep() >>
+                    # CreateCodebaseIndexChroma() >>
                     StepsOnStartupStep())
 
         observation = history.get_current().observation
diff --git a/continuedev/src/continuedev/core/sdk.py b/continuedev/src/continuedev/core/sdk.py
index 8317a3d1..690949f1 100644
--- a/continuedev/src/continuedev/core/sdk.py
+++ b/continuedev/src/continuedev/core/sdk.py
@@ -11,7 +11,7 @@ from ..libs.llm.hf_inference_api import HuggingFaceInferenceAPI
 from ..libs.llm.openai import OpenAI
 from .observation import Observation
 from ..server.ide_protocol import AbstractIdeProtocolServer
-from .main import History, Step
+from .main import Context, ContinueCustomException, History, Step
 from ..steps.core.core import *
 
 
@@ -44,6 +44,7 @@ class ContinueSDK(AbstractContinueSDK):
     ide: AbstractIdeProtocolServer
     steps: ContinueSDKSteps
     models: Models
+    context: Context
     __autopilot: Autopilot
 
     def __init__(self, autopilot: Autopilot):
@@ -51,6 +52,7 @@ class ContinueSDK(AbstractContinueSDK):
         self.__autopilot = autopilot
         self.steps = ContinueSDKSteps(self)
         self.models = Models(self)
+        self.context = autopilot.context
 
     @property
     def history(self) -> History:
@@ -64,8 +66,8 @@ class ContinueSDK(AbstractContinueSDK):
     async def run_step(self, step: Step) -> Coroutine[Observation, None, None]:
         return await self.__autopilot._run_singular_step(step)
 
-    async def apply_filesystem_edit(self, edit: FileSystemEdit):
-        return await self.run_step(FileSystemEditStep(edit=edit))
+    async def apply_filesystem_edit(self, edit: FileSystemEdit, name: str = None, description: str = None):
+        return await self.run_step(FileSystemEditStep(edit=edit, description=description, **({'name': name} if name else {})))
 
     async def wait_for_user_input(self) -> str:
         return await self.__autopilot.wait_for_user_input()
@@ -73,18 +75,20 @@ class ContinueSDK(AbstractContinueSDK):
     async def wait_for_user_confirmation(self, prompt: str):
         return await self.run_step(WaitForUserConfirmationStep(prompt=prompt))
 
-    async def run(self, commands: Union[List[str], str], cwd: str = None):
+    async def run(self, commands: Union[List[str], str], cwd: str = None, name: str = None, description: str = None):
         commands = commands if isinstance(commands, List) else [commands]
-        return await self.run_step(ShellCommandsStep(cmds=commands, cwd=cwd))
+        return await self.run_step(ShellCommandsStep(cmds=commands, cwd=cwd, description=description, **({'name': name} if name else {})))
 
-    async def edit_file(self, filename: str, prompt: str):
+    async def edit_file(self, filename: str, prompt: str, name: str = None, description: str = None):
         filepath = await self._ensure_absolute_path(filename)
 
         await self.ide.setFileOpen(filepath)
         contents = await self.ide.readFile(filepath)
         await self.run_step(EditCodeStep(
             range_in_files=[RangeInFile.from_entire_file(filepath, contents)],
-            prompt=f'Here is the code before:\n\n{{code}}\n\nHere is the user request:\n\n{prompt}\n\nHere is the code edited to perfectly solve the user request:\n\n'
+            prompt=f'Here is the code before:\n\n{{code}}\n\nHere is the user request:\n\n{prompt}\n\nHere is the code edited to perfectly solve the user request:\n\n',
+            description=description,
+            **({'name': name} if name else {})
         ))
 
     async def append_to_file(self, filename: str, content: str):
@@ -126,3 +130,6 @@ class ContinueSDK(AbstractContinueSDK):
     def set_loading_message(self, message: str):
         # self.__autopilot.set_loading_message(message)
         raise NotImplementedError()
+
+    def raise_exception(self, message: str):
+        raise ContinueCustomException(message)
diff --git a/continuedev/src/continuedev/recipes/CreatePipelineRecipe/main.py b/continuedev/src/continuedev/recipes/CreatePipelineRecipe/main.py
index 55c25da4..1206db0e 100644
--- a/continuedev/src/continuedev/recipes/CreatePipelineRecipe/main.py
+++ b/continuedev/src/continuedev/recipes/CreatePipelineRecipe/main.py
@@ -12,7 +12,7 @@ class CreatePipelineRecipe(Step):
 
     async def run(self, sdk: ContinueSDK):
         text_observation = await sdk.run_step(
-            MessageStep(message=dedent("""\
+            MessageStep(name="Building your first dlt pipeline", message=dedent("""\
                 This recipe will walk you through the process of creating a dlt pipeline for your chosen data source. With the help of Continue, you will:
                 - Create a Python virtual environment with dlt installed
                 - Run `dlt init` to generate a pipeline template
diff --git a/continuedev/src/continuedev/recipes/CreatePipelineRecipe/steps.py b/continuedev/src/continuedev/recipes/CreatePipelineRecipe/steps.py
index ef5e3b43..3c8277c0 100644
--- a/continuedev/src/continuedev/recipes/CreatePipelineRecipe/steps.py
+++ b/continuedev/src/continuedev/recipes/CreatePipelineRecipe/steps.py
@@ -1,8 +1,12 @@
+import os
+import subprocess
 from textwrap import dedent
 
+from ...models.main import Range
+from ...models.filesystem import RangeInFile
 from ...steps.main import MessageStep
 from ...core.sdk import Models
-from ...core.observation import DictObservation
+from ...core.observation import DictObservation, InternalErrorObservation
 from ...models.filesystem_edit import AddFile
 from ...core.main import Step
 from ...core.sdk import ContinueSDK
@@ -33,45 +37,69 @@ class SetupPipelineStep(Step):
             f'dlt init {source_name} duckdb',
             'Y',
             'pip install -r requirements.txt'
-        ])
-
+        ], description=dedent(f"""\
+            Running the following commands:
+            - `python3 -m venv env`: Create a Python virtual environment
+            - `source env/bin/activate`: Activate the virtual environment
+            - `pip install dlt`: Install dlt
+            - `dlt init {source_name} duckdb`: Create a new dlt pipeline called {source_name} that loads data into a local DuckDB instance
+            - `pip install -r requirements.txt`: Install the Python dependencies for the pipeline"""))
+
+        await sdk.wait_for_user_confirmation("Wait for the commands to finish running, then press `Continue`")
         # editing the resource function to call the requested API
+        await sdk.ide.highlightCode(RangeInFile(filepath=os.path.join(await sdk.ide.getWorkspaceDirectory(), filename), range=Range.from_shorthand(15, 0, 30, 0)), "#00ff0022")
+
         await sdk.edit_file(
             filename=filename,
-            prompt=f'Edit the resource function to call the API described by this: {self.api_description}'
+            prompt=f'Edit the resource function to call the API described by this: {self.api_description}',
+            name="Edit the resource function to call the API"
         )
 
         # wait for user to put API key in secrets.toml
         await sdk.ide.setFileOpen(await sdk.ide.getWorkspaceDirectory() + "/.dlt/secrets.toml")
         await sdk.wait_for_user_confirmation("If this service requires an API key, please add it to the `secrets.toml` file and then press `Continue`")
-        return DictObservation(values={"source_name": source_name})
+
+        sdk.context.set("source_name", source_name)
 
 
 class ValidatePipelineStep(Step):
     hide: bool = True
 
     async def run(self, sdk: ContinueSDK):
-        source_name = sdk.history.last_observation().values["source_name"]
+        workspace_dir = await sdk.ide.getWorkspaceDirectory()
+        source_name = sdk.context.get("source_name")
         filename = f'{source_name}.py'
 
-        await sdk.run_step(MessageStep(message=dedent("""\
-                This step will validate that your dlt pipeline is working as expected:
-                - Test that the API call works
-                - Load the data into a local DuckDB instance
-                - Write a query to view the data
-                """)))
+        # await sdk.run_step(MessageStep(name="Validate the pipeline", message=dedent("""\
+        #         Next, we will validate that your dlt pipeline is working as expected:
+        #         - Test that the API call works
+        #         - Load the data into a local DuckDB instance
+        #         - Write a query to view the data
+        #         """)))
 
         # test that the API call works
-        await sdk.run(f'python3 {filename}')
+
+        p = subprocess.run(
+            ['python3', f'{filename}'], capture_output=True, text=True, cwd=workspace_dir)
+        err = p.stderr
+
+        # If it fails, return the error
+        if err is not None and err != "":
+            sdk.raise_exception(
+                f"Error while running pipeline. Fix the resource function in {filename} and rerun this step: \n\n" + err)
+
+        await sdk.run(f'python3 {filename}', name="Test the pipeline", description=f"Running python3 {filename} to test loading data from the API")
 
         # remove exit() from the main main function
         await sdk.edit_file(
             filename=filename,
-            prompt='Remove exit() from the main function'
+            prompt='Remove exit() from the main function',
+            name="Remove early exit() from main function",
+            description="Remove the `exit()` call from the main function in the pipeline file so that the data is loaded into DuckDB"
         )
 
         # load the data into the DuckDB instance
-        await sdk.run(f'python3 {filename}')
+        await sdk.run(f'python3 {filename}', name="Load data into DuckDB", description=f"Running python3 {filename} to load data into DuckDB")
 
         table_name = f"{source_name}.{source_name}_resource"
         tables_query_code = dedent(f'''\
@@ -89,5 +117,5 @@ class ValidatePipelineStep(Step):
         ''')
 
         query_filename = (await sdk.ide.getWorkspaceDirectory()) + "/query.py"
-        await sdk.apply_filesystem_edit(AddFile(filepath=query_filename, content=tables_query_code))
-        await sdk.run('env/bin/python3 query.py')
+        await sdk.apply_filesystem_edit(AddFile(filepath=query_filename, content=tables_query_code), name="Add query.py file", description="Adding a file called `query.py` to the workspace that will run a test query on the DuckDB instance")
+        await sdk.run('env/bin/python3 query.py', name="Run test query", description="Running `env/bin/python3 query.py` to test that the data was loaded into DuckDB as expected")
diff --git a/continuedev/src/continuedev/server/ide.py b/continuedev/src/continuedev/server/ide.py
index eec5b607..073e1dba 100644
--- a/continuedev/src/continuedev/server/ide.py
+++ b/continuedev/src/continuedev/server/ide.py
@@ -133,6 +133,17 @@ class IdeProtocolServer(AbstractIdeProtocolServer):
             "sessionId": session_id
         })
 
+    async def highlightCode(self, range_in_file: RangeInFile, color: str):
+        await self._send_json("highlightCode", {
+            "rangeInFile": range_in_file.dict(),
+            "color": color
+        })
+
+    async def runCommand(self, command: str):
+        await self._send_json("runCommand", {
+            "command": command
+        })
+
     async def showSuggestionsAndWait(self, suggestions: List[FileEdit]) -> bool:
         ids = [str(uuid.uuid4()) for _ in suggestions]
         for i in range(len(suggestions)):
diff --git a/continuedev/src/continuedev/server/ide_protocol.py b/continuedev/src/continuedev/server/ide_protocol.py
index 8f155415..f42de68f 100644
--- a/continuedev/src/continuedev/server/ide_protocol.py
+++ b/continuedev/src/continuedev/server/ide_protocol.py
@@ -82,3 +82,11 @@ class AbstractIdeProtocolServer(ABC):
     @abstractmethod
     async def getUserSecret(self, key: str):
         """Get a user secret"""
+
+    @abstractmethod
+    async def highlightCode(self, range_in_file: RangeInFile, color: str):
+        """Highlight code"""
+
+    @abstractmethod
+    async def runCommand(self, command: str):
+        """Run a command"""
diff --git a/continuedev/src/continuedev/steps/core/core.py b/continuedev/src/continuedev/steps/core/core.py
index ad468595..99786b00 100644
--- a/continuedev/src/continuedev/steps/core/core.py
+++ b/continuedev/src/continuedev/steps/core/core.py
@@ -55,18 +55,21 @@ class ShellCommandsStep(Step):
     async def run(self, sdk: ContinueSDK) -> Coroutine[Observation, None, None]:
         cwd = await sdk.ide.getWorkspaceDirectory() if self.cwd is None else self.cwd
 
-        process = subprocess.Popen(
-            '/bin/bash', stdin=subprocess.PIPE, stdout=subprocess.PIPE, cwd=cwd)
+        for cmd in self.cmds:
+            await sdk.ide.runCommand(cmd)
 
-        stdin_input = "\n".join(self.cmds)
-        out, err = process.communicate(stdin_input.encode())
+        # process = subprocess.Popen(
+        #     '/bin/bash', stdin=subprocess.PIPE, stdout=subprocess.PIPE, cwd=cwd)
 
-        # If it fails, return the error
-        if err is not None and err != "":
-            self._err_text = err
-            return TextObservation(text=err)
+        # stdin_input = "\n".join(self.cmds)
+        # out, err = process.communicate(stdin_input.encode())
 
-        return None
+        # # If it fails, return the error
+        # if err is not None and err != "":
+        #     self._err_text = err
+        #     return TextObservation(text=err)
+
+        # return None
 
 
 class EditCodeStep(Step):
@@ -197,10 +200,10 @@ class WaitForUserInputStep(Step):
         if self._response is None:
             return self.prompt
         else:
-            return self.prompt + "\n\n" + self._response
+            return f"{self.prompt}\n\n`{self._response}`"
 
     async def run(self, sdk: ContinueSDK) -> Coroutine[Observation, None, None]:
-        self._description = self.prompt
+        self.description = self.prompt
         resp = await sdk.wait_for_user_input()
         self._response = resp
         return TextObservation(text=resp)
@@ -214,7 +217,7 @@ class WaitForUserConfirmationStep(Step):
         return self.prompt
 
     async def run(self, sdk: ContinueSDK) -> Coroutine[Observation, None, None]:
-        self._description = self.prompt
+        self.description = self.prompt
         resp = await sdk.wait_for_user_input()
         self.hide = True
         return TextObservation(text=resp)