feat: Enhance workflow logging and progress tracking across multiple workflows

- Added custom logging for workflow step start, progress, and completion in the content studio workflow.
- Implemented detailed progress updates for each step in the document processing workflow, including error handling.
- Enhanced the financial report workflow with custom logging for fetching price data, including parallel progress tracking.
- Improved the governed RAG index workflow with detailed progress updates during document indexing.
- Updated the weather workflow to include progress tracking for fetching weather data and planning activities.
- Standardized the logging format across workflows to include step IDs and status messages for better traceability.

Author: ssdeanx

app/globals.css

@@ -1,6 +1,6 @@
 @import 'tailwindcss';
 @import 'tw-animate-css';
-
+@source "../node_modules/streamdown/dist/index.js";
 /* Tailwind CSS v4 Custom Variants */
 @custom-variant dark (&:is(.dark *));
 @custom-variant starting (&:is(:popover-open, :modal, [data-state="open"]):starting-style);

app/workflows/providers/workflow-context.tsx

@@ -11,6 +11,7 @@ import {
   useEffect,
   type ReactNode,
 } from "react"
+import type { AgentDataPart } from "@mastra/ai-sdk"
 import {
   getWorkflowConfig,
   WORKFLOW_CONFIGS,
@@ -61,6 +62,15 @@ export interface WorkflowRun {
   error?: string
 }
 
+export interface WorkflowDataPart {
+  messageId: string
+  partIndex: number
+  part: {
+    type: string
+    data?: unknown
+  }
+}
+
 /* eslint-disable no-unused-vars */
 export interface WorkflowContextValue {
   selectedWorkflow: WorkflowId
@@ -70,6 +80,7 @@ export interface WorkflowContextValue {
   activeStepIndex: number
   progressEvents: WorkflowProgressEvent[]
   suspendPayload: WorkflowSuspendPayload | null
+  dataParts: WorkflowDataPart[]
   selectWorkflow: (workflowId: WorkflowId) => void
   runWorkflow: (inputData?: Record<string, unknown>) => void
   pauseWorkflow: () => void
@@ -204,6 +215,7 @@ export function WorkflowProvider({
   const [activeStepIndex, setActiveStepIndex] = useState(-1)
   const [progressEvents, setProgressEvents] = useState<WorkflowProgressEvent[]>([])
   const [suspendPayload, setSuspendPayload] = useState<WorkflowSuspendPayload | null>(null)
+  const [dataParts, setDataParts] = useState<WorkflowDataPart[]>([])
 
   const workflowConfig = useMemo(
     () => getWorkflowConfig(selectedWorkflow),
@@ -253,14 +265,24 @@ export function WorkflowProvider({
     }
   }, [status, workflowStatus])
 
-  // Extract progress events from custom data parts
+  // Extract progress events and data parts from custom data parts
   useEffect(() => {
     const allProgressEvents: WorkflowProgressEvent[] = []
+    const allDataParts: WorkflowDataPart[] = []
     let detectedSuspendPayload: WorkflowSuspendPayload | null = null
 
     for (const message of messages) {
       if (message.role === "assistant" && message.parts !== null) {
         for (const [partIndex, part] of message.parts.entries()) {
+          // Collect all data parts for custom component rendering
+          if (typeof part.type === "string" && part.type.startsWith("data-")) {
+            allDataParts.push({
+              messageId: message.id,
+              partIndex,
+              part,
+            })
+          }
+
           // Handle workflow progress events (data-workflow, data-tool-workflow)
           if (part.type === "data-workflow" || part.type === "data-tool-workflow") {
             const workflowPart = part as { data?: { text?: string; status?: string; stepId?: string } }
@@ -279,6 +301,77 @@ export function WorkflowProvider({
             }
           }
 
+          // Handle network aggregation events (data-network)
+          if (part.type === "data-network") {
+            const networkPart = part as { data?: { text?: string; status?: string; networkId?: string } }
+            const eventData = networkPart.data
+
+            if (eventData && eventData.text !== null && typeof eventData.text === "string" && eventData.text.trim()) {
+              allProgressEvents.push({
+                id: `${message.id}-${part.type}-${partIndex}`,
+                stage: "network",
+                status: "in-progress",
+                message: eventData.text,
+                stepId: eventData.networkId,
+                timestamp: new Date(),
+                data: eventData,
+              })
+            }
+          }
+
+          // Handle tool agent events (data-tool-agent)
+          if (part.type === "data-tool-agent") {
+            const agentPart = part as { data?: AgentDataPart }
+            const eventData = agentPart.data
+
+            if (eventData) {
+              allProgressEvents.push({
+                id: `${message.id}-${part.type}-${partIndex}`,
+                stage: "tool agent",
+                status: "in-progress",
+                message: `Agent executing tool`,
+                timestamp: new Date(),
+                data: eventData,
+              })
+            }
+          }
+
+          // Handle nested workflow events (data-tool-workflow)
+          if (part.type === "data-tool-workflow") {
+            const nestedWorkflowPart = part as { data?: { text?: string; status?: string; workflowId?: string } }
+            const eventData = nestedWorkflowPart.data
+
+            if (eventData && eventData.text !== null && typeof eventData.text === "string" && eventData.text.trim()) {
+              allProgressEvents.push({
+                id: `${message.id}-${part.type}-${partIndex}`,
+                stage: "nested workflow",
+                status: "in-progress",
+                message: eventData.text,
+                stepId: eventData.workflowId,
+                timestamp: new Date(),
+                data: eventData,
+              })
+            }
+          }
+
+          // Handle nested network events (data-tool-network)
+          if (part.type === "data-tool-network") {
+            const nestedNetworkPart = part as { data?: { text?: string; status?: string; networkId?: string } }
+            const eventData = nestedNetworkPart.data
+
+            if (eventData && eventData.text !== null && typeof eventData.text === "string" && eventData.text.trim()) {
+              allProgressEvents.push({
+                id: `${message.id}-${part.type}-${partIndex}`,
+                stage: "nested network",
+                status: "in-progress",
+                message: eventData.text,
+                stepId: eventData.networkId,
+                timestamp: new Date(),
+                data: eventData,
+              })
+            }
+          }
+
           // Handle suspend payloads (data-workflow-suspend)
           if (part.type === "data-workflow-suspend") {
             const suspendPart = part as { data?: { message?: string; requestId?: string; stepId?: string } }
@@ -313,11 +406,29 @@ export function WorkflowProvider({
               })
             }
           }
+
+          // Handle custom tool progress events
+          if (typeof part.type === "string" && part.type.startsWith("data-tool-progress")) {
+            const toolProgressPart = part as { type: string; data?: { status?: string; message?: string; toolName?: string } }
+            const eventData = toolProgressPart.data
+
+            if (eventData && eventData.status !== null && typeof eventData.status === "string") {
+              allProgressEvents.push({
+                id: `${message.id}-${part.type}-${partIndex}`,
+                stage: "tool",
+                status: eventData.status === "success" ? "done" : eventData.status === "pending" ? "in-progress" : "error",
+                message: eventData.message ?? `${eventData.toolName ?? 'Tool'} ${eventData.status}`,
+                timestamp: new Date(),
+                data: eventData,
+              })
+            }
+          }
         }
       }
     }
 
     setProgressEvents(allProgressEvents)
+    setDataParts(allDataParts)
     if (detectedSuspendPayload) {
       setSuspendPayload(detectedSuspendPayload)
     }
@@ -331,6 +442,7 @@ export function WorkflowProvider({
       setActiveStepIndex(-1)
       setProgressEvents([])
       setSuspendPayload(null)
+      setDataParts([])
     }
   }, [])
 
@@ -351,6 +463,7 @@ export function WorkflowProvider({
       setActiveStepIndex(0)
       setProgressEvents([]) // Clear previous progress events
       setSuspendPayload(null) // Clear any previous suspend state
+      setDataParts([]) // Clear previous data parts
 
       // Send message to trigger workflow via AI SDK
       const inputText = inputData?.input?.toString() ?? `Run ${workflowConfig.name}`
@@ -394,6 +507,7 @@ export function WorkflowProvider({
     setActiveStepIndex(-1)
     setProgressEvents([])
     setSuspendPayload(null)
+    setDataParts([])
   }, [stop])
 
   const runStep = useCallback(
@@ -475,6 +589,7 @@ export function WorkflowProvider({
       activeStepIndex,
       progressEvents,
       suspendPayload,
+      dataParts,
       selectWorkflow,
       runWorkflow,
       pauseWorkflow,
@@ -496,6 +611,7 @@ export function WorkflowProvider({
       activeStepIndex,
       progressEvents,
       suspendPayload,
+      dataParts,
       selectWorkflow,
       runWorkflow,
       pauseWorkflow,

src/mastra/config/upstash.ts

@@ -18,11 +18,13 @@ export const upstashVector = new UpstashVector({
   token: process.env.UPSTASH_VECTOR_REST_TOKEN ?? 'your-vector-token'
 });
 
-await upstashVector.createIndex({
-  indexName: "vector_messages",
-  dimension: 1536,
-  metric: 'cosine'
-});
+export async function initializeUpstashVector() {
+  await upstashVector.createIndex({
+    indexName: "vector_messages",
+    dimension: 1536,
+    metric: 'cosine'
+  });
+}
 
 /**
  * Shared Mastra agent memory instance using Upstash for distributed storage and [Pinecone] for vector search.
@@ -154,7 +156,7 @@ export const upstashQueryTool = createVectorQueryTool({
     'PostgreSQL vector similarity search using PgVector for semantic content retrieval and question answering.',
   // Supported vector store and index options
   vectorStoreName: 'vector_messages',
-  indexName: 'messages',
+  indexName: 'vector_messages',
   model: google.textEmbedding('gemini-embedding-001'),
   // Supported database configuration for PgVector
   providerOptions: {

src/mastra/tools/document-chunking.tool.ts

@@ -7,6 +7,7 @@ import {
   rerankWithScorer as rerank,
   MastraAgentRelevanceScorer,
 } from '@mastra/rag';
+import { ModelRouterEmbeddingModel } from "@mastra/core/llm";
 import { embed, embedMany } from 'ai';
 import { z } from 'zod';
 import {
@@ -777,6 +778,7 @@ Use this tool to improve retrieval quality by re-ranking initial search results.
           processingTimeMs: processingTime,
         }
       }
+      // Create a relevance scorer
 
       // Step 3: Re-rank results using semantic relevance scorer
       await context?.writer?.custom({ type: 'data-tool-progress', data: { message: `⚖️ Reranking ${initialResults.length} documents` } });
@@ -789,7 +791,7 @@ Use this tool to improve retrieval quality by re-ranking initial search results.
           score: result.score || 0,
         })),
         query: inputData.userQuery,
-        scorer: new MastraAgentRelevanceScorer('relevance-scorer', google('gemini-1.5-flash')),
+        scorer: new MastraAgentRelevanceScorer('relevance-scorer', google('gemini-2.5-flash')),
         options: {
           weights: {
             semantic: inputData.semanticWeight,