From ad35b32f5becd4842600bc5009f9b20bacf99310 Mon Sep 17 00:00:00 2001 From: mrhid6 Date: Mon, 20 Jul 2026 11:05:43 +0100 Subject: [PATCH] docs: add Server Workflows implementation plan --- .../plans/2026-07-20-server-workflows.md | 1823 +++++++++++++++++ 1 file changed, 1823 insertions(+) create mode 100644 docs/superpowers/plans/2026-07-20-server-workflows.md diff --git a/docs/superpowers/plans/2026-07-20-server-workflows.md b/docs/superpowers/plans/2026-07-20-server-workflows.md new file mode 100644 index 0000000..2d959cc --- /dev/null +++ b/docs/superpowers/plans/2026-07-20-server-workflows.md @@ -0,0 +1,1823 @@ +# Server Workflows Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Let operators compose reusable Bash/PowerShell steps into workflows and run them across many servers in parallel, passing data between steps via `$WORKFLOW_ENV`, with full run history and secret injection. + +**Architecture:** Extends the existing bidirectional `CommandStream` gRPC infra. A new `RunStepCmd` command is pushed to agents; agents exec the script with a `$WORKFLOW_ENV` file and reply with a new `StepResult` (stdout/stderr/exit/output_env). The server runner fans out one goroutine per target server (parallel), runs steps serially per server, merges output env forward, and applies per-step failure policy. A pending-result registry correlates `StepResult` back to the awaiting runner by `command_id`. + +**Tech Stack:** Go (gin, mongo-driver v2), hand-written JSON-codec gRPC structs (no protoc), Next.js 16 app-router + react-query + Tailwind, MongoDB. + +## Global Constraints + +- **No tests this iteration** — do not write `*_test.go` or frontend tests. Verify each task with `go build ./...`, `go vet ./...`, and (frontend) `npm run build`. +- gRPC uses a **JSON codec** — proto messages are hand-written Go structs in **two** files that must stay identical: `server/internal/grpc/pb/vantage.pb.go` and `agent/internal/grpc/pb/vantage.pb.go`. There is no codegen step. Also update `proto/vantage/v1/vantage.proto` as documentation. +- Mongo access pattern: `db.Col("collection_name")` with `context.WithTimeout`. Follow `server/internal/services/secrets.go`. +- Audit every mutation with `services.LogEvent(action, actor, serverID, targetID, message)`. +- REST handlers: gin, JSON, register under the session-authed `apiGroup` in `server/internal/api/handlers.go` (or a new `RegisterWorkflowRoutes(apiGroup)` called from there). Actor via `actorFromCtx(c)`. +- Frontend: use `@/lib/api` typed client, `@/components/ui` (`Button`, `Card`, `Table`/`Thead`/`Tbody`/`Tr`/`Th`/`Td`), Tailwind tokens (`text-primary`, `text-secondary`, `surface`, `surface-2`, `accent`, `border`, `danger`), react-query for data. +- Secret values must never be written into persisted run logs (`stdout`/`stderr`/`run_env`). Mask by literal replacement before persisting. +- Interpreter values are the literals `"bash"` and `"powershell"` everywhere. +- Go module path: `github.com/mrhid6/vantage`. + +--- + +## Task 1: Proto/pb structs — RunStepCmd + StepResult + +**Files:** +- Modify: `proto/vantage/v1/vantage.proto` +- Modify: `server/internal/grpc/pb/vantage.pb.go` +- Modify: `agent/internal/grpc/pb/vantage.pb.go` + +**Interfaces:** +- Produces: `pb.RunStepCmd{Interpreter string, Script string, Env map[string]string, TimeoutSeconds int}`, `pb.StepResult{CommandId string, ExitCode int, Stdout string, Stderr string, OutputEnv map[string]string}`. `pb.ServerCommand` gains field `RunStep *RunStepCmd`. `pb.AgentMessage` gains field `StepResult *StepResult`. + +- [ ] **Step 1: Document in the proto file** + +In `proto/vantage/v1/vantage.proto`, add to the `ServerCommand` oneof: `RunStepCmd run_step = 6;`. Add to the `AgentMessage` oneof: `StepResult step_result = 5;`. Add the two messages: + +```protobuf +message RunStepCmd { + string interpreter = 1; // "bash" | "powershell" + string script = 2; + map env = 3; + int32 timeout_seconds = 4; +} + +message StepResult { + string command_id = 1; + int32 exit_code = 2; + string stdout = 3; + string stderr = 4; + map output_env = 5; +} +``` + +- [ ] **Step 2: Add structs to server pb file** + +In `server/internal/grpc/pb/vantage.pb.go`, add `RunStep` to `ServerCommand` and `StepResult` to `AgentMessage`, then add the two new structs: + +```go +// in type ServerCommand struct { ... } add: + RunStep *RunStepCmd `json:"run_step,omitempty"` + +// in type AgentMessage struct { ... } add: + StepResult *StepResult `json:"step_result,omitempty"` + +type RunStepCmd struct { + Interpreter string `json:"interpreter"` + Script string `json:"script"` + Env map[string]string `json:"env,omitempty"` + TimeoutSeconds int `json:"timeout_seconds,omitempty"` +} + +type StepResult struct { + CommandId string `json:"command_id"` + ExitCode int `json:"exit_code"` + Stdout string `json:"stdout,omitempty"` + Stderr string `json:"stderr,omitempty"` + OutputEnv map[string]string `json:"output_env,omitempty"` +} +``` + +- [ ] **Step 3: Mirror the exact same additions into the agent pb file** + +Apply the identical struct field additions and new types to `agent/internal/grpc/pb/vantage.pb.go`. + +- [ ] **Step 4: Verify build** + +Run: `cd server && go build ./... && cd ../agent && go build ./...` +Expected: both succeed, no errors. + +- [ ] **Step 5: Commit** + +```bash +git add proto/vantage/v1/vantage.proto server/internal/grpc/pb/vantage.pb.go agent/internal/grpc/pb/vantage.pb.go +git commit -m "feat(proto): add RunStepCmd and StepResult messages" +``` + +--- + +## Task 2: Pending-result registry (server correlation) + +**Files:** +- Create: `server/internal/services/stepresults.go` + +**Interfaces:** +- Consumes: `pb.StepResult` (Task 1). +- Produces: package-level `var StepResults *stepResultRegistry` with methods `Await(commandID string) <-chan *pb.StepResult`, `Cancel(commandID string)`, `Deliver(res *pb.StepResult)`. + +- [ ] **Step 1: Write the registry** + +```go +package services + +import ( + "sync" + + "github.com/mrhid6/vantage/server/internal/grpc/pb" +) + +type stepResultRegistry struct { + mu sync.Mutex + pending map[string]chan *pb.StepResult +} + +// StepResults correlates agent StepResult replies back to the workflow runner +// goroutine that dispatched the matching RunStepCmd, keyed by command_id. +var StepResults = &stepResultRegistry{pending: make(map[string]chan *pb.StepResult)} + +// Await registers interest in a command's result BEFORE the command is +// dispatched, and returns a buffered channel that receives the single result. +func (r *stepResultRegistry) Await(commandID string) <-chan *pb.StepResult { + ch := make(chan *pb.StepResult, 1) + r.mu.Lock() + r.pending[commandID] = ch + r.mu.Unlock() + return ch +} + +// Cancel removes a pending waiter (call on timeout to avoid leaks). +func (r *stepResultRegistry) Cancel(commandID string) { + r.mu.Lock() + delete(r.pending, commandID) + r.mu.Unlock() +} + +// Deliver routes an incoming StepResult to its waiter, if any. +func (r *stepResultRegistry) Deliver(res *pb.StepResult) { + if res == nil { + return + } + r.mu.Lock() + ch, ok := r.pending[res.CommandId] + if ok { + delete(r.pending, res.CommandId) + } + r.mu.Unlock() + if ok { + ch <- res + } +} +``` + +- [ ] **Step 2: Wire delivery into the CommandStream receive loop** + +In `server/internal/grpc/server.go`, inside the background `stream.Recv()` goroutine (around line 122-133), after the existing `if m.Result != nil { ... }` block, add: + +```go + if m.StepResult != nil { + services.StepResults.Deliver(m.StepResult) + } +``` + +- [ ] **Step 3: Verify build** + +Run: `cd server && go build ./... && go vet ./...` +Expected: success. + +- [ ] **Step 4: Commit** + +```bash +git add server/internal/services/stepresults.go server/internal/grpc/server.go +git commit -m "feat(server): add pending step-result registry and stream delivery" +``` + +--- + +## Task 3: Agent — execute RunStepCmd + +**Files:** +- Create: `agent/internal/exec/exec.go` +- Modify: the agent command-stream loop that handles `ServerCommand` (search: `cmd.GenerateKey != nil` / `cmd.UpdateAgent != nil`; likely `agent/internal/sync/sync.go` or `agent/internal/updates/updates.go`). + +**Interfaces:** +- Consumes: `pb.RunStepCmd` (Task 1). +- Produces: `exec.RunStep(cmd *pb.RunStepCmd) *pb.StepResult` — runs the script, returns populated result. The agent loop sends it back via the existing stream `Send(&pb.AgentMessage{ServerId, AgentToken, StepResult: res})`. + +- [ ] **Step 1: Write the executor** + +```go +package exec + +import ( + "bufio" + "bytes" + "context" + "os" + "os/exec" + "path/filepath" + "runtime" + "strings" + "time" + + "github.com/mrhid6/vantage/agent/internal/grpc/pb" +) + +// RunStep writes the script to a temp file, provides a WORKFLOW_ENV file for +// the script to append KEY=value output to, executes it under the requested +// interpreter, and returns captured output plus parsed output env. +func RunStep(cmd *pb.RunStepCmd) *pb.StepResult { + res := &pb.StepResult{CommandId: "", OutputEnv: map[string]string{}} + + dir, err := os.MkdirTemp("", "vantage-step-") + if err != nil { + res.ExitCode = 1 + res.Stderr = "create temp dir: " + err.Error() + return res + } + defer os.RemoveAll(dir) + + envFile := filepath.Join(dir, "workflow_env") + if err := os.WriteFile(envFile, nil, 0600); err != nil { + res.ExitCode = 1 + res.Stderr = "create env file: " + err.Error() + return res + } + + var scriptPath string + var c *exec.Cmd + timeout := time.Duration(cmd.TimeoutSeconds) * time.Second + if timeout <= 0 { + timeout = 30 * time.Minute + } + ctx, cancel := context.WithTimeout(context.Background(), timeout) + defer cancel() + + switch cmd.Interpreter { + case "powershell": + scriptPath = filepath.Join(dir, "step.ps1") + if err := os.WriteFile(scriptPath, []byte(cmd.Script), 0600); err != nil { + res.ExitCode = 1 + res.Stderr = err.Error() + return res + } + shell := "pwsh" + if runtime.GOOS == "windows" { + if _, err := exec.LookPath("pwsh"); err != nil { + shell = "powershell.exe" + } + } + c = exec.CommandContext(ctx, shell, "-NoProfile", "-NonInteractive", "-File", scriptPath) + default: // "bash" + scriptPath = filepath.Join(dir, "step.sh") + if err := os.WriteFile(scriptPath, []byte(cmd.Script), 0700); err != nil { + res.ExitCode = 1 + res.Stderr = err.Error() + return res + } + c = exec.CommandContext(ctx, "bash", scriptPath) + } + + c.Env = append(os.Environ(), "WORKFLOW_ENV="+envFile) + for k, v := range cmd.Env { + c.Env = append(c.Env, k+"="+v) + } + + var stdout, stderr bytes.Buffer + c.Stdout = &stdout + c.Stderr = &stderr + runErr := c.Run() + + res.Stdout = stdout.String() + res.Stderr = stderr.String() + if ctx.Err() == context.DeadlineExceeded { + res.ExitCode = 124 + res.Stderr += "\n[vantage] step timed out" + } else if ee, ok := runErr.(*exec.ExitError); ok { + res.ExitCode = ee.ExitCode() + } else if runErr != nil { + res.ExitCode = 1 + res.Stderr += "\n[vantage] " + runErr.Error() + } + + res.OutputEnv = parseEnvFile(envFile) + return res +} + +// parseEnvFile reads KEY=value lines (last write wins). Blank lines and lines +// without '=' are ignored. +func parseEnvFile(path string) map[string]string { + out := map[string]string{} + f, err := os.Open(path) + if err != nil { + return out + } + defer f.Close() + sc := bufio.NewScanner(f) + sc.Buffer(make([]byte, 0, 64*1024), 1024*1024) + for sc.Scan() { + line := sc.Text() + i := strings.IndexByte(line, '=') + if i <= 0 { + continue + } + out[line[:i]] = line[i+1:] + } + return out +} +``` + +- [ ] **Step 2: Handle the command in the agent loop** + +Find the agent's `ServerCommand` handling switch (where `cmd.GenerateKey`, `cmd.UpdateAgent`, `cmd.ApplyUpdates` are dispatched). Add a branch. `stream` is the `pb.Vantage_CommandStreamClient`; `serverID`/`agentToken` are in scope there (match how `AgentReady` was sent): + +```go + if cmd.RunStep != nil { + res := exec.RunStep(cmd.RunStep) + res.CommandId = cmd.CommandId + _ = stream.Send(&pb.AgentMessage{ + ServerId: serverID, + AgentToken: agentToken, + StepResult: res, + }) + continue + } +``` + +Add the import `"github.com/mrhid6/vantage/agent/internal/exec"`. + +- [ ] **Step 3: Verify build** + +Run: `cd agent && go build ./... && go vet ./...` +Expected: success. + +- [ ] **Step 4: Commit** + +```bash +git add agent/internal/exec/exec.go agent/internal/ +git commit -m "feat(agent): execute RunStepCmd with WORKFLOW_ENV capture" +``` + +--- + +## Task 4: Models — steps, workflows, runs + +**Files:** +- Create: `server/internal/models/workflow.go` + +**Interfaces:** +- Produces: structs `WorkflowStep`, `Workflow`, `WorkflowStepRef`, `WorkflowRun`, `ServerRun`, `StepRun` with bson+json tags matching spec §3. + +- [ ] **Step 1: Write the models** + +```go +package models + +import "time" + +type WorkflowStep struct { + ID string `bson:"_id,omitempty" json:"-"` + StepID string `bson:"step_id" json:"step_id"` + Name string `bson:"name" json:"name"` + Description string `bson:"description" json:"description"` + Interpreter string `bson:"interpreter" json:"interpreter"` // "bash" | "powershell" + Script string `bson:"script" json:"script"` + DeclaredOutputs []string `bson:"declared_outputs" json:"declared_outputs"` + SecretRefs []string `bson:"secret_refs" json:"secret_refs"` + CreatedAt time.Time `bson:"created_at" json:"created_at"` + UpdatedAt time.Time `bson:"updated_at" json:"updated_at"` +} + +type WorkflowStepRef struct { + StepID string `bson:"step_id" json:"step_id"` + Order int `bson:"order" json:"order"` + OnFailure string `bson:"on_failure" json:"on_failure"` // "stop" | "continue" | "retry" + MaxRetries int `bson:"max_retries" json:"max_retries"` + Overrides *StepOverride `bson:"overrides,omitempty" json:"overrides,omitempty"` +} + +type StepOverride struct { + Script *string `bson:"script,omitempty" json:"script,omitempty"` + SecretRefs []string `bson:"secret_refs,omitempty" json:"secret_refs,omitempty"` +} + +type Workflow struct { + ID string `bson:"_id,omitempty" json:"-"` + WorkflowID string `bson:"workflow_id" json:"workflow_id"` + Name string `bson:"name" json:"name"` + TargetServerIDs []string `bson:"target_server_ids" json:"target_server_ids"` + Steps []WorkflowStepRef `bson:"steps" json:"steps"` + CreatedAt time.Time `bson:"created_at" json:"created_at"` + UpdatedAt time.Time `bson:"updated_at" json:"updated_at"` +} + +// ResolvedStep is a step frozen into a run snapshot (library step + overrides applied). +type ResolvedStep struct { + Order int `bson:"order" json:"order"` + Name string `bson:"name" json:"name"` + Interpreter string `bson:"interpreter" json:"interpreter"` + Script string `bson:"script" json:"script"` + SecretRefs []string `bson:"secret_refs" json:"secret_refs"` + OnFailure string `bson:"on_failure" json:"on_failure"` + MaxRetries int `bson:"max_retries" json:"max_retries"` +} + +type StepRun struct { + Order int `bson:"order" json:"order"` + Name string `bson:"name" json:"name"` + Status string `bson:"status" json:"status"` // queued|running|success|failed|skipped + Attempts int `bson:"attempts" json:"attempts"` + ExitCode int `bson:"exit_code" json:"exit_code"` + Stdout string `bson:"stdout" json:"stdout"` + Stderr string `bson:"stderr" json:"stderr"` + OutputEnv map[string]string `bson:"output_env" json:"output_env"` + StartedAt *time.Time `bson:"started_at,omitempty" json:"started_at,omitempty"` + FinishedAt *time.Time `bson:"finished_at,omitempty" json:"finished_at,omitempty"` +} + +type ServerRun struct { + ServerID string `bson:"server_id" json:"server_id"` + Hostname string `bson:"hostname" json:"hostname"` + Status string `bson:"status" json:"status"` // queued|running|success|failed|skipped + StartedAt *time.Time `bson:"started_at,omitempty" json:"started_at,omitempty"` + FinishedAt *time.Time `bson:"finished_at,omitempty" json:"finished_at,omitempty"` + RunEnv map[string]string `bson:"run_env" json:"run_env"` + Steps []StepRun `bson:"steps" json:"steps"` +} + +type WorkflowRun struct { + ID string `bson:"_id,omitempty" json:"-"` + RunID string `bson:"run_id" json:"run_id"` + WorkflowID string `bson:"workflow_id" json:"workflow_id"` + Name string `bson:"name" json:"name"` + Steps []ResolvedStep `bson:"steps_snapshot" json:"steps_snapshot"` + Status string `bson:"status" json:"status"` // running|success|failed|cancelled + TriggeredBy string `bson:"triggered_by" json:"triggered_by"` + StartedAt time.Time `bson:"started_at" json:"started_at"` + FinishedAt *time.Time `bson:"finished_at,omitempty" json:"finished_at,omitempty"` + ServerRuns []ServerRun `bson:"server_runs" json:"server_runs"` +} +``` + +- [ ] **Step 2: Verify build** + +Run: `cd server && go build ./...` +Expected: success. + +- [ ] **Step 3: Commit** + +```bash +git add server/internal/models/workflow.go +git commit -m "feat(models): add workflow, step, and run models" +``` + +--- + +## Task 5: Step library + workflow CRUD services + +**Files:** +- Create: `server/internal/services/workflows.go` + +**Interfaces:** +- Consumes: models (Task 4), `db.Col`. +- Produces: + - `EnsureWorkflowIndexes() error` + - `ListSteps() ([]models.WorkflowStep, error)`, `CreateStep(s models.WorkflowStep) (*models.WorkflowStep, error)`, `UpdateStep(stepID string, s models.WorkflowStep) error`, `DeleteStep(stepID string) error` + - `ListWorkflows() ([]models.Workflow, error)`, `GetWorkflow(id string) (*models.Workflow, error)`, `CreateWorkflow(w models.Workflow) (*models.Workflow, error)`, `UpdateWorkflow(id string, w models.Workflow) error`, `DeleteWorkflow(id string) error` + +- [ ] **Step 1: Write CRUD service** + +```go +package services + +import ( + "context" + "fmt" + "time" + + "github.com/google/uuid" + "github.com/mrhid6/vantage/server/internal/db" + "github.com/mrhid6/vantage/server/internal/models" + "go.mongodb.org/mongo-driver/v2/bson" + "go.mongodb.org/mongo-driver/v2/mongo" + "go.mongodb.org/mongo-driver/v2/mongo/options" +) + +func wfCtx() (context.Context, context.CancelFunc) { + return context.WithTimeout(context.Background(), 10*time.Second) +} + +func EnsureWorkflowIndexes() error { + ctx, cancel := wfCtx() + defer cancel() + if _, err := db.Col("workflow_steps").Indexes().CreateOne(ctx, mongo.IndexModel{ + Keys: bson.D{{Key: "step_id", Value: 1}}, Options: options.Index().SetUnique(true), + }); err != nil { + return err + } + if _, err := db.Col("workflows").Indexes().CreateOne(ctx, mongo.IndexModel{ + Keys: bson.D{{Key: "workflow_id", Value: 1}}, Options: options.Index().SetUnique(true), + }); err != nil { + return err + } + _, err := db.Col("workflow_runs").Indexes().CreateOne(ctx, mongo.IndexModel{ + Keys: bson.D{{Key: "run_id", Value: 1}}, Options: options.Index().SetUnique(true), + }) + return err +} + +// ---- Steps ---- + +func ListSteps() ([]models.WorkflowStep, error) { + ctx, cancel := wfCtx() + defer cancel() + cur, err := db.Col("workflow_steps").Find(ctx, bson.M{}, + options.Find().SetSort(bson.D{{Key: "name", Value: 1}})) + if err != nil { + return nil, err + } + defer cur.Close(ctx) + steps := []models.WorkflowStep{} + if err := cur.All(ctx, &steps); err != nil { + return nil, err + } + return steps, nil +} + +func CreateStep(s models.WorkflowStep) (*models.WorkflowStep, error) { + ctx, cancel := wfCtx() + defer cancel() + s.StepID = uuid.New().String() + s.CreatedAt = time.Now() + s.UpdatedAt = s.CreatedAt + if s.DeclaredOutputs == nil { + s.DeclaredOutputs = []string{} + } + if s.SecretRefs == nil { + s.SecretRefs = []string{} + } + if _, err := db.Col("workflow_steps").InsertOne(ctx, s); err != nil { + return nil, err + } + return &s, nil +} + +func UpdateStep(stepID string, s models.WorkflowStep) error { + ctx, cancel := wfCtx() + defer cancel() + _, err := db.Col("workflow_steps").UpdateOne(ctx, bson.M{"step_id": stepID}, bson.M{"$set": bson.M{ + "name": s.Name, + "description": s.Description, + "interpreter": s.Interpreter, + "script": s.Script, + "declared_outputs": s.DeclaredOutputs, + "secret_refs": s.SecretRefs, + "updated_at": time.Now(), + }}) + return err +} + +func DeleteStep(stepID string) error { + ctx, cancel := wfCtx() + defer cancel() + _, err := db.Col("workflow_steps").DeleteOne(ctx, bson.M{"step_id": stepID}) + return err +} + +func getStep(ctx context.Context, stepID string) (*models.WorkflowStep, error) { + var s models.WorkflowStep + err := db.Col("workflow_steps").FindOne(ctx, bson.M{"step_id": stepID}).Decode(&s) + if err == mongo.ErrNoDocuments { + return nil, fmt.Errorf("step %s not found", stepID) + } + return &s, err +} + +// ---- Workflows ---- + +func ListWorkflows() ([]models.Workflow, error) { + ctx, cancel := wfCtx() + defer cancel() + cur, err := db.Col("workflows").Find(ctx, bson.M{}, + options.Find().SetSort(bson.D{{Key: "name", Value: 1}})) + if err != nil { + return nil, err + } + defer cur.Close(ctx) + wfs := []models.Workflow{} + if err := cur.All(ctx, &wfs); err != nil { + return nil, err + } + return wfs, nil +} + +func GetWorkflow(id string) (*models.Workflow, error) { + ctx, cancel := wfCtx() + defer cancel() + var w models.Workflow + err := db.Col("workflows").FindOne(ctx, bson.M{"workflow_id": id}).Decode(&w) + if err == mongo.ErrNoDocuments { + return nil, fmt.Errorf("workflow not found") + } + return &w, err +} + +func CreateWorkflow(w models.Workflow) (*models.Workflow, error) { + ctx, cancel := wfCtx() + defer cancel() + w.WorkflowID = uuid.New().String() + w.CreatedAt = time.Now() + w.UpdatedAt = w.CreatedAt + if w.TargetServerIDs == nil { + w.TargetServerIDs = []string{} + } + if w.Steps == nil { + w.Steps = []models.WorkflowStepRef{} + } + if _, err := db.Col("workflows").InsertOne(ctx, w); err != nil { + return nil, err + } + return &w, nil +} + +func UpdateWorkflow(id string, w models.Workflow) error { + ctx, cancel := wfCtx() + defer cancel() + _, err := db.Col("workflows").UpdateOne(ctx, bson.M{"workflow_id": id}, bson.M{"$set": bson.M{ + "name": w.Name, + "target_server_ids": w.TargetServerIDs, + "steps": w.Steps, + "updated_at": time.Now(), + }}) + return err +} + +func DeleteWorkflow(id string) error { + ctx, cancel := wfCtx() + defer cancel() + _, err := db.Col("workflows").DeleteOne(ctx, bson.M{"workflow_id": id}) + return err +} +``` + +- [ ] **Step 2: Register indexes at startup** + +Find where `EnsureSecretIndexes()` is called (search `EnsureSecretIndexes` in `server/cmd/main.go`) and add `EnsureWorkflowIndexes()` alongside it with the same error handling. + +- [ ] **Step 3: Verify build** + +Run: `cd server && go build ./... && go vet ./...` +Expected: success. + +- [ ] **Step 4: Commit** + +```bash +git add server/internal/services/workflows.go server/cmd/main.go +git commit -m "feat(server): step library and workflow CRUD services" +``` + +--- + +## Task 6: Workflow runner (orchestration) + +**Files:** +- Create: `server/internal/services/workflow_runner.go` + +**Interfaces:** +- Consumes: `Dispatcher` (dispatch.go), `StepResults` (Task 2), `GetSecretGroupDecrypted`/secrets, `getStep` (Task 5), models (Task 4), `pb`. +- Produces: `TriggerWorkflow(workflowID, actor string) (string, error)` returning the new `run_id`; `GetRun(runID string) (*models.WorkflowRun, error)`; `ListRuns(workflowID string, limit int64) ([]models.WorkflowRun, error)`; `CancelRun(runID string) error`. + +Notes: +- The dispatcher is fire-and-forget; add a small dispatch helper that pushes a `ServerCommand{RunStep}` for a given server. Reuse `Dispatcher` via a new exported method or replicate the `dispatch` pattern. Add to `dispatch.go`: + +```go +// DispatchRunStep pushes a RunStepCmd to a server's agent. Caller must have +// registered StepResults.Await(commandID) first. +func DispatchRunStep(serverID, commandID string, cmd *pb.RunStepCmd) error { + return Dispatcher.dispatch(serverID, &pb.ServerCommand{CommandId: commandID, RunStep: cmd}) +} +``` + +- [ ] **Step 1: Add `DispatchRunStep` to `dispatch.go`** + +Add the function above to `server/internal/services/dispatch.go` (it needs no new imports; `pb` is already imported). + +- [ ] **Step 2: Write the runner** + +```go +package services + +import ( + "context" + "fmt" + "strings" + "time" + + "github.com/google/uuid" + "github.com/mrhid6/vantage/server/internal/db" + "github.com/mrhid6/vantage/server/internal/grpc/pb" + "github.com/mrhid6/vantage/server/internal/models" + "go.mongodb.org/mongo-driver/v2/bson" + "go.mongodb.org/mongo-driver/v2/mongo" + "go.mongodb.org/mongo-driver/v2/mongo/options" +) + +const stepDispatchGrace = 15 * time.Second + +// TriggerWorkflow snapshots the workflow, creates a run doc, and starts a +// background goroutine per target server (parallel fan-out). Returns run_id. +func TriggerWorkflow(workflowID, actor string) (string, error) { + wf, err := GetWorkflow(workflowID) + if err != nil { + return "", err + } + if len(wf.TargetServerIDs) == 0 { + return "", fmt.Errorf("workflow has no target servers") + } + if len(wf.Steps) == 0 { + return "", fmt.Errorf("workflow has no steps") + } + + // Reject a concurrent run of the same workflow. + ctx, cancel := wfCtx() + running := db.Col("workflow_runs").FindOne(ctx, bson.M{"workflow_id": workflowID, "status": "running"}) + cancel() + if running.Err() == nil { + return "", fmt.Errorf("workflow already has a run in progress") + } + + resolved, err := resolveSteps(wf) + if err != nil { + return "", err + } + + run := models.WorkflowRun{ + RunID: uuid.New().String(), + WorkflowID: workflowID, + Name: wf.Name, + Steps: resolved, + Status: "running", + TriggeredBy: actor, + StartedAt: time.Now(), + ServerRuns: make([]models.ServerRun, 0, len(wf.TargetServerIDs)), + } + for _, sid := range wf.TargetServerIDs { + hostname := sid + if s, e := GetServer(sid); e == nil { + hostname = s.Hostname + } + sr := models.ServerRun{ServerID: sid, Hostname: hostname, Status: "queued", RunEnv: map[string]string{}} + for _, rs := range resolved { + sr.Steps = append(sr.Steps, models.StepRun{Order: rs.Order, Name: rs.Name, Status: "queued", OutputEnv: map[string]string{}}) + } + run.ServerRuns = append(run.ServerRuns, sr) + } + + ictx, icancel := wfCtx() + defer icancel() + if _, err := db.Col("workflow_runs").InsertOne(ictx, run); err != nil { + return "", err + } + + go executeRun(run.RunID) + return run.RunID, nil +} + +// resolveSteps freezes each workflow step ref into a ResolvedStep by loading the +// library step and applying overrides. +func resolveSteps(wf *models.Workflow) ([]models.ResolvedStep, error) { + ctx, cancel := wfCtx() + defer cancel() + out := make([]models.ResolvedStep, 0, len(wf.Steps)) + for _, ref := range wf.Steps { + lib, err := getStep(ctx, ref.StepID) + if err != nil { + return nil, err + } + rs := models.ResolvedStep{ + Order: ref.Order, + Name: lib.Name, + Interpreter: lib.Interpreter, + Script: lib.Script, + SecretRefs: lib.SecretRefs, + OnFailure: ref.OnFailure, + MaxRetries: ref.MaxRetries, + } + if ref.Overrides != nil { + if ref.Overrides.Script != nil { + rs.Script = *ref.Overrides.Script + } + if ref.Overrides.SecretRefs != nil { + rs.SecretRefs = ref.Overrides.SecretRefs + } + } + if rs.OnFailure == "" { + rs.OnFailure = "stop" + } + out = append(out, rs) + } + return out, nil +} + +// executeRun fans out one goroutine per server run and waits for all to finish. +func executeRun(runID string) { + run, err := GetRun(runID) + if err != nil { + return + } + done := make(chan int, len(run.ServerRuns)) + for i := range run.ServerRuns { + go func(idx int) { + runServer(runID, idx, run.Steps, run.ServerRuns[idx].ServerID) + done <- idx + }(i) + } + for range run.ServerRuns { + <-done + } + + // Aggregate status. + final, _ := GetRun(runID) + status := "success" + for _, sr := range final.ServerRuns { + if sr.Status == "failed" { + status = "failed" + } + } + now := time.Now() + ctx, cancel := wfCtx() + defer cancel() + _, _ = db.Col("workflow_runs").UpdateOne(ctx, bson.M{"run_id": runID}, + bson.M{"$set": bson.M{"status": status, "finished_at": now}}) +} + +// runServer executes the resolved steps sequentially on one server, threading +// output env forward and applying per-step failure policy. +func runServer(runID string, srvIdx int, steps []models.ResolvedStep, serverID string) { + now := time.Now() + setServerRun(runID, srvIdx, bson.M{"server_runs.$.status": "running", "server_runs.$.started_at": now}) + + if !Dispatcher.IsConnected(serverID) { + fin := time.Now() + setServerRun(runID, srvIdx, bson.M{"server_runs.$.status": "skipped", "server_runs.$.finished_at": fin}) + return + } + + runEnv := map[string]string{} + serverFailed := false + + for i, step := range steps { + startStep(runID, serverID, i, "running") + var res *pb.StepResult + attempts := 0 + maxAttempts := 1 + if step.OnFailure == "retry" { + maxAttempts = step.MaxRetries + 1 + } + + // Merge secrets into command env (kept out of persisted logs). + secretVals := resolveSecrets(step.SecretRefs) + cmdEnv := map[string]string{} + for k, v := range runEnv { + cmdEnv[k] = v + } + for k, v := range secretVals { + cmdEnv[k] = v + } + + for attempts < maxAttempts { + attempts++ + res = dispatchAndWait(serverID, &pb.RunStepCmd{ + Interpreter: step.Interpreter, + Script: step.Script, + Env: cmdEnv, + TimeoutSeconds: 0, + }) + if res != nil && res.ExitCode == 0 { + break + } + } + + // Mask secret values before persisting. + stdout, stderr := "", "" + exit := 1 + outEnv := map[string]string{} + if res != nil { + stdout = maskSecrets(res.Stdout, secretVals) + stderr = maskSecrets(res.Stderr, secretVals) + exit = res.ExitCode + for k, v := range res.OutputEnv { + outEnv[k] = v + runEnv[k] = v // implicit: all outputs flow to all later steps + } + } else { + stderr = "[vantage] agent did not return a result" + } + + status := "success" + if exit != 0 { + status = "failed" + } + finishStep(runID, serverID, i, status, attempts, exit, stdout, stderr, outEnv) + + if exit != 0 { + switch step.OnFailure { + case "continue": + // keep going + default: // "stop" or exhausted "retry" + serverFailed = true + } + if serverFailed { + markRemainingSkipped(runID, serverID, i+1) + break + } + } + } + + fin := time.Now() + status := "success" + if serverFailed { + status = "failed" + } + setServerRun(runID, srvIdx, bson.M{ + "server_runs.$.status": status, + "server_runs.$.finished_at": fin, + "server_runs.$.run_env": runEnv, + }) +} + +// dispatchAndWait registers a waiter, dispatches the step, and blocks for the +// result or a timeout. +func dispatchAndWait(serverID string, cmd *pb.RunStepCmd) *pb.StepResult { + commandID := uuid.New().String() + ch := StepResults.Await(commandID) + if err := DispatchRunStep(serverID, commandID, cmd); err != nil { + StepResults.Cancel(commandID) + return &pb.StepResult{ExitCode: 1, Stderr: "[vantage] dispatch failed: " + err.Error()} + } + wait := time.Duration(cmd.TimeoutSeconds)*time.Second + stepDispatchGrace + if cmd.TimeoutSeconds == 0 { + wait = 30*time.Minute + stepDispatchGrace + } + select { + case res := <-ch: + return res + case <-time.After(wait): + StepResults.Cancel(commandID) + return &pb.StepResult{ExitCode: 124, Stderr: "[vantage] timed out waiting for agent result"} + } +} + +func resolveSecrets(refs []string) map[string]string { + out := map[string]string{} + for _, ref := range refs { + // ref format "group/KEY"; resolve via RevealSecret. + parts := strings.SplitN(ref, "/", 2) + if len(parts) != 2 { + continue + } + if v, err := RevealSecret(parts[0], parts[1]); err == nil { + out[parts[1]] = v + } + } + return out +} + +func maskSecrets(s string, secrets map[string]string) string { + for _, v := range secrets { + if v == "" { + continue + } + s = strings.ReplaceAll(s, v, "***") + } + return s +} + +// ---- run doc mutation helpers ---- + +func setServerRun(runID string, srvIdx int, set bson.M) { + ctx, cancel := wfCtx() + defer cancel() + _, _ = db.Col("workflow_runs").UpdateOne(ctx, + bson.M{"run_id": runID, "server_runs.server_id": serverIDAt(runID, srvIdx)}, + bson.M{"$set": set}) +} + +// serverIDAt returns the server_id at an index (positional operator needs a match). +func serverIDAt(runID string, srvIdx int) string { + r, err := GetRun(runID) + if err != nil || srvIdx >= len(r.ServerRuns) { + return "" + } + return r.ServerRuns[srvIdx].ServerID +} + +func startStep(runID, serverID string, order int, status string) { + now := time.Now() + updateStep(runID, serverID, order, bson.M{ + "server_runs.$[s].steps.$[t].status": status, + "server_runs.$[s].steps.$[t].started_at": now, + }) +} + +func finishStep(runID, serverID string, order int, status string, attempts, exit int, stdout, stderr string, outEnv map[string]string) { + now := time.Now() + updateStep(runID, serverID, order, bson.M{ + "server_runs.$[s].steps.$[t].status": status, + "server_runs.$[s].steps.$[t].attempts": attempts, + "server_runs.$[s].steps.$[t].exit_code": exit, + "server_runs.$[s].steps.$[t].stdout": stdout, + "server_runs.$[s].steps.$[t].stderr": stderr, + "server_runs.$[s].steps.$[t].output_env": outEnv, + "server_runs.$[s].steps.$[t].finished_at": now, + }) +} + +func markRemainingSkipped(runID, serverID string, fromOrder int) { + ctx, cancel := wfCtx() + defer cancel() + _, _ = db.Col("workflow_runs").UpdateMany(ctx, + bson.M{"run_id": runID}, + bson.M{"$set": bson.M{"server_runs.$[s].steps.$[t].status": "skipped"}}, + options.UpdateMany().SetArrayFilters([]interface{}{ + bson.M{"s.server_id": serverID}, + bson.M{"t.order": bson.M{"$gte": fromOrder}, "t.status": "queued"}, + }), + ) +} + +func updateStep(runID, serverID string, order int, set bson.M) { + ctx, cancel := wfCtx() + defer cancel() + _, _ = db.Col("workflow_runs").UpdateOne(ctx, + bson.M{"run_id": runID}, + bson.M{"$set": set}, + options.UpdateOne().SetArrayFilters([]interface{}{ + bson.M{"s.server_id": serverID}, + bson.M{"t.order": order}, + }), + ) +} + +// ---- reads ---- + +func GetRun(runID string) (*models.WorkflowRun, error) { + ctx, cancel := wfCtx() + defer cancel() + var r models.WorkflowRun + err := db.Col("workflow_runs").FindOne(ctx, bson.M{"run_id": runID}).Decode(&r) + if err == mongo.ErrNoDocuments { + return nil, fmt.Errorf("run not found") + } + return &r, err +} + +func ListRuns(workflowID string, limit int64) ([]models.WorkflowRun, error) { + ctx, cancel := wfCtx() + defer cancel() + cur, err := db.Col("workflow_runs").Find(ctx, bson.M{"workflow_id": workflowID}, + options.Find().SetSort(bson.D{{Key: "started_at", Value: -1}}).SetLimit(limit)) + if err != nil { + return nil, err + } + defer cur.Close(ctx) + runs := []models.WorkflowRun{} + if err := cur.All(ctx, &runs); err != nil { + return nil, err + } + return runs, nil +} + +func CancelRun(runID string) error { + now := time.Now() + ctx, cancel := wfCtx() + defer cancel() + _, err := db.Col("workflow_runs").UpdateOne(ctx, + bson.M{"run_id": runID, "status": "running"}, + bson.M{"$set": bson.M{"status": "cancelled", "finished_at": now}}) + return err +} +``` + +Note on `setServerRun`: the positional `$` requires the query to match the array element, so it queries `server_runs.server_id`. `serverIDAt` resolves the id from the index (steps use `srvIdx` only to find the id, then everything else keys off `serverID`). + +- [ ] **Step 3: Verify build** + +Run: `cd server && go build ./... && go vet ./...` +Expected: success. Fix any unused-import or signature mismatches surfaced. + +- [ ] **Step 4: Commit** + +```bash +git add server/internal/services/workflow_runner.go server/internal/services/dispatch.go +git commit -m "feat(server): workflow runner with parallel fan-out and env threading" +``` + +--- + +## Task 7: REST API + routes + +**Files:** +- Create: `server/internal/api/workflows.go` +- Modify: `server/internal/api/handlers.go` (register routes) + +**Interfaces:** +- Consumes: services (Tasks 5, 6). +- Produces: HTTP endpoints per spec §8. + +- [ ] **Step 1: Write handlers** + +```go +package api + +import ( + "fmt" + "net/http" + "strconv" + + "github.com/gin-gonic/gin" + "github.com/mrhid6/vantage/server/internal/models" + "github.com/mrhid6/vantage/server/internal/services" +) + +func registerWorkflowRoutes(g *gin.RouterGroup) { + g.GET("/steps", listSteps) + g.POST("/steps", createStep) + g.PUT("/steps/:id", updateStep) + g.DELETE("/steps/:id", deleteStep) + + g.GET("/workflows", listWorkflows) + g.POST("/workflows", createWorkflow) + g.GET("/workflows/:id", getWorkflow) + g.PUT("/workflows/:id", updateWorkflow) + g.DELETE("/workflows/:id", deleteWorkflow) + g.POST("/workflows/:id/run", runWorkflow) + g.GET("/workflows/:id/runs", listWorkflowRuns) + + g.GET("/runs/:runId", getRun) + g.POST("/runs/:runId/cancel", cancelRun) +} + +func listSteps(c *gin.Context) { + steps, err := services.ListSteps() + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + c.JSON(http.StatusOK, steps) +} + +func createStep(c *gin.Context) { + var s models.WorkflowStep + if err := c.ShouldBindJSON(&s); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + out, err := services.CreateStep(s) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + services.LogEvent("workflow.step_created", actorFromCtx(c), "", out.StepID, fmt.Sprintf("step '%s' created", out.Name)) + c.JSON(http.StatusCreated, out) +} + +func updateStep(c *gin.Context) { + var s models.WorkflowStep + if err := c.ShouldBindJSON(&s); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + if err := services.UpdateStep(c.Param("id"), s); err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + services.LogEvent("workflow.step_updated", actorFromCtx(c), "", c.Param("id"), "step updated") + c.JSON(http.StatusOK, gin.H{"updated": true}) +} + +func deleteStep(c *gin.Context) { + if err := services.DeleteStep(c.Param("id")); err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + services.LogEvent("workflow.step_deleted", actorFromCtx(c), "", c.Param("id"), "step deleted") + c.JSON(http.StatusOK, gin.H{"deleted": true}) +} + +func listWorkflows(c *gin.Context) { + wfs, err := services.ListWorkflows() + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + c.JSON(http.StatusOK, wfs) +} + +func createWorkflow(c *gin.Context) { + var w models.Workflow + if err := c.ShouldBindJSON(&w); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + out, err := services.CreateWorkflow(w) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + services.LogEvent("workflow.created", actorFromCtx(c), "", out.WorkflowID, fmt.Sprintf("workflow '%s' created", out.Name)) + c.JSON(http.StatusCreated, out) +} + +func getWorkflow(c *gin.Context) { + w, err := services.GetWorkflow(c.Param("id")) + if err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": err.Error()}) + return + } + c.JSON(http.StatusOK, w) +} + +func updateWorkflow(c *gin.Context) { + var w models.Workflow + if err := c.ShouldBindJSON(&w); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + if err := services.UpdateWorkflow(c.Param("id"), w); err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + services.LogEvent("workflow.updated", actorFromCtx(c), "", c.Param("id"), "workflow updated") + c.JSON(http.StatusOK, gin.H{"updated": true}) +} + +func deleteWorkflow(c *gin.Context) { + if err := services.DeleteWorkflow(c.Param("id")); err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + services.LogEvent("workflow.deleted", actorFromCtx(c), "", c.Param("id"), "workflow deleted") + c.JSON(http.StatusOK, gin.H{"deleted": true}) +} + +func runWorkflow(c *gin.Context) { + runID, err := services.TriggerWorkflow(c.Param("id"), actorFromCtx(c)) + if err != nil { + c.JSON(http.StatusServiceUnavailable, gin.H{"error": err.Error()}) + return + } + services.LogEvent("workflow.run_triggered", actorFromCtx(c), "", c.Param("id"), fmt.Sprintf("run %s triggered", runID)) + c.JSON(http.StatusAccepted, gin.H{"run_id": runID}) +} + +func listWorkflowRuns(c *gin.Context) { + limit := int64(50) + if l := c.Query("limit"); l != "" { + if n, err := strconv.ParseInt(l, 10, 64); err == nil && n > 0 { + limit = n + } + } + runs, err := services.ListRuns(c.Param("id"), limit) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + c.JSON(http.StatusOK, runs) +} + +func getRun(c *gin.Context) { + r, err := services.GetRun(c.Param("runId")) + if err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": err.Error()}) + return + } + c.JSON(http.StatusOK, r) +} + +func cancelRun(c *gin.Context) { + if err := services.CancelRun(c.Param("runId")); err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + c.JSON(http.StatusOK, gin.H{"cancelled": true}) +} +``` + +- [ ] **Step 2: Register the group** + +In `server/internal/api/handlers.go`, inside the `apiGroup { ... }` block (after the console routes, before the closing brace), add: + +```go + registerWorkflowRoutes(apiGroup) +``` + +- [ ] **Step 3: Verify build** + +Run: `cd server && go build ./... && go vet ./...` +Expected: success. + +- [ ] **Step 4: Commit** + +```bash +git add server/internal/api/workflows.go server/internal/api/handlers.go +git commit -m "feat(api): workflow, step, and run REST endpoints" +``` + +--- + +## Task 8: API client + types (frontend) + +**Files:** +- Modify: `web/lib/api.ts` + +**Interfaces:** +- Produces: TS types `WorkflowStep`, `WorkflowStepRef`, `Workflow`, `WorkflowRun`, `ServerRun`, `StepRun`; `api` methods for all Task 7 endpoints. + +- [ ] **Step 1: Add types and methods** + +Match the existing `api` object style in `web/lib/api.ts` (same fetch/`apiFetch` helper the other methods use — inspect the file and reuse it). Add: + +```ts +export interface WorkflowStep { + step_id: string; + name: string; + description: string; + interpreter: "bash" | "powershell"; + script: string; + declared_outputs: string[]; + secret_refs: string[]; +} + +export interface WorkflowStepRef { + step_id: string; + order: number; + on_failure: "stop" | "continue" | "retry"; + max_retries: number; + overrides?: { script?: string; secret_refs?: string[] }; +} + +export interface Workflow { + workflow_id: string; + name: string; + target_server_ids: string[]; + steps: WorkflowStepRef[]; +} + +export interface StepRun { + order: number; + name: string; + status: string; + attempts: number; + exit_code: number; + stdout: string; + stderr: string; + output_env: Record; + started_at?: string; + finished_at?: string; +} + +export interface ServerRun { + server_id: string; + hostname: string; + status: string; + run_env: Record; + steps: StepRun[]; + started_at?: string; + finished_at?: string; +} + +export interface WorkflowRun { + run_id: string; + workflow_id: string; + name: string; + status: string; + triggered_by: string; + started_at: string; + finished_at?: string; + server_runs: ServerRun[]; +} +``` + +Then add methods to the `api` object (use the file's existing request helper; shown here with a generic `req`): + +```ts + listSteps: () => req("/api/steps"), + createStep: (s: Partial) => req("/api/steps", { method: "POST", body: JSON.stringify(s) }), + updateStep: (id: string, s: Partial) => req(`/api/steps/${id}`, { method: "PUT", body: JSON.stringify(s) }), + deleteStep: (id: string) => req(`/api/steps/${id}`, { method: "DELETE" }), + + listWorkflows: () => req("/api/workflows"), + getWorkflow: (id: string) => req(`/api/workflows/${id}`), + createWorkflow: (w: Partial) => req("/api/workflows", { method: "POST", body: JSON.stringify(w) }), + updateWorkflow: (id: string, w: Partial) => req(`/api/workflows/${id}`, { method: "PUT", body: JSON.stringify(w) }), + deleteWorkflow: (id: string) => req(`/api/workflows/${id}`, { method: "DELETE" }), + runWorkflow: (id: string) => req<{ run_id: string }>(`/api/workflows/${id}/run`, { method: "POST" }), + listRuns: (id: string) => req(`/api/workflows/${id}/runs`), + getRun: (runId: string) => req(`/api/runs/${runId}`), + cancelRun: (runId: string) => req(`/api/runs/${runId}/cancel`, { method: "POST" }), +``` + +Adapt `req`/method names to whatever the file already defines (e.g. it may use `apiFetch` or per-verb helpers). Keep the existing patterns. + +- [ ] **Step 2: Verify build** + +Run: `cd web && npm run build` +Expected: type-checks and builds. Fix type mismatches against the real helper signature. + +- [ ] **Step 3: Commit** + +```bash +git add web/lib/api.ts +git commit -m "feat(web): workflow API client types and methods" +``` + +--- + +## Task 9: Workflows list page + sidebar link + +**Files:** +- Create: `web/app/workflows/page.tsx` +- Modify: `web/components/Sidebar.tsx` (add a Workflows nav item next to Secrets/Servers) + +**Interfaces:** +- Consumes: `api.listWorkflows`, `api.listRuns`, `api.runWorkflow`, `api.createWorkflow` (Task 8). + +- [ ] **Step 1: Add sidebar link** + +In `web/components/Sidebar.tsx`, add a nav entry `{ href: "/workflows", label: "Workflows" }` following the existing item structure/icon pattern used for Servers and Secrets. + +- [ ] **Step 2: Write the list page** + +```tsx +"use client"; + +import { useState } from "react"; +import Link from "next/link"; +import { useQuery, useMutation, useQueryClient } from "@tanstack/react-query"; +import { api, Workflow } from "@/lib/api"; +import { Button, Card } from "@/components/ui"; +import { Table, Thead, Tbody, Tr, Th, Td } from "@/components/ui"; + +export default function WorkflowsPage() { + const qc = useQueryClient(); + const [creating, setCreating] = useState(false); + + const { data: workflows, isLoading, error } = useQuery({ + queryKey: ["workflows"], + queryFn: api.listWorkflows, + }); + + const { mutate: create, isPending } = useMutation({ + mutationFn: () => api.createWorkflow({ name: "Untitled workflow", target_server_ids: [], steps: [] }), + onSuccess: () => { + qc.invalidateQueries({ queryKey: ["workflows"] }); + setCreating(false); + }, + }); + + return ( +
+
+
+

