Update Pipe Widget documentation and add support for creating and running pipes using DSL

pipe/README.md

@@ -1,5 +1,80 @@
 # Pipe
 
-**Warning: This function under development and is not yet publicly available.**
+Pipe Widget is used for complex logic orchestration, serving as an alternative to Flow.
 
-A new workflow orchestration engine designed to solve complex workflow orchestration problems. This is an alternative to **Flow**.
+**Warning**:
+
+Pipe Widget is an experimental feature and not recommended for production use.
+
+**Usage Scenario**
+
+Generating DSL from a graphical interface, implementing simple functional logic extensions on the application side.
+
+## DSL
+
+CLI: https://github.com/YaoApp/yao-dev-app/blob/main/pipes/cli/translator.pip.yao
+
+WEB: https://github.com/YaoApp/yao-dev-app/blob/main/pipes/web/translator.pip.yao
+
+## Node Types
+
+| Type        | options                      | Description                                               |
+| ----------- | ---------------------------- | --------------------------------------------------------- |
+| Yao Process | `name`, `args`               | run yao process                                           |
+| Switch      |                              | conditional branch                                        |
+| AI          | `prompts`, `model`, `option` | AI interface                                              |
+| Request     |                              | HTTP request (not supported yet, use yao process instead) |
+| User Input  | `ui` (cli/web/...)           | user input interface                                      |
+
+for more details, refer to the DSL demo.
+
+## Process
+
+Refer to unit test programs for examples.
+
+### pipes.<Widget.ID>
+
+Run Pipe
+
+```bash
+yao run pipes.<Widget.ID> [args...]
+```
+
+If interrupted by user input interface, it returns a context ID for resuming execution.
+
+### pipe.run
+
+Run Pipe, equivalent to `pipes.<Widget.ID>`
+
+```bash
+yao run pipe.run <Widget.ID> [args...]
+```
+
+### pipe.create
+
+Pass DSL text to create and run Pipe
+
+```bash
+yao run pipe.create <DSL> [args...]
+```
+
+### pipe.Resume
+
+Resume execution, used for context restoration
+
+```bash
+yao run pipe.Resume <Context.ID> [args...]
+```
+
+### pipe.Close
+
+Close Pipe
+
+```bash
+yao run pipe.Close <Context.ID>
+```
+
+## Todo
+
+[] Progress report for hook integration
+[] Support for Http Request Node

pipe/pipe.go

@@ -43,14 +43,14 @@ func Load(cfg config.Config) error {
 // New create Pipe
 func New(source []byte) (*Pipe, error) {
 	pipe := Pipe{}
-	err := application.Parse("<source>", source, &pipe)
+	err := application.Parse("<source>.yao", source, &pipe)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("parse pipe: %s", err)
 	}
 
 	err = (&pipe).build()
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("build pipe: %s", err)
 	}
 
 	return &pipe, nil

pipe/process.go

@@ -9,6 +9,7 @@ func init() {
 	process.Register("pipes", processPipes)
 	process.RegisterGroup("pipe", map[string]process.Handler{
 		"run":    processRun,
+		"create": processCreate,
 		"resume": processResume,
 		"close":  processClose,
 	})
@@ -22,14 +23,26 @@ func processPipes(process *process.Process) interface{} {
 		exception.New("pipes.%s not loaded", 404, process.ID).Throw()
 		return nil
 	}
-
 	ctx := pipe.Create().WithGlobal(process.Global).WithSid(process.Sid)
-	res, err := ctx.Exec(process.Args...)
+	return ctx.Run(process.Args...)
+}
+
+// processCreate process the create pipe.create <pipe.id> [...args]
+func processCreate(process *process.Process) interface{} {
+	process.ValidateArgNums(1)
+	dsl := process.ArgsString(0)
+	args := []any{}
+	if len(process.Args) > 1 {
+		args = process.Args[1:]
+	}
+
+	pipe, err := New([]byte(dsl))
 	if err != nil {
 		exception.New(err.Error(), 500).Throw()
 	}
 
-	return res
+	ctx := pipe.Create().WithGlobal(process.Global).WithSid(process.Sid)
+	return ctx.Run(args...)
 }
 
 // processRun process the resume pipe.run <pipe.id> [...args]

pipe/process_test.go

@@ -53,6 +53,39 @@ func TestProcessRun(t *testing.T) {
 	assert.Len(t, res.Get("switch"), 2)
 }
 
+func TestProcessCreate(t *testing.T) {
+
+	prepare(t)
+	defer test.Clean()
+
+	dsl := `{
+		"whitelist": ["utils.fmt.Print"],
+		"name": "test",
+		"label": "Test",
+		"nodes": [
+			{
+				"name": "print",
+				"process": {"name":"utils.fmt.Print", "args": "{{ $in }}"},
+				"output": "print"
+			}
+		],
+		"output": {"input": "{{ $input }}" }
+	}`
+
+	p, err := process.Of("pipe.Create", dsl, "hello world")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	output, err := p.Exec()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	res := any.Of(output).Map().MapStrAny.Dot()
+	assert.Equal(t, "hello world", res.Get("input[0]"))
+}
+
 func TestProcessResume(t *testing.T) {
 	prepare(t)
 	defer test.Clean()