Cl 1679 release 2025 09 rpa crud workflow executions

src/askui/chat/api/app.py

@@ -22,6 +22,7 @@
 from askui.chat.api.messages.router import router as messages_router
 from askui.chat.api.runs.router import router as runs_router
 from askui.chat.api.threads.router import router as threads_router
+from askui.chat.api.workflow_executions.router import router as executions_router
 from askui.chat.api.workflows.router import router as workflows_router
 from askui.utils.api_utils import (
     ConflictError,
@@ -54,6 +55,7 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:  # noqa: ARG001
 # Include routers
 v1_router = APIRouter(prefix="/v1")
 v1_router.include_router(assistants_router)
+v1_router.include_router(executions_router)
 v1_router.include_router(threads_router)
 v1_router.include_router(messages_router)
 v1_router.include_router(runs_router)

src/askui/chat/api/assistants/seeds.py

@@ -19,6 +19,10 @@
         "list_displays",
         "set_active_display",
         "retrieve_active_display",
+        "create_workflow_v1_workflows_post",
+        "list_workflows_v1_workflows_get",
+        "modify_workflow_v1_workflows",
+        "create_workflow_execution_v1_workflow_executions",
     ],
 )
 

src/askui/chat/api/models.py

@@ -6,11 +6,14 @@
 from askui.utils.id_utils import IdField
 
 AssistantId = Annotated[str, IdField("asst")]
+WorkflowExecutionId = Annotated[str, IdField("wfexec")]
 McpConfigId = Annotated[str, IdField("mcpcnf")]
 FileId = Annotated[str, IdField("file")]
 MessageId = Annotated[str, IdField("msg")]
 RunId = Annotated[str, IdField("run")]
 ThreadId = Annotated[str, IdField("thread")]
+WorkflowId = Annotated[str, IdField("wf")]
+
 WorkspaceId = UUID4
 
 

src/askui/chat/api/workflow_executions/__init__.py

@@ -0,0 +1 @@
+"""Execution models and services for the chat API."""

src/askui/chat/api/workflow_executions/dependencies.py

@@ -0,0 +1,21 @@
+from fastapi import Depends
+
+from askui.chat.api.dependencies import SettingsDep
+from askui.chat.api.settings import Settings
+from askui.chat.api.threads.dependencies import ThreadFacadeDep
+from askui.chat.api.threads.facade import ThreadFacade
+from askui.chat.api.workflow_executions.service import ExecutionService
+from askui.chat.api.workflows.dependencies import WorkflowServiceDep
+from askui.chat.api.workflows.service import WorkflowService
+
+
+def get_execution_service(
+    settings: Settings = SettingsDep,
+    workflow_service: WorkflowService = WorkflowServiceDep,
+    thread_facade: ThreadFacade = ThreadFacadeDep,
+) -> ExecutionService:
+    """Get ExecutionService instance."""
+    return ExecutionService(settings.data_dir, workflow_service, thread_facade)
+
+
+ExecutionServiceDep = Depends(get_execution_service)

src/askui/chat/api/workflow_executions/models.py

@@ -0,0 +1,63 @@
+import datetime
+from typing import Literal
+
+from pydantic import BaseModel
+
+from askui.chat.api.models import (
+    RunId,
+    ThreadId,
+    WorkflowExecutionId,
+    WorkspaceId,
+    WorkspaceResource,
+)
+from askui.chat.api.workflows.models import WorkflowId
+from askui.utils.datetime_utils import now
+from askui.utils.id_utils import generate_time_ordered_id
+
+
+class WorkflowExecutionCreateParams(BaseModel):
+    """
+    Parameters for creating a workflow execution via API.
+    """
+
+    workflow_id: WorkflowId
+
+
+class WorkflowExecution(WorkspaceResource):
+    """
+    A workflow execution resource in the chat API.
+
+    Args:
+        id (WorkflowExecutionId): The id of the execution. Must start with the 'exec_' prefix and be
+            followed by one or more alphanumerical characters.
+        object (Literal['execution']): The object type, always 'execution'.
+        created_at (datetime.datetime): The creation time as a datetime.
+        workflow (WorkflowId): The id of the workflow being executed. Must start with the 'wf_' prefix.
+        thread (ThreadId): The id of the thread this execution is associated with. Must start with the 'thread_' prefix.
+        status (ExecutionStatus): The current status of the workflow execution.
+        workspace_id (WorkspaceId | None, optional): The workspace this execution belongs to.
+    """
+
+    id: WorkflowExecutionId
+    object: Literal["workflow_execution"] = "workflow_execution"
+    created_at: datetime.datetime
+    workflow_id: WorkflowId
+    thread_id: ThreadId
+    run_id: RunId
+
+    @classmethod
+    def create(
+        cls,
+        workspace_id: WorkspaceId,
+        workflow_id: WorkflowId,
+        run_id: RunId,
+        thread_id: ThreadId,
+    ) -> "WorkflowExecution":
+        return cls(
+            id=generate_time_ordered_id("wfexec"),
+            created_at=now(),
+            workspace_id=workspace_id,
+            run_id=run_id,
+            thread_id=thread_id,
+            workflow_id=workflow_id,
+        )

