Fixed metadata system

Author: ericallam

packages/cli-v3/src/entryPoints/dev-run-worker.ts

@@ -378,6 +378,8 @@ const zodIpc = new ZodIpcConnection({
           return;
         }
 
+        runMetadataManager.runId = execution.run.id;
+
         const executor = new TaskExecutor(task, {
           tracer,
           tracingSDK,

packages/cli-v3/src/entryPoints/managed-run-worker.ts

@@ -376,6 +376,8 @@ const zodIpc = new ZodIpcConnection({
           return;
         }
 
+        runMetadataManager.runId = execution.run.id;
+
         const executor = new TaskExecutor(task, {
           tracer,
           tracingSDK,

packages/react-hooks/src/hooks/useWaitToken.ts

@@ -17,6 +17,8 @@ export interface WaitTokenInstance<TOutput> {
   isCompleted: boolean;
   /** Any error that occurred during completion */
   error?: Error;
+  /** Whether the waitpoint is ready to be completed */
+  isReady: boolean;
 }
 
 /**
@@ -36,7 +38,7 @@ export interface WaitTokenInstance<TOutput> {
  * ```
  */
 export function useWaitToken<TOutput>(
-  waitpointId: string,
+  waitpointId?: string,
   options?: UseApiClientOptions
 ): WaitTokenInstance<TOutput> {
   const apiClient = useApiClient(options);
@@ -46,6 +48,10 @@ export function useWaitToken<TOutput>(
       throw new Error("Could not complete waitpoint in useWaitToken: Missing access token");
     }
 
+    if (!waitpointId) {
+      throw new Error("Could not complete waitpoint in useWaitToken: Missing waitpoint ID");
+    }
+
     const result = await apiClient.completeWaitpointToken(waitpointId, {
       data: output,
     });
@@ -62,6 +68,7 @@ export function useWaitToken<TOutput>(
     },
     isLoading: mutation.isMutating,
     isCompleted: !!mutation.data?.success,
+    isReady: !!waitpointId,
     error: mutation.error,
   };
 }

references/agent-loop/src/components/main-app.tsx

@@ -6,36 +6,79 @@ import { useState } from "react";
 import { Send } from "lucide-react";
 import ChatInterface from "@/components/chat-interface";
 import InitialPrompt from "@/components/initial-prompt";
-import { useRealtimeTaskTriggerWithStreams } from "@trigger.dev/react-hooks";
+import { useRealtimeTaskTriggerWithStreams, useWaitToken } from "@trigger.dev/react-hooks";
 import type { agentLoopExample } from "@/trigger/agent";
+import { AgentLoopMetadata } from "@/trigger/schemas";
+import type { TextStreamPart } from "ai";
 
 type ChatConversation = Array<{ role: "user" | "assistant"; content: string }>;
 
+type ResponseStreams = {
+  [K in `responses.${number | string}`]: TextStreamPart<{}>;
+};
+
 export function useAgentLoop({ publicAccessToken }: { publicAccessToken: string }) {
-  const triggerInstance = useRealtimeTaskTriggerWithStreams<typeof agentLoopExample>(
-    "agent-loop-example",
-    {
-      accessToken: publicAccessToken,
-      baseURL: process.env.NEXT_PUBLIC_TRIGGER_API_URL,
-    }
-  );
+  const triggerInstance = useRealtimeTaskTriggerWithStreams<
+    typeof agentLoopExample,
+    ResponseStreams
+  >("agent-loop-example", {
+    accessToken: publicAccessToken,
+    baseURL: process.env.NEXT_PUBLIC_TRIGGER_API_URL,
+  });
   const [conversation, setConversation] = useState<ChatConversation>([]);
   const [isLoading, setIsLoading] = useState(false);
+  const [waitToken, setWaitToken] = useState<AgentLoopMetadata | null>(null);
+
+  const waitTokenInstance = useWaitToken(waitToken?.waitToken.id, {
+    enabled: !!waitToken?.waitToken.id,
+    accessToken: waitToken?.waitToken.publicAccessToken,
+    baseURL: process.env.NEXT_PUBLIC_TRIGGER_API_URL,
+  });
+
+  const waitTokenStream = waitToken?.waitToken.id
+    ? triggerInstance.streams[`responses.${waitToken?.waitToken.id}`]
+    : undefined;
+
+  const textStream = waitTokenStream
+    ?.map((part) => {
+      if (part.type === "text-delta") {
+        return part.textDelta;
+      }
+    })
+    .join("");
+
+  console.log("textStream", textStream);
 
   if (triggerInstance.run) {
     console.log("run", triggerInstance.run);
+
+    const metadata = AgentLoopMetadata.safeParse(triggerInstance.run.metadata);
+
+    if (!metadata.success) {
+      console.error("Failed to parse metadata", metadata.error);
+    } else {
+      console.log("metadata", metadata.data);
+
+      setWaitToken(metadata.data);
+    }
   }
 
   return {
     continueConversation: (prompt: string) => {
-      const result = triggerInstance.submit({
-        model: "gpt-4o-mini",
-        prompt,
-      });
+      if (waitTokenInstance.isReady) {
+        waitTokenInstance.complete({
+          message: prompt,
+        });
+      } else {
+        const result = triggerInstance.submit({
+          model: "gpt-4o-mini",
+          prompt,
+        });
 
-      setConversation((prev) => [...prev, { role: "user", content: prompt }]);
+        setConversation((prev) => [...prev, { role: "user", content: prompt }]);
 
-      return result;
+        return result;
+      }
     },
     conversation,
     isLoading,

references/agent-loop/src/trigger/agent.ts

@@ -39,6 +39,7 @@ export const agentLoopExample = schemaTask({
           const token = await wait.createToken({ timeout: "1d" });
 
           metadata.set("waitToken", token);
+          await metadata.flush();
 
           const result = streamText({
             model: aiModel,

references/agent-loop/src/trigger/schemas.ts

@@ -0,0 +1,10 @@
+import { z } from "zod";
+
+export const AgentLoopMetadata = z.object({
+  waitToken: z.object({
+    id: z.string(),
+    publicAccessToken: z.string(),
+  }),
+});
+
+export type AgentLoopMetadata = z.infer<typeof AgentLoopMetadata>;