diff --git a/configs/config.example.yaml b/configs/config.example.yaml index b348e73..5fa03bb 100644 --- a/configs/config.example.yaml +++ b/configs/config.example.yaml @@ -15,14 +15,14 @@ scheduler: max_count_per_job: 10 # submit async 单个 job 最多生成多少张图 maintenance_interval: 5m # 后台维护任务执行间隔 task_lease_timeout: 30m # 后台任务租约超时时间 - max_attempts: 3 # submit async 单张图片失败后的最大重试次数 + max_attempts: 3 # 总尝试次数;3 表示首次尝试加最多两次重试 backend: type: built_in_codex # 使用本机 Codex CLI 调用内置 $imagegen command: codex # Codex CLI 命令名或可执行文件路径 model: "" # 为空时使用 Codex CLI 默认模型;需要固定模型时再填写 cwd: "" # Codex CLI 工作目录;为空时使用当前进程工作目录 - timeout: 90s # 单次 Codex/imagegen 调用超时时间 + timeout: 300s # 单次 Codex/imagegen 调用超时时间 delivery_dir: "" # 可选:生成后复制图片到该目录;OpenClaw/TG 可指向其允许发送的 workspace/media 目录 delivery_max_files: 200 # delivery_dir 非空时最多保留的投递文件数;0 表示不自动清理 cleanup_source_thread_dir: false # delivery_dir 非空且复制成功后,是否删除 Codex generated_images 中本次图片所在的 thread 目录 diff --git a/internal/codex/runner.go b/internal/codex/runner.go index 00c4ccb..e5f9d82 100644 --- a/internal/codex/runner.go +++ b/internal/codex/runner.go @@ -53,10 +53,15 @@ func (Runner) Run(ctx context.Context, req Request) (RunResult, error) { req.RecordPhase(phase, time.Now(), detail) } + if err := ctx.Err(); err != nil { + return RunResult{}, err + } + record("process.starting", "") - cmd := exec.CommandContext(ctx, req.Command, req.Args...) + cmd := exec.Command(req.Command, req.Args...) cmd.Dir = req.Dir cmd.Env = envWithCodexHome(req.Env, req.CodexHome) + configureCommandProcess(cmd) stdoutPipe, err := cmd.StdoutPipe() if err != nil { @@ -101,8 +106,27 @@ func (Runner) Run(ctx context.Context, req Request) (RunResult, error) { stderrErr = readLines(stderrPipe, &stderr, stderrPhaseRecorder(record)) }) - wg.Wait() - waitErr := cmd.Wait() + pipesDone := make(chan struct{}) + go func() { + wg.Wait() + close(pipesDone) + }() + + var waitErr error + select { + case <-pipesDone: + waitErr = cmd.Wait() + case <-ctx.Done(): + killErr := killCommandProcess(cmd) + waitErr = cmd.Wait() + <-pipesDone + if killErr != nil && !commandProcessDoneError(killErr) { + waitErr = errors.Join(ctx.Err(), killErr) + } else { + waitErr = ctx.Err() + } + } + runCleanups() record("process.exited", "") diff --git a/internal/codex/runner_process_other.go b/internal/codex/runner_process_other.go new file mode 100644 index 0000000..4105230 --- /dev/null +++ b/internal/codex/runner_process_other.go @@ -0,0 +1,22 @@ +//go:build !windows && !unix + +package codex + +import ( + "errors" + "os" + "os/exec" +) + +func configureCommandProcess(cmd *exec.Cmd) {} + +func killCommandProcess(cmd *exec.Cmd) error { + if cmd == nil || cmd.Process == nil { + return nil + } + return cmd.Process.Kill() +} + +func commandProcessDoneError(err error) bool { + return errors.Is(err, os.ErrProcessDone) +} diff --git a/internal/codex/runner_process_unix.go b/internal/codex/runner_process_unix.go new file mode 100644 index 0000000..ec7e540 --- /dev/null +++ b/internal/codex/runner_process_unix.go @@ -0,0 +1,25 @@ +//go:build unix + +package codex + +import ( + "errors" + "os" + "os/exec" + "syscall" +) + +func configureCommandProcess(cmd *exec.Cmd) { + cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} +} + +func killCommandProcess(cmd *exec.Cmd) error { + if cmd == nil || cmd.Process == nil { + return nil + } + return syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL) +} + +func commandProcessDoneError(err error) bool { + return errors.Is(err, os.ErrProcessDone) || errors.Is(err, syscall.ESRCH) +} diff --git a/internal/codex/runner_process_windows.go b/internal/codex/runner_process_windows.go new file mode 100644 index 0000000..4eca298 --- /dev/null +++ b/internal/codex/runner_process_windows.go @@ -0,0 +1,22 @@ +//go:build windows + +package codex + +import ( + "errors" + "os" + "os/exec" +) + +func configureCommandProcess(cmd *exec.Cmd) {} + +func killCommandProcess(cmd *exec.Cmd) error { + if cmd == nil || cmd.Process == nil { + return nil + } + return cmd.Process.Kill() +} + +func commandProcessDoneError(err error) bool { + return errors.Is(err, os.ErrProcessDone) +} diff --git a/internal/codex/runner_test.go b/internal/codex/runner_test.go index f8fd95c..0cece7c 100644 --- a/internal/codex/runner_test.go +++ b/internal/codex/runner_test.go @@ -6,6 +6,7 @@ import ( "os" "path/filepath" "reflect" + "runtime" "strings" "testing" "time" @@ -22,6 +23,52 @@ func TestRunnerRunReturnsDeadlineExceededOnTimeout(t *testing.T) { } } +func TestRunnerRunDoesNotStartCommandWhenContextCanceled(t *testing.T) { + sentinel := filepath.Join(t.TempDir(), "started") + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + var phases []string + _, err := (Runner{}).Run(ctx, Request{ + Command: "sh", + Args: []string{"-c", "printf started > \"$SENTINEL\""}, + Env: []string{"SENTINEL=" + sentinel, "PATH=/bin:/usr/bin"}, + RecordPhase: func(phase string, occurredAt time.Time, detail string) { + phases = append(phases, phase) + }, + }) + + if !errors.Is(err, context.Canceled) { + t.Fatalf("expected context canceled, got %v", err) + } + if len(phases) != 0 { + t.Fatalf("expected no process phases because command was not started, got %#v", phases) + } + if _, statErr := os.Stat(sentinel); !errors.Is(statErr, os.ErrNotExist) { + t.Fatalf("sentinel file exists or stat failed unexpectedly: %v", statErr) + } +} + +func TestRunnerRunTimeoutKillsProcessGroupBeforeWaitingForPipes(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("process group timeout behavior is Unix-specific") + } + started := time.Now() + _, err := (Runner{}).Run(context.Background(), Request{ + Command: "sh", + Args: []string{"-c", "sh -c 'sleep 2' & printf 'parent ready\\n'; wait"}, + Timeout: 50 * time.Millisecond, + }) + elapsed := time.Since(started) + + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("expected deadline exceeded, got %v", err) + } + if elapsed > 500*time.Millisecond { + t.Fatalf("Run returned after %s, want it to return promptly after timeout", elapsed) + } +} + func TestRunnerRunRecordsStreamingPhases(t *testing.T) { var phases []string result, err := (Runner{}).Run(context.Background(), Request{ diff --git a/internal/config/config.go b/internal/config/config.go index bf81dcf..6bd4b69 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -109,7 +109,7 @@ func Default() Config { Backend: BackendConfig{ Type: "built_in_codex", Command: "codex", - Timeout: 90 * time.Second, + Timeout: 300 * time.Second, DeliveryMaxFiles: 200, Prompt: PromptConfig{ Prefix: "$imagegen", diff --git a/internal/config/load_test.go b/internal/config/load_test.go index 1d93e4b..39510e4 100644 --- a/internal/config/load_test.go +++ b/internal/config/load_test.go @@ -35,6 +35,9 @@ func TestLoadMissingDefaultPathReturnsDefaults(t *testing.T) { if cfg.Server.Listen != Default().Server.Listen { t.Fatalf("listen = %q", cfg.Server.Listen) } + if cfg.Backend.Timeout != 300*time.Second { + t.Fatalf("backend.timeout = %s, want 300s", cfg.Backend.Timeout) + } } func TestDefaultRealtimeConfig(t *testing.T) { @@ -297,6 +300,9 @@ func TestLoadExampleConfig(t *testing.T) { if cfg.Backend.Command != "codex" { t.Fatalf("backend.command = %q", cfg.Backend.Command) } + if cfg.Backend.Timeout != 300*time.Second { + t.Fatalf("backend.timeout = %s, want 300s", cfg.Backend.Timeout) + } if cfg.Backend.Prompt.Prefix != "$imagegen" { t.Fatalf("backend.prompt.prefix = %q", cfg.Backend.Prompt.Prefix) }