Adding all changes before testing

Author: jasonyang101

examples/tutorials/10_agentic/10_temporal/000_hello_acp/project/run_worker.py

@@ -9,6 +9,7 @@
 from workflow import At000HelloAcpWorkflow
 
 
+
 environment_variables = EnvironmentVariables.refresh()
 
 logger = make_logger(__name__)

examples/tutorials/10_agentic/10_temporal/030_custom_activities/project/custom_activites.py

@@ -15,11 +15,18 @@ class ProcessBatchEventsActivityParams(BaseModel):
   events: List[Any]
   batch_number: int
 
+
 REPORT_PROGRESS_ACTIVITY = "report_progress"
 class ReportProgressActivityParams(BaseModel):
   num_batches_processed: int
   num_batches_failed: int
   num_batches_running: int
+  task_id: str
+
+
+COMPLETE_WORKFLOW_ACTIVITY = "complete_workflow"
+class CompleteWorkflowActivityParams(BaseModel):
+  task_id: str
 
 
 class CustomActivities:
@@ -82,9 +89,22 @@ async def report_progress(self, params: ReportProgressActivityParams) -> None:
     logger.info(f"📊 Progress Update - num_batches_processed: {params.num_batches_processed}, num_batches_failed: {params.num_batches_failed}, num_batches_running: {params.num_batches_running}")
 
     await adk.messages.create(
-        task_id=task_id,
+        task_id=params.task_id,
         content=TextContent(
             author="agent",
             content=f"📊 Progress Update - num_batches_processed: {params.num_batches_processed}, num_batches_failed: {params.num_batches_failed}, num_batches_running: {params.num_batches_running}",
         ),
     )
+
+  @activity.defn(name=COMPLETE_WORKFLOW_ACTIVITY)
+  async def complete_workflow(self, params: CompleteWorkflowActivityParams) -> None:
+    """
+    This activity will complete the workflow.
+
+    Typically here you may do anything like:
+    - Send a final email to the user
+    - Send a final message to the user
+    - Update a job status in a database to completed
+    """
+    logger.info(f"🎉 Workflow Complete! Task ID: {params.task_id}")
+

examples/tutorials/10_agentic/10_temporal/030_custom_activities/project/run_worker.py

@@ -7,6 +7,7 @@
 from agentex.lib.environment_variables import EnvironmentVariables
 
 from project.workflow import At030CustomActivitiesWorkflow
+from project.custom_activites import CustomActivities
 
 
 environment_variables = EnvironmentVariables.refresh()
@@ -27,8 +28,17 @@ async def main():
         task_queue=task_queue_name,
     )
 
+    agentex_activities = get_all_activities()
+
+    custom_activities_use_case = CustomActivities()
+    all_activites = [
+        custom_activities_use_case.report_progress, 
+        custom_activities_use_case.process_batch_events,
+        *agentex_activities, 
+    ]
+
     await worker.run(
-        activities=get_all_activities(),
+        activities=all_activites,
         workflow=At030CustomActivitiesWorkflow,
     )
 

examples/tutorials/10_agentic/10_temporal/030_custom_activities/project/workflow.py

@@ -2,7 +2,8 @@
 from typing import List, Any, override
 
 from temporalio import workflow
-from pydantic import BaseModel
+from temporalio.common import RetryPolicy
+from datetime import timedelta
 
 from agentex.lib import adk
 from agentex.lib.types.acp import CreateTaskParams, SendEventParams
@@ -15,7 +16,7 @@
 
 from project.workflow_utils import BatchProcessingUtils
 from project.shared_models import StateModel, IncomingEventData
-
+from project.custom_activites import REPORT_PROGRESS_ACTIVITY, ReportProgressActivityParams, COMPLETE_WORKFLOW_ACTIVITY, CompleteWorkflowActivityParams
 
 environment_variables = EnvironmentVariables.refresh()
 
@@ -97,7 +98,7 @@ async def on_task_create(self, params: CreateTaskParams) -> None:
             task_id=params.task.id,
             content=TextContent(
                 author="agent",
-                content=f"🚀 Starting batch processing! I'll collect events into batches of {self._batch_size} and process them using custom activities.",
+                content=f"🚀 Starting batch processing! I'll collect events into batches of {self._batch_size} and process them using custom activities. I'll also report progress you as I go..",
             ),
         )
 
@@ -172,7 +173,39 @@ async def on_task_create(self, params: CreateTaskParams) -> None:
 
         # Wait for all remaining tasks to complete, with real-time progress updates
         await BatchProcessingUtils.wait_for_remaining_tasks(self._processing_tasks, self._state, params.task.id)
