wip: feat(engine): dags-as-durable-tasks

api/v1/server/handlers/v1/tasks/get.go

@@ -31,6 +31,12 @@ func (t *TasksService) V1TaskGet(ctx echo.Context, request gen.V1TaskGetRequestO
 		return nil, echo.NewHTTPError(500, "Task type assertion failed")
 	}
 
+	if task.IsDagOrchestrator {
+		return gen.V1TaskGet404JSONResponse{
+			Errors: []gen.APIError{{Description: "task not found"}},
+		}, nil
+	}
+
 	attempt := request.Params.Attempt
 
 	var retryCount *int

api/v1/server/handlers/v1/tasks/list_by_dag_id.go

@@ -35,6 +35,7 @@ func (t *TasksService) V1DagListTasks(ctx echo.Context, request gen.V1DagListTas
 		tenantId,
 		dagIds,
 		false,
+		nil,
 	)
 
 	if err != nil {

api/v1/server/handlers/v1/workflow-runs/get.go

@@ -46,12 +46,17 @@ func (t *V1WorkflowRunsService) getWorkflowRunDetails(
 	taskMetadata := rawWorkflowRun.TaskMetadata
 	workflowRunId := workflowRun.ExternalID
 
-	taskRunEvents, err := t.config.V1.OLAP().ListTaskRunEventsByWorkflowRunId(
-		ctx,
-		tenantId,
-		workflowRunId,
+	var (
+		taskRunEvents []*v1.TaskEventWithPayloads
+		err           error
 	)
 
+	if workflowRun.Kind == sqlcv1.V1RunKindTASK {
+		taskRunEvents, err = t.config.V1.OLAP().ListTaskRunEventsByTaskIds(ctx, tenantId, taskMetadata)
+	} else {
+		taskRunEvents, err = t.config.V1.OLAP().ListTaskRunEventsByWorkflowRunId(ctx, tenantId, workflowRunId)
+	}
+
 	if err != nil {
 		return nil, err
 	}

api/v1/server/handlers/v1/workflow-runs/list.go

@@ -177,18 +177,46 @@ func (t *V1WorkflowRunsService) WithDags(ctx context.Context, request gen.V1Work
 	}
 
 	dagExternalIds := make([]uuid.UUID, 0)
+	dagDurableTaskExternalIds := make(map[uuid.UUID]struct{}, 0)
 
 	for _, dag := range dags {
 		if dag.Kind == sqlcv1.V1RunKindDAG {
 			dagExternalIds = append(dagExternalIds, dag.ExternalID)
 		}
+
+		if dag.Kind == sqlcv1.V1RunKindDAG && dag.IsDagOrchestratorTask {
+			dagDurableTaskExternalIds[dag.ExternalID] = struct{}{}
+		}
+	}
+
+	orchestratorChildExternalIds := make(map[uuid.UUID]struct{})
+	idsInsertedAtsForDagChildren := make([]v1.IdInsertedAt, 0)
+	for _, dag := range dags {
+		if dag.IsDagOrchestratorTask || dag.ParentTaskExternalId == nil {
+			continue
+		}
+		if _, parentIsOrchestrator := dagDurableTaskExternalIds[*dag.ParentTaskExternalId]; parentIsOrchestrator {
+			orchestratorChildExternalIds[dag.ExternalID] = struct{}{}
+			idsInsertedAtsForDagChildren = append(idsInsertedAtsForDagChildren, v1.IdInsertedAt{
+				ID:         dag.ID,
+				InsertedAt: dag.InsertedAt,
+			})
+		}
+	}
+
+	filteredDags := make([]*v1.WorkflowRunData, 0, len(dags))
+	for _, dag := range dags {
+		if _, isChild := orchestratorChildExternalIds[dag.ExternalID]; !isChild {
+			filteredDags = append(filteredDags, dag)
+		}
 	}
 
 	tasks, taskIdToDagExternalId, err := t.config.V1.OLAP().ListTasksByDAGId(
 		ctx,
 		tenantId,
 		dagExternalIds,
 		includePayloads,
+		idsInsertedAtsForDagChildren,
 	)
 
 	if err != nil {
@@ -197,7 +225,7 @@ func (t *V1WorkflowRunsService) WithDags(ctx context.Context, request gen.V1Work
 
 	pgWorkflowIds := make([]uuid.UUID, 0)
 
-	for _, wf := range dags {
+	for _, wf := range filteredDags {
 		pgWorkflowIds = append(pgWorkflowIds, wf.WorkflowID)
 	}
 
@@ -236,7 +264,7 @@ func (t *V1WorkflowRunsService) WithDags(ctx context.Context, request gen.V1Work
 		}
 	}
 
-	result := transformers.ToWorkflowRunMany(dags, dagChildren, taskIdToActionId, workflowNames, total, limit, offset)
+	result := transformers.ToWorkflowRunMany(filteredDags, dagChildren, taskIdToActionId, workflowNames, total, limit, offset)
 
 	// Search for api errors to see how we handle errors in other cases
 	return gen.V1WorkflowRunList200JSONResponse(

cmd/hatchet-engine/engine/run.go

@@ -408,6 +408,7 @@ func runV0Config(ctx context.Context, sc *server.ServerConfig, cleanup *cleanup.
 			adminv1.WithOptimisticSchedulingEnabled(sc.Runtime.OptimisticSchedulingEnabled),
 			adminv1.WithGrpcTriggersEnabled(sc.Runtime.GRPCTriggerWritesEnabled),
 			adminv1.WithGrpcTriggerSlots(sc.Runtime.GRPCTriggerWriteSlots),
+			adminv1.WithDagOperatorEnabled(sc.Runtime.DagOperatorEnabled),
 			adminv1.WithPrometheusGate(sc.PrometheusGate),
 		)
 
@@ -850,6 +851,7 @@ func runV1Config(ctx context.Context, sc *server.ServerConfig, cleanup *cleanup.
 			adminv1.WithOptimisticSchedulingEnabled(sc.Runtime.OptimisticSchedulingEnabled),
 			adminv1.WithGrpcTriggersEnabled(sc.Runtime.GRPCTriggerWritesEnabled),
 			adminv1.WithGrpcTriggerSlots(sc.Runtime.GRPCTriggerWriteSlots),
+			adminv1.WithDagOperatorEnabled(sc.Runtime.DagOperatorEnabled),
 			adminv1.WithPrometheusGate(sc.PrometheusGate),
 		)
 

cmd/hatchet-migrate/migrate/migrations/20260708133531_v1_0_126.sql

@@ -0,0 +1,58 @@
+-- +goose Up
+-- +goose StatementBegin
+ALTER TABLE "Worker" ADD COLUMN "operatorId" UUID;
+
+ALTER TABLE "Step" ADD COLUMN "isDagOrchestrator" BOOLEAN NOT NULL DEFAULT false;
+
+ALTER TABLE v1_task
+    ADD COLUMN is_dag_orchestrator BOOLEAN NOT NULL DEFAULT false,
+    ADD COLUMN is_dag_subtask BOOLEAN NOT NULL DEFAULT false
+;
+
+ALTER TABLE v1_tasks_olap
+    ADD COLUMN is_dag_orchestrator BOOLEAN NOT NULL DEFAULT false,
+    ADD COLUMN is_dag_subtask BOOLEAN NOT NULL DEFAULT false
+;
+
+ALTER TABLE "WorkflowVersion"
+    ADD COLUMN "isUsingDagOperator" BOOLEAN NOT NULL DEFAULT false,
+    ADD COLUMN "dagShape" JSONB
+;
+
+CREATE TYPE v1_operator_kind AS ENUM ('HTTP_API', 'DAG');
+
+CREATE TABLE v1_operator (
+    id UUID NOT NULL DEFAULT gen_random_uuid(),
+    tenant_id UUID NOT NULL,
+    name TEXT NOT NULL,
+    kind v1_operator_kind NOT NULL,
+    config JSONB NOT NULL,
+    worker_id UUID,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    CONSTRAINT v1_operator_pkey PRIMARY KEY (id)
+);
+-- +goose StatementEnd
+
+-- +goose Down
+-- +goose StatementBegin
+ALTER TABLE "Step" DROP COLUMN "isDagOrchestrator";
+ALTER TABLE "WorkflowVersion"
+    DROP COLUMN "isUsingDagOperator",
+    DROP COLUMN "dagShape"
+;
+ALTER TABLE v1_tasks_olap
+    DROP COLUMN is_dag_orchestrator,
+    DROP COLUMN is_dag_subtask
+;
+ALTER TABLE v1_task
+    DROP COLUMN is_dag_orchestrator,
+    DROP COLUMN is_dag_subtask
+;
+
+DROP TABLE v1_operator;
+
+DROP TYPE v1_operator_kind;
+
+ALTER TABLE "Worker" DROP COLUMN "operatorId";
+-- +goose StatementEnd

frontend/app/src/pages/main/v1/workflow-runs-v1/$run/index.tsx

@@ -274,7 +274,7 @@ function ExpandedWorkflowRun({ id }: { id: string }) {
     [open],
   );
 
-  const { workflowRun, shape, taskRuns, isLoading, isError } =
+  const { workflowRun, shape, taskRuns, taskEvents, isLoading, isError } =
     useWorkflowDetails();
 
   const tasksForSynthesis = useMemo((): TaskSummaryForSynthesis[] => {
@@ -303,8 +303,11 @@ function ExpandedWorkflowRun({ id }: { id: string }) {
   );
 
   const durableTaskIds = useMemo(
-    () => taskRuns.filter((t) => t.isDurable).map((t) => t.taskExternalId),
-    [taskRuns],
+    () =>
+      taskRuns
+        .filter((t) => t.isDurable && t.taskExternalId !== id)
+        .map((t) => t.taskExternalId),
+    [taskRuns, id],
   );
 
   if (isLoading || isError || !workflowRun) {
@@ -381,6 +384,7 @@ function ExpandedWorkflowRun({ id }: { id: string }) {
                   fallbackTaskDisplayName={workflowRun.displayName}
                   onClick={handleTaskRunExpand}
                   durableTaskIds={durableTaskIds}
+                  events={taskEvents.filter((e) => e.taskId !== id)}
                 />
               </TabsContent>
               <TabsContent

frontend/app/src/pages/main/v1/workflow-runs-v1/$run/v2components/step-run-events-for-workflow-run.tsx

@@ -36,11 +36,13 @@ export function StepRunEvents({
   workflowRunId,
   isDurable,
   durableTaskIds,
+  events: externalEvents,
 }: {
   taskRunId?: string;
   workflowRunId?: string;
   isDurable?: boolean;
   durableTaskIds?: string[];
+  events?: V1TaskEvent[];
   taskDisplayName?: string;
   fallbackTaskDisplayName?: string;
   onClick?: (stepRunId: string) => void;
@@ -61,7 +63,8 @@ export function StepRunEvents({
       taskRunId,
       workflowRunId,
     ),
-    refetchInterval,
+    enabled: !externalEvents,
+    refetchInterval: externalEvents ? false : refetchInterval,
   });
 
   // fixme: this is an n+1 query, would be better to have a bulk getter
@@ -76,15 +79,15 @@ export function StepRunEvents({
 
   const logs = useMemo(() => {
     const taskLines = toTaskEventLogLines(
-      eventsQuery.data?.rows ?? [],
+      externalEvents ?? eventsQuery.data?.rows ?? [],
       isDag,
       tenantId,
     );
     const durableEventLogLines = durableLogsQueries.flatMap((q) =>
       toDurableEventLogLines(q.data ?? []),
     );
     return mergeByTimestamp(taskLines, durableEventLogLines);
-  }, [eventsQuery.data, isDag, tenantId, durableLogsQueries]);
+  }, [externalEvents, eventsQuery.data, isDag, tenantId, durableLogsQueries]);
 
   const handleTaskRunExpand = useCallback(
     (taskRunId: string) => {
@@ -112,7 +115,7 @@ export function StepRunEvents({
     [open],
   );
 
-  if (eventsQuery.isLoading) {
+  if (!externalEvents && eventsQuery.isLoading) {
     return <Loading />;
   }
 
@@ -327,6 +330,9 @@ function entryMessage(entry: V1DurableEventLogEntry): string {
   const userMessage = entry.userMessage?.trim();
 
   if (entry.kind === V1DurableEventLogKind.RUN) {
+    if (userMessage) {
+      return `spawned child ${userMessage}`;
+    }
     const item = entry.waitData?.[0];
     if (
       entry.waitData?.length === 1 &&
@@ -337,7 +343,7 @@ function entryMessage(entry: V1DurableEventLogEntry): string {
         ? `spawned child ${item.workflowName}`
         : 'spawned child';
     }
-    return userMessage || 'spawned child';
+    return 'spawned child';
   }
   if (userMessage) {
     return userMessage;
@@ -350,6 +356,10 @@ function entryMessage(entry: V1DurableEventLogEntry): string {
 
 function entryCompletionMessage(entry: V1DurableEventLogEntry): string {
   if (entry.kind === V1DurableEventLogKind.RUN) {
+    const userMessage = entry.userMessage?.trim();
+    if (userMessage) {
+      return `child ${userMessage} completed`;
+    }
     const item = entry.waitData?.[0];
     if (
       entry.waitData?.length === 1 &&
@@ -412,6 +422,10 @@ function childNamesFromEntries(
   entries: V1DurableEventLogEntry[],
 ): (string | null)[] {
   return entries.map((e) => {
+    const userMessage = e.userMessage?.trim();
+    if (userMessage) {
+      return userMessage;
+    }
     const item = e.waitData?.[0];
     if (
       e.waitData?.length === 1 &&