example changed with var names

Author: dewmal

bindings/ceylon/ceylon/base/playground.py

@@ -114,6 +114,8 @@ async def play(self, workers: Optional[List[BaseAgent]] = None):
             logger.warning("Received Ctrl+C, initiating force close...")
             await self.force_close()
         except Exception as e:
+            import traceback
+            traceback.print_exc()
             logger.error(f"Error in playground: {e}")
             await self.force_close()
         finally:

bindings/ceylon/ceylon/llm/agent.py

@@ -80,7 +80,7 @@ async def execute_task(self, task: TaskMessage) -> None:
 
             if response:
                 # Cache successful response
-                self.response_cache[task.task_id] = response
+                self.response_cache[task.id] = response
 
                 # Print the response
                 logger.info("\nGenerated Content:")
@@ -100,10 +100,10 @@ async def execute_task(self, task: TaskMessage) -> None:
                 task.metadata['response_timestamp'] = response.timestamp
                 task.metadata.update(response.metadata)
 
-                logger.info(f"{self.name}: Completed task {task.task_id}")
+                logger.info(f"{self.name}: Completed task {task.id}")
 
                 # Remove from active tasks and broadcast completion
-                del self.active_tasks[task.task_id]
+                del self.active_tasks[task.id]
                 await self.broadcast_message(task)
 
                 # Request new task
@@ -112,7 +112,7 @@ async def execute_task(self, task: TaskMessage) -> None:
                 raise Exception("Failed to get valid LLM response")
 
         except Exception as e:
-            logger.error(f"Error executing LLM task {task.task_id}: {e}")
+            logger.error(f"Error executing LLM task {task.id}: {e}")
             task.status = TaskStatus.FAILED
             task.metadata = task.metadata or {}
             task.metadata['error'] = str(e)
