Callback

Author: caelinsutch

apps/execution-worker/src/context.ts

@@ -3,12 +3,23 @@ import type { ExecutionAgent } from "./durable-objects/execution-agent";
 
 // Worker context and environment types
 
+// Type for InteractionService RPC interface
+export type InteractionServiceRpc = {
+  handleAgentCompletion(input: {
+    agentId: string;
+    conversationId: string;
+    success: boolean;
+    result?: string;
+    error?: string;
+  }): Promise<void>;
+};
+
 // Re-export the Env from the global Cloudflare namespace with proper DO typing
 export type WorkerEnv = SharedHonoEnv &
   Cloudflare.Env & {
     EXECUTION_AGENT: DurableObjectNamespace<ExecutionAgent>;
     HYPERDRIVE: Hyperdrive;
-    INTERACTION_WORKER: Fetcher;
+    INTERACTION_WORKER: Fetcher & InteractionServiceRpc;
   };
 
 export type Variables = SharedHonoVariables;

apps/execution-worker/src/durable-objects/execution-agent.ts

@@ -15,29 +15,9 @@ export class ExecutionAgent extends Agent<WorkerEnv, ExecutionState> {
   };
 
   /**
-   * Execute a task using an agentic loop with tools
+   * Execute task in background and ping interaction worker when done
    */
-  @callable()
-  async executeTask(
-    input: TaskInput,
-  ): Promise<{ success: boolean; result?: unknown; error?: string }> {
-    // Check if already executing to prevent concurrent runs
-    if (this.state.isExecuting) {
-      logger
-        .withTags({
-          agentId: input.agentId,
-          conversationId: input.conversationId,
-        })
-        .warn("ExecutionAgent: Attempted concurrent execution", {
-          taskDescription: input.taskDescription.substring(0, 100),
-        });
-
-      return {
-        success: false,
-        error: "Agent is already executing a task",
-      };
-    }
-
+  private async executeTaskInBackground(input: TaskInput): Promise<void> {
     this.setState({ isExecuting: true });
 
     logger
@@ -204,7 +184,40 @@ ${input.taskDescription}`,
         });
 
       this.setState({ isExecuting: false });
-      return { success: true, result: finalResult };
+
+      // Ping interaction worker with completion via RPC
+      try {
+        logger
+          .withTags({
+            agentId: input.agentId,
+            conversationId: input.conversationId,
+          })
+          .info("ExecutionAgent: Pinging interaction worker with completion");
+
+        await this.env.INTERACTION_WORKER.handleAgentCompletion({
+          agentId: input.agentId,
+          conversationId: input.conversationId,
+          success: true,
+          result: result.text,
+        });
+
+        logger
+          .withTags({
+            agentId: input.agentId,
+            conversationId: input.conversationId,
+          })
+          .info("ExecutionAgent: Successfully pinged interaction worker");
+      } catch (rpcError) {
+        logger
+          .withTags({
+            agentId: input.agentId,
+            conversationId: input.conversationId,
+          })
+          .error("ExecutionAgent: Failed to ping interaction worker", {
+            error:
+              rpcError instanceof Error ? rpcError.message : String(rpcError),
+          });
+      }
     } catch (error) {
       const errorMessage =
         error instanceof Error ? error.message : String(error);
@@ -251,7 +264,89 @@ ${input.taskDescription}`,
       }
 
       this.setState({ isExecuting: false });
-      return { success: false, error: errorMessage };
+
+      // Ping interaction worker with error via RPC
+      try {
+        logger
+          .withTags({
+            agentId: input.agentId,
+            conversationId: input.conversationId,
+          })
+          .info("ExecutionAgent: Pinging interaction worker with error");
+
+        await this.env.INTERACTION_WORKER.handleAgentCompletion({
+          agentId: input.agentId,
+          conversationId: input.conversationId,
+          success: false,
+          error: errorMessage,
+        });
+
+        logger
+          .withTags({
+            agentId: input.agentId,
+            conversationId: input.conversationId,
+          })
+          .info(
+            "ExecutionAgent: Successfully pinged interaction worker with error",
+          );
+      } catch (rpcError) {
+        logger
+          .withTags({
+            agentId: input.agentId,
+            conversationId: input.conversationId,
+          })
+          .error(
+            "ExecutionAgent: Failed to ping interaction worker with error",
+            {
+              error:
+                rpcError instanceof Error ? rpcError.message : String(rpcError),
+            },
+          );
+      }
     }
   }
+
+  /**
+   * Execute a task using an agentic loop with tools
+   * Returns immediately and runs task in background
+   */
+  @callable()
+  async executeTask(
+    input: TaskInput,
+  ): Promise<{ success: boolean; message: string }> {
+    // Check if already executing to prevent concurrent runs
+    if (this.state.isExecuting) {
+      logger
+        .withTags({
+          agentId: input.agentId,
+          conversationId: input.conversationId,
+        })
+        .warn("ExecutionAgent: Attempted concurrent execution", {
+          taskDescription: input.taskDescription.substring(0, 100),
+        });
+
+      return {
+        success: false,
+        message: "Agent is already executing a task",
+      };
+    }
+
+    logger
+      .withTags({
+        agentId: input.agentId,
+        conversationId: input.conversationId,
+      })
+      .info("ExecutionAgent: Starting task execution in background", {
+        taskDescription: input.taskDescription.substring(0, 200),
+        taskLength: input.taskDescription.length,
+      });
+
+    // Start execution in background
+    this.ctx.waitUntil(this.executeTaskInBackground(input));
+
+    return {
+      success: true,
+      message: "Task execution started in background",
+    };
+  }
 }

apps/interaction-worker/src/index.ts

@@ -1,3 +1,4 @@
+import { WorkerEntrypoint } from "cloudflare:workers";
 import {
   useWorkersLogger,
   withDefaultCors,
@@ -6,14 +7,26 @@ import {
 } from "@poppy/hono-helpers";
 import { loopMessageWebhookPayloadSchema } from "@poppy/schemas";
 import { Hono } from "hono";
-import type { App } from "./context";
+import type { App, WorkerEnv } from "./context";
 import { createDatabaseClient } from "./db/client";
 import { logger } from "./helpers/logger";
 import { handleMessageInbound } from "./services/loop/loop-message-inbound-handler";
+import {
+  type AgentCompletionInput,
+  processAgentCompletion,
+} from "./services/process-message/process-agent-completion";
 
 // Export Durable Object
 export { MessageDebouncer } from "./durable-objects/message-debouncer";
 
+// WorkerEntrypoint for RPC calls from other workers
+export class InteractionService extends WorkerEntrypoint<WorkerEnv> {
+  async handleAgentCompletion(input: AgentCompletionInput): Promise<void> {
+    const db = createDatabaseClient(this.env);
+    await processAgentCompletion(input, db, this.env);
+  }
+}
+
 const app = new Hono<App>();
 
 // Middleware