-        
-        # Final summary with complete statistics
-        await BatchProcessingUtils.send_final_summary(self._state, params.task.id)
-        return
+        await workflow.execute_activity(
+            REPORT_PROGRESS_ACTIVITY,
+            ReportProgressActivityParams(
+                num_batches_processed=self._state.num_batches_processed,
+                num_batches_failed=self._state.num_batches_failed,
+                num_batches_running=0,
+                task_id=params.task.id
+            ),
+            start_to_close_timeout=timedelta(minutes=1),
+            retry_policy=RetryPolicy(maximum_attempts=3)
+        )
+
+        final_summary = f"✅ Workflow Complete! Final Summary:\n"
+                        f"• Batches completed successfully: {self._state.num_batches_processed} ✅\n" 
+                        f"• Batches failed: {self._state.num_batches_failed} ❌\n"
+                        f"• Total events processed: {self._state.total_events_processed}\n"
+                        f"• Events dropped (queue full): {self._state.total_events_dropped}\n"
+                        f"📝 Tutorial completed - you learned how to use asyncio.create_task() with Temporal custom activities!"
+
+        await adk.messages.create(
+            task_id=task_id,
+            content=TextContent(
+                author="agent",
+                content=final_summary
+            ),
+        )
+
+        await workflow.execute_activity(
+            COMPLETE_WORKFLOW_ACTIVITY,
+            CompleteWorkflowActivityParams(
+                task_id=params.task.id
+            ),
+            start_to_close_timeout=timedelta(minutes=1),
+            retry_policy=RetryPolicy(maximum_attempts=3)            
+        )        
+

examples/tutorials/10_agentic/10_temporal/030_custom_activities/project/workflow_utils.py

@@ -123,7 +123,8 @@ async def update_progress(processing_tasks: List[asyncio.Task[Any]], state: Stat
                 ReportProgressActivityParams(
                     num_batches_processed=state.num_batches_processed,
                     num_batches_failed=state.num_batches_failed,
-                    num_batches_running=len(still_running)
+                    num_batches_running=len(still_running),
+                    task_id=task_id,
                 ),
                 start_to_close_timeout=timedelta(minutes=1),
                 retry_policy=RetryPolicy(maximum_attempts=3)
@@ -197,22 +198,4 @@ async def wait_for_remaining_tasks(processing_tasks: List[asyncio.Task[Any]], st
             except asyncio.TimeoutError:
                 # Some tasks still running, update progress and continue waiting
                 processing_tasks[:] = await BatchProcessingUtils.update_progress(processing_tasks, state, task_id)
-                continue
-
-    @staticmethod
-    async def send_final_summary(state: Any, task_id: str) -> None:
-        """
-        Send the final workflow completion summary with statistics.
-        """
-        await adk.messages.create(
-            task_id=task_id,
-            content=TextContent(
-                author="agent",
-                content=f"✅ Workflow Complete! Final Summary:\n"
-                       f"• Batches completed successfully: {state.num_batches_processed} ✅\n" 
-                       f"• Batches failed: {state.num_batches_failed} ❌\n"
-                       f"• Total events processed: {state.total_events_processed}\n"
-                       f"• Events dropped (queue full): {state.total_events_dropped}\n"
-                       f"📝 Tutorial completed - you learned how to use asyncio.create_task() with Temporal custom activities!",
-            ),
-        )
+                continue

examples/tutorials/10_agentic/10_temporal/030_custom_activities/pyproject.toml

@@ -15,6 +15,7 @@ dependencies = [
 
 [project.optional-dependencies]
 dev = [
+    "jupyter",
     "pytest",
     "black",
     "isort",
@@ -31,4 +32,4 @@ target-version = ['py312']
 
 [tool.isort]
 profile = "black"
-line_length = 88
+line_length = 88

src/agentex/lib/cli/handlers/run_handlers.py

@@ -86,7 +86,7 @@ async def cleanup_processes(self):
 
 
 async def start_temporal_worker_with_reload(
-    worker_path: Path, env: dict[str, str], process_manager: ProcessManager
+    worker_path: Path, env: dict[str, str], process_manager: ProcessManager, manifest_dir: Path
 ) -> asyncio.Task[None]:
     """Start temporal worker with auto-reload using watchfiles"""
 
@@ -143,7 +143,7 @@ async def start_worker() -> asyncio.subprocess.Process:
                 except asyncio.CancelledError:
                     pass
 
-            current_process = await start_temporal_worker(worker_path, env)
+            current_process = await start_temporal_worker(worker_path, env, manifest_dir)
             process_manager.add_process(current_process)
             console.print("[green]Temporal worker started[/green]")
             return current_process
@@ -222,7 +222,7 @@ async def start_acp_server(
 
 
 async def start_temporal_worker(
-    worker_path: Path, env: dict[str, str]
+    worker_path: Path, env: dict[str, str], manifest_dir: Path
 ) -> asyncio.subprocess.Process:
     """Start the temporal worker process"""
     cmd = [sys.executable, "-m", "run_worker"]
@@ -231,7 +231,7 @@ async def start_temporal_worker(
 
     return await asyncio.create_subprocess_exec(
         *cmd,
-        cwd=worker_path.parent,  # Use worker directory as CWD for imports to work
+        cwd=manifest_dir,  # Use worker directory as CWD for imports to work
         env=env,
         stdout=asyncio.subprocess.PIPE,
         stderr=asyncio.subprocess.STDOUT,
@@ -320,7 +320,7 @@ async def run_agent(manifest_path: str, debug_config: "DebugConfig | None" = Non
                 worker_task = asyncio.create_task(stream_process_output(worker_process, "WORKER"))
             else:
                 # Normal mode with auto-reload
-                worker_task = await start_temporal_worker_with_reload(file_paths["worker"], agent_env, process_manager)
+                worker_task = await start_temporal_worker_with_reload(file_paths["worker"], agent_env, process_manager, manifest_dir)
             tasks.append(worker_task)
 
         console.print(