@@ -177,7 +177,7 @@ async def _call_llm(self, task: TaskMessage) -> LLMResponse:
                 return LLMResponse(
                     content=response_text,
                     metadata={
-                        'task_id': task.task_id,
+                        'task_id': task.id,
                         'usage': usage.__dict__,
                         'model_name': self.llm_model.model_name
                     }

bindings/ceylon/ceylon/llm/task_pl.py

@@ -0,0 +1,180 @@
+from dataclasses import dataclass, field
+from typing import Any, Dict, Optional, List
+import asyncio
+import uuid
+
+from loguru import logger
+from ceylon import on, on_connect, Worker, AgentDetail
+
+from .base_playground import BasePlayGround
+from .manager import TaskManager, TaskMessage, TaskGroup, TaskStatus
+
+
+@dataclass
+class TaskRequest:
+    task_id: str
+    task_type: str
+    instructions: str
+    required_role: str
+    metadata: Dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass
+class TaskResponse:
+    task_id: str
+    status: TaskStatus
+    result: Optional[Any] = None
+    error_message: Optional[str] = None
+    runtime_stats: Dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass
+class TaskProgressUpdate:
+    task_id: str
+    progress: float  # 0.0 to 1.0
+    status: TaskStatus
+    message: Optional[str] = None
+
+
+class TaskWorker(Worker):
+    def __init__(self, name: str, role: str):
+        super().__init__(name=name, role=role)
+        self.active_task: Optional[TaskMessage] = None
+
+    @on(TaskRequest)
+    async def handle_task_request(self, request: TaskRequest, time: int):
+        try:
+            if self.active_task:
+                # Already processing a task
+                return
+
+            logger.info(f"Worker {self.name} received task: {request.task_id}")
+
+            # Process task (simulated work)
+            self.active_task = TaskMessage(
+                task_id=request.task_id,
+                name=f"Task-{request.task_id[:8]}",
+                instructions=request.instructions,
+                required_role=request.required_role
+            )
+
+            # Send progress updates
+            await self.broadcast_message(TaskProgressUpdate(
+                task_id=request.task_id,
+                progress=0.0,
+                status=TaskStatus.IN_PROGRESS,
+                message="Starting task"
+            ))
+
+            # Simulate work
+            await asyncio.sleep(2)
+
+            # Send completion
+            response = TaskResponse(
+                task_id=request.task_id,
+                status=TaskStatus.COMPLETED,
+                result={"processed": True},
+                runtime_stats={
+                    "duration": 2.0,
+                    "memory_used": "100MB"
+                }
+            )
+            await self.broadcast_message(response)
+            self.active_task = None
+
+        except Exception as e:
+            logger.error(f"Error processing task {request.task_id}: {e}")
+            await self.broadcast_message(TaskResponse(
+                task_id=request.task_id,
+                status=TaskStatus.FAILED,
+                error_message=str(e)
+            ))
+            self.active_task = None
+
+
+class TaskPlayGround(BasePlayGround):
+    def __init__(self, name="task_playground", port=8888):
+        super().__init__(name=name, port=port)
+        self.task_manager = TaskManager()
+        self.task_responses: Dict[str, TaskResponse] = {}
+        self.task_events: Dict[str, asyncio.Event] = {}
+        self.task_progress: Dict[str, float] = {}
+
+    @on(TaskResponse)
+    async def handle_task_response(self, response: TaskResponse, time: int):
+        """Handle task completion responses from workers"""
+        logger.info(f"Received task response for {response.task_id}: {response.status}")
+
+        self.task_responses[response.task_id] = response
+        if response.task_id in self.task_events:
+            self.task_events[response.task_id].set()
+
+        if response.status == TaskStatus.COMPLETED:
+            task = self.task_manager.tasks.get(response.task_id)
+            if task:
+                task.completed = True
+                all_completed = await self.task_manager.handle_task_completion(task)
+                if all_completed:
+                    logger.info("All tasks completed")
+                    await self.finish()
+
+    @on(TaskProgressUpdate)
+    async def handle_progress_update(self, update: TaskProgressUpdate, time: int):
+        """Handle task progress updates"""
+        self.task_progress[update.task_id] = update.progress
+        logger.debug(f"Task {update.task_id} progress: {update.progress:.1%}")
+
+    @on_connect("*")
+    async def handle_worker_connection(self, topic: str, agent: AgentDetail):
+        """Register new workers with the task manager"""
+        self.task_manager.register_worker(agent.name, agent.role)
+        await super().on_llm_agent_connected(topic, agent)
+
+    async def submit_task(self, task_type: str, instructions: str, role: str,
+                          metadata: Optional[Dict[str, Any]] = None) -> TaskResponse:
+        """Submit a task and wait for its completion"""
+        task_id = str(uuid.uuid4())
+        request = TaskRequest(
+            task_id=task_id,
+            task_type=task_type,
+            instructions=instructions,
+            required_role=role,
+            metadata=metadata or {}
+        )
+
+        # Setup completion event
+        self.task_events[task_id] = asyncio.Event()
+
+        # Send request
+        await self.broadcast_message(request)
+
+        try:
+            # Wait for completion
+            await asyncio.wait_for(self.task_events[task_id].wait(), timeout=30.0)
+            return self.task_responses[task_id]
+        except asyncio.TimeoutError:
+            return TaskResponse(
+                task_id=task_id,
+                status=TaskStatus.FAILED,
+                error_message="Task timed out"
+            )
+        finally:
+            # Cleanup
+            self.task_events.pop(task_id, None)
+
+    def get_task_progress(self, task_id: str) -> float:
+        """Get current progress for a task"""
+        return self.task_progress.get(task_id, 0.0)
+
+    async def close(self):
+        """Clean shutdown of playground"""
+        # Cancel any pending tasks
+        for task_id, event in self.task_events.items():
+            event.set()
+
+        # Clear state
+        self.task_responses.clear()
+        self.task_events.clear()
+        self.task_progress.clear()
+
+        await self.force_close()

bindings/ceylon/ceylon/task/data.py

@@ -2,12 +2,9 @@
 #  Licensed under the Apache License, Version 2.0 (See LICENSE.md or http://www.apache.org/licenses/LICENSE-2.0).
 #
 #
-import uuid
-from dataclasses import dataclass, field
+from dataclasses import dataclass
 from enum import Enum
-from typing import Optional, Set, List, Dict, Callable, Any
-
-from loguru import logger
+from typing import Any, Dict, Optional
 
 
 class TaskStatus(Enum):
@@ -17,85 +14,18 @@ class TaskStatus(Enum):
     FAILED = "failed"
 
 
-@dataclass
-class TaskMessage:
-    task_id: str
-    name: str
-    instructions: str
-    required_role: str=None
-    parent_id: Optional[str] = None
-    group_id: Optional[str] = None  # Reference to TaskGroup
-    subtask_ids: Set[str] = field(default_factory=set)
-    assigned_to: Optional[str] = None
-    completed: bool = False
-    start_time: Optional[float] = None
-    end_time: Optional[float] = None
-    max_concurrent: int = 3
-    status: TaskStatus = TaskStatus.PENDING
-    metadata: Optional[Dict[str, Any]] = None
-    result: Optional[Any] = None
-
-
 @dataclass
 class TaskRequest:
-    requester: str
-    role: str
-    task_type: str
-    priority: int = 1
-
-
-@dataclass
-class TaskStatusUpdate:
     task_id: str
-    status: TaskStatus
-    message: Optional[str] = None
-    group_id: Optional[str] = None
-
-
-class GoalStatus(Enum):
-    NOT_STARTED = "not_started"
-    IN_PROGRESS = "in_progress"
-    ACHIEVED = "achieved"
-    FAILED = "failed"
-
-
-@dataclass
-class TaskGroupGoal:
-    name: str
-    description: str
-    check_condition: Callable[[Dict[str, 'TaskGroup'], Dict[str, TaskMessage]], bool]
-    success_message: str
-    failure_message: str
-    status: GoalStatus = GoalStatus.NOT_STARTED
+    task_type: str
+    data: Any
+    metadata: Dict[str, Any] = None
 
 
 @dataclass
-class TaskGroup:
+class TaskResponse:
     task_id: str
-    name: str
-    description: str
-    subtasks: List[TaskMessage]
-    goal: Optional[TaskGroupGoal] = None
-    id: str = field(default_factory=lambda: str(uuid.uuid4()))
-    dependencies: Dict[str, List[str]] = field(default_factory=dict)
-    depends_on: List[str] = field(default_factory=list)
-    required_by: List[str] = field(default_factory=list)
-    status: TaskStatus = TaskStatus.PENDING
-    priority: int = 1
-
-    def check_goal(self, task_groups: Dict[str, 'TaskGroup'], completed_tasks: Dict[str, TaskMessage]) -> bool:
-        """Check if the group's goal has been achieved"""
-        print(f"Checking goal for group {self.name} {self.goal} ")
-        if not self.goal:
-            return False
-
-        if self.goal.status == GoalStatus.ACHIEVED:
-            return True
-
-        if self.goal.check_condition(task_groups, completed_tasks):
-            self.goal.status = GoalStatus.ACHIEVED
-            logger.debug(f"\nGoal Achieved for group {self.name}: {self.goal.name}")
-            logger.debug(self.goal.success_message)
-            return True
-
-        return False
+    result: Any
+    status: str  # 'success' or 'error'
+    error_message: Optional[str] = None
+    metadata: Dict[str, Any] = None