Workflows

+

+ {workflows?.length ?? 0} workflow{workflows?.length !== 1 ? "s" : ""} · run reusable steps across servers +

+
+ +
+ + + {isLoading ? ( +
+
+
+ ) : error ? ( +
Failed to load workflows.
+ ) : workflows && workflows.length > 0 ? ( + + + + + + {workflows.map((w: Workflow) => ( + + + + + + + ))} + +
NameTargetsSteps
{w.name}{w.target_server_ids.length} server{w.target_server_ids.length !== 1 ? "s" : ""}{w.steps.length} + + + +
+ ) : ( +
+

No workflows yet.

+ +
+ )} + +
+ ); +} +``` + +- [ ] **Step 3: Verify build** + +Run: `cd web && npm run build` +Expected: success. + +- [ ] **Step 4: Commit** + +```bash +git add web/app/workflows/page.tsx web/components/Sidebar.tsx +git commit -m "feat(web): workflows list page and sidebar link" +``` + +--- + +## Task 10: Workflow builder page (three-pane) + +**Files:** +- Create: `web/app/workflows/[id]/page.tsx` + +**Interfaces:** +- Consumes: `api.getWorkflow`, `api.updateWorkflow`, `api.listSteps`, `api.createStep`, `api.runWorkflow`, `api.listServers`, `api.listSecretGroups`. + +Reference the approved mockup (`workflow-builder.html`) for layout: left library, center canvas of ordered nodes with env chips on wires, right inspector. Implement with Tailwind tokens; drag can be simplified to add/reorder buttons for v1 (HTML5 drag optional). + +- [ ] **Step 1: Write the builder page** + +Implement a client component with three columns (CSS grid `grid-cols-[264px_1fr_320px]`): +- **Left (Library):** `api.listSteps()` list with `bash`/`pwsh` badges; an "Add" button opens an inline form calling `api.createStep`; clicking a library step appends a `WorkflowStepRef` to local workflow state. +- **Center (Canvas):** render `workflow.steps` (sorted by `order`) as node cards showing the resolved step name, interpreter badge, and the step's script preview. Between nodes render a "passes" chip row derived from each step's `declared_outputs` (union of all prior outputs). Provide up/down reorder and remove buttons. A header shows workflow name (editable input), a target-servers multiselect (`api.listServers`), Save (`api.updateWorkflow`) and Run (`api.runWorkflow`, then route to the run detail page). +- **Right (Inspector):** for the selected node: name, script editor (`