src/askui/chat/api/workflow_executions/router.py

@@ -0,0 +1,69 @@
+from collections.abc import AsyncGenerator
+from typing import Annotated
+
+from fastapi import APIRouter, BackgroundTasks, Header, Query, Response, status
+from fastapi.responses import JSONResponse, StreamingResponse
+from pydantic import BaseModel
+
+from askui.chat.api.dependencies import ListQueryDep
+from askui.chat.api.models import ThreadId, WorkspaceId
+from askui.chat.api.workflow_executions.dependencies import ExecutionServiceDep
+from askui.chat.api.workflow_executions.models import (
+    WorkflowExecution,
+    WorkflowExecutionCreateParams,
+    WorkflowExecutionId,
+)
+from askui.chat.api.workflow_executions.service import ExecutionService
+from askui.chat.api.workflows.models import WorkflowId
+from askui.utils.api_utils import ListQuery, ListResponse
+
+router = APIRouter(prefix="/workflow-executions", tags=["workflow-executions"])
+
+
+@router.get("/")
+def list_workflow_executions(
+    askui_workspace: Annotated[WorkspaceId, Header()],
+    query: ListQuery = ListQueryDep,
+    workflow_id: Annotated[WorkflowId | None, Query()] = None,
+    thread_id: Annotated[ThreadId | None, Query()] = None,
+    execution_service: ExecutionService = ExecutionServiceDep,
+) -> ListResponse[WorkflowExecution]:
+    """List executions with optional filtering by workflow and/or thread."""
+    return execution_service.list_(
+        workspace_id=askui_workspace,
+        query=query,
+        workflow_id=workflow_id,
+        thread_id=thread_id,
+    )
+
+
+@router.post("/")
+async def create_workflow_execution(
+    askui_workspace: Annotated[WorkspaceId, Header()],
+    params: WorkflowExecutionCreateParams,
+    background_tasks: BackgroundTasks,
+    execution_service: ExecutionService = ExecutionServiceDep,
+) -> WorkflowExecution:
+    """Create a new workflow execution."""
+    execution, async_generator = await execution_service.create(
+        workspace_id=askui_workspace, params=params
+    )
+
+    async def _run_async_generator() -> None:
+        async for _ in async_generator:
+            pass
+
+    background_tasks.add_task(_run_async_generator)
+    return execution
+
+
+@router.get("/{execution_id}")
+def retrieve_workflow_execution(
+    askui_workspace: Annotated[WorkspaceId, Header()],
+    execution_id: WorkflowExecutionId,
+    execution_service: ExecutionService = ExecutionServiceDep,
+) -> WorkflowExecution:
+    """Retrieve a specific execution by ID."""
+    return execution_service.retrieve(
+        workspace_id=askui_workspace, execution_id=execution_id
+    )

src/askui/chat/api/workflow_executions/service.py

