diff --git a/cmd/flow/main.go b/cmd/flow/main.go index 856981a..c34a0e3 100644 --- a/cmd/flow/main.go +++ b/cmd/flow/main.go @@ -219,6 +219,7 @@ func serve() int { Runs: mongo.Runs(), Agents: mongo.Agents(), Credentials: mongo.Credentials(), + Environments: mongo.Environments(), Events: eventBus, Logs: logs, }), diff --git a/pkg/api/agents.go b/pkg/api/agents.go index b4f1a1b..832cc19 100644 --- a/pkg/api/agents.go +++ b/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 ---------------------------------------------------- diff --git a/pkg/api/environments.go b/pkg/api/environments.go new file mode 100644 index 0000000..fd832bf --- /dev/null +++ b/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) +} diff --git a/pkg/api/runs_bus.go b/pkg/api/runs_bus.go index 753b55a..f1264b3 100644 --- a/pkg/api/runs_bus.go +++ b/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"` } diff --git a/pkg/api/server.go b/pkg/api/server.go index e5b11da..ce69b01 100644 --- a/pkg/api/server.go +++ b/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) diff --git a/pkg/executors/approval.go b/pkg/executors/approval.go index 397a5a9..e1e8179 100644 --- a/pkg/executors/approval.go +++ b/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), ) diff --git a/pkg/executors/deploy.go b/pkg/executors/deploy.go index a5e7486..b266fbc 100644 --- a/pkg/executors/deploy.go +++ b/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, diff --git a/pkg/storage/mongo.go b/pkg/storage/mongo.go index 30946af..9d5c787 100644 --- a/pkg/storage/mongo.go +++ b/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 +} diff --git a/pkg/storage/storage.go b/pkg/storage/storage.go index 53ea048..5be8a74 100644 --- a/pkg/storage/storage.go +++ b/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) +} diff --git a/web/app/agents/view/page.tsx b/web/app/agents/view/page.tsx index 7a67ca0..7ceb2f9 100644 --- a/web/app/agents/view/page.tsx +++ b/web/app/agents/view/page.tsx @@ -34,6 +34,7 @@ import { api, type Agent, type AuthStatus, + type Environment, type FlowSummary, type PublicCredential, type Run, @@ -56,27 +57,35 @@ function AgentDetail() { const [agent, setAgent] = useState(null); const [config, setConfig] = useState(null); + const [environments, setEnvironments] = useState([]); const [pipelines, setPipelines] = useState([]); const [runs, setRuns] = useState([]); const [error, setError] = useState(null); const [busy, setBusy] = useState(null); // which action is in flight - const [showPipelinePicker, setShowPipelinePicker] = useState(false); + const [showEnvPicker, setShowEnvPicker] = useState(false); async function load() { if (!id) return; try { - const [a, cfg, allPipes] = await Promise.all([ + const [a, cfg, allEnvs, allPipes] = await Promise.all([ api.getAgent(id), api.getConfig().catch(() => null), + api.listEnvironments().catch(() => []), api.listFlows().catch(() => []), ]); setAgent(a); setConfig(cfg); + setEnvironments(allEnvs); setPipelines(allPipes); - // Pull recent runs across all attached pipelines. - if (a.attachedPipelines?.length) { + // Recent runs across the pipelines of every followed env. + const followed = (a.environments ?? []) + .map((n) => allEnvs.find((e) => e.name === n)) + .filter((e): e is Environment => Boolean(e)); + const pipelineIds = new Set(); + followed.forEach((e) => (e.pipelineIds ?? []).forEach((pid) => pipelineIds.add(pid))); + if (pipelineIds.size) { const lists = await Promise.all( - a.attachedPipelines.map((pid) => + [...pipelineIds].map((pid) => api.listRuns({ pipelineId: pid, limit: 5 }).catch(() => []) ) ); @@ -134,7 +143,10 @@ function AgentDetail() { } throw new Error( fails - .map((f) => `${f.pipelineId}: ${f.reason}${f.error ? " — " + f.error : ""}`) + .map((f) => { + const where = [f.environment, f.pipelineId].filter(Boolean).join("/"); + return `${where || "?"}: ${f.reason}${f.error ? " — " + f.error : ""}`; + }) .join("; ") ); }); @@ -162,18 +174,18 @@ function AgentDetail() { }); } - async function onAttachPipeline(pipelineId: string) { - await withBusy("attach", async () => { - await api.attachPipeline(id, pipelineId); - setShowPipelinePicker(false); + async function onFollowEnv(envName: string) { + await withBusy("follow-env", async () => { + await api.agentFollowEnv(id, envName); + setShowEnvPicker(false); await load(); }); } - async function onDetachPipeline(pipelineId: string) { - if (!confirm("Detach this pipeline from the agent?")) return; - await withBusy("detach", async () => { - await api.detachPipeline(id, pipelineId); + async function onUnfollowEnv(envName: string) { + if (!confirm(`Stop following environment "${envName}"? This agent will no longer dispatch its pipelines.`)) return; + await withBusy("unfollow-env", async () => { + await api.agentUnfollowEnv(id, envName); await load(); }); } @@ -207,12 +219,14 @@ function AgentDetail() { } const lastRun = runs[0]; - const attachedPipelineDetails = (agent.attachedPipelines ?? []) - .map((pid) => pipelines.find((p) => p.id === pid)) - .filter((p): p is FlowSummary => Boolean(p)); - const attachable = pipelines.filter( - (p) => !agent.attachedPipelines?.includes(p.id) + const followedEnvs = (agent.environments ?? []) + .map((n) => environments.find((e) => e.name === n)) + .filter((e): e is Environment => Boolean(e)); + const followableEnvs = environments.filter( + (e) => !agent.environments?.includes(e.name) ); + const pipelineName = (pid: string) => + pipelines.find((p) => p.id === pid)?.name ?? pid; return (
@@ -263,16 +277,16 @@ function AgentDetail() { disabled={ busy === "trigger" || !config?.orchestratorEnabled || - !agent.attachedPipelines?.length + !agent.environments?.length } title={ !config?.orchestratorEnabled ? "Orchestrator not configured (Restate unreachable)" - : !agent.attachedPipelines?.length - ? "Attach a pipeline first" + : !agent.environments?.length + ? "Follow an environment first" : busy === "trigger" ? "Dispatching…" - : "Trigger a run on every attached pipeline" + : "Trigger a run across the followed environments' pipelines" } > @@ -307,28 +321,28 @@ function AgentDetail() { No runs yet. )} - - {attachedPipelineDetails.length === 0 ? ( + + {followedEnvs.length === 0 ? ( - None attached.{" "} - - Create one + None followed.{" "} + + Manage environments . ) : ( - {attachedPipelineDetails.length} attached + {followedEnvs.map((e) => e.name).join(", ")} )}
- {agent.attachedPipelines?.length ? ( + {agent.environments?.length ? (

Pushes to{" "} {agent.name} route through - this agent’s pipelines (matched by branch). + the pipelines of the followed environments (matched by branch).

) : null} @@ -430,58 +444,66 @@ function AgentDetail() { - {/* Pipelines ----------------------------------------------------- */} + {/* Environments -------------------------------------------------- */} - Pipelines - {attachable.length > 0 ? ( + Environments followed + {followableEnvs.length > 0 ? ( + ) : environments.length === 0 ? ( + ) : ( )} - {attachedPipelineDetails.length === 0 ? ( + {followedEnvs.length === 0 ? (

- No pipelines attached. Click “Add pipeline” to bind one - (or create one in{" "} - - /flows/new + Not following any environment. Follow one to dispatch its + pipelines for this agent. Manage envs on the{" "} + + Environments page - ). + .

) : (
    - {attachedPipelineDetails.map((p) => ( + {followedEnvs.map((e) => (
  • - {p.name || "Untitled"} + {e.name}
    - {p.nodeCount} nodes · {formatDate(p.updatedAt)} + {(e.pipelineIds ?? []).length === 0 + ? "no pipelines" + : (e.pipelineIds ?? []) + .map(pipelineName) + .join(" → ")}
    @@ -490,23 +512,23 @@ function AgentDetail() {
)} - {showPipelinePicker && attachable.length > 0 && ( + {showEnvPicker && followableEnvs.length > 0 && (
- Attach a pipeline + Follow an environment
    - {attachable.map((p) => ( -
  • + {followableEnvs.map((e) => ( +
    • diff --git a/web/app/credentials/page.tsx b/web/app/credentials/page.tsx index 5f1a308..e2ff5be 100644 --- a/web/app/credentials/page.tsx +++ b/web/app/credentials/page.tsx @@ -48,7 +48,7 @@ export default function CredentialsPage() { } return ( -
    +

    Credentials

    diff --git a/web/app/environments/edit/page.tsx b/web/app/environments/edit/page.tsx new file mode 100644 index 0000000..4aaf1bc --- /dev/null +++ b/web/app/environments/edit/page.tsx @@ -0,0 +1,101 @@ +"use client"; + +import { Suspense, useEffect, useState } from "react"; +import { useRouter, useSearchParams } from "next/navigation"; +import Link from "next/link"; +import { ArrowLeft, Layers } from "lucide-react"; + +import { Button } from "@/components/ui/button"; +import { + Card, + CardContent, + CardDescription, + CardHeader, + CardTitle, +} from "@/components/ui/card"; +import { EnvironmentForm } from "@/components/environments/environment-form"; +import { api, type Environment } from "@/lib/api"; + +export default function EditEnvironmentPage() { + return ( + Loading…

    }> + + + ); +} + +function EditEnvironment() { + const router = useRouter(); + const params = useSearchParams(); + const name = params.get("name") ?? ""; + + const [env, setEnv] = useState(null); + const [error, setError] = useState(null); + + useEffect(() => { + if (!name) return; + api + .getEnvironment(name) + .then(setEnv) + .catch((err) => setError(err instanceof Error ? err.message : "load failed")); + }, [name]); + + if (!name) { + return ( +
    + Missing name query param. +
    + ); + } + + return ( +
    + + +
    +
    + + Edit environment +
    +

    {name}

    +

    + Update the description. Pipelines and their order are managed on the + environments list. Name is immutable. +

    +
    + + {error && ( +

    + {error} +

    + )} + + + + Configuration + Name + description. + + + {env === null ? ( +

    Loading…

    + ) : ( + router.push("/environments")} + onSubmit={async (body) => { + await api.updateEnvironment(name, body); + router.push("/environments"); + }} + onError={setError} + /> + )} +     +     +
    + ); +} diff --git a/web/app/environments/new/page.tsx b/web/app/environments/new/page.tsx new file mode 100644 index 0000000..edd11ed --- /dev/null +++ b/web/app/environments/new/page.tsx @@ -0,0 +1,71 @@ +"use client"; + +import { useState } from "react"; +import { useRouter } from "next/navigation"; +import Link from "next/link"; +import { ArrowLeft, Layers } from "lucide-react"; + +import { Button } from "@/components/ui/button"; +import { + Card, + CardContent, + CardDescription, + CardHeader, + CardTitle, +} from "@/components/ui/card"; +import { EnvironmentForm } from "@/components/environments/environment-form"; +import { api } from "@/lib/api"; + +export default function NewEnvironmentPage() { + const router = useRouter(); + const [error, setError] = useState(null); + + return ( +
    + + +
    +
    + + New environment +
    +

    New environment

    +

    + A global deploy stage. Add pipelines to it from the environments + list; agents follow this environment and run its pipelines. +

    +
    + + {error && ( +

    + {error} +

    + )} + + + + Configuration + + Name is how this env is referenced. Description is free-text. + + + + router.push("/environments")} + onSubmit={async (body) => { + await api.createEnvironment(body); + router.push("/environments"); + }} + onError={setError} + /> + + +
    + ); +} diff --git a/web/app/environments/page.tsx b/web/app/environments/page.tsx index edba132..ae2624b 100644 --- a/web/app/environments/page.tsx +++ b/web/app/environments/page.tsx @@ -1,122 +1,268 @@ "use client"; -import { Layers, Lock, ScanFace, ShieldCheck } from "lucide-react"; +import { useEffect, useState } from "react"; +import Link from "next/link"; +import { ArrowDown, ArrowUp, Layers, Plus } from "lucide-react"; -// Static preview of the Environments concept. Not wired to storage yet — -// this page is the contract we show clients before the runtime work -// lands. When the storage / routing actually exists, replace the three -// demo tiles with live env records from the API. - -const ENVS: { name: string; description: string; accent: string }[] = [ - { - name: "DEV", - description: - "Auto-deploy on every push, smoke evals only, no approval gates.", - accent: "border-foreground/60", - }, - { - name: "STAGING", - description: - "Full eval suite, optional approval, canary or progressive rollout.", - accent: "border-amber-400/60", - }, - { - name: "PROD", - description: - "Strict policy gates, human approval, audit log, SLO-backed rollback.", - accent: "border-rose-400/60", - }, -]; +import { Button } from "@/components/ui/button"; +import { + Card, + CardContent, + CardDescription, + CardHeader, + CardTitle, +} from "@/components/ui/card"; +import { api, type Environment, type FlowSummary } from "@/lib/api"; export default function EnvironmentsPage() { + const [envs, setEnvs] = useState(null); + const [pipelines, setPipelines] = useState([]); + const [error, setError] = useState(null); + + async function refresh() { + setError(null); + try { + const [e, p] = await Promise.all([api.listEnvironments(), api.listFlows()]); + setEnvs(e); + setPipelines(p); + } catch (err) { + setError(err instanceof Error ? err.message : "load failed"); + } + } + + useEffect(() => { + refresh(); + }, []); + + async function handleDelete(name: string) { + if (!confirm(`Delete environment "${name}"? Agents following it will stop dispatching its pipelines.`)) return; + try { + await api.deleteEnvironment(name); + await refresh(); + } catch (e) { + setError(e instanceof Error ? e.message : "delete failed"); + } + } + + const pipelineName = (id: string) => + pipelines.find((p) => p.id === id)?.name ?? id; + return ( -
    -
    -
    - - Environments +
    +
    +
    +
    + + Environments +
    +

    Environments

    +

    + A deploy stage is a named, ordered list of pipelines — the + promotion sequence. Agents follow environments; + triggering an agent runs its followed envs’ pipelines (each + still gated by its Trigger node’s branch). +

    -

    Environments

    -

    - Each environment owns its runtime target, credentials, secrets, - scaling, and approval policy. Pipelines reference envs by name; - promotion moves an artifact from one env’s pipeline to the - next. + +

    + + {error && ( +

    + {error}

    -
    + )} -
    - {/* Three env tiles */} -
    - {ENVS.map((e) => ( -
    -
    - - {e.name} - - tier + {envs === null &&

    Loading…

    } + {envs?.length === 0 && ( +

    + No environments yet. Create dev,{" "} + staging, and{" "} + prod to get started. +

    + )} + +
    + {envs?.map((env) => ( + + +
    + {env.name} + {env.description && {env.description}}
    -

    {e.description}

    -
    - ))} -
    - - {/* Concept rows */} -
    - - - -
    - - {/* Footer */} -
    - -
    +
    + + +
    + + + + + + ))}
    ); } -function ConceptRow({ - icon: Icon, - title, - body, +// ─── pipelines-in-env sub-component ───────────────────────────────────────── + +function PipelinesInEnv({ + env, + allPipelines, + pipelineName, + onChanged, + onError, }: { - icon: React.ComponentType<{ className?: string }>; - title: string; - body: string; + env: Environment; + allPipelines: FlowSummary[]; + pipelineName: (id: string) => string; + onChanged: () => void | Promise; + onError: (m: string | null) => void; }) { + const [picking, setPicking] = useState(false); + const ids = env.pipelineIds ?? []; + const inEnv = new Set(ids); + const available = allPipelines.filter((p) => !inEnv.has(p.id)); + + async function reorder(next: string[]) { + try { + await api.reorderEnvPipelines(env.name, next); + await onChanged(); + } catch (e) { + onError(e instanceof Error ? e.message : "reorder failed"); + } + } + function move(i: number, dir: -1 | 1) { + const j = i + dir; + if (j < 0 || j >= ids.length) return; + const next = ids.slice(); + [next[i], next[j]] = [next[j], next[i]]; + reorder(next); + } + return ( -
    -
    - -
    -
    -
    {title}
    -

    {body}

    +
    +
    +
    + Pipelines in this environment (promotion order) +
    +
    + + {ids.length === 0 && ( +

    + No pipelines yet. Build one on the{" "} + Pipelines page and add it + here. +

    + )} + + {ids.length > 0 && ( +
      + {ids.map((pid, i) => ( +
    • + + {i + 1} + + + {pipelineName(pid)} + +
      + + + +
      +
      • + ))} +
    + )} + + {picking && ( +
    + {available.length === 0 ? ( +

    + All pipelines are already in this environment. +

    + ) : ( +
      + {available.map((p) => ( +
    • + {p.name} + +
      • + ))} +
    + )} +
    + )}
    ); } diff --git a/web/app/gateway/page.tsx b/web/app/gateway/page.tsx index 7afd023..b8476d1 100644 --- a/web/app/gateway/page.tsx +++ b/web/app/gateway/page.tsx @@ -9,7 +9,7 @@ import { Activity, ArrowLeftRight, Coins, Gauge, Radio } from "lucide-react"; export default function GatewayPage() { return ( -
    +
    diff --git a/web/components/canvas/node-form.tsx b/web/components/canvas/node-form.tsx index c719462..c02ee3a 100644 --- a/web/components/canvas/node-form.tsx +++ b/web/components/canvas/node-form.tsx @@ -486,6 +486,9 @@ function ApprovalForm({ node, onChange }: NodeFormProps) { const reason = getString(node, "reason", "Manual review"); const reviewers = getStringArray(node, "reviewers"); const text = reviewers.join("\n"); + const method = getString(node, "method", ""); + const minApprovers = getNumber(node, "minApprovers", 0); + const timeout = getNumber(node, "timeoutSeconds", 0); function commit(t: string) { const parsed = t.split("\n").map((s) => s.trim()).filter(Boolean); @@ -518,6 +521,58 @@ function ApprovalForm({ node, onChange }: NodeFormProps) { placeholder="user@example.com" />
    + +
    + + +

    + auto emits straight to the approved output without + pausing. Quorum N>1 enforcement is surfaced to reviewers but not + yet hard-enforced. +

    +
    + + {method === "quorum" && ( +
    + + + onChange(setParam(node, "minApprovers", Number(e.target.value))) + } + placeholder="(inherit)" + className="font-mono text-xs" + /> +
    + )} + +
    + + + onChange(setParam(node, "timeoutSeconds", Number(e.target.value))) + } + className="font-mono text-xs" + /> +

    + When set, the run is auto-rejected after this long (routed to the + rejected output). +

    +
    ); } @@ -1127,8 +1182,8 @@ function TargetRow({ } function DeployForm({ node, onChange }: NodeFormProps) { - const target = getString(node, "target", "agentcore"); - const credentialName = getString(node, "credentialName", "aws"); + const target = getString(node, "target", ""); + const credentialName = getString(node, "credentialName", ""); const runtimeName = getString(node, "runtimeName", ""); const image = getString(node, "image", ""); const timeout = getNumber(node, "timeoutSeconds", 600); @@ -1156,6 +1211,7 @@ function DeployForm({ node, onChange }: NodeFormProps) { onChange={(e) => onChange(setParam(node, "target", e.target.value))} className="h-9 w-full rounded-md border border-input bg-transparent px-2 text-sm" > +

    - AgentCore deploys the upstream Push image. AWS region + account + - cross-account role come from the named credential below. The deploy - summary will include the public invoke URL. + AgentCore deploys the upstream Push image; the deploy summary + includes the public invoke URL.

    @@ -1183,8 +1238,8 @@ function DeployForm({ node, onChange }: NodeFormProps) { className="font-mono text-xs" />

    - Must match a credential of type aws on the Agent - (Credentials section on the agent page). + Defaults to aws. Must match a credential of type{" "} + aws in the global pool or an agent override.

    diff --git a/web/components/environments/environment-form.tsx b/web/components/environments/environment-form.tsx new file mode 100644 index 0000000..33651cc --- /dev/null +++ b/web/components/environments/environment-form.tsx @@ -0,0 +1,85 @@ +"use client"; + +import { useState } from "react"; + +import { Button } from "@/components/ui/button"; +import { Input } from "@/components/ui/input"; +import { Label } from "@/components/ui/label"; +import type { Environment, EnvironmentBody } from "@/lib/api"; + +// Shared environment create/edit form — name + description only. Per-deploy +// config (credential / runtime target / approval method) lives on the nodes, +// not the environment. Pipelines are managed on the environments list page. +// +// Used by /environments/new and /environments/edit as the full-page body. + +export type EnvironmentFormProps = { + initial: Environment | null; + onSubmit: (body: EnvironmentBody) => Promise; + onCancel: () => void; + onError: (m: string | null) => void; +}; + +export function EnvironmentForm({ initial, onSubmit, onCancel, onError }: EnvironmentFormProps) { + const isEdit = !!initial; + const [name, setName] = useState(initial?.name ?? ""); + const [description, setDescription] = useState(initial?.description ?? ""); + const [saving, setSaving] = useState(false); + + async function handleSave() { + onError(null); + setSaving(true); + try { + const trimmed = name.trim(); + if (!trimmed) throw new Error("name is required"); + await onSubmit({ name: trimmed, description: description.trim() || undefined }); + } catch (e) { + onError(e instanceof Error ? e.message : "save failed"); + } finally { + setSaving(false); + } + } + + return ( +
    +
    + + setName(e.target.value)} + placeholder="dev" + disabled={isEdit} + className="font-mono" + /> + {isEdit && ( +

    + Name is immutable. Delete + re-create to rename. +

    + )} +
    + +
    + + setDescription(e.target.value)} + placeholder="Auto-deploy on push, smoke evals only" + /> +

    + Pipelines and their order are managed on the environments list. +

    +
    + +
    + + +
    +
    + ); +} diff --git a/web/lib/api.ts b/web/lib/api.ts index 757af30..6b0e444 100644 --- a/web/lib/api.ts +++ b/web/lib/api.ts @@ -54,6 +54,24 @@ export type CredentialBody = { kv?: Record; }; +// An Environment is a sequencing container: a name + description + an +// ordered list of pipeline IDs (the promotion sequence, reorderable). +// Per-deploy config (credential / runtime target / approval method) +// lives on the nodes, not the env. +export type Environment = { + id: string; + name: string; + description?: string; + pipelineIds?: string[]; + createdAt: string; + updatedAt: string; +}; + +export type EnvironmentBody = { + name: string; + description?: string; +}; + export type Agent = { id: string; name: string; @@ -66,7 +84,7 @@ export type Agent = { webhookInstalledAt?: string; authStatus?: AuthStatus; authCheckedAt?: string; - attachedPipelines?: string[]; + environments?: string[]; credentials?: PublicCredential[]; createdAt: string; updatedAt: string; @@ -196,13 +214,13 @@ export const api = { uninstallAgentWebhook: (id: string) => fetch(`${base}/api/agents/${id}/webhook`, { method: "DELETE" }).then(handle), - attachPipeline: (id: string, pipelineId: string) => - fetch(`${base}/api/agents/${id}/pipelines/${pipelineId}`, { + agentFollowEnv: (id: string, envName: string) => + fetch(`${base}/api/agents/${id}/environments/${encodeURIComponent(envName)}`, { method: "POST", }).then(handle), - detachPipeline: (id: string, pipelineId: string) => - fetch(`${base}/api/agents/${id}/pipelines/${pipelineId}`, { + agentUnfollowEnv: (id: string, envName: string) => + fetch(`${base}/api/agents/${id}/environments/${encodeURIComponent(envName)}`, { method: "DELETE", }).then(handle), @@ -210,10 +228,60 @@ export const api = { fetch(`${base}/api/agents/${id}/trigger`, { method: "POST" }).then( handle<{ executionIds: string[]; - failures?: { pipelineId: string; reason: string; error?: string }[]; + failures?: { + environment?: string; + pipelineId?: string; + reason: string; + error?: string; + }[]; }> ), + // ── Environments ───────────────────────────────────────────────────── + listEnvironments: () => + fetch(`${base}/api/environments`).then(handle), + + getEnvironment: (name: string) => + fetch(`${base}/api/environments/${encodeURIComponent(name)}`).then( + handle + ), + + createEnvironment: (body: EnvironmentBody) => + fetch(`${base}/api/environments`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify(body), + }).then(handle), + + updateEnvironment: (name: string, body: EnvironmentBody) => + fetch(`${base}/api/environments/${encodeURIComponent(name)}`, { + method: "PUT", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify(body), + }).then(handle), + + deleteEnvironment: (name: string) => + fetch(`${base}/api/environments/${encodeURIComponent(name)}`, { + method: "DELETE", + }).then(handle), + + envAddPipeline: (envName: string, pipelineId: string) => + fetch(`${base}/api/environments/${encodeURIComponent(envName)}/pipelines/${pipelineId}`, { + method: "POST", + }).then(handle), + + envRemovePipeline: (envName: string, pipelineId: string) => + fetch(`${base}/api/environments/${encodeURIComponent(envName)}/pipelines/${pipelineId}`, { + method: "DELETE", + }).then(handle), + + reorderEnvPipelines: (envName: string, pipelineIds: string[]) => + fetch(`${base}/api/environments/${encodeURIComponent(envName)}/pipelines`, { + method: "PUT", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ pipelineIds }), + }).then(handle), + listCredentials: (agentId: string) => fetch(`${base}/api/agents/${agentId}/credentials`).then( handle diff --git a/web/lib/node-catalog.ts b/web/lib/node-catalog.ts index 16aa9bb..ecd9668 100644 --- a/web/lib/node-catalog.ts +++ b/web/lib/node-catalog.ts @@ -216,8 +216,8 @@ export const CATALOG: CatalogEntry[] = [ color: "bg-rose-500", outputs: 1, defaults: { - target: "agentcore", - credentialName: "aws", + target: "", + credentialName: "", runtimeName: "", image: "", envVars: {},