refactor(edge): 提取 runcontrol 统一 REST/MCP 的 run 创建规则

edge-server/internal/api/handlers.go

@@ -6,7 +6,6 @@ import (
 	"net/http"
 	"strings"
 	"sync"
-	"time"
 
 	"github.com/gorilla/websocket"
 
@@ -80,7 +79,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
@@ -102,9 +100,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.
@@ -125,7 +120,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
 	}
@@ -144,16 +138,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()

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)
 // ---------------------------------------------------------------------------

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)))

edge-server/internal/api/handlers_test.go

@@ -23,6 +23,7 @@ import (
 	"github.com/agenthub/edge-server/internal/jwtutil"
 	"github.com/agenthub/edge-server/internal/lifecycle"
 	"github.com/agenthub/edge-server/internal/permission"
+	"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"
@@ -1182,8 +1183,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") {
@@ -1468,7 +1469,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")
@@ -1506,7 +1507,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))
 	}
 }