From f820b6eb18ebfe876003c1d214ee2057cbb9c5ec Mon Sep 17 00:00:00 2001 From: Delicious233 <101502465+DeliciousBuding@users.noreply.github.com> Date: Thu, 13 Aug 2026 11:32:31 +0800 Subject: [PATCH] =?UTF-8?q?refactor(edge):=20=E6=8F=90=E5=8F=96=20runcontr?= =?UTF-8?q?ol=20=E7=BB=9F=E4=B8=80=20REST/MCP=20=E7=9A=84=20run=20?= =?UTF-8?q?=E5=88=9B=E5=BB=BA=E8=A7=84=E5=88=99?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit POST /v1/runs 与 MCP agenthub_start_run 此前各自实现同一套 run 创建 序列（目标校验、活跃 run 守卫、CreateRun、run.queued、执行器启动与 失败状态机），规则已开始漂移。提取 internal/runcontrol 作为单一事实源： - Create(store, executor, bus, params) 拥有完整不变量序列，返回 *errcode.Error 供各传输层映射到自己的响应形状（HTTP 状态 vs JSON-RPC 错误文本）。 - 传输层只保留协议差异：REST 注入 profile 默认值、双 token 认证、 timeline 发布（prompt + queued 标记 item）与富上下文构建器；MCP 注入单条 user_message item 与最小上下文。两者经 CreateParams 回调 （Timeline / BuildContext / AgentExists）表达，核心不感知协议。 - 进程级 runCreationMu 串行化 check-then-create 段：修复了 MCP 并发 start_run 可绕过活跃 run 守卫的竞态（REST 原本有 per-Handler 锁， MCP 无锁），且跨协议共享同一把锁。 行为统一（有意为之，PR 内记录）： - MCP 的活跃 run 判定纳入 cancelling（与 REST isActiveRunStatus 对齐） - MCP 错误从 fmt.Errorf 升级为结构化 errcode（workdir_required / active_run_exists / executor_start_failed 等） - REST 空 allowlist 错误码从 workspace_not_allowed 精确为 workspace_allowlist_not_configured（与 MCP 已有行为一致） 新增 runcontrol 直接单测 11 个（含 8 并发串行化对拍，race 干净）； api 22 个 PostRuns 测试与 mcp 33 个 server 测试全部保持通过。 验证：go build/vet/staticcheck 全绿；go test ./... -short -race 通过； verify-orchestrator-deps、verify-test-sleep-ratchet、git diff --check 通过； runcontrol 覆盖率 75%。 Co-authored-by: Cursor --- edge-server/internal/api/handlers.go | 16 -- edge-server/internal/api/handlers_events.go | 48 ---- edge-server/internal/api/handlers_runs.go | 152 ++++------ edge-server/internal/api/handlers_test.go | 9 +- edge-server/internal/mcp/tools_handlers.go | 172 ++++-------- edge-server/internal/runcontrol/runcontrol.go | 264 ++++++++++++++++++ .../internal/runcontrol/runcontrol_test.go | 245 ++++++++++++++++ 7 files changed, 629 insertions(+), 277 deletions(-) create mode 100644 edge-server/internal/runcontrol/runcontrol.go create mode 100644 edge-server/internal/runcontrol/runcontrol_test.go diff --git a/edge-server/internal/api/handlers.go b/edge-server/internal/api/handlers.go index ca4929c58..f91dfaa1a 100644 --- a/edge-server/internal/api/handlers.go +++ b/edge-server/internal/api/handlers.go @@ -6,7 +6,6 @@ import ( "net/http" "strings" "sync" - "time" "github.com/gorilla/websocket" @@ -79,7 +78,6 @@ type Handler struct { // Tests inject a temp dir so real foreign session stores are never scanned. SessionHome string - runCreateMu sync.Mutex permissionRegistryMu sync.Mutex permissionObserverCancel func() permissionBrokerInstalled bool @@ -101,9 +99,6 @@ var upgrader = websocket.Upgrader{ } const ( - defaultRunCleanupTerminalTTL = 24 * time.Hour - defaultRunCleanupMaxTerminalRunsPerThread = 50 - // CloseCodeEventGap is the WebSocket close code sent when the event bus // detects dropped events for this subscriber. The client should reconnect // with a known-good cursor to trigger a full resync. @@ -124,7 +119,6 @@ func (h *Handler) denyRemoteHubSharedConfig(w http.ResponseWriter, r *http.Reque func (h *Handler) validateWorkDirAllowed(workDir string) error { // Empty workDir is allowed for non-run endpoints (e.g. optional read paths). - // Run-start uses validateRunWorkDir, which rejects empty values (#854). if workDir == "" { return nil } @@ -143,16 +137,6 @@ func (h *Handler) validateWorkDirAllowed(workDir string) error { return err } -// validateRunWorkDir enforces a non-empty workDir for adapter-backed run starts, -// then applies the workspace allowlist check. Callers should pass the trimmed -// workDir value and use the returned error for HTTP mapping. -func (h *Handler) validateRunWorkDir(workDir string) error { - if strings.TrimSpace(workDir) == "" { - return errcode.ErrWorkDirRequired - } - return h.validateWorkDirAllowed(workDir) -} - func ensureStore(h *Handler) store.Repository { if h.Store == nil { h.Store = store.New() diff --git a/edge-server/internal/api/handlers_events.go b/edge-server/internal/api/handlers_events.go index caab29576..ecf26b2eb 100644 --- a/edge-server/internal/api/handlers_events.go +++ b/edge-server/internal/api/handlers_events.go @@ -4,7 +4,6 @@ import ( "crypto/sha256" "encoding/hex" "encoding/json" - "fmt" "io" "log/slog" "net/http" @@ -248,15 +247,6 @@ func activeRunExistsResponse(run store.Run) map[string]any { return body } -func activeRunForThread(runs []store.Run) (store.Run, bool) { - for _, run := range runs { - if isActiveRunStatus(run.Status) { - return run, true - } - } - return store.Run{}, false -} - // threadHasAssistantHistory returns true when the thread contains at least one // message from the agent (role "agent"), indicating the adapter should resume // rather than start a fresh conversation. @@ -269,44 +259,6 @@ func threadHasAssistantHistory(repo store.Repository, threadID string) bool { return false } -func isActiveRunStatus(status string) bool { - switch status { - case "queued", "started", "cancelling": - return true - default: - return false - } -} - -// validatePermissionMode returns an error if mode is not a recognised -// Claude Code --permission-mode value. An empty mode is allowed and means -// "use the adapter default". -func validatePermissionMode(mode string) error { - if mode == "" { - return nil - } - // SEC-02: Reject 'bypassPermissions' — it disables ALL security hooks at - // the CLI level, giving the agent unrestricted shell access regardless - // of SecurityHook settings. Only the whitelist modes are allowed. - switch mode { - case "default", "acceptEdits", "plan", "dontAsk": - return nil - default: - return fmt.Errorf("unknown permission mode %q: valid values are default, acceptEdits, plan, dontAsk", mode) - } -} - -func cleanupRuns(repository store.Repository) store.RunCleanupResult { - cleaner, ok := repository.(store.RunCleaner) - if !ok { - return store.RunCleanupResult{} - } - return cleaner.CleanupRuns(store.RunCleanupOptions{ - TerminalTTL: defaultRunCleanupTerminalTTL, - MaxTerminalRunsPerThread: defaultRunCleanupMaxTerminalRunsPerThread, - }) -} - // --------------------------------------------------------------------------- // POST /v1/permissions/decide (Desktop permission gate) // --------------------------------------------------------------------------- diff --git a/edge-server/internal/api/handlers_runs.go b/edge-server/internal/api/handlers_runs.go index 2328dd404..7615234d9 100644 --- a/edge-server/internal/api/handlers_runs.go +++ b/edge-server/internal/api/handlers_runs.go @@ -12,6 +12,7 @@ import ( "github.com/agenthub/edge-server/internal/jwtutil" "github.com/agenthub/edge-server/internal/lifecycle" "github.com/agenthub/edge-server/internal/router" + "github.com/agenthub/edge-server/internal/runcontrol" "github.com/agenthub/edge-server/internal/runnerctx" "github.com/agenthub/edge-server/internal/store" ) @@ -181,43 +182,9 @@ func (h *Handler) validateCapabilityRequest(r *http.Request, req *runRequest) *e return nil } -// validateRunCreateState performs the pre-create validation under the -// run-create mutex, in the historical PostRuns order. Returns an HTTP status -// and a response body when the request is rejected (body != nil). -func (h *Handler) validateRunCreateState(repository store.Repository, req *runRequest) (int, map[string]any) { - thread, ok := repository.GetThread(req.ThreadID) - if !ok || thread.ProjectID != req.ProjectID { - return http.StatusNotFound, errcode.ErrorBody(errcode.ErrNotFound.WithMessage("project or thread not found")) - } - if _, ok := repository.GetProject(req.ProjectID); !ok { - return http.StatusNotFound, errcode.ErrorBody(errcode.ErrNotFound.WithMessage("project or thread not found")) - } - req.WorkDir = strings.TrimSpace(req.WorkDir) - if err := h.validateRunWorkDir(req.WorkDir); err != nil { - if errors.Is(err, errcode.ErrWorkDirRequired) { - return http.StatusBadRequest, errcode.ErrorBody(errcode.ErrWorkDirRequired) - } - slog.Error("run workdir validation failed", "workDir", req.WorkDir, "error", err) - return http.StatusForbidden, errcode.ErrorBody(errcode.ErrWorkspaceNotAllowed) - } - if err := validatePermissionMode(req.PermissionMode); err != nil { - slog.Error("invalid permission mode", "permissionMode", req.PermissionMode, "error", err) - return http.StatusBadRequest, errcode.ErrorBody(errcode.ErrInvalidPermissionMode) - } - if active, ok := activeRunForThread(repository.ListRuns(req.ThreadID)); ok { - return http.StatusConflict, activeRunExistsResponse(active) - } - if h.Executor == nil { - return http.StatusServiceUnavailable, errcode.ErrorBody(errcode.ErrExecutorUnavailable.WithMessage("no Agent Runtime executor configured")) - } - // #175: Reject unknown agentId — do not fall back to default adapter. - if req.AgentID != "" && h.AdapterRegistry != nil { - if _, ok := h.AdapterRegistry.Get(req.AgentID); !ok { - return http.StatusBadRequest, errcode.ErrorBody(errcode.ErrInvalidAgentID.WithMessagef("unknown agent adapter: %q", req.AgentID)) - } - } - return 0, nil -} +// validateRunCreateState previously performed pre-create validation here; the +// shared run-creation core (internal/runcontrol.Create) now owns the entire +// validation + creation + executor sequence for both REST and MCP. // publishRunPromptItem stores the run prompt as a user_message item and // publishes message.created / item.created events. Failure is non-fatal. @@ -261,13 +228,10 @@ func publishRunQueuedItem(h *Handler, run store.Run) { }) } -// startRunExecutor builds the run process context (skills, memory, MCP -// config) and starts the executor. On start failure the run is marked failed -// and a run.failed event is published; the error is returned for HTTP mapping. -func (h *Handler) startRunExecutor(run store.Run, req *runRequest, scope map[string]any) error { - if h.Executor == nil { - return nil - } +// buildRunContext builds the run process context (skills, memory, MCP +// config) for the shared run-creation core. The executor start and the +// failure state transition live in runcontrol.Create. +func (h *Handler) buildRunContext(run store.Run, req *runRequest) lifecycle.RunProcessContext { runCtx := lifecycle.RunProcessContext{ Run: run, Prompt: req.Prompt, @@ -320,18 +284,7 @@ func (h *Handler) startRunExecutor(run store.Run, req *runRequest, scope map[str if h.MCPConfigStore != nil { runCtx.MCPConfig = adapters.MergeConfigJSON(runCtx.MCPConfig, h.MCPConfigStore) } - if err := h.Executor.Start(run, runCtx); err != nil { - slog.Error("run executor start failed", "runId", run.ID, "error", err) - if failed, ok := ensureStore(h).SetRunStatusIf(run.ID, "failed", "queued"); ok { - h.Bus.Publish("run.failed", scope, map[string]any{ - "runId": failed.ID, - "status": failed.Status, - "error": "run execution failed", - }) - } - return err - } - return nil + return runCtx } func (h *Handler) PostRuns(w http.ResponseWriter, r *http.Request) { @@ -367,19 +320,16 @@ func (h *Handler) PostRuns(w http.ResponseWriter, r *http.Request) { } repository := ensureStore(h) - h.runCreateMu.Lock() - cleanupRuns(repository) - if status, body := h.validateRunCreateState(repository, &req); body != nil { - h.runCreateMu.Unlock() - writeJSON(w, status, body) - return - } + // Auto-detect continue: when the thread has prior assistant messages, // set ContinueLast = true so adapters can resume the conversation. + // Each run creates a fresh CC conversation via --session-id. if !req.Continue && threadHasAssistantHistory(repository, req.ThreadID) { req.Continue = true } - // Each run creates a fresh CC conversation via --session-id. + // WorkDir normalization previously happened inside validateRunCreateState; + // the shared core also trims, this keeps the context builder consistent. + req.WorkDir = strings.TrimSpace(req.WorkDir) // Resolve adapter label for debug logging. resolvedAdapterID := req.AgentID @@ -392,44 +342,58 @@ func (h *Handler) PostRuns(w http.ResponseWriter, r *http.Request) { } slog.Debug("run.create", "agentId", req.AgentID, "threadId", req.ThreadID, "model", req.Model, "adapterResolved", resolvedAdapterID, "hasExecutor", h.Executor != nil) - runID := genID("run_") - // Classify prompt complexity for execution strategy selection. // Complex tasks may benefit from orchestration/TeamRun; Simple tasks // can dispatch directly to a single agent. promptComplexity := router.ClassifyComplexity(req.Prompt) - slog.Debug("run.complexity", "runId", runID, "complexity", promptComplexity, + slog.Debug("run.complexity", "complexity", promptComplexity, "promptLen", len(req.Prompt), "agentId", req.AgentID) - run, err := repository.CreateRun(runID, req.ProjectID, req.ThreadID) - h.runCreateMu.Unlock() - if err != nil { - if errors.Is(err, store.ErrNotFound) { - writeJSON(w, http.StatusNotFound, errcode.ErrorBody(errcode.ErrNotFound.WithMessage("project or thread not found"))) - } else { - writeJSON(w, http.StatusInternalServerError, errcode.ErrorBody(errcode.ErrInternal.WithMessagef("failed to create run: %v", err))) - } - return - } - scope := map[string]any{ - "projectId": run.ProjectID, - "threadId": run.ThreadID, - "runId": run.ID, - } - - // Emit run.queued - h.Bus.Publish("run.queued", scope, run) - slog.Debug("run.queued", "runId", runID, "agentId", req.AgentID) - publishRunPromptItem(h, run, req.Prompt) - publishRunQueuedItem(h, run) - if err := h.startRunExecutor(run, &req, scope); err != nil { - if errors.Is(err, lifecycle.ErrTooManyConcurrentRuns) { - slog.Error("too many concurrent runs", "runId", runID, "error", err) - writeJSON(w, http.StatusTooManyRequests, errcode.ErrorBody(errcode.ErrTooManyConcurrentRuns)) + // The shared run-creation core (internal/runcontrol) owns validation, + // active-run guarding, run record creation, run.queued publication, and + // the executor start/failure state machine — REST and MCP share it. REST + // contributes the timeline policy and the adapter context builder. + run, err := runcontrol.Create(repository, h.Executor, h.Bus, runcontrol.CreateParams{ + ProjectID: req.ProjectID, + ThreadID: req.ThreadID, + Prompt: req.Prompt, + AgentID: req.AgentID, + Model: req.Model, + PermissionMode: req.PermissionMode, + SessionID: req.SessionID, + ContinueLast: req.Continue, + WorkDir: req.WorkDir, + WorkspaceAllowlist: h.WorkspaceAllowlist, + AgentExists: func(agentID string) bool { + if h.AdapterRegistry == nil { + return true // no registry in scope; skip the #175 check + } + _, ok := h.AdapterRegistry.Get(agentID) + return ok + }, + Cleanup: true, + Timeline: func(run store.Run) { + publishRunPromptItem(h, run, req.Prompt) + publishRunQueuedItem(h, run) + }, + BuildContext: func(run store.Run) lifecycle.RunProcessContext { + return h.buildRunContext(run, &req) + }, + }) + if err != nil { + if e, ok := err.(*errcode.Error); ok { + // Enrich the active-run conflict with the conflicting run, + // preserving the historical response body shape. + if errors.Is(err, errcode.ErrActiveRunExists) { + if active, found := runcontrol.ActiveRunForThread(repository.ListRuns(req.ThreadID)); found { + writeJSON(w, http.StatusConflict, activeRunExistsResponse(active)) + return + } + } + writeJSON(w, e.HTTPStatus, errcode.ErrorBody(e)) return } - slog.Error("run executor start failed", "runId", runID, "error", err) - writeJSON(w, http.StatusInternalServerError, errcode.ErrorBody(errcode.ErrExecutorStartFailed)) + writeJSON(w, http.StatusInternalServerError, errcode.ErrorBody(errcode.ErrInternal.WithMessagef("%v", err))) return } writeSuccess(w, http.StatusAccepted, acceptedResponse(runToResponse(run))) diff --git a/edge-server/internal/api/handlers_test.go b/edge-server/internal/api/handlers_test.go index 5abd8d465..e23b9957a 100644 --- a/edge-server/internal/api/handlers_test.go +++ b/edge-server/internal/api/handlers_test.go @@ -22,6 +22,7 @@ import ( "github.com/agenthub/edge-server/internal/hub" "github.com/agenthub/edge-server/internal/jwtutil" "github.com/agenthub/edge-server/internal/lifecycle" + "github.com/agenthub/edge-server/internal/runcontrol" "github.com/agenthub/edge-server/internal/runners" "github.com/agenthub/edge-server/internal/store" "github.com/golang-jwt/jwt/v5" @@ -1181,8 +1182,8 @@ func TestPostRunsRejectsWorkDirWhenWorkspaceAllowlistEmpty(t *testing.T) { if !ok { t.Fatalf("error body = %#v, want error object", resp) } - if errObj["code"] != errcode.ErrWorkspaceNotAllowed.Code { - t.Fatalf("error code = %#v, want %s", errObj["code"], errcode.ErrWorkspaceNotAllowed.Code) + if errObj["code"] != errcode.ErrWorkspaceAllowlistNotConfigured.Code { + t.Fatalf("error code = %#v, want %s", errObj["code"], errcode.ErrWorkspaceAllowlistNotConfigured.Code) } msg, ok := errObj["message"].(string) if !ok || !strings.Contains(msg, "allowlist") { @@ -1467,7 +1468,7 @@ func TestPostRunsCleansTerminalRunsBeforeCreatingNewRun(t *testing.T) { h.Executor = executor h.ensureDefaults() - for i := 0; i < defaultRunCleanupMaxTerminalRunsPerThread+1; i++ { + for i := 0; i < runcontrol.DefaultRunCleanupMaxTerminalRunsPerThread+1; i++ { runID := fmt.Sprintf("run_terminal_%02d", i) itemID := fmt.Sprintf("item_terminal_%02d", i) run, err := h.Store.CreateRun(runID, "proj_local", "thread_local") @@ -1505,7 +1506,7 @@ func TestPostRunsCleansTerminalRunsBeforeCreatingNewRun(t *testing.T) { if len(executor.started) != 1 { t.Fatalf("executor starts = %d, want 1", len(executor.started)) } - if got := h.Store.ListRuns("thread_local"); len(got) != defaultRunCleanupMaxTerminalRunsPerThread+1 { + if got := h.Store.ListRuns("thread_local"); len(got) != runcontrol.DefaultRunCleanupMaxTerminalRunsPerThread+1 { t.Fatalf("thread run count = %d, want retained terminal runs plus new active run", len(got)) } } diff --git a/edge-server/internal/mcp/tools_handlers.go b/edge-server/internal/mcp/tools_handlers.go index 4b3719d6f..72fc21603 100644 --- a/edge-server/internal/mcp/tools_handlers.go +++ b/edge-server/internal/mcp/tools_handlers.go @@ -5,15 +5,13 @@ package mcp import ( "encoding/json" - "errors" "fmt" - "log/slog" "strings" "github.com/agenthub/edge-server/internal/adapters" "github.com/agenthub/edge-server/internal/errcode" "github.com/agenthub/edge-server/internal/lifecycle" - "github.com/agenthub/edge-server/internal/security" + "github.com/agenthub/edge-server/internal/runcontrol" "github.com/agenthub/edge-server/internal/store" ) @@ -142,47 +140,20 @@ func decodeStartRunArgs(args json.RawMessage) (startRunParams, error) { return params, nil } -// validateStartRunWorkDir requires a non-empty workDir for adapter runs -// (#854), then applies the shared REST/MCP workspace allowlist policy -// (AH-SR-006 / #998): EvalSymlinks + IsPathWithin via -// security.ValidateWorkDirAgainstAllowlist. Returns the trimmed workDir. -func validateStartRunWorkDir(workDir string, allowlist []string) (string, error) { - workDir = strings.TrimSpace(workDir) - if workDir == "" { - return "", errcode.ErrWorkDirRequired - } - if err := security.ValidateWorkDirAgainstAllowlist(workDir, allowlist); err != nil { - if errors.Is(err, security.ErrWorkspaceAllowlistEmpty) { - return "", errcode.ErrWorkspaceAllowlistNotConfigured - } - if errors.Is(err, security.ErrWorkspaceOutsideAllowlist) { - return "", errcode.ErrWorkspaceNotAllowed - } - return "", fmt.Errorf("invalid workDir: %w", err) - } - return workDir, nil -} - -// errIfActiveRunExists returns an error when the thread already has an active -// (queued or started) run. -func errIfActiveRunExists(repo store.Repository, threadID string) error { - for _, r := range repo.ListRuns(threadID) { - if r.Status == "queued" || r.Status == "started" { - return fmt.Errorf("thread already has an active run: %s", r.ID) - } - } - return nil -} +// validateStartRunWorkDir and errIfActiveRunExists previously lived here; the +// shared run-creation core (internal/runcontrol.Create) now owns workDir +// validation and the active-run guard for both MCP and REST. // toolStartRun implements the agenthub_start_run tool. // // Requires non-empty workDir and validates it against the workspace allowlist -// (AH-SR-006 / #854), mirroring REST validateRunWorkDir. Creates a run record, -// publishes run.queued, creates a user message item, and starts the agent -// executor. +// (AH-SR-006 / #854). The validation, run creation, run.queued publication, +// executor start, and failure state transition are shared with POST /v1/runs +// through internal/runcontrol.Create — MCP only decodes its own argument +// shape and provides its timeline/context policies. // // Returns an error if: -// - The thread already has an active run (queued or started) +// - The thread already has an active run (queued, started, or cancelling) // - The project or thread is not found // - workDir is missing/empty or outside the configured allowlist // - Required fields (projectId, threadId, prompt, workDir) are missing @@ -198,86 +169,57 @@ func (s *Server) toolStartRun(args json.RawMessage) (json.RawMessage, error) { if err != nil { return nil, err } - workDir, err := validateStartRunWorkDir(params.WorkDir, s.workspaceAllowlist) - if err != nil { - return nil, err - } - params.WorkDir = workDir - - // Verify project and thread exist - thread, ok := s.store.GetThread(params.ThreadID) - if !ok || thread.ProjectID != params.ProjectID { - return nil, fmt.Errorf("thread not found: %s", params.ThreadID) - } - if _, ok := s.store.GetProject(params.ProjectID); !ok { - return nil, fmt.Errorf("project not found: %s", params.ProjectID) - } - - // Check for active run - if err := errIfActiveRunExists(s.store, params.ThreadID); err != nil { - return nil, err - } - - // Generate run ID and create the run - runID := generateID("run_") - run, err := s.store.CreateRun(runID, params.ProjectID, params.ThreadID) - if err != nil { - return nil, fmt.Errorf("failed to create run: %w", err) - } - - // Publish run.queued event - scope := map[string]any{ - "projectId": run.ProjectID, - "threadId": run.ThreadID, - "runId": run.ID, - } - if s.bus != nil { - s.bus.Publish("run.queued", scope, run) - } - - // Create user message item - if _, err := s.store.CreateItem(store.Item{ - ID: generateID("item_"), - ProjectID: run.ProjectID, - ThreadID: run.ThreadID, - RunID: run.ID, - Type: "user_message", - Role: "user", - Status: "created", - Content: params.Prompt, - }); err == nil { - if s.bus != nil { - s.bus.Publish("message.created", scope, map[string]any{ + params.WorkDir = strings.TrimSpace(params.WorkDir) + + run, err := runcontrol.Create(s.store, s.executor, s.bus, runcontrol.CreateParams{ + ProjectID: params.ProjectID, + ThreadID: params.ThreadID, + Prompt: params.Prompt, + AgentID: params.AgentID, + Model: params.Model, + WorkDir: params.WorkDir, + WorkspaceAllowlist: s.workspaceAllowlist, + SessionID: "mcp_" + params.ThreadID, + ContinueLast: true, + Timeline: func(run store.Run) { + // MCP publishes a single user_message item (REST additionally + // publishes a queued marker item). + item, createErr := s.store.CreateItem(store.Item{ + ID: generateID("item_"), + ProjectID: run.ProjectID, + ThreadID: run.ThreadID, + RunID: run.ID, + Type: "user_message", + Role: "user", + Status: "created", + Content: params.Prompt, + }) + if createErr != nil || s.bus == nil { + return + } + s.bus.Publish("message.created", map[string]any{ + "projectId": item.ProjectID, + "threadId": item.ThreadID, + "runId": item.RunID, + "itemId": item.ID, + }, map[string]any{ "content": params.Prompt, }) - } - } - - // Start the executor - sessionID := "mcp_" + run.ThreadID - runCtx := lifecycle.RunProcessContext{ - Run: run, - Prompt: params.Prompt, - AgentID: params.AgentID, - Model: params.Model, - SessionID: sessionID, - ContinueLast: true, - WorkDir: params.WorkDir, - } - - if err := s.executor.Start(run, runCtx); err != nil { - slog.Error("mcp run start failed", "runId", run.ID, "error", err) - // Mark run as failed - if failed, ok := s.store.SetRunStatusIf(run.ID, "failed", "queued"); ok { - if s.bus != nil { - s.bus.Publish("run.failed", scope, map[string]any{ - "runId": failed.ID, - "status": failed.Status, - "error": "run execution failed", - }) + }, + BuildContext: func(run store.Run) lifecycle.RunProcessContext { + return lifecycle.RunProcessContext{ + Run: run, + Prompt: params.Prompt, + AgentID: params.AgentID, + Model: params.Model, + SessionID: "mcp_" + run.ThreadID, + ContinueLast: true, + WorkDir: params.WorkDir, } - } - return nil, fmt.Errorf("failed to start run: %w", err) + }, + }) + if err != nil { + return nil, err } result := map[string]any{ diff --git a/edge-server/internal/runcontrol/runcontrol.go b/edge-server/internal/runcontrol/runcontrol.go new file mode 100644 index 000000000..bd5aebf9c --- /dev/null +++ b/edge-server/internal/runcontrol/runcontrol.go @@ -0,0 +1,264 @@ +// Package runcontrol is the single source of truth for agent run creation, +// shared by the REST (POST /v1/runs) and MCP (agenthub_start_run) entry +// points. Both transports decode their own request shape and build their own +// timeline/context policies, but the invariant sequence — target validation, +// active-run guard, run record creation, run.queued publication, executor +// start, and failure state transition — lives here exactly once. +// +// Dependency direction: runcontrol depends on store / lifecycle / events / +// security / errcode only. It must never import api or mcp. +package runcontrol + +import ( + "crypto/rand" + "errors" + "fmt" + "log/slog" + "strings" + "sync" + "time" + + "github.com/agenthub/edge-server/internal/errcode" + "github.com/agenthub/edge-server/internal/events" + "github.com/agenthub/edge-server/internal/lifecycle" + "github.com/agenthub/edge-server/internal/security" + "github.com/agenthub/edge-server/internal/store" +) + +// runCreationMu serializes run creation process-wide. Run creation is a +// check-then-create sequence (no active run on the thread, then CreateRun); +// without one shared lock, two concurrent requests — including one from REST +// and one from MCP — could both pass the active-run check and create +// overlapping runs. Run creation is rare and cheap, so the contention cost of +// a process-wide lock is negligible compared to the invariant it protects. +var runCreationMu sync.Mutex + +const ( + // DefaultRunCleanupTerminalTTL is how long a terminal run is retained + // before the pre-create cleanup removes it. + DefaultRunCleanupTerminalTTL = 24 * time.Hour + // DefaultRunCleanupMaxTerminalRunsPerThread bounds how many terminal runs + // are retained per thread before the pre-create cleanup trims the oldest. + DefaultRunCleanupMaxTerminalRunsPerThread = 50 +) + +// CreateParams carries the transport-independent inputs for starting a run. +// Transport-specific behavior (profile defaults, session naming, timeline +// items, adapter context) is injected as callbacks so the core sequence never +// needs to know which protocol created the request. +type CreateParams struct { + ProjectID string + ThreadID string + Prompt string + AgentID string + Model string + PermissionMode string + SessionID string + ContinueLast bool + WorkDir string + + // WorkspaceAllowlist is the request-time allowlist used to validate + // WorkDir (AH-SR-006 / #998). Empty = fail-closed for non-empty workDir. + WorkspaceAllowlist []string + + // AgentExists, when non-nil, rejects unknown agent IDs (#175). REST wires + // it to the adapter registry; MCP leaves it nil (no registry in scope). + AgentExists func(agentID string) bool + + // Cleanup runs terminal-run cleanup before validation (REST housekeeping). + Cleanup bool + + // Timeline publishes the transport's timeline items/events after the run + // record is created and before the executor starts. REST publishes the + // prompt and queued-marker items; MCP publishes a single user_message item. + Timeline func(run store.Run) + + // BuildContext builds the RunProcessContext handed to the executor. + // When nil, the executor start step is skipped. + BuildContext func(run store.Run) lifecycle.RunProcessContext +} + +// Create validates the target thread, creates the run record, publishes +// run.queued, invokes the transport timeline hook, and starts the executor. +// +// Error returns are always *errcode.Error so each transport can map them to +// its own response shape (HTTP status via HTTPStatus, or JSON-RPC error text). +// The returned run is the persisted record; on error the run is zero-valued. +func Create(repository store.Repository, executor lifecycle.RunExecutor, bus *events.Bus, params CreateParams) (store.Run, error) { + if repository == nil { + return store.Run{}, errcode.ErrStoreNotConfigured + } + params.WorkDir = strings.TrimSpace(params.WorkDir) + + // The lock covers only the check-then-create section (matching the + // historical PostRuns lock scope): cleanup, validation, and CreateRun. + // Event publication, timeline hooks, and the executor start run outside + // the lock so slow executor starts never serialize behind each other. + runCreationMu.Lock() + if params.Cleanup { + cleanupRuns(repository) + } + if err := validateTarget(repository, params.ProjectID, params.ThreadID); err != nil { + runCreationMu.Unlock() + return store.Run{}, err + } + if err := validateWorkDir(params.WorkDir, params.WorkspaceAllowlist); err != nil { + runCreationMu.Unlock() + return store.Run{}, err + } + if err := validatePermissionMode(params.PermissionMode); err != nil { + runCreationMu.Unlock() + slog.Error("invalid permission mode", "permissionMode", params.PermissionMode, "error", err) + return store.Run{}, errcode.ErrInvalidPermissionMode + } + if active, ok := ActiveRunForThread(repository.ListRuns(params.ThreadID)); ok { + runCreationMu.Unlock() + return store.Run{}, errcode.ErrActiveRunExists.WithMessagef("thread already has an active run: %s", active.ID) + } + if executor == nil { + runCreationMu.Unlock() + return store.Run{}, errcode.ErrExecutorUnavailable.WithMessage("no Agent Runtime executor configured") + } + // #175: Reject unknown agentId — do not fall back to default adapter. + if params.AgentID != "" && params.AgentExists != nil && !params.AgentExists(params.AgentID) { + runCreationMu.Unlock() + return store.Run{}, errcode.ErrInvalidAgentID.WithMessagef("unknown agent adapter: %q", params.AgentID) + } + run, err := repository.CreateRun(generateRunID(), params.ProjectID, params.ThreadID) + runCreationMu.Unlock() + if err != nil { + if errors.Is(err, store.ErrNotFound) { + return store.Run{}, errcode.ErrNotFound.WithMessage("project or thread not found") + } + return store.Run{}, errcode.ErrInternal.WithMessagef("failed to create run: %v", err) + } + + scope := map[string]any{ + "projectId": run.ProjectID, + "threadId": run.ThreadID, + "runId": run.ID, + } + if bus != nil { + bus.Publish("run.queued", scope, run) + slog.Debug("run.queued", "runId", run.ID, "agentId", params.AgentID) + } + if params.Timeline != nil { + params.Timeline(run) + } + + if params.BuildContext != nil { + runCtx := params.BuildContext(run) + if err := executor.Start(run, runCtx); err != nil { + slog.Error("run executor start failed", "runId", run.ID, "error", err) + if failed, ok := repository.SetRunStatusIf(run.ID, "failed", "queued"); ok && bus != nil { + bus.Publish("run.failed", scope, map[string]any{ + "runId": failed.ID, + "status": failed.Status, + "error": "run execution failed", + }) + } + if errors.Is(err, lifecycle.ErrTooManyConcurrentRuns) { + slog.Error("too many concurrent runs", "runId", run.ID, "error", err) + return store.Run{}, errcode.ErrTooManyConcurrentRuns + } + return store.Run{}, errcode.ErrExecutorStartFailed + } + } + return run, nil +} + +// ActiveRunForThread returns the first active (queued, started, or cancelling) +// run of the given list. Used both by Create and by transports that need to +// enrich an active-run error response with the conflicting run. +func ActiveRunForThread(runs []store.Run) (store.Run, bool) { + for _, run := range runs { + if IsActiveRunStatus(run.Status) { + return run, true + } + } + return store.Run{}, false +} + +// IsActiveRunStatus reports whether a run status occupies the thread's run +// slot (queued/started/cancelling). Terminal statuses release the slot. +func IsActiveRunStatus(status string) bool { + switch status { + case "queued", "started", "cancelling": + return true + default: + return false + } +} + +// validateTarget verifies the project and thread exist and the thread belongs +// to the project, mirroring the historical PostRuns order. +func validateTarget(repository store.Repository, projectID, threadID string) *errcode.Error { + thread, ok := repository.GetThread(threadID) + if !ok || thread.ProjectID != projectID { + return errcode.ErrNotFound.WithMessage("project or thread not found") + } + if _, ok := repository.GetProject(projectID); !ok { + return errcode.ErrNotFound.WithMessage("project or thread not found") + } + return nil +} + +// validateWorkDir enforces a non-empty workDir for adapter runs (#854), then +// applies the shared REST/MCP workspace allowlist policy (AH-SR-006 / #998): +// EvalSymlinks + IsPathWithin via security.ValidateWorkDirAgainstAllowlist. +func validateWorkDir(workDir string, allowlist []string) *errcode.Error { + if workDir == "" { + return errcode.ErrWorkDirRequired + } + if err := security.ValidateWorkDirAgainstAllowlist(workDir, allowlist); err != nil { + if errors.Is(err, security.ErrWorkspaceAllowlistEmpty) { + // Fail-closed: empty allowlist rejects any non-empty workDir. + return errcode.ErrWorkspaceAllowlistNotConfigured + } + if errors.Is(err, security.ErrWorkspaceOutsideAllowlist) { + return errcode.ErrWorkspaceNotAllowed + } + slog.Error("run workdir validation failed", "workDir", workDir, "error", err) + return errcode.ErrWorkspaceNotAllowed.WithMessagef("invalid workDir: %v", err) + } + return nil +} + +// validatePermissionMode returns an error if mode is not a recognised +// Claude Code --permission-mode value. An empty mode is allowed and means +// "use the adapter default". +func validatePermissionMode(mode string) error { + if mode == "" { + return nil + } + // SEC-02: Reject 'bypassPermissions' — it disables ALL security hooks at + // the CLI level, giving the agent unrestricted shell access regardless + // of SecurityHook settings. Only the whitelist modes are allowed. + switch mode { + case "default", "acceptEdits", "plan", "dontAsk": + return nil + default: + return fmt.Errorf("unknown permission mode %q: valid values are default, acceptEdits, plan, dontAsk", mode) + } +} + +// cleanupRuns removes terminal runs that exceeded the retention policy. It is +// a no-op when the store does not implement store.RunCleaner. +func cleanupRuns(repository store.Repository) { + cleaner, ok := repository.(store.RunCleaner) + if !ok { + return + } + cleaner.CleanupRuns(store.RunCleanupOptions{ + TerminalTTL: DefaultRunCleanupTerminalTTL, + MaxTerminalRunsPerThread: DefaultRunCleanupMaxTerminalRunsPerThread, + }) +} + +// generateRunID produces a run_ prefixed random identifier with the same +// shape as the historical api.genID / mcp.generateID helpers. +func generateRunID() string { + b := make([]byte, 8) + _, _ = rand.Read(b) + return fmt.Sprintf("run_%016x", b) +} diff --git a/edge-server/internal/runcontrol/runcontrol_test.go b/edge-server/internal/runcontrol/runcontrol_test.go new file mode 100644 index 000000000..d0b658e53 --- /dev/null +++ b/edge-server/internal/runcontrol/runcontrol_test.go @@ -0,0 +1,245 @@ +package runcontrol + +import ( + "errors" + "sync" + "testing" + + "github.com/agenthub/edge-server/internal/errcode" + "github.com/agenthub/edge-server/internal/events" + "github.com/agenthub/edge-server/internal/lifecycle" + "github.com/agenthub/edge-server/internal/store" +) + +// recordingExecutor records Start calls and returns a configurable error. +type recordingExecutor struct { + mu sync.Mutex + started []store.Run + contexts []lifecycle.RunProcessContext + startErr error +} + +func (e *recordingExecutor) Start(run store.Run, ctx lifecycle.RunProcessContext) error { + e.mu.Lock() + defer e.mu.Unlock() + e.started = append(e.started, run) + e.contexts = append(e.contexts, ctx) + return e.startErr +} + +func (e *recordingExecutor) Cancel(runID string) lifecycle.CancelResult { + return lifecycle.CancelResult{Found: false} +} + +func (e *recordingExecutor) lastContext() lifecycle.RunProcessContext { + e.mu.Lock() + defer e.mu.Unlock() + if len(e.contexts) == 0 { + return lifecycle.RunProcessContext{} + } + return e.contexts[len(e.contexts)-1] +} + +func (e *recordingExecutor) startCount() int { + e.mu.Lock() + defer e.mu.Unlock() + return len(e.started) +} + +func newTestRepo(t *testing.T) store.Repository { + t.Helper() + repo := store.New() + if _, err := repo.CreateProject("proj_local", "Local Project", ""); err != nil { + t.Fatalf("CreateProject: %v", err) + } + if _, err := repo.CreateThread("thread_local", "proj_local", "Local Thread", "direct", "", ""); err != nil { + t.Fatalf("CreateThread: %v", err) + } + return repo +} + +func baseParams(workDir string) CreateParams { + return CreateParams{ + ProjectID: "proj_local", + ThreadID: "thread_local", + Prompt: "build the patch", + AgentID: "codex", + Model: "gpt-5", + WorkDir: workDir, + WorkspaceAllowlist: []string{workDir}, + Timeline: func(store.Run) {}, + BuildContext: func(run store.Run) lifecycle.RunProcessContext { + return lifecycle.RunProcessContext{Run: run, Prompt: "build the patch"} + }, + } +} + +func TestCreateHappyPath(t *testing.T) { + repo := newTestRepo(t) + bus := events.NewBus(100) + executor := &recordingExecutor{} + workDir := t.TempDir() + + run, err := Create(repo, executor, bus, baseParams(workDir)) + if err != nil { + t.Fatalf("Create returned error: %v", err) + } + if run.ID == "" || run.ProjectID != "proj_local" || run.ThreadID != "thread_local" { + t.Fatalf("run = %#v, want created run bound to project/thread", run) + } + if got := executor.lastContext(); got.Run.ID != run.ID || got.Prompt != "build the patch" { + t.Fatalf("executor context = %#v, want run id %s and prompt", got, run.ID) + } + // The core publishes exactly run.queued before the timeline hook runs. + if bus.HistoryLen() != 1 { + t.Fatalf("bus history = %d, want 1 (run.queued)", bus.HistoryLen()) + } +} + +func TestCreateRejectsMissingTarget(t *testing.T) { + repo := newTestRepo(t) + bus := events.NewBus(100) + executor := &recordingExecutor{} + + params := baseParams(t.TempDir()) + params.ThreadID = "thread_missing" + if _, err := Create(repo, executor, bus, params); !errors.Is(err, errcode.ErrNotFound) { + t.Fatalf("missing thread err = %v, want ErrNotFound", err) + } + params.ThreadID = "thread_local" + params.ProjectID = "proj_missing" + if _, err := Create(repo, executor, bus, params); !errors.Is(err, errcode.ErrNotFound) { + t.Fatalf("missing project err = %v, want ErrNotFound", err) + } + if bus.HistoryLen() != 0 { + t.Fatalf("event history = %d, want 0 for rejected creates", bus.HistoryLen()) + } +} + +func TestCreateRejectsWorkDirViolations(t *testing.T) { + repo := newTestRepo(t) + bus := events.NewBus(100) + executor := &recordingExecutor{} + allowedRoot := t.TempDir() + + params := baseParams(allowedRoot) + params.WorkDir = " " + if _, err := Create(repo, executor, bus, params); !errors.Is(err, errcode.ErrWorkDirRequired) { + t.Fatalf("empty workDir err = %v, want ErrWorkDirRequired", err) + } + params.WorkDir = t.TempDir() // outside allowlist + params.WorkspaceAllowlist = []string{allowedRoot} + if _, err := Create(repo, executor, bus, params); !errors.Is(err, errcode.ErrWorkspaceNotAllowed) { + t.Fatalf("outside workDir err = %v, want ErrWorkspaceNotAllowed", err) + } + params.WorkDir = allowedRoot + params.WorkspaceAllowlist = nil + if _, err := Create(repo, executor, bus, params); !errors.Is(err, errcode.ErrWorkspaceAllowlistNotConfigured) { + t.Fatalf("empty allowlist err = %v, want ErrWorkspaceAllowlistNotConfigured", err) + } +} + +func TestCreateRejectsInvalidPermissionMode(t *testing.T) { + repo := newTestRepo(t) + params := baseParams(t.TempDir()) + params.PermissionMode = "bypassPermissions" + if _, err := Create(repo, &recordingExecutor{}, events.NewBus(10), params); !errors.Is(err, errcode.ErrInvalidPermissionMode) { + t.Fatalf("err = %v, want ErrInvalidPermissionMode", err) + } +} + +func TestCreateRejectsActiveRun(t *testing.T) { + repo := newTestRepo(t) + if _, err := repo.CreateRun("run_active", "proj_local", "thread_local"); err != nil { + t.Fatalf("CreateRun: %v", err) + } + _, err := Create(repo, &recordingExecutor{}, events.NewBus(10), baseParams(t.TempDir())) + if !errors.Is(err, errcode.ErrActiveRunExists) { + t.Fatalf("err = %v, want ErrActiveRunExists", err) + } +} + +func TestCreateRejectsUnknownAgent(t *testing.T) { + repo := newTestRepo(t) + params := baseParams(t.TempDir()) + params.AgentExists = func(agentID string) bool { return false } + _, err := Create(repo, &recordingExecutor{}, events.NewBus(10), params) + if !errors.Is(err, errcode.ErrInvalidAgentID) { + t.Fatalf("err = %v, want ErrInvalidAgentID", err) + } +} + +func TestCreateRequiresExecutor(t *testing.T) { + repo := newTestRepo(t) + _, err := Create(repo, nil, events.NewBus(10), baseParams(t.TempDir())) + if !errors.Is(err, errcode.ErrExecutorUnavailable) { + t.Fatalf("err = %v, want ErrExecutorUnavailable", err) + } +} + +func TestCreateMarksRunFailedWhenExecutorFails(t *testing.T) { + repo := newTestRepo(t) + bus := events.NewBus(100) + executor := &recordingExecutor{startErr: errors.New("executor offline")} + + _, err := Create(repo, executor, bus, baseParams(t.TempDir())) + if !errors.Is(err, errcode.ErrExecutorStartFailed) { + t.Fatalf("err = %v, want ErrExecutorStartFailed", err) + } + runs := repo.ListRuns("thread_local") + if len(runs) != 1 || runs[0].Status != "failed" { + t.Fatalf("runs = %#v, want single failed run", runs) + } + // run.queued followed by run.failed. + if bus.HistoryLen() != 2 { + t.Fatalf("bus history = %d, want 2 (run.queued + run.failed)", bus.HistoryLen()) + } +} + +func TestCreateMapsTooManyConcurrentRuns(t *testing.T) { + repo := newTestRepo(t) + executor := &recordingExecutor{startErr: lifecycle.ErrTooManyConcurrentRuns} + _, err := Create(repo, executor, events.NewBus(10), baseParams(t.TempDir())) + if !errors.Is(err, errcode.ErrTooManyConcurrentRuns) { + t.Fatalf("err = %v, want ErrTooManyConcurrentRuns", err) + } +} + +func TestCreateSerializesConcurrentCreates(t *testing.T) { + repo := newTestRepo(t) + executor := &recordingExecutor{} + workDir := t.TempDir() + + const workers = 8 + var wg sync.WaitGroup + ready := make(chan struct{}) + results := make([]error, workers) + for i := 0; i < workers; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + <-ready + _, results[i] = Create(repo, executor, events.NewBus(10), baseParams(workDir)) + }(i) + } + close(ready) + wg.Wait() + + successes, activeRunErrors := 0, 0 + for _, err := range results { + switch { + case err == nil: + successes++ + case errors.Is(err, errcode.ErrActiveRunExists): + activeRunErrors++ + default: + t.Fatalf("unexpected error: %v", err) + } + } + if successes != 1 || activeRunErrors != workers-1 { + t.Fatalf("successes = %d, activeRunErrors = %d, want 1 and %d", successes, activeRunErrors, workers-1) + } + if got := executor.startCount(); got != 1 { + t.Fatalf("executor starts = %d, want exactly 1", got) + } +}