feat: Add task list tracking system with LLM tools and Progress tab UI

agentic_connector_builder_webapp/agentic_connector_builder_webapp.py

@@ -21,6 +21,7 @@
     requirements_tab_content,
     save_publish_tab_content,
 )
+from .task_list import TaskList, create_default_connector_task_list
 
 env_file = Path.cwd() / ".env"
 if env_file.exists():
@@ -44,6 +45,8 @@ class ConnectorBuilderState(rx.State):
     functional_requirements: str = ""
     test_list: str = ""
 
+    task_list_json: str = ""
+
     settings_modal_open: bool = False
     openai_api_key_input: str = ""
 
@@ -84,6 +87,18 @@ def get_content_length(self) -> int:
         """Get the content length."""
         return len(self.yaml_content)
 
+    def get_task_list(self) -> TaskList:
+        """Get the task list object from JSON."""
+        if not self.task_list_json:
+            default_task_list = create_default_connector_task_list()
+            self.task_list_json = default_task_list.model_dump_json()
+            return default_task_list
+        return TaskList.model_validate_json(self.task_list_json)
+
+    def update_task_list(self, task_list: TaskList):
+        """Update the task list from a TaskList object."""
+        self.task_list_json = task_list.model_dump_json()
+
     def update_yaml_content(self, content: str):
         """Update the YAML content when editor changes."""
         self.yaml_content = content
@@ -182,6 +197,191 @@ def has_env_api_key(self) -> bool:
         """Check if an API key is available from environment variables (not UI input)."""
         return bool(os.environ.get("OPENAI_API_KEY", ""))
 
