diff --git a/docs/roadmap.md b/docs/roadmap.md index 24f397ca8..d8f3c307c 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -4,7 +4,7 @@ ## 当前总目标 -推进 M2 Edge 本地数据层，让前端、后端、客户端三条线能围绕稳定的 Project / Thread / Run / Item / Event 模型并行开发。当前客户端 PR #30 已提供内存态最小实现，`feat/client-thread-messages-delicious233` 已补 message/item 写入链路，`feat/client-run-lifecycle-delicious233` 已抽出 Runner lifecycle 边界，`feat/client-store-boundary-delicious233` 已抽象 Edge store 接口边界，`feat/client-store-persistence-delicious233` 已提供轻量 JSON 文件持久化实现，`feat/client-edge-store-file-flag-delicious233` 已接入 Edge 启动参数 `--store-file`，下一步重点是真实 Runner adapter。 +推进 M2 Edge 本地数据层，让前端、后端、客户端三条线能围绕稳定的 Project / Thread / Run / Item / Event 模型并行开发。当前客户端 PR #30 已提供内存态最小实现，`feat/client-thread-messages-delicious233` 已补 message/item 写入链路，`feat/client-run-lifecycle-delicious233` 已抽出 Runner lifecycle 边界，`feat/client-store-boundary-delicious233` 已抽象 Edge store 接口边界，`feat/client-store-persistence-delicious233` 已提供轻量 JSON 文件持久化实现，`feat/client-edge-store-file-flag-delicious233` 已接入 Edge 启动参数 `--store-file`，`feat/client-runner-process-adapter-delicious233` 已补本地进程 executor 边界，下一步重点是真实 Runner adapter。 ## 路线图分层 @@ -26,7 +26,7 @@ - [x] M1 客户端本地链路：Desktop Shell + Local Edge + Mock Runner + smoke test。 - [ ] M2 Edge 本地数据层：Project / Thread / Run / Item / EventStore。最小内存实现已在 PR #30，message/item 写入链路、Runner lifecycle 边界、store 接口边界、轻量 JSON 文件持久化实现和 `--store-file` 启动参数已补齐，SQLite 仍是后续可选评估项。 -- [ ] M3 真实 Runner：CLI Agent 进程、取消、日志、错误映射。 +- [ ] M3 真实 Runner：CLI Agent 进程、取消、日志、错误映射。本地进程 executor 边界已补齐，但还不是 Claude Code / Codex / OpenCode 的完整 adapter。 - [ ] M4 Workspace 能力：worktree、diff、preview、artifact、approval。 - [ ] M5 Hub 协作链路：Edge-Hub sync、远程查看、远程审批。 @@ -34,7 +34,7 @@ - 前端：从 Mock 数据过渡到真实 REST / WebSocket client，承接 UI 同学设计。 - 后端：实现 Hub Server、Edge-Hub 通信、账号/群聊/同步/中继能力。 -- 客户端：PR #30 推进 Edge 本地数据层，`feat/client-thread-messages-delicious233` 补齐 message/item 写入链路，`feat/client-run-lifecycle-delicious233` 和 `feat/client-store-boundary-delicious233` 分别补齐 lifecycle/store 可替换边界，`feat/client-store-persistence-delicious233` 增加轻量 JSON 文件持久化 store，`feat/client-edge-store-file-flag-delicious233` 将文件 store 接入 Edge 启动入口，后续继续做真实 Runner adapter。 +- 客户端：PR #30 推进 Edge 本地数据层，`feat/client-thread-messages-delicious233` 补齐 message/item 写入链路，`feat/client-run-lifecycle-delicious233` 和 `feat/client-store-boundary-delicious233` 分别补齐 lifecycle/store 可替换边界，`feat/client-store-persistence-delicious233` 增加轻量 JSON 文件持久化 store，`feat/client-edge-store-file-flag-delicious233` 将文件 store 接入 Edge 启动入口，`feat/client-runner-process-adapter-delicious233` 增加可测试的本地进程 executor，后续继续做真实 Runner adapter。 ## 验收门槛 @@ -50,6 +50,7 @@ - [x] 增加 Edge store 轻量 JSON 文件持久化实现；SQLite 依赖获取问题保留为后续可选评估。 - [x] 为 Edge 启动入口接入可选 `--store-file `，未传时继续使用内存 store。 - [x] 在客户端 M2 基础上补 `POST /v1/threads/{threadId}/messages` 到 Item / event 的写入链路。 +- [x] 增加 Edge 本地进程 executor 边界，支持 stdout/stderr、成功、失败和取消事件映射。 - [ ] 将 Runner 真正接入 Edge Run lifecycle，替换 handler 内置 mock flow。 - [ ] M2 完成后归档或更新 `docs/client-roadmap.md`，避免路线图重复。 - [ ] 为 Runner 真实 CLI adapter 规划最小测试夹具。 diff --git a/docs/roadmaps/branches/feat-client-runner-process-adapter-delicious233.md b/docs/roadmaps/branches/feat-client-runner-process-adapter-delicious233.md new file mode 100644 index 000000000..72fe10855 --- /dev/null +++ b/docs/roadmaps/branches/feat-client-runner-process-adapter-delicious233.md @@ -0,0 +1,42 @@ +# feat/client-runner-process-adapter-delicious233 路线图 + +最后更新：2026-05-23 + +## 当前目标 + +- [x] 为 Edge Run lifecycle 增加可测试的本地进程 executor。 +- [x] 保持未配置 runner 命令时仍使用 MockExecutor。 +- [x] 记录这是本地进程 executor 边界，不是完整真实 CLI adapter。 + +## 写入范围 + +- `edge-server/internal/lifecycle/` +- `edge-server/cmd/agenthub-edge/` +- `edge-server/internal/httpserver/` +- `docs/roadmap.md` +- `docs/roadmaps/client.md` +- `docs/roadmaps/branches/feat-client-runner-process-adapter-delicious233.md` + +## 已完成 + +- 新增 `ProcessExecutor`，通过 `exec.CommandContext` 启动可配置命令。 +- 将 stdout / stderr 转为 `run.output.batch` 事件。 +- 正常退出发布 `run.finished`，非零退出发布 `run.failed`，取消发布 `run.cancelled`。 +- 增加并发 run map，重复启动返回 `ErrRunAlreadyStarted`，`Cancel(runID)` 可取消运行中的进程。 +- `agenthub-edge` 增加 `--runner-command` 和可重复 `--runner-arg`，未配置时仍走 MockExecutor。 +- 子进程会收到 `AGENTHUB_RUN_ID` / `AGENTHUB_PROJECT_ID` / `AGENTHUB_THREAD_ID`，用于后续真实 adapter 读取执行上下文。 +- 修复交叉 review 发现的边界：缺失 run 不会启动外部进程，nil bus/store 会在构造期返回错误。 +- 测试使用 Go test 自进程 helper，不依赖 shell、PowerShell、Claude Code、Codex 或 OpenCode。 + +## 验收 + +- [x] `cd edge-server; go test -count=1 ./internal/lifecycle ./cmd/agenthub-edge ./internal/httpserver` +- [x] 交叉 review 发现 1 个 High、1 个 Medium，均已修复。 +- [x] `git diff --check` +- [x] `python -c "import yaml, pathlib; yaml.safe_load(pathlib.Path('api/openapi.yaml').read_text(encoding='utf-8')); print('yaml ok')"` +- [x] `cd edge-server; go test -count=1 ./...` +- [x] `cd runner; go test -count=1 ./...` + +## 下一步 + +- [ ] 在真实 Runner adapter 任务中定义 CLI 入参、工作目录、环境变量、路径保护和错误映射。 diff --git a/docs/roadmaps/client.md b/docs/roadmaps/client.md index 86f686d2e..997861a2d 100644 --- a/docs/roadmaps/client.md +++ b/docs/roadmaps/client.md @@ -12,7 +12,7 @@ ## 当前目标 -推进 M2 Edge 本地数据层，把 M1 的内存事件流升级为 Project / Thread / Run / Item / Event 模型。当前 PR #30 已完成内存态最小模型，`feat/client-thread-messages-delicious233` 已补 message/item 写入链路，`feat/client-run-lifecycle-delicious233` 已抽出 Edge Run lifecycle executor 边界，`feat/client-store-boundary-delicious233` 已抽象可替换 store 接口，`feat/client-store-persistence-delicious233` 已提供轻量 JSON 文件持久化实现，`feat/client-edge-store-file-flag-delicious233` 已将文件 store 接入 Edge 启动参数，后续继续补真实 Runner adapter。 +推进 M2 Edge 本地数据层，把 M1 的内存事件流升级为 Project / Thread / Run / Item / Event 模型。当前 PR #30 已完成内存态最小模型，`feat/client-thread-messages-delicious233` 已补 message/item 写入链路，`feat/client-run-lifecycle-delicious233` 已抽出 Edge Run lifecycle executor 边界，`feat/client-store-boundary-delicious233` 已抽象可替换 store 接口，`feat/client-store-persistence-delicious233` 已提供轻量 JSON 文件持久化实现，`feat/client-edge-store-file-flag-delicious233` 已将文件 store 接入 Edge 启动参数，`feat/client-runner-process-adapter-delicious233` 已补本地进程 executor 边界，后续继续补真实 Runner adapter。 ## 近期任务 @@ -28,6 +28,7 @@ - [ ] 后续按需要评估 SQLite 持久化方案。 - [x] 补齐 `POST /v1/threads/{threadId}/messages` 到 Item / event 的写入链路。 - [x] 抽出 Edge Run lifecycle executor 边界，替换 handler 内置 mock flow。 +- [x] 增加可测试的本地进程 executor，覆盖 stdout/stderr 输出、正常退出、非零退出、取消和重复启动。 - [ ] 将真实 Runner adapter 接入 Edge Run lifecycle。 - [ ] 细化 Project / Thread / Item 的 OpenAPI 响应 schema。 diff --git a/edge-server/cmd/agenthub-edge/main.go b/edge-server/cmd/agenthub-edge/main.go index 624826b8a..e4b8a16d0 100644 --- a/edge-server/cmd/agenthub-edge/main.go +++ b/edge-server/cmd/agenthub-edge/main.go @@ -5,14 +5,29 @@ import ( "fmt" "log/slog" "os" + "strings" "github.com/agenthub/edge-server/internal/httpserver" + "github.com/agenthub/edge-server/internal/lifecycle" "github.com/agenthub/edge-server/internal/store" ) type config struct { - Addr string - StoreFile string + Addr string + StoreFile string + RunnerCommand string + RunnerArgs repeatedString +} + +type repeatedString []string + +func (v *repeatedString) String() string { + return fmt.Sprint([]string(*v)) +} + +func (v *repeatedString) Set(value string) error { + *v = append(*v, value) + return nil } func main() { @@ -32,7 +47,15 @@ func main() { os.Exit(1) } - if err := httpserver.Run(httpserver.Config{Addr: cfg.Addr, Store: repository}); err != nil { + serverConfig := httpserver.Config{Addr: cfg.Addr, Store: repository} + if cfg.RunnerCommand != "" { + serverConfig.ProcessExecutor = lifecycle.ProcessExecutorConfig{ + Command: cfg.RunnerCommand, + Args: append([]string(nil), cfg.RunnerArgs...), + } + } + + if err := httpserver.Run(serverConfig); err != nil { slog.Error("server exited with error", "err", err) os.Exit(1) } @@ -45,9 +68,15 @@ func buildConfig(args []string) (config, error) { cfg := config{} fs.StringVar(&cfg.Addr, "addr", "127.0.0.1:3210", "listen address") fs.StringVar(&cfg.StoreFile, "store-file", "", "JSON store snapshot file path") + fs.StringVar(&cfg.RunnerCommand, "runner-command", "", "local process command to execute for each run; empty uses the mock executor") + fs.Var(&cfg.RunnerArgs, "runner-arg", "argument passed to --runner-command; may be repeated") if err := fs.Parse(args); err != nil { return config{}, err } + cfg.RunnerCommand = strings.TrimSpace(cfg.RunnerCommand) + if cfg.RunnerCommand == "" && len(cfg.RunnerArgs) > 0 { + return config{}, fmt.Errorf("--runner-arg requires --runner-command") + } if fs.NArg() != 0 { return config{}, fmt.Errorf("unexpected positional arguments: %v", fs.Args()) } diff --git a/edge-server/cmd/agenthub-edge/main_test.go b/edge-server/cmd/agenthub-edge/main_test.go index 5554eab4e..6c31c6602 100644 --- a/edge-server/cmd/agenthub-edge/main_test.go +++ b/edge-server/cmd/agenthub-edge/main_test.go @@ -21,10 +21,22 @@ func TestBuildConfigDefaultsToMemoryStore(t *testing.T) { if cfg.StoreFile != "" { t.Fatalf("StoreFile = %q, want empty", cfg.StoreFile) } + if cfg.RunnerCommand != "" { + t.Fatalf("RunnerCommand = %q, want empty", cfg.RunnerCommand) + } + if len(cfg.RunnerArgs) != 0 { + t.Fatalf("RunnerArgs = %#v, want empty", cfg.RunnerArgs) + } } func TestBuildConfigParsesStoreFile(t *testing.T) { - cfg, err := buildConfig([]string{"--addr", "127.0.0.1:4321", "--store-file", "edge-store.json"}) + cfg, err := buildConfig([]string{ + "--addr", "127.0.0.1:4321", + "--store-file", "edge-store.json", + "--runner-command", "agenthub-runner", + "--runner-arg", "--mock", + "--runner-arg", "--addr=127.0.0.1:0", + }) if err != nil { t.Fatalf("buildConfig returned error: %v", err) } @@ -35,6 +47,12 @@ func TestBuildConfigParsesStoreFile(t *testing.T) { if cfg.StoreFile != "edge-store.json" { t.Fatalf("StoreFile = %q, want parsed path", cfg.StoreFile) } + if cfg.RunnerCommand != "agenthub-runner" { + t.Fatalf("RunnerCommand = %q, want parsed command", cfg.RunnerCommand) + } + if got, want := []string(cfg.RunnerArgs), []string{"--mock", "--addr=127.0.0.1:0"}; strings.Join(got, "\x00") != strings.Join(want, "\x00") { + t.Fatalf("RunnerArgs = %#v, want %#v", got, want) + } } func TestBuildConfigRejectsUnexpectedArguments(t *testing.T) { @@ -44,6 +62,13 @@ func TestBuildConfigRejectsUnexpectedArguments(t *testing.T) { } } +func TestBuildConfigRejectsRunnerArgsWithoutCommand(t *testing.T) { + _, err := buildConfig([]string{"--runner-arg", "--mock"}) + if err == nil || !strings.Contains(err.Error(), "--runner-arg requires --runner-command") { + t.Fatalf("buildConfig error = %v, want runner command requirement", err) + } +} + func TestNewStoreFromConfigUsesMemoryStoreByDefault(t *testing.T) { repository, err := newStoreFromConfig(config{}) if err != nil { diff --git a/edge-server/internal/httpserver/server.go b/edge-server/internal/httpserver/server.go index 8148078c6..a922137d2 100644 --- a/edge-server/internal/httpserver/server.go +++ b/edge-server/internal/httpserver/server.go @@ -11,6 +11,7 @@ import ( "github.com/agenthub/edge-server/internal/api" "github.com/agenthub/edge-server/internal/events" + "github.com/agenthub/edge-server/internal/lifecycle" "github.com/agenthub/edge-server/internal/runners" "github.com/agenthub/edge-server/internal/security" "github.com/agenthub/edge-server/internal/store" @@ -18,8 +19,9 @@ import ( // Config holds server configuration. type Config struct { - Addr string - Store store.Repository + Addr string + Store store.Repository + ProcessExecutor lifecycle.ProcessExecutorConfig } // Run starts the HTTP server and blocks until a shutdown signal is received. @@ -27,14 +29,27 @@ func Run(cfg Config) error { if cfg.Addr == "" { cfg.Addr = "127.0.0.1:3210" } + if cfg.Store == nil { + cfg.Store = store.New() + } bus := events.NewBus(10000) registry := runners.NewRegistry() + var executor lifecycle.RunExecutor + if cfg.ProcessExecutor.Command != "" { + processExecutor, err := lifecycle.NewProcessExecutor(bus, cfg.Store, cfg.ProcessExecutor) + if err != nil { + return err + } + executor = processExecutor + } + handler := &api.Handler{ Bus: bus, Registry: registry, Store: cfg.Store, + Executor: executor, } mux := http.NewServeMux() diff --git a/edge-server/internal/lifecycle/process_executor.go b/edge-server/internal/lifecycle/process_executor.go new file mode 100644 index 000000000..219b2101c --- /dev/null +++ b/edge-server/internal/lifecycle/process_executor.go @@ -0,0 +1,228 @@ +package lifecycle + +import ( + "context" + "errors" + "fmt" + "io" + "os" + "os/exec" + "sync" + + "github.com/agenthub/edge-server/internal/events" + "github.com/agenthub/edge-server/internal/store" +) + +var ErrProcessBusRequired = errors.New("process event bus is required") +var ErrProcessCommandRequired = errors.New("process command is required") +var ErrProcessStoreRequired = errors.New("process store is required") + +type ProcessExecutorConfig struct { + Command string + Args []string + Env []string +} + +type ProcessExecutor struct { + bus *events.Bus + store store.RunLifecycleStore + command string + args []string + env []string + + mu sync.Mutex + running map[string]context.CancelFunc +} + +func NewProcessExecutor(bus *events.Bus, store store.RunLifecycleStore, cfg ProcessExecutorConfig) (*ProcessExecutor, error) { + if bus == nil { + return nil, ErrProcessBusRequired + } + if store == nil { + return nil, ErrProcessStoreRequired + } + if cfg.Command == "" { + return nil, ErrProcessCommandRequired + } + return &ProcessExecutor{ + bus: bus, + store: store, + command: cfg.Command, + args: append([]string(nil), cfg.Args...), + env: append([]string(nil), cfg.Env...), + running: make(map[string]context.CancelFunc), + }, nil +} + +func (e *ProcessExecutor) Start(run store.Run) error { + current, ok := e.store.GetRun(run.ID) + if !ok { + return store.ErrNotFound + } + if current.Status != "queued" { + return ErrRunAlreadyStarted + } + + ctx, cancel := context.WithCancel(context.Background()) + e.mu.Lock() + if _, ok := e.running[run.ID]; ok { + e.mu.Unlock() + cancel() + return ErrRunAlreadyStarted + } + e.running[run.ID] = cancel + e.mu.Unlock() + + go e.run(ctx, run) + return nil +} + +func (e *ProcessExecutor) Cancel(runID string) CancelResult { + run, ok := e.store.GetRun(runID) + if !ok { + return CancelResult{Found: false, Status: "not_found"} + } + switch run.Status { + case "queued", "started", "cancelling": + default: + return CancelResult{Run: run, Found: true, Status: run.Status} + } + + e.mu.Lock() + cancel, ok := e.running[runID] + e.mu.Unlock() + if !ok { + return CancelResult{Found: false, Status: "not_running"} + } + cancel() + + run, ok = e.store.SetRunStatusIf(runID, "cancelling", "queued", "started", "cancelling") + if !ok { + if current, found := e.store.GetRun(runID); found { + return CancelResult{Run: current, Found: true, Status: current.Status} + } + return CancelResult{Found: false, Status: "not_found"} + } + return CancelResult{Run: run, Found: true, Status: run.Status} +} + +func (e *ProcessExecutor) run(ctx context.Context, run store.Run) { + defer e.finish(run.ID) + + cmd := exec.CommandContext(ctx, e.command, e.args...) + cmd.Env = e.envForRun(run) + stdout, err := cmd.StdoutPipe() + if err != nil { + e.publishFailed(run, fmt.Errorf("open stdout pipe: %w", err)) + return + } + stderr, err := cmd.StderrPipe() + if err != nil { + e.publishFailed(run, fmt.Errorf("open stderr pipe: %w", err)) + return + } + if err := cmd.Start(); err != nil { + if ctx.Err() != nil { + e.publishCancelled(run) + return + } + e.publishFailed(run, err) + return + } + + started, ok := e.store.SetRunStatusIf(run.ID, "started", "queued") + if ok { + e.bus.Publish("run.started", runScope(started), RunResponse(started)) + } + + var wg sync.WaitGroup + wg.Add(2) + go e.publishOutput(&wg, run, "stdout", stdout) + go e.publishOutput(&wg, run, "stderr", stderr) + + waitErr := cmd.Wait() + wg.Wait() + + if ctx.Err() != nil || e.runStatus(run.ID) == "cancelling" { + e.publishCancelled(run) + return + } + if waitErr != nil { + e.publishFailed(run, waitErr) + return + } + finished, ok := e.store.SetRunStatusIf(run.ID, "finished", "started") + if ok { + e.bus.Publish("run.finished", runScope(finished), RunResponse(finished)) + } +} + +func (e *ProcessExecutor) publishOutput(wg *sync.WaitGroup, run store.Run, stream string, reader io.Reader) { + defer wg.Done() + + buf := make([]byte, 32*1024) + offset := 0 + for { + n, err := reader.Read(buf) + if n > 0 { + text := string(buf[:n]) + e.bus.Publish("run.output.batch", runScope(run), map[string]any{ + "runId": run.ID, + "stream": stream, + "chunks": []map[string]any{ + {"offset": offset, "text": text}, + }, + }) + offset += n + } + if err != nil { + return + } + } +} + +func (e *ProcessExecutor) envForRun(run store.Run) []string { + env := e.env + if env == nil { + env = os.Environ() + } else { + env = append([]string(nil), env...) + } + return append(env, + "AGENTHUB_RUN_ID="+run.ID, + "AGENTHUB_PROJECT_ID="+run.ProjectID, + "AGENTHUB_THREAD_ID="+run.ThreadID, + ) +} + +func (e *ProcessExecutor) publishFailed(run store.Run, err error) { + failed, ok := e.store.SetRunStatusIf(run.ID, "failed", "queued", "started") + if ok { + e.bus.Publish("run.failed", runScope(failed), map[string]any{ + "runId": failed.ID, + "status": failed.Status, + "error": err.Error(), + }) + } +} + +func (e *ProcessExecutor) publishCancelled(run store.Run) { + cancelled, ok := e.store.SetRunStatusIf(run.ID, "cancelled", "queued", "started", "cancelling") + if ok { + e.bus.Publish("run.cancelled", runScope(cancelled), RunResponse(cancelled)) + } +} + +func (e *ProcessExecutor) runStatus(runID string) string { + run, ok := e.store.GetRun(runID) + if !ok { + return "" + } + return run.Status +} + +func (e *ProcessExecutor) finish(runID string) { + e.mu.Lock() + delete(e.running, runID) + e.mu.Unlock() +} diff --git a/edge-server/internal/lifecycle/process_executor_test.go b/edge-server/internal/lifecycle/process_executor_test.go new file mode 100644 index 000000000..eb4385891 --- /dev/null +++ b/edge-server/internal/lifecycle/process_executor_test.go @@ -0,0 +1,278 @@ +package lifecycle + +import ( + "errors" + "fmt" + "os" + "strings" + "testing" + "time" + + "github.com/agenthub/edge-server/internal/events" + "github.com/agenthub/edge-server/internal/store" +) + +func TestProcessExecutorRequiresCommand(t *testing.T) { + _, err := NewProcessExecutor(events.NewBus(10), store.New(), ProcessExecutorConfig{}) + if !errors.Is(err, ErrProcessCommandRequired) { + t.Fatalf("NewProcessExecutor error = %v, want ErrProcessCommandRequired", err) + } +} + +func TestProcessExecutorRequiresDependencies(t *testing.T) { + _, err := NewProcessExecutor(nil, store.New(), ProcessExecutorConfig{Command: os.Args[0]}) + if !errors.Is(err, ErrProcessBusRequired) { + t.Fatalf("NewProcessExecutor nil bus error = %v, want ErrProcessBusRequired", err) + } + _, err = NewProcessExecutor(events.NewBus(10), nil, ProcessExecutorConfig{Command: os.Args[0]}) + if !errors.Is(err, ErrProcessStoreRequired) { + t.Fatalf("NewProcessExecutor nil store error = %v, want ErrProcessStoreRequired", err) + } +} + +func TestProcessExecutorRejectsMissingRun(t *testing.T) { + bus := events.NewBus(100) + s := store.New() + run := store.Run{ + ID: "run_missing", + ProjectID: "proj_missing", + ThreadID: "thread_missing", + Status: "queued", + } + _, ch, _ := bus.Subscribe(0) + executor := newTestProcessExecutor(t, bus, s, "success") + + if err := executor.Start(run); !errors.Is(err, store.ErrNotFound) { + t.Fatalf("Start missing run error = %v, want store.ErrNotFound", err) + } + select { + case evt := <-ch: + t.Fatalf("unexpected event after missing run start: %s", evt.Type) + case <-time.After(50 * time.Millisecond): + } +} + +func TestProcessExecutorPublishesOutputAndFinished(t *testing.T) { + bus := events.NewBus(100) + s := store.New() + run := newExecutorTestRun(t, s) + _, ch, _ := bus.Subscribe(0) + executor := newTestProcessExecutor(t, bus, s, "success") + + if err := executor.Start(run); err != nil { + t.Fatalf("Start returned error: %v", err) + } + + seenOutput := map[string]bool{} + var stdoutText string + for { + evt := nextEvent(t, ch) + if evt.Scope["runId"] != run.ID { + t.Fatalf("event scope runId = %#v, want %q", evt.Scope["runId"], run.ID) + } + switch evt.Type { + case "run.started": + case "run.output.batch": + payload, ok := evt.Payload.(map[string]any) + if !ok { + t.Fatalf("output payload = %T, want map", evt.Payload) + } + stream, _ := payload["stream"].(string) + seenOutput[stream] = true + if stream == "stdout" { + chunks, ok := payload["chunks"].([]map[string]any) + if !ok || len(chunks) == 0 { + t.Fatalf("output chunks = %#v, want non-empty []map[string]any", payload["chunks"]) + } + text, _ := chunks[0]["text"].(string) + stdoutText += text + } + case "run.finished": + if !seenOutput["stdout"] || !seenOutput["stderr"] { + t.Fatalf("seen output streams = %#v, want stdout and stderr", seenOutput) + } + for _, want := range []string{ + "run=" + run.ID, + "project=" + run.ProjectID, + "thread=" + run.ThreadID, + } { + if !strings.Contains(stdoutText, want) { + t.Fatalf("stdout text = %q, want %q", stdoutText, want) + } + } + stored, ok := s.GetRun(run.ID) + if !ok { + t.Fatalf("run %q was not stored", run.ID) + } + if stored.Status != "finished" || stored.StartedAt == "" || stored.FinishedAt == "" { + t.Fatalf("stored run = %#v, want finished with timestamps", stored) + } + return + default: + t.Fatalf("unexpected event type %q", evt.Type) + } + } +} + +func TestProcessExecutorPublishesFailedForNonZeroExit(t *testing.T) { + bus := events.NewBus(100) + s := store.New() + run := newExecutorTestRun(t, s) + _, ch, _ := bus.Subscribe(0) + executor := newTestProcessExecutor(t, bus, s, "fail") + + if err := executor.Start(run); err != nil { + t.Fatalf("Start returned error: %v", err) + } + + var sawStarted bool + for { + evt := nextEvent(t, ch) + switch evt.Type { + case "run.started": + sawStarted = true + case "run.output.batch": + case "run.failed": + if !sawStarted { + t.Fatal("run.failed arrived before run.started") + } + payload, ok := evt.Payload.(map[string]any) + if !ok { + t.Fatalf("failed payload = %T, want map", evt.Payload) + } + if payload["status"] != "failed" || payload["error"] == "" { + t.Fatalf("failed payload = %#v, want failed status and error", payload) + } + return + default: + t.Fatalf("unexpected event type %q", evt.Type) + } + } +} + +func TestProcessExecutorPublishesFailedWhenCommandCannotStart(t *testing.T) { + bus := events.NewBus(100) + s := store.New() + run := newExecutorTestRun(t, s) + _, ch, _ := bus.Subscribe(0) + executor, err := NewProcessExecutor(bus, s, ProcessExecutorConfig{Command: "agenthub-missing-command-for-test"}) + if err != nil { + t.Fatalf("NewProcessExecutor returned error: %v", err) + } + + if err := executor.Start(run); err != nil { + t.Fatalf("Start returned error: %v", err) + } + + evt := nextEvent(t, ch) + if evt.Type != "run.failed" { + t.Fatalf("event type = %q, want run.failed", evt.Type) + } + payload, ok := evt.Payload.(map[string]any) + if !ok { + t.Fatalf("failed payload = %T, want map", evt.Payload) + } + if payload["status"] != "failed" || payload["error"] == "" { + t.Fatalf("failed payload = %#v, want failed status and error", payload) + } + stored, ok := s.GetRun(run.ID) + if !ok { + t.Fatalf("run %q was not stored", run.ID) + } + if stored.Status != "failed" { + t.Fatalf("stored run status = %q, want failed", stored.Status) + } +} + +func TestProcessExecutorRejectsDuplicateStart(t *testing.T) { + bus := events.NewBus(100) + s := store.New() + run := newExecutorTestRun(t, s) + executor := newTestProcessExecutor(t, bus, s, "sleep") + + if err := executor.Start(run); err != nil { + t.Fatalf("first Start returned error: %v", err) + } + if err := executor.Start(run); !errors.Is(err, ErrRunAlreadyStarted) { + t.Fatalf("second Start error = %v, want ErrRunAlreadyStarted", err) + } + _ = executor.Cancel(run.ID) +} + +func TestProcessExecutorCancelPublishesCancelledEvent(t *testing.T) { + bus := events.NewBus(100) + s := store.New() + run := newExecutorTestRun(t, s) + _, ch, _ := bus.Subscribe(0) + executor := newTestProcessExecutor(t, bus, s, "sleep") + + if err := executor.Start(run); err != nil { + t.Fatalf("Start returned error: %v", err) + } + + result := executor.Cancel(run.ID) + if !result.Found || result.Status != "cancelling" { + t.Fatalf("Cancel result = %#v, want found cancelling", result) + } + + for { + evt := nextEvent(t, ch) + if evt.Type == "run.cancelled" { + stored, ok := s.GetRun(run.ID) + if !ok { + t.Fatalf("run %q was not stored", run.ID) + } + if stored.Status != "cancelled" { + t.Fatalf("stored run status = %q, want cancelled", stored.Status) + } + return + } + } +} + +func TestProcessExecutorCancelMissingRun(t *testing.T) { + executor := newTestProcessExecutor(t, events.NewBus(10), store.New(), "success") + + result := executor.Cancel("run_missing") + if result.Found || result.Status != "not_found" { + t.Fatalf("Cancel missing result = %#v, want not_found", result) + } +} + +func newTestProcessExecutor(t *testing.T, bus *events.Bus, s store.RunLifecycleStore, mode string) *ProcessExecutor { + t.Helper() + + executor, err := NewProcessExecutor(bus, s, ProcessExecutorConfig{ + Command: os.Args[0], + Args: []string{"-test.run=TestProcessExecutorHelper", "--", mode}, + Env: append(os.Environ(), "AGENTHUB_PROCESS_EXECUTOR_HELPER=1"), + }) + if err != nil { + t.Fatalf("NewProcessExecutor returned error: %v", err) + } + return executor +} + +func TestProcessExecutorHelper(t *testing.T) { + if os.Getenv("AGENTHUB_PROCESS_EXECUTOR_HELPER") != "1" { + return + } + mode := os.Args[len(os.Args)-1] + switch mode { + case "success": + fmt.Fprint(os.Stdout, "stdout chunk\n") + fmt.Fprintf(os.Stdout, "run=%s\n", os.Getenv("AGENTHUB_RUN_ID")) + fmt.Fprintf(os.Stdout, "project=%s\n", os.Getenv("AGENTHUB_PROJECT_ID")) + fmt.Fprintf(os.Stdout, "thread=%s\n", os.Getenv("AGENTHUB_THREAD_ID")) + fmt.Fprint(os.Stderr, "stderr chunk\n") + case "fail": + fmt.Fprint(os.Stderr, "failure chunk\n") + os.Exit(7) + case "sleep": + time.Sleep(5 * time.Second) + default: + fmt.Fprintf(os.Stderr, "unknown helper mode %q\n", mode) + os.Exit(2) + } + os.Exit(0) +}