fix(dify-agent): add reliable shellctl stdio mode (#39997)

dify-agent-runtime/README.md

@@ -14,6 +14,21 @@ cmd/
 internal/            - internal implementations
 ```
 
+## Job execution modes
+
+`POST /v1/jobs/run` accepts an optional `mode` field:
+
+- `pty` (default) keeps the interactive tmux PTY path. stdout and stderr are
+  merged, sanitized, and written to `output.log`; the job accepts `/input`.
+- `stdio` keeps tmux as the lifecycle owner but gives the child `/dev/null` as
+  stdin and captures stdout and stderr through separate pipes. Public output
+  and pagination read stdout from `output.log`; private diagnostics are written
+  to `stderr.log`. A stdio job completes only after both streams reach EOF and
+  does not accept `/input`.
+
+The response models are identical in both modes. Use `stdio` for bounded,
+machine-readable control commands and `pty` for interactive jobs.
+
 ## Building
 
 ```bash

dify-agent-runtime/cmd/runner/main.go

@@ -1,9 +1,9 @@
 // shellctl-runner is the Go replacement for the previously generated bash+python
 // runner script.  It is invoked by tmux as:
 //
-//	shellctl-runner <job_dir> <job_id> <cwd>
+//	shellctl-runner <job_dir> <job_id> <cwd> [pty|stdio]
 //
-// The binary operates in two modes:
+// The binary operates in two process roles:
 //
 //  1. Parent mode (default): waits for start-gate, loads env, forks child,
 //     waits for exit, writes exit artifacts.
@@ -17,16 +17,19 @@ package main
 import (
 	"encoding/json"
 	"fmt"
+	"io"
 	"os"
 	"os/exec"
 	"os/signal"
 	"path/filepath"
 	"strings"
+	"sync"
 	"syscall"
 	"time"
 
 	"github.com/langgenius/dify/dify-agent-runtime/internal/cmdutil"
 	"github.com/langgenius/dify/dify-agent-runtime/internal/envvar"
+	"github.com/langgenius/dify/dify-agent-runtime/internal/jobmode"
 	"github.com/langgenius/dify/dify-agent-runtime/internal/landlock"
 )
 
@@ -39,21 +42,25 @@ func main() {
 }
 
 // parentMode is the entry point when called by tmux.
-// Args: shellctl-runner <job_dir> <job_id> <cwd>
+// Args: shellctl-runner <job_dir> <job_id> <cwd> [pty|stdio]
 func parentMode() {
 	if len(os.Args) < 4 {
-		cmdutil.HandleError(fmt.Errorf("bad args"), 125, "usage: shellctl-runner <job_dir> <job_id> <cwd>")
+		cmdutil.HandleError(fmt.Errorf("bad args"), 125, "usage: shellctl-runner <job_dir> <job_id> <cwd> [pty|stdio]")
 	}
 
 	jobDir := os.Args[1]
 	// jobID := os.Args[2] // unused in parent but passed for compat
 	cwd := os.Args[3]
+	modeRaw := ""
+	if len(os.Args) >= 5 {
+		modeRaw = os.Args[4]
+	}
+	mode, err := jobmode.Parse(modeRaw)
+	cmdutil.HandleError(err, 125, "parse job mode")
 
 	scriptPath := filepath.Join(jobDir, "script")
 	envPath := filepath.Join(jobDir, ".job-env.json")
 	startGate := filepath.Join(jobDir, "start-gate")
-	exitCodePath := filepath.Join(jobDir, "runner-exit-code")
-	endedAtPath := filepath.Join(jobDir, "runner-ended-at")
 
 	// Wait for start-gate.
 	for {
@@ -104,9 +111,6 @@ func parentMode() {
 
 	cmd := exec.Command(self, childArgs...)
 	cmd.Env = env
-	cmd.Stdin = os.Stdin
-	cmd.Stdout = os.Stdout
-	cmd.Stderr = os.Stderr
 	cmd.Dir = cwd
 
 	// Forward signals to child.
@@ -120,23 +124,154 @@ func parentMode() {
 		}
 	}()
 
-	err := cmd.Run()
+	exitCode := runCommandAndRecordExit(cmd, jobDir, mode)
+	os.Exit(exitCode)
+}
 
-	exitCode := 0
-	if err != nil {
-		if exitErr, ok := err.(*exec.ExitError); ok {
-			exitCode = exitErr.ExitCode()
-		} else {
-			exitCode = 125
-		}
+// runCommandAndRecordExit publishes the existing runner artifacts after the
+// child path returns. In stdio mode that return includes both stream drains;
+// PTY mode still relies on its separate pipe-drain finalizer.
+func runCommandAndRecordExit(cmd *exec.Cmd, jobDir string, mode jobmode.Mode) int {
+	var exitCode int
+	if mode == jobmode.Stdio {
+		exitCode = runStdio(cmd, jobDir)
+	} else {
+		cmd.Stdin = os.Stdin
+		cmd.Stdout = os.Stdout
+		cmd.Stderr = os.Stderr
+		exitCode = runPTY(cmd)
 	}
 
 	endedAt := time.Now().UTC().Format("2006-01-02T15:04:05Z")
 
-	writeAtomic(exitCodePath, fmt.Sprintf("%d", exitCode))
-	writeAtomic(endedAtPath, endedAt)
+	writeAtomic(filepath.Join(jobDir, "runner-exit-code"), fmt.Sprintf("%d", exitCode))
+	writeAtomic(filepath.Join(jobDir, "runner-ended-at"), endedAt)
+	return exitCode
+}
 
-	os.Exit(exitCode)
+func runPTY(cmd *exec.Cmd) int {
+	return commandExitCode(cmd.Run())
+}
+
+// runStdio captures stdout and stderr independently and does not return until
+// both streams reach EOF and their files are closed.
+func runStdio(cmd *exec.Cmd, jobDir string) int {
+	outputFile, err := openCaptureFile(filepath.Join(jobDir, "output.log"))
+	if err != nil {
+		return runnerError("open stdout capture", err)
+	}
+	stderrFile, err := openCaptureFile(filepath.Join(jobDir, "stderr.log"))
+	if err != nil {
+		_ = outputFile.Close()
+		return runnerError("open stderr capture", err)
+	}
+
+	stdoutReader, stdoutWriter, err := os.Pipe()
+	if err != nil {
+		_ = outputFile.Close()
+		_ = stderrFile.Close()
+		return runnerError("create stdout pipe", err)
+	}
+	stderrReader, stderrWriter, err := os.Pipe()
+	if err != nil {
+		_ = stdoutReader.Close()
+		_ = stdoutWriter.Close()
+		_ = outputFile.Close()
+		_ = stderrFile.Close()
+		return runnerError("create stderr pipe", err)
+	}
+	stdinFile, err := os.Open(os.DevNull)
+	if err != nil {
+		_ = stdoutReader.Close()
+		_ = stdoutWriter.Close()
+		_ = stderrReader.Close()
+		_ = stderrWriter.Close()
+		_ = outputFile.Close()
+		_ = stderrFile.Close()
+		return runnerError("open stdin", err)
+	}
+
+	cmd.Stdin = stdinFile
+	cmd.Stdout = stdoutWriter
+	cmd.Stderr = stderrWriter
+
+	var captureErrors []error
+	var captureErrorsMu sync.Mutex
+	recordError := func(operation string, err error) {
+		if err != nil {
+			captureErrorsMu.Lock()
+			captureErrors = append(captureErrors, fmt.Errorf("%s: %w", operation, err))
+			captureErrorsMu.Unlock()
+		}
+	}
+
+	var drains sync.WaitGroup
+	drains.Add(2)
+	go func() {
+		defer drains.Done()
+		_, copyErr := io.Copy(outputFile, stdoutReader)
+		recordError("copy stdout", copyErr)
+		recordError("close stdout reader", stdoutReader.Close())
+	}()
+	go func() {
+		defer drains.Done()
+		_, copyErr := io.Copy(stderrFile, stderrReader)
+		recordError("copy stderr", copyErr)
+		recordError("close stderr reader", stderrReader.Close())
+	}()
+
+	startErr := cmd.Start()
+	recordError("close stdout writer", stdoutWriter.Close())
+	recordError("close stderr writer", stderrWriter.Close())
+	recordError("close stdin", stdinFile.Close())
+
+	var waitErr error
+	if startErr != nil {
+		recordError("start child", startErr)
+	} else {
+		waitErr = cmd.Wait()
+	}
+
+	// Descendants may retain either write end after the direct child exits. In
+	// that case the runner intentionally remains alive until both reach EOF.
+	drains.Wait()
+	recordError("close stdout capture", outputFile.Close())
+	recordError("close stderr capture", stderrFile.Close())
+
+	for _, captureErr := range captureErrors {
+		fmt.Fprintf(os.Stderr, "shellctl-runner: %v\n", captureErr)
+	}
+	if len(captureErrors) > 0 {
+		return 125
+	}
+	return commandExitCode(waitErr)
+}
+
+func openCaptureFile(path string) (*os.File, error) {
+	file, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0600)
+	if err != nil {
+		return nil, err
+	}
+	if err := file.Chmod(0600); err != nil {
+		_ = file.Close()
+		return nil, err
+	}
+	return file, nil
+}
+
+func runnerError(operation string, err error) int {
+	fmt.Fprintf(os.Stderr, "shellctl-runner: %s: %v\n", operation, err)
+	return 125
+}
+
+func commandExitCode(err error) int {
+	if err == nil {
+		return 0
+	}
+	if exitErr, ok := err.(*exec.ExitError); ok {
+		return exitErr.ExitCode()
+	}
+	return 125
 }
 
 // childMode applies Landlock (if --landlock flag) and exec's the user script.

dify-agent-runtime/cmd/runner/main_test.go

@@ -0,0 +1,294 @@
+package main
+
+import (
+	"bytes"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strconv"
+	"syscall"
+	"testing"
+	"time"
+
+	"github.com/langgenius/dify/dify-agent-runtime/internal/jobmode"
+)
+
+func TestRunStdioCapturesCompleteSeparatedStreams(t *testing.T) {
+	jobDir := t.TempDir()
+	cmd := exec.Command(os.Args[0], "-test.run=^TestStdioHelperProcess$")
+	cmd.Env = append(os.Environ(), "SHELLCTL_STDIO_HELPER=large-output")
+
+	if exitCode := runStdio(cmd, jobDir); exitCode != 0 {
+		t.Fatalf("runStdio exit code = %d, want 0", exitCode)
+	}
+
+	wantStdout := bytes.Repeat([]byte("stdout-payload\n"), 16*1024)
+	wantStderr := bytes.Repeat([]byte("stderr-payload\n"), 16*1024)
+	gotStdout, err := os.ReadFile(filepath.Join(jobDir, "output.log"))
+	if err != nil {
+		t.Fatalf("read output.log: %v", err)
+	}
+	gotStderr, err := os.ReadFile(filepath.Join(jobDir, "stderr.log"))
+	if err != nil {
+		t.Fatalf("read stderr.log: %v", err)
+	}
+	if !bytes.Equal(gotStdout, wantStdout) {
+		t.Errorf("stdout capture length = %d, want %d", len(gotStdout), len(wantStdout))
+	}
+	if !bytes.Equal(gotStderr, wantStderr) {
+		t.Errorf("stderr capture length = %d, want %d", len(gotStderr), len(wantStderr))
+	}
+	for _, name := range []string{"output.log", "stderr.log"} {
+		info, err := os.Stat(filepath.Join(jobDir, name))
+		if err != nil {
+			t.Fatalf("stat %s: %v", name, err)
+		}
+		if got := info.Mode().Perm(); got != 0600 {
+			t.Errorf("%s permissions = %#o, want 0600", name, got)
+		}
+	}
+}
+
+func TestRunStdioUsesNonTTYStreams(t *testing.T) {
+	jobDir := t.TempDir()
+	cmd := exec.Command("sh", "-c", `if [ -t 0 ] || [ -t 1 ] || [ -t 2 ]; then exit 1; fi; printf 'stdout-only'; printf 'warning' >&2`)
+
+	if exitCode := runStdio(cmd, jobDir); exitCode != 0 {
+		t.Fatalf("runStdio exit code = %d, want 0", exitCode)
+	}
+	stdout, err := os.ReadFile(filepath.Join(jobDir, "output.log"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	stderr, err := os.ReadFile(filepath.Join(jobDir, "stderr.log"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if string(stdout) != "stdout-only" {
+		t.Errorf("stdout = %q, want stdout-only", stdout)
+	}
+	if string(stderr) != "warning" {
+		t.Errorf("stderr = %q, want warning", stderr)
+	}
+}
+
+func TestRunStdioWaitsForBothDescendantStreamsBeforePublishingExit(t *testing.T) {
+	jobDir := t.TempDir()
+	cmd := exec.Command(os.Args[0], "-test.run=^TestStdioHelperProcess$")
+	cmd.Env = mergeEnv(os.Environ(), map[string]string{
+		"SHELLCTL_STDIO_HELPER":  "spawn-descendants",
+		"SHELLCTL_STDIO_JOB_DIR": jobDir,
+	})
+	done := make(chan struct{})
+	var exitCode int
+	go func() {
+		exitCode = runCommandAndRecordExit(cmd, jobDir, jobmode.Stdio)
+		close(done)
+	}()
+
+	stdoutRelease := filepath.Join(jobDir, "release-stdout")
+	stderrRelease := filepath.Join(jobDir, "release-stderr")
+	t.Cleanup(func() {
+		_ = os.WriteFile(stdoutRelease, nil, 0600)
+		_ = os.WriteFile(stderrRelease, nil, 0600)
+		for _, name := range []string{"stdout-closed", "stderr-closed"} {
+			if !waitForPath(filepath.Join(jobDir, name), 5*time.Second) {
+				t.Errorf("cleanup timed out waiting for %s", name)
+			}
+		}
+		select {
+		case <-done:
+		case <-time.After(5 * time.Second):
+			t.Error("cleanup timed out waiting for runner completion")
+		}
+	})
+
+	waitForTestFile(t, filepath.Join(jobDir, "direct-child-exited"))
+	assertRunnerRemainsIncomplete(t, done)
+	assertExitArtifactsAbsent(t, jobDir)
+
+	if err := os.WriteFile(stdoutRelease, nil, 0600); err != nil {
+		t.Fatal(err)
+	}
+	waitForTestFile(t, filepath.Join(jobDir, "stdout-closed"))
+	assertRunnerRemainsIncomplete(t, done)
+	assertExitArtifactsAbsent(t, jobDir)
+
+	if err := os.WriteFile(stderrRelease, nil, 0600); err != nil {
+		t.Fatal(err)
+	}
+	waitForTestFile(t, filepath.Join(jobDir, "stderr-closed"))
+
+	select {
+	case <-done:
+		if exitCode != 0 {
+			t.Fatalf("exit code = %d, want 0", exitCode)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("runner did not complete after both descendant streams reached EOF")
+	}
+	waitForTestFile(t, filepath.Join(jobDir, "runner-exit-code"))
+	waitForTestFile(t, filepath.Join(jobDir, "runner-ended-at"))
+
+	stdout, err := os.ReadFile(filepath.Join(jobDir, "output.log"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	stderr, err := os.ReadFile(filepath.Join(jobDir, "stderr.log"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if string(stdout) != "stdout-tail" {
+		t.Errorf("stdout = %q, want stdout-tail", stdout)
+	}
+	if string(stderr) != "stderr-tail" {
+		t.Errorf("stderr = %q, want stderr-tail", stderr)
+	}
+}
+
+func TestRunStdioPreservesNonZeroExitCode(t *testing.T) {
+	jobDir := t.TempDir()
+	cmd := exec.Command("sh", "-c", "exit 23")
+
+	if exitCode := runCommandAndRecordExit(cmd, jobDir, jobmode.Stdio); exitCode != 23 {
+		t.Fatalf("exit code = %d, want 23", exitCode)
+	}
+	exitCodeArtifact, err := os.ReadFile(filepath.Join(jobDir, "runner-exit-code"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if string(exitCodeArtifact) != "23\n" {
+		t.Errorf("runner-exit-code = %q, want 23", exitCodeArtifact)
+	}
+}
+
+func TestStdioHelperProcess(t *testing.T) {
+	switch os.Getenv("SHELLCTL_STDIO_HELPER") {
+	case "":
+		return
+	case "large-output":
+		stdout := bytes.Repeat([]byte("stdout-payload\n"), 16*1024)
+		stderr := bytes.Repeat([]byte("stderr-payload\n"), 16*1024)
+		_, _ = os.Stdout.Write(stdout)
+		_, _ = os.Stderr.Write(stderr)
+		os.Exit(0)
+	case "spawn-descendants":
+		spawnStdioDescendants()
+	case "hold-stdout", "hold-stderr":
+		holdStdioStream(os.Getenv("SHELLCTL_STDIO_HELPER"))
+	default:
+		os.Exit(125)
+	}
+}
+
+func spawnStdioDescendants() {
+	jobDir := os.Getenv("SHELLCTL_STDIO_JOB_DIR")
+	parentPID := strconv.Itoa(os.Getpid())
+	stdoutHolder := exec.Command(os.Args[0], "-test.run=^TestStdioHelperProcess$")
+	stdoutHolder.Env = mergeEnv(os.Environ(), map[string]string{
+		"SHELLCTL_STDIO_HELPER":     "hold-stdout",
+		"SHELLCTL_STDIO_JOB_DIR":    jobDir,
+		"SHELLCTL_STDIO_PARENT_PID": parentPID,
+	})
+	stdoutHolder.Stdout = os.Stdout
+	if err := stdoutHolder.Start(); err != nil {
+		os.Exit(125)
+	}
+
+	stderrHolder := exec.Command(os.Args[0], "-test.run=^TestStdioHelperProcess$")
+	stderrHolder.Env = mergeEnv(os.Environ(), map[string]string{
+		"SHELLCTL_STDIO_HELPER":  "hold-stderr",
+		"SHELLCTL_STDIO_JOB_DIR": jobDir,
+	})
+	stderrHolder.Stderr = os.Stderr
+	if err := stderrHolder.Start(); err != nil {
+		os.Exit(125)
+	}
+
+	if !waitForPath(filepath.Join(jobDir, "stdout-ready"), 5*time.Second) ||
+		!waitForPath(filepath.Join(jobDir, "stderr-ready"), 5*time.Second) {
+		os.Exit(125)
+	}
+	os.Exit(0)
+}
+
+func holdStdioStream(mode string) {
+	jobDir := os.Getenv("SHELLCTL_STDIO_JOB_DIR")
+	streamName := mode[len("hold-"):]
+	if err := os.WriteFile(filepath.Join(jobDir, streamName+"-ready"), nil, 0600); err != nil {
+		os.Exit(125)
+	}
+	if mode == "hold-stdout" {
+		parentPID, err := strconv.Atoi(os.Getenv("SHELLCTL_STDIO_PARENT_PID"))
+		if err != nil || !waitForProcessExit(parentPID, 5*time.Second) {
+			os.Exit(125)
+		}
+		if err := os.WriteFile(filepath.Join(jobDir, "direct-child-exited"), nil, 0600); err != nil {
+			os.Exit(125)
+		}
+	}
+
+	if !waitForPath(filepath.Join(jobDir, "release-"+streamName), 5*time.Second) {
+		os.Exit(125)
+	}
+	if mode == "hold-stdout" {
+		_, _ = os.Stdout.WriteString("stdout-tail")
+		_ = os.Stdout.Close()
+	} else {
+		_, _ = os.Stderr.WriteString("stderr-tail")
+		_ = os.Stderr.Close()
+	}
+	if err := os.WriteFile(filepath.Join(jobDir, streamName+"-closed"), nil, 0600); err != nil {
+		os.Exit(125)
+	}
+	os.Exit(0)
+}
+
+func waitForProcessExit(pid int, timeout time.Duration) bool {
+	deadline := time.Now().Add(timeout)
+	for time.Now().Before(deadline) {
+		if err := syscall.Kill(pid, 0); err == syscall.ESRCH {
+			return true
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+	return false
+}
+
+func waitForPath(path string, timeout time.Duration) bool {
+	deadline := time.Now().Add(timeout)
+	for time.Now().Before(deadline) {
+		if _, err := os.Stat(path); err == nil {
+			return true
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+	return false
+}
+
+func waitForTestFile(t *testing.T, path string) {
+	t.Helper()
+	if !waitForPath(path, 5*time.Second) {
+		t.Fatalf("timed out waiting for %s", filepath.Base(path))
+	}
+}
+
+func assertRunnerRemainsIncomplete(t *testing.T, done <-chan struct{}) {
+	t.Helper()
+	timer := time.NewTimer(100 * time.Millisecond)
+	defer timer.Stop()
+	select {
+	case <-done:
+		t.Fatal("runner completed before both streams reached EOF")
+	case <-timer.C:
+	}
+}
+
+func assertExitArtifactsAbsent(t *testing.T, jobDir string) {
+	t.Helper()
+	for _, name := range []string{"runner-exit-code", "runner-ended-at"} {
+		if _, err := os.Stat(filepath.Join(jobDir, name)); !os.IsNotExist(err) {
+			t.Fatalf("%s became visible before both streams reached EOF: %v", name, err)
+		}
+	}
+}

dify-agent-runtime/internal/jobmode/mode.go

@@ -0,0 +1,29 @@
+// Package jobmode defines the execution modes shared by the shellctl server
+// and runner.
+package jobmode
+
+import "fmt"
+
+// Mode selects how the runner connects a job's standard streams.
+type Mode string
+
+const (
+	PTY   Mode = "pty"
+	Stdio Mode = "stdio"
+)
+
+// Parse validates a mode received at a process or API boundary. An empty value
+// preserves the historical PTY behavior for callers that omit the mode.
+func Parse(raw string) (Mode, error) {
+	if raw == "" {
+		return PTY, nil
+	}
+
+	mode := Mode(raw)
+	switch mode {
+	case PTY, Stdio:
+		return mode, nil
+	default:
+		return "", fmt.Errorf("invalid job mode %q", raw)
+	}
+}

dify-agent-runtime/internal/jobmode/mode_test.go

@@ -0,0 +1,29 @@
+package jobmode
+
+import "testing"
+
+func TestParse(t *testing.T) {
+	tests := []struct {
+		name    string
+		raw     string
+		want    Mode
+		wantErr bool
+	}{
+		{name: "omitted", raw: "", want: PTY},
+		{name: "pty", raw: "pty", want: PTY},
+		{name: "stdio", raw: "stdio", want: Stdio},
+		{name: "unknown", raw: "stdout", wantErr: true},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := Parse(tt.raw)
+			if (err != nil) != tt.wantErr {
+				t.Fatalf("Parse(%q) error = %v, wantErr %v", tt.raw, err, tt.wantErr)
+			}
+			if got != tt.want {
+				t.Errorf("Parse(%q) = %q, want %q", tt.raw, got, tt.want)
+			}
+		})
+	}
+}

dify-agent-runtime/internal/server/api.go

@@ -9,6 +9,8 @@ import (
 	"strconv"
 	"strings"
 	"time"
+
+	"github.com/langgenius/dify/dify-agent-runtime/internal/jobmode"
 )
 
 // Handler creates the HTTP handler (mux) for the shellctl API.
@@ -45,6 +47,12 @@ func handleRunJob(svc *Service) http.HandlerFunc {
 			writeError(w, 400, "invalid_request", "script is required")
 			return
 		}
+		mode, err := jobmode.Parse(string(req.Mode))
+		if err != nil {
+			writeError(w, 422, "validation_error", err.Error())
+			return
+		}
+		req.Mode = mode
 		// Validate env
 		if req.Env != nil {
 			for name, value := range req.Env {

dify-agent-runtime/internal/server/api_test.go

@@ -152,6 +152,25 @@ func TestHealthzHandler(t *testing.T) {
 	}
 }
 
+func TestRunJobRejectsInvalidModeBeforeCallingService(t *testing.T) {
+	handler := handleRunJob(nil)
+	req := httptest.NewRequest("POST", "/v1/jobs/run", strings.NewReader(`{"script":"true","mode":"stdout"}`))
+	w := httptest.NewRecorder()
+
+	handler(w, req)
+
+	if w.Code != http.StatusUnprocessableEntity {
+		t.Fatalf("expected 422, got %d", w.Code)
+	}
+	var result ErrorResponse
+	if err := json.NewDecoder(w.Body).Decode(&result); err != nil {
+		t.Fatal(err)
+	}
+	if result.Error.Code != "validation_error" {
+		t.Errorf("expected validation_error, got %q", result.Error.Code)
+	}
+}
+
 func TestServerErrorFormat(t *testing.T) {
 	err := NewServerError(422, "validation_error", "bad input")
 	expected := "[422] validation_error: bad input"

dify-agent-runtime/internal/server/db.go

@@ -5,9 +5,19 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/langgenius/dify/dify-agent-runtime/internal/jobmode"
+
 	_ "modernc.org/sqlite"
 )
 
+const latestSchemaVersion = 1
+
+type schemaMigration func(*sql.Tx) error
+
+var schemaMigrations = []schemaMigration{
+	migrateSchemaV1,
+}
+
 // JobStatusName represents the lifecycle states of a shellctl job.
 type JobStatusName string
 
@@ -35,6 +45,7 @@ type JobRow struct {
 	JobID        string
 	ScriptPath   string
 	OutputPath   string
+	Mode         jobmode.Mode
 	Cwd          string
 	TerminalCols int
 	TerminalRows int
@@ -72,9 +83,63 @@ func (d *DB) Close() error {
 	return d.db.Close()
 }
 
-// InitSchema creates the jobs table if it does not exist.
+// InitSchema creates the v0 baseline and applies pending schema migrations.
 func (d *DB) InitSchema() error {
-	_, err := d.db.Exec(`
+	var currentVersion int
+	if err := d.db.QueryRow("PRAGMA user_version").Scan(&currentVersion); err != nil {
+		return fmt.Errorf("read schema version: %w", err)
+	}
+	if currentVersion > latestSchemaVersion {
+		return fmt.Errorf(
+			"database schema version %d is newer than supported version %d",
+			currentVersion,
+			latestSchemaVersion,
+		)
+	}
+
+	if currentVersion == 0 {
+		if err := d.createSchemaV0(); err != nil {
+			return err
+		}
+	}
+
+	for currentVersion < latestSchemaVersion {
+		targetVersion := currentVersion + 1
+		if err := d.applySchemaMigration(targetVersion, schemaMigrations[targetVersion-1]); err != nil {
+			return err
+		}
+		currentVersion = targetVersion
+	}
+	return nil
+}
+
+// applySchemaMigration commits migration SQL and PRAGMA user_version together,
+// rolling back the entire version if either operation fails.
+func (d *DB) applySchemaMigration(targetVersion int, migration schemaMigration) error {
+	tx, err := d.db.Begin()
+	if err != nil {
+		return fmt.Errorf("begin schema migration v%d: %w", targetVersion, err)
+	}
+	if err := migration(tx); err != nil {
+		_ = tx.Rollback()
+		return fmt.Errorf("apply schema migration v%d: %w", targetVersion, err)
+	}
+	if _, err := tx.Exec(fmt.Sprintf("PRAGMA user_version = %d", targetVersion)); err != nil {
+		_ = tx.Rollback()
+		return fmt.Errorf("record schema migration v%d: %w", targetVersion, err)
+	}
+	if err := tx.Commit(); err != nil {
+		return fmt.Errorf("commit schema migration v%d: %w", targetVersion, err)
+	}
+	return nil
+}
+
+func (d *DB) createSchemaV0() error {
+	tx, err := d.db.Begin()
+	if err != nil {
+		return fmt.Errorf("begin schema v0: %w", err)
+	}
+	if _, err := tx.Exec(`
 		CREATE TABLE IF NOT EXISTS jobs (
 			job_id       TEXT PRIMARY KEY,
 			script_path  TEXT NOT NULL,
@@ -93,18 +158,29 @@ func (d *DB) InitSchema() error {
 			ended_at     TEXT,
 			updated_at   TEXT NOT NULL
 		)
-	`)
+	`); err != nil {
+		_ = tx.Rollback()
+		return fmt.Errorf("create schema v0: %w", err)
+	}
+	if err := tx.Commit(); err != nil {
+		return fmt.Errorf("commit schema v0: %w", err)
+	}
+	return nil
+}
+
+func migrateSchemaV1(tx *sql.Tx) error {
+	_, err := tx.Exec(`ALTER TABLE jobs ADD COLUMN mode TEXT NOT NULL DEFAULT 'pty'`)
 	return err
 }
 
 // InsertJob inserts a new job row. Returns false if the job_id already exists.
 func (d *DB) InsertJob(row *JobRow) (bool, error) {
 	_, err := d.db.Exec(`
-		INSERT INTO jobs (job_id, script_path, output_path, cwd, terminal_cols, terminal_rows,
+		INSERT INTO jobs (job_id, script_path, output_path, mode, cwd, terminal_cols, terminal_rows,
 			status, session_name, pane_target, exit_code, reason, message,
 			created_at, started_at, ended_at, updated_at)
-		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
-		row.JobID, row.ScriptPath, row.OutputPath, row.Cwd,
+		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+		row.JobID, row.ScriptPath, row.OutputPath, string(row.Mode), row.Cwd,
 		row.TerminalCols, row.TerminalRows,
 		string(row.Status), row.SessionName, row.PaneTarget,
 		row.ExitCode, row.Reason, row.Message,
@@ -123,7 +199,7 @@ func (d *DB) InsertJob(row *JobRow) (bool, error) {
 // GetJob retrieves a single job row by ID.
 func (d *DB) GetJob(jobID string) (*JobRow, error) {
 	row := d.db.QueryRow(`
-		SELECT job_id, script_path, output_path, cwd, terminal_cols, terminal_rows,
+		SELECT job_id, script_path, output_path, mode, cwd, terminal_cols, terminal_rows,
 			status, session_name, pane_target, exit_code, reason, message,
 			created_at, started_at, ended_at, updated_at
 		FROM jobs WHERE job_id = ?`, jobID)
@@ -136,7 +212,7 @@ func (d *DB) ListJobs(statuses []JobStatusName) ([]*JobRow, error) {
 	var err error
 
 	if len(statuses) == 0 {
-		rows, err = d.db.Query(`SELECT job_id, script_path, output_path, cwd,
+		rows, err = d.db.Query(`SELECT job_id, script_path, output_path, mode, cwd,
 			terminal_cols, terminal_rows, status, session_name, pane_target,
 			exit_code, reason, message, created_at, started_at, ended_at, updated_at
 			FROM jobs ORDER BY created_at DESC`)
@@ -150,7 +226,7 @@ func (d *DB) ListJobs(statuses []JobStatusName) ([]*JobRow, error) {
 			}
 			placeholders += "?"
 		}
-		query := fmt.Sprintf(`SELECT job_id, script_path, output_path, cwd,
+		query := fmt.Sprintf(`SELECT job_id, script_path, output_path, mode, cwd,
 			terminal_cols, terminal_rows, status, session_name, pane_target,
 			exit_code, reason, message, created_at, started_at, ended_at, updated_at
 			FROM jobs WHERE status IN (%s) ORDER BY created_at DESC`, placeholders)
@@ -313,9 +389,9 @@ type TransitionOpts struct {
 
 func scanJobRow(row *sql.Row) (*JobRow, error) {
 	var jr JobRow
-	var status string
+	var mode, status string
 	err := row.Scan(
-		&jr.JobID, &jr.ScriptPath, &jr.OutputPath, &jr.Cwd,
+		&jr.JobID, &jr.ScriptPath, &jr.OutputPath, &mode, &jr.Cwd,
 		&jr.TerminalCols, &jr.TerminalRows,
 		&status, &jr.SessionName, &jr.PaneTarget,
 		&jr.ExitCode, &jr.Reason, &jr.Message,
@@ -327,15 +403,16 @@ func scanJobRow(row *sql.Row) (*JobRow, error) {
 	if err != nil {
 		return nil, err
 	}
+	jr.Mode = jobmode.Mode(mode)
 	jr.Status = JobStatusName(status)
 	return &jr, nil
 }
 
 func scanJobRows(rows *sql.Rows) (*JobRow, error) {
 	var jr JobRow
-	var status string
+	var mode, status string
 	err := rows.Scan(
-		&jr.JobID, &jr.ScriptPath, &jr.OutputPath, &jr.Cwd,
+		&jr.JobID, &jr.ScriptPath, &jr.OutputPath, &mode, &jr.Cwd,
 		&jr.TerminalCols, &jr.TerminalRows,
 		&status, &jr.SessionName, &jr.PaneTarget,
 		&jr.ExitCode, &jr.Reason, &jr.Message,
@@ -344,6 +421,7 @@ func scanJobRows(rows *sql.Rows) (*JobRow, error) {
 	if err != nil {
 		return nil, err
 	}
+	jr.Mode = jobmode.Mode(mode)
 	jr.Status = JobStatusName(status)
 	return &jr, nil
 }

dify-agent-runtime/internal/server/db_test.go

@@ -1,9 +1,13 @@
 package server
 
 import (
+	"database/sql"
+	"errors"
 	"os"
 	"path/filepath"
 	"testing"
+
+	"github.com/langgenius/dify/dify-agent-runtime/internal/jobmode"
 )
 
 func TestJobStatusIsTerminal(t *testing.T) {
@@ -40,6 +44,7 @@ func TestOpenDBAndInitSchema(t *testing.T) {
 		JobID:        "test-job-1",
 		ScriptPath:   "jobs/test-job-1/script",
 		OutputPath:   "jobs/test-job-1/output.log",
+		Mode:         jobmode.PTY,
 		Cwd:          "/tmp",
 		TerminalCols: 80,
 		TerminalRows: 24,
@@ -66,6 +71,104 @@ func TestOpenDBAndInitSchema(t *testing.T) {
 	if ok {
 		t.Error("expected duplicate insert to return ok=false")
 	}
+	if got := schemaVersion(t, db); got != latestSchemaVersion {
+		t.Errorf("schema version = %d, want %d", got, latestSchemaVersion)
+	}
+}
+
+func TestInitSchemaMigratesV0JobsToPTY(t *testing.T) {
+	db := openTestDB(t, t.TempDir())
+	defer func() { _ = db.Close() }()
+
+	if err := db.createSchemaV0(); err != nil {
+		t.Fatalf("createSchemaV0: %v", err)
+	}
+	_, err := db.db.Exec(`
+		INSERT INTO jobs (
+			job_id, script_path, output_path, cwd, terminal_cols, terminal_rows,
+			status, session_name, pane_target, created_at, updated_at
+		) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+		"legacy-job", "jobs/legacy-job/script", "jobs/legacy-job/output.log", "/tmp",
+		80, 24, "created", "shellctl-legacy-job", "shellctl-legacy-job:0.0",
+		"2025-01-01T00:00:00Z", "2025-01-01T00:00:00Z",
+	)
+	if err != nil {
+		t.Fatalf("insert legacy job: %v", err)
+	}
+
+	if err := db.InitSchema(); err != nil {
+		t.Fatalf("InitSchema: %v", err)
+	}
+	row, err := db.GetJob("legacy-job")
+	if err != nil {
+		t.Fatalf("GetJob: %v", err)
+	}
+	if row.Mode != jobmode.PTY {
+		t.Errorf("legacy job mode = %q, want %q", row.Mode, jobmode.PTY)
+	}
+	if got := schemaVersion(t, db); got != latestSchemaVersion {
+		t.Errorf("schema version = %d, want %d", got, latestSchemaVersion)
+	}
+}
+
+func TestInitSchemaAtLatestVersionIsIdempotent(t *testing.T) {
+	db := setupTestDB(t, t.TempDir())
+	defer func() { _ = db.Close() }()
+
+	if err := db.InitSchema(); err != nil {
+		t.Fatalf("second InitSchema: %v", err)
+	}
+	if got := schemaVersion(t, db); got != latestSchemaVersion {
+		t.Errorf("schema version = %d, want %d", got, latestSchemaVersion)
+	}
+}
+
+func TestInitSchemaRejectsNewerDatabaseWithoutDDL(t *testing.T) {
+	db := openTestDB(t, t.TempDir())
+	defer func() { _ = db.Close() }()
+	if _, err := db.db.Exec("PRAGMA user_version = 2"); err != nil {
+		t.Fatalf("set future schema version: %v", err)
+	}
+
+	if err := db.InitSchema(); err == nil {
+		t.Fatal("InitSchema unexpectedly accepted a newer schema")
+	}
+	var tableCount int
+	if err := db.db.QueryRow(`SELECT count(*) FROM sqlite_master WHERE type='table' AND name='jobs'`).Scan(&tableCount); err != nil {
+		t.Fatalf("query jobs table: %v", err)
+	}
+	if tableCount != 0 {
+		t.Errorf("jobs table count = %d, want 0", tableCount)
+	}
+}
+
+func TestApplySchemaMigrationFailureRollsBackDDLAndVersion(t *testing.T) {
+	db := openTestDB(t, t.TempDir())
+	defer func() { _ = db.Close() }()
+	if err := db.createSchemaV0(); err != nil {
+		t.Fatalf("createSchemaV0: %v", err)
+	}
+
+	sentinel := errors.New("sentinel migration failure")
+	err := db.applySchemaMigration(1, func(tx *sql.Tx) error {
+		if _, err := tx.Exec(`ALTER TABLE jobs ADD COLUMN rollback_probe TEXT`); err != nil {
+			return err
+		}
+		return sentinel
+	})
+	if !errors.Is(err, sentinel) {
+		t.Fatalf("migration error = %v, want sentinel failure", err)
+	}
+	var probeColumns int
+	if err := db.db.QueryRow(`SELECT count(*) FROM pragma_table_info('jobs') WHERE name = 'rollback_probe'`).Scan(&probeColumns); err != nil {
+		t.Fatalf("query rollback probe column: %v", err)
+	}
+	if probeColumns != 0 {
+		t.Errorf("rollback_probe column count = %d, want 0", probeColumns)
+	}
+	if got := schemaVersion(t, db); got != 0 {
+		t.Errorf("schema version = %d, want 0", got)
+	}
 }
 
 func TestGetJob(t *testing.T) {
@@ -88,6 +191,9 @@ func TestGetJob(t *testing.T) {
 	if row.TerminalCols != 80 {
 		t.Errorf("expected cols=80, got %d", row.TerminalCols)
 	}
+	if row.Mode != jobmode.PTY {
+		t.Errorf("expected mode=pty, got %s", row.Mode)
+	}
 }
 
 func TestGetJobNotFound(t *testing.T) {
@@ -243,6 +349,7 @@ func TestRecordRunnerExitIdempotent(t *testing.T) {
 	exitCode := 10
 	row := &JobRow{
 		JobID: "job-exit-2", ScriptPath: "x", OutputPath: "y", Cwd: "/tmp",
+		Mode:         jobmode.PTY,
 		TerminalCols: 80, TerminalRows: 24, Status: StatusExited,
 		SessionName: "s", PaneTarget: "p", ExitCode: &exitCode,
 		CreatedAt: "2025-01-01T00:00:00Z", UpdatedAt: "2025-01-01T00:00:00Z",
@@ -267,24 +374,40 @@ func TestRecordRunnerExitIdempotent(t *testing.T) {
 // Helpers
 
 func setupTestDB(t *testing.T, dir string) *DB {
+	t.Helper()
+	db := openTestDB(t, dir)
+	if err := db.InitSchema(); err != nil {
+		t.Fatalf("InitSchema: %v", err)
+	}
+	return db
+}
+
+func openTestDB(t *testing.T, dir string) *DB {
 	t.Helper()
 	dbPath := filepath.Join(dir, "shellctl.db")
 	db, err := OpenDB(dbPath, 5000)
 	if err != nil {
 		t.Fatalf("OpenDB: %v", err)
 	}
-	if err := db.InitSchema(); err != nil {
-		t.Fatalf("InitSchema: %v", err)
-	}
 	return db
 }
 
+func schemaVersion(t *testing.T, db *DB) int {
+	t.Helper()
+	var version int
+	if err := db.db.QueryRow("PRAGMA user_version").Scan(&version); err != nil {
+		t.Fatalf("read schema version: %v", err)
+	}
+	return version
+}
+
 func insertTestJob(t *testing.T, db *DB, jobID string, status JobStatusName) {
 	t.Helper()
 	row := &JobRow{
 		JobID:        jobID,
 		ScriptPath:   "jobs/" + jobID + "/script",
 		OutputPath:   "jobs/" + jobID + "/output.log",
+		Mode:         jobmode.PTY,
 		Cwd:          "/tmp",
 		TerminalCols: 80,
 		TerminalRows: 24,

dify-agent-runtime/internal/server/service.go

@@ -11,6 +11,8 @@ import (
 	"strings"
 	"sync"
 	"time"
+
+	"github.com/langgenius/dify/dify-agent-runtime/internal/jobmode"
 )
 
 // Service is the core job lifecycle manager backed by SQLite and tmux.
@@ -105,7 +107,9 @@ func (s *Service) StartBackgroundGC() {
 	}()
 }
 
-// StartBackgroundPipeMonitor starts the periodic pipe health check goroutine.
+// StartBackgroundPipeMonitor periodically reconciles mode-aware runtime state:
+// PTY pane pipe/drain state, and stdio session state plus completion
+// materialization from exit artifacts.
 func (s *Service) StartBackgroundPipeMonitor() {
 	ctx, cancel := context.WithCancel(context.Background())
 	s.cancelMon = cancel
@@ -194,6 +198,7 @@ func (s *Service) RunJob(req *RunJobRequest) (*JobResult, error) {
 		JobID:        jobID,
 		ScriptPath:   fmt.Sprintf("jobs/%s/script", jobID),
 		OutputPath:   fmt.Sprintf("jobs/%s/output.log", jobID),
+		Mode:         req.Mode,
 		Cwd:          cwd,
 		TerminalCols: cols,
 		TerminalRows: rows,
@@ -220,9 +225,10 @@ func (s *Service) RunJob(req *RunJobRequest) (*JobResult, error) {
 		Target:      StatusStarting,
 	})
 
-	// Create tmux session and enable output pipe
+	// Start the mode-specific runtime: PTY manages pane piping and drain, while
+	// stdio tracks the session and materializes completion from exit artifacts.
 	log.Printf("RunJob [%s]: starting job, cwd=%s", jobID, cwd)
-	startErr := s.startJob(jobID, jobDir, cwd, cols, rows)
+	startErr := s.startJob(jobID, jobDir, cwd, cols, rows, req.Mode)
 	if startErr != nil {
 		log.Printf("RunJob [%s]: start failed: %v", jobID, startErr)
 		reason := "start_failed"
@@ -251,24 +257,26 @@ func (s *Service) RunJob(req *RunJobRequest) (*JobResult, error) {
 	})
 }
 
-func (s *Service) startJob(jobID, jobDir, cwd string, cols, rows int) error {
+func (s *Service) startJob(jobID, jobDir, cwd string, cols, rows int, mode jobmode.Mode) error {
 	log.Printf("startJob [%s]: creating tmux session", jobID)
-	if err := s.tmux.CreateJobSession(jobID, jobDir, cwd, cols, rows); err != nil {
+	if err := s.tmux.CreateJobSession(jobID, jobDir, cwd, cols, rows, mode); err != nil {
 		log.Printf("startJob [%s]: tmux session failed: %v", jobID, err)
 		return err
 	}
 
 	pipeReadyPath := filepath.Join(jobDir, ".pipe-ready")
-	log.Printf("startJob [%s]: enabling output pipe", jobID)
-	if err := s.tmux.EnableOutputPipe(jobID, jobDir, pipeReadyPath); err != nil {
-		log.Printf("startJob [%s]: pipe-pane failed: %v", jobID, err)
-		return err
-	}
+	if mode == jobmode.PTY {
+		log.Printf("startJob [%s]: enabling output pipe", jobID)
+		if err := s.tmux.EnableOutputPipe(jobID, jobDir, pipeReadyPath); err != nil {
+			log.Printf("startJob [%s]: pipe-pane failed: %v", jobID, err)
+			return err
+		}
 
-	// Wait for pipe ready handshake
-	if err := s.waitForPipeReady(jobID, pipeReadyPath); err != nil {
-		log.Printf("startJob [%s]: pipe-ready timeout: %v", jobID, err)
-		return err
+		// Wait for pipe ready handshake
+		if err := s.waitForPipeReady(jobID, pipeReadyPath); err != nil {
+			log.Printf("startJob [%s]: pipe-ready timeout: %v", jobID, err)
+			return err
+		}
 	}
 
 	// Open start gate
@@ -285,8 +293,9 @@ func (s *Service) startJob(jobID, jobDir, cwd string, cols, rows int) error {
 		RequireExitCodeNull: true,
 	})
 
-	// Clean up ready file
-	_ = os.Remove(pipeReadyPath)
+	if mode == jobmode.PTY {
+		_ = os.Remove(pipeReadyPath)
+	}
 	return nil
 }
 
@@ -425,11 +434,15 @@ func (s *Service) TailJob(jobID string, outputLimit int) (*JobResult, error) {
 
 // GetJobStatus materializes the current status from SQLite + live tmux state.
 func (s *Service) GetJobStatus(jobID string) (*JobStatusView, error) {
-	sessionExists, pipeActive, err := s.liveRuntimeState(jobID)
+	row, err := s.db.GetJob(jobID)
 	if err != nil {
 		return nil, err
 	}
-	return s.materializeStatusView(jobID, sessionExists, pipeActive)
+	sessionExists, pipeActive, err := s.liveRuntimeState(row)
+	if err != nil {
+		return nil, err
+	}
+	return s.materializeStatusView(row, sessionExists, pipeActive)
 }
 
 // ListJobs returns recent jobs, optionally filtered by status.
@@ -474,6 +487,13 @@ func (s *Service) SendInput(jobID string, req *InputJobRequest) (*JobResult, err
 	if view.Done {
 		return nil, NewServerError(409, "job_not_running", fmt.Sprintf("Job %s is already terminal", jobID))
 	}
+	row, err := s.db.GetJob(jobID)
+	if err != nil {
+		return nil, err
+	}
+	if row.Mode == jobmode.Stdio {
+		return nil, NewServerError(409, "input_unsupported", "stdio jobs do not support input")
+	}
 
 	if err := s.tmux.SendInput(jobID, req.Text); err != nil {
 		// Check if job became terminal in the meantime
@@ -594,7 +614,8 @@ func (s *Service) GCOnce() error {
 	return nil
 }
 
-// CheckRunningJobsPipeHealth fails running jobs whose pipe died.
+// CheckRunningJobsPipeHealth reconciles PTY pane pipe/drain state and stdio
+// session state, materializing completion from exit artifacts.
 func (s *Service) CheckRunningJobsPipeHealth() {
 	rows, _ := s.db.ListJobs([]JobStatusName{StatusRunning})
 	for _, row := range rows {
@@ -602,12 +623,8 @@ func (s *Service) CheckRunningJobsPipeHealth() {
 	}
 }
 
-func (s *Service) materializeStatusView(jobID string, sessionExists bool, pipeActive *bool) (*JobStatusView, error) {
-	row, err := s.db.GetJob(jobID)
-	if err != nil {
-		return nil, err
-	}
-
+func (s *Service) materializeStatusView(row *JobRow, sessionExists bool, pipeActive *bool) (*JobStatusView, error) {
+	jobID := row.JobID
 	status := row.Status
 
 	if status.IsTerminal() {
@@ -631,14 +648,14 @@ func (s *Service) materializeStatusView(jobID string, sessionExists bool, pipeAc
 		}); err == nil {
 			row = r
 		}
-	} else if exit := s.drainedNormalExitMetadata(jobID); exit != nil {
-		// Recover from drained exit artifacts
+	} else if exit := s.completedExitMetadata(row); exit != nil {
+		// Recover from mode-specific completed exit artifacts.
 		_ = s.db.RecordRunnerExit(jobID, exit.exitCode, exit.endedAt)
 		if r, err := s.db.GetJob(jobID); err == nil {
 			row = r
 		}
 	} else if sessionExists {
-		if pipeActive != nil && !*pipeActive {
+		if row.Mode == jobmode.PTY && pipeActive != nil && !*pipeActive {
 			s.mu.Lock()
 			isStarting := s.startingJobs[jobID]
 			s.mu.Unlock()
@@ -670,7 +687,7 @@ func (s *Service) materializeStatusView(jobID string, sessionExists bool, pipeAc
 		}
 	} else {
 		// No session
-		if s.normalExitCommitPending(jobID) {
+		if s.normalExitCommitPending(row) {
 			// Wait for pipe drain finalizer
 		} else {
 			s.mu.Lock()
@@ -694,15 +711,18 @@ func (s *Service) materializeStatusView(jobID string, sessionExists bool, pipeAc
 	return s.statusViewFromRow(row), nil
 }
 
-func (s *Service) liveRuntimeState(jobID string) (bool, *bool, error) {
-	exists, err := s.tmux.SessionExists(JobSessionName(jobID))
+func (s *Service) liveRuntimeState(row *JobRow) (bool, *bool, error) {
+	exists, err := s.tmux.SessionExists(row.SessionName)
 	if err != nil {
 		return false, nil, err
 	}
 	if !exists {
 		return false, nil, nil
 	}
-	active, err := s.tmux.IsOutputPipeActive(jobID)
+	if row.Mode == jobmode.Stdio {
+		return true, nil, nil
+	}
+	active, err := s.tmux.IsOutputPipeActive(row.JobID)
 	if err != nil {
 		return false, nil, err
 	}
@@ -717,13 +737,16 @@ type exitMetadata struct {
 	endedAt  string
 }
 
-func (s *Service) drainedNormalExitMetadata(jobID string) *exitMetadata {
-	jobDir := filepath.Join(s.config.JobsDir(), jobID)
+func (s *Service) completedExitMetadata(row *JobRow) *exitMetadata {
+	jobDir := filepath.Join(s.config.JobsDir(), row.JobID)
 	drainedPath := filepath.Join(jobDir, ".pipe-drained")
 	exitCodePath := filepath.Join(jobDir, "runner-exit-code")
 	endedAtPath := filepath.Join(jobDir, "runner-ended-at")
 
-	if !fileExists(drainedPath) || !fileExists(exitCodePath) || !fileExists(endedAtPath) {
+	if row.Mode == jobmode.PTY && !fileExists(drainedPath) {
+		return nil
+	}
+	if !fileExists(exitCodePath) || !fileExists(endedAtPath) {
 		return nil
 	}
 
@@ -750,8 +773,11 @@ func (s *Service) drainedNormalExitMetadata(jobID string) *exitMetadata {
 	return &exitMetadata{exitCode: code, endedAt: endedAtStr}
 }
 
-func (s *Service) normalExitCommitPending(jobID string) bool {
-	jobDir := filepath.Join(s.config.JobsDir(), jobID)
+func (s *Service) normalExitCommitPending(row *JobRow) bool {
+	if row.Mode != jobmode.PTY {
+		return false
+	}
+	jobDir := filepath.Join(s.config.JobsDir(), row.JobID)
 	return fileExists(filepath.Join(jobDir, "runner-exit-code")) &&
 		fileExists(filepath.Join(jobDir, "runner-ended-at")) &&
 		!fileExists(filepath.Join(jobDir, ".pipe-drained")) &&

dify-agent-runtime/internal/server/service_test.go

@@ -0,0 +1,111 @@
+package server
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/langgenius/dify/dify-agent-runtime/internal/jobmode"
+)
+
+func TestMaterializeStdioStatusDoesNotRequirePanePipe(t *testing.T) {
+	service, row := setupModeTestService(t, jobmode.Stdio, StatusRunning)
+	pipeInactive := false
+
+	view, err := service.materializeStatusView(row, true, &pipeInactive)
+	if err != nil {
+		t.Fatalf("materializeStatusView: %v", err)
+	}
+	if view.Status != StatusRunning {
+		t.Errorf("status = %q, want %q", view.Status, StatusRunning)
+	}
+}
+
+func TestMaterializeStdioStatusUsesExitArtifactsWithoutPipeDrainMarker(t *testing.T) {
+	service, row := setupModeTestService(t, jobmode.Stdio, StatusRunning)
+	jobDir := filepath.Join(service.config.JobsDir(), row.JobID)
+	if err := os.WriteFile(filepath.Join(jobDir, "runner-exit-code"), []byte("7\n"), 0600); err != nil {
+		t.Fatal(err)
+	}
+	const endedAt = "2026-08-04T10:00:00Z"
+	if err := os.WriteFile(filepath.Join(jobDir, "runner-ended-at"), []byte(endedAt+"\n"), 0600); err != nil {
+		t.Fatal(err)
+	}
+
+	view, err := service.materializeStatusView(row, false, nil)
+	if err != nil {
+		t.Fatalf("materializeStatusView: %v", err)
+	}
+	if view.Status != StatusExited || !view.Done {
+		t.Errorf("view = status %q done %v, want exited and done", view.Status, view.Done)
+	}
+	if view.ExitCode == nil || *view.ExitCode != 7 {
+		t.Errorf("exit code = %v, want 7", view.ExitCode)
+	}
+	if view.EndedAt == nil || *view.EndedAt != endedAt {
+		t.Errorf("ended_at = %v, want %s", view.EndedAt, endedAt)
+	}
+}
+
+func TestMaterializeStdioStatusMarksMissingSessionWithIncompleteArtifactsLost(t *testing.T) {
+	service, row := setupModeTestService(t, jobmode.Stdio, StatusRunning)
+	jobDir := filepath.Join(service.config.JobsDir(), row.JobID)
+	if err := os.WriteFile(filepath.Join(jobDir, "runner-exit-code"), []byte("0\n"), 0600); err != nil {
+		t.Fatal(err)
+	}
+
+	view, err := service.materializeStatusView(row, false, nil)
+	if err != nil {
+		t.Fatalf("materializeStatusView: %v", err)
+	}
+	if view.Status != StatusLost {
+		t.Errorf("status = %q, want %q", view.Status, StatusLost)
+	}
+}
+
+func setupModeTestService(t *testing.T, mode jobmode.Mode, status JobStatusName) (*Service, *JobRow) {
+	t.Helper()
+	stateDir := t.TempDir()
+	config := DefaultConfig()
+	config.StateDir = stateDir
+	config.RuntimeDir = filepath.Join(stateDir, "runtime")
+	if err := os.MkdirAll(config.JobsDir(), 0700); err != nil {
+		t.Fatal(err)
+	}
+	db := setupTestDB(t, stateDir)
+	t.Cleanup(func() { _ = db.Close() })
+	service := NewService(config)
+	service.db = db
+
+	const jobID = "mode-test-job"
+	jobDir := filepath.Join(config.JobsDir(), jobID)
+	if err := os.MkdirAll(jobDir, 0700); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(filepath.Join(jobDir, "output.log"), nil, 0600); err != nil {
+		t.Fatal(err)
+	}
+	row := &JobRow{
+		JobID:        jobID,
+		ScriptPath:   "jobs/mode-test-job/script",
+		OutputPath:   "jobs/mode-test-job/output.log",
+		Mode:         mode,
+		Cwd:          "/tmp",
+		TerminalCols: 80,
+		TerminalRows: 24,
+		Status:       status,
+		SessionName:  JobSessionName(jobID),
+		PaneTarget:   JobPaneTarget(jobID),
+		CreatedAt:    "2026-08-04T09:00:00Z",
+		UpdatedAt:    "2026-08-04T09:00:00Z",
+	}
+	inserted, err := db.InsertJob(row)
+	if err != nil || !inserted {
+		t.Fatalf("InsertJob: inserted=%v err=%v", inserted, err)
+	}
+	persisted, err := db.GetJob(jobID)
+	if err != nil {
+		t.Fatalf("GetJob: %v", err)
+	}
+	return service, persisted
+}

dify-agent-runtime/internal/server/tmux.go

@@ -7,6 +7,8 @@ import (
 	"os/exec"
 	"path/filepath"
 	"strings"
+
+	"github.com/langgenius/dify/dify-agent-runtime/internal/jobmode"
 )
 
 // TmuxController manages tmux sessions for shellctl jobs via a dedicated socket.
@@ -82,9 +84,13 @@ func (t *TmuxController) IsOutputPipeActive(jobID string) (*bool, error) {
 }
 
 // CreateJobSession creates a new tmux session for a job.
-func (t *TmuxController) CreateJobSession(jobID, jobDir, cwd string, cols, rows int) error {
+func (t *TmuxController) CreateJobSession(
+	jobID, jobDir, cwd string,
+	cols, rows int,
+	mode jobmode.Mode,
+) error {
 	runnerCmd := shellJoin([]string{
-		t.config.RunnerPath(), jobDir, jobID, cwd,
+		t.config.RunnerPath(), jobDir, jobID, cwd, string(mode),
 	})
 	result, err := t.runTmuxNoCheck(
 		"-f", "/dev/null",

dify-agent-runtime/internal/server/types.go

@@ -1,11 +1,14 @@
 package server
 
+import "github.com/langgenius/dify/dify-agent-runtime/internal/jobmode"
+
 // RunJobRequest is the HTTP request body for POST /v1/jobs/run.
 type RunJobRequest struct {
 	Script           string            `json:"script"`
 	Cwd              *string           `json:"cwd,omitempty"`
 	Env              map[string]string `json:"env,omitempty"`
 	Terminal         *TerminalSize     `json:"terminal,omitempty"`
+	Mode             jobmode.Mode      `json:"mode,omitempty"`
 	Timeout          float64           `json:"timeout,omitempty"`
 	OutputLimit      int               `json:"output_limit,omitempty"`
 	IdleFlushSeconds float64           `json:"idle_flush_seconds,omitempty"`

dify-agent-runtime/tests/acceptance_test.go

@@ -170,6 +170,84 @@ func TestRunSimpleScript(t *testing.T) {
 	}
 }
 
+func TestPTYModesMergeStdoutAndStderr(t *testing.T) {
+	for _, tgt := range targets() {
+		for _, tc := range []struct {
+			name string
+			mode string
+		}{
+			{name: "default"},
+			{name: "explicit", mode: "pty"},
+		} {
+			t.Run(tgt.name+"/"+tc.name, func(t *testing.T) {
+				payload := map[string]any{
+					"script":  "printf 'stdout-marker\\n'; printf 'stderr-marker\\n' >&2",
+					"timeout": 10,
+				}
+				if tc.mode != "" {
+					payload["mode"] = tc.mode
+				}
+				result := runJob(t, tgt, payload)
+				assertJobDone(t, result)
+				assertExitCode(t, result, 0)
+				output := result["output"].(string)
+				if !strings.Contains(output, "stdout-marker") || !strings.Contains(output, "stderr-marker") {
+					t.Fatalf("PTY output did not merge stdout and stderr: %q", output)
+				}
+			})
+		}
+	}
+}
+
+func TestRunStdioSeparatesStdoutAndStderr(t *testing.T) {
+	for _, tgt := range targets() {
+		t.Run(tgt.name, func(t *testing.T) {
+			result := runJob(t, tgt, map[string]any{
+				"script":  "printf '{\"ok\":true}'\nprintf 'warning' >&2",
+				"mode":    "stdio",
+				"timeout": 10,
+			})
+			assertJobDone(t, result)
+			assertExitCode(t, result, 0)
+			if output := result["output"].(string); output != `{"ok":true}` {
+				t.Errorf("stdio output = %q, want stdout-only JSON", output)
+			}
+		})
+	}
+}
+
+func TestStdioInputIsRejected(t *testing.T) {
+	for _, tgt := range targets() {
+		t.Run(tgt.name, func(t *testing.T) {
+			result := runJob(t, tgt, map[string]any{
+				"script":  "sleep 60",
+				"mode":    "stdio",
+				"timeout": 0.1,
+			})
+			jobID := result["job_id"].(string)
+			defer func() {
+				resp := doPost(t, tgt, fmt.Sprintf("/v1/jobs/%s/terminate", jobID), map[string]any{"grace_seconds": 0}, true)
+				resp.Body.Close()
+			}()
+
+			resp := doPost(t, tgt, fmt.Sprintf("/v1/jobs/%s/input", jobID), map[string]any{
+				"text":    "ignored\n",
+				"offset":  0,
+				"timeout": 1,
+			}, true)
+			assertStatus(t, resp, http.StatusConflict)
+			body := readBody(t, resp)
+			var failure map[string]map[string]string
+			if err := json.Unmarshal(body, &failure); err != nil {
+				t.Fatal(err)
+			}
+			if code := failure["error"]["code"]; code != "input_unsupported" {
+				t.Errorf("error code = %q, want input_unsupported", code)
+			}
+		})
+	}
+}
+
 func TestRunWithEnv(t *testing.T) {
 	for _, tgt := range targets() {
 		t.Run(tgt.name, func(t *testing.T) {
@@ -431,10 +509,12 @@ func TestSendInput(t *testing.T) {
 				"timeout": 2, // Will timeout waiting for input
 			})
 			jobID := result["job_id"].(string)
+			t.Cleanup(func() {
+				cleanupJobBestEffort(tgt, jobID)
+			})
 
 			if result["done"] == true {
-				// Already finished (possible race), skip
-				t.Skip("job completed before input could be sent")
+				t.Fatalf("interactive PTY job completed before input was sent: %#v", result)
 			}
 
 			// Send input
@@ -451,7 +531,7 @@ func TestSendInput(t *testing.T) {
 			json.Unmarshal(body, &inputResult)
 			output := inputResult["output"].(string)
 			if !strings.Contains(output, "got:hello-input") {
-				t.Logf("output after input: %q (may need more wait time)", output)
+				t.Fatalf("input result did not contain command echo: %q", output)
 			}
 		})
 	}
@@ -720,6 +800,25 @@ func doPost(t *testing.T, tgt target, path string, payload map[string]any, withA
 	return resp
 }
 
+func cleanupJobBestEffort(tgt target, jobID string) {
+	client := &http.Client{Timeout: 5 * time.Second}
+	body, _ := json.Marshal(map[string]any{"grace_seconds": 0})
+	req, _ := http.NewRequest("POST", fmt.Sprintf("%s/v1/jobs/%s/terminate", tgt.baseURL, jobID), bytes.NewReader(body))
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("Authorization", "Bearer "+authToken)
+	if resp, err := client.Do(req); err == nil {
+		_, _ = io.Copy(io.Discard, resp.Body)
+		_ = resp.Body.Close()
+	}
+
+	req, _ = http.NewRequest("DELETE", fmt.Sprintf("%s/v1/jobs/%s?force=true&grace_seconds=0", tgt.baseURL, jobID), nil)
+	req.Header.Set("Authorization", "Bearer "+authToken)
+	if resp, err := client.Do(req); err == nil {
+		_, _ = io.Copy(io.Discard, resp.Body)
+		_ = resp.Body.Close()
+	}
+}
+
 func readBody(t *testing.T, resp *http.Response) []byte {
 	t.Helper()
 	defer resp.Body.Close()

dify-agent/src/dify_agent/adapters/shell/__init__.py

@@ -9,6 +9,7 @@ from dify_agent.adapters.shell.protocols import (
     ShellCommandProtocol,
     ShellCommandResult,
     ShellCommandStatus,
+    ShellExecutionMode,
     ShellPromptObservation,
     ShellProviderError,
 )
@@ -27,6 +28,7 @@ __all__ = [
     "ShellCommandProtocol",
     "ShellCommandResult",
     "ShellCommandStatus",
+    "ShellExecutionMode",
     "ShellPromptObservation",
     "ShellProviderError",
 ]

dify-agent/src/dify_agent/adapters/shell/protocols.py

@@ -4,6 +4,9 @@ from dataclasses import dataclass
 from typing import Literal, Protocol
 
 
+type ShellExecutionMode = Literal["pty", "stdio"]
+
+
 @dataclass(frozen=True, slots=True)
 class ShellCommandResult:
     job_id: str
@@ -63,6 +66,7 @@ class ShellCommandProtocol(Protocol):
         cwd: str | None = None,
         env: dict[str, str] | None = None,
         timeout: float,
+        mode: ShellExecutionMode = "pty",
     ) -> ShellCommandResult: ...
 
     async def wait(

dify-agent/src/dify_agent/adapters/shell/shellctl.py

@@ -14,12 +14,13 @@ from typing import Protocol, TypeVar, cast
 
 import httpx2 as httpx
 from shellctl.client import ShellctlClientError
-from shellctl.shared import HealthResponse
+from shellctl.shared import HealthResponse, JobMode
 
 from dify_agent.adapters.shell.protocols import (
     ShellCommandProtocol,
     ShellCommandResult,
     ShellCommandStatus,
+    ShellExecutionMode,
     ShellProviderError,
 )
 
@@ -60,6 +61,7 @@ class ShellctlClientProtocol(Protocol):
         cwd: str | None = None,
         env: dict[str, str] | None = None,
         timeout: float = _DEFAULT_TIMEOUT_SECONDS,
+        mode: JobMode = JobMode.PTY,
     ) -> ShellctlJobResult: ...
 
     async def wait(
@@ -114,6 +116,7 @@ class ShellctlCommands(ShellCommandProtocol):
         cwd: str | None = None,
         env: dict[str, str] | None = None,
         timeout: float,
+        mode: ShellExecutionMode = "pty",
     ) -> ShellCommandResult:
         resolved_cwd = _resolve_lease_cwd(
             cwd,
@@ -122,7 +125,15 @@ class ShellctlCommands(ShellCommandProtocol):
         )
         resolved_env = _lease_env(env, home_dir=self.home_dir)
         return _from_job_result(
-            await _run_client_call(self.client.run(script, cwd=resolved_cwd, env=resolved_env, timeout=timeout))
+            await _run_client_call(
+                self.client.run(
+                    script,
+                    cwd=resolved_cwd,
+                    env=resolved_env,
+                    timeout=timeout,
+                    mode=JobMode(mode),
+                )
+            )
         )
 
     async def wait(

dify-agent/src/dify_agent/layers/shell/layer.py

@@ -460,6 +460,7 @@ class DifyShellLayer(PydanticAILayer[DifyShellLayerDeps, object, DifyShellLayerC
             ),
             timeout=timeout,
             max_output_bytes=max_output_bytes,
+            mode="stdio",
         )
 
     async def run_remote_script(
@@ -488,6 +489,7 @@ class DifyShellLayer(PydanticAILayer[DifyShellLayerDeps, object, DifyShellLayerC
             env=self._build_shell_command_env(include_agent_stub_env=False),
             timeout=DEFAULT_TIMEOUT_SECONDS,
             max_output_bytes=_REMOTE_COMPLETE_OUTPUT_MAX_BYTES,
+            mode="stdio",
         )
 
     def _require_resource(self) -> RuntimeLease:

dify-agent/src/dify_agent/runtime/command_runner.py

@@ -10,6 +10,7 @@ from dify_agent.adapters.shell.protocols import (
     CompleteShellCommandResult,
     ShellCommandProtocol,
     ShellCommandResult,
+    ShellExecutionMode,
 )
 from dify_agent.layers.shell.output_text import utf8_prefix
 
@@ -26,6 +27,7 @@ async def execute_complete_with_commands(
     env: dict[str, str] | None,
     timeout: float,
     max_output_bytes: int,
+    mode: ShellExecutionMode,
 ) -> CompleteShellCommandResult:
     """Run a command to completion with bounded output and deterministic cleanup."""
 
@@ -36,7 +38,13 @@ async def execute_complete_with_commands(
     captured_bytes = 0
     incomplete_reason: Literal["output_limit", "timeout"] | None = None
     try:
-        result = await commands.run(script, cwd=cwd, env=env, timeout=_remaining_time(deadline))
+        result = await commands.run(
+            script,
+            cwd=cwd,
+            env=env,
+            timeout=_remaining_time(deadline),
+            mode=mode,
+        )
         job_id = result.job_id
         while True:
             remaining_bytes = max(max_output_bytes - captured_bytes, 0)

dify-agent/src/dify_agent/runtime_backend/shellctl.py

@@ -112,8 +112,8 @@ async def run_shellctl_control_command(
     *,
     timeout: float = 30.0,
 ) -> CompleteShellCommandResult:
-    """Run one bounded driver control command and always delete its transient job."""
-    result = await commands.run(script, cwd=None, env=None, timeout=timeout)
+    """Run one bounded control command through stdout-only stdio and delete its transient job."""
+    result = await commands.run(script, cwd=None, env=None, timeout=timeout, mode="stdio")
     job_id = result.job_id
     output_parts = [result.output]
     try:

dify-agent/src/dify_agent/server/binding_files.py

@@ -166,6 +166,7 @@ class BindingFileService:
                     env={"HOME": lease.layout.home_dir},
                     timeout=_BROWSE_TIMEOUT_SECONDS,
                     max_output_bytes=_BROWSE_OUTPUT_MAX_BYTES,
+                    mode="stdio",
                 )
             payload = _require_browse_payload(result, operation="list")
             try:
@@ -193,6 +194,7 @@ class BindingFileService:
                     env={"HOME": lease.layout.home_dir},
                     timeout=_BROWSE_TIMEOUT_SECONDS,
                     max_output_bytes=_BROWSE_OUTPUT_MAX_BYTES,
+                    mode="stdio",
                 )
             payload = _require_browse_payload(result, operation="read")
             try:
@@ -243,6 +245,7 @@ class BindingFileService:
                         env=env,
                         timeout=_DOWNLOAD_TIMEOUT_SECONDS,
                         max_output_bytes=_DOWNLOAD_OUTPUT_MAX_BYTES,
+                        mode="stdio",
                     )
                 except ShellProviderError as exc:
                     if exc.code == "timeout":

dify-agent/src/shellctl/__init__.py

@@ -33,6 +33,7 @@ if TYPE_CHECKING:
         HealthResponse,
         InputJobRequest,
         JobInfo,
+        JobMode,
         JobResult,
         JobStatusName,
         JobStatusView,
@@ -63,6 +64,7 @@ __all__ = [
     "HealthResponse",
     "InputJobRequest",
     "JobInfo",
+    "JobMode",
     "JobResult",
     "JobStatusName",
     "JobStatusView",
@@ -97,6 +99,7 @@ _EXPORTS = {
     "HealthResponse": "shellctl.shared",
     "InputJobRequest": "shellctl.shared",
     "JobInfo": "shellctl.shared",
+    "JobMode": "shellctl.shared",
     "JobResult": "shellctl.shared",
     "JobStatusName": "shellctl.shared",
     "JobStatusView": "shellctl.shared",

dify-agent/src/shellctl/client/sdk.py

@@ -29,6 +29,7 @@ from shellctl.shared.schemas import (
     DeleteJobResponse,
     HealthResponse,
     JobInfo,
+    JobMode,
     JobResult,
     JobStatusView,
     ListJobsResponse,
@@ -144,11 +145,14 @@ class ShellctlClient:
         env: dict[str, str] | None = None,
         timeout: float = DEFAULT_TIMEOUT_SECONDS,
         terminal: TerminalSize | None = None,
+        mode: JobMode = JobMode.PTY,
     ) -> JobResult:
         """Create a new job and wait for initial output or completion.
 
         `cwd` and `env` preset the script's working directory and environment
-        overlay on the server side.
+        overlay on the server side. `mode="stdio"` provides stdout-only public
+        output for non-interactive commands; the default PTY mode remains
+        interactive and merges stdout with stderr.
         """
 
         payload = RunJobRequest(
@@ -156,6 +160,7 @@ class ShellctlClient:
             cwd=cwd,
             env=env,
             terminal=terminal,
+            mode=mode,
             timeout=timeout,
             output_limit=self.output_limit,
             idle_flush_seconds=self.idle_flush_seconds,

dify-agent/src/shellctl/shared/__init__.py

@@ -60,6 +60,7 @@ if TYPE_CHECKING:
         HealthResponse,
         InputJobRequest,
         JobInfo,
+        JobMode,
         JobResult,
         JobStatusName,
         JobStatusView,
@@ -101,6 +102,7 @@ __all__ = [
     "HealthResponse",
     "InputJobRequest",
     "JobInfo",
+    "JobMode",
     "JobResult",
     "JobStatusName",
     "JobStatusView",
@@ -166,6 +168,7 @@ _EXPORTS = {
     "HealthResponse": "shellctl.shared.schemas",
     "InputJobRequest": "shellctl.shared.schemas",
     "JobInfo": "shellctl.shared.schemas",
+    "JobMode": "shellctl.shared.schemas",
     "JobResult": "shellctl.shared.schemas",
     "JobStatusName": "shellctl.shared.schemas",
     "JobStatusView": "shellctl.shared.schemas",

dify-agent/src/shellctl/shared/schemas.py

@@ -42,6 +42,13 @@ class JobStatusName(StrEnum):
     LOST = "lost"
 
 
+class JobMode(StrEnum):
+    """Standard-stream wiring used to execute a shellctl job."""
+
+    PTY = "pty"
+    STDIO = "stdio"
+
+
 TERMINAL_JOB_STATUSES = frozenset(
     {
         JobStatusName.EXITED,
@@ -139,6 +146,7 @@ class RunJobRequest(ShellctlModel):
     cwd: str | None = None
     env: dict[str, str] | None = None
     terminal: TerminalSize | None = None
+    mode: JobMode = JobMode.PTY
     timeout: float = Field(default=DEFAULT_TIMEOUT_SECONDS, gt=0, le=SHELL_TOOL_HARD_TIMEOUT_SECONDS)
     output_limit: int = Field(default=DEFAULT_OUTPUT_LIMIT_BYTES, ge=1, le=MAX_OUTPUT_LIMIT_BYTES)
     idle_flush_seconds: float = Field(default=DEFAULT_IDLE_FLUSH_SECONDS, ge=0, le=30)
@@ -201,6 +209,7 @@ __all__ = [
     "HealthResponse",
     "InputJobRequest",
     "JobInfo",
+    "JobMode",
     "JobResult",
     "JobStatusName",
     "JobStatusView",

dify-agent/tests/local/dify_agent/adapters/shell/test_shellctl.py

@@ -9,6 +9,7 @@ from typing import cast
 import httpx2 as httpx
 import pytest
 from shellctl.client import ShellctlClientError
+from shellctl.shared import JobMode
 
 from dify_agent.adapters.shell.protocols import ShellProviderError
 from dify_agent.adapters.shell.shellctl import ShellctlClientProtocol, ShellctlCommands
@@ -39,12 +40,20 @@ class _Status:
 class _Client:
     run_result: object = field(default_factory=_Job)
     delete_error: Exception | None = None
-    run_calls: list[tuple[str, str | None, dict[str, str] | None, float]] = field(default_factory=list)
+    run_calls: list[tuple[str, str | None, dict[str, str] | None, float, JobMode]] = field(default_factory=list)
     wait_calls: list[tuple[str, int, float]] = field(default_factory=list)
     delete_calls: list[tuple[str, bool, float | None]] = field(default_factory=list)
 
-    async def run(self, script: str, *, cwd=None, env=None, timeout=30.0):
-        self.run_calls.append((script, cwd, env, timeout))
+    async def run(
+        self,
+        script: str,
+        *,
+        cwd=None,
+        env=None,
+        timeout=30.0,
+        mode: JobMode = JobMode.PTY,
+    ):
+        self.run_calls.append((script, cwd, env, timeout, mode))
         if isinstance(self.run_result, Exception):
             raise self.run_result
         return self.run_result
@@ -85,7 +94,20 @@ def test_commands_apply_runtime_layout_and_home_environment() -> None:
         assert result.output == "ok"
 
     asyncio.run(scenario())
-    assert client.run_calls == [("pwd", "/workspace/reports", {"TOKEN": "value", "HOME": "/home/binding"}, 2.5)]
+    assert client.run_calls == [
+        ("pwd", "/workspace/reports", {"TOKEN": "value", "HOME": "/home/binding"}, 2.5, JobMode.PTY)
+    ]
+
+
+def test_commands_forward_stdio_mode() -> None:
+    client = _Client()
+
+    async def scenario() -> None:
+        commands = ShellctlCommands(_client(client))
+        await commands.run("printf result", timeout=2.5, mode="stdio")
+
+    asyncio.run(scenario())
+    assert client.run_calls == [("printf result", None, None, 2.5, JobMode.STDIO)]
 
 
 def test_commands_reject_cwd_outside_runtime_layout() -> None:

dify-agent/tests/local/dify_agent/layers/shell/test_layer.py

@@ -26,6 +26,7 @@ from dify_agent.layers.shell.layer import (
 from dify_agent.adapters.shell.protocols import (
     ShellCommandResult,
     ShellCommandStatus,
+    ShellExecutionMode,
     ShellProviderError,
 )
 from dify_agent.layers.execution_context import DifyExecutionContextLayerConfig
@@ -113,6 +114,7 @@ class RunCall:
     cwd: str | None
     env: Mapping[str, str] | None
     timeout: float
+    mode: ShellExecutionMode = "pty"
 
 
 @dataclass(slots=True)
@@ -167,8 +169,16 @@ class FakeCommands:
     interrupt_calls: list[InterruptCall] = field(default_factory=list)
     delete_calls: list[DeleteCall] = field(default_factory=list)
 
-    async def run(self, script: str, *, cwd: str | None = None, env: dict[str, str] | None = None, timeout: float):
-        self.run_calls.append(RunCall(script=script, cwd=cwd, env=env, timeout=timeout))
+    async def run(
+        self,
+        script: str,
+        *,
+        cwd: str | None = None,
+        env: dict[str, str] | None = None,
+        timeout: float,
+        mode: ShellExecutionMode = "pty",
+    ):
+        self.run_calls.append(RunCall(script=script, cwd=cwd, env=env, timeout=timeout, mode=mode))
         if self.run_handler is None:
             raise AssertionError("Unexpected run() call")
         return self.run_handler(script, cwd, env, timeout)
@@ -499,6 +509,7 @@ def test_shell_layer_tools_map_inputs_and_maintain_offsets_with_tail_end() -> No
     asyncio.run(scenario())
 
     assert layer.runtime_state.job_offsets == {"user-job": 34}
+    assert commands.run_calls[0].mode == "pty"
     assert commands.tail_calls == [TailCall(job_id="user-job"), TailCall(job_id="user-job")]
 
 
@@ -936,6 +947,7 @@ def test_run_remote_script_complete_uses_read_output_before_wait_and_deletes_job
 
     asyncio.run(scenario())
     assert events == ["run", "read_output", "wait"]
+    assert commands.run_calls[0].mode == "stdio"
     assert [call.job_id for call in commands.delete_calls] == ["remote-job"]
 
 

dify-agent/tests/local/dify_agent/runtime/test_command_runner.py

@@ -6,7 +6,7 @@ from dataclasses import dataclass, field
 
 import pytest
 
-from dify_agent.adapters.shell.protocols import ShellCommandResult
+from dify_agent.adapters.shell.protocols import ShellCommandResult, ShellExecutionMode
 from dify_agent.runtime.command_runner import execute_complete_with_commands
 
 
@@ -16,11 +16,20 @@ class _BlockingCommands:
     wait_forever: asyncio.Event = field(default_factory=asyncio.Event)
     deletes: list[tuple[str, bool]] = field(default_factory=list)
 
-    async def run(self, script: str, *, cwd: str | None, env: dict[str, str] | None, timeout: float):
+    async def run(
+        self,
+        script: str,
+        *,
+        cwd: str | None,
+        env: dict[str, str] | None,
+        timeout: float,
+        mode: ShellExecutionMode = "pty",
+    ):
         assert script == "long-running"
         assert cwd == "/workspace"
         assert env == {"HOME": "/home/agent"}
         assert timeout > 0
+        assert mode == "stdio"
         return ShellCommandResult(
             job_id="job-1",
             status="running",
@@ -66,6 +75,7 @@ async def test_cancellation_deletes_job_returned_before_blocking_wait() -> None:
             env={"HOME": "/home/agent"},
             timeout=60.0,
             max_output_bytes=4096,
+            mode="stdio",
         )
     )
     try:

dify-agent/tests/local/dify_agent/runtime_backend/test_local.py

@@ -5,7 +5,7 @@ import shlex
 from typing import Mapping
 
 import pytest
-from shellctl.shared import DeleteJobResponse, JobResult, JobStatusName, JobStatusView
+from shellctl.shared import DeleteJobResponse, JobMode, JobResult, JobStatusName, JobStatusView
 
 from dify_agent.runtime_backend import (
     BindingCreateError,
@@ -22,6 +22,7 @@ class _RunCall:
     commands: tuple[tuple[str, ...], ...]
     cwd: str | None
     env: Mapping[str, str] | None
+    mode: JobMode
 
 
 @dataclass(slots=True)
@@ -40,12 +41,13 @@ class _Client:
         cwd: str | None = None,
         env: Mapping[str, str] | None = None,
         timeout: float = 10.0,
+        mode: JobMode = JobMode.PTY,
     ) -> JobResult:
         del timeout
         commands = tuple(
             tuple(shlex.split(line)) for line in script.splitlines() if line.strip() and line.strip() != "set -eu"
         )
-        self.runs.append(_RunCall(commands=commands, cwd=cwd, env=env))
+        self.runs.append(_RunCall(commands=commands, cwd=cwd, env=env, mode=mode))
         return JobResult(
             job_id=f"job-{len(self.runs)}",
             status=JobStatusName.EXITED,
@@ -165,6 +167,7 @@ async def test_local_binding_create_materializes_home_and_new_workspace() -> Non
     assert ("mkdir", "-p", "/homes/binding-1") in factory.commands
     assert ("cp", "-a", "/snapshots/home-home-1/.", "/homes/binding-1/") in factory.commands
     assert ("chmod", "700", "/homes/binding-1", "/workspaces/workspace-1") in factory.commands
+    assert all(run.mode is JobMode.STDIO for run in factory.runs)
 
 
 @pytest.mark.anyio
@@ -245,6 +248,7 @@ async def test_local_binding_acquire_scopes_commands_to_materialized_home_and_wo
     pwd_run = next(run for run in factory.runs if run.commands == (("pwd",),))
     assert pwd_run.cwd == "/workspaces/workspace-1"
     assert pwd_run.env == {"HOME": "/homes/binding-1"}
+    assert pwd_run.mode is JobMode.PTY
     with pytest.raises(ValueError, match="outside this RuntimeLease"):
         await lease.commands.run("cat secret", cwd="/homes/other", timeout=10.0)
     await backend.release(lease)

dify-agent/tests/local/dify_agent/runtime_backend/test_shellctl_backend.py

@@ -5,7 +5,7 @@ from typing import cast
 
 import pytest
 
-from dify_agent.adapters.shell.protocols import ShellCommandResult, ShellCommandStatus
+from dify_agent.adapters.shell.protocols import ShellCommandResult, ShellCommandStatus, ShellExecutionMode
 from dify_agent.adapters.shell.shellctl import ShellctlClientProtocol
 from dify_agent.runtime_backend.protocols import RuntimeLayout
 from dify_agent.runtime_backend.shellctl import (
@@ -43,6 +43,7 @@ class _FakeCommands:
     wait_error: Exception | None = None
     delete_error: Exception | None = None
     delete_calls: list[tuple[str, bool]] = field(default_factory=list)
+    run_modes: list[ShellExecutionMode] = field(default_factory=list)
 
     async def run(
         self,
@@ -51,8 +52,10 @@ class _FakeCommands:
         cwd: str | None = None,
         env: dict[str, str] | None = None,
         timeout: float,
+        mode: ShellExecutionMode = "pty",
     ) -> ShellCommandResult:
         del script, cwd, env, timeout
+        self.run_modes.append(mode)
         return self.initial
 
     async def wait(self, job_id: str, *, offset: int, timeout: float) -> ShellCommandResult:
@@ -169,6 +172,7 @@ async def test_control_command_success_is_preserved_when_delete_fails(
         result = await run_shellctl_control_command(commands, "true")
 
     assert result.output == "ok"
+    assert commands.run_modes == ["stdio"]
     assert commands.delete_calls == [("job-1", True)]
     assert "delete failed" in caplog.text
 
@@ -188,4 +192,5 @@ async def test_control_command_error_is_preserved_when_delete_also_fails(
             _ = await run_shellctl_control_command(commands, "false")
 
     assert commands.delete_calls == [("job-1", True)]
+    assert commands.run_modes == ["stdio"]
     assert "delete failed" in caplog.text

dify-agent/tests/local/dify_agent/server/test_binding_files.py

@@ -15,7 +15,7 @@ import pytest
 from fastapi import FastAPI
 from fastapi.testclient import TestClient
 
-from dify_agent.adapters.shell.protocols import ShellCommandResult, ShellProviderError
+from dify_agent.adapters.shell.protocols import ShellCommandResult, ShellExecutionMode, ShellProviderError
 from dify_agent.layers.execution_context import DifyExecutionContextLayerConfig
 from dify_agent.protocol import BindingFileDownloadRequest, BindingFileListRequest, BindingFileReadRequest
 from dify_agent.runtime_backend import BindingAcquireError, BindingLostError, RuntimeLayout, RuntimeLease
@@ -38,7 +38,16 @@ class _Commands:
     calls: list[tuple[str, str | None, dict[str, str] | None, float]] = field(default_factory=list)
     deletes: list[str] = field(default_factory=list)
 
-    async def run(self, script: str, *, cwd: str | None = None, env=None, timeout: float) -> ShellCommandResult:
+    async def run(
+        self,
+        script: str,
+        *,
+        cwd: str | None = None,
+        env=None,
+        timeout: float,
+        mode: ShellExecutionMode = "pty",
+    ) -> ShellCommandResult:
+        assert mode == "stdio"
         self.calls.append((script, cwd, env, timeout))
         output = self.outputs.pop(0)
         exit_code = self.exit_codes.pop(0) if self.exit_codes else 0
@@ -77,7 +86,16 @@ class _ProviderErrorCommands(_Commands):
     phase: Literal["run", "wait"] = "run"
     error_code: str = "timeout"
 
-    async def run(self, script: str, *, cwd: str | None = None, env=None, timeout: float) -> ShellCommandResult:
+    async def run(
+        self,
+        script: str,
+        *,
+        cwd: str | None = None,
+        env=None,
+        timeout: float,
+        mode: ShellExecutionMode = "pty",
+    ) -> ShellCommandResult:
+        assert mode == "stdio"
         self.calls.append((script, cwd, env, timeout))
         if self.phase == "run":
             raise ShellProviderError("shell provider failed", code=self.error_code)
@@ -99,7 +117,16 @@ class _ProviderErrorCommands(_Commands):
 
 @dataclass(slots=True)
 class _LocalCommands(_Commands):
-    async def run(self, script: str, *, cwd: str | None = None, env=None, timeout: float) -> ShellCommandResult:
+    async def run(
+        self,
+        script: str,
+        *,
+        cwd: str | None = None,
+        env=None,
+        timeout: float,
+        mode: ShellExecutionMode = "pty",
+    ) -> ShellCommandResult:
+        assert mode == "stdio"
         self.calls.append((script, cwd, env, timeout))
         process = await asyncio.create_subprocess_shell(
             script,

dify-agent/tests/local/shellctl/test_shellctl_client.py

@@ -10,6 +10,7 @@ from shellctl.client import sdk as shellctl_sdk
 from shellctl.shared import (
     DEFAULT_TERMINATE_GRACE_SECONDS,
     HealthResponse,
+    JobMode,
     JobStatusName,
 )
 
@@ -29,12 +30,14 @@ class ForcedDeleteKwargs(TypedDict, total=False):
                 cwd="/tmp",
                 env={"HELLO": "world"},
                 timeout=12,
+                mode=JobMode.STDIO,
             ),
             "/v1/jobs/run",
             {
                 "script": "printf ready\\n",
                 "cwd": "/tmp",
                 "env": {"HELLO": "world"},
+                "mode": "stdio",
                 "timeout": 12.0,
                 "output_limit": 4096,
                 "idle_flush_seconds": 0.25,

dify-agent/tests/local/shellctl/test_shellctl_shared.py

@@ -7,6 +7,7 @@ from pydantic import ValidationError
 
 from shellctl.shared import (
     JOB_ID_ALPHABET,
+    JobMode,
     MAX_WAIT_TIMEOUT_SECONDS,
     RunJobRequest,
     SHELL_TOOL_HARD_TIMEOUT_SECONDS,
@@ -18,6 +19,13 @@ from shellctl.shared import (
 )
 
 
+def test_run_job_request_defaults_to_pty_and_rejects_unknown_mode() -> None:
+    assert RunJobRequest(script="true").mode is JobMode.PTY
+
+    with pytest.raises(ValidationError):
+        RunJobRequest(script="true", mode="stdout")  # pyright: ignore[reportArgumentType]
+
+
 def test_shell_tool_timeout_budget_has_one_source_of_truth() -> None:
     assert MAX_WAIT_TIMEOUT_SECONDS == SHELL_TOOL_HARD_TIMEOUT_SECONDS == 300
     assert SHELL_TOOL_HTTP_TIMEOUT_GRACE_SECONDS == 10