@@ -0,0 +1,146 @@
+from collections.abc import AsyncGenerator
+from pathlib import Path
+from typing import Callable
+
+from askui.chat.api.messages.models import MessageCreateParams
+from askui.chat.api.models import ThreadId, WorkspaceId
+from askui.chat.api.runs.models import ThreadAndRunCreateParams
+from askui.chat.api.runs.runner.events.events import Events
+from askui.chat.api.threads.facade import ThreadFacade
+from askui.chat.api.threads.models import ThreadCreateParams
+from askui.chat.api.utils import build_workspace_filter_fn
+from askui.chat.api.workflow_executions.models import (
+    WorkflowExecution,
+    WorkflowExecutionCreateParams,
+    WorkflowExecutionId,
+)
+from askui.chat.api.workflows.models import WorkflowId
+from askui.chat.api.workflows.service import WorkflowService
+from askui.utils.api_utils import (
+    ConflictError,
+    ListQuery,
+    ListResponse,
+    NotFoundError,
+    list_resources,
+)
+
+
+def _build_execution_filter_fn(
+    workspace_id: WorkspaceId,
+    workflow_id: WorkflowId | None = None,
+    thread_id: ThreadId | None = None,
+) -> Callable[[WorkflowExecution], bool]:
+    """Build filter function for executions with optional workflow and thread filters."""
+    workspace_filter: Callable[[WorkflowExecution], bool] = build_workspace_filter_fn(
+        workspace_id, WorkflowExecution
+    )
+
+    def filter_fn(execution: WorkflowExecution) -> bool:
+        if not workspace_filter(execution):
+            return False
+        if workflow_id is not None and execution.workflow_id != workflow_id:
+            return False
+        if thread_id is not None and execution.thread_id != thread_id:
+            return False
+        return True
+
+    return filter_fn
+
+
+class ExecutionService:
+    """Service for managing Execution resources with filesystem persistence."""
+
+    def __init__(
+        self,
+        base_dir: Path,
+        workflow_service: WorkflowService,
+        thread_facade: ThreadFacade,
+    ) -> None:
+        self._base_dir = base_dir
+        self._executions_dir = base_dir / "executions"
+        self._workflow_service = workflow_service
+        self._thread_facade = thread_facade
+
+    def _get_execution_path(
+        self, execution_id: WorkflowExecutionId, new: bool = False
+    ) -> Path:
+        """Get the file path for an execution."""
+        execution_path = self._executions_dir / f"{execution_id}.json"
+        exists = execution_path.exists()
+        if new and exists:
+            error_msg = f"Execution {execution_id} already exists"
+            raise ConflictError(error_msg)
+        if not new and not exists:
+            error_msg = f"Execution {execution_id} not found"
+            raise NotFoundError(error_msg)
+        return execution_path
+
+    def list_(
+        self,
+        workspace_id: WorkspaceId,
+        query: ListQuery,
+        workflow_id: WorkflowId | None = None,
+        thread_id: ThreadId | None = None,
+    ) -> ListResponse[WorkflowExecution]:
+        """List executions with optional filtering by workflow and/or thread."""
+        return list_resources(
+            base_dir=self._executions_dir,
+            query=query,
+            resource_type=WorkflowExecution,
+            filter_fn=_build_execution_filter_fn(workspace_id, workflow_id, thread_id),
+        )
+
+    def retrieve(
+        self, workspace_id: WorkspaceId, execution_id: WorkflowExecutionId
+    ) -> WorkflowExecution:
+        """Retrieve a specific execution by ID."""
+        try:
+            execution_path = self._get_execution_path(execution_id)
+            execution = WorkflowExecution.model_validate_json(
+                execution_path.read_text()
+            )
+
+            # Check workspace access - allow if workspace_id is None (global access)
+            # or if execution workspace matches or execution has no workspace
+            if execution.workspace_id != workspace_id:
+                error_msg = f"Execution {execution_id} not found"
+                raise NotFoundError(error_msg)
+
+        except FileNotFoundError as e:
+            error_msg = f"Execution {execution_id} not found"
+            raise NotFoundError(error_msg) from e
+        else:
+            return execution
+
+    async def create(
+        self, workspace_id: WorkspaceId, params: WorkflowExecutionCreateParams
+    ) -> tuple[WorkflowExecution, AsyncGenerator[Events, None]]:
+        """Create a new execution."""
+        # Validate that the workflow exists in the same workspace
+        workflow = self._workflow_service.retrieve(workspace_id, params.workflow_id)
+
+        # Create a thread and a run
+        run, async_generator = await self._thread_facade.create_thread_and_run(
+            workspace_id,
+            ThreadAndRunCreateParams(
+                assistant_id=workflow.assistant_id,
+                thread=ThreadCreateParams(
+                    name=workflow.name,
+                    messages=[
+                        MessageCreateParams(role="user", content=workflow.description)
+                    ],
+                ),
+            ),
+        )
+
+        execution = WorkflowExecution.create(
+            workspace_id, params.workflow_id, run.id, run.thread_id
+        )
+        self._save(execution, new=True)
+        return execution, async_generator
+
+    def _save(self, execution: WorkflowExecution, new: bool = False) -> None:
+        """Save execution to filesystem."""
+        self._executions_dir.mkdir(parents=True, exist_ok=True)
+        execution_file = self._get_execution_path(execution.id, new=new)
+        execution_file.write_text(execution.model_dump_json(), encoding="utf-8")

src/askui/chat/api/workflows/models.py

@@ -2,12 +2,15 @@
 
 from pydantic import BaseModel, Field
 
-from askui.chat.api.models import WorkspaceId, WorkspaceResource
+from askui.chat.api.models import (
+    AssistantId,
+    WorkflowId,
+    WorkspaceId,
+    WorkspaceResource,
+)
 from askui.utils.datetime_utils import UnixDatetime, now
 from askui.utils.id_utils import IdField, generate_time_ordered_id
 
-WorkflowId = Annotated[str, IdField("wf")]
-
 
 class WorkflowCreateParams(BaseModel):
     """
@@ -16,6 +19,7 @@ class WorkflowCreateParams(BaseModel):
 
     name: str
     description: str
+    assistant_id: AssistantId
     tags: list[str] = Field(default_factory=list)
 
 
@@ -42,19 +46,20 @@ class Workflow(WorkspaceResource):
         description (str): A detailed description of the workflow's purpose and steps.
         tags (list[str], optional): Tags associated with the workflow for filtering or
             categorization. Default is an empty list.
-        workspace_id (WorkspaceId | None, optional): The workspace this workflow belongs to.
+        workspace_id (WorkspaceId, optional): The workspace this workflow belongs to.
     """
 
     id: WorkflowId
     object: Literal["workflow"] = "workflow"
+    assistant_id: AssistantId
     created_at: UnixDatetime
     name: str
     description: str
     tags: list[str] = Field(default_factory=list)
 
     @classmethod
     def create(
-        cls, workspace_id: WorkspaceId | None, params: WorkflowCreateParams
+        cls, workspace_id: WorkspaceId, params: WorkflowCreateParams
     ) -> "Workflow":
         return cls(
             id=generate_time_ordered_id("wf"),