feat: implement environment management features

- Added environment creation and editing pages with forms for name and description.
- Integrated environment listing with options to edit and delete environments.
- Updated agent detail page to manage environments followed by agents.
- Enhanced API to support environment operations including listing, creating, updating, and deleting environments.
- Refactored related components and state management for improved clarity and functionality.

pkg/api/agents.go

@@ -58,7 +58,7 @@ type Agent struct {
 	WebhookInstalledAt *time.Time         `json:"webhookInstalledAt,omitempty"`
 	AuthStatus         storage.AuthStatus `json:"authStatus,omitempty"`
 	AuthCheckedAt      *time.Time         `json:"authCheckedAt,omitempty"`
-	AttachedPipelines  []string           `json:"attachedPipelines,omitempty"`
+	Environments       []string           `json:"environments,omitempty"`
 	Credentials        []PublicCredential `json:"credentials,omitempty"`
 
 	CreatedAt time.Time `json:"createdAt"`
@@ -108,7 +108,7 @@ func (s *Server) publicAgent(a *storage.Agent) Agent {
 		WebhookInstalledAt: a.WebhookInstalledAt,
 		AuthStatus:         a.AuthStatus,
 		AuthCheckedAt:      a.AuthCheckedAt,
-		AttachedPipelines:  a.AttachedPipelines,
+		Environments:       a.Environments,
 		Credentials:        creds,
 		CreatedAt:          a.CreatedAt,
 		UpdatedAt:          a.UpdatedAt,
@@ -342,27 +342,31 @@ func (s *Server) handleUninstallWebhook(w http.ResponseWriter, r *http.Request)
 	writeJSON(w, http.StatusOK, s.publicAgent(a))
 }
 
-// --- pipeline attachments ------------------------------------------------
+// --- environment subscriptions -------------------------------------------
 
-func (s *Server) handleAttachPipeline(w http.ResponseWriter, r *http.Request) {
+func (s *Server) handleAgentFollowEnv(w http.ResponseWriter, r *http.Request) {
 	id := r.PathValue("id")
-	pipelineID := r.PathValue("pipelineId")
+	envName := r.PathValue("envName")
 	a, err := s.agents.Get(r.Context(), id)
 	if err != nil {
 		writeStorageErr(w, err, "agent not found")
 		return
 	}
-	if _, err := s.pipelines.Get(r.Context(), pipelineID); err != nil {
-		writeStorageErr(w, err, "pipeline not found")
+	if s.environments == nil {
+		writeError(w, http.StatusServiceUnavailable, errors.New("environments store not configured"))
 		return
 	}
-	for _, existing := range a.AttachedPipelines {
-		if existing == pipelineID {
-			writeJSON(w, http.StatusOK, s.publicAgent(a)) // already attached
+	if _, err := s.environments.GetByName(r.Context(), envName); err != nil {
+		writeStorageErr(w, err, "environment not found")
+		return
+	}
+	for _, e := range a.Environments {
+		if strings.EqualFold(e, envName) {
+			writeJSON(w, http.StatusOK, s.publicAgent(a)) // already following
 			return
 		}
 	}
-	a.AttachedPipelines = append(a.AttachedPipelines, pipelineID)
+	a.Environments = append(a.Environments, envName)
 	a.UpdatedAt = time.Now().UTC()
 	if err := s.agents.Update(r.Context(), a); err != nil {
 		writeError(w, http.StatusInternalServerError, err)
@@ -371,21 +375,21 @@ func (s *Server) handleAttachPipeline(w http.ResponseWriter, r *http.Request) {
 	writeJSON(w, http.StatusOK, s.publicAgent(a))
 }
 
-func (s *Server) handleDetachPipeline(w http.ResponseWriter, r *http.Request) {
+func (s *Server) handleAgentUnfollowEnv(w http.ResponseWriter, r *http.Request) {
 	id := r.PathValue("id")
-	pipelineID := r.PathValue("pipelineId")
+	envName := r.PathValue("envName")
 	a, err := s.agents.Get(r.Context(), id)
 	if err != nil {
 		writeStorageErr(w, err, "agent not found")
 		return
 	}
-	out := a.AttachedPipelines[:0]
-	for _, p := range a.AttachedPipelines {
-		if p != pipelineID {
-			out = append(out, p)
+	out := a.Environments[:0]
+	for _, e := range a.Environments {
+		if !strings.EqualFold(e, envName) {
+			out = append(out, e)
 		}
 	}
-	a.AttachedPipelines = out
+	a.Environments = out
 	a.UpdatedAt = time.Now().UTC()
 	if err := s.agents.Update(r.Context(), a); err != nil {
 		writeError(w, http.StatusInternalServerError, err)
@@ -396,8 +400,8 @@ func (s *Server) handleDetachPipeline(w http.ResponseWriter, r *http.Request) {
 
 // --- manual trigger ------------------------------------------------------
 
-// handleTriggerAgent dispatches a run on each attached pipeline. Trigger
-// data describes who/what triggered the run (manual / webhook / etc.).
+// handleTriggerAgent dispatches runs across the agent's followed
+// environments. Trigger data describes who/what triggered the run.
 func (s *Server) handleTriggerAgent(w http.ResponseWriter, r *http.Request) {
 	if s.orch == nil {
 		writeError(w, http.StatusServiceUnavailable, errors.New("orchestrator not configured"))
@@ -409,8 +413,8 @@ func (s *Server) handleTriggerAgent(w http.ResponseWriter, r *http.Request) {
 		writeStorageErr(w, err, "agent not found")
 		return
 	}
-	if len(a.AttachedPipelines) == 0 {
-		writeError(w, http.StatusBadRequest, errors.New("agent has no attached pipelines"))
+	if len(a.Environments) == 0 {
+		writeError(w, http.StatusBadRequest, errors.New("agent follows no environments"))
 		return
 	}
 	triggerData := []map[string]any{{
@@ -437,140 +441,149 @@ func (s *Server) handleTriggerAgent(w http.ResponseWriter, r *http.Request) {
 	})
 }
 
-// dispatchAgent runs each attached pipeline asynchronously. Returns the
-// execution IDs collected and a per-pipeline failure list so the caller
-// can surface skip reasons to the user instead of silently returning [].
+// dispatchAgent fans out across the agent's followed environments. For
+// each env it iterates the env's pipelines, applies the per-pipeline
+// branch filter (for push triggers), stamps the env name + agent info
+// into the trigger payload, and submits the run. Returns the execution
+// IDs and a per-(env,pipeline) failure list so callers can surface skips.
 func (s *Server) dispatchAgent(ctx context.Context, a *storage.Agent, trigger any) ([]string, []DispatchFailure, error) {
-	triggerJSON, _ := json.Marshal(trigger)
-	var triggerItems []models.Item
-	if items, ok := trigger.([]map[string]any); ok {
-		for _, m := range items {
-			triggerItems = append(triggerItems, models.Item(m))
-		}
-	}
-
-	// Per-pipeline branch filter — only applies when the trigger came from a
-	// git push event. Manual triggers fan out to every attached pipeline so
-	// you can still kick a run from the UI without first hand-editing every
-	// Trigger node. The filter compares the pushed branch against the
-	// pipeline's Trigger node `fromBranch` — wildcard ("*", empty) on either
-	// side disables the filter for that pipeline.
+	// pushedBranch is set only for github_push triggers; it gates which
+	// pipelines run (a pipeline's Trigger.fromBranch must match, or be a
+	// wildcard). Manual triggers run every pipeline in every followed env.
 	pushedBranch := ""
 	isPush := false
+	source := ""
 	if items, ok := trigger.([]map[string]any); ok && len(items) > 0 {
-		if src, _ := items[0]["source"].(string); src == "github_push" {
-			isPush = true
-			if b, _ := items[0]["branch"].(string); b != "" {
-				pushedBranch = b
-			} else if b, _ := items[0]["ref"].(string); b != "" {
-				pushedBranch = b
+		if src, _ := items[0]["source"].(string); src != "" {
+			source = src
+			if src == "github_push" {
+				isPush = true
+				if b, _ := items[0]["branch"].(string); b != "" {
+					pushedBranch = b
+				} else if b, _ := items[0]["ref"].(string); b != "" {
+					pushedBranch = b
+				}
 			}
 		}
 	}
 
-	out := make([]string, 0, len(a.AttachedPipelines))
+	var out []string
 	var failures []DispatchFailure
-	for _, pid := range a.AttachedPipelines {
-		p, err := s.pipelines.Get(ctx, pid)
+
+	for _, envName := range a.Environments {
+		env, err := s.environments.GetByName(ctx, envName)
 		if err != nil {
-			slog.WarnContext(ctx, "agent_dispatch_pipeline_missing",
-				slog.String("agent_id", a.ID),
-				slog.String("pipeline_id", pid),
-				slog.Any("error", err),
-			)
+			slog.WarnContext(ctx, "agent_dispatch_env_missing",
+				slog.String("agent_id", a.ID), slog.String("env", envName), slog.Any("error", err))
 			failures = append(failures, DispatchFailure{
-				PipelineID: pid, Reason: "pipeline not found", Error: err.Error(),
+				Environment: envName, Reason: "environment not found", Error: err.Error(),
 			})
 			continue
 		}
-		wf, err := engine.ParseWorkflow(p.Definition)
-		if err != nil {
-			slog.WarnContext(ctx, "agent_dispatch_parse_failed",
-				slog.String("pipeline_id", pid),
-				slog.Any("error", err),
-			)
-			failures = append(failures, DispatchFailure{
-				PipelineID: pid, Reason: "parse failed", Error: err.Error(),
-			})
-			continue
-		}
-		if isPush {
-			triggerBranch := pipelineTriggerBranch(wf)
-			if triggerBranch != "" && triggerBranch != "*" && triggerBranch != pushedBranch {
-				slog.InfoContext(ctx, "agent_dispatch_branch_filtered",
-					slog.String("pipeline_id", pid),
-					slog.String("pushed", pushedBranch),
-					slog.String("trigger_branch", triggerBranch),
-				)
+		for _, pid := range env.PipelineIDs {
+			p, err := s.pipelines.Get(ctx, pid)
+			if err != nil {
 				failures = append(failures, DispatchFailure{
-					PipelineID: pid,
-					Reason:     "branch filtered",
-					Error:      fmt.Sprintf("trigger fromBranch=%q ≠ pushed=%q", triggerBranch, pushedBranch),
+					Environment: envName, PipelineID: pid, Reason: "pipeline not found", Error: err.Error(),
 				})
 				continue
 			}
-		}
-		runCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
-		execID, err := s.orch.RunAsync(runCtx, &orchestrator.RunRequest{
-			RequestMeta: orchestrator.RequestMeta{WorkflowID: pid},
-			Workflow:    wf,
-			TriggerData: triggerItems,
-		})
-		cancel()
-		if err != nil {
-			slog.WarnContext(ctx, "agent_dispatch_failed",
-				slog.String("pipeline_id", pid),
-				slog.Any("error", err),
-			)
-			failures = append(failures, DispatchFailure{
-				PipelineID: pid, Reason: "orchestrator submit", Error: err.Error(),
-			})
-			continue
-		}
-		// Hook the log archiver onto the new execution so per-node lines
-		// land in MinIO when each node finishes.
-		startLogArchiver(context.Background(), s.logs, s.events, execID)
-
-		// Broadcast so /runs etc. light up without polling. We extract
-		// `source` from the trigger payload (manual / github_push).
-		source := ""
-		if items, ok := trigger.([]map[string]any); ok && len(items) > 0 {
-			if s, _ := items[0]["source"].(string); s != "" {
-				source = s
+			wf, err := engine.ParseWorkflow(p.Definition)
+			if err != nil {
+				failures = append(failures, DispatchFailure{
+					Environment: envName, PipelineID: pid, Reason: "parse failed", Error: err.Error(),
+				})
+				continue
+			}
+			if isPush {
+				tb := pipelineTriggerBranch(wf)
+				if tb != "" && tb != "*" && tb != pushedBranch {
+					failures = append(failures, DispatchFailure{
+						Environment: envName, PipelineID: pid, Reason: "branch filtered",
+						Error: fmt.Sprintf("trigger fromBranch=%q ≠ pushed=%q", tb, pushedBranch),
+					})
+					continue
+				}
 			}
-		}
-		s.runsBus.Publish(RunCreatedEvent{
-			Type:         "run_created",
-			ExecutionID:  execID,
-			PipelineID:   pid,
-			PipelineName: p.Name,
-			AgentID:      a.ID,
-			Source:       source,
-			StartedAt:    time.Now().UTC(),
-		})
 
-		if s.runs != nil {
-			_ = s.runs.Insert(ctx, &storage.Run{
-				ID:           execID,
+			// Per-(env, pipeline) trigger payload: clone the base items and
+			// stamp the env name so downstream nodes (Deploy / Approval) can
+			// inherit env defaults.
+			perRunItems := stampTriggerEnv(trigger, envName)
+			triggerJSON, _ := json.Marshal(perRunItems)
+			triggerItems := make([]models.Item, 0, len(perRunItems))
+			for _, m := range perRunItems {
+				triggerItems = append(triggerItems, models.Item(m))
+			}
+
+			runCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
+			execID, err := s.orch.RunAsync(runCtx, &orchestrator.RunRequest{
+				RequestMeta: orchestrator.RequestMeta{WorkflowID: pid},
+				Workflow:    wf,
+				TriggerData: triggerItems,
+			})
+			cancel()
+			if err != nil {
+				failures = append(failures, DispatchFailure{
+					Environment: envName, PipelineID: pid, Reason: "orchestrator submit", Error: err.Error(),
+				})
+				continue
+			}
+			startLogArchiver(context.Background(), s.logs, s.events, execID)
+			s.runsBus.Publish(RunCreatedEvent{
+				Type:         "run_created",
+				ExecutionID:  execID,
 				PipelineID:   pid,
 				PipelineName: p.Name,
-				Status:       "running",
+				AgentID:      a.ID,
+				Environment:  envName,
+				Source:       source,
 				StartedAt:    time.Now().UTC(),
-				TriggerData:  triggerJSON,
 			})
+			if s.runs != nil {
+				_ = s.runs.Insert(ctx, &storage.Run{
+					ID:           execID,
+					PipelineID:   pid,
+					PipelineName: p.Name,
+					Status:       "running",
+					StartedAt:    time.Now().UTC(),
+					TriggerData:  triggerJSON,
+				})
+			}
+			out = append(out, execID)
 		}
-		out = append(out, execID)
 	}
 	return out, failures, nil
 }
 
-// DispatchFailure describes why a single attached pipeline was skipped at
-// dispatch time. We surface these to the API caller so trigger failures
+// stampTriggerEnv returns a copy of the trigger items (each a map) with
+// `environment` set to envName. The base items are not mutated so the
+// same trigger payload can be reused across environments.
+func stampTriggerEnv(trigger any, envName string) []map[string]any {
+	items, ok := trigger.([]map[string]any)
+	if !ok || len(items) == 0 {
+		return []map[string]any{{"environment": envName}}
+	}
+	out := make([]map[string]any, 0, len(items))
+	for _, m := range items {
+		cp := make(map[string]any, len(m)+1)
+		for k, v := range m {
+			cp[k] = v
+		}
+		cp["environment"] = envName
+		out = append(out, cp)
+	}
+	return out
+}
+
+// DispatchFailure describes why a single (environment, pipeline) pair was
+// skipped at dispatch time. Surfaced to API callers so trigger failures
 // don't appear as silent no-ops.
 type DispatchFailure struct {
-	PipelineID string `json:"pipelineId"`
-	Reason     string `json:"reason"`
-	Error      string `json:"error,omitempty"`
+	Environment string `json:"environment,omitempty"`
+	PipelineID  string `json:"pipelineId,omitempty"`
+	Reason      string `json:"reason"`
+	Error       string `json:"error,omitempty"`
 }
 
 // --- webhook receiver ----------------------------------------------------

pkg/api/environments.go

@@ -0,0 +1,261 @@
+package api
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net/http"
+	"strings"
+	"time"
+
+	"github.com/lyzrai/flow/pkg/storage"
+)
+
+// Environments API — global, named deploy stages. An environment is a
+// sequencing container: it owns an ordered list of pipelines (the
+// promotion sequence) plus a description. Per-deploy config lives on the
+// nodes, not here. Agents subscribe to environments (see agents.go);
+// dispatch fans out across the followed envs' pipelines.
+//
+// Routes:
+//   GET    /api/environments
+//   POST   /api/environments
+//   GET    /api/environments/{name}
+//   PUT    /api/environments/{name}                          (name + description)
+//   DELETE /api/environments/{name}
+//   POST   /api/environments/{name}/pipelines/{pipelineId}   (bring a pipeline in — appends)
+//   DELETE /api/environments/{name}/pipelines/{pipelineId}   (remove a pipeline)
+//   PUT    /api/environments/{name}/pipelines                (reorder; body {"pipelineIds":[...]})
+
+// environmentBody is the create/update wire shape. PipelineIDs are
+// managed via the /pipelines sub-routes.
+type environmentBody struct {
+	Name        string `json:"name"`
+	Description string `json:"description,omitempty"`
+}
+
+func (b *environmentBody) normalize() (storage.Environment, error) {
+	name := strings.TrimSpace(b.Name)
+	if name == "" {
+		return storage.Environment{}, errors.New("name is required")
+	}
+	now := time.Now().UTC()
+	return storage.Environment{
+		Name:        name,
+		Description: strings.TrimSpace(b.Description),
+		CreatedAt:   now,
+		UpdatedAt:   now,
+	}, nil
+}
+
+func (s *Server) handleListEnvironments(w http.ResponseWriter, r *http.Request) {
+	if s.environments == nil {
+		writeError(w, http.StatusServiceUnavailable, errors.New("environments store not configured"))
+		return
+	}
+	list, err := s.environments.List(r.Context())
+	if err != nil {
+		writeError(w, http.StatusInternalServerError, err)
+		return
+	}
+	if list == nil {
+		list = []*storage.Environment{}
+	}
+	writeJSON(w, http.StatusOK, list)
+}
+
+func (s *Server) handleGetEnvironment(w http.ResponseWriter, r *http.Request) {
+	if s.environments == nil {
+		writeError(w, http.StatusServiceUnavailable, errors.New("environments store not configured"))
+		return
+	}
+	e, err := s.environments.GetByName(r.Context(), r.PathValue("name"))
+	if err != nil {
+		writeStorageErr(w, err, "environment not found")
+		return
+	}
+	writeJSON(w, http.StatusOK, e)
+}
+
+func (s *Server) handleCreateEnvironment(w http.ResponseWriter, r *http.Request) {
+	if s.environments == nil {
+		writeError(w, http.StatusServiceUnavailable, errors.New("environments store not configured"))
+		return
+	}
+	var body environmentBody
+	if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
+		writeError(w, http.StatusBadRequest, err)
+		return
+	}
+	env, err := body.normalize()
+	if err != nil {
+		writeError(w, http.StatusBadRequest, err)
+		return
+	}
+	env.ID = newID()
+	if err := s.environments.Create(r.Context(), &env); err != nil {
+		if errors.Is(err, storage.ErrAlreadyExists) {
+			writeError(w, http.StatusConflict, fmt.Errorf("environment %q already exists", env.Name))
+			return
+		}
+		writeError(w, http.StatusInternalServerError, err)
+		return
+	}
+	writeJSON(w, http.StatusCreated, &env)
+}
+
+func (s *Server) handleUpdateEnvironment(w http.ResponseWriter, r *http.Request) {
+	if s.environments == nil {
+		writeError(w, http.StatusServiceUnavailable, errors.New("environments store not configured"))
+		return
+	}
+	name := r.PathValue("name")
+	existing, err := s.environments.GetByName(r.Context(), name)
+	if err != nil {
+		writeStorageErr(w, err, "environment not found")
+		return
+	}
+	var body environmentBody
+	if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
+		writeError(w, http.StatusBadRequest, err)
+		return
+	}
+	if body.Name == "" {
+		body.Name = name
+	}
+	updated, err := body.normalize()
+	if err != nil {
+		writeError(w, http.StatusBadRequest, err)
+		return
+	}
+	// Name is immutable through PUT; carry forward identity + pipeline set.
+	updated.Name = existing.Name
+	updated.ID = existing.ID
+	updated.PipelineIDs = existing.PipelineIDs
+	updated.CreatedAt = existing.CreatedAt
+	updated.UpdatedAt = time.Now().UTC()
+	if err := s.environments.Update(r.Context(), &updated); err != nil {
+		writeError(w, http.StatusInternalServerError, err)
+		return
+	}
+	writeJSON(w, http.StatusOK, &updated)
+}
+
+func (s *Server) handleDeleteEnvironment(w http.ResponseWriter, r *http.Request) {
+	if s.environments == nil {
+		writeError(w, http.StatusServiceUnavailable, errors.New("environments store not configured"))
+		return
+	}
+	if err := s.environments.Delete(r.Context(), r.PathValue("name")); err != nil {
+		writeStorageErr(w, err, "environment not found")
+		return
+	}
+	w.WriteHeader(http.StatusNoContent)
+}
+
+func (s *Server) handleEnvAddPipeline(w http.ResponseWriter, r *http.Request) {
+	if s.environments == nil {
+		writeError(w, http.StatusServiceUnavailable, errors.New("environments store not configured"))
+		return
+	}
+	name := r.PathValue("name")
+	pipelineID := r.PathValue("pipelineId")
+	env, err := s.environments.GetByName(r.Context(), name)
+	if err != nil {
+		writeStorageErr(w, err, "environment not found")
+		return
+	}
+	if _, err := s.pipelines.Get(r.Context(), pipelineID); err != nil {
+		writeStorageErr(w, err, "pipeline not found")
+		return
+	}
+	if env.HasPipeline(pipelineID) {
+		writeJSON(w, http.StatusOK, env) // already in this env
+		return
+	}
+	env.PipelineIDs = append(env.PipelineIDs, pipelineID)
+	env.UpdatedAt = time.Now().UTC()
+	if err := s.environments.Update(r.Context(), env); err != nil {
+		writeError(w, http.StatusInternalServerError, err)
+		return
+	}
+	writeJSON(w, http.StatusOK, env)
+}
+
+func (s *Server) handleEnvRemovePipeline(w http.ResponseWriter, r *http.Request) {
+	if s.environments == nil {
+		writeError(w, http.StatusServiceUnavailable, errors.New("environments store not configured"))
+		return
+	}
+	name := r.PathValue("name")
+	pipelineID := r.PathValue("pipelineId")
+	env, err := s.environments.GetByName(r.Context(), name)
+	if err != nil {
+		writeStorageErr(w, err, "environment not found")
+		return
+	}
+	out := env.PipelineIDs[:0]
+	for _, p := range env.PipelineIDs {
+		if p != pipelineID {
+			out = append(out, p)
+		}
+	}
+	env.PipelineIDs = out
+	env.UpdatedAt = time.Now().UTC()
+	if err := s.environments.Update(r.Context(), env); err != nil {
+		writeError(w, http.StatusInternalServerError, err)
+		return
+	}
+	w.WriteHeader(http.StatusNoContent)
+}
+
+// handleEnvReorderPipelines replaces the env's pipeline order. The body
+// must be a permutation of the env's current pipeline set — extra or
+// missing IDs are rejected so a stale client can't silently drop a
+// pipeline.
+func (s *Server) handleEnvReorderPipelines(w http.ResponseWriter, r *http.Request) {
+	if s.environments == nil {
+		writeError(w, http.StatusServiceUnavailable, errors.New("environments store not configured"))
+		return
+	}
+	name := r.PathValue("name")
+	env, err := s.environments.GetByName(r.Context(), name)
+	if err != nil {
+		writeStorageErr(w, err, "environment not found")
+		return
+	}
+	var body struct {
+		PipelineIDs []string `json:"pipelineIds"`
+	}
+	if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
+		writeError(w, http.StatusBadRequest, err)
+		return
+	}
+	cur := map[string]bool{}
+	for _, p := range env.PipelineIDs {
+		cur[p] = true
+	}
+	if len(body.PipelineIDs) != len(env.PipelineIDs) {
+		writeError(w, http.StatusBadRequest, fmt.Errorf("expected a permutation of %d pipeline ids, got %d", len(env.PipelineIDs), len(body.PipelineIDs)))
+		return
+	}
+	seen := map[string]bool{}
+	for _, p := range body.PipelineIDs {
+		if !cur[p] {
+			writeError(w, http.StatusBadRequest, fmt.Errorf("pipeline %q is not in this environment", p))
+			return
+		}
+		if seen[p] {
+			writeError(w, http.StatusBadRequest, fmt.Errorf("pipeline %q listed twice", p))
+			return
+		}
+		seen[p] = true
+	}
+	env.PipelineIDs = body.PipelineIDs
+	env.UpdatedAt = time.Now().UTC()
+	if err := s.environments.Update(r.Context(), env); err != nil {
+		writeError(w, http.StatusInternalServerError, err)
+		return
+	}
+	writeJSON(w, http.StatusOK, env)
+}

pkg/api/runs_bus.go

@@ -14,6 +14,7 @@ type RunCreatedEvent struct {
 	PipelineID   string    `json:"pipelineId,omitempty"`
 	PipelineName string    `json:"pipelineName,omitempty"`
 	AgentID      string    `json:"agentId,omitempty"`
+	Environment  string    `json:"environment,omitempty"`
 	Source       string    `json:"source,omitempty"` // "manual" | "github_push"
 	StartedAt    time.Time `json:"startedAt"`
 }

pkg/api/server.go

@@ -58,10 +58,11 @@ type ServerDeps struct {
 	// "https://abcd.trycloudflare.com"). Used to render webhook callback
 	// URLs that GitHub can hit. Empty means webhook install is disabled.
 	PublicURL string
-	Pipelines   storage.PipelineStore
-	Runs        storage.RunStore
-	Agents      storage.AgentStore
-	Credentials storage.CredentialStore
+	Pipelines    storage.PipelineStore
+	Runs         storage.RunStore
+	Agents       storage.AgentStore
+	Credentials  storage.CredentialStore
+	Environments storage.EnvironmentStore
 	// Events is the in-memory pub/sub bus the orchestrator publishes
 	// per-node lifecycle events to. The SSE handler subscribes per
 	// execution ID. Nil disables /api/executions/{id}/stream.
@@ -88,12 +89,13 @@ type Server struct {
 	corsOrigins   []string
 	publicURL     string
 
-	pipelines   storage.PipelineStore
-	runs        storage.RunStore
-	agents      storage.AgentStore
-	credentials storage.CredentialStore
-	events      EventSubscriber
-	logs        logstore.Store
+	pipelines    storage.PipelineStore
+	runs         storage.RunStore
+	agents       storage.AgentStore
+	credentials  storage.CredentialStore
+	environments storage.EnvironmentStore
+	events       EventSubscriber
+	logs         logstore.Store
 
 	// runsBus broadcasts run_created events to every UI tab subscribed to
 	// /api/runs/stream. Used so a webhook-triggered run shows up live in
@@ -111,11 +113,12 @@ func NewServer(deps ServerDeps) *Server {
 		restateIngres: deps.RestateIngressURL,
 		corsOrigins:   deps.CORSOrigins,
 		publicURL:     strings.TrimRight(deps.PublicURL, "/"),
-		pipelines:   deps.Pipelines,
-		runs:        deps.Runs,
-		agents:      deps.Agents,
-		credentials: deps.Credentials,
-		events:      deps.Events,
+		pipelines:    deps.Pipelines,
+		runs:         deps.Runs,
+		agents:       deps.Agents,
+		credentials:  deps.Credentials,
+		environments: deps.Environments,
+		events:       deps.Events,
 		logs:          deps.Logs,
 		runsBus:       newRunsBus(),
 	}
@@ -166,8 +169,8 @@ func (s *Server) routes() {
 	s.mux.HandleFunc("POST /api/agents/{id}/test-auth", s.handleTestAgentAuth)
 	s.mux.HandleFunc("POST /api/agents/{id}/webhook", s.handleInstallWebhook)
 	s.mux.HandleFunc("DELETE /api/agents/{id}/webhook", s.handleUninstallWebhook)
-	s.mux.HandleFunc("POST /api/agents/{id}/pipelines/{pipelineId}", s.handleAttachPipeline)
-	s.mux.HandleFunc("DELETE /api/agents/{id}/pipelines/{pipelineId}", s.handleDetachPipeline)
+	s.mux.HandleFunc("POST /api/agents/{id}/environments/{envName}", s.handleAgentFollowEnv)
+	s.mux.HandleFunc("DELETE /api/agents/{id}/environments/{envName}", s.handleAgentUnfollowEnv)
 	s.mux.HandleFunc("POST /api/agents/{id}/trigger", s.handleTriggerAgent)
 
 	// Credentials API — global org-wide pool. Lookup by name is shared
@@ -187,6 +190,16 @@ func (s *Server) routes() {
 	s.mux.HandleFunc("PUT /api/agents/{id}/credentials/{name}", s.handleUpdateCredential)
 	s.mux.HandleFunc("DELETE /api/agents/{id}/credentials/{name}", s.handleDeleteCredential)
 
+	// Environments API — global deploy stages owning pipelines + config.
+	s.mux.HandleFunc("GET /api/environments", s.handleListEnvironments)
+	s.mux.HandleFunc("POST /api/environments", s.handleCreateEnvironment)
+	s.mux.HandleFunc("GET /api/environments/{name}", s.handleGetEnvironment)
+	s.mux.HandleFunc("PUT /api/environments/{name}", s.handleUpdateEnvironment)
+	s.mux.HandleFunc("DELETE /api/environments/{name}", s.handleDeleteEnvironment)
+	s.mux.HandleFunc("POST /api/environments/{name}/pipelines/{pipelineId}", s.handleEnvAddPipeline)
+	s.mux.HandleFunc("DELETE /api/environments/{name}/pipelines/{pipelineId}", s.handleEnvRemovePipeline)
+	s.mux.HandleFunc("PUT /api/environments/{name}/pipelines", s.handleEnvReorderPipelines)
+
 	// Public webhook receiver. GitHub posts here; HMAC signature is the
 	// authentication. Must NOT require CORS / API auth.
 	s.mux.HandleFunc("POST /webhooks/github/{id}", s.handleGitHubWebhook)

pkg/executors/approval.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"log/slog"
+	"time"
 
 	"github.com/google/uuid"
 	restate "github.com/restatedev/sdk-go"
@@ -17,6 +18,13 @@ import (
 // It pauses the workflow using a Restate Awakeable and blocks until an external
 // caller resolves it via POST /api/executions/{id}/resume.
 //
+// Node params (all optional):
+//   - method "ui" (default) | "quorum" | "auto" — "auto" emits on output 0
+//     immediately without pausing. quorum N>1 is surfaced in the reviewer
+//     context; strict N-approver enforcement is a follow-up.
+//   - minApprovers — for method "quorum".
+//   - timeoutSeconds — reject automatically after that long.
+//
 // Two outputs:
 //   - Output 0: approved — items flow with human-supplied data merged in
 //   - Output 1: rejected — items flow with rejection_reason field
@@ -39,6 +47,27 @@ func (e *ApprovalExecutor) Execute(
 		inputItems = []models.Item{{}}
 	}
 
+	// Resolve the effective approval policy from node params.
+	method := strParam(node.Parameters, "method", "")
+	if method == "" {
+		method = "ui"
+	}
+	minApprovers := intParam(node.Parameters, "minApprovers", 1)
+	timeoutSeconds := intParam(node.Parameters, "timeoutSeconds", 0)
+
+	// Auto method: don't pause at all — emit straight to the approved output.
+	if method == "auto" {
+		slog.InfoContext(ctx, "approval_auto", slog.String("node", node.Name))
+		var out []models.Item
+		for _, item := range inputItems {
+			m := copyItem(item)
+			m["approved"] = true
+			m["approval_method"] = "auto"
+			out = append(out, m)
+		}
+		return map[int][]models.Item{0: out}, nil
+	}
+
 	// Approval requires Restate for durable blocking.
 	raw := durability.RestateCtxFromContext(ctx)
 	rctx, ok := raw.(restate.WorkflowContext)
@@ -69,6 +98,13 @@ func (e *ApprovalExecutor) Execute(
 	if reason, ok := resolved["reason"].(string); ok && reason != "" {
 		approvalCtx["reason"] = reason
 	}
+	approvalCtx["method"] = method
+	if method == "quorum" {
+		approvalCtx["minApprovers"] = minApprovers
+	}
+	if timeoutSeconds > 0 {
+		approvalCtx["timeoutSeconds"] = timeoutSeconds
+	}
 	if len(inputItems) == 1 {
 		approvalCtx["inputs"] = map[string]any(inputItems[0])
 	} else if len(inputItems) > 1 {
@@ -111,10 +147,29 @@ func (e *ApprovalExecutor) Execute(
 		slog.String("awakeable_id", awakeableID),
 	)
 
-	// Block on the Awakeable. Durable across crashes / restarts.
-	approvalData, err := awakeable.Result()
-	if err != nil {
-		return nil, fmt.Errorf("approval node %q: %w", node.Name, err)
+	// Block on the Awakeable. Durable across crashes / restarts. If a
+	// timeout is set, race it against a durable Restate timer — whichever
+	// fires first wins; on timeout we route to the rejected output.
+	var approvalData map[string]any
+	timedOut := false
+	if timeoutSeconds > 0 {
+		selector := restate.Select(rctx, awakeable, restate.After(rctx, time.Duration(timeoutSeconds)*time.Second))
+		switch winner := selector.Select(); winner {
+		case awakeable:
+			d, err := awakeable.Result()
+			if err != nil {
+				return nil, fmt.Errorf("approval node %q: %w", node.Name, err)
+			}
+			approvalData = d
+		default:
+			timedOut = true
+		}
+	} else {
+		d, err := awakeable.Result()
+		if err != nil {
+			return nil, fmt.Errorf("approval node %q: %w", node.Name, err)
+		}
+		approvalData = d
 	}
 
 	// Clear pending markers now that we've resumed.
@@ -122,6 +177,19 @@ func (e *ApprovalExecutor) Execute(
 	restate.Clear(rctx, "pending_approval_id")
 	restate.Clear(rctx, "pending_approval_context")
 
+	if timedOut {
+		slog.InfoContext(ctx, "approval_timed_out",
+			slog.String("node", node.Name), slog.Int("timeout_seconds", timeoutSeconds))
+		var rejected []models.Item
+		for _, item := range inputItems {
+			r := copyItem(item)
+			r["approved"] = false
+			r["rejection_reason"] = fmt.Sprintf("approval timed out after %ds", timeoutSeconds)
+			rejected = append(rejected, r)
+		}
+		return map[int][]models.Item{1: rejected}, nil
+	}
+
 	slog.InfoContext(ctx, "workflow_resumed",
 		slog.String("node", node.Name),
 	)

pkg/executors/deploy.go

@@ -44,16 +44,6 @@ func (e *DeployExecutor) Execute(ctx context.Context, node models.NodeDef, input
 	}
 	logger := engine.NodeLoggerFromContext(ctx)
 
-	target := strings.ToLower(strParam(node.Parameters, "target", "agentcore"))
-	switch target {
-	case "agentcore":
-		// supported below
-	case "k8s", "kubernetes", "vertex", "":
-		return nil, fmt.Errorf("deploy: target %q not yet implemented (only \"agentcore\" today)", target)
-	default:
-		return nil, fmt.Errorf("deploy: unknown target %q", target)
-	}
-
 	trigger := firstItem(inputs)
 	agentID, _ := trigger["agentId"].(string)
 	if agentID == "" {
@@ -63,18 +53,33 @@ func (e *DeployExecutor) Execute(ctx context.Context, node models.NodeDef, input
 	if err != nil {
 		return nil, fmt.Errorf("deploy: load agent %q: %w", agentID, err)
 	}
-	credName := strParam(node.Parameters, "credentialName", "aws")
+	// envName is purely informational here (recorded in the summary).
+	envName, _ := trigger["environment"].(string)
+
+	// target: node param → "agentcore".
+	target := strings.ToLower(strParam(node.Parameters, "target", ""))
+	if target == "" {
+		target = "agentcore"
+	}
+	switch target {
+	case "agentcore":
+		// supported below
+	case "k8s", "kubernetes", "vertex":
+		return nil, fmt.Errorf("deploy: target %q not yet implemented (only \"agentcore\" today)", target)
+	default:
+		return nil, fmt.Errorf("deploy: unknown target %q", target)
+	}
+
+	// credentialName: node param → "aws".
+	credName := strParam(node.Parameters, "credentialName", "")
+	if credName == "" {
+		credName = "aws"
+	}
 	cred, credScope, err := e.lookupCredential(ctx, a, credName)
 	if err != nil {
 		return nil, fmt.Errorf("deploy: %w", err)
 	}
-	logger.Log(fmt.Sprintf("[deploy] using %s credential %q", credScope, credName))
-	if cred.Type != storage.CredentialAWS {
-		return nil, fmt.Errorf("deploy: credential %q is type %q; target=agentcore needs an aws credential", credName, cred.Type)
-	}
-	if cred.AwsRegion == "" || cred.AwsAccountID == "" || cred.AwsCrossAccountRoleArn == "" {
-		return nil, fmt.Errorf("deploy: aws credential %q is missing region / accountId / crossAccountRoleArn", credName)
-	}
+	logger.Log(fmt.Sprintf("[deploy] target=%s credential=%s (%s)", target, credName, credScope))
 
 	image := resolveDeployImage(node.Parameters, inputs)
 	if image == "" {
@@ -143,6 +148,7 @@ func (e *DeployExecutor) Execute(ctx context.Context, node models.NodeDef, input
 
 	summary := map[string]any{
 		"target":          "agentcore",
+		"environment":     envName,
 		"agentId":         agentID,
 		"agentName":       a.Name,
 		"credentialName":  cred.Name,

pkg/storage/mongo.go

@@ -72,6 +72,12 @@ func (m *Mongo) Credentials() CredentialStore {
 	return &mongoCredentials{coll: m.db.Collection("credentials")}
 }
 
+// Environments returns the global EnvironmentStore backed by this Mongo
+// connection.
+func (m *Mongo) Environments() EnvironmentStore {
+	return &mongoEnvironments{coll: m.db.Collection("environments")}
+}
+
 func (m *Mongo) ensureIndexes(ctx context.Context) error {
 	if _, err := m.db.Collection("pipelines").Indexes().CreateMany(ctx, []mongo.IndexModel{
 		{Keys: bson.D{{Key: "updated_at", Value: -1}}},
@@ -95,6 +101,12 @@ func (m *Mongo) ensureIndexes(ctx context.Context) error {
 	}); err != nil {
 		return fmt.Errorf("credentials indexes: %w", err)
 	}
+	if _, err := m.db.Collection("environments").Indexes().CreateMany(ctx, []mongo.IndexModel{
+		{Keys: bson.D{{Key: "name", Value: 1}}, Options: options.Index().SetUnique(true)},
+		{Keys: bson.D{{Key: "updated_at", Value: -1}}},
+	}); err != nil {
+		return fmt.Errorf("environments indexes: %w", err)
+	}
 	return nil
 }
 
@@ -522,3 +534,63 @@ func (s *mongoCredentials) List(ctx context.Context) ([]*Credential, error) {
 	}
 	return out, nil
 }
+
+// --- environments (global) -----------------------------------------------
+
+type mongoEnvironments struct{ coll *mongo.Collection }
+
+func (s *mongoEnvironments) Create(ctx context.Context, e *Environment) error {
+	if _, err := s.coll.InsertOne(ctx, e); err != nil {
+		if mongo.IsDuplicateKeyError(err) {
+			return ErrAlreadyExists
+		}
+		return err
+	}
+	return nil
+}
+
+func (s *mongoEnvironments) GetByName(ctx context.Context, name string) (*Environment, error) {
+	var e Environment
+	if err := s.coll.FindOne(ctx, bson.M{"name": name}).Decode(&e); err != nil {
+		if errors.Is(err, mongo.ErrNoDocuments) {
+			return nil, ErrNotFound
+		}
+		return nil, err
+	}
+	return &e, nil
+}
+
+func (s *mongoEnvironments) Update(ctx context.Context, e *Environment) error {
+	res, err := s.coll.ReplaceOne(ctx, bson.M{"name": e.Name}, e)
+	if err != nil {
+		return err
+	}
+	if res.MatchedCount == 0 {
+		return ErrNotFound
+	}
+	return nil
+}
+
+func (s *mongoEnvironments) Delete(ctx context.Context, name string) error {
+	res, err := s.coll.DeleteOne(ctx, bson.M{"name": name})
+	if err != nil {
+		return err
+	}
+	if res.DeletedCount == 0 {
+		return ErrNotFound
+	}
+	return nil
+}
+
+func (s *mongoEnvironments) List(ctx context.Context) ([]*Environment, error) {
+	cur, err := s.coll.Find(ctx, bson.M{}, options.Find().SetSort(bson.D{{Key: "name", Value: 1}}))
+	if err != nil {
+		return nil, err
+	}
+	defer cur.Close(ctx)
+	var out []*Environment
+	if err := cur.All(ctx, &out); err != nil {
+		return nil, err
+	}
+	return out, nil
+}

pkg/storage/storage.go

@@ -132,9 +132,14 @@ type Agent struct {
 	WebhookInstalledAt *time.Time `json:"webhookInstalledAt,omitempty" bson:"webhook_installed_at,omitempty"`
 	AuthStatus         AuthStatus `json:"authStatus,omitempty"        bson:"auth_status,omitempty"`
 	AuthCheckedAt      *time.Time `json:"authCheckedAt,omitempty"     bson:"auth_checked_at,omitempty"`
-	AttachedPipelines  []string   `json:"attachedPipelines,omitempty" bson:"attached_pipelines,omitempty"`
+	// Environments this agent follows by name. Triggering the agent runs
+	// the pipelines of these environments (filtered by branch). Replaces
+	// the older flat AttachedPipelines list — pipelines now live on the
+	// environment, and agents subscribe to environments.
+	Environments []string `json:"environments,omitempty" bson:"environments,omitempty"`
 
 	// Named credentials — referenced by name from Deploy / future nodes.
+	// These are agent-specific overrides of the global credential pool.
 	Credentials []Credential `json:"credentials,omitempty" bson:"credentials,omitempty"`
 
 	CreatedAt time.Time `json:"createdAt"         bson:"created_at"`
@@ -177,3 +182,51 @@ type CredentialStore interface {
 	Delete(ctx context.Context, name string) error
 	List(ctx context.Context) ([]*Credential, error)
 }
+
+// Environment is a global, named deploy stage (dev / staging / prod /
+// custom — the name is free-form). It is purely a sequencing container:
+// it owns an ordered list of pipeline IDs (the promotion sequence) plus
+// a description. Per-deploy config (credential, runtime target, approval
+// method) lives on the nodes themselves, not here.
+//
+// Agents subscribe to environments by name (agent.Environments);
+// triggering an agent runs the pipelines of the environments it follows
+// (each pipeline still gated by its Trigger node's branch filter).
+//
+// A pipeline may appear in more than one environment.
+type Environment struct {
+	ID          string `json:"id"                    bson:"_id"`
+	Name        string `json:"name"                  bson:"name"` // unique
+	Description string `json:"description,omitempty"  bson:"description,omitempty"`
+
+	// PipelineIDs is ordered — the order is the promotion sequence and is
+	// reorderable via the API. Dispatch still applies each pipeline's own
+	// branch filter; the order is the documented progression.
+	PipelineIDs []string `json:"pipelineIds,omitempty" bson:"pipeline_ids,omitempty"`
+
+	CreatedAt time.Time `json:"createdAt" bson:"created_at"`
+	UpdatedAt time.Time `json:"updatedAt" bson:"updated_at"`
+}
+
+// HasPipeline reports whether the pipeline ID is brought into this env.
+func (e *Environment) HasPipeline(pipelineID string) bool {
+	if e == nil {
+		return false
+	}
+	for _, p := range e.PipelineIDs {
+		if p == pipelineID {
+			return true
+		}
+	}
+	return false
+}
+
+// EnvironmentStore persists global environments. Lookup is by name (the
+// user-facing identifier — pipelines/agents reference envs by name).
+type EnvironmentStore interface {
+	Create(ctx context.Context, e *Environment) error
+	GetByName(ctx context.Context, name string) (*Environment, error)
+	Update(ctx context.Context, e *Environment) error
+	Delete(ctx context.Context, name string) error
+	List(ctx context.Context) ([]*Environment, error)
+}