Merge pull request #11 from jhd3197/dev

Migrate Prompture integration to async API

Author: jhd3197

agentsite/agents/designer.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from prompture import Agent
+from prompture import AsyncAgent as Agent
 
 from ..engine.capabilities import supports_structured_output
 from ..models import StyleSpec

agentsite/agents/developer.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from prompture import Agent
+from prompture import AsyncAgent as Agent
 
 from ..engine.capabilities import supports_tools
 from .personas import DEVELOPER_PERSONA

agentsite/agents/orchestrator.py

@@ -4,7 +4,7 @@
 
 from typing import Any
 
-from prompture import GroupCallbacks, LoopGroup, SequentialGroup
+from prompture import AsyncLoopGroup, AsyncSequentialGroup, GroupCallbacks
 
 from ..config import settings
 from ..models import AgentConfig
@@ -21,7 +21,7 @@ def create_pipeline(
     max_review_iterations: int | None = None,
     review_threshold: int | None = None,
     agent_configs: dict[str, AgentConfig] | None = None,
-) -> SequentialGroup:
+) -> AsyncSequentialGroup:
     """Build the full generation pipeline (static — all 4 agents).
 
     Pipeline structure::
@@ -60,7 +60,7 @@ def _exit_condition(state: dict[str, Any], iteration: int) -> bool:
         lower = feedback_text.lower()
         return '"approved": true' in lower or '"approved":true' in lower
 
-    build_review_loop = LoopGroup(
+    build_review_loop = AsyncLoopGroup(
         [
             (
                 developer,
@@ -92,7 +92,7 @@ def _exit_condition(state: dict[str, Any], iteration: int) -> bool:
     )
 
     # Full pipeline
-    pipeline = SequentialGroup(
+    pipeline = AsyncSequentialGroup(
         [
             (pm, "{prompt}"),
             (
@@ -120,7 +120,7 @@ def create_dynamic_pipeline(
     review_threshold: int | None = None,
     agent_configs: dict[str, AgentConfig] | None = None,
     error_policy: Any = None,
-) -> SequentialGroup:
+) -> AsyncSequentialGroup:
     """Build a dynamic pipeline based on PM's required_agents output.
 
     PM always runs before this. This builds the remaining pipeline steps
@@ -169,7 +169,7 @@ def _exit_condition(state: dict[str, Any], iteration: int) -> bool:
                 return True
             return False
 
-        build_review_loop = LoopGroup(
+        build_review_loop = AsyncLoopGroup(
             [
                 (
                     developer,
@@ -220,7 +220,7 @@ def _exit_condition(state: dict[str, Any], iteration: int) -> bool:
     kwargs: dict[str, Any] = {"callbacks": callbacks}
     if error_policy is not None:
         kwargs["error_policy"] = error_policy
-    return SequentialGroup(steps, **kwargs)
+    return AsyncSequentialGroup(steps, **kwargs)
 
 
 def _agent_model(

agentsite/agents/pm.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from prompture import Agent
+from prompture import AsyncAgent as Agent
 
 from ..engine.capabilities import supports_structured_output
 from ..models import SitePlan

agentsite/agents/reviewer.py

@@ -4,7 +4,7 @@
 
 import json
 
-from prompture import Agent
+from prompture import AsyncAgent as Agent
 
 from ..engine.capabilities import supports_structured_output, supports_tools
 from ..models import ReviewFeedback

agentsite/api/routes/generate.py

@@ -4,7 +4,6 @@
 
 import asyncio
 import logging
-from concurrent.futures import ThreadPoolExecutor
 from datetime import datetime, timezone
 
 from fastapi import APIRouter, Depends, HTTPException, WebSocket, WebSocketDisconnect
@@ -18,10 +17,6 @@
 logger = logging.getLogger("agentsite.api.generate")
 router = APIRouter(tags=["generate"])
 
-# Thread pool for running sync Prompture pipelines
-_executor = ThreadPoolExecutor(max_workers=2, thread_name_prefix="agentsite-gen")
-
-
 class GenerateRequest(BaseModel):
     prompt: str = ""
     model: str = ""
@@ -82,9 +77,7 @@ async def start_generation(
     )
     await version_repo.create(version)
 
-    # Get event loop for WS bridge
-    loop = asyncio.get_running_loop()
-    on_event = ws_manager.make_callback(project_id, loop)
+    on_event = ws_manager.make_callback(project_id)
 
     # Load agent configs from DB for pipeline customization
     configs_list = await agent_config_repo.list_all()
@@ -93,17 +86,13 @@ async def start_generation(
     # Build pipeline
     pipeline = GenerationPipeline(pm, on_event=on_event, agent_configs=agent_configs)
 
-    # Run in thread pool (Prompture groups are synchronous)
     async def _run():
         try:
-            result = await loop.run_in_executor(
-                _executor,
-                lambda: pipeline.generate(
-                    project,
-                    slug=slug,
-                    version_number=version_number,
-                    page_prompt=prompt,
-                ),
+            result = await pipeline.generate(
+                project,
+                slug=slug,
+                version_number=version_number,
+                page_prompt=prompt,
             )
             # Read generated files from disk into version record
             file_list = pm.list_version_files(project.id, slug, version_number)

agentsite/api/websocket.py

@@ -2,10 +2,7 @@
 
 from __future__ import annotations
 
-import asyncio
-import json
 import logging
-from typing import Any
 
 from fastapi import WebSocket
 
@@ -50,16 +47,11 @@ async def broadcast(self, project_id: str, event: WSEvent) -> None:
         for ws in dead:
             self.disconnect(project_id, ws)
 
-    def make_callback(self, project_id: str, loop: asyncio.AbstractEventLoop):
-        """Create a sync callback that bridges to async broadcast.
+    def make_callback(self, project_id: str):
+        """Return an async callback for GenerationPipeline.on_event."""
 
-        Returns a callable suitable for GenerationPipeline.on_event.
-        """
-
-        def _on_event(event: WSEvent) -> None:
-            asyncio.run_coroutine_threadsafe(
-                self.broadcast(project_id, event), loop
-            )
+        async def _on_event(event: WSEvent) -> None:
+            await self.broadcast(project_id, event)
 
         return _on_event
 

agentsite/cli.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import json
 import sys
 
@@ -99,11 +100,13 @@ def _on_event(event):
     pipeline = GenerationPipeline(pm, on_event=_on_event)
 
     try:
-        result = pipeline.generate(
-            project,
-            slug=page,
-            version_number=1,
-            page_prompt=prompt,
+        result = asyncio.run(
+            pipeline.generate(
+                project,
+                slug=page,
+                version_number=1,
+                page_prompt=prompt,
+            )
         )
         version_dir = pm.version_dir(project.id, page, 1)