add client GetWorkflowSteps

Author: maxdml

dbos/client.go

@@ -31,6 +31,7 @@ type Client interface {
 	CancelWorkflow(workflowID string) error
 	ResumeWorkflow(workflowID string) (WorkflowHandle[any], error)
 	ForkWorkflow(input ForkWorkflowInput) (WorkflowHandle[any], error)
+	GetWorkflowSteps(workflowID string) ([]StepInfo, error)
 	Shutdown(timeout time.Duration) // Simply close the system DB connection pool
 }
 
@@ -295,6 +296,11 @@ func (c *client) ForkWorkflow(input ForkWorkflowInput) (WorkflowHandle[any], err
 	return c.dbosCtx.ForkWorkflow(c.dbosCtx, input)
 }
 
+// GetWorkflowSteps retrieves the execution steps of a workflow.
+func (c *client) GetWorkflowSteps(workflowID string) ([]StepInfo, error) {
+	return c.dbosCtx.GetWorkflowSteps(c.dbosCtx, workflowID)
+}
+
 // Shutdown gracefully shuts down the client and closes the system database connection.
 func (c *client) Shutdown(timeout time.Duration) {
 	// Get the concrete dbosContext to access internal fields

dbos/client_test.go

@@ -928,3 +928,69 @@ func TestListWorkflows(t *testing.T) {
 	// Verify all queue entries are cleaned up
 	require.True(t, queueEntriesAreCleanedUp(serverCtx), "expected queue entries to be cleaned up after list workflows tests")
 }
+
+func TestGetWorkflowSteps(t *testing.T) {
+	// Setup server context
+	serverCtx := setupDBOS(t, true, true)
+
+	// Create queue for communication
+	queue := NewWorkflowQueue(serverCtx, "get-workflow-steps-queue")
+
+	// Workflow with one step
+	stepFunction := func(ctx context.Context) (string, error) {
+		return "abc", nil
+	}
+
+	testWorkflow := func(ctx DBOSContext, input string) (string, error) {
+		result, err := RunAsStep(ctx, stepFunction)
+		if err != nil {
+			return "", err
+		}
+		return result, nil
+	}
+	RegisterWorkflow(serverCtx, testWorkflow, WithWorkflowName("TestWorkflow"))
+
+	// Launch server
+	err := Launch(serverCtx)
+	require.NoError(t, err)
+
+	// Setup client
+	databaseURL := getDatabaseURL()
+	config := ClientConfig{
+		DatabaseURL: databaseURL,
+	}
+	client, err := NewClient(context.Background(), config)
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		if client != nil {
+			client.Shutdown(30 * time.Second)
+		}
+	})
+
+	// Enqueue and run the workflow
+	workflowID := "test-get-workflow-steps"
+	handle, err := Enqueue[string, string](client, queue.Name, "TestWorkflow", "test-input", WithEnqueueWorkflowID(workflowID))
+	require.NoError(t, err)
+
+	// Wait for workflow to complete
+	result, err := handle.GetResult()
+	require.NoError(t, err)
+	assert.Equal(t, "abc", result)
+
+	// Test GetWorkflowSteps with loadOutput = true
+	stepsWithOutput, err := client.GetWorkflowSteps(workflowID)
+	require.NoError(t, err)
+	require.Len(t, stepsWithOutput, 1, "expected exactly 1 step")
+
+	step := stepsWithOutput[0]
+	assert.Equal(t, 0, step.StepID, "expected step ID to be 0")
+	assert.NotEmpty(t, step.StepName, "expected step name to be set")
+	assert.Nil(t, step.Error, "expected no error in step")
+	assert.Equal(t, "", step.ChildWorkflowID, "expected no child workflow ID")
+
+	// Verify the output wasn't loaded
+	require.Nil(t, step.Output, "expected output not to be loaded")
+
+	// Verify all queue entries are cleaned up
+	require.True(t, queueEntriesAreCleanedUp(serverCtx), "expected queue entries to be cleaned up after get workflow steps test")
+}

dbos/system_database.go

@@ -51,7 +51,7 @@ type systemDatabase interface {
 	// Steps
 	recordOperationResult(ctx context.Context, input recordOperationResultDBInput) error
 	checkOperationExecution(ctx context.Context, input checkOperationExecutionDBInput) (*recordedResult, error)
-	getWorkflowSteps(ctx context.Context, workflowID string) ([]StepInfo, error)
+	getWorkflowSteps(ctx context.Context, input getWorkflowStepsInput) ([]StepInfo, error)
 
 	// Communication (special steps)
 	send(ctx context.Context, input WorkflowSendInput) error
@@ -1457,13 +1457,18 @@ type StepInfo struct {
 	ChildWorkflowID string // The ID of a child workflow spawned by this step (if applicable)
 }
 
-func (s *sysDB) getWorkflowSteps(ctx context.Context, workflowID string) ([]StepInfo, error) {
+type getWorkflowStepsInput struct {
+	workflowID string
+	loadOutput bool
+}
+
+func (s *sysDB) getWorkflowSteps(ctx context.Context, input getWorkflowStepsInput) ([]StepInfo, error) {
 	query := fmt.Sprintf(`SELECT function_id, function_name, output, error, child_workflow_id
 			  FROM %s.operation_outputs
 			  WHERE workflow_uuid = $1
 			  ORDER BY function_id ASC`, pgx.Identifier{s.schema}.Sanitize())
 
-	rows, err := s.pool.Query(ctx, query, workflowID)
+	rows, err := s.pool.Query(ctx, query, input.workflowID)
 	if err != nil {
 		return nil, fmt.Errorf("failed to query workflow steps: %w", err)
 	}
@@ -1481,8 +1486,8 @@ func (s *sysDB) getWorkflowSteps(ctx context.Context, workflowID string) ([]Step
 			return nil, fmt.Errorf("failed to scan step row: %w", err)
 		}
 
-		// Deserialize output if present
-		if outputString != nil {
+		// Deserialize output if present and loadOutput is true
+		if input.loadOutput && outputString != nil {
 			output, err := deserialize(outputString)
 			if err != nil {
 				return nil, fmt.Errorf("failed to deserialize output: %w", err)

dbos/workflow.go

@@ -1956,14 +1956,25 @@ func ListWorkflows(ctx DBOSContext, opts ...ListWorkflowsOption) ([]WorkflowStat
 }
 
 func (c *dbosContext) GetWorkflowSteps(_ DBOSContext, workflowID string) ([]StepInfo, error) {
+	var loadOutput bool
+	if c.launched.Load() {
+		loadOutput = true
+	} else {
+		loadOutput = false
+	}
+	getWorkflowStepsInput := getWorkflowStepsInput{
+		workflowID: workflowID,
+		loadOutput: loadOutput,
+	}
+
 	workflowState, ok := c.Value(workflowStateKey).(*workflowState)
 	isWithinWorkflow := ok && workflowState != nil
 	if isWithinWorkflow {
 		return RunAsStep(c, func(ctx context.Context) ([]StepInfo, error) {
-			return c.systemDB.getWorkflowSteps(ctx, workflowID)
+			return c.systemDB.getWorkflowSteps(ctx, getWorkflowStepsInput)
 		}, WithStepName("DBOS.getWorkflowSteps"))
 	} else {
-		return c.systemDB.getWorkflowSteps(c, workflowID)
+		return c.systemDB.getWorkflowSteps(c, getWorkflowStepsInput)
 	}
 }