+    @rx.var
+    def has_task_list(self) -> bool:
+        """Check if a task list has been initialized."""
+        return bool(self.task_list_json)
+
+    @rx.var
+    def connector_tasks_view(self) -> list[dict[str, Any]]:
+        """Get connector tasks as JSON-serializable view data."""
+        if not self.task_list_json:
+            return []
+        try:
+            task_list = TaskList.model_validate_json(self.task_list_json)
+            result = []
+            for task in task_list.tasks:
+                if task.task_type == "stream":
+                    continue
+                status = task.status.value
+                icon, color = {
+                    "not_started": ("○", "gray.400"),
+                    "in_progress": ("◐", "blue.400"),
+                    "completed": ("●", "green.400"),
+                    "failed": ("✗", "red.400"),
+                }.get(status, ("?", "gray.400"))
+                result.append(
+                    {
+                        "id": task.id,
+                        "title": task.title,
+                        "details": task.details or "",
+                        "status": status,
+                        "icon": icon,
+                        "color": color,
+                    }
+                )
+            return result
+        except Exception:
+            return []
+
+    @rx.var
+    def stream_tasks_view(self) -> list[dict[str, Any]]:
+        """Get stream tasks as JSON-serializable view data, sorted by stream name."""
+        if not self.task_list_json:
+            return []
+        try:
+            task_list = TaskList.model_validate_json(self.task_list_json)
+            result = []
+            for task in task_list.tasks:
+                if task.task_type != "stream":
+                    continue
+                status = task.status.value
+                icon, color = {
+                    "not_started": ("○", "gray.400"),
+                    "in_progress": ("◐", "blue.400"),
+                    "completed": ("●", "green.400"),
+                    "failed": ("✗", "red.400"),
+                }.get(status, ("?", "gray.400"))
+                result.append(
+                    {
+                        "id": task.id,
+                        "stream_name": task.stream_name,
+                        "title": task.title,
+                        "details": task.details or "",
+                        "status": status,
+                        "icon": icon,
+                        "color": color,
+                    }
+                )
+            result.sort(key=lambda r: (r["stream_name"], r["title"]))
+            return result
+        except Exception:
+            return []
+
+    @rx.var
+    def task_list_header(self) -> dict[str, Any]:
+        """Get task list header information."""
+        if not self.task_list_json:
+            return {
+                "name": "",
+                "description": "",
+                "summary": {"total": 0, "completed": 0, "in_progress": 0, "failed": 0},
+            }
+        try:
+            task_list = TaskList.model_validate_json(self.task_list_json)
+            return {
+                "name": task_list.name,
+                "description": task_list.description,
+                "summary": task_list.get_summary(),
+            }
+        except Exception:
+            return {
+                "name": "Task List",
+                "description": "Error parsing task list",
+                "summary": {"total": 0, "completed": 0, "in_progress": 0, "failed": 0},
+            }
+
+    @rx.var
+    def has_connector_tasks(self) -> bool:
+        """Check if there are any connector tasks."""
+        return len(self.connector_tasks_view) > 0
+
+    @rx.var
+    def has_stream_tasks(self) -> bool:
+        """Check if there are any stream tasks."""
+        return len(self.stream_tasks_view) > 0
+
+    @rx.var
+    def task_list_name(self) -> str:
+        """Get the task list name."""
+        if not self.task_list_json:
+            return ""
+        try:
+            task_list = TaskList.model_validate_json(self.task_list_json)
+            return task_list.name
+        except Exception:
+            return "Task List"
+
+    @rx.var
+    def task_list_description(self) -> str:
+        """Get the task list description."""
+        if not self.task_list_json:
+            return ""
+        try:
+            task_list = TaskList.model_validate_json(self.task_list_json)
+            return task_list.description
+        except Exception:
+            return "Error parsing task list"
+
+    @rx.var
+    def task_total_count(self) -> int:
+        """Get total number of tasks."""
+        if not self.task_list_json:
+            return 0
+        try:
+            task_list = TaskList.model_validate_json(self.task_list_json)
+            return task_list.get_summary()["total"]
+        except Exception:
+            return 0
+
+    @rx.var
+    def task_completed_count(self) -> int:
+        """Get number of completed tasks."""
+        if not self.task_list_json:
+            return 0
+        try:
+            task_list = TaskList.model_validate_json(self.task_list_json)
+            return task_list.get_summary()["completed"]
+        except Exception:
+            return 0
+
+    @rx.var
+    def task_in_progress_count(self) -> int:
+        """Get number of in-progress tasks."""
+        if not self.task_list_json:
+            return 0
+        try:
+            task_list = TaskList.model_validate_json(self.task_list_json)
+            return task_list.get_summary()["in_progress"]
+        except Exception:
+            return 0
+
+    @rx.var
+    def task_failed_count(self) -> int:
+        """Get number of failed tasks."""
+        if not self.task_list_json:
+            return 0
+        try:
+            task_list = TaskList.model_validate_json(self.task_list_json)
+            return task_list.get_summary()["failed"]
+        except Exception:
+            return 0
+
+    @rx.var
+    def completed_of_total_text(self) -> str:
+        """Get formatted completed/total text."""
+        return f"{self.task_completed_count}/{self.task_total_count} Completed"
+
+    @rx.var
+    def in_progress_text(self) -> str:
+        """Get formatted in-progress text."""
+        return f"{self.task_in_progress_count} In Progress"
+
+    @rx.var
+    def failed_text(self) -> str:
+        """Get formatted failed text."""
+        return f"{self.task_failed_count} Failed"
+
     _cached_agent: Any = None
     _cached_api_key: str | None = None
     _agent_started: bool = False
@@ -235,13 +435,16 @@ async def send_message(self):
         self.current_streaming_message = ""
         yield
 
+        self.get_task_list()
+
         session_deps = SessionDeps(
             yaml_content=self.yaml_content,
             connector_name=self.connector_name,
             source_api_name=self.source_api_name,
             documentation_urls=self.documentation_urls,
             functional_requirements=self.functional_requirements,
             test_list=self.test_list,
+            task_list_json=self.task_list_json,
             set_source_api_name=self.set_source_api_name,
             set_connector_name=self.set_connector_name,
         )
@@ -282,6 +485,10 @@ async def send_message(self):
                 self.yaml_content = session_deps.yaml_content
                 yield
 
+            if session_deps.task_list_json != self.task_list_json:
+                self.task_list_json = session_deps.task_list_json
+                yield
+
         except Exception as e:
             self.chat_messages.append(
                 {