From 8fed823b869f6763a7dd159cde9b31be3dba3b2d Mon Sep 17 00:00:00 2001 From: Delicious233 Date: Sat, 23 May 2026 06:32:13 +0800 Subject: [PATCH] =?UTF-8?q?feat(client):=20=E5=A2=9E=E5=8A=A0=20Runner=20a?= =?UTF-8?q?dapter=20profile?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/roadmap.md | 7 +- ...ent-runner-adapter-profile-delicious233.md | 34 +++ docs/roadmaps/client.md | 3 +- edge-server/cmd/agenthub-edge/main.go | 15 +- edge-server/cmd/agenthub-edge/main_test.go | 27 ++ .../internal/lifecycle/process_executor.go | 45 ++-- .../lifecycle/process_executor_test.go | 239 ++++++++++++++++++ .../internal/lifecycle/process_profile.go | 164 ++++++++++++ 8 files changed, 508 insertions(+), 26 deletions(-) create mode 100644 docs/roadmaps/branches/feat-client-runner-adapter-profile-delicious233.md create mode 100644 edge-server/internal/lifecycle/process_profile.go diff --git a/docs/roadmap.md b/docs/roadmap.md index a06d1d8d8..2454eb9ae 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`，`feat/client-runner-process-adapter-delicious233` 已补本地进程 executor 边界，`feat/client-runner-workdir-delicious233` 已补本地进程工作目录边界，下一步重点是真实 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 边界，`feat/client-runner-workdir-delicious233` 已补本地进程工作目录边界，`feat/client-runner-adapter-profile-delicious233` 已补 generic adapter profile / 命令模板最小层，下一步重点是真实 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 进程、取消、日志、错误映射。本地进程 executor 和本地进程工作目录边界已补齐，但还不是 Claude Code / Codex / OpenCode 的完整 adapter。 +- [ ] M3 真实 Runner：CLI Agent 进程、取消、日志、错误映射。本地进程 executor、本地进程工作目录边界和 generic adapter profile / 命令模板最小层已补齐，但还不是 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 启动入口，`feat/client-runner-process-adapter-delicious233` 增加可测试的本地进程 executor，`feat/client-runner-workdir-delicious233` 增加本地进程工作目录配置，后续继续做真实 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，`feat/client-runner-workdir-delicious233` 增加本地进程工作目录配置，`feat/client-runner-adapter-profile-delicious233` 增加可测试的 generic adapter profile / 命令模板最小层，后续继续做真实 Runner adapter。 ## 验收门槛 @@ -52,6 +52,7 @@ - [x] 在客户端 M2 基础上补 `POST /v1/threads/{threadId}/messages` 到 Item / event 的写入链路。 - [x] 增加 Edge 本地进程 executor 边界，支持 stdout/stderr、成功、失败和取消事件映射。 - [x] 为 Edge 本地进程 executor 增加工作目录配置边界；这只是本地进程 workdir 能力，不是完整真实 Runner adapter。 +- [x] 为 Edge 本地进程 executor 增加 generic adapter profile / 命令模板最小层，支持从 Run 上下文展开 args/env。 - [ ] 将 Runner 真正接入 Edge Run lifecycle，替换 handler 内置 mock flow。 - [ ] M2 完成后归档或更新 `docs/client-roadmap.md`，避免路线图重复。 - [ ] 为 Runner 真实 CLI adapter 规划最小测试夹具。 diff --git a/docs/roadmaps/branches/feat-client-runner-adapter-profile-delicious233.md b/docs/roadmaps/branches/feat-client-runner-adapter-profile-delicious233.md new file mode 100644 index 000000000..9cabe1a28 --- /dev/null +++ b/docs/roadmaps/branches/feat-client-runner-adapter-profile-delicious233.md @@ -0,0 +1,34 @@ +# feat/client-runner-adapter-profile-delicious233 路线图 + +最后更新：2026-05-23 + +## 当前目标 + +- [x] 为 Edge ProcessExecutor 增加 generic adapter profile / 命令模板最小层，支持从 Run 上下文展开 CLI args/env。 + +## 写入范围 + +- `edge-server/internal/lifecycle/` +- `edge-server/cmd/agenthub-edge/` +- `docs/roadmap.md` +- `docs/roadmaps/client.md` +- `docs/roadmaps/branches/feat-client-runner-adapter-profile-delicious233.md` + +## 已完成 + +- [x] 新增 `RunnerProfile`、`CommandTemplate`、`RunProcessContext` 内部边界。 +- [x] 支持 `{{run.id}}`、`{{run.projectId}}`、`{{run.threadId}}` 在 args/env 中展开。 +- [x] 保留 `--runner-command`、`--runner-arg`、`--runner-workdir` 行为，并新增 repeatable `--runner-env KEY=VALUE`。 +- [x] 构造期校验空 command、未知占位符和 env 格式错误。 +- [x] 完成交叉 review 后修复环境继承、env 错误脱敏和 `ExtraEnv` 边界，系统环境原样继承，只有显式 `--runner-env` 参与模板展开。 + +## 下一步 + +- [ ] 后续分支接入 Claude Code / Codex / OpenCode 具体 adapter profile。 + +## 验收 + +- [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 ./...` diff --git a/docs/roadmaps/client.md b/docs/roadmaps/client.md index e49f94f8b..9849a7f15 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 启动参数，`feat/client-runner-process-adapter-delicious233` 已补本地进程 executor 边界，`feat/client-runner-workdir-delicious233` 已补本地进程工作目录配置边界，后续继续补真实 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 边界，`feat/client-runner-workdir-delicious233` 已补本地进程工作目录配置边界，`feat/client-runner-adapter-profile-delicious233` 已补 generic adapter profile / 命令模板最小层，后续继续补真实 Runner adapter。 ## 近期任务 @@ -30,6 +30,7 @@ - [x] 抽出 Edge Run lifecycle executor 边界，替换 handler 内置 mock flow。 - [x] 增加可测试的本地进程 executor，覆盖 stdout/stderr 输出、正常退出、非零退出、取消和重复启动。 - [x] 增加本地进程工作目录配置，覆盖构造期目录验证和子进程实际运行目录；这不是完整 Claude Code / Codex / OpenCode adapter。 +- [x] 增加 generic adapter profile / 命令模板最小层，覆盖 args/env 的 Run 占位符展开、未知占位符错误、固定 args 兼容和 workdir 不回退。 - [ ] 将真实 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 418ef9044..f5874777e 100644 --- a/edge-server/cmd/agenthub-edge/main.go +++ b/edge-server/cmd/agenthub-edge/main.go @@ -17,6 +17,7 @@ type config struct { StoreFile string RunnerCommand string RunnerArgs repeatedString + RunnerEnv repeatedString RunnerWorkDir string } @@ -51,9 +52,10 @@ func main() { serverConfig := httpserver.Config{Addr: cfg.Addr, Store: repository} if cfg.RunnerCommand != "" { serverConfig.ProcessExecutor = lifecycle.ProcessExecutorConfig{ - Command: cfg.RunnerCommand, - Args: append([]string(nil), cfg.RunnerArgs...), - WorkDir: cfg.RunnerWorkDir, + Command: cfg.RunnerCommand, + Args: append([]string(nil), cfg.RunnerArgs...), + ExtraEnv: append([]string(nil), cfg.RunnerEnv...), + WorkDir: cfg.RunnerWorkDir, } } @@ -73,6 +75,7 @@ func buildConfig(args []string) (config, error) { fs.StringVar(&cfg.RunnerCommand, "runner-command", "", "local process command to execute for each run; empty uses the mock executor") fs.StringVar(&cfg.RunnerWorkDir, "runner-workdir", "", "working directory for --runner-command; empty inherits the edge process working directory") fs.Var(&cfg.RunnerArgs, "runner-arg", "argument passed to --runner-command; may be repeated") + fs.Var(&cfg.RunnerEnv, "runner-env", "environment variable passed to --runner-command as KEY=VALUE; may be repeated") if err := fs.Parse(args); err != nil { return config{}, err } @@ -80,9 +83,15 @@ func buildConfig(args []string) (config, error) { if cfg.RunnerCommand == "" && len(cfg.RunnerArgs) > 0 { return config{}, fmt.Errorf("--runner-arg requires --runner-command") } + if cfg.RunnerCommand == "" && len(cfg.RunnerEnv) > 0 { + return config{}, fmt.Errorf("--runner-env requires --runner-command") + } if cfg.RunnerCommand == "" && cfg.RunnerWorkDir != "" { return config{}, fmt.Errorf("--runner-workdir requires --runner-command") } + if _, err := lifecycle.NewCommandTemplate(nil, cfg.RunnerEnv); err != nil { + return config{}, fmt.Errorf("--runner-env: %w", err) + } 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 c979b8615..58e594e5d 100644 --- a/edge-server/cmd/agenthub-edge/main_test.go +++ b/edge-server/cmd/agenthub-edge/main_test.go @@ -30,6 +30,9 @@ func TestBuildConfigDefaultsToMemoryStore(t *testing.T) { if len(cfg.RunnerArgs) != 0 { t.Fatalf("RunnerArgs = %#v, want empty", cfg.RunnerArgs) } + if len(cfg.RunnerEnv) != 0 { + t.Fatalf("RunnerEnv = %#v, want empty", cfg.RunnerEnv) + } } func TestBuildConfigParsesStoreFile(t *testing.T) { @@ -40,6 +43,8 @@ func TestBuildConfigParsesStoreFile(t *testing.T) { "--runner-workdir", "workspace", "--runner-arg", "--mock", "--runner-arg", "--addr=127.0.0.1:0", + "--runner-env", "AGENTHUB_PROFILE_RUN={{run.id}}", + "--runner-env", "AGENTHUB_PROFILE_THREAD={{run.threadId}}", }) if err != nil { t.Fatalf("buildConfig returned error: %v", err) @@ -60,6 +65,9 @@ func TestBuildConfigParsesStoreFile(t *testing.T) { 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) } + if got, want := []string(cfg.RunnerEnv), []string{"AGENTHUB_PROFILE_RUN={{run.id}}", "AGENTHUB_PROFILE_THREAD={{run.threadId}}"}; strings.Join(got, "\x00") != strings.Join(want, "\x00") { + t.Fatalf("RunnerEnv = %#v, want %#v", got, want) + } } func TestBuildConfigRejectsUnexpectedArguments(t *testing.T) { @@ -76,6 +84,25 @@ func TestBuildConfigRejectsRunnerArgsWithoutCommand(t *testing.T) { } } +func TestBuildConfigRejectsRunnerEnvWithoutCommand(t *testing.T) { + _, err := buildConfig([]string{"--runner-env", "AGENTHUB_PROFILE_RUN={{run.id}}"}) + if err == nil || !strings.Contains(err.Error(), "--runner-env requires --runner-command") { + t.Fatalf("buildConfig error = %v, want runner command requirement", err) + } +} + +func TestBuildConfigRejectsInvalidRunnerEnv(t *testing.T) { + tests := []string{"AGENTHUB_PROFILE_RUN", "=value"} + for _, value := range tests { + t.Run(value, func(t *testing.T) { + _, err := buildConfig([]string{"--runner-command", "agenthub-runner", "--runner-env", value}) + if err == nil || !strings.Contains(err.Error(), "--runner-env") { + t.Fatalf("buildConfig error = %v, want runner env validation error", err) + } + }) + } +} + func TestBuildConfigRejectsRunnerWorkDirWithoutCommand(t *testing.T) { _, err := buildConfig([]string{"--runner-workdir", "workspace"}) if err == nil || !strings.Contains(err.Error(), "--runner-workdir requires --runner-command") { diff --git a/edge-server/internal/lifecycle/process_executor.go b/edge-server/internal/lifecycle/process_executor.go index cab725675..27fe3f1ab 100644 --- a/edge-server/internal/lifecycle/process_executor.go +++ b/edge-server/internal/lifecycle/process_executor.go @@ -18,19 +18,17 @@ 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 - WorkDir string + Command string + Args []string + Env []string + ExtraEnv []string + WorkDir string } type ProcessExecutor struct { bus *events.Bus store store.RunLifecycleStore - command string - args []string - env []string - workDir string + profile RunnerProfile mu sync.Mutex running map[string]context.CancelFunc @@ -43,8 +41,9 @@ func NewProcessExecutor(bus *events.Bus, store store.RunLifecycleStore, cfg Proc if store == nil { return nil, ErrProcessStoreRequired } - if cfg.Command == "" { - return nil, ErrProcessCommandRequired + profile, err := NewGenericRunnerProfile(cfg.Command, cfg.Args, cfg.Env, cfg.ExtraEnv, cfg.WorkDir) + if err != nil { + return nil, err } if cfg.WorkDir != "" { info, err := os.Stat(cfg.WorkDir) @@ -58,10 +57,7 @@ func NewProcessExecutor(bus *events.Bus, store store.RunLifecycleStore, cfg Proc return &ProcessExecutor{ bus: bus, store: store, - command: cfg.Command, - args: append([]string(nil), cfg.Args...), - env: append([]string(nil), cfg.Env...), - workDir: cfg.WorkDir, + profile: profile, running: make(map[string]context.CancelFunc), }, nil } @@ -121,9 +117,19 @@ func (e *ProcessExecutor) Cancel(runID string) CancelResult { func (e *ProcessExecutor) run(ctx context.Context, run store.Run) { defer e.finish(run.ID) - cmd := exec.CommandContext(ctx, e.command, e.args...) - cmd.Dir = e.workDir - cmd.Env = e.envForRun(run) + args, env, err := e.profile.Template.Expand(RunProcessContext{Run: run}) + if err != nil { + e.publishFailed(run, err) + return + } + _, extraEnv, err := e.profile.ExtraEnvTemplate.Expand(RunProcessContext{Run: run}) + if err != nil { + e.publishFailed(run, err) + return + } + cmd := exec.CommandContext(ctx, e.profile.Command, args...) + cmd.Dir = e.profile.WorkDir + cmd.Env = e.envForRun(run, env, extraEnv) stdout, err := cmd.StdoutPipe() if err != nil { e.publishFailed(run, fmt.Errorf("open stdout pipe: %w", err)) @@ -194,13 +200,14 @@ func (e *ProcessExecutor) publishOutput(wg *sync.WaitGroup, run store.Run, strea } } -func (e *ProcessExecutor) envForRun(run store.Run) []string { - env := e.env +func (e *ProcessExecutor) envForRun(run store.Run, profileEnv, extraEnv []string) []string { + env := profileEnv if env == nil { env = os.Environ() } else { env = append([]string(nil), env...) } + env = append(env, extraEnv...) return append(env, "AGENTHUB_RUN_ID="+run.ID, "AGENTHUB_PROJECT_ID="+run.ProjectID, diff --git a/edge-server/internal/lifecycle/process_executor_test.go b/edge-server/internal/lifecycle/process_executor_test.go index fecfbaa31..a3e0bb0ca 100644 --- a/edge-server/internal/lifecycle/process_executor_test.go +++ b/edge-server/internal/lifecycle/process_executor_test.go @@ -206,6 +206,207 @@ func TestProcessExecutorRunsCommandInConfiguredWorkDir(t *testing.T) { } } +func TestProcessExecutorExpandsRunPlaceholdersInArgs(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: os.Args[0], + Args: []string{ + "-test.run=TestProcessExecutorHelper", + "--", + "--run={{run.id}}", + "--project={{ run.projectId }}", + "--thread={{run.threadId}}", + "args", + }, + Env: append(os.Environ(), "AGENTHUB_PROCESS_EXECUTOR_HELPER=1"), + }) + if err != nil { + t.Fatalf("NewProcessExecutor returned error: %v", err) + } + + if err := executor.Start(run); err != nil { + t.Fatalf("Start returned error: %v", err) + } + + stdoutText := collectStdoutUntilFinished(t, ch) + 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) + } + } +} + +func TestProcessExecutorExpandsRunPlaceholdersInEnv(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: os.Args[0], + Args: []string{"-test.run=TestProcessExecutorHelper", "--", "env"}, + Env: append(os.Environ(), + "AGENTHUB_PROCESS_EXECUTOR_HELPER=1", + "PROFILE_RUN={{run.id}}", + "PROFILE_PROJECT={{run.projectId}}", + "PROFILE_THREAD={{run.threadId}}", + ), + }) + if err != nil { + t.Fatalf("NewProcessExecutor returned error: %v", err) + } + + if err := executor.Start(run); err != nil { + t.Fatalf("Start returned error: %v", err) + } + + stdoutText := collectStdoutUntilFinished(t, ch) + for _, want := range []string{ + "profileRun=" + run.ID, + "profileProject=" + run.ProjectID, + "profileThread=" + run.ThreadID, + } { + if !strings.Contains(stdoutText, want) { + t.Fatalf("stdout text = %q, want %q", stdoutText, want) + } + } +} + +func TestProcessExecutorExpandsRunPlaceholdersInExtraEnv(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: os.Args[0], + Args: []string{"-test.run=TestProcessExecutorHelper", "--", "env"}, + Env: nil, + ExtraEnv: []string{ + "AGENTHUB_PROCESS_EXECUTOR_HELPER=1", + "PROFILE_RUN={{run.id}}", + "PROFILE_PROJECT={{run.projectId}}", + "PROFILE_THREAD={{run.threadId}}", + }, + }) + if err != nil { + t.Fatalf("NewProcessExecutor returned error: %v", err) + } + + if err := executor.Start(run); err != nil { + t.Fatalf("Start returned error: %v", err) + } + + stdoutText := collectStdoutUntilFinished(t, ch) + for _, want := range []string{ + "profileRun=" + run.ID, + "profileProject=" + run.ProjectID, + "profileThread=" + run.ThreadID, + } { + if !strings.Contains(stdoutText, want) { + t.Fatalf("stdout text = %q, want %q", stdoutText, want) + } + } +} + +func TestProcessExecutorExtraEnvDoesNotTemplateParentEnvironment(t *testing.T) { + t.Setenv("AGENTHUB_PARENT_ENV_WITH_BRACES", "{{not.a.template") + + _, err := NewProcessExecutor(events.NewBus(10), store.New(), ProcessExecutorConfig{ + Command: os.Args[0], + Args: []string{"-test.run=TestProcessExecutorHelper", "--", "env"}, + Env: nil, + ExtraEnv: []string{"PROFILE_RUN={{run.id}}"}, + }) + if err != nil { + t.Fatalf("NewProcessExecutor returned error: %v", err) + } +} + +func TestProcessExecutorNilEnvInheritsParentEnvironment(t *testing.T) { + t.Setenv("AGENTHUB_PROCESS_EXECUTOR_HELPER", "1") + t.Setenv("AGENTHUB_INHERITED_ENV_FOR_TEST", "expected") + + bus := events.NewBus(100) + s := store.New() + run := newExecutorTestRun(t, s) + _, ch, _ := bus.Subscribe(0) + executor, err := NewProcessExecutor(bus, s, ProcessExecutorConfig{ + Command: os.Args[0], + Args: []string{"-test.run=TestProcessExecutorHelper", "--", "inherited-env"}, + Env: nil, + }) + if err != nil { + t.Fatalf("NewProcessExecutor returned error: %v", err) + } + + if err := executor.Start(run); err != nil { + t.Fatalf("Start returned error: %v", err) + } + + stdoutText := collectStdoutUntilFinished(t, ch) + if !strings.Contains(stdoutText, "inherited=expected") { + t.Fatalf("stdout text = %q, want inherited env value", stdoutText) + } +} + +func TestProcessExecutorRejectsUnknownPlaceholder(t *testing.T) { + _, err := NewProcessExecutor(events.NewBus(10), store.New(), ProcessExecutorConfig{ + Command: os.Args[0], + Args: []string{"--bad={{run.workspaceId}}"}, + }) + if err == nil || !strings.Contains(err.Error(), "unknown placeholder") { + t.Fatalf("NewProcessExecutor error = %v, want unknown placeholder", err) + } +} + +func TestProcessExecutorRejectsInvalidEnvTemplate(t *testing.T) { + tests := []struct { + name string + env []string + want string + }{ + {name: "missing equals", env: []string{"PROFILE_RUN"}, want: "KEY=VALUE"}, + {name: "empty key", env: []string{"=value"}, want: "key is required"}, + {name: "whitespace in key", env: []string{"PROFILE RUN=value"}, want: "invalid whitespace"}, + {name: "unknown placeholder", env: []string{"PROFILE_RUN={{run.workspaceId}}"}, want: "unknown placeholder"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + _, err := NewProcessExecutor(events.NewBus(10), store.New(), ProcessExecutorConfig{ + Command: os.Args[0], + Env: tt.env, + }) + if err == nil || !strings.Contains(err.Error(), tt.want) { + t.Fatalf("NewProcessExecutor error = %v, want containing %q", err, tt.want) + } + }) + } +} + +func TestProcessExecutorRedactsEnvTemplateValueInErrors(t *testing.T) { + secretValue := "token-secret-should-not-appear" + _, err := NewProcessExecutor(events.NewBus(10), store.New(), ProcessExecutorConfig{ + Command: os.Args[0], + Env: []string{"PROFILE_TOKEN=" + secretValue + "{{run.workspaceId}}"}, + }) + if err == nil { + t.Fatal("NewProcessExecutor returned nil error for invalid env placeholder") + } + if strings.Contains(err.Error(), secretValue) { + t.Fatalf("NewProcessExecutor error = %q, must not include env value", err.Error()) + } + if !strings.Contains(err.Error(), "PROFILE_TOKEN") || !strings.Contains(err.Error(), "unknown placeholder") { + t.Fatalf("NewProcessExecutor error = %q, want key and placeholder error", err.Error()) + } +} + func TestProcessExecutorPublishesFailedForNonZeroExit(t *testing.T) { bus := events.NewBus(100) s := store.New() @@ -369,9 +570,47 @@ func TestProcessExecutorHelper(t *testing.T) { os.Exit(3) } fmt.Fprintf(os.Stdout, "cwd=%s\n", filepath.Clean(cwd)) + case "args": + fmt.Fprintf(os.Stdout, "args=%s\n", strings.Join(os.Args, "\n")) + case "env": + fmt.Fprintf(os.Stdout, "profileRun=%s\n", os.Getenv("PROFILE_RUN")) + fmt.Fprintf(os.Stdout, "profileProject=%s\n", os.Getenv("PROFILE_PROJECT")) + fmt.Fprintf(os.Stdout, "profileThread=%s\n", os.Getenv("PROFILE_THREAD")) + case "inherited-env": + fmt.Fprintf(os.Stdout, "inherited=%s\n", os.Getenv("AGENTHUB_INHERITED_ENV_FOR_TEST")) default: fmt.Fprintf(os.Stderr, "unknown helper mode %q\n", mode) os.Exit(2) } os.Exit(0) } + +func collectStdoutUntilFinished(t *testing.T, ch <-chan events.EventEnvelope) string { + t.Helper() + + var stdoutText string + for { + evt := nextEvent(t, ch) + 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) + } + if payload["stream"] != "stdout" { + continue + } + 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": + return stdoutText + default: + t.Fatalf("unexpected event type %q", evt.Type) + } + } +} diff --git a/edge-server/internal/lifecycle/process_profile.go b/edge-server/internal/lifecycle/process_profile.go new file mode 100644 index 000000000..35f716376 --- /dev/null +++ b/edge-server/internal/lifecycle/process_profile.go @@ -0,0 +1,164 @@ +package lifecycle + +import ( + "fmt" + "strings" + + "github.com/agenthub/edge-server/internal/store" +) + +type RunnerProfile struct { + Command string + Template CommandTemplate + ExtraEnvTemplate CommandTemplate + WorkDir string +} + +type CommandTemplate struct { + Args []string + Env []string +} + +type RunProcessContext struct { + Run store.Run +} + +func NewGenericRunnerProfile(command string, args, env, extraEnv []string, workDir string) (RunnerProfile, error) { + command = strings.TrimSpace(command) + if command == "" { + return RunnerProfile{}, ErrProcessCommandRequired + } + template, err := NewCommandTemplate(args, env) + if err != nil { + return RunnerProfile{}, err + } + extraEnvTemplate, err := NewCommandTemplate(nil, extraEnv) + if err != nil { + return RunnerProfile{}, err + } + return RunnerProfile{ + Command: command, + Template: template, + ExtraEnvTemplate: extraEnvTemplate, + WorkDir: workDir, + }, nil +} + +func NewCommandTemplate(args, env []string) (CommandTemplate, error) { + template := CommandTemplate{ + Args: append([]string(nil), args...), + Env: append([]string(nil), env...), + } + for _, arg := range template.Args { + if err := validatePlaceholders(arg); err != nil { + return CommandTemplate{}, fmt.Errorf("process arg template %q: %w", arg, err) + } + } + for _, entry := range template.Env { + key, value, err := parseEnvTemplate(entry) + if err != nil { + return CommandTemplate{}, err + } + if err := validatePlaceholders(value); err != nil { + return CommandTemplate{}, fmt.Errorf("process env %q template: %w", key, err) + } + } + return template, nil +} + +func (t CommandTemplate) Expand(ctx RunProcessContext) ([]string, []string, error) { + args := make([]string, 0, len(t.Args)) + for _, arg := range t.Args { + expanded, err := expandPlaceholders(arg, ctx) + if err != nil { + return nil, nil, fmt.Errorf("expand process arg template %q: %w", arg, err) + } + args = append(args, expanded) + } + + var env []string + if t.Env != nil { + env = make([]string, 0, len(t.Env)) + } + for _, entry := range t.Env { + key, value, err := parseEnvTemplate(entry) + if err != nil { + return nil, nil, err + } + expanded, err := expandPlaceholders(value, ctx) + if err != nil { + return nil, nil, fmt.Errorf("expand process env %q template: %w", key, err) + } + env = append(env, key+"="+expanded) + } + return args, env, nil +} + +func parseEnvTemplate(entry string) (string, string, error) { + key, value, ok := strings.Cut(entry, "=") + if !ok { + return "", "", fmt.Errorf("process env entry must use KEY=VALUE") + } + if err := validateEnvKey(key); err != nil { + return "", "", err + } + return key, value, nil +} + +func validateEnvKey(key string) error { + if key == "" { + return fmt.Errorf("process env key is required") + } + if strings.TrimSpace(key) != key { + return fmt.Errorf("process env key %q must not contain surrounding whitespace", key) + } + for _, r := range key { + if r == '=' || r == 0 || r == ' ' || r == '\t' || r == '\n' || r == '\r' { + return fmt.Errorf("process env key %q contains invalid whitespace or separator", key) + } + } + return nil +} + +func validatePlaceholders(value string) error { + _, err := expandPlaceholders(value, RunProcessContext{}) + return err +} + +func expandPlaceholders(value string, ctx RunProcessContext) (string, error) { + var out strings.Builder + for { + start := strings.Index(value, "{{") + if start == -1 { + out.WriteString(value) + return out.String(), nil + } + out.WriteString(value[:start]) + value = value[start+2:] + + end := strings.Index(value, "}}") + if end == -1 { + return "", fmt.Errorf("unterminated placeholder") + } + name := strings.TrimSpace(value[:end]) + replacement, ok := runPlaceholderValue(name, ctx.Run) + if !ok { + return "", fmt.Errorf("unknown placeholder %q", name) + } + out.WriteString(replacement) + value = value[end+2:] + } +} + +func runPlaceholderValue(name string, run store.Run) (string, bool) { + switch name { + case "run.id": + return run.ID, true + case "run.projectId": + return run.ProjectID, true + case "run.threadId": + return run.ThreadID, true + default: + return "", false + } +}