Compare commits

..
40 Commits
Author SHA1 Message Date
mrhid6 351ad59dd8 fix(web): reseed edit-workflow modal state on open
Server Deploy / deploy (push) Successful in 1m23s
2026-07-20 14:56:14 +01:00
mrhid6 dcc901b0d2 feat(web): workflow runs list page and navigation links
Server Deploy / deploy (push) Successful in 1m29s
2026-07-20 14:52:13 +01:00
mrhid6 99bf093f00 feat(web): rebuild workflow builder — mockup styling, drag-and-drop, inputs inspector 2026-07-20 14:48:29 +01:00
mrhid6 619ccd28cb feat(web): edit-workflow modal (name/targets/delete) 2026-07-20 14:44:59 +01:00
mrhid6 f22f0a4729 feat(web): edit-base-step modal with inputs/outputs editor 2026-07-20 14:42:13 +01:00
mrhid6 78194daf5f feat(web): builder tokens, Modal primitive, input-param types 2026-07-20 14:40:07 +01:00
mrhid6 f141767fc2 feat(workflows): inject step inputs into env; update returns workflow 2026-07-20 14:37:36 +01:00
mrhid6 05cd8e154b feat(workflows): step input params model + cascade step delete 2026-07-20 14:34:50 +01:00
mrhid6 004cc03ba6 docs: add workflow builder v2 plan 2026-07-20 14:33:46 +01:00
mrhid6 236e89989f docs: add workflow builder v2 spec 2026-07-20 14:30:55 +01:00
mrhid6 b0a2de8ca1 fix: Fixed workflow style topbar
Server Deploy / deploy (push) Successful in 1m20s
2026-07-20 13:55:17 +01:00
mrhid6 e9ac7be8c3 feat: updates to workflow page
Server Deploy / deploy (push) Successful in 36s
2026-07-20 13:35:35 +01:00
mrhid6 47690c58d9 fix: Fixed workflow id
Server Deploy / deploy (push) Successful in 1m14s
2026-07-20 12:49:43 +01:00
mrhid6 5d72088837 feat(agent): stream step output chunks over CommandStream
Agent Release / build (push) Successful in 44s
Server Deploy / deploy (push) Successful in 1m51s
Agent Release / msi (push) Successful in 1m23s
2026-07-20 12:36:37 +01:00
mrhid6 7a60295bc1 feat(proto): add StepOutputChunk streaming message 2026-07-20 12:34:32 +01:00
mrhid6 b48467fb6e docs: add workflow log streaming plan 2026-07-20 12:32:57 +01:00
mrhid6 b5e828c9e8 docs: add workflow log streaming spec 2026-07-20 12:30:11 +01:00
mrhid6 98284f4387 fix(workflows): mask output env in persisted logs, preserve cancelled status, run agent step async 2026-07-20 12:01:35 +01:00
mrhid6 e35e8fc839 feat(web): workflow run detail page with live logs 2026-07-20 11:53:30 +01:00
mrhid6 2cd9bc1c89 feat(web): three-pane workflow builder 2026-07-20 11:50:00 +01:00
mrhid6 39980581b1 feat(web): workflows list page and sidebar link 2026-07-20 11:46:44 +01:00
mrhid6 6f478eb817 feat(web): workflow API client types and methods 2026-07-20 11:44:10 +01:00
mrhid6 ff3a94b888 fix(api): audit workflow run cancellation 2026-07-20 11:42:31 +01:00
mrhid6 631894084a feat(api): workflow, step, and run REST endpoints 2026-07-20 11:40:36 +01:00
mrhid6 296e0179cb feat(server): workflow runner with parallel fan-out and env threading 2026-07-20 11:37:04 +01:00
mrhid6 600126a913 feat(server): step library and workflow CRUD services 2026-07-20 11:33:33 +01:00
mrhid6 4872a26786 feat(models): add workflow, step, and run models 2026-07-20 11:31:37 +01:00
mrhid6 f0c86a3bdf feat(agent): execute RunStepCmd with WORKFLOW_ENV capture 2026-07-20 11:29:14 +01:00
mrhid6 1b286762f6 feat(server): add pending step-result registry and stream delivery 2026-07-20 11:26:35 +01:00
mrhid6 3c77c20de8 feat(proto): add RunStepCmd and StepResult messages 2026-07-20 11:24:08 +01:00
mrhid6 9e53f21746 docs: add Fleet Inventory and SaaS auth/orgs specs + plans 2026-07-20 11:16:19 +01:00
mrhid6 ad35b32f5b docs: add Server Workflows implementation plan 2026-07-20 11:05:43 +01:00
mrhid6 d20d3b08fa docs: add Server Workflows design spec 2026-07-20 10:55:21 +01:00
mrhid6 c3c58581cc fix: Fixed scale and mouse handler
Server Deploy / deploy (push) Successful in 1m38s
2026-07-20 09:49:20 +01:00
mrhid6 c558b81471 fix: Fixed keyboard disconnect
Server Deploy / deploy (push) Successful in 41s
2026-07-20 09:42:07 +01:00
mrhid6 963fa9c877 fix: Fixed mouse position on console
Server Deploy / deploy (push) Successful in 39s
2026-07-20 09:36:57 +01:00
mrhid6 a02747d02e fix: Fixed console resolution
Server Deploy / deploy (push) Successful in 44s
2026-07-20 09:30:44 +01:00
mrhid6 db5b5e173f fix: Ci and agent update
Server Deploy / deploy (push) Successful in 13s
Agent Release / build (push) Successful in 10m32s
Agent Release / msi (push) Successful in 36s
2026-07-17 16:43:57 +01:00
mrhid6 2657780e7a fix: agent msi agent version upgrade
Server Deploy / deploy (push) Successful in 11s
Agent Release / build (push) Successful in 10m38s
Agent Release / msi (push) Successful in 58s
2026-07-17 16:19:42 +01:00
mrhid6 129be23a9d fix: Fixed msi version
Server Deploy / deploy (push) Successful in 13s
Agent Release / build (push) Successful in 33s
Agent Release / msi (push) Successful in 33s
2026-07-17 16:10:19 +01:00
42 changed files with 8711 additions and 40 deletions
-16
View File
@@ -86,22 +86,6 @@ jobs:
$env:GOOS = "windows"; $env:GOARCH = "amd64"
go build -ldflags="-s -w -X main.Version=$env:VERSION" -o ../installer/vantage-agent-windows-amd64.exe ./cmd
- name: Cache nssm
id: cache-nssm
uses: actions/cache@v4
with:
path: installer/nssm.exe
key: nssm-2.24-win64
- name: Fetch nssm
if: steps.cache-nssm.outputs.cache-hit != 'true'
working-directory: installer
shell: pwsh
run: |
Invoke-WebRequest -Uri https://nssm.cc/release/nssm-2.24.zip -OutFile nssm.zip
Expand-Archive -Path nssm.zip -DestinationPath nssm-extract -Force
Copy-Item nssm-extract/nssm-2.24/win64/nssm.exe -Destination nssm.exe
- name: Install WiX
shell: pwsh
run: dotnet tool install --global wix --version 5.*
+1 -1
View File
@@ -4,7 +4,7 @@ build
.env
docs
.superpowers
installer/*.exe
installer/vantage-agent-windows-amd64.exe
installer/*.msi
installer/nssm.zip
installer/checksums-msi.txt
+139
View File
@@ -0,0 +1,139 @@
package exec
import (
"bufio"
"context"
"os"
"os/exec"
"path/filepath"
"runtime"
"strings"
"sync"
"time"
"github.com/mrhid6/vantage/agent/internal/grpc/pb"
)
// streamWriter forwards every write to emit() as an ordered chunk. Used as both
// Stdout and Stderr so output interleaves in real execution order. The mutex
// ensures a single stdout/stderr write is not interleaved mid-slice with another.
type streamWriter struct {
mu sync.Mutex
seq uint64
emit func(seq uint64, data []byte)
}
func (w *streamWriter) Write(p []byte) (int, error) {
w.mu.Lock()
defer w.mu.Unlock()
if w.emit != nil {
buf := make([]byte, len(p))
copy(buf, p)
w.emit(w.seq, buf)
w.seq++
}
return len(p), nil
}
// 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 streams output via emit, returning the terminal result
// with empty stdout/stderr but populated exit_code/output_env.
func RunStep(cmd *pb.RunStepCmd, emit func(seq uint64, data []byte)) *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)
}
sw := &streamWriter{emit: emit}
c.Stdout = sw
c.Stderr = sw
runErr := c.Run()
// stdout/stderr are streamed via emit, not returned in the result.
if ctx.Err() == context.DeadlineExceeded {
res.ExitCode = 124
res.Stderr = "[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 = "[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
}
+29 -4
View File
@@ -68,6 +68,7 @@ type ServerCommand struct {
DeleteKey *DeleteKeyCmd `json:"delete_key,omitempty"`
UpdateAgent *UpdateAgentCmd `json:"update_agent,omitempty"`
ApplyUpdates *ApplyUpdatesCmd `json:"apply_updates,omitempty"`
RunStep *RunStepCmd `json:"run_step,omitempty"`
}
type DeleteKeyCmd struct {
@@ -88,10 +89,12 @@ type GenerateKeyCmd struct {
}
type AgentMessage struct {
ServerId string `json:"server_id"`
AgentToken string `json:"agent_token"`
Ready *AgentReady `json:"ready,omitempty"`
Result *CommandResult `json:"result,omitempty"`
ServerId string `json:"server_id"`
AgentToken string `json:"agent_token"`
Ready *AgentReady `json:"ready,omitempty"`
Result *CommandResult `json:"result,omitempty"`
StepResult *StepResult `json:"step_result,omitempty"`
StepOutput *StepOutputChunk `json:"step_output,omitempty"`
}
type AgentReady struct{}
@@ -102,6 +105,28 @@ type CommandResult struct {
Message string `json:"message"`
}
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"`
}
type StepOutputChunk struct {
CommandId string `json:"command_id"`
Seq uint64 `json:"seq"`
Data []byte `json:"data,omitempty"`
Eof bool `json:"eof,omitempty"`
}
// CommandStream client-side interface
type Vantage_CommandStreamClient interface {
+37
View File
@@ -14,9 +14,11 @@ import (
"path/filepath"
"runtime"
"strings"
"sync"
"time"
"github.com/mrhid6/vantage/agent/internal/config"
agentexec "github.com/mrhid6/vantage/agent/internal/exec"
grpcclient "github.com/mrhid6/vantage/agent/internal/grpc"
"github.com/mrhid6/vantage/agent/internal/grpc/pb"
"github.com/mrhid6/vantage/agent/internal/keys"
@@ -168,6 +170,16 @@ func connectAndHandleStream(ctx context.Context, cfg *config.Config) error {
log.Println("command stream connected")
// grpc streams are not safe for concurrent Send; RunStep results are sent
// from per-command goroutines, so all sends on this stream must go through
// this mutex-protected helper.
var sendMu sync.Mutex
send := func(msg *pb.AgentMessage) error {
sendMu.Lock()
defer sendMu.Unlock()
return stream.Send(msg)
}
for {
cmd, err := stream.Recv()
if err != nil {
@@ -186,6 +198,31 @@ func connectAndHandleStream(ctx context.Context, cfg *config.Config) error {
if cmd.ApplyUpdates != nil {
go handleApplyUpdates(cfg, cmd)
}
if cmd.RunStep != nil {
go func(rc *pb.RunStepCmd, cid string) {
emit := func(seq uint64, data []byte) {
_ = send(&pb.AgentMessage{
ServerId: cfg.ServerID,
AgentToken: cfg.AgentToken,
StepOutput: &pb.StepOutputChunk{CommandId: cid, Seq: seq, Data: data},
})
}
res := agentexec.RunStep(rc, emit)
res.CommandId = cid
// Final eof marker so the server closes the log file.
_ = send(&pb.AgentMessage{
ServerId: cfg.ServerID,
AgentToken: cfg.AgentToken,
StepOutput: &pb.StepOutputChunk{CommandId: cid, Eof: true},
})
_ = send(&pb.AgentMessage{
ServerId: cfg.ServerID,
AgentToken: cfg.AgentToken,
StepResult: res,
})
}(cmd.RunStep, cmd.CommandId)
continue
}
}
}
@@ -0,0 +1,653 @@
# Fleet Inventory 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:** Agents collect CPU/RAM/swap/disk/partition inventory and report it to the server via a new `ReportInventory` RPC; the server stores the latest snapshot per server and the UI displays it.
**Architecture:** New unary gRPC `ReportInventory` (mirrors existing `ReportUpdates`). Agent runs a 30s metrics ticker (CPU/RAM/swap usage) and, every 15 min, a full static collection (disks, partitions, CPU model, kernel). Server upserts an embedded `inventory` sub-doc on the `servers` document with merge rules that preserve static fields between slow ticks.
**Tech Stack:** Go (gin, mongo-driver v2, hand-written JSON-codec gRPC), `/proc` readers, Next.js 16 + react-query + Tailwind.
## Global Constraints
- **No tests this iteration.** Verify with `go build ./...`, `go vet ./...`, `npm run build`.
- gRPC uses a JSON codec: edit **both** `server/internal/grpc/pb/vantage.pb.go` and `agent/internal/grpc/pb/vantage.pb.go` identically, plus `proto/vantage/v1/vantage.proto` as documentation. No codegen. Mirror the existing `ReportUpdates` RPC wiring exactly (service interface, `_Vantage_*_Handler`, client method, `Vantage_ServiceDesc`).
- Mongo: `db.Col("servers")`, `context.WithTimeout`. Follow `server/internal/services/servers.go`.
- Agent already runs as root; `/proc` is readable. Linux is primary; Windows collectors may return empty.
- Module path `github.com/mrhid6/vantage`.
- Do not add heavy dependencies; implement `/proc` parsing directly.
---
## Task 1: Inventory model + gRPC messages
**Files:**
- Modify: `server/internal/models/server.go`
- Modify: `proto/vantage/v1/vantage.proto`
- Modify: `server/internal/grpc/pb/vantage.pb.go`
- Modify: `agent/internal/grpc/pb/vantage.pb.go`
**Interfaces:**
- Produces: `models.Inventory` (+ `CPUInfo`, `MemInfo`, `Partition`) and `Server.Inventory *Inventory`. pb structs `InventoryReport`, `CPUReport`, `MemReport`, `PartitionReport`, `InventoryReportResponse`. Service method `ReportInventory` on both client and server interfaces.
- [ ] **Step 1: Add model structs**
In `server/internal/models/server.go` add (keep the existing `import "time"`):
```go
type CPUInfo struct {
Model string `bson:"model,omitempty" json:"model,omitempty"`
Cores int `bson:"cores,omitempty" json:"cores,omitempty"`
UsagePct float64 `bson:"usage_pct" json:"usage_pct"`
Load1 float64 `bson:"load1,omitempty" json:"load1,omitempty"`
}
type MemInfo struct {
TotalBytes uint64 `bson:"total_bytes" json:"total_bytes"`
UsedBytes uint64 `bson:"used_bytes" json:"used_bytes"`
}
type Partition struct {
Device string `bson:"device" json:"device"`
Mountpoint string `bson:"mountpoint" json:"mountpoint"`
Fstype string `bson:"fstype,omitempty" json:"fstype,omitempty"`
TotalBytes uint64 `bson:"total_bytes" json:"total_bytes"`
UsedBytes uint64 `bson:"used_bytes" json:"used_bytes"`
}
type Inventory struct {
CPU CPUInfo `bson:"cpu" json:"cpu"`
Memory MemInfo `bson:"memory" json:"memory"`
SwapTotalBytes uint64 `bson:"swap_total_bytes" json:"swap_total_bytes"`
SwapUsedBytes uint64 `bson:"swap_used_bytes" json:"swap_used_bytes"`
Partitions []Partition `bson:"partitions,omitempty" json:"partitions,omitempty"`
Kernel string `bson:"kernel,omitempty" json:"kernel,omitempty"`
MetricsAt *time.Time `bson:"metrics_at,omitempty" json:"metrics_at,omitempty"`
StaticAt *time.Time `bson:"static_at,omitempty" json:"static_at,omitempty"`
}
```
Add to the `Server` struct: `Inventory *Inventory \`bson:"inventory,omitempty" json:"inventory,omitempty"\``.
- [ ] **Step 2: Document RPC in proto**
In `proto/vantage/v1/vantage.proto`, add to the service: `rpc ReportInventory(InventoryReport) returns (InventoryReportResponse);` and the messages `InventoryReport`, `CPUReport`, `MemReport`, `PartitionReport`, `InventoryReportResponse` per spec §4.
- [ ] **Step 3: Add pb structs + RPC wiring (server pb)**
In `server/internal/grpc/pb/vantage.pb.go` add the message structs:
```go
type CPUReport struct {
Model string `json:"model,omitempty"`
Cores int `json:"cores,omitempty"`
UsagePct float64 `json:"usage_pct"`
Load1 float64 `json:"load1,omitempty"`
}
type MemReport struct {
TotalBytes uint64 `json:"total_bytes"`
UsedBytes uint64 `json:"used_bytes"`
}
type PartitionReport struct {
Device string `json:"device"`
Mountpoint string `json:"mountpoint"`
Fstype string `json:"fstype,omitempty"`
TotalBytes uint64 `json:"total_bytes"`
UsedBytes uint64 `json:"used_bytes"`
}
type InventoryReport struct {
ServerId string `json:"server_id"`
AgentToken string `json:"agent_token"`
IncludeStatic bool `json:"include_static"`
CPU *CPUReport `json:"cpu,omitempty"`
Memory *MemReport `json:"memory,omitempty"`
SwapTotal uint64 `json:"swap_total"`
SwapUsed uint64 `json:"swap_used"`
Partitions []PartitionReport `json:"partitions,omitempty"`
Kernel string `json:"kernel,omitempty"`
}
type InventoryReportResponse struct{}
```
Then mirror the `ReportUpdates` RPC plumbing for `ReportInventory`. Locate every `ReportUpdates` reference in this file and add the parallel `ReportInventory`:
- `VantageServer` interface: add `ReportInventory(context.Context, *InventoryReport) (*InventoryReportResponse, error)`.
- `UnimplementedVantageServer`: add the stub returning `Unimplemented`.
- `VantageClient` interface + `keyManagerClient`: add the client method `Invoke`-ing `/vantage.v1.Vantage/ReportInventory`.
- `Vantage_ServiceDesc.Methods`: add `{MethodName: "ReportInventory", Handler: _Vantage_ReportInventory_Handler}`.
- Add `_Vantage_ReportInventory_Handler` copied from `_Vantage_ReportUpdates_Handler` with types swapped.
- [ ] **Step 4: Mirror pb structs + wiring (agent pb)**
Apply the identical additions to `agent/internal/grpc/pb/vantage.pb.go`.
- [ ] **Step 5: Verify build**
Run: `cd server && go build ./... && cd ../agent && go build ./...`
Expected: both succeed.
- [ ] **Step 6: Commit**
```bash
git add server/internal/models/server.go 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 ReportInventory RPC and inventory model"
```
---
## Task 2: Server handler + store service
**Files:**
- Create: `server/internal/services/inventory.go`
- Modify: `server/internal/grpc/server.go`
**Interfaces:**
- Consumes: `pb.InventoryReport` (T1), `db.Col("servers")`.
- Produces: `services.StoreInventory(serverID string, r *pb.InventoryReport) error`; gRPC method `(*vantageServer).ReportInventory`.
- [ ] **Step 1: Write the store service**
```go
package services
import (
"context"
"time"
"github.com/mrhid6/vantage/server/internal/db"
"github.com/mrhid6/vantage/server/internal/grpc/pb"
"go.mongodb.org/mongo-driver/v2/bson"
)
// StoreInventory upserts the latest inventory snapshot onto the server document.
// Metrics fields update every call; static fields only when r.IncludeStatic.
func StoreInventory(serverID string, r *pb.InventoryReport) error {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
now := time.Now()
set := bson.M{"inventory.metrics_at": now}
if r.CPU != nil {
set["inventory.cpu.usage_pct"] = r.CPU.UsagePct
set["inventory.cpu.load1"] = r.CPU.Load1
}
if r.Memory != nil {
set["inventory.memory.used_bytes"] = r.Memory.UsedBytes
}
set["inventory.swap_used_bytes"] = r.SwapUsed
if r.IncludeStatic {
set["inventory.static_at"] = now
set["inventory.swap_total_bytes"] = r.SwapTotal
set["inventory.kernel"] = r.Kernel
if r.CPU != nil {
set["inventory.cpu.model"] = r.CPU.Model
set["inventory.cpu.cores"] = r.CPU.Cores
}
if r.Memory != nil {
set["inventory.memory.total_bytes"] = r.Memory.TotalBytes
}
parts := make([]bson.M, 0, len(r.Partitions))
for _, p := range r.Partitions {
parts = append(parts, bson.M{
"device": p.Device, "mountpoint": p.Mountpoint, "fstype": p.Fstype,
"total_bytes": p.TotalBytes, "used_bytes": p.UsedBytes,
})
}
set["inventory.partitions"] = parts
}
_, err := db.Col("servers").UpdateOne(ctx, bson.M{"server_id": serverID}, bson.M{"$set": set})
return err
}
```
- [ ] **Step 2: Add the gRPC handler**
In `server/internal/grpc/server.go`, add (mirroring the existing `ReportUpdates` handler that validates the agent token):
```go
func (s *vantageServer) ReportInventory(ctx context.Context, req *pb.InventoryReport) (*pb.InventoryReportResponse, error) {
srv, err := services.ValidateAgentToken(req.ServerId, req.AgentToken)
if err != nil {
return nil, status.Errorf(codes.Unauthenticated, "invalid agent token")
}
if err := services.StoreInventory(srv.ServerID, req); err != nil {
log.Printf("store inventory for %s: %v", srv.ServerID, err)
}
return &pb.InventoryReportResponse{}, nil
}
```
Confirm `status`, `codes`, `log` are already imported in the file (they are, used by other handlers).
- [ ] **Step 3: Verify build**
Run: `cd server && go build ./... && go vet ./...`
Expected: success.
- [ ] **Step 4: Commit**
```bash
git add server/internal/services/inventory.go server/internal/grpc/server.go
git commit -m "feat(server): store inventory and handle ReportInventory RPC"
```
---
## Task 3: Agent collectors
**Files:**
- Create: `agent/internal/inventory/collect_linux.go`
- Create: `agent/internal/inventory/collect_other.go`
- Create: `agent/internal/inventory/inventory.go`
**Interfaces:**
- Produces: `inventory.Collect(includeStatic bool) *pb.InventoryReport`.
- [ ] **Step 1: Common entry (`inventory.go`)**
```go
package inventory
import "github.com/mrhid6/vantage/agent/internal/grpc/pb"
// Collect gathers metrics always and static hardware info when includeStatic.
// Platform specifics are provided by collect_linux.go / collect_other.go.
func Collect(includeStatic bool) *pb.InventoryReport {
r := &pb.InventoryReport{IncludeStatic: includeStatic, CPU: &pb.CPUReport{}, Memory: &pb.MemReport{}}
collect(r, includeStatic)
return r
}
```
- [ ] **Step 2: Linux collector (`collect_linux.go`)**
Build-tagged `//go:build linux`. Implement `collect(r *pb.InventoryReport, includeStatic bool)`:
- CPU usage: read `/proc/stat` first line twice ~100ms apart, compute `1 - idleDelta/totalDelta` × 100 → `r.CPU.UsagePct`.
- Load: first field of `/proc/loadavg` → `r.CPU.Load1`.
- Mem/swap: parse `/proc/meminfo` (`MemTotal`, `MemAvailable`, `SwapTotal`, `SwapFree`; used = total available; swap used = swaptotal swapfree) → `r.Memory.*`, `r.SwapUsed`, and on static `r.SwapTotal`.
- Static only: `/proc/cpuinfo` (`model name`, count `processor` lines) → `r.CPU.Model/Cores`; `/proc/meminfo MemTotal` → `r.Memory.TotalBytes`; kernel via `syscall.Uname` or read `/proc/sys/kernel/osrelease` → `r.Kernel`; partitions from `/proc/mounts` filtered to fstypes in {ext4,xfs,btrfs,zfs,vfat,ntfs} then `syscall.Statfs` for total/used → `r.Partitions`.
```go
//go:build linux
package inventory
import (
"bufio"
"os"
"strconv"
"strings"
"syscall"
"time"
"github.com/mrhid6/vantage/agent/internal/grpc/pb"
)
func collect(r *pb.InventoryReport, includeStatic bool) {
r.CPU.UsagePct = cpuUsage()
r.CPU.Load1 = load1()
memTotal, memAvail, swapTotal, swapFree := meminfo()
if memTotal > memAvail {
r.Memory.UsedBytes = memTotal - memAvail
}
if swapTotal > swapFree {
r.SwapUsed = swapTotal - swapFree
}
if includeStatic {
r.Memory.TotalBytes = memTotal
r.SwapTotal = swapTotal
r.CPU.Model, r.CPU.Cores = cpuStatic()
r.Kernel = kernel()
r.Partitions = partitions()
}
}
func readProc(path string) string { b, _ := os.ReadFile(path); return string(b) }
func cpuSample() (idle, total uint64) {
f, err := os.Open("/proc/stat")
if err != nil {
return
}
defer f.Close()
sc := bufio.NewScanner(f)
if sc.Scan() {
fields := strings.Fields(sc.Text()) // cpu user nice system idle iowait ...
for i, v := range fields[1:] {
n, _ := strconv.ParseUint(v, 10, 64)
total += n
if i == 3 { // idle
idle = n
}
}
}
return
}
func cpuUsage() float64 {
i1, t1 := cpuSample()
time.Sleep(100 * time.Millisecond)
i2, t2 := cpuSample()
dt := float64(t2 - t1)
if dt <= 0 {
return 0
}
return (1 - float64(i2-i1)/dt) * 100
}
func load1() float64 {
fields := strings.Fields(readProc("/proc/loadavg"))
if len(fields) > 0 {
v, _ := strconv.ParseFloat(fields[0], 64)
return v
}
return 0
}
func meminfo() (total, avail, swapTotal, swapFree uint64) {
f, err := os.Open("/proc/meminfo")
if err != nil {
return
}
defer f.Close()
sc := bufio.NewScanner(f)
for sc.Scan() {
fields := strings.Fields(sc.Text())
if len(fields) < 2 {
continue
}
kb, _ := strconv.ParseUint(fields[1], 10, 64)
b := kb * 1024
switch strings.TrimSuffix(fields[0], ":") {
case "MemTotal":
total = b
case "MemAvailable":
avail = b
case "SwapTotal":
swapTotal = b
case "SwapFree":
swapFree = b
}
}
return
}
func cpuStatic() (model string, cores int) {
f, err := os.Open("/proc/cpuinfo")
if err != nil {
return
}
defer f.Close()
sc := bufio.NewScanner(f)
for sc.Scan() {
line := sc.Text()
if strings.HasPrefix(line, "processor") {
cores++
} else if strings.HasPrefix(line, "model name") && model == "" {
if i := strings.Index(line, ":"); i >= 0 {
model = strings.TrimSpace(line[i+1:])
}
}
}
return
}
func kernel() string {
return strings.TrimSpace(readProc("/proc/sys/kernel/osrelease"))
}
func partitions() []pb.PartitionReport {
allowed := map[string]bool{"ext4": true, "xfs": true, "btrfs": true, "zfs": true, "vfat": true, "ntfs": true, "ext3": true}
f, err := os.Open("/proc/mounts")
if err != nil {
return nil
}
defer f.Close()
var out []pb.PartitionReport
seen := map[string]bool{}
sc := bufio.NewScanner(f)
for sc.Scan() {
fields := strings.Fields(sc.Text())
if len(fields) < 3 || !allowed[fields[2]] || seen[fields[1]] {
continue
}
seen[fields[1]] = true
var st syscall.Statfs_t
if syscall.Statfs(fields[1], &st) != nil {
continue
}
total := st.Blocks * uint64(st.Bsize)
free := st.Bavail * uint64(st.Bsize)
out = append(out, pb.PartitionReport{
Device: fields[0], Mountpoint: fields[1], Fstype: fields[2],
TotalBytes: total, UsedBytes: total - free,
})
}
return out
}
```
- [ ] **Step 3: Non-linux stub (`collect_other.go`)**
```go
//go:build !linux
package inventory
import "github.com/mrhid6/vantage/agent/internal/grpc/pb"
// collect is a no-op best-effort stub on non-Linux platforms.
func collect(r *pb.InventoryReport, includeStatic bool) {}
```
- [ ] **Step 4: Verify build**
Run: `cd agent && go build ./... && go vet ./...`
Expected: success (build both native and, if convenient, `GOOS=windows go build ./...`).
- [ ] **Step 5: Commit**
```bash
git add agent/internal/inventory/
git commit -m "feat(agent): /proc-based inventory collectors"
```
---
## Task 4: Agent client method + scheduler
**Files:**
- Modify: `agent/internal/grpc/client.go`
- Modify: the agent main loop (`agent/cmd/main.go` or `agent/internal/sync/sync.go` — wherever the poll loop/tickers live).
**Interfaces:**
- Consumes: `inventory.Collect` (T3), pb (T1).
- Produces: `(*Client).ReportInventory(report *pb.InventoryReport) error`; a running ticker that reports metrics every 30s and static every 15 min.
- [ ] **Step 1: Add client method**
In `agent/internal/grpc/client.go`, mirroring `ReportUpdates`:
```go
func (c *Client) ReportInventory(report *pb.InventoryReport) error {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
_, err := c.client.ReportInventory(ctx, report)
return err
}
```
The report already carries `ServerId`/`AgentToken`; ensure the caller sets them (see Step 2).
- [ ] **Step 2: Add the scheduler to the agent loop**
Find where the agent starts its poll loop (the goroutine that calls `SyncKeys`/`ReportUpdates`). Add a parallel inventory ticker. `serverID`, `agentToken`, and the `*Client` are in scope there:
```go
go func() {
tick := 0
t := time.NewTicker(30 * time.Second)
defer t.Stop()
report := func(static bool) {
r := inventory.Collect(static)
r.ServerId = serverID
r.AgentToken = agentToken
if err := client.ReportInventory(r); err != nil {
log.Printf("report inventory: %v", err)
}
}
report(true) // send a full snapshot on startup
for range t.C {
tick++
report(tick%30 == 0) // every 30th tick = 15 min → include static
}
}()
```
Add imports `"github.com/mrhid6/vantage/agent/internal/inventory"`, `time`, `log` if missing. Match variable names to the actual loop (e.g. the client may be named `c`).
- [ ] **Step 3: Verify build**
Run: `cd agent && go build ./... && go vet ./...`
Expected: success.
- [ ] **Step 4: Commit**
```bash
git add agent/internal/grpc/client.go agent/
git commit -m "feat(agent): schedule inventory reporting (30s metrics, 15m static)"
```
---
## Task 5: Frontend — inventory panel on server detail
**Files:**
- Modify: `web/lib/api.ts` (extend the `Server`/server-detail type with `inventory`)
- Modify: `web/app/servers/[id]/page.tsx` (add panel; enable polling)
**Interfaces:**
- Consumes: server-detail query.
- [ ] **Step 1: Add the inventory type**
In `web/lib/api.ts`, add and attach to the server type used by the detail page:
```ts
export interface Inventory {
cpu: { model?: string; cores?: number; usage_pct: number; load1?: number };
memory: { total_bytes: number; used_bytes: number };
swap_total_bytes: number;
swap_used_bytes: number;
partitions?: { device: string; mountpoint: string; fstype?: string; total_bytes: number; used_bytes: number }[];
kernel?: string;
metrics_at?: string;
static_at?: string;
}
```
Add `inventory?: Inventory;` to the server detail interface.
- [ ] **Step 2: Add a `formatBytes` helper + Inventory panel**
In `web/app/servers/[id]/page.tsx`, add a helper and a panel component. Enable polling on the server-detail `useQuery` with `refetchInterval: 30000`.
```tsx
function formatBytes(n: number): string {
if (!n) return "0 B";
const u = ["B", "KB", "MB", "GB", "TB"];
const i = Math.floor(Math.log(n) / Math.log(1024));
return `${(n / Math.pow(1024, i)).toFixed(1)} ${u[i]}`;
}
function UsageBar({ used, total }: { used: number; total: number }) {
const pct = total > 0 ? Math.min(100, (used / total) * 100) : 0;
return (
<div className="h-2 w-full overflow-hidden rounded-full bg-surface-2">
<div className={`h-full rounded-full ${pct > 90 ? "bg-danger" : "bg-accent"}`} style={{ width: `${pct}%` }} />
</div>
);
}
function InventoryPanel({ inv }: { inv: Inventory }) {
return (
<Card>
<h2 className="mb-4 text-lg font-semibold text-text-primary">Inventory</h2>
<div className="grid gap-4 sm:grid-cols-2">
<div>
<div className="mb-1 flex justify-between text-sm"><span className="text-text-secondary">CPU</span><span className="text-text-primary">{inv.cpu.usage_pct.toFixed(0)}%</span></div>
<UsageBar used={inv.cpu.usage_pct} total={100} />
<p className="mt-1 text-xs text-text-secondary">{inv.cpu.model} · {inv.cpu.cores} cores · load {inv.cpu.load1?.toFixed(2)}</p>
</div>
<div>
<div className="mb-1 flex justify-between text-sm"><span className="text-text-secondary">Memory</span><span className="text-text-primary">{formatBytes(inv.memory.used_bytes)} / {formatBytes(inv.memory.total_bytes)}</span></div>
<UsageBar used={inv.memory.used_bytes} total={inv.memory.total_bytes} />
<div className="mb-1 mt-3 flex justify-between text-sm"><span className="text-text-secondary">Swap</span><span className="text-text-primary">{formatBytes(inv.swap_used_bytes)} / {formatBytes(inv.swap_total_bytes)}</span></div>
<UsageBar used={inv.swap_used_bytes} total={inv.swap_total_bytes} />
</div>
</div>
{inv.partitions && inv.partitions.length > 0 && (
<div className="mt-5">
<h3 className="mb-2 text-sm font-medium text-text-secondary">Partitions</h3>
<div className="space-y-3">
{inv.partitions.map((p) => (
<div key={p.mountpoint}>
<div className="mb-1 flex justify-between text-xs">
<span className="font-mono text-text-primary">{p.mountpoint}</span>
<span className="text-text-secondary">{formatBytes(p.used_bytes)} / {formatBytes(p.total_bytes)} · {p.fstype}</span>
</div>
<UsageBar used={p.used_bytes} total={p.total_bytes} />
</div>
))}
</div>
</div>
)}
{inv.kernel && <p className="mt-4 text-xs text-text-secondary">Kernel {inv.kernel}</p>}
</Card>
);
}
```
Render `{server.inventory && <InventoryPanel inv={server.inventory} />}` in the page body (ensure `Card`, `Inventory` are imported). Match how the page currently reads the server object.
- [ ] **Step 3: Verify build**
Run: `cd web && npm run build`
Expected: success.
- [ ] **Step 4: Commit**
```bash
git add web/lib/api.ts web/app/servers/[id]/page.tsx
git commit -m "feat(web): inventory panel on server detail"
```
---
## Task 6: End-to-end manual verification
- [ ] **Step 1: Build all**
Run: `cd server && go build ./... && cd ../agent && go build ./... && cd ../web && npm run build`
Expected: all succeed.
- [ ] **Step 2: Smoke (if environment available)**
With server + Mongo + a connected Linux agent: within ~30s the server detail page shows CPU %, RAM/swap bars; within 15 min (or on agent restart, which sends a full snapshot immediately) partitions, CPU model and kernel appear. Confirm metrics update roughly every 30s.
- [ ] **Step 3: Commit any fixes**
```bash
git add -A
git commit -m "fix: fleet inventory verification fixes"
```
---
## Self-Review Notes
- **Spec coverage:** §3 model → T1; §4 RPC → T1; §5 collectors + scheduler → T3, T4; §6 handler/store → T2; §7 frontend → T5. Split cadence (30s metrics / 15m static) in T4 scheduler; merge rules preserving static in T2 `StoreInventory`. Tests omitted per Global Constraints.
- **Startup snapshot:** agent sends `Collect(true)` immediately so static fields populate without waiting 15 min.
- **Types consistent:** `InventoryReport` field names identical across proto, both pb files, store service, and TS interface (`usage_pct`, `used_bytes`, `total_bytes`, `swap_*`).
- **Follow-ups (out of scope):** time-series history, usage alerting, Windows collectors, servers-list CPU/RAM badges.
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,747 @@
# Workflow Builder v2 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:** Rebuild the workflow builder to match the approved mockup with drag-and-drop, base-step editing/deletion (cascade), step input parameters, an edit-workflow modal, runs navigation, and fix the save crash.
**Architecture:** Backend gains input-parameter fields on step/ref/resolved models, cascade delete, runner env injection, and a workflow-update handler that returns the updated workflow. Frontend adds `bash`/`pwsh`/`signal` tokens and a `Modal` primitive, then rebuilds the builder page (dotted canvas, 340px node cards, wire env chips, kicker/field inspector, HTML5 drag-and-drop), plus Edit-base-step and Edit-workflow modals and a runs list page.
**Tech Stack:** Go (gin, mongo-driver v2), Next.js 16 app-router + react-query + Tailwind, HTML5 Drag-and-Drop, MongoDB.
## Global Constraints
- **No tests this iteration** — no `*_test.go` or frontend tests. Verify with `go build ./...`, `go vet ./...`, `npm run build`.
- Mongo access pattern: `db.Col("collection_name")` + `context.WithTimeout`. Follow `server/internal/services/workflows.go`.
- Audit every mutation with `services.LogEvent(action, actor, serverID, targetID, message)`.
- Interpreter literals are `"bash"` and `"powershell"`.
- Go module path: `github.com/mrhid6/vantage`.
- Visual target: approved mockup. Builder palette adds amber signal `#f5a524` (`signal`), `#241800` (`signal-ink`), bash `#3fb950`, pwsh `#5b9bff`; keep existing `surface`/`surface-2`/`border`/`text-primary`/`text-secondary`/`danger`/`accent` tokens for panels. Node cards 340px; dotted-grid canvas; dashed-amber `passes` chips on wires; kicker/field inspector.
- Frontend uses `@/lib/api` typed client, `@/components/ui`, react-query. Input styling follows the existing `inputClass` pattern in `web/app/secrets/page.tsx`.
---
## Task 1: Models + step CRUD + cascade delete
**Files:**
- Modify: `server/internal/models/workflow.go`
- Modify: `server/internal/services/workflows.go`
**Interfaces:**
- Produces: `models.InputParam{Name,Default,Description}`; `WorkflowStep.DeclaredInputs`, `WorkflowStepRef.Inputs`, `ResolvedStep.Inputs`. `DeleteStep(stepID)` now cascades to workflows.
- [ ] **Step 1: Add the model fields**
In `server/internal/models/workflow.go` add:
```go
type InputParam struct {
Name string `bson:"name" json:"name"`
Default string `bson:"default" json:"default"`
Description string `bson:"description" json:"description"`
}
```
- In `WorkflowStep`, add after `DeclaredOutputs`: `DeclaredInputs []InputParam `bson:"declared_inputs" json:"declared_inputs"``
- In `WorkflowStepRef`, add: `Inputs map[string]string `bson:"inputs,omitempty" json:"inputs,omitempty"``
- In `ResolvedStep`, add: `Inputs map[string]string `bson:"inputs" json:"inputs"``
- [ ] **Step 2: Persist declared_inputs in CreateStep/UpdateStep**
In `server/internal/services/workflows.go`:
- In `CreateStep`, after the `SecretRefs` nil-guard add:
```go
if s.DeclaredInputs == nil {
s.DeclaredInputs = []models.InputParam{}
}
```
- In `UpdateStep`'s `$set`, add: `"declared_inputs": s.DeclaredInputs,`
- [ ] **Step 3: Cascade DeleteStep to workflows**
Replace the body of `DeleteStep` with a version that also strips the step from every workflow:
```go
func DeleteStep(stepID string) error {
ctx, cancel := wfCtx()
defer cancel()
if _, err := db.Col("workflow_steps").DeleteOne(ctx, bson.M{"step_id": stepID}); err != nil {
return err
}
// Cascade: remove this step from every workflow that references it, re-sequencing orders.
cur, err := db.Col("workflows").Find(ctx, bson.M{"steps.step_id": stepID})
if err != nil {
return err
}
defer cur.Close(ctx)
var wfs []models.Workflow
if err := cur.All(ctx, &wfs); err != nil {
return err
}
for _, w := range wfs {
kept := make([]models.WorkflowStepRef, 0, len(w.Steps))
for _, ref := range w.Steps {
if ref.StepID == stepID {
continue
}
kept = append(kept, ref)
}
for i := range kept {
kept[i].Order = i
}
if _, err := db.Col("workflows").UpdateOne(ctx,
bson.M{"workflow_id": w.WorkflowID},
bson.M{"$set": bson.M{"steps": kept, "updated_at": time.Now()}},
); err != nil {
return err
}
}
return nil
}
```
Confirm `models` and `time` are imported in this file (they are).
- [ ] **Step 4: Verify build**
Run: `cd server && go build ./... && go vet ./...`
Expected: success.
- [ ] **Step 5: Commit**
```bash
git add server/internal/models/workflow.go server/internal/services/workflows.go
git commit -m "feat(workflows): step input params model + cascade step delete"
```
---
## Task 2: Runner input injection + update handler returns workflow
**Files:**
- Modify: `server/internal/services/workflow_runner.go`
- Modify: `server/internal/api/workflows.go`
**Interfaces:**
- Consumes: `models.ResolvedStep.Inputs`, `WorkflowStep.DeclaredInputs`, `WorkflowStepRef.Inputs` (Task 1).
- Produces: runner injects input env; `PUT /api/workflows/:id` returns the `Workflow`.
- [ ] **Step 1: Resolve inputs in `resolveSteps`**
In `server/internal/services/workflow_runner.go`, in `resolveSteps`, after loading `lib` and before/where the `ResolvedStep` is built, compute the input env and set it. Add inside the loop (after `lib, err := getStep(...)` succeeds):
```go
inputs := map[string]string{}
for _, p := range lib.DeclaredInputs {
if ref.Inputs != nil {
if v, ok := ref.Inputs[p.Name]; ok {
inputs[p.Name] = v
continue
}
}
inputs[p.Name] = p.Default
}
```
and set `Inputs: inputs,` on the `models.ResolvedStep{...}` literal.
- [ ] **Step 2: Inject inputs into cmdEnv in `runServer`**
In `runServer`, where `cmdEnv` is built (currently: copy `runEnv`, then overlay `secretVals`), change the layering so inputs are the base layer:
```go
cmdEnv := map[string]string{}
for k, v := range step.Inputs {
cmdEnv[k] = v
}
for k, v := range runEnv {
cmdEnv[k] = v
}
for k, v := range secretVals {
cmdEnv[k] = v
}
```
(Inputs first = lowest precedence; upstream outputs override; secrets win. Keep the existing `secretVals`/`allSecrets` masking logic unchanged.)
- [ ] **Step 3: `updateWorkflow` handler returns the workflow**
In `server/internal/api/workflows.go`, change `updateWorkflow` so that after a successful `services.UpdateWorkflow`, it re-fetches and returns the workflow:
```go
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")
updated, err := services.GetWorkflow(c.Param("id"))
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
}
c.JSON(http.StatusOK, updated)
}
```
- [ ] **Step 4: Verify build**
Run: `cd server && go build ./... && go vet ./...`
Expected: success.
- [ ] **Step 5: Commit**
```bash
git add server/internal/services/workflow_runner.go server/internal/api/workflows.go
git commit -m "feat(workflows): inject step inputs into env; update returns workflow"
```
---
## Task 3: Frontend tokens, Modal primitive, API types
**Files:**
- Modify: `web/tailwind.config.ts`
- Create: `web/components/ui/Modal.tsx`
- Modify: `web/components/ui/index.ts`
- Modify: `web/lib/api.ts`
**Interfaces:**
- Produces: tokens `bash`/`pwsh`/`signal`/`signal-ink`; `<Modal open title onClose>children</Modal>`; TS `InputParam`, `WorkflowStep.declared_inputs`, `WorkflowStepRef.inputs`.
- [ ] **Step 1: Add tokens**
In `web/tailwind.config.ts`, add to `theme.extend.colors`:
```ts
bash: "#3fb950",
pwsh: "#5b9bff",
signal: "#f5a524",
"signal-ink": "#241800",
```
- [ ] **Step 2: Create the Modal primitive**
`web/components/ui/Modal.tsx`:
```tsx
"use client";
import { useEffect } from "react";
export function Modal({
open,
title,
onClose,
children,
wide,
}: {
open: boolean;
title: string;
onClose: () => void;
children: React.ReactNode;
wide?: boolean;
}) {
useEffect(() => {
if (!open) return;
const onKey = (e: KeyboardEvent) => e.key === "Escape" && onClose();
window.addEventListener("keydown", onKey);
return () => window.removeEventListener("keydown", onKey);
}, [open, onClose]);
if (!open) return null;
return (
<div className="fixed inset-0 z-50 flex items-center justify-center p-4">
<div className="absolute inset-0 bg-black/60" onClick={onClose} />
<div
className={`relative z-10 w-full ${wide ? "max-w-2xl" : "max-w-md"} max-h-[90vh] overflow-auto rounded-xl border border-border bg-surface shadow-2xl`}
role="dialog"
aria-modal="true"
>
<div className="flex items-center justify-between border-b border-border px-5 py-3">
<h2 className="text-sm font-bold text-text-primary">{title}</h2>
<button onClick={onClose} className="text-text-secondary hover:text-text-primary" aria-label="Close">
</button>
</div>
<div className="p-5">{children}</div>
</div>
</div>
);
}
```
- [ ] **Step 3: Export it**
In `web/components/ui/index.ts` add: `export { Modal } from "./Modal";`
- [ ] **Step 4: API types**
In `web/lib/api.ts`:
- Add interface:
```ts
export interface InputParam {
name: string;
default: string;
description: string;
}
```
- In `WorkflowStep`, add `declared_inputs: InputParam[];`
- In `WorkflowStepRef`, add `inputs?: Record<string, string>;`
(`updateWorkflow` already typed to return `Workflow`; the backend now honors it.)
- [ ] **Step 5: Verify build**
Run: `cd web && npm run build`
Expected: success.
- [ ] **Step 6: Commit**
```bash
git add web/tailwind.config.ts web/components/ui/Modal.tsx web/components/ui/index.ts web/lib/api.ts
git commit -m "feat(web): builder tokens, Modal primitive, input-param types"
```
---
## Task 4: Edit-base-step modal
**Files:**
- Create: `web/components/workflows/EditStepModal.tsx`
**Interfaces:**
- Consumes: `api.createStep`, `api.updateStep`, `api.deleteStep`, `Modal`, types (Task 3).
- Produces: `<EditStepModal open step onClose />` where `step` is a `WorkflowStep` (edit) or `null` (new).
- [ ] **Step 1: Write the modal**
Client component. Local form state seeded from `step` (or blank for new). Fields:
- **Name** (text), **Interpreter** (select bash/powershell), **Script** (`<textarea>` mono).
- **Outputs** (`declared_outputs`): a list of text chips with add/remove — an input + "Add" appends a name; each name shows an `✕` to remove.
- **Inputs** (`declared_inputs`): rows, each with `name` / `default` / `description` inputs and a remove button; an "Add input" button appends a blank `{name:"",default:"",description:""}`.
- Hint: "Reusable steps are shared across all workflows. Editing here changes it everywhere."
Actions:
- **Save**: build the `WorkflowStep` payload (`declared_outputs`, `declared_inputs`, `secret_refs: step?.secret_refs ?? []`, `description: ""` if absent). If editing (`step` truthy) call `api.updateStep(step.step_id, payload)`, else `api.createStep(payload)`. On success `queryClient.invalidateQueries({queryKey:["steps"]})` and `onClose()`.
- **Delete** (edit mode only): a confirm (`window.confirm`) then `api.deleteStep(step.step_id)`, invalidate `["steps"]` AND `["workflow"]` (broad — deleted step is pulled from workflows server-side), `onClose()`.
Use the `inputClass` styling pattern and `Button` variants (`primary` save, `danger` delete, `ghost` cancel). Follow existing token names. Filter empty input rows (blank `name`) out of the payload on save.
Skeleton:
```tsx
"use client";
import { useState } from "react";
import { useQueryClient } from "@tanstack/react-query";
import { api, WorkflowStep, InputParam } from "@/lib/api";
import { Button, Modal } from "@/components/ui";
const inputClass =
"w-full rounded-lg border border-border bg-surface-2 px-3 py-2 text-sm text-text-primary placeholder-text-secondary/50 focus:border-signal focus:outline-none focus:ring-1 focus:ring-signal";
export function EditStepModal({ open, step, onClose }: { open: boolean; step: WorkflowStep | null; onClose: () => void }) {
const qc = useQueryClient();
const [name, setName] = useState(step?.name ?? "");
const [interpreter, setInterpreter] = useState<"bash" | "powershell">(step?.interpreter ?? "bash");
const [script, setScript] = useState(step?.script ?? "");
const [outputs, setOutputs] = useState<string[]>(step?.declared_outputs ?? []);
const [inputs, setInputs] = useState<InputParam[]>(step?.declared_inputs ?? []);
const [newOut, setNewOut] = useState("");
const [busy, setBusy] = useState(false);
const [error, setError] = useState<string | null>(null);
// NOTE: because state is seeded from props, render the modal conditionally
// (parent mounts it only when opening) OR key it by step_id so it re-seeds.
const save = async () => {
setBusy(true); setError(null);
try {
const payload = {
name: name.trim(), description: step?.description ?? "", interpreter, script,
declared_outputs: outputs, declared_inputs: inputs.filter((i) => i.name.trim() !== ""),
secret_refs: step?.secret_refs ?? [],
};
if (step) await api.updateStep(step.step_id, payload);
else await api.createStep(payload);
qc.invalidateQueries({ queryKey: ["steps"] });
onClose();
} catch (e) { setError((e as Error).message); } finally { setBusy(false); }
};
const del = async () => {
if (!step || !window.confirm("Delete this step? It will be removed from every workflow that uses it.")) return;
setBusy(true); setError(null);
try {
await api.deleteStep(step.step_id);
qc.invalidateQueries({ queryKey: ["steps"] });
qc.invalidateQueries({ queryKey: ["workflow"] });
onClose();
} catch (e) { setError((e as Error).message); } finally { setBusy(false); }
};
return (
<Modal open={open} onClose={onClose} title={step ? "Edit base step" : "New step"} wide>
<div className="space-y-4">
{error && <div className="rounded border border-danger/30 bg-danger/10 px-3 py-2 text-sm text-danger">{error}</div>}
<div>
<label className="mb-1 block text-xs uppercase text-text-secondary">Name</label>
<input className={inputClass} value={name} onChange={(e) => setName(e.target.value)} />
</div>
<div>
<label className="mb-1 block text-xs uppercase text-text-secondary">Interpreter</label>
<select className={inputClass} value={interpreter} onChange={(e) => setInterpreter(e.target.value as "bash" | "powershell")}>
<option value="bash">bash</option>
<option value="powershell">powershell</option>
</select>
</div>
<div>
<label className="mb-1 block text-xs uppercase text-text-secondary">Script</label>
<textarea className={`${inputClass} h-40 font-mono text-xs`} value={script} onChange={(e) => setScript(e.target.value)} />
<p className="mt-1 text-xs text-text-secondary">Write <code className="text-signal">KEY=value</code> to <code className="text-signal">$WORKFLOW_ENV</code> to expose it to later steps.</p>
</div>
<div>
<label className="mb-1 block text-xs uppercase text-text-secondary">Outputs</label>
<div className="mb-2 flex flex-wrap gap-1">
{outputs.map((o) => (
<span key={o} className="flex items-center gap-1 rounded bg-signal px-2 py-0.5 font-mono text-[11px] text-signal-ink">
{o}<button onClick={() => setOutputs(outputs.filter((x) => x !== o))}></button>
</span>
))}
</div>
<div className="flex gap-2">
<input className={inputClass} placeholder="OUTPUT_NAME" value={newOut} onChange={(e) => setNewOut(e.target.value)} />
<Button variant="ghost" size="sm" onClick={() => { if (newOut.trim()) { setOutputs([...outputs, newOut.trim()]); setNewOut(""); } }}>Add</Button>
</div>
</div>
<div>
<label className="mb-1 block text-xs uppercase text-text-secondary">Inputs</label>
<div className="space-y-2">
{inputs.map((inp, i) => (
<div key={i} className="flex gap-2">
<input className={inputClass} placeholder="name" value={inp.name} onChange={(e) => setInputs(inputs.map((x, j) => j === i ? { ...x, name: e.target.value } : x))} />
<input className={inputClass} placeholder="default" value={inp.default} onChange={(e) => setInputs(inputs.map((x, j) => j === i ? { ...x, default: e.target.value } : x))} />
<input className={inputClass} placeholder="description" value={inp.description} onChange={(e) => setInputs(inputs.map((x, j) => j === i ? { ...x, description: e.target.value } : x))} />
<Button variant="ghost" size="sm" onClick={() => setInputs(inputs.filter((_, j) => j !== i))}></Button>
</div>
))}
</div>
<Button variant="ghost" size="sm" className="mt-2" onClick={() => setInputs([...inputs, { name: "", default: "", description: "" }])}>Add input</Button>
</div>
<div className="flex items-center justify-between pt-2">
{step ? <Button variant="danger" onClick={del} loading={busy}>Delete step</Button> : <span />}
<div className="flex gap-2">
<Button variant="ghost" onClick={onClose}>Cancel</Button>
<Button variant="primary" onClick={save} loading={busy} disabled={!name.trim()}>Save</Button>
</div>
</div>
</div>
</Modal>
);
}
```
**Important:** because the form state seeds from `step` props, the parent must either mount `EditStepModal` only while open, or pass a React `key={step?.step_id ?? "new"}` so the state re-initializes each time a different step is edited. Note this in the component with a comment; the builder (Task 6) will mount it with a `key`.
- [ ] **Step 2: Verify build**
Run: `cd web && npm run build`
Expected: success (the component may be unused until Task 6 — that's fine, but an unused import-free component builds).
- [ ] **Step 3: Commit**
```bash
git add web/components/workflows/EditStepModal.tsx
git commit -m "feat(web): edit-base-step modal with inputs/outputs editor"
```
---
## Task 5: Edit-workflow modal
**Files:**
- Create: `web/components/workflows/EditWorkflowModal.tsx`
**Interfaces:**
- Consumes: `api.updateWorkflow`, `api.deleteWorkflow`, `api.listServers`, `Modal`, `Workflow` (Task 3).
- Produces: `<EditWorkflowModal open workflow onSaved onClose />` — saves name/targets immediately, returns the updated workflow via `onSaved`.
- [ ] **Step 1: Write the modal**
Client component. Seeds name + selected target servers from `workflow`. Uses `api.listServers` (react-query) for the multiselect (checkbox chips keyed by `server_id`, labelled `hostname`). Actions:
- **Save**: `const updated = await api.updateWorkflow(workflow.workflow_id, { ...workflow, name, target_server_ids })`; call `onSaved(updated)`; `onClose()`. (Backend now returns the workflow.)
- **Delete workflow**: confirm, `api.deleteWorkflow(workflow.workflow_id)`, then `router.push("/workflows")`.
```tsx
"use client";
import { useState } from "react";
import { useRouter } from "next/navigation";
import { useQuery } from "@tanstack/react-query";
import { api, Workflow } from "@/lib/api";
import { Button, Modal } from "@/components/ui";
const inputClass =
"w-full rounded-lg border border-border bg-surface-2 px-3 py-2 text-sm text-text-primary focus:border-signal focus:outline-none focus:ring-1 focus:ring-signal";
export function EditWorkflowModal({ open, workflow, onSaved, onClose }: { open: boolean; workflow: Workflow; onSaved: (w: Workflow) => void; onClose: () => void }) {
const router = useRouter();
const [name, setName] = useState(workflow.name);
const [targets, setTargets] = useState<string[]>(workflow.target_server_ids);
const [busy, setBusy] = useState(false);
const [error, setError] = useState<string | null>(null);
const { data: servers } = useQuery({ queryKey: ["servers"], queryFn: api.listServers });
const toggle = (id: string) => setTargets((t) => (t.includes(id) ? t.filter((x) => x !== id) : [...t, id]));
const save = async () => {
setBusy(true); setError(null);
try {
const updated = await api.updateWorkflow(workflow.workflow_id, { ...workflow, name, target_server_ids: targets });
onSaved(updated); onClose();
} catch (e) { setError((e as Error).message); } finally { setBusy(false); }
};
const del = async () => {
if (!window.confirm("Delete this workflow? This cannot be undone.")) return;
setBusy(true); setError(null);
try { await api.deleteWorkflow(workflow.workflow_id); router.push("/workflows"); }
catch (e) { setError((e as Error).message); setBusy(false); }
};
return (
<Modal open={open} onClose={onClose} title="Edit workflow">
<div className="space-y-4">
{error && <div className="rounded border border-danger/30 bg-danger/10 px-3 py-2 text-sm text-danger">{error}</div>}
<div>
<label className="mb-1 block text-xs uppercase text-text-secondary">Name</label>
<input className={inputClass} value={name} onChange={(e) => setName(e.target.value)} />
</div>
<div>
<label className="mb-1 block text-xs uppercase text-text-secondary">Target servers</label>
<div className="flex flex-wrap gap-2">
{servers?.map((s) => {
const on = targets.includes(s.server_id);
return (
<label key={s.server_id} className={`flex cursor-pointer items-center gap-2 rounded-lg border px-2 py-1 text-sm ${on ? "border-signal bg-signal/10 text-text-primary" : "border-border text-text-secondary"}`}>
<input type="checkbox" className="accent-signal" checked={on} onChange={() => toggle(s.server_id)} />
{s.hostname}
</label>
);
})}
{servers && servers.length === 0 && <p className="text-xs text-text-secondary">No servers registered.</p>}
</div>
</div>
<div className="flex items-center justify-between pt-2">
<Button variant="danger" onClick={del} loading={busy}>Delete workflow</Button>
<div className="flex gap-2">
<Button variant="ghost" onClick={onClose}>Cancel</Button>
<Button variant="primary" onClick={save} loading={busy} disabled={!name.trim()}>Save</Button>
</div>
</div>
</div>
</Modal>
);
}
```
- [ ] **Step 2: Verify build**
Run: `cd web && npm run build`
Expected: success.
- [ ] **Step 3: Commit**
```bash
git add web/components/workflows/EditWorkflowModal.tsx
git commit -m "feat(web): edit-workflow modal (name/targets/delete)"
```
---
## Task 6: Builder rewrite — mockup styling + drag-and-drop + inspector
**Files:**
- Modify: `web/app/workflows/[id]/page.tsx`
**Interfaces:**
- Consumes: `api.getWorkflow/updateWorkflow/listSteps/runWorkflow`, `EditStepModal` (Task 4), `EditWorkflowModal` (Task 5), tokens (Task 3).
Rebuild the page to the approved mockup. Reference values from the mockup:
- Canvas dotted grid: `background: radial-gradient(circle at 1px 1px, <border> 1px, transparent 0) 0 0 / 22px 22px` over `bg-background`.
- Node card: width `340px`, `rounded-[10px] border border-border bg-surface`, selected → `border-signal` + `ring-2 ring-signal/40`. Top row: index badge (mono, boxed), title, shell badge, then body with a mono script preview (`bg-surface-2 border border-border rounded p-2 text-xs`).
- Shell badges: bash → `text-bash bg-bash/15`, pwsh (powershell) → `text-pwsh bg-pwsh/15`, mono uppercase `text-[10px] px-1.5 py-0.5 rounded`.
- Wire: a 2px vertical `bg-border` segment ~26px tall between nodes.
- `passes` chip row: dashed amber pill `border border-dashed border-signal/55 bg-surface rounded-full px-3 py-1`, label "passes" (`text-[10px] uppercase text-text-secondary`), each output name a chip `bg-signal text-signal-ink font-mono text-[11px] rounded px-2 py-0.5`.
- Inspector fields: kicker (`text-[11px] uppercase tracking-wide text-text-secondary font-bold`), title with shell badge; each field block separated by `border-b border-border` with an uppercase label.
- [ ] **Step 1: Rewrite the builder page**
Requirements (keep everything in React state; **Save** persists via `api.updateWorkflow` and sets `wf` to the returned workflow):
1. **State/data:** load workflow (`["workflow", id]`), library (`["steps"]`), servers (`["servers"]`). Seed local `wf` from the query once. Keep `selected` (index in sorted order). `sortedSteps = [...wf.steps].sort(order)`.
2. **Topbar:** brand dot + `Workflows /` crumb + `wf.name` + `· draft`. Right: a **Targets** chip showing `${wf.target_server_ids.length} servers`; a **Runs** link (`<Link href={`/workflows/${id}/runs`}>`); an **Edit** button (opens `EditWorkflowModal`); **Save** (`variant="secondary"`, persists, `setWf(returned)`); **Run workflow** (amber: `className` using `bg-signal text-signal-ink`, or `variant="primary"` acceptable) → `api.runWorkflow` then route to the run detail.
3. **Library (left):** header "Step Library" + `+` button opening `EditStepModal` in new mode (`step={null}`). A search input filters `library` by name (case-insensitive). Group by interpreter with labels `Shared · Bash` / `Shared · PowerShell`. Each card: shell badge, name, `description`, a grip glyph (`⠿`), `cursor-grab`, and `draggable`. On `dragstart` set `dataTransfer` to `JSON.stringify({kind:"lib", stepId})`. Clicking the card still appends the step. A small edit affordance (e.g. a pencil `✎` button on hover) opens `EditStepModal` with that library step.
4. **Canvas (center):** dotted grid. For each `sortedSteps[i]`: render a drop target above it (a thin zone; on `drop` insert at `i`), then the node card. Node card is `draggable` (`dragstart` sets `{kind:"move", from:i}`). Between consecutive nodes render the wire + `passes` chips = union of `sortedSteps.slice(0,i)` `declared_outputs` (deduped). After the last node render an end drop zone styled `+ Drop a step here` (dashed). Handle `drop`:
- parse `dataTransfer`; if `kind==="lib"` insert a new `WorkflowStepRef{step_id, order:<pos>, on_failure:"stop", max_retries:0}` at the drop position; if `kind==="move"` move `from``pos`; then re-sequence all `order` to array index. Update `wf`.
- `dragover` must `e.preventDefault()` on drop zones to allow dropping.
- Selecting a node (click) sets `selected`.
5. **Inspector (right):** for the selected placement (`selectedRef` = `sortedSteps[selected]`, `selectedLib` = library by `step_id`):
- Kicker `Step ${selected+1} · Inspector`, title = shell badge + `selectedLib.name`.
- **Command** `<textarea>` bound to `selectedRef.overrides?.script ?? selectedLib.script`; edits write `overrides.script` on the ref (per-placement override). Hint about `$WORKFLOW_ENV`.
- **Inputs** (this step's `declared_inputs`): one row per param — label = `param.name` (+ description as sub-text), an input bound to `selectedRef.inputs?.[name] ?? ""` with `placeholder={param.default}`; edits write `ref.inputs[name]`.
- **Inputs · from upstream** (read-only): the union of prior steps' `declared_outputs` as `IN` rows.
- **Outputs · to $WORKFLOW_ENV** (read-only): this step's `declared_outputs` as `OUT` rows.
- **Secret refs**: keep the existing group/KEY checklist behavior (port it over) writing `overrides.secret_refs`.
- **On failure** select + **Max retries** (when retry).
- **Remove from workflow** button (danger) → remove the ref, re-sequence orders, clear selection.
6. **Modals:** render `<EditWorkflowModal>` (open state) and `<EditStepModal key={editingStep?.step_id ?? "new"} open step={editingStep} onClose>`; the library `+` and per-card edit set `editingStep`.
7. **Bug fix:** `save()` sets `wf` to the awaited `updateWorkflow` result (now a full workflow); no direct crash. Additionally guard: if the returned object lacks `steps`, keep the prior `wf` and surface an error.
Provide a complete, working implementation (this is a full rewrite of the file). Preserve the secret-group lazy-fetch (`api.getSecretGroup`) logic from the current file for the secret-refs checklist. Use `inputClass` styling. Do not leave TODOs.
- [ ] **Step 2: Verify build**
Run: `cd web && npm run build`
Expected: type-checks and builds. Manually confirm no `st.stdout`-style dead references and no unused imports.
- [ ] **Step 3: Commit**
```bash
git add web/app/workflows/[id]/page.tsx
git commit -m "feat(web): rebuild workflow builder — mockup styling, drag-and-drop, inputs inspector"
```
---
## Task 7: Runs list page + navigation links
**Files:**
- Create: `web/app/workflows/[id]/runs/page.tsx`
- Modify: `web/app/workflows/page.tsx`
**Interfaces:**
- Consumes: `api.listRuns`, `api.getWorkflow`.
- [ ] **Step 1: Runs list page**
`web/app/workflows/[id]/runs/page.tsx` — client component. Load `api.getWorkflow(id)` (for the name) and `api.listRuns(id)`. Render a header "Runs · <name>" with a "← Back to builder" `Link` to `/workflows/${id}`, and a table (`@/components/ui` Table) of runs: short run id, a status `Badge` (map success→success, failed→danger, running→accent/warning, cancelled→neutral), `started_at` (localized), `triggered_by`, and server count (`server_runs.length`). Each row links to `/workflows/${id}/runs/${run.run_id}`. Empty state "No runs yet."
Use the existing Badge variants (success/warning/danger/neutral/accent) — verify names in `web/components/ui/Badge.tsx`.
Skeleton:
```tsx
"use client";
import Link from "next/link";
import { useParams } from "next/navigation";
import { useQuery } from "@tanstack/react-query";
import { api, WorkflowRun } from "@/lib/api";
import { Card, Table, Thead, Tbody, Tr, Th, Td, Badge } from "@/components/ui";
const statusVariant: Record<string, string> = {
success: "success", failed: "danger", running: "warning", cancelled: "neutral", queued: "neutral",
};
export default function WorkflowRunsPage() {
const { id } = useParams<{ id: string }>();
const { data: wf } = useQuery({ queryKey: ["workflow", id], queryFn: () => api.getWorkflow(id) });
const { data: runs } = useQuery({ queryKey: ["runs", id], queryFn: () => api.listRuns(id) });
return (
<div className="p-8">
<div className="mb-6">
<Link href={`/workflows/${id}`} className="text-sm text-text-secondary hover:text-text-primary"> Back to builder</Link>
<h1 className="mt-2 text-2xl font-bold text-text-primary">Runs · {wf?.name ?? ""}</h1>
</div>
<Card padding={false}>
{runs && runs.length > 0 ? (
<Table>
<Thead><Tr><Th>Run</Th><Th>Status</Th><Th>Started</Th><Th>By</Th><Th>Servers</Th></Tr></Thead>
<Tbody>
{runs.map((r: WorkflowRun) => (
<Tr key={r.run_id}>
<Td><Link href={`/workflows/${id}/runs/${r.run_id}`} className="font-mono text-text-primary hover:text-signal">{r.run_id.slice(0, 8)}</Link></Td>
<Td><Badge variant={(statusVariant[r.status] ?? "neutral") as never}>{r.status}</Badge></Td>
<Td className="text-text-secondary">{new Date(r.started_at).toLocaleString()}</Td>
<Td className="text-text-secondary">{r.triggered_by}</Td>
<Td className="text-text-secondary">{r.server_runs.length}</Td>
</Tr>
))}
</Tbody>
</Table>
) : (
<div className="py-16 text-center text-text-secondary">No runs yet.</div>
)}
</Card>
</div>
);
}
```
Adapt `Card padding={false}` / Badge variant prop to the real component signatures (check `Card.tsx`/`Badge.tsx`).
- [ ] **Step 2: Add a Runs link on the workflows list**
In `web/app/workflows/page.tsx`, in each workflow row add a "Runs" link/button next to "Open" → `Link href={`/workflows/${w.workflow_id}/runs`}`. Match the existing row action styling.
- [ ] **Step 3: Verify build**
Run: `cd web && npm run build`
Expected: success.
- [ ] **Step 4: Commit**
```bash
git add web/app/workflows/[id]/runs/page.tsx web/app/workflows/page.tsx
git commit -m "feat(web): workflow runs list page and navigation links"
```
---
## Task 8: End-to-end verification
**Files:** none (verification only).
- [ ] **Step 1: Build everything**
Run: `cd server && go build ./... && go vet ./... && cd ../web && npm run build`
Expected: all succeed.
- [ ] **Step 2: Manual smoke (documented, run if an environment is available)**
1. Open a workflow builder; confirm it matches the mockup (dotted canvas, node cards, amber passes chips, kicker/field inspector).
2. Drag a library step onto the canvas; drag to reorder; confirm order persists after Save (no "steps is not iterable" crash).
3. Open Edit base step; add an input param (name/default/description) and an output; Save. Place the step; set the input value in the inspector; Run; confirm the script sees the input env var and downstream steps see the output.
4. Delete a shared step from the Edit-base-step modal; confirm it disappears from every workflow that used it.
5. Open Edit workflow; rename, change target servers, Save; confirm persisted. Delete a throwaway workflow; confirm redirect to `/workflows`.
6. From the workflows list and the builder topbar, navigate to Runs; open a run.
- [ ] **Step 3: Commit any fixes found**
```bash
git add -A
git commit -m "fix: workflow builder v2 e2e fixes"
```
---
## Self-Review Notes
- **Spec coverage:** §3 model → T1; §4.1 cascade/CRUD → T1; §4.2 update-returns-workflow → T2; §4.3 runner inputs → T2; §6.1 tokens/Modal → T3; §6.2 api types → T3; §6.4 edit-step modal → T4; §6.5 edit-workflow modal → T5; §6.3 builder restyle/DnD/inspector → T6; §6.6 runs page + nav → T7. Save crash fixed by T2 (server) + T6 (client guard). Tests omitted per Global Constraints.
- **Dependency order:** modals (T4, T5) land before the builder (T6) that imports them; tokens/Modal/api (T3) first.
- **Interpreter literals** `"bash"`/`"powershell"` consistent across T1/T4/T6.
- **Modal state seeding:** EditStepModal re-seeds via `key` in the builder (documented in T4/T6).
- **Open follow-ups (out of scope):** typed/required inputs, drag-to-trash removal, live status in canvas, keyboard reordering.
```
@@ -0,0 +1,887 @@
# Workflow Log Streaming 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:** Stream workflow step output live from agents to per-server-run log files on the server, tail them live in the UI over SSE, and auto-expire them on a configurable retention period.
**Architecture:** Agent streams interleaved stdout/stderr chunks over the existing `CommandStream` (`AgentMessage.StepOutput`). Server appends secret-masked chunks to `<logdir>/<run_id>/<server_id>.log` via a per-command log-writer registry, records a per-step byte offset, and stops persisting log bodies in Mongo. UI tails via an SSE endpoint while running and fetches the whole file after. An hourly sweeper deletes run-log dirs older than the retention setting.
**Tech Stack:** Go (gin, mongo-driver v2), hand-written JSON-codec gRPC structs (no protoc), Next.js 16 app-router + react-query + EventSource, MongoDB, local filesystem for logs.
## 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`. No codegen. Also update `proto/vantage/v1/vantage.proto` as documentation.
- Mongo access pattern: `db.Col("collection_name")` with `context.WithTimeout`. Follow `server/internal/services/workflows.go`.
- Secret values must never be written into log files unmasked — mask by literal `***` replacement at write time, boundary-safe via a carry buffer.
- Interpreter values are the literals `"bash"` and `"powershell"`.
- Go module path: `github.com/mrhid6/vantage`.
- Log dir from env `VANTAGE_WORKFLOW_LOG_DIR`, default `<data>/workflow-logs`; files `0600`, dirs `0700`.
- Retention default **30** days, stored `settings.workflow_log_retention_days`; `0`/negative = keep forever.
- The agent's stream `Send` is only safe through the existing per-connection mutex-guarded `send()` closure in `connectAndHandleStream` — all `StepOutput`/`StepResult` sends MUST go through it.
---
## Task 1: Proto/pb — StepOutputChunk
**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.StepOutputChunk{CommandId string, Seq uint64, Data []byte, Eof bool}`; `pb.AgentMessage` gains `StepOutput *StepOutputChunk`.
- [ ] **Step 1: Document in the proto file**
In `proto/vantage/v1/vantage.proto`, add to the `AgentMessage` oneof: `StepOutputChunk step_output = 6;` and add the message:
```protobuf
message StepOutputChunk {
string command_id = 1;
uint64 seq = 2;
bytes data = 3;
bool eof = 4;
}
```
- [ ] **Step 2: Add struct + field to server pb file**
In `server/internal/grpc/pb/vantage.pb.go`, add to `type AgentMessage struct { ... }`:
```go
StepOutput *StepOutputChunk `json:"step_output,omitempty"`
```
and add the new struct:
```go
type StepOutputChunk struct {
CommandId string `json:"command_id"`
Seq uint64 `json:"seq"`
Data []byte `json:"data,omitempty"`
Eof bool `json:"eof,omitempty"`
}
```
- [ ] **Step 3: Mirror identical additions into the agent pb file**
Apply the identical `AgentMessage.StepOutput` field and `StepOutputChunk` struct to `agent/internal/grpc/pb/vantage.pb.go`.
- [ ] **Step 4: Verify build**
Run: `cd server && go build ./... && cd ../agent && go build ./...`
Expected: both succeed.
- [ ] **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 StepOutputChunk streaming message"
```
---
## Task 2: Agent — stream step output
**Files:**
- Modify: `agent/internal/exec/exec.go`
- Modify: `agent/internal/sync/sync.go` (the `cmd.RunStep != nil` goroutine)
**Interfaces:**
- Consumes: `pb.RunStepCmd`, `pb.StepResult`, `pb.StepOutputChunk` (Task 1).
- Produces: `exec.RunStep(cmd *pb.RunStepCmd, emit func(seq uint64, data []byte)) *pb.StepResult` — streams output via `emit`, returns terminal result with empty stdout/stderr but populated exit_code/output_env.
- [ ] **Step 1: Rework `exec.RunStep` to stream**
In `agent/internal/exec/exec.go`, change the signature and replace the two `bytes.Buffer`s with a single mutex-guarded streaming writer. Full new body of the run/capture section (keep the existing temp-dir, env-file, interpreter-selection, timeout, and `parseEnvFile` logic exactly as-is):
Add this type at package scope:
```go
// streamWriter forwards every write to emit() as an ordered chunk. Used as both
// Stdout and Stderr so output interleaves in real execution order. The mutex
// ensures a single stdout/stderr write is not interleaved mid-slice with another.
type streamWriter struct {
mu sync.Mutex
seq uint64
emit func(seq uint64, data []byte)
}
func (w *streamWriter) Write(p []byte) (int, error) {
w.mu.Lock()
defer w.mu.Unlock()
if w.emit != nil {
buf := make([]byte, len(p))
copy(buf, p)
w.emit(w.seq, buf)
w.seq++
}
return len(p), nil
}
```
Add `"sync"` to the imports. Change the signature to:
```go
func RunStep(cmd *pb.RunStepCmd, emit func(seq uint64, data []byte)) *pb.StepResult {
```
Replace the block that currently declares `var stdout, stderr bytes.Buffer`, assigns `c.Stdout`/`c.Stderr`, and sets `res.Stdout`/`res.Stderr` from them, with:
```go
sw := &streamWriter{emit: emit}
c.Stdout = sw
c.Stderr = sw
runErr := c.Run()
// stdout/stderr are streamed via emit, not returned in the result.
if ctx.Err() == context.DeadlineExceeded {
res.ExitCode = 124
res.Stderr = "[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 = "[vantage] " + runErr.Error()
}
res.OutputEnv = parseEnvFile(envFile)
return res
```
Remove the now-unused `"bytes"` and `"bufio"` imports **only if** they are no longer referenced (`parseEnvFile` uses `bufio` + `os` — keep `bufio`; `bytes` is likely now unused — remove it if so). Verify with `go build`.
- [ ] **Step 2: Wire streaming into the agent loop**
In `agent/internal/sync/sync.go`, the `cmd.RunStep != nil` goroutine currently calls `agentexec.RunStep(rc)` and sends one `StepResult` via `send()`. Change it to pass an `emit` closure that streams chunks, then send an eof chunk, then the terminal result — all through the existing mutex-guarded `send()`:
```go
if cmd.RunStep != nil {
go func(rc *pb.RunStepCmd, cid string) {
emit := func(seq uint64, data []byte) {
_ = send(&pb.AgentMessage{
ServerId: cfg.ServerID,
AgentToken: cfg.AgentToken,
StepOutput: &pb.StepOutputChunk{CommandId: cid, Seq: seq, Data: data},
})
}
res := agentexec.RunStep(rc, emit)
res.CommandId = cid
// Final eof marker so the server closes the log file.
_ = send(&pb.AgentMessage{
ServerId: cfg.ServerID,
AgentToken: cfg.AgentToken,
StepOutput: &pb.StepOutputChunk{CommandId: cid, Eof: true},
})
_ = send(&pb.AgentMessage{
ServerId: cfg.ServerID,
AgentToken: cfg.AgentToken,
StepResult: res,
})
}(cmd.RunStep, cmd.CommandId)
continue
}
```
(Match the exact field names already used by the existing `send()` calls in this function — `cfg.ServerID`, `cfg.AgentToken`, and the `send` closure. If the existing RunStep branch used different local names, keep those.)
- [ ] **Step 3: Verify build**
Run: `cd agent && go build ./... && go vet ./...`
Expected: success. Resolve any leftover unused-import error from Step 1.
- [ ] **Step 4: Commit**
```bash
git add agent/internal/exec/exec.go agent/internal/sync/sync.go
git commit -m "feat(agent): stream step output chunks over CommandStream"
```
---
## Task 3: Server log-writer registry + retention sweeper
**Files:**
- Create: `server/internal/services/steplogs.go`
**Interfaces:**
- Consumes: `settings` service (retention), `db.Col("workflow_runs")` (sweeper), env `VANTAGE_WORKFLOW_LOG_DIR`.
- Produces:
- `WorkflowLogDir() string` — resolved base dir (env or default), created on first call.
- `ServerRunLogPath(runID, serverID string) string``<logdir>/<runID>/<serverID>.log`.
- `AppendMarker(runID, serverID, line string) (int64, error)` — appends a marker line, returns the byte offset **before** the write (the step's `log_offset`).
- `var StepLogs *stepLogRegistry` with `Open(commandID, path string, secrets []string) error`, `Append(commandID string, data []byte)`, `Close(commandID string)`.
- `StartLogSweeper()` — launches the hourly retention goroutine; also sweeps once immediately.
- [ ] **Step 1: Write the registry, paths, and sweeper**
```go
package services
import (
"os"
"path/filepath"
"strings"
"sync"
"time"
"github.com/mrhid6/vantage/server/internal/db"
"go.mongodb.org/mongo-driver/v2/bson"
)
// WorkflowLogDir returns the base directory for workflow step logs, creating it.
func WorkflowLogDir() string {
dir := os.Getenv("VANTAGE_WORKFLOW_LOG_DIR")
if dir == "" {
dir = filepath.Join("data", "workflow-logs")
}
_ = os.MkdirAll(dir, 0700)
return dir
}
// ServerRunLogPath is the per-server-run log file path.
func ServerRunLogPath(runID, serverID string) string {
return filepath.Join(WorkflowLogDir(), runID, serverID+".log")
}
// AppendMarker appends a line to the server-run log and returns the byte offset
// at which the write began (used as a step's log_offset).
func AppendMarker(runID, serverID, line string) (int64, error) {
path := ServerRunLogPath(runID, serverID)
if err := os.MkdirAll(filepath.Dir(path), 0700); err != nil {
return 0, err
}
f, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0600)
if err != nil {
return 0, err
}
defer f.Close()
off, _ := f.Seek(0, 2) // current end = offset before write
if _, err := f.WriteString(line); err != nil {
return off, err
}
return off, nil
}
// ---- streamed chunk writer, boundary-safe secret masking ----
type stepLogWriter struct {
mu sync.Mutex
f *os.File
carry []byte
secrets []string
maxSecret int
}
type stepLogRegistry struct {
mu sync.Mutex
writers map[string]*stepLogWriter
}
var StepLogs = &stepLogRegistry{writers: make(map[string]*stepLogWriter)}
// Open opens (append) the server-run file for a step's streamed chunks.
func (r *stepLogRegistry) Open(commandID, path string, secrets []string) error {
if err := os.MkdirAll(filepath.Dir(path), 0700); err != nil {
return err
}
f, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0600)
if err != nil {
return err
}
max := 0
for _, s := range secrets {
if len(s) > max {
max = len(s)
}
}
w := &stepLogWriter{f: f, secrets: secrets, maxSecret: max}
r.mu.Lock()
r.writers[commandID] = w
r.mu.Unlock()
return nil
}
func (r *stepLogRegistry) get(commandID string) *stepLogWriter {
r.mu.Lock()
defer r.mu.Unlock()
return r.writers[commandID]
}
// Append masks and writes a chunk, holding back the last maxSecret-1 bytes so a
// secret split across a chunk boundary is still masked on the next append/close.
func (r *stepLogRegistry) Append(commandID string, data []byte) {
w := r.get(commandID)
if w == nil {
return
}
w.mu.Lock()
defer w.mu.Unlock()
if len(w.secrets) == 0 || w.maxSecret <= 1 {
_, _ = w.f.Write(data)
return
}
buf := append(w.carry, data...)
hold := w.maxSecret - 1
if len(buf) <= hold {
w.carry = buf
return
}
flush := buf[:len(buf)-hold]
w.carry = append([]byte{}, buf[len(buf)-hold:]...)
_, _ = w.f.Write(maskBytes(flush, w.secrets))
}
// Close flushes the carry (masked) and closes the file.
func (r *stepLogRegistry) Close(commandID string) {
r.mu.Lock()
w := r.writers[commandID]
delete(r.writers, commandID)
r.mu.Unlock()
if w == nil {
return
}
w.mu.Lock()
defer w.mu.Unlock()
if len(w.carry) > 0 {
_, _ = w.f.Write(maskBytes(w.carry, w.secrets))
w.carry = nil
}
_ = w.f.Close()
}
func maskBytes(b []byte, secrets []string) []byte {
s := string(b)
for _, v := range secrets {
if v == "" {
continue
}
s = strings.ReplaceAll(s, v, "***")
}
return []byte(s)
}
// ---- retention sweeper ----
// StartLogSweeper sweeps expired run-log dirs hourly (and once now).
func StartLogSweeper() {
go func() {
sweepLogs()
t := time.NewTicker(time.Hour)
defer t.Stop()
for range t.C {
sweepLogs()
}
}()
}
func sweepLogs() {
days := retentionDays()
if days <= 0 {
return
}
cutoff := time.Now().AddDate(0, 0, -days)
base := WorkflowLogDir()
entries, err := os.ReadDir(base)
if err != nil {
return
}
for _, e := range entries {
if !e.IsDir() {
continue
}
runID := e.Name()
dir := filepath.Join(base, runID)
if runExpired(runID, dir, cutoff) {
_ = os.RemoveAll(dir)
}
}
}
// runExpired is true when the run finished before cutoff (falling back to dir
// mtime when the run doc is gone).
func runExpired(runID, dir string, cutoff time.Time) bool {
ctx, cancel := wfCtx()
defer cancel()
var run struct {
FinishedAt *time.Time `bson:"finished_at"`
}
err := db.Col("workflow_runs").FindOne(ctx, bson.M{"run_id": runID}).Decode(&run)
if err == nil {
if run.FinishedAt == nil {
return false // still running / never finished — keep
}
return run.FinishedAt.Before(cutoff)
}
// run doc gone: use dir mtime
if fi, e := os.Stat(dir); e == nil {
return fi.ModTime().Before(cutoff)
}
return false
}
func retentionDays() int {
if v, err := GetWorkflowLogRetentionDays(); err == nil {
return v
}
return 30
}
```
Note: `wfCtx` is defined in `workflows.go` (same package) — reuse it. `GetWorkflowLogRetentionDays` is added in Task 4; this file references it (same package, compiles together).
- [ ] **Step 2: Verify build**
Run: `cd server && go build ./... && go vet ./...`
Expected: FAIL — `GetWorkflowLogRetentionDays` undefined until Task 4. This is expected; proceed to commit the file so Task 4 completes it. (If you prefer a green build, do Task 4's settings accessor first, then return — but committing here is fine since Task 4 immediately follows.)
Actually to keep every commit buildable: **temporarily** add a local stub at the bottom of this file and remove it in Task 4:
```go
// TEMP stub, replaced in Task 4.
func GetWorkflowLogRetentionDays() (int, error) { return 30, nil }
```
Then `cd server && go build ./... && go vet ./...` must succeed.
- [ ] **Step 3: Commit**
```bash
git add server/internal/services/steplogs.go
git commit -m "feat(server): workflow log-writer registry, paths, retention sweeper"
```
---
## Task 4: Settings — retention accessor + startup wiring
**Files:**
- Modify: `server/internal/services/settings.go` (or wherever settings get/set lives — search `settings` collection usage)
- Modify: `server/internal/services/steplogs.go` (remove the temp stub)
- Modify: `server/cmd/main.go` (start the sweeper)
**Interfaces:**
- Produces: `GetWorkflowLogRetentionDays() (int, error)` (default 30 when unset), `SetWorkflowLogRetentionDays(int) error`. If settings are exposed as a single document/struct, add the field there and derive these accessors.
- [ ] **Step 1: Inspect the settings service**
Read the existing settings service (search for the `settings` collection: `grep -rn "\"settings\"" server/internal/services`). Determine whether settings are a typed struct document or key/value. Match that pattern.
- [ ] **Step 2: Add the retention accessor**
If settings are a **typed document** (e.g. a `GetSettings()/UpdateSettings()`), add a field `WorkflowLogRetentionDays int `bson:"workflow_log_retention_days" json:"workflow_log_retention_days"`` to the settings struct and implement:
```go
func GetWorkflowLogRetentionDays() (int, error) {
s, err := GetSettings() // use the real accessor name
if err != nil {
return 30, err
}
if s.WorkflowLogRetentionDays == 0 && /* unset sentinel */ !s.WorkflowLogRetentionSet {
return 30, nil
}
return s.WorkflowLogRetentionDays, nil
}
```
Simplify to match reality: if the settings doc uses zero-value-means-unset and you cannot distinguish "0 = keep forever" from "unset", store the retention as a pointer `*int` or default at read: **treat a missing field as 30, an explicit 0 as keep-forever.** Prefer `*int` in the struct so the three states (unset→30, 0→forever, N→N) are representable. Implement `GetWorkflowLogRetentionDays` to return 30 when the pointer is nil, else its value. `SetWorkflowLogRetentionDays(n int)` sets the pointer.
If settings are **key/value**, implement both accessors against that store with the same nil→30 / 0→forever semantics (store empty/absent = 30).
- [ ] **Step 3: Remove the temp stub from `steplogs.go`**
Delete the `// TEMP stub` `GetWorkflowLogRetentionDays` added in Task 3 so the real one is used.
- [ ] **Step 4: Start the sweeper at boot**
In `server/cmd/main.go`, next to `EnsureWorkflowIndexes()`, add `services.StartLogSweeper()`.
- [ ] **Step 5: Verify build**
Run: `cd server && go build ./... && go vet ./...`
Expected: success (real accessor now resolves the reference from Task 3).
- [ ] **Step 6: Commit**
```bash
git add server/internal/services/settings.go server/internal/services/steplogs.go server/cmd/main.go
git commit -m "feat(server): workflow log retention setting + sweeper startup"
```
---
## Task 5: Runner + model — write to files, drop log bodies from Mongo
**Files:**
- Modify: `server/internal/models/workflow.go` (`StepRun`)
- Modify: `server/internal/services/workflow_runner.go`
- Modify: `server/internal/grpc/server.go` (stream delivery of `StepOutput`)
**Interfaces:**
- Consumes: `StepLogs`, `AppendMarker`, `ServerRunLogPath` (Task 3), `pb.StepOutputChunk` (Task 1).
- Produces: runner writes markers + streams chunks to files; `StepRun.LogOffset` persisted; `StepRun.Stdout/Stderr` removed.
- [ ] **Step 1: Update the `StepRun` model**
In `server/internal/models/workflow.go`, in `type StepRun struct`:
- Remove the `Stdout` and `Stderr` fields.
- Add: `LogOffset int64 `bson:"log_offset" json:"log_offset"``
- [ ] **Step 2: Deliver StepOutput chunks in the gRPC receive loop**
In `server/internal/grpc/server.go`, after the existing `if m.StepResult != nil { services.StepResults.Deliver(m.StepResult) }` block, add:
```go
if m.StepOutput != nil {
if m.StepOutput.Eof {
services.StepLogs.Close(m.StepOutput.CommandId)
} else {
services.StepLogs.Append(m.StepOutput.CommandId, m.StepOutput.Data)
}
}
```
- [ ] **Step 3: Rework `runServer` to open logs + write markers, drop persisted bodies**
In `server/internal/services/workflow_runner.go`, `runServer`:
Inside the per-step loop, **before** `dispatchAndWait`, add marker + open (compute `secretVals` first, which already exists in the loop):
```go
// Write the step marker and remember the offset for later slicing.
marker := fmt.Sprintf("\n===== step %d: %s =====\n", step.Order, step.Name)
offset, _ := AppendMarker(runID, serverID, marker)
logPath := ServerRunLogPath(runID, serverID)
_ = StepLogs.Open(commandID_placeholder, logPath, secretsSlice(secretVals))
```
There is a chicken-and-egg with `commandID`: today `dispatchAndWait` generates the `commandID` internally. Refactor so the runner owns the `commandID`:
1. Change `dispatchAndWait(serverID string, cmd *pb.RunStepCmd)` to `dispatchAndWait(serverID, commandID string, cmd *pb.RunStepCmd)` and remove its internal `commandID := uuid.New().String()` (use the passed one).
2. In `runServer`, generate `commandID := uuid.New().String()` at the top of each attempt-group (before the marker/open), open the log with it, then call `dispatchAndWait(serverID, commandID, cmd)`.
3. After the step completes (result received), call `StepLogs.Close(commandID)` defensively (idempotent — the agent's eof usually closed it already; Close on a missing key is a no-op).
Add a helper to convert the `secretVals map[string]string` to a `[]string` of values:
```go
func secretsSlice(m map[string]string) []string {
out := make([]string, 0, len(m))
for _, v := range m {
out = append(out, v)
}
return out
}
```
Update `finishStep(...)` call + signature: **remove** the `stdout, stderr string` params and the `output_env` masking stays. Persist `log_offset` instead. New `finishStep`:
```go
func finishStep(runID, serverID string, order int, status string, attempts, exit int, logOffset int64, 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].log_offset": logOffset,
"server_runs.$[s].steps.$[t].output_env": outEnv,
"server_runs.$[s].steps.$[t].finished_at": now,
})
}
```
In the loop, after receiving `res`, drop the `stdout, stderr := ...` masking of `res.Stdout/res.Stderr` (those are now streamed to file). Keep the `outEnv` build **with existing masking** (`maskSecrets(v, allSecrets)` per the merged secret-leak fix) — `output_env`/`run_env` masking is unchanged. Call:
```go
finishStep(runID, serverID, i, status, attempts, exit, offset, outEnv)
```
where `offset` is the marker offset captured before dispatch. If `res == nil`, still write a short note to the file so failures are visible:
```go
if res == nil {
_, _ = AppendMarker(runID, serverID, "[vantage] agent did not return a result\n")
}
```
Remove the initial `StepRun{... Status:"queued"}` `Stdout/Stderr` references if any (the model no longer has them — the queued StepRun in `TriggerWorkflow` set only `Order/Name/Status/OutputEnv`, so no change needed there; verify).
Ensure `fmt` is imported (it already is).
- [ ] **Step 4: Verify build**
Run: `cd server && go build ./... && go vet ./...`
Expected: success. Fix any remaining references to the removed `Stdout`/`Stderr` fields or the old `finishStep`/`dispatchAndWait` signatures.
- [ ] **Step 5: Commit**
```bash
git add server/internal/models/workflow.go server/internal/services/workflow_runner.go server/internal/grpc/server.go
git commit -m "feat(server): stream step logs to files, drop log bodies from run docs"
```
---
## Task 6: REST — log fetch + SSE stream endpoints
**Files:**
- Modify: `server/internal/api/workflows.go`
**Interfaces:**
- Consumes: `ServerRunLogPath`, `GetRun` (existing).
- Produces: `GET /api/runs/:runId/servers/:serverId/logs` and `GET /api/runs/:runId/servers/:serverId/logs/stream` (SSE).
- [ ] **Step 1: Add the two handlers + routes**
In `registerWorkflowRoutes`, add:
```go
g.GET("/runs/:runId/servers/:serverId/logs", getServerRunLog)
g.GET("/runs/:runId/servers/:serverId/logs/stream", streamServerRunLog)
```
Add a UUID-ish validator and the handlers:
```go
var uuidLike = regexp.MustCompile(`^[a-zA-Z0-9-]{1,64}$`)
func getServerRunLog(c *gin.Context) {
runID, serverID := c.Param("runId"), c.Param("serverId")
if !uuidLike.MatchString(runID) || !uuidLike.MatchString(serverID) {
c.JSON(http.StatusBadRequest, gin.H{"error": "invalid id"})
return
}
path := services.ServerRunLogPath(runID, serverID)
b, err := os.ReadFile(path)
if err != nil {
c.JSON(http.StatusNotFound, gin.H{"error": "no logs"})
return
}
c.Data(http.StatusOK, "text/plain; charset=utf-8", b)
}
func streamServerRunLog(c *gin.Context) {
runID, serverID := c.Param("runId"), c.Param("serverId")
if !uuidLike.MatchString(runID) || !uuidLike.MatchString(serverID) {
c.JSON(http.StatusBadRequest, gin.H{"error": "invalid id"})
return
}
path := services.ServerRunLogPath(runID, serverID)
c.Writer.Header().Set("Content-Type", "text/event-stream")
c.Writer.Header().Set("Cache-Control", "no-cache")
c.Writer.Header().Set("Connection", "keep-alive")
c.Writer.Header().Set("X-Accel-Buffering", "no")
flusher, ok := c.Writer.(http.Flusher)
if !ok {
c.JSON(http.StatusInternalServerError, gin.H{"error": "stream unsupported"})
return
}
var offset int64
sendNew := func() bool {
f, err := os.Open(path)
if err != nil {
return true // file may not exist yet; keep waiting
}
defer f.Close()
if _, err := f.Seek(offset, 0); err != nil {
return true
}
buf := make([]byte, 32*1024)
for {
n, _ := f.Read(buf)
if n <= 0 {
break
}
offset += int64(n)
// SSE data frame; split on newlines to keep frames well-formed.
for _, line := range splitSSE(buf[:n]) {
_, _ = c.Writer.WriteString("data: " + line + "\n")
}
_, _ = c.Writer.WriteString("\n")
flusher.Flush()
}
return true
}
ctx := c.Request.Context()
ticker := time.NewTicker(500 * time.Millisecond)
defer ticker.Stop()
for {
sendNew()
if serverRunTerminal(runID, serverID) {
sendNew() // final drain
_, _ = c.Writer.WriteString("event: done\ndata: end\n\n")
flusher.Flush()
return
}
select {
case <-ctx.Done():
return
case <-ticker.C:
}
}
}
// serverRunTerminal reports whether the given server-run has reached a terminal status.
func serverRunTerminal(runID, serverID string) bool {
r, err := services.GetRun(runID)
if err != nil {
return true
}
for _, sr := range r.ServerRuns {
if sr.ServerID == serverID {
switch sr.Status {
case "success", "failed", "skipped", "cancelled":
return true
}
return false
}
}
return true
}
// splitSSE turns a raw byte slice into SSE-safe payload lines (newlines become
// separate data lines; carriage returns stripped).
func splitSSE(b []byte) []string {
s := strings.ReplaceAll(string(b), "\r", "")
return strings.Split(s, "\n")
}
```
Add imports: `"os"`, `"regexp"`, `"strings"`, `"time"`, `"net/http"` (already present). Confirm `services.GetRun` and `ServerRun.Status`/`ServerID` fields exist (they do from the Workflows feature).
- [ ] **Step 2: Verify build**
Run: `cd server && go build ./... && go vet ./...`
Expected: success.
- [ ] **Step 3: Commit**
```bash
git add server/internal/api/workflows.go
git commit -m "feat(api): server-run log fetch and SSE stream endpoints"
```
---
## Task 7: Frontend — live SSE tail + retention setting
**Files:**
- Modify: `web/lib/api.ts`
- Modify: `web/app/workflows/[id]/runs/[runId]/page.tsx`
- Modify: `web/app/settings/page.tsx`
**Interfaces:**
- Consumes: SSE endpoint, logs endpoint, settings mutation.
- [ ] **Step 1: Update API types + helpers**
In `web/lib/api.ts`:
- In `StepRun`, remove `stdout` and `stderr`; add `log_offset: number`.
- Add: `getServerRunLog: (runId: string, serverId: string) => request<string>(...)` — but the logs endpoint returns `text/plain`, so add a dedicated fetch that reads text. If `request<T>` assumes JSON, add a sibling:
```ts
async getServerRunLog(runId: string, serverId: string): Promise<string> {
const res = await fetch(`${API_BASE}/api/runs/${runId}/servers/${serverId}/logs`, { credentials: "include" });
if (!res.ok) throw new Error("no logs");
return res.text();
},
```
(Use the file's real base-URL constant / credentials pattern — inspect how `request` builds URLs and mirror it. If the app is same-origin with a rewrite, a relative `/api/...` fetch is fine.)
- Export a helper to build the SSE URL: `serverRunLogStreamUrl(runId, serverId)` returning the `/api/runs/:runId/servers/:serverId/logs/stream` URL against the same base.
- In the Settings type, add `workflow_log_retention_days?: number | null`.
- [ ] **Step 2: Live tail in the run detail page**
In `web/app/workflows/[id]/runs/[runId]/page.tsx`:
- Remove all use of `st.stdout` / `st.stderr` (fields gone). Step `<details>` now show status/exit/attempts pills only.
- Add a per-server live terminal. For each `server_run`, render a `<pre>` and, while `sr.status === "running"`, subscribe via `EventSource`:
```tsx
function ServerLog({ runId, serverId, status }: { runId: string; serverId: string; status: string }) {
const [text, setText] = useState("");
const preRef = useRef<HTMLPreElement>(null);
const running = status === "running";
useEffect(() => {
if (running) {
const es = new EventSource(api.serverRunLogStreamUrl(runId, serverId), { withCredentials: true });
es.onmessage = (e) => setText((t) => t + e.data + "\n");
es.addEventListener("done", () => es.close());
es.onerror = () => es.close();
return () => es.close();
}
// terminal: fetch the whole file once
api.getServerRunLog(runId, serverId).then(setText).catch(() => setText(""));
}, [running, runId, serverId]);
useEffect(() => { preRef.current?.scrollTo(0, preRef.current.scrollHeight); }, [text]);
return (
<pre ref={preRef} className="mt-2 max-h-80 overflow-auto rounded bg-black/40 p-2 font-mono text-xs text-text-secondary whitespace-pre-wrap">
{text || (running ? "Waiting for output…" : "No output.")}
</pre>
);
}
```
Render `<ServerLog runId={run.run_id} serverId={sr.server_id} status={sr.status} />` inside each server card, below the step pills. Keep the existing react-query `refetchInterval` on the run (drives status pills); the SSE handles live text.
- [ ] **Step 3: Retention field in Settings**
In `web/app/settings/page.tsx`, add a "Workflow log retention (days)" number input bound to `workflow_log_retention_days`, saved through the existing settings save mutation. Add helper text: "0 = keep forever." Match the page's existing input styling.
- [ ] **Step 4: Verify build**
Run: `cd web && npm run build`
Expected: type-checks and builds. Fix any lingering `st.stdout`/`st.stderr` references.
- [ ] **Step 5: Commit**
```bash
git add web/lib/api.ts web/app/workflows/[id]/runs/[runId]/page.tsx web/app/settings/page.tsx
git commit -m "feat(web): live SSE log tail and log retention setting"
```
---
## Task 8: End-to-end verification
**Files:** none (verification only).
- [ ] **Step 1: Build everything**
Run: `cd server && go build ./... && go vet ./... && cd ../agent && go build ./... && go vet ./... && cd ../web && npm run build`
Expected: all succeed.
- [ ] **Step 2: Manual smoke (documented, run if an environment is available)**
With server + MongoDB + a connected agent:
1. Run a workflow with a step that emits output slowly (e.g. `for i in $(seq 1 10); do echo "line $i"; sleep 1; done`). Open the run detail page while running; confirm lines appear live (SSE), not only at the end.
2. Confirm `<logdir>/<run_id>/<server_id>.log` exists on the server with step markers and the output.
3. Confirm `workflow_runs` doc no longer stores stdout/stderr bodies; `steps[].log_offset` is set.
4. Add a secret ref and echo it; confirm the file shows `***`, including when the secret would straddle a chunk boundary.
5. Set retention to 0 in Settings → confirm sweeper keeps files; set to a small value and backdate a run's `finished_at` → confirm the dir is removed within the hour (or call `sweepLogs` path manually).
- [ ] **Step 3: Commit any fixes found**
```bash
git add -A
git commit -m "fix: workflow log streaming e2e fixes"
```
---
## Self-Review Notes
- **Spec coverage:** §3 proto → T1; §4 agent streaming → T2; §5.1 registry + §7 sweeper → T3; §7.1 setting + startup → T4; §5.3/§5.4 runner+model → T5; §6 REST/SSE → T6; §8 frontend → T7. Tests omitted per Global Constraints.
- **Masking** boundary-safe carry buffer in `StepLogs.Append`, flushed in `Close` (T3); `output_env`/`run_env` masking unchanged (T5 keeps the merged fix).
- **commandID ownership** moved to the runner so the log file can be opened before dispatch (T5) — mirrors the `StepResults.Await`-before-dispatch ordering.
- **Buildable commits:** T3 adds a temp stub for `GetWorkflowLogRetentionDays`, removed in T4.
- **Removed fields** `StepRun.Stdout/Stderr` — every reader updated in T5 (runner) and T7 (frontend).
- **Open follow-ups (out of scope):** per-step SSE channels, log download/zip, compression, pre-existing runs have no files.
```
@@ -0,0 +1,142 @@
# Fleet Inventory — Design
**Date:** 2026-07-20
**Status:** Approved (design) — ready for implementation planning
**Scope:** Fleet Inventory only. Server Workflows and SaaS/auth are separate sub-projects.
---
## 1. Summary
Each agent collects hardware/OS inventory about its host and reports it to the server, which stores the latest snapshot per server and surfaces it in the UI. Two cadences:
- **Metrics (near-real-time):** CPU load/usage, RAM used/total, swap used/total — every **30s** (aligned with existing poll rhythm).
- **Static inventory (slow):** disks, partitions and their usage, CPU model/cores, total RAM, OS details — every **15 min**.
Transport: a **new unary gRPC `ReportInventory` RPC** (mirrors the existing `ReportUpdates` pattern). No streaming.
---
## 2. Locked decisions
| Topic | Decision |
|-------|----------|
| Transport | New `ReportInventory` unary RPC. |
| Cadence | Metrics every 30s; static inventory every 15 min. One RPC carries both, but static fields are only populated on the 15-min tick (empty/omitted otherwise → server keeps prior static snapshot). |
| Storage | Latest snapshot embedded on the `servers` document (`inventory` sub-doc). No history/time-series in v1. |
| Collection | Pure-Go where practical (`/proc`, `gopsutil`-style). Agent already runs as root. |
| Platform | Linux primary; Windows agent populates what it can, leaves the rest empty. |
---
## 3. Data model
Add an `Inventory` sub-document to the existing `Server` model (`server/internal/models/server.go`):
```go
type CPUInfo struct {
Model string `bson:"model,omitempty" json:"model,omitempty"`
Cores int `bson:"cores,omitempty" json:"cores,omitempty"`
UsagePct float64 `bson:"usage_pct" json:"usage_pct"` // metrics tick
Load1 float64 `bson:"load1,omitempty" json:"load1,omitempty"`
}
type MemInfo struct {
TotalBytes uint64 `bson:"total_bytes" json:"total_bytes"`
UsedBytes uint64 `bson:"used_bytes" json:"used_bytes"` // metrics tick
}
type Partition struct {
Device string `bson:"device" json:"device"`
Mountpoint string `bson:"mountpoint" json:"mountpoint"`
Fstype string `bson:"fstype,omitempty" json:"fstype,omitempty"`
TotalBytes uint64 `bson:"total_bytes" json:"total_bytes"`
UsedBytes uint64 `bson:"used_bytes" json:"used_bytes"`
}
type Inventory struct {
CPU CPUInfo `bson:"cpu" json:"cpu"`
Memory MemInfo `bson:"memory" json:"memory"`
SwapTotalBytes uint64 `bson:"swap_total_bytes" json:"swap_total_bytes"`
SwapUsedBytes uint64 `bson:"swap_used_bytes" json:"swap_used_bytes"`
Partitions []Partition `bson:"partitions,omitempty" json:"partitions,omitempty"`
Kernel string `bson:"kernel,omitempty" json:"kernel,omitempty"`
MetricsAt *time.Time `bson:"metrics_at,omitempty" json:"metrics_at,omitempty"`
StaticAt *time.Time `bson:"static_at,omitempty" json:"static_at,omitempty"`
}
```
Add `Inventory *Inventory` field to `Server`.
Server-side update rules:
- Metrics fields (`cpu.usage_pct`, `cpu.load1`, `memory.used_bytes`, swap used) always updated + `metrics_at`.
- Static fields (`cpu.model/cores`, `memory.total_bytes`, `partitions`, `kernel`, swap total) updated only when the report includes them (non-zero/non-empty) + `static_at`.
---
## 4. gRPC protocol (`proto/vantage/v1/vantage.proto` + both `pb.go` files)
```protobuf
rpc ReportInventory(InventoryReport) returns (InventoryReportResponse);
message InventoryReport {
string server_id = 1;
string agent_token = 2;
bool include_static = 3; // true on the 15-min tick
CPUReport cpu = 4;
MemReport memory = 5;
uint64 swap_total = 6;
uint64 swap_used = 7;
repeated PartitionReport partitions = 8; // only when include_static
string kernel = 9; // only when include_static
}
message CPUReport { string model = 1; int32 cores = 2; double usage_pct = 3; double load1 = 4; }
message MemReport { uint64 total_bytes = 1; uint64 used_bytes = 2; }
message PartitionReport { string device = 1; string mountpoint = 2; string fstype = 3; uint64 total_bytes = 4; uint64 used_bytes = 5; }
message InventoryReportResponse {}
```
Hand-written JSON-codec structs added to `server/internal/grpc/pb/vantage.pb.go` and `agent/internal/grpc/pb/vantage.pb.go`, plus the RPC method wiring (service interface, client method, handler registration) mirroring `ReportUpdates`.
---
## 5. Agent collection (`agent/internal/inventory/`)
- `Collect(includeStatic bool) *pb.InventoryReport` — reads:
- CPU usage: sample `/proc/stat` delta; load from `/proc/loadavg`; model/cores from `/proc/cpuinfo` (static).
- Memory/swap: `/proc/meminfo`.
- Partitions: `/proc/mounts` filtered to real filesystems + `statfs` for total/used (static).
- Kernel: `uname` / `/proc/version` (static).
- Windows: best-effort via `wmic`/PS or leave empty.
- Scheduler in the agent main loop: a 30s ticker calls `Collect(false)` and `ReportInventory`; every 30th tick (15 min) calls `Collect(true)`.
- Reuse existing gRPC client; add `Client.ReportInventory(...)` like `ReportUpdates`.
Prefer implementing the `/proc` readers directly (no new heavy deps) unless a `gopsutil` dependency is already vendored.
---
## 6. Server handler + service
- gRPC handler `ReportInventory` in `server/internal/grpc/server.go`: validate agent token (`ValidateAgentToken`), then call `services.StoreInventory(serverID, report)`.
- `services.StoreInventory` (in `server/internal/services/inventory.go`): builds the `$set` per the update rules in §3 and `UpdateOne` on `servers`.
---
## 7. Frontend
Surface inventory on the existing server detail page (`web/app/servers/[id]/page.tsx`) — add an "Inventory" panel:
- CPU usage gauge + model/cores, load.
- RAM used/total bar, swap bar.
- Partitions table: device, mount, fstype, used/total with a usage bar.
- "Updated Xs ago" from `metrics_at`/`static_at`.
Optionally add compact CPU/RAM badges to the servers list (`web/app/servers/page.tsx`). Reuse `@/components/ui` + Tailwind tokens. Poll the server detail query while the page is open (react-query `refetchInterval` ~30s) so metrics stay fresh.
---
## 8. Out of scope
- Time-series history / graphs (only latest snapshot stored).
- Alerting thresholds on usage (settings/alerts is a separate concern).
- Per-process / network / GPU inventory.
- Tests (skipped, consistent with the Workflows iteration).
@@ -0,0 +1,142 @@
# SaaS: Auth + Organizations — Design
**Date:** 2026-07-20
**Status:** Approved (design) — ready for implementation planning
**Scope:** Local auth + organizations + per-org OIDC, and org-scoping of existing data. Billing/plan-limits explicitly deferred. Fleet Inventory and Server Workflows are separate sub-projects.
---
## 1. Summary
Turn Vantage from a single-admin, single global-OIDC tool into a multi-tenant app:
1. **Replace** the global Authentik/env-based OIDC with **local email/password accounts** as the primary login.
2. **Organizations** — every user belongs to an org; every domain object (servers, keys, secrets, assignments, workflows, steps, runs, audit) carries an `org_id` and all queries are scoped to the caller's org.
3. **Per-org OpenID** — an org admin can configure their own OIDC provider (issuer/client id/secret); users in that org can then sign in through it.
No billing, no seat/server limits this iteration (schema leaves room).
---
## 2. Locked decisions
| Topic | Decision |
|-------|----------|
| Primary auth | Local email + password (bcrypt). Replaces global Authentik. |
| Org SSO | Per-org OIDC provider, configured by org admin, resolved dynamically at login. |
| Isolation | `org_id` on every collection; every service query filtered by org. Enforced in the request layer via session→org. |
| Roles | `owner`, `admin`, `member` (v1: owner/admin can manage users + org OIDC + all resources; member can use resources). Keep minimal. |
| Bootstrapping | First-run creates the initial org + owner account (setup flow) when no users exist. |
| Sessions | Keep existing Redis session store; session now carries `user_id`, `org_id`, `role`, `email`. |
| Agent auth | Unchanged (per-server agent tokens). Servers gain `org_id`; agent RPCs resolve org from the server record. |
---
## 3. Data model
### `orgs`
```json
{ "_id":"ObjectId", "org_id":"uuid", "name":"Acme", "created_at":"ISODate" }
```
### `users`
```json
{
"_id":"ObjectId", "user_id":"uuid", "org_id":"uuid",
"email":"a@b.com", "password_hash":"bcrypt...", "role":"owner|admin|member",
"auth_source":"local|oidc", "created_at":"ISODate", "last_login":"ISODate|null"
}
```
Unique index on `email` (global — email identifies the account and its org).
### `org_oidc` (per-org provider config)
```json
{
"_id":"ObjectId", "org_id":"uuid",
"issuer":"https://id.acme.com", "client_id":"...",
"client_secret_enc":"AES...", // encrypted with existing crypto.go
"redirect_url":"https://vantage.../auth/oidc/callback",
"enabled": true, "updated_at":"ISODate"
}
```
### Existing collections — add `org_id`
`servers`, `keys`, `assignments`, `secrets`, `workflows`, `workflow_steps`, `workflow_runs`, `audit` each gain `org_id string`. A **migration** backfills all existing documents into a default org (see §7).
---
## 4. Auth flows
### Local
- `POST /auth/register` — only allowed during first-run bootstrap (creates org + owner) OR by an org admin inviting a user (see below). Not open self-serve.
- `POST /auth/login` — email + password → verify bcrypt → create session with `{user_id, org_id, role, email}`.
- `POST /auth/logout` — destroy session.
- `GET /auth/me` — returns current user + org.
### Org-admin user management
- `GET /api/org/users` / `POST /api/org/users` (create local user in caller's org) / `PUT /api/org/users/:id/role` / `DELETE /api/org/users/:id`.
### Per-org OIDC
- `GET/PUT /api/org/oidc` — read/save the caller org's provider config (admin only). Secret stored encrypted.
- `GET /auth/oidc/start?org=<org_id or slug>` — look up org's `org_oidc`, build the OIDC provider on demand (cache per org), redirect to authorize.
- `GET /auth/oidc/callback` — exchange code, match/provision the user by email **within that org**, create session.
- If the email exists in the org → log in. If not → provision a `member` with `auth_source=oidc` (org admin can promote). Reject if email belongs to a different org.
### First-run bootstrap
- `GET /auth/bootstrap-status``{ needs_setup: bool }` (true when `users` is empty).
- Setup page collects org name + owner email/password → creates org + owner → session.
---
## 5. Request scoping
- `auth.Middleware` already loads the session; extend `Session` to include `OrgID`, `UserID`, `Role`. Add helper `auth.OrgID(c) string`.
- **Every service function that reads/writes a scoped collection takes an `orgID` argument** and adds `"org_id": orgID` to its filter and on insert. Handlers pass `auth.OrgID(c)`.
- Add a `requireRole(role)` gin middleware for admin-only routes (org user mgmt, org OIDC).
- Agent-facing gRPC: resolve `org_id` from the `servers` record (already tied to `server_id`); inventory/keys/sync operate on that org implicitly.
---
## 6. Removing global Authentik
- Delete/retire env-driven `InitOIDC` global provider (`OIDC_ISSUER` etc.). Keep the `go-oidc`/`oauth2` machinery but move it behind the per-org resolver.
- `authEnabled` global replaced by "auth always on" (there is always local auth). Update `middleware.go` accordingly (no more `if !authEnabled { next }` bypass — except the bootstrap endpoints and login/register which are unauthenticated).
- Login page (`web/app/login` or existing) offers: email/password form + "Sign in with your organization's SSO" (enter org, redirect to `/auth/oidc/start`).
---
## 7. Migration
One-shot migration run at startup (idempotent):
1. If `orgs` is empty AND `servers`/`keys`/etc. contain documents without `org_id`: create a **default org** ("Default").
2. Set `org_id = <default>` on all existing `servers`, `keys`, `assignments`, `secrets`, `workflows`, `workflow_steps`, `workflow_runs`, `audit` documents missing it.
3. If `OIDC_ISSUER` env was set previously and an admin email is known, optionally seed an owner user (documented manual step) — otherwise first-run bootstrap handles owner creation.
Guard with a marker (e.g. a `migrations` collection entry) so it runs once.
---
## 8. Frontend
- **Login/Setup:** `web/app/login/page.tsx` (email/password + org SSO entry) and `web/app/setup/page.tsx` (first-run). Redirect logic based on `bootstrap-status` and `auth/me`.
- **Org settings:** `web/app/settings/org/` — members list + invite/create user + role management; OIDC provider form (issuer/client id/secret/enabled).
- Existing pages unchanged functionally but now implicitly org-scoped by the backend. Show current org + user in the sidebar/header.
---
## 9. Security
- Passwords: bcrypt (cost ≥ 12). Never returned.
- Org OIDC client secret encrypted at rest (reuse `services/crypto.go` AES).
- Cross-org access prevented at the service layer (org_id in every filter) — the primary isolation boundary. Handlers must never accept an `org_id` from the client; always derive from session.
- OIDC callback must bind the returned identity to the org that initiated the flow (state carries org_id) to prevent org-mixing.
- Role checks on all org-admin mutations.
---
## 10. Out of scope
- Billing, plans, seat/server limits.
- Cross-org resource sharing, org switching for a single user (one user = one org in v1).
- SCIM / directory sync, SAML.
- Email delivery for invites (create-user sets a password or invite token; email sending deferred — document as manual/console output).
- Tests (skipped, consistent with prior iterations).
@@ -0,0 +1,238 @@
# Server Workflows — Design
**Date:** 2026-07-20
**Status:** Approved (design) — ready for implementation planning
**Scope:** Server Workflows only. Fleet Inventory and SaaS/local-auth are separate sub-projects with their own specs.
Approved UI mockup: three-pane builder (Step Library · Canvas · Inspector), env vars shown riding the wire between nodes.
---
## 1. Summary
Let operators compose **reusable shell steps** (Bash or PowerShell) into **workflows** and run them across many managed servers in parallel. Steps pass data to later steps through a `$WORKFLOW_ENV` file (GitHub-Actions style). Every run is recorded with full per-step logs. Steps can reference org secrets, injected as environment variables at runtime.
Builds directly on the existing `CommandStream` gRPC infrastructure (`dispatch.go`, `ServerCommand` oneof, agent command loop).
---
## 2. Locked decisions
| Topic | Decision |
|-------|----------|
| Data passing | Implicit. Every step's `$WORKFLOW_ENV` outputs merge into the run's env and are exposed to **all** later steps as `$KEY`. No explicit port wiring. |
| Failure model | Per-step policy: `stop` (default), `continue`, `retry` (with max attempt count). |
| Targets | Fan-out. Same step sequence runs on N target servers **in parallel**. Steps within one server run **sequentially**. |
| History/logs | Every run persisted: status, timing, per-server per-step stdout/stderr/exit code, captured output env. |
| Secrets | Steps declare needed secret keys; resolved from existing `secrets` store and injected as env vars at exec time. Never persisted into run logs. |
| Testing | **Skipped** for this iteration per request. No test files written. |
---
## 3. Data model (MongoDB)
### `workflow_steps` — reusable step library
```json
{
"_id": "ObjectId",
"step_id": "uuid",
"name": "Restart service",
"description": "Restart-Service by name, wait ready",
"interpreter": "bash | powershell",
"script": "Restart-Service vantage-api\n...",
"declared_outputs": ["STARTED_AT"], // documentation/UI hints; not enforced
"secret_refs": ["DEPLOY_TOKEN"], // secret keys this step needs injected
"org_id": "uuid", // for future multi-tenant; single-org for now
"created_at": "ISODate",
"updated_at": "ISODate"
}
```
### `workflows` — ordered composition
```json
{
"_id": "ObjectId",
"workflow_id": "uuid",
"name": "Deploy & Restart API",
"target_server_ids": ["uuid", "uuid"],
"steps": [
{
"step_id": "uuid", // reference to library step
"order": 0,
"on_failure": "stop | continue | retry",
"max_retries": 0, // used when on_failure = retry
"overrides": { // optional local fork of the library step
"script": null,
"secret_refs": null
}
}
],
"created_at": "ISODate",
"updated_at": "ISODate"
}
```
Editing a library step from the Inspector writes an `overrides` block on that workflow step (a local fork) rather than mutating the shared step.
### `workflow_runs` — execution records
```json
{
"_id": "ObjectId",
"run_id": "uuid",
"workflow_id": "uuid",
"workflow_snapshot": { }, // frozen copy of workflow + resolved steps at trigger time
"status": "running | success | failed | cancelled",
"triggered_by": "user-id",
"started_at": "ISODate",
"finished_at": "ISODate | null",
"server_runs": [
{
"server_id": "uuid",
"status": "queued | running | success | failed | skipped",
"started_at": "ISODate | null",
"finished_at": "ISODate | null",
"run_env": { "VERSION": "a1b9f0" }, // accumulated non-secret output env
"steps": [
{
"order": 0,
"name": "Git pull & build",
"status": "success | failed | running | queued | skipped",
"attempts": 1,
"exit_code": 0,
"stdout": "…",
"stderr": "…",
"output_env": { "VERSION": "a1b9f0" },
"started_at": "ISODate",
"finished_at": "ISODate"
}
]
}
]
}
```
Secret values are never written to `stdout`/`stderr`/`run_env` by us; masking of known secret values in captured output is applied before persistence.
---
## 4. gRPC protocol changes (`proto/vantage/v1/vantage.proto`)
### New command in the `ServerCommand` oneof
```protobuf
message RunStepCmd {
string interpreter = 1; // "bash" | "powershell"
string script = 2;
map<string, string> env = 3; // inputs = accumulated run env + injected secrets
int32 timeout_seconds = 4;
}
```
Add `RunStepCmd run_step = 6;` to the `ServerCommand` oneof.
### Richer result — new `AgentMessage` payload
Current `CommandResult{command_id, success, message}` is too thin. Add a dedicated step result:
```protobuf
message StepResult {
string command_id = 1;
int32 exit_code = 2;
string stdout = 3;
string stderr = 4;
map<string, string> output_env = 5; // parsed $WORKFLOW_ENV KEY=value lines
}
```
Add `StepResult step_result = 5;` to the `AgentMessage` oneof (alongside existing `ready` / `result`).
---
## 5. Agent execution (`agent/internal/...`)
New handler for `RunStepCmd` in the agent command loop:
1. Create a temp dir; create empty `WORKFLOW_ENV` file inside it.
2. Write `script` to a temp script file.
3. Build the process environment: inherited env + `cmd.env` (run env + secrets) + `WORKFLOW_ENV=<path to env file>`.
4. Execute:
- `bash``bash <script>`
- `powershell``pwsh -NoProfile -File <script>` (fallback `powershell.exe` on Windows if `pwsh` absent).
5. Capture stdout, stderr, exit code. Enforce `timeout_seconds` (kill on exceed → non-zero exit, stderr note).
6. Parse the `WORKFLOW_ENV` file: each `KEY=value` line becomes an `output_env` entry (last write wins; supports multi-line via simple `KEY<<EOF` heredoc form, optional for v1 — start with single-line `KEY=value`).
7. Reply with `StepResult`. Delete temp dir.
Agent runs as root (existing), so no privilege change. Script content is trusted operator input.
---
## 6. Server orchestration (`server/internal/services/workflows.go`)
Runner responsibilities:
1. On trigger: snapshot the workflow (resolve each library step + overrides), create a `workflow_runs` doc with one `server_run` per target, all `queued`.
2. Spawn one goroutine **per target server** (parallel fan-out). Each goroutine:
- Verifies the agent is connected (`Dispatcher.IsConnected`); if not → `server_run.status = skipped`, reason recorded.
- Maintains a `run_env map[string]string`, seeded empty.
- For each step in order:
- Resolve `secret_refs` from the secrets service → merge into the command env (kept separate from persisted `run_env`).
- Dispatch `RunStepCmd{env: run_env + secrets}` via a **correlated** send — needs a way to await the matching `StepResult` by `command_id` (see §7).
- On result: persist step record (stdout/stderr/exit, masked); merge `output_env` into `run_env`.
- Apply `on_failure` on non-zero exit: `stop` (fail server_run, break), `continue` (mark failed, proceed), `retry` (re-dispatch up to `max_retries`).
3. Aggregate: run `status = success` if all server_runs succeeded, else `failed`. Set `finished_at`.
### Concurrency / queue
- One workflow run per workflow at a time (reject or queue concurrent triggers — v1: reject with clear error).
- Per-server step dispatch is serial; servers are parallel.
---
## 7. Correlated command results
The existing dispatcher is fire-and-forget; workflows need request/response by `command_id`. Add a small **pending-result registry** alongside `Dispatcher`:
- `AwaitResult(commandID) <-chan *pb.StepResult` — registers a channel before dispatch.
- The `CommandStream` receive loop, on a `StepResult`, looks up the pending channel by `command_id` and delivers it (falls back to existing `CommandResult` handling for other command types).
- Timeout guard on the server side (step `timeout_seconds` + grace) so a dead agent can't hang a run.
This is additive; existing `CommandResult` flow for key/update commands is unchanged.
---
## 8. REST API (`server/internal/api/workflows.go`)
| Method + path | Purpose |
|---------------|---------|
| `GET /api/steps` / `POST` / `PUT /:id` / `DELETE /:id` | Reusable step library CRUD |
| `GET /api/workflows` / `POST` / `PUT /:id` / `DELETE /:id` | Workflow CRUD (name, targets, ordered steps) |
| `POST /api/workflows/:id/run` | Trigger a run; returns `run_id` |
| `GET /api/workflows/:id/runs` | Run history (summary list) |
| `GET /api/runs/:run_id` | Full run detail incl. per-server per-step logs |
| `POST /api/runs/:run_id/cancel` | Best-effort cancel |
Secrets are referenced by key only through these APIs; values never returned.
---
## 9. Frontend (`web/app/workflows/`)
- `/workflows` — list workflows, last run status/time, Run button.
- `/workflows/[id]` — the three-pane builder from the approved mockup:
- **Library** (left): reusable steps, `bash`/`pwsh` badges, search, add.
- **Canvas** (center): ordered nodes, env chips on wires, live status pills.
- **Inspector** (right): name, command editor, declared inputs/outputs, `secret_refs` picker, `on_failure` + retry count.
- `/workflows/[id]/runs/[runId]` — run detail: per-server columns, expandable per-step stdout/stderr, exit codes, timing. Live-updating while `running` (poll, consistent with existing 30s-poll ethos — or reuse whatever the console screen uses).
Reuse existing web components/styling patterns (there is already `servers`, `secrets`, `audit`, console UI to match).
---
## 10. Security notes
- Scripts are trusted operator input executed as root — same trust level as the existing console feature. No new sandbox in v1.
- Secret values injected as env only; masked from all persisted logs (`stdout`/`stderr`/`run_env`) by literal replacement before write.
- Run triggering and step/workflow CRUD gated behind existing auth (`server/internal/auth`).
- Audit: emit audit-log entries (existing `audit` service) on workflow create/edit/delete and run trigger.
---
## 11. Out of scope (this iteration)
- Tests (explicitly skipped).
- Branching/conditional steps, matrix per-server conditionals (fan-out only).
- Scheduled/cron triggers (manual run only for v1).
- Multi-org isolation enforcement (schema carries `org_id` for later; single-org behavior now).
- Artifact upload/collection beyond env vars.
@@ -0,0 +1,150 @@
# Workflow Builder v2 — Design
**Date:** 2026-07-20
**Status:** Approved (design) — ready for implementation planning
**Scope:** Overhaul the workflow builder UI to match the approved mockup, add drag-and-drop (library→canvas + reorder), base-step editing/deletion with cascade, step input parameters, an Edit-Workflow modal, runs navigation, and fix the save crash. Enhancement to the merged Server Workflows feature. Independent of the in-flight log-streaming work.
---
## 1. Summary
The shipped builder diverges from the approved mockup and is missing interactions. This iteration:
1. **Restyle** the builder (`/workflows/[id]`) to the approved mockup: dotted-grid canvas, 340px node cards with index badge + shell badge + status, wire connectors with dashed-amber "passes" env chips, a kicker/field inspector, and a library of grabbable step cards with descriptions.
2. **Drag-and-drop**: drag a library step onto the canvas to add it; drag nodes to reorder. Remove the up/down/remove buttons.
3. **Base-step editing**: an "Edit base step" modal edits the shared library step (name/interpreter/script/outputs/inputs/secret refs) and **saves**; deleting a shared step **cascades** — it is pulled from every workflow that references it.
4. **Input parameters**: a base step can declare inputs (`name` + `default` + `description`); when placed, each placement sets values; the runner injects them into the step's environment.
5. **Env visibility**: show the output variables passed between steps as chips on the wires (already partially present — align to the mockup).
6. **Edit-Workflow modal**: edit name, target servers, delete the workflow, and other workflow settings.
7. **Runs navigation**: a runs list page per workflow, linked from the builder and the workflows list.
8. **Bug fix**: `updateWorkflow` returns `{updated:true}`, which the builder stores as the workflow and then crashes on `[...wf.steps]` ("d.steps is not iterable"). Fix the endpoint to return the updated workflow and harden the client.
---
## 2. Locked decisions
| Topic | Decision |
|-------|----------|
| Visual target | The approved mockup (artifact `61ab5256`). Adopt its layout + **amber (`#f5a524`) as the builder signal/focus color**, plus bash-green (`#3fb950`) / pwsh-blue (`#5b9bff`) badge colors. Keep the app's existing `surface`/`border`/`text-*` tokens for panels so it integrates with the dark theme. |
| Drag-and-drop | Native HTML5 DnD. Library cards are `draggable`; the canvas has drop targets (between nodes + end zone) to insert; nodes are `draggable` to reorder. Clicking a library card still appends (keyboard/fallback). |
| Step removal | No per-node buttons. Remove a placed step from the **inspector** ("Remove from workflow"). |
| Input params | `WorkflowStep.declared_inputs: [{name, default, description}]`. `WorkflowStepRef.inputs: map[name]value`. Runner resolves `value = ref.inputs[name] ?? default` and injects as env vars. |
| Edit scope | Inspector script edit = **per-placement override** (existing `overrides`, "forks a local copy"). A separate **Edit base step** modal updates the shared library step for all workflows. |
| Cascade delete | Deleting a library step pulls its `step_id` from every `workflow.steps` and re-sequences remaining `order`s. |
| Env chips | Names passed between steps = union of prior steps' `declared_outputs`. Shown on the wire between nodes. |
| Edit workflow | Modal launched from the topbar: name, target-servers multiselect, delete workflow. |
| Runs nav | New page `/workflows/[id]/runs` (list); links from the builder topbar and the workflows list page. |
| Save fix | `PUT /api/workflows/:id` returns the full updated `Workflow`. Client also guards against non-workflow responses. |
| Proto | **No proto change** — input params travel through the existing `RunStepCmd.Env`. |
---
## 3. Data model (`server/internal/models/workflow.go`)
Add an input-parameter type and fields:
```go
type InputParam struct {
Name string `bson:"name" json:"name"`
Default string `bson:"default" json:"default"`
Description string `bson:"description" json:"description"`
}
```
- `WorkflowStep` gains: `DeclaredInputs []InputParam `bson:"declared_inputs" json:"declared_inputs"``.
- `WorkflowStepRef` gains: `Inputs map[string]string `bson:"inputs,omitempty" json:"inputs,omitempty"`` (per-placement values).
- `ResolvedStep` gains: `Inputs map[string]string `bson:"inputs" json:"inputs"`` (frozen resolved input env for the run).
`declared_inputs` defaults to `[]` on create (like `declared_outputs`).
---
## 4. Services
### 4.1 Step CRUD (`server/internal/services/workflows.go`)
- `CreateStep`: default `DeclaredInputs` to `[]InputParam{}` when nil; persist it.
- `UpdateStep`: add `declared_inputs` to the `$set`.
- `DeleteStep`**cascade**. New behavior: within the delete, also update every workflow that references the step:
1. `DeleteOne` on `workflow_steps` by `step_id` (as today).
2. Load all workflows containing the step (`workflows` where `steps.step_id == stepID`); for each, remove the matching `WorkflowStepRef`(s), re-sequence remaining `order` values to `0..n-1`, and `UpdateWorkflow`.
Keep it a single service call `DeleteStep(stepID)` so the handler is unchanged. Audit both the step deletion and each affected workflow via `LogEvent`.
### 4.2 Workflow update returns the workflow (`server/internal/services/workflows.go` + handler)
- `UpdateWorkflow(id, w)` stays, but the **handler** `updateWorkflow` re-fetches and returns the full workflow: after `services.UpdateWorkflow`, call `services.GetWorkflow(id)` and return it (200 with the `Workflow` JSON) instead of `{"updated": true}`. This is the crash fix's server half.
### 4.3 Runner input injection (`server/internal/services/workflow_runner.go`)
- `resolveSteps`: when freezing each `ResolvedStep`, compute `Inputs`: for each `InputParam` on the library step, `value = ref.Inputs[name]` if present else `param.Default`; store the resulting `map[string]string` on `ResolvedStep.Inputs`.
- `runServer`: when building `cmdEnv`, merge `step.Inputs` **first** (base layer), then `runEnv` (upstream outputs), then `secretVals` (highest precedence). Input values are not secrets and are not masked (they are user-provided config, not secret material) — unless a value coincidentally equals a secret literal, existing masking still catches it in logs.
---
## 5. REST API
No new endpoints. Changes:
- `updateWorkflow` handler returns the `Workflow` (see §4.2).
- `deleteStep` handler unchanged (cascade lives in the service).
- Existing `GET /api/workflows/:id/runs`, `GET /api/runs/:runId` power the runs pages.
- `createStep`/`updateStep` accept `declared_inputs` via the existing `ShouldBindJSON(&models.WorkflowStep)` (no handler change once the model has the field).
---
## 6. Frontend
### 6.1 Tokens + primitives
- `tailwind.config.ts`: add `bash: "#3fb950"`, `pwsh: "#5b9bff"`. Amber signal reuses a new `signal: "#f5a524"` token (add it) for the builder's focus ring / env chips / run button; `signal-ink: "#241800"` for text on amber chips.
- Add a `Modal` primitive to `components/ui` (overlay + centered panel, `onClose`, ESC + backdrop click, `title`, children, exported from `index.ts`). Used by the Edit-base-step and Edit-workflow modals.
### 6.2 API client (`web/lib/api.ts`)
- `WorkflowStep`: add `declared_inputs: InputParam[]`.
- New `InputParam { name: string; default: string; description: string }`.
- `WorkflowStepRef`: add `inputs?: Record<string, string>`.
- `updateWorkflow` return type stays `Workflow` (now actually returns one). Add a client guard: if a mutation is expected to return a workflow but the body lacks `steps`, treat it as an error / refetch.
- No other method changes; `deleteStep`, `updateStep`, `createStep`, `listRuns`, `getRun` already exist.
### 6.3 Builder restyle + DnD (`web/app/workflows/[id]/page.tsx`)
Rebuild the three-pane builder to the mockup:
- **Topbar:** brand dot + `Workflows /` crumb + workflow name + `· draft · edited …`; right side a **Targets** chip (read-only summary, e.g. "3 servers"), a **Runs** link (→ `/workflows/[id]/runs`), an **Edit** button (opens Edit-workflow modal), **Save**, **Run workflow** (amber). Remove the inline name input and the in-canvas Target-servers card (both move into the Edit-workflow modal).
- **Library (left):** section header with `+` (opens Edit-base-step modal in "new" mode), a search box that filters by name, grouped `Shared · Bash` / `Shared · PowerShell`, each a `.step-card` with shell badge, name, description, and a grip glyph; `draggable`. Clicking still appends. Each card has an edit affordance (pencil / context) that opens the Edit-base-step modal for that step.
- **Canvas (center):** dotted-grid background. Render placed steps as 340px node cards: index badge, title, shell badge, (build-time) no status pill — status is a run concern; in the builder show the interpreter badge only. Node body shows a syntax-lite script preview. Between nodes render a wire + a dashed-amber `passes` chip row listing the union of prior `declared_outputs` (names only in the builder). Nodes are `draggable` to reorder; drop targets sit on the wires and at the end ("+ Drop a step here"). Selecting a node opens it in the inspector.
- **Inspector (right):** kicker "Step N · Inspector", title with shell badge + name. Fields: **Step name** (edits the placement label? — placements don't have a name; show the library name read-only, edit happens in the base-step modal), **Command** (script `<textarea>` — this is the per-placement override; empty = inherit base), a hint "Write `KEY=value` to `$WORKFLOW_ENV`…", **Inputs** (one row per `declared_input` with an input to set the placement value, showing the default as placeholder), **Inputs · from upstream** (read-only list of upstream `declared_outputs` available to this step), **Outputs · to $WORKFLOW_ENV** (read-only list of this step's `declared_outputs`), **Secret refs** (existing group/KEY checklist), **On failure** (stop/continue/retry + max retries), and **Remove from workflow**.
DnD detail: use `dataTransfer` with a payload discriminating "library step" (carries `step_id`) vs "reorder" (carries the placement index). On drop at position `k`, insert/move and re-sequence `order`. Keep everything in React state; **Save** persists via `updateWorkflow` (which now returns the workflow → `setWf(returned)` no longer crashes).
### 6.4 Edit base step modal
A `Modal` with fields: name, interpreter (bash/powershell), script (`<textarea>` mono), declared **outputs** (chip/list editor — add/remove names), declared **inputs** (rows of name/default/description, add/remove), secret refs (optional). Actions: **Save** (`api.createStep` in new mode / `api.updateStep` in edit mode) then invalidate `["steps"]`; **Delete** (edit mode only) → confirm, `api.deleteStep` (cascades server-side), invalidate `["steps"]` and `["workflow", id]` (a deleted step vanishes from the canvas after refetch). Editing here changes the shared step for all workflows (surface the "shared across all workflows" hint).
### 6.5 Edit workflow modal
A `Modal` launched from the topbar **Edit** button: workflow **name** input, **target servers** multiselect (the chips currently in the canvas), and a **Delete workflow** action (confirm → `api.deleteWorkflow` → route to `/workflows`). Save applies name/targets to local `wf` state (persisted on the builder's Save) or immediately via `updateWorkflow` — immediate is simpler and avoids losing the change; use immediate save for the modal, then `setWf(returned)`.
### 6.6 Runs list page (`web/app/workflows/[id]/runs/page.tsx`)
New page: header "Runs · <workflow name>", a table of `api.listRuns(id)` rows — run id (short), status badge, started_at, triggered_by, server count — each linking to `/workflows/[id]/runs/[runId]`. A "Back to builder" link. Also add a **Runs** action/link on the workflows list page (`web/app/workflows/page.tsx`) per row and the **Runs** link in the builder topbar.
---
## 7. Security
- Input parameter values are user config, injected as env; not masked (not secret). Secret masking (existing) still applies to logs and to any value equal to a secret literal.
- Cascade delete is an authenticated mutation; audited via `LogEvent` for the step and each affected workflow.
- Modals perform the same session-authed API calls; no new trust boundary.
---
## 8. Out of scope
- Live run status pills inside the builder canvas (status belongs to the run detail page).
- Typed inputs (all inputs are strings), required/validation rules, secret-typed inputs.
- Multi-select drag of several steps, copy/paste of steps, undo/redo.
- Reworking the run-detail page (covered by the separate log-streaming iteration).
- Reordering via keyboard.
- Tests (skipped, consistent with prior iterations).
```
@@ -0,0 +1,193 @@
# Workflow Log Streaming — Design
**Date:** 2026-07-20
**Status:** Approved (design) — ready for implementation planning
**Scope:** Stream step stdout/stderr live from agent to server-side log files, tail them live in the UI, and auto-expire them on a retention period. Enhancement to the already-merged Server Workflows feature. No auth/orgs, no inventory.
---
## 1. Summary
Today a workflow step buffers all stdout/stderr in agent RAM, ships it in one terminal `StepResult`, and the server persists the whole body into the `workflow_runs` Mongo document. Long/chatty steps risk: agent memory blow-up, the gRPC 4MB message ceiling, and the Mongo 16MB document cap.
Change to **live streaming**:
1. Agent streams output chunks over the existing `CommandStream` as the process runs.
2. Server appends chunks (secret-masked) to a **per-server-run log file** on disk — not Mongo.
3. UI tails the file live via **SSE** while a server-run is running; slices per-step by byte offset after completion.
4. A **retention sweeper** deletes old run-log directories on a configurable period (default 30 days, set in Settings).
`workflow_runs` documents shrink: they no longer carry `stdout`/`stderr` bodies, only status/exit/attempts/output_env/timestamps plus a per-step `log_offset`.
---
## 2. Locked decisions
| Topic | Decision |
|-------|----------|
| Transport | Reuse bidirectional `CommandStream`. New `AgentMessage.StepOutput` chunk message. |
| Chunk shape | `{command_id, seq, data, eof}`. Interleaved stdout+stderr in execution order. |
| Terminal result | `StepResult` still sent at step end, now carries only `exit_code` + `output_env` (no stdout/stderr). |
| Log granularity | **One file per server-run**: `<logdir>/<run_id>/<server_id>.log`, with a marker line before each step. |
| Streams | **Interleaved** — single synchronized writer on the agent, terminal-order output. |
| Masking | **Server-side** (agent can't tell secret env from normal env). Per-stream carry buffer of `maxSecretLen-1` bytes so a secret split across a chunk boundary still masks; flushed on EOF. |
| Live tail | **SSE** at per-server-run granularity: `GET /api/runs/:runId/servers/:serverId/logs/stream`. Post-run whole-file fetch + per-step offset slice. |
| Retention | `settings.workflow_log_retention_days`, default **30**, editable in `/settings`. Hourly sweeper deletes `<logdir>/<run_id>/` dirs older than retention by run `finished_at`. |
| Log dir | Env `VANTAGE_WORKFLOW_LOG_DIR`, default `<data>/workflow-logs`. Created `0700`. |
| Mongo | No log bodies in `workflow_runs`. Disk is the source of truth for output. |
---
## 3. gRPC protocol (`proto/vantage/v1/vantage.proto` + both `pb.go` files)
Add to the `AgentMessage` oneof: `StepOutputChunk step_output = 6;`
```protobuf
message StepOutputChunk {
string command_id = 1;
uint64 seq = 2; // monotonic per command_id, 0-based
bytes data = 3; // raw interleaved stdout+stderr bytes
bool eof = 4; // true on the final (empty) chunk
}
```
`StepResult` is unchanged in shape but `stdout`/`stderr` are now left empty by the agent (kept in the message for backward-compat / error notes only — server ignores them for log content). The server still reads `exit_code` and `output_env` from `StepResult`.
Hand-written JSON-codec struct added to **both** `server/internal/grpc/pb/vantage.pb.go` and `agent/internal/grpc/pb/vantage.pb.go`, identical. `AgentMessage` gains `StepOutput *StepOutputChunk` in both.
`data` is `[]byte` in the Go structs (JSON-codec base64-encodes it, which is fine).
---
## 4. Agent (`agent/internal/exec/exec.go`)
`RunStep` signature gains a chunk sink:
```go
func RunStep(cmd *pb.RunStepCmd, emit func(seq uint64, data []byte)) *pb.StepResult
```
- Replace the two `bytes.Buffer`s with a single `streamWriter` set as **both** `c.Stdout` and `c.Stderr`. Its `Write` takes a mutex (so stdout+stderr interleave without interleaving *within* a write), assigns the next `seq`, and calls `emit(seq, copyOfBytes)`. Chunks are whatever the OS pipe delivers (typically ≤64KB); no extra buffering/line-assembly.
- `StepResult` returns with `Stdout`/`Stderr` empty; `ExitCode` and `OutputEnv` populated as today (env parsing unchanged).
- On timeout/exec error, put the short note in `StepResult.Stderr` (terminal, not streamed) so the runner can still surface a failure reason even if nothing streamed.
Agent loop (`agent/internal/sync/sync.go`, the `cmd.RunStep != nil` goroutine): pass an `emit` closure that sends `AgentMessage{ServerId, AgentToken, StepOutput: &pb.StepOutputChunk{CommandId, Seq, Data}}` through the existing mutex-guarded `send()`. After `RunStep` returns, send a final `StepOutput{eof:true, seq:last+1}` then the terminal `StepResult` (both via `send()`). Ordering: all chunks, then eof, then StepResult.
---
## 5. Server write path
### 5.1 Log writer registry (`server/internal/services/steplogs.go`)
Parallel to `StepResults`. Keyed by `command_id`:
```go
type stepLogWriter struct {
f *os.File
mu sync.Mutex
carry []byte // held-back tail for boundary-safe masking
secrets []string // secret literals to mask
maxSecret int
}
var StepLogs = &stepLogRegistry{ ... }
func (r *stepLogRegistry) Open(commandID, path string, secrets []string) (*stepLogWriter, error)
func (r *stepLogRegistry) Append(commandID string, data []byte) // masked write
func (r *stepLogRegistry) Close(commandID string) // flush carry, close file
```
- `Append` masking: concatenate `carry+data`, mask all secret literals (`ReplaceAll(v,"***")`), then write everything except the last `maxSecret-1` bytes; keep those as the new `carry`. `Close` masks+writes the remaining carry. If `secrets` empty, write straight through (no carry).
- The file handle is opened append-only (`O_APPEND|O_CREATE|O_WRONLY`, `0600`); dir `0700`.
### 5.2 Stream delivery (`server/internal/grpc/server.go`)
In the receive loop, after the `m.StepResult` block, add:
```go
if m.StepOutput != nil {
if m.StepOutput.Eof {
services.StepLogs.Close(m.StepOutput.CommandId)
} else {
services.StepLogs.Append(m.StepOutput.CommandId, m.StepOutput.Data)
}
}
```
### 5.3 Runner changes (`server/internal/services/workflow_runner.go`)
- Resolve the log dir + run/server file path once per server-run; ensure `<logdir>/<run_id>/` exists.
- Before dispatching each step: write the step marker line to the file (`\n===== step <order>: <name> =====\n`), record the current file byte offset as the step's `log_offset` (persisted on the `StepRun`), and `StepLogs.Open(commandID, path, secretVals)` **before** `DispatchRunStep` (same ordering rule as `StepResults.Await`).
- `dispatchAndWait` no longer expects stdout/stderr in the result. On terminal `StepResult`, `StepLogs.Close(commandID)` is driven by the agent's eof; the runner also calls `Close` defensively on timeout/dispatch-failure (idempotent).
- **Drop** `stdout`/`stderr` from `finishStep` persistence. Masking of the streamed body is done in `Append`; `run_env`/`output_env` masking (existing, from the merged fix) stays.
- The shared server-run file is written by two writers that never overlap in time (steps are serial, and the runner writes each step marker *before* `StepLogs.Open`): (a) the runner writes markers directly to the path, serially between steps; (b) `StepLogs` writes chunks during a step. **Resolved approach:** `Open(commandID, path, secrets)` opens the path fresh with `O_APPEND|O_CREATE|O_WRONLY` for that step and `Close(commandID)` closes it on eof. One handle live at a time per server-run (serial steps guarantee this), so there is no shared-handle race and no ref-counting. The runner's marker write is a separate short `O_APPEND` open/write/close on the same path.
### 5.4 Data model (`server/internal/models/workflow.go`)
`StepRun`:
- **Remove** `Stdout`, `Stderr` string fields.
- **Add** `LogOffset int64 `bson:"log_offset" json:"log_offset"`` — byte offset in the server-run file where this step's marker begins.
`ServerRun` gains nothing structural (its file path is derivable: `<logdir>/<run_id>/<server_id>.log`).
---
## 6. REST API (`server/internal/api/workflows.go`)
- `GET /api/runs/:runId/servers/:serverId/logs` — returns the whole server-run log file (`text/plain`). 404 if absent. Used post-run and as SSE fallback.
- `GET /api/runs/:runId/servers/:serverId/logs/stream`**SSE**. Opens the file, streams existing content as `data:` events, then polls for appends (~500ms) emitting new bytes, until the server-run status is terminal (success/failed/skipped/cancelled) AND no more bytes, then sends a final `event: done` and closes. Sets `Content-Type: text/event-stream`, disables gin's buffering. Guards against path traversal (runId/serverId are used as literal path segments — validate they are UUIDs / contain no separators).
Log content served by these endpoints is already masked (masking happens at write time), so no masking needed on read.
---
## 7. Retention
### 7.1 Setting
`settings` collection gains `workflow_log_retention_days int` (default 30 when unset). Read/write via the existing settings service + surfaced in `/settings` UI as a number input. `0` or negative disables sweeping (keep forever) — document this.
### 7.2 Sweeper (`server/internal/services/steplogs.go` or `logsweeper.go`)
- `StartLogSweeper()` launched at server startup (next to index setup): hourly `time.Ticker`.
- Each tick: read retention setting; if ≤0 skip. Compute cutoff = `now - retentionDays`. For each `<logdir>/<run_id>/` dir, look up the run's `finished_at` (query `workflow_runs` by run_id); if finished and older than cutoff, `os.RemoveAll` the dir. Fallback to dir mtime if the run doc is gone.
- Also run once at startup.
---
## 8. Frontend
### 8.1 API client (`web/lib/api.ts`)
- `StepRun`: remove `stdout`/`stderr`; add `log_offset: number`.
- Add `getServerRunLog(runId, serverId): Promise<string>` (GET .../logs).
- SSE consumed directly via `EventSource` in the component (not through the `request` helper), URL built from the same base.
- Settings type gains `workflow_log_retention_days`.
### 8.2 Run detail (`web/app/workflows/[id]/runs/[runId]/page.tsx`)
- Per-server card: while the server-run is `running`, open an `EventSource` to the stream endpoint and render a live `<pre>` terminal that appends incoming chunks (auto-scroll). Close the source on `event: done`, unmount, or terminal status.
- After completion: fetch the whole file once and render it; step `<details>` still list status/exit/attempts pills. (Per-step slicing by `log_offset` is optional polish — v1 may show the whole server log under the card and keep step pills as the status summary.)
- Remove reliance on `st.stdout`/`st.stderr` (fields gone).
### 8.3 Settings (`web/app/settings/page.tsx`)
- Add a "Workflow log retention (days)" number input bound to `workflow_log_retention_days`, saved via the existing settings mutation. Note that `0` = keep forever.
---
## 9. Security
- Secret masking moves to the streaming write path but remains server-side and boundary-safe (carry buffer). Same `***` replacement.
- Log files `0600`, dirs `0700`, under a dedicated log dir.
- SSE/read endpoints validate `runId`/`serverId` as UUID-shaped path segments to prevent traversal; they are session-authed (same `apiGroup`).
- Terminal `StepResult.Stderr` (error notes only) is still masked before any persistence (it is no longer persisted as log body; if surfaced, mask against secretVals).
---
## 10. Out of scope
- Per-step (rather than per-server) live SSE channels.
- Log compression / rotation within a run, remote log storage (S3), download-as-zip.
- Full-text search over logs.
- Backfilling/migrating already-existing `workflow_runs` stdout/stderr into files (pre-existing runs keep whatever they had; new field just won't be set — acceptable, feature is new).
- Tests (skipped, consistent with the Workflows iteration).
```
Binary file not shown.
+8 -2
View File
@@ -32,6 +32,10 @@ function Invoke-Native {
function Invoke-NativeSoft {
param([string]$File, [string[]]$Arguments)
Write-Log ("RUN(soft): {0} {1}" -f $File, ($Arguments -join " "))
# Native stderr merged via 2>&1 becomes terminating errors under
# ErrorActionPreference=Stop; force Continue in this scope so a benign nssm
# message (e.g. "service has not been started") never aborts setup.
$ErrorActionPreference = "Continue"
$out = & $File @Arguments 2>&1
if ($out) { Write-Log ("OUT: {0}" -f ($out -join "`n")) }
Write-Log ("EXIT: {0}" -f $LASTEXITCODE)
@@ -121,8 +125,10 @@ tls: true
Invoke-Native -File $nssm -Arguments @("set", "VantageAgent", "AppRotateOnline", "1")
Invoke-Native -File $nssm -Arguments @("set", "VantageAgent", "AppRotateBytes", "1048576")
# restart (not just start) so an upgrade picks up the new binary
Invoke-NativeSoft -File $nssm -Arguments @("restart", "VantageAgent")
# Service is freshly (re)installed and stopped here (teardown removed the old
# one on upgrade), so start it. "restart" would try to stop a not-running
# service and emit a stderr error.
Invoke-NativeSoft -File $nssm -Arguments @("start", "VantageAgent")
Write-Log "=== setup ok ==="
exit 0
+4 -1
View File
@@ -1,10 +1,13 @@
<?xml version="1.0" encoding="UTF-8"?>
<?ifndef Version ?>
<?define Version = "0.0.0.0" ?>
<?endif?>
<Wix xmlns="http://wixtoolset.org/schemas/v4/wxs">
<Package Name="Vantage Agent" Manufacturer="Vantage"
Version="$(var.Version)" UpgradeCode="7d1e6d2c-2a5f-4b3e-9c3a-8a1b2c3d4e5f"
Scope="perMachine">
<MajorUpgrade DowngradeErrorMessage="A newer version is already installed." />
<MajorUpgrade DowngradeErrorMessage="A newer version is already installed."
Schedule="afterInstallInitialize" />
<MediaTemplate EmbedCab="yes" />
<!-- Public properties settable via msiexec: SERVERID, TOKEN, SERVERURL -->
+25
View File
@@ -55,6 +55,8 @@ message AgentMessage {
oneof payload {
AgentReady ready = 3;
CommandResult result = 4;
StepResult step_result = 5;
StepOutputChunk step_output = 6;
}
}
@@ -89,6 +91,7 @@ message ServerCommand {
DeleteKeyCmd delete_key = 3;
UpdateAgentCmd update_agent = 4;
ApplyUpdatesCmd apply_updates = 5;
RunStepCmd run_step = 6;
}
}
@@ -108,3 +111,25 @@ message GenerateKeyCmd {
string passphrase = 4; // empty = no passphrase
string comment = 5; // embedded in public key
}
message RunStepCmd {
string interpreter = 1; // "bash" | "powershell"
string script = 2;
map<string, string> env = 3;
int32 timeout_seconds = 4;
}
message StepResult {
string command_id = 1;
int32 exit_code = 2;
string stdout = 3;
string stderr = 4;
map<string, string> output_env = 5;
}
message StepOutputChunk {
string command_id = 1;
uint64 seq = 2;
bytes data = 3;
bool eof = 4;
}
+4
View File
@@ -27,6 +27,10 @@ func main() {
log.Printf("warning: failed to ensure secret indexes: %v", err)
}
if err := services.EnsureWorkflowIndexes(); err != nil {
log.Printf("warning: failed to ensure workflow indexes: %v", err)
}
redisAddr := getEnv("REDIS_ADDR", "localhost:6379")
if err := auth.InitRedis(redisAddr); err != nil {
log.Fatalf("failed to connect to Redis: %v", err)
+2
View File
@@ -78,6 +78,8 @@ func RegisterRoutes(r *gin.Engine) {
apiGroup.POST("/console/connect", consoleConnect)
apiGroup.GET("/console/tunnel", consoleTunnel)
registerWorkflowRoutes(apiGroup)
}
}
+180
View File
@@ -0,0 +1,180 @@
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")
updated, err := services.GetWorkflow(c.Param("id"))
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
}
c.JSON(http.StatusOK, updated)
}
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
}
services.LogEvent("workflow.run_cancelled", actorFromCtx(c), "", c.Param("runId"), "run cancelled")
c.JSON(http.StatusOK, gin.H{"cancelled": true})
}
+29 -4
View File
@@ -71,6 +71,7 @@ type ServerCommand struct {
DeleteKey *DeleteKeyCmd `json:"delete_key,omitempty"`
UpdateAgent *UpdateAgentCmd `json:"update_agent,omitempty"`
ApplyUpdates *ApplyUpdatesCmd `json:"apply_updates,omitempty"`
RunStep *RunStepCmd `json:"run_step,omitempty"`
}
type DeleteKeyCmd struct {
@@ -91,10 +92,12 @@ type GenerateKeyCmd struct {
}
type AgentMessage struct {
ServerId string `json:"server_id"`
AgentToken string `json:"agent_token"`
Ready *AgentReady `json:"ready,omitempty"`
Result *CommandResult `json:"result,omitempty"`
ServerId string `json:"server_id"`
AgentToken string `json:"agent_token"`
Ready *AgentReady `json:"ready,omitempty"`
Result *CommandResult `json:"result,omitempty"`
StepResult *StepResult `json:"step_result,omitempty"`
StepOutput *StepOutputChunk `json:"step_output,omitempty"`
}
type AgentReady struct{}
@@ -105,6 +108,28 @@ type CommandResult struct {
Message string `json:"message"`
}
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"`
}
type StepOutputChunk struct {
CommandId string `json:"command_id"`
Seq uint64 `json:"seq"`
Data []byte `json:"data,omitempty"`
Eof bool `json:"eof,omitempty"`
}
// CommandStream server-side interface
type Vantage_CommandStreamServer interface {
+3
View File
@@ -129,6 +129,9 @@ func (s *vantageServer) CommandStream(stream pb.Vantage_CommandStreamServer) err
r := m.Result
log.Printf("agent %s cmd %s: success=%v %s", srv.ServerID, r.CommandId, r.Success, r.Message)
}
if m.StepResult != nil {
services.StepResults.Deliver(m.StepResult)
}
}
}()
+99
View File
@@ -0,0 +1,99 @@
package models
import (
"time"
"go.mongodb.org/mongo-driver/v2/bson"
)
type InputParam struct {
Name string `bson:"name" json:"name"`
Default string `bson:"default" json:"default"`
Description string `bson:"description" json:"description"`
}
type WorkflowStep struct {
ID bson.ObjectID `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"`
DeclaredInputs []InputParam `bson:"declared_inputs" json:"declared_inputs"`
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"`
Inputs map[string]string `bson:"inputs,omitempty" json:"inputs,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 bson.ObjectID `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"`
Inputs map[string]string `bson:"inputs" json:"inputs"`
}
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 bson.ObjectID `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"`
}
+6
View File
@@ -62,6 +62,12 @@ func (d *commandDispatcher) dispatch(serverID string, cmd *pb.ServerCommand) err
}
}
// 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})
}
// KeyGenParams carries all options for a generate-key command.
type KeyGenParams struct {
Label string
+49
View File
@@ -0,0 +1,49 @@
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
}
}
+415
View File
@@ -0,0 +1,415 @@
package services
import (
"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
}
inputs := map[string]string{}
for _, p := range lib.DeclaredInputs {
if ref.Inputs != nil {
if v, ok := ref.Inputs[p.Name]; ok {
inputs[p.Name] = v
continue
}
}
inputs[p.Name] = p.Default
}
rs := models.ResolvedStep{
Order: ref.Order,
Name: lib.Name,
Interpreter: lib.Interpreter,
Script: lib.Script,
SecretRefs: lib.SecretRefs,
OnFailure: ref.OnFailure,
MaxRetries: ref.MaxRetries,
Inputs: inputs,
}
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, "status": "running"},
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{}
allSecrets := 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)
for k, v := range secretVals {
allSecrets[k] = v
}
cmdEnv := map[string]string{}
for k, v := range step.Inputs {
cmdEnv[k] = v
}
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{} // masked copy, safe to persist
if res != nil {
stdout = maskSecrets(res.Stdout, allSecrets)
stderr = maskSecrets(res.Stderr, allSecrets)
exit = res.ExitCode
for k, v := range res.OutputEnv {
runEnv[k] = v // real, unmasked value threads forward to later steps
outEnv[k] = maskSecrets(v, allSecrets)
}
} 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"
}
// Persist only a masked copy of runEnv; the real (unmasked) runEnv was already
// used above to build cmdEnv for each step and must never be written to the DB.
maskedRunEnv := make(map[string]string, len(runEnv))
for k, v := range runEnv {
maskedRunEnv[k] = maskSecrets(v, allSecrets)
}
setServerRun(runID, srvIdx, bson.M{
"server_runs.$.status": status,
"server_runs.$.finished_at": fin,
"server_runs.$.run_env": maskedRunEnv,
})
}
// 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
}
+204
View File
@@ -0,0 +1,204 @@
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 s.DeclaredInputs == nil {
s.DeclaredInputs = []models.InputParam{}
}
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,
"declared_inputs": s.DeclaredInputs,
"secret_refs": s.SecretRefs,
"updated_at": time.Now(),
}})
return err
}
func DeleteStep(stepID string) error {
ctx, cancel := wfCtx()
defer cancel()
if _, err := db.Col("workflow_steps").DeleteOne(ctx, bson.M{"step_id": stepID}); err != nil {
return err
}
// Cascade: remove this step from every workflow that references it, re-sequencing orders.
cur, err := db.Col("workflows").Find(ctx, bson.M{"steps.step_id": stepID})
if err != nil {
return err
}
defer cur.Close(ctx)
var wfs []models.Workflow
if err := cur.All(ctx, &wfs); err != nil {
return err
}
for _, w := range wfs {
kept := make([]models.WorkflowStepRef, 0, len(w.Steps))
for _, ref := range w.Steps {
if ref.StepID == stepID {
continue
}
kept = append(kept, ref)
}
for i := range kept {
kept[i].Order = i
}
if _, err := db.Col("workflows").UpdateOne(ctx,
bson.M{"workflow_id": w.WorkflowID},
bson.M{"$set": bson.M{"steps": kept, "updated_at": time.Now()}},
); err != nil {
return err
}
}
return nil
}
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
}
+44 -6
View File
@@ -15,7 +15,11 @@ export default function ServerConsolePage() {
const serverId = params.id as string;
const containerRef = useRef<HTMLDivElement>(null);
const connectionRef = useRef<{ disconnect: () => void } | null>(null);
const connectionRef = useRef<{
disconnect: () => void;
setScale: (scale: number) => void;
resize: (width: number, height: number) => void;
} | null>(null);
const [protocol, setProtocol] = useState<string>(searchParams.get("protocol") || "");
const [keyId, setKeyId] = useState<string>("");
@@ -27,6 +31,8 @@ export default function ServerConsolePage() {
const [connected, setConnected] = useState(false);
const [error, setError] = useState<string | null>(null);
const [pending, setPending] = useState<{ token: string; wsPath: string } | null>(null);
const [zoom, setZoom] = useState(1);
const dprRef = useRef(1);
// Inject the vendored Guacamole client script once.
useEffect(() => {
@@ -103,17 +109,36 @@ export default function ServerConsolePage() {
const wsProto = location.protocol === "https:" ? "wss" : "ws";
const wsUrl = `${wsProto}://${location.host}${pending.wsPath}`;
const rect = containerRef.current.getBoundingClientRect();
const dpi = Math.round(96 * (window.devicePixelRatio || 1));
const dpr = window.devicePixelRatio || 1;
dprRef.current = dpr;
// Request the remote at device-pixel resolution with a fixed 96 dpi, then
// scale the display back down by dpr. Folding dpr into `dpi` instead makes
// the remote enlarge everything, which reads as a zoomed-in view.
const connectData =
`token=${encodeURIComponent(pending.token)}` +
`&width=${Math.floor(rect.width)}` +
`&height=${Math.floor(rect.height)}` +
`&dpi=${dpi}`;
`&width=${Math.floor(rect.width * dpr)}` +
`&height=${Math.floor(rect.height * dpr)}` +
`&dpi=96`;
connectionRef.current = openConsole(containerRef.current, wsUrl, connectData);
connectionRef.current.setScale(zoom / dpr);
setPending(null);
}, [connected, pending]);
// Apply zoom live without reconnecting: resize the remote to a resolution
// that, once scaled to fit the container, yields the requested zoom. Higher
// zoom = fewer remote pixels rendered larger. Display always fits the
// container exactly, so no scrollbars appear.
useEffect(() => {
if (!connectionRef.current || !containerRef.current) return;
const rect = containerRef.current.getBoundingClientRect();
const dpr = dprRef.current;
const remoteW = Math.floor((rect.width * dpr) / zoom);
const remoteH = Math.floor((rect.height * dpr) / zoom);
connectionRef.current.resize(remoteW, remoteH);
connectionRef.current.setScale(zoom / dpr);
}, [zoom]);
function handleDisconnect() {
connectionRef.current?.disconnect();
connectionRef.current = null;
@@ -246,12 +271,25 @@ export default function ServerConsolePage() {
<Button variant="danger" onClick={handleDisconnect}>
Disconnect
</Button>
<label className="text-sm text-text-secondary">Scale</label>
<select
value={zoom}
onChange={(e) => setZoom(Number(e.target.value))}
className="rounded-lg border border-border bg-surface-2 px-3 py-2 text-sm text-text-primary focus:border-accent/50 focus:outline-none focus:ring-1 focus:ring-accent/30"
>
<option value={0.5}>50%</option>
<option value={0.75}>75%</option>
<option value={1}>100%</option>
<option value={1.25}>125%</option>
<option value={1.5}>150%</option>
<option value={2}>200%</option>
</select>
</div>
)}
<div
ref={containerRef}
className="min-h-[500px] flex-1 rounded-lg border border-border bg-black"
className="min-h-[500px] flex-1 overflow-hidden rounded-lg border border-border bg-black"
/>
</div>
);
+560
View File
@@ -0,0 +1,560 @@
"use client";
import { useEffect, useState } from "react";
import Link from "next/link";
import { useParams, useRouter } from "next/navigation";
import { useQuery, useQueryClient } from "@tanstack/react-query";
import { api, Workflow, WorkflowStep, WorkflowStepRef, SecretGroupSummary } from "@/lib/api";
import { Button } from "@/components/ui";
import { EditStepModal } from "@/components/workflows/EditStepModal";
import { EditWorkflowModal } from "@/components/workflows/EditWorkflowModal";
const inputClass =
"w-full rounded-lg border border-border bg-surface-2 px-3 py-2 text-sm text-text-primary placeholder-text-secondary/50 focus:border-signal focus:outline-none focus:ring-1 focus:ring-signal";
type DragPayload = { kind: "lib"; stepId: string } | { kind: "move"; from: number };
function ShellBadge({ interpreter }: { interpreter: "bash" | "powershell" }) {
const isBash = interpreter === "bash";
return (
<span
className={`rounded px-1.5 py-0.5 font-mono text-[10px] uppercase ${
isBash ? "bg-bash/15 text-bash" : "bg-pwsh/15 text-pwsh"
}`}
>
{isBash ? "bash" : "pwsh"}
</span>
);
}
export default function WorkflowBuilder() {
const params = useParams<{ id: string }>();
const id = params.id;
const router = useRouter();
const queryClient = useQueryClient();
const [wf, setWf] = useState<Workflow | null>(null);
const [selected, setSelected] = useState<number | null>(null);
const [search, setSearch] = useState("");
const [saving, setSaving] = useState(false);
const [running, setRunning] = useState(false);
const [error, setError] = useState<string | null>(null);
const [groupKeys, setGroupKeys] = useState<Record<string, string[]>>({});
const [editWorkflowOpen, setEditWorkflowOpen] = useState(false);
const [editingStep, setEditingStep] = useState<WorkflowStep | null>(null);
const [editStepOpen, setEditStepOpen] = useState(false);
const [dragOverZone, setDragOverZone] = useState<number | null>(null);
const { data: loaded } = useQuery({
queryKey: ["workflow", id],
queryFn: () => api.getWorkflow(id),
});
const { data: library } = useQuery({ queryKey: ["steps"], queryFn: api.listSteps });
const { data: secretGroups } = useQuery({
queryKey: ["secret-groups"],
queryFn: api.listSecretGroups,
});
useEffect(() => {
if (loaded && !wf) setWf(loaded);
// eslint-disable-next-line react-hooks/exhaustive-deps
}, [loaded]);
// Lazily fetch the keys for every secret group so the inspector's
// secret-ref checklist can offer "group/KEY" options.
useEffect(() => {
if (!secretGroups) return;
secretGroups.forEach((g: SecretGroupSummary) => {
if (groupKeys[g.group] !== undefined) return;
api.getSecretGroup(g.group)
.then((res) =>
setGroupKeys((prev) => ({
...prev,
[g.group]: res.secrets.map((s) => s.key),
})),
)
.catch(() => {
setGroupKeys((prev) => ({ ...prev, [g.group]: [] }));
});
});
// eslint-disable-next-line react-hooks/exhaustive-deps
}, [secretGroups]);
if (!wf) {
return <div className="p-8 text-text-secondary">Loading</div>;
}
const libById = (sid: string) => library?.find((l) => l.step_id === sid);
const sortedSteps = [...wf.steps].sort((a, b) => a.order - b.order);
const selectedRef = selected !== null ? sortedSteps[selected] : null;
const selectedLib = selectedRef ? libById(selectedRef.step_id) : null;
const selectedIdxInWf = selectedRef ? wf.steps.indexOf(selectedRef) : -1;
const save = async () => {
setSaving(true);
setError(null);
try {
const updated = await api.updateWorkflow(id, wf);
if (!updated || !Array.isArray(updated.steps)) {
setError("Save failed: server returned an unexpected response.");
return;
}
setWf(updated);
} catch (e) {
setError((e as Error).message);
} finally {
setSaving(false);
}
};
const run = async () => {
setRunning(true);
setError(null);
try {
const { run_id } = await api.runWorkflow(id);
router.push(`/workflows/${id}/runs/${run_id}`);
} catch (e) {
setError((e as Error).message);
setRunning(false);
}
};
const resequence = (steps: WorkflowStepRef[]) => steps.map((r, i) => ({ ...r, order: i }));
const insertLibStep = (stepId: string, pos: number) => {
const next = [...sortedSteps];
next.splice(pos, 0, { step_id: stepId, order: 0, on_failure: "stop", max_retries: 0 });
setWf({ ...wf, steps: resequence(next) });
};
const moveStep = (from: number, pos: number) => {
const next = [...sortedSteps];
const [item] = next.splice(from, 1);
const target = from < pos ? pos - 1 : pos;
next.splice(target, 0, item);
setWf({ ...wf, steps: resequence(next) });
if (selected === from) setSelected(target);
else if (selected !== null) {
if (from < selected && target >= selected) setSelected(selected - 1);
else if (from > selected && target <= selected) setSelected(selected + 1);
}
};
const handleDrop = (e: React.DragEvent, pos: number) => {
e.preventDefault();
setDragOverZone(null);
const raw = e.dataTransfer.getData("text/plain");
if (!raw) return;
let payload: DragPayload;
try {
payload = JSON.parse(raw);
} catch {
return;
}
if (payload.kind === "lib") {
insertLibStep(payload.stepId, pos);
} else if (payload.kind === "move") {
moveStep(payload.from, pos);
}
};
const updateRef = (idx: number, patch: Partial<WorkflowStepRef>) =>
setWf({
...wf,
steps: wf.steps.map((r, i) => (i === idx ? { ...r, ...patch } : r)),
});
const removeStep = (idx: number) => {
const remaining = resequence(wf.steps.filter((_, i) => i !== idx));
setWf({ ...wf, steps: remaining });
setSelected(null);
};
const toggleSecretRef = (ref: string) => {
if (selectedIdxInWf === -1 || !selectedRef) return;
const current = selectedRef.overrides?.secret_refs ?? [];
const next = current.includes(ref) ? current.filter((r) => r !== ref) : [...current, ref];
updateRef(selectedIdxInWf, { overrides: { ...selectedRef.overrides, secret_refs: next } });
};
const filteredLibrary = (library ?? []).filter((s) => s.name.toLowerCase().includes(search.toLowerCase()));
const bashSteps = filteredLibrary.filter((s) => s.interpreter === "bash");
const pwshSteps = filteredLibrary.filter((s) => s.interpreter === "powershell");
const upstreamOutputsFor = (i: number) =>
Array.from(new Set(sortedSteps.slice(0, i).flatMap((r) => libById(r.step_id)?.declared_outputs ?? [])));
const DropZone = ({ pos }: { pos: number }) => (
<div
className={`h-3 w-full transition-all ${dragOverZone === pos ? "h-8 rounded bg-signal/15 border border-dashed border-signal/50" : ""}`}
onDragOver={(e) => {
e.preventDefault();
setDragOverZone(pos);
}}
onDragLeave={() => setDragOverZone((z) => (z === pos ? null : z))}
onDrop={(e) => handleDrop(e, pos)}
/>
);
return (
<>
<div className="flex items-center gap-3 border-b border-border bg-surface px-4 py-3">
<span className="h-2.5 w-2.5 rounded-full bg-signal" />
<div className="flex items-center gap-1.5 text-sm">
<span className="text-text-secondary">Workflows /</span>
<span className="font-medium text-text-primary">{wf.name}</span>
<span className="text-text-secondary">· draft</span>
</div>
<div className="ml-auto flex items-center gap-2">
<span className="rounded-full border border-border bg-surface-2 px-3 py-1 text-xs text-text-secondary">
{wf.target_server_ids.length} servers
</span>
<Link
href={`/workflows/${id}/runs`}
className="rounded-lg border border-border bg-surface-2 px-3 py-1.5 text-sm text-text-secondary hover:text-text-primary"
>
Runs
</Link>
<Button variant="secondary" size="sm" onClick={() => setEditWorkflowOpen(true)}>
Edit
</Button>
<Button variant="secondary" size="sm" loading={saving} onClick={save}>
Save
</Button>
<Button
size="sm"
loading={running}
onClick={run}
className="bg-signal text-signal-ink border-transparent hover:bg-signal/90"
>
Run workflow
</Button>
</div>
</div>
{error && (
<div className="border-b border-danger/30 bg-danger/10 px-4 py-2 text-sm text-danger">{error}</div>
)}
<div className="grid h-[calc(100vh-53px)] grid-cols-[264px_1fr_320px]">
{/* LEFT: library */}
<aside className="overflow-auto border-r border-border bg-surface p-3">
<div className="mb-2 flex items-center justify-between">
<h2 className="text-xs font-bold uppercase tracking-wide text-text-secondary">Step Library</h2>
<Button
variant="ghost"
size="sm"
onClick={() => {
setEditingStep(null);
setEditStepOpen(true);
}}
>
+
</Button>
</div>
<input
className={`${inputClass} mb-3`}
placeholder="Search steps…"
value={search}
onChange={(e) => setSearch(e.target.value)}
/>
{bashSteps.length > 0 && (
<>
<h3 className="mb-1 mt-2 text-[11px] font-bold uppercase tracking-wide text-text-secondary">
Shared · Bash
</h3>
{bashSteps.map((s) => (
<LibraryCard key={s.step_id} step={s} onAdd={() => insertLibStep(s.step_id, sortedSteps.length)} onEdit={() => {
setEditingStep(s);
setEditStepOpen(true);
}} />
))}
</>
)}
{pwshSteps.length > 0 && (
<>
<h3 className="mb-1 mt-3 text-[11px] font-bold uppercase tracking-wide text-text-secondary">
Shared · PowerShell
</h3>
{pwshSteps.map((s) => (
<LibraryCard key={s.step_id} step={s} onAdd={() => insertLibStep(s.step_id, sortedSteps.length)} onEdit={() => {
setEditingStep(s);
setEditStepOpen(true);
}} />
))}
</>
)}
{filteredLibrary.length === 0 && <p className="mt-2 text-xs text-text-secondary">No steps found.</p>}
</aside>
{/* CENTER: canvas */}
<main
className="overflow-auto bg-background bg-[radial-gradient(circle_at_1px_1px,theme(colors.border)_1px,transparent_0)] bg-[length:22px_22px] p-8"
>
<div className="mx-auto flex w-[340px] flex-col items-center">
<DropZone pos={0} />
{sortedSteps.map((ref, i) => {
const lib = libById(ref.step_id);
const outs = upstreamOutputsFor(i);
const script = ref.overrides?.script ?? lib?.script ?? "";
const wfIdx = wf.steps.indexOf(ref);
const isSelected = selected === i;
return (
<div key={wfIdx} className="w-full">
{i > 0 && (
<div className="flex flex-col items-center py-1">
<div className="h-[13px] w-0.5 bg-border" />
{outs.length > 0 && (
<div className="flex w-fit max-w-[300px] flex-wrap items-center justify-center gap-1 rounded-full border border-dashed border-signal/55 bg-surface px-3 py-1">
<span className="text-[10px] uppercase text-text-secondary">passes</span>
{outs.map((o) => (
<span
key={o}
className="rounded bg-signal px-2 py-0.5 font-mono text-[11px] text-signal-ink"
>
{o}
</span>
))}
</div>
)}
<div className="h-[13px] w-0.5 bg-border" />
</div>
)}
<div
draggable
onDragStart={(e) => {
e.dataTransfer.setData("text/plain", JSON.stringify({ kind: "move", from: i }));
}}
onClick={() => setSelected(i)}
className={`w-[340px] cursor-pointer rounded-[10px] border bg-surface p-3 ${
isSelected ? "border-signal ring-2 ring-signal/40" : "border-border"
}`}
>
<div className="mb-2 flex items-center gap-2">
<span className="grid h-5 w-5 place-items-center rounded border border-border font-mono text-[10px] text-text-secondary">
{i + 1}
</span>
<span className="text-sm font-medium text-text-primary">{lib?.name ?? ref.step_id}</span>
{lib && <ShellBadge interpreter={lib.interpreter} />}
</div>
<pre className="max-h-16 overflow-hidden text-ellipsis whitespace-pre-wrap rounded border border-border bg-surface-2 p-2 font-mono text-xs text-text-secondary">
{script.slice(0, 200)}
</pre>
</div>
<DropZone pos={i + 1} />
</div>
);
})}
{sortedSteps.length === 0 && (
<button
onDragOver={(e) => e.preventDefault()}
onDrop={(e) => handleDrop(e, 0)}
className="mt-2 w-full rounded-[10px] border border-dashed border-border bg-surface py-6 text-sm text-text-secondary hover:border-signal/50 hover:text-text-primary"
>
+ Drop a step here
</button>
)}
</div>
</main>
{/* RIGHT: inspector */}
<aside className="overflow-auto border-l border-border bg-surface p-4">
{selected === null || !selectedRef ? (
<p className="text-sm text-text-secondary">Select a step to configure it.</p>
) : (
<div className="space-y-4">
<div>
<div className="mb-1 text-[11px] font-bold uppercase tracking-wide text-text-secondary">
Step {selected + 1} · Inspector
</div>
<div className="flex items-center gap-2">
{selectedLib && <ShellBadge interpreter={selectedLib.interpreter} />}
<h2 className="text-sm font-bold text-text-primary">{selectedLib?.name ?? selectedRef.step_id}</h2>
</div>
</div>
<div className="border-b border-border pb-4">
<label className="mb-1 block text-xs uppercase text-text-secondary">Command</label>
<textarea
className={`${inputClass} h-32 font-mono text-xs`}
value={selectedRef.overrides?.script ?? selectedLib?.script ?? ""}
onChange={(e) =>
updateRef(selectedIdxInWf, {
overrides: { ...selectedRef.overrides, script: e.target.value },
})
}
/>
<p className="mt-1 text-xs text-text-secondary">
Write <code className="text-signal">KEY=value</code> to <code className="text-signal">$WORKFLOW_ENV</code> to expose
it to later steps.
</p>
</div>
{(selectedLib?.declared_inputs ?? []).length > 0 && (
<div className="border-b border-border pb-4">
<label className="mb-2 block text-xs uppercase text-text-secondary">Inputs</label>
<div className="space-y-2">
{selectedLib?.declared_inputs.map((param) => (
<div key={param.name}>
<div className="mb-1 font-mono text-xs text-text-primary">{param.name}</div>
{param.description && (
<div className="mb-1 text-[11px] text-text-secondary">{param.description}</div>
)}
<input
className={inputClass}
placeholder={param.default}
value={selectedRef.inputs?.[param.name] ?? ""}
onChange={(e) =>
updateRef(selectedIdxInWf, {
inputs: { ...selectedRef.inputs, [param.name]: e.target.value },
})
}
/>
</div>
))}
</div>
</div>
)}
<div className="border-b border-border pb-4">
<label className="mb-2 block text-xs uppercase text-text-secondary">Inputs · from upstream</label>
<div className="flex flex-wrap gap-1">
{upstreamOutputsFor(selected).length === 0 && (
<p className="text-xs text-text-secondary">No upstream outputs.</p>
)}
{upstreamOutputsFor(selected).map((o) => (
<span key={o} className="flex items-center gap-1 rounded bg-surface-2 border border-border px-2 py-0.5 font-mono text-[11px] text-text-primary">
<span className="text-[9px] uppercase text-text-secondary">in</span>
{o}
</span>
))}
</div>
</div>
<div className="border-b border-border pb-4">
<label className="mb-2 block text-xs uppercase text-text-secondary">Outputs · to $WORKFLOW_ENV</label>
<div className="flex flex-wrap gap-1">
{(selectedLib?.declared_outputs ?? []).length === 0 && (
<p className="text-xs text-text-secondary">No declared outputs.</p>
)}
{(selectedLib?.declared_outputs ?? []).map((o) => (
<span key={o} className="flex items-center gap-1 rounded bg-signal px-2 py-0.5 font-mono text-[11px] text-signal-ink">
<span className="text-[9px] uppercase">out</span>
{o}
</span>
))}
</div>
</div>
<div className="border-b border-border pb-4">
<label className="mb-2 block text-xs uppercase text-text-secondary">Secret refs</label>
<div className="max-h-56 space-y-2 overflow-auto rounded-lg border border-border p-2">
{secretGroups?.map((g) => (
<div key={g.group}>
<div className="font-mono text-[11px] font-semibold text-text-secondary">{g.group}</div>
{(groupKeys[g.group] ?? []).map((key) => {
const ref = `${g.group}/${key}`;
const checked = (selectedRef.overrides?.secret_refs ?? []).includes(ref);
return (
<label key={ref} className="ml-2 flex cursor-pointer items-center gap-2 text-xs text-text-primary">
<input
type="checkbox"
className="accent-signal"
checked={checked}
onChange={() => toggleSecretRef(ref)}
/>
<span className="font-mono">{key}</span>
</label>
);
})}
{(groupKeys[g.group] ?? []).length === 0 && <p className="ml-2 text-[11px] text-text-secondary">No keys.</p>}
</div>
))}
{secretGroups && secretGroups.length === 0 && <p className="text-xs text-text-secondary">No secret groups yet.</p>}
</div>
</div>
<div className="border-b border-border pb-4">
<label className="mb-1 block text-xs uppercase text-text-secondary">On failure</label>
<select
className={inputClass}
value={selectedRef.on_failure}
onChange={(e) =>
updateRef(selectedIdxInWf, {
on_failure: e.target.value as WorkflowStepRef["on_failure"],
})
}
>
<option value="stop">Stop workflow</option>
<option value="continue">Continue</option>
<option value="retry">Retry</option>
</select>
{selectedRef.on_failure === "retry" && (
<div className="mt-2">
<label className="mb-1 block text-xs uppercase text-text-secondary">Max retries</label>
<input
type="number"
min={1}
className={inputClass}
value={selectedRef.max_retries}
onChange={(e) => updateRef(selectedIdxInWf, { max_retries: parseInt(e.target.value || "0", 10) })}
/>
</div>
)}
</div>
<Button variant="danger" size="sm" onClick={() => removeStep(selectedIdxInWf)}>
Remove from workflow
</Button>
</div>
)}
</aside>
</div>
<EditWorkflowModal open={editWorkflowOpen} workflow={wf} onSaved={(w) => setWf(w)} onClose={() => setEditWorkflowOpen(false)} />
<EditStepModal
key={editingStep?.step_id ?? "new"}
open={editStepOpen}
step={editingStep}
onClose={() => {
setEditStepOpen(false);
queryClient.invalidateQueries({ queryKey: ["steps"] });
}}
/>
</>
);
}
function LibraryCard({ step, onAdd, onEdit }: { step: WorkflowStep; onAdd: () => void; onEdit: () => void }) {
return (
<div
draggable
onDragStart={(e) => {
e.dataTransfer.setData("text/plain", JSON.stringify({ kind: "lib", stepId: step.step_id }));
}}
onClick={onAdd}
className="group relative mb-2 cursor-grab rounded-lg border border-border bg-surface-2 p-2 text-left hover:border-signal/50"
>
<div className="mb-1 flex items-center gap-2">
<span className="text-text-secondary"></span>
<ShellBadge interpreter={step.interpreter} />
<span className="text-sm font-medium text-text-primary">{step.name}</span>
<button
onClick={(e) => {
e.stopPropagation();
onEdit();
}}
className="ml-auto hidden text-text-secondary hover:text-text-primary group-hover:block"
title="Edit step"
>
</button>
</div>
{step.description && <p className="text-xs text-text-secondary">{step.description}</p>}
</div>
);
}
@@ -0,0 +1,102 @@
"use client";
import { useParams } from "next/navigation";
import { useQuery, useQueryClient } from "@tanstack/react-query";
import { api, ServerRun, StepRun } from "@/lib/api";
import { Button, Badge, Card } from "@/components/ui";
type BadgeVariant = "success" | "warning" | "danger" | "neutral" | "accent";
const statusVariant: Record<string, BadgeVariant> = {
success: "success",
failed: "danger",
running: "accent",
queued: "neutral",
skipped: "neutral",
cancelled: "warning",
};
function StatusBadge({ status }: { status: string }) {
return <Badge variant={statusVariant[status] ?? "neutral"}>{status}</Badge>;
}
export default function RunDetail() {
const { runId } = useParams<{ runId: string }>();
const queryClient = useQueryClient();
const { data: run, isLoading } = useQuery({
queryKey: ["run", runId],
queryFn: () => api.getRun(runId),
refetchInterval: (query) => (query.state.data?.status === "running" ? 2000 : false),
});
const cancel = async () => {
await api.cancelRun(runId);
queryClient.invalidateQueries({ queryKey: ["run", runId] });
};
if (isLoading || !run) {
return <div className="p-8 text-text-secondary">Loading</div>;
}
return (
<div className="p-8">
<div className="mb-6 flex items-center justify-between">
<div>
<h1 className="text-2xl font-bold text-text-primary">{run.name}</h1>
<p className="mt-1 flex items-center gap-2 text-sm text-text-secondary">
<span>Run {run.run_id.slice(0, 8)}</span>
<StatusBadge status={run.status} />
</p>
</div>
{run.status === "running" && (
<Button variant="danger" onClick={cancel}>
Cancel
</Button>
)}
</div>
<div className="grid gap-4 md:grid-cols-2 lg:grid-cols-3">
{run.server_runs.map((sr: ServerRun) => (
<Card key={sr.server_id}>
<div className="mb-3 flex items-center justify-between">
<span className="font-medium text-text-primary">{sr.hostname}</span>
<StatusBadge status={sr.status} />
</div>
<div className="space-y-2">
{sr.steps.map((st: StepRun) => (
<details
key={st.order}
className="rounded-lg border border-border bg-surface-2 p-2"
>
<summary className="flex cursor-pointer items-center justify-between gap-2">
<span className="text-sm text-text-primary">{st.name}</span>
<span className="flex items-center gap-2">
<span className="text-xs text-text-secondary">
attempts: {st.attempts}
{st.status === "failed" ? ` · exit ${st.exit_code}` : ""}
</span>
<StatusBadge status={st.status} />
</span>
</summary>
{(st.stdout || st.stderr) && (
<pre className="mt-2 max-h-64 overflow-auto rounded bg-black/40 p-2 font-mono text-xs text-text-secondary">
{st.stdout}
{st.stderr ? `\n${st.stderr}` : ""}
</pre>
)}
</details>
))}
{sr.steps.length === 0 && (
<p className="text-xs text-text-secondary">No steps yet.</p>
)}
</div>
</Card>
))}
{run.server_runs.length === 0 && (
<p className="text-text-secondary">No servers targeted by this run.</p>
)}
</div>
</div>
);
}
+78
View File
@@ -0,0 +1,78 @@
"use client";
import Link from "next/link";
import { useParams } from "next/navigation";
import { useQuery } from "@tanstack/react-query";
import { api, WorkflowRun } from "@/lib/api";
import { Card, Table, Thead, Tbody, Tr, Th, Td, Badge } from "@/components/ui";
type BadgeVariant = "success" | "warning" | "danger" | "neutral" | "accent";
const statusVariant: Record<string, BadgeVariant> = {
success: "success",
failed: "danger",
running: "warning",
cancelled: "neutral",
queued: "neutral",
};
export default function WorkflowRunsPage() {
const { id } = useParams<{ id: string }>();
const { data: wf } = useQuery({ queryKey: ["workflow", id], queryFn: () => api.getWorkflow(id) });
const { data: runs, isLoading, error } = useQuery({ queryKey: ["runs", id], queryFn: () => api.listRuns(id) });
return (
<div className="p-8">
<div className="mb-6">
<Link href={`/workflows/${id}`} className="text-sm text-text-secondary hover:text-text-primary">
Back to builder
</Link>
<h1 className="mt-2 text-2xl font-bold text-text-primary">Runs · {wf?.name ?? ""}</h1>
</div>
<Card padding={false}>
{isLoading ? (
<div className="flex items-center justify-center py-20">
<div className="h-8 w-8 animate-spin rounded-full border-2 border-border border-t-accent" />
</div>
) : error ? (
<div className="py-20 text-center text-danger">Failed to load runs. Is the backend running?</div>
) : runs && runs.length > 0 ? (
<Table>
<Thead>
<Tr>
<Th>Run</Th>
<Th>Status</Th>
<Th>Started</Th>
<Th>By</Th>
<Th>Servers</Th>
</Tr>
</Thead>
<Tbody>
{runs.map((r: WorkflowRun) => (
<Tr key={r.run_id}>
<Td>
<Link
href={`/workflows/${id}/runs/${r.run_id}`}
className="font-mono text-text-primary hover:text-signal"
>
{r.run_id.slice(0, 8)}
</Link>
</Td>
<Td>
<Badge variant={statusVariant[r.status] ?? "neutral"}>{r.status}</Badge>
</Td>
<Td className="text-text-secondary">{new Date(r.started_at).toLocaleString()}</Td>
<Td className="text-text-secondary">{r.triggered_by}</Td>
<Td className="text-text-secondary">{r.server_runs.length}</Td>
</Tr>
))}
</Tbody>
</Table>
) : (
<div className="py-16 text-center text-text-secondary">No runs yet.</div>
)}
</Card>
</div>
);
}
+114
View File
@@ -0,0 +1,114 @@
"use client";
import { useState } from "react";
import Link from "next/link";
import { useRouter } from "next/navigation";
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 router = useRouter();
const [error, setError] = useState<string | null>(null);
const { data: workflows, isLoading, error: loadError } = useQuery({
queryKey: ["workflows"],
queryFn: api.listWorkflows,
});
const { mutate: create, isPending } = useMutation({
mutationFn: () => api.createWorkflow({ name: "Untitled workflow", target_server_ids: [], steps: [] }),
onSuccess: (workflow) => {
qc.invalidateQueries({ queryKey: ["workflows"] });
router.push(`/workflows/${workflow.workflow_id}`);
},
onError: (err) => setError((err as Error).message),
});
return (
<div className="p-8">
<div className="mb-6 flex items-center justify-between">
<div>
<h1 className="text-2xl font-bold text-text-primary">Workflows</h1>
<p className="mt-1 text-sm text-text-secondary">
{workflows?.length ?? 0} workflow{workflows?.length !== 1 ? "s" : ""} · run reusable steps across servers
</p>
</div>
<Button variant="primary" loading={isPending} onClick={() => create()}>
<svg className="h-4 w-4" fill="none" viewBox="0 0 24 24" stroke="currentColor" strokeWidth={2}>
<path strokeLinecap="round" strokeLinejoin="round" d="M12 4.5v15m7.5-7.5h-15" />
</svg>
New Workflow
</Button>
</div>
{error && (
<div className="mb-4 rounded-lg border border-danger/30 bg-danger/10 px-3 py-2 text-sm text-danger">
{error}
</div>
)}
<Card padding={false}>
{isLoading ? (
<div className="flex items-center justify-center py-20">
<div className="h-8 w-8 animate-spin rounded-full border-2 border-border border-t-accent" />
</div>
) : loadError ? (
<div className="py-20 text-center text-danger">Failed to load workflows. Is the backend running?</div>
) : workflows && workflows.length > 0 ? (
<Table>
<Thead>
<Tr>
<Th>Name</Th>
<Th>Targets</Th>
<Th>Steps</Th>
<Th />
</Tr>
</Thead>
<Tbody>
{workflows.map((w: Workflow) => (
<Tr key={w.workflow_id}>
<Td>
<span className="font-medium text-text-primary">{w.name}</span>
</Td>
<Td>
<span className="text-text-secondary">
{w.target_server_ids.length} server{w.target_server_ids.length !== 1 ? "s" : ""}
</span>
</Td>
<Td>
<span className="text-text-secondary">{w.steps.length}</span>
</Td>
<Td>
<div className="flex items-center justify-end gap-2">
<Link href={`/workflows/${w.workflow_id}/runs`}>
<Button variant="ghost" size="sm">Runs</Button>
</Link>
<Link href={`/workflows/${w.workflow_id}`}>
<Button variant="ghost" size="sm">Open </Button>
</Link>
</div>
</Td>
</Tr>
))}
</Tbody>
</Table>
) : (
<div className="py-20 text-center">
<div className="mx-auto mb-4 flex h-12 w-12 items-center justify-center rounded-full bg-surface-2">
<svg className="h-6 w-6 text-text-secondary" fill="none" viewBox="0 0 24 24" stroke="currentColor" strokeWidth={1.5}>
<path strokeLinecap="round" strokeLinejoin="round" d="M9.594 3.94c.09-.542.56-.94 1.11-.94h2.593c.55 0 1.02.398 1.11.94l.213 1.281M15 12a3 3 0 11-6 0 3 3 0 016 0z" />
</svg>
</div>
<p className="text-text-secondary">No workflows yet.</p>
<Button variant="primary" size="sm" className="mt-4" loading={isPending} onClick={() => create()}>
Create your first workflow
</Button>
</div>
)}
</Card>
</div>
);
}
+9
View File
@@ -35,6 +35,14 @@ function SecretIcon() {
);
}
function WorkflowIcon() {
return (
<svg className="h-5 w-5" fill="none" viewBox="0 0 24 24" stroke="currentColor" strokeWidth={1.5}>
<path strokeLinecap="round" strokeLinejoin="round" d="M9 3.75H6.912a2.25 2.25 0 00-2.15 1.588L2.35 13.177a2.25 2.25 0 00-.1.661V18a2.25 2.25 0 002.25 2.25h15A2.25 2.25 0 0021.75 18v-4.162c0-.224-.034-.447-.1-.661L19.24 5.338a2.25 2.25 0 00-2.15-1.588H15M2.25 13.5h3.86a2.25 2.25 0 012.012 1.244l.256.512a2.25 2.25 0 002.012 1.244h3.22a2.25 2.25 0 002.012-1.244l.256-.512a2.25 2.25 0 012.012-1.244h3.86M12 3v8.25m0 0l-3-3m3 3l3-3" />
</svg>
);
}
function AuditIcon() {
return (
<svg className="h-5 w-5" fill="none" viewBox="0 0 24 24" stroke="currentColor" strokeWidth={1.5}>
@@ -56,6 +64,7 @@ const navItems: NavItem[] = [
{ href: "/servers", label: "Servers", icon: <ServerIcon /> },
{ href: "/keys", label: "SSH Keys", icon: <KeyIcon /> },
{ href: "/secrets", label: "Secrets", icon: <SecretIcon /> },
{ href: "/workflows", label: "Workflows", icon: <WorkflowIcon /> },
{ href: "/audit", label: "Audit Log", icon: <AuditIcon /> },
{ href: "/settings", label: "Settings", icon: <SettingsIcon /> },
];
+45
View File
@@ -0,0 +1,45 @@
"use client";
import { useEffect } from "react";
export function Modal({
open,
title,
onClose,
children,
wide,
}: {
open: boolean;
title: string;
onClose: () => void;
children: React.ReactNode;
wide?: boolean;
}) {
useEffect(() => {
if (!open) return;
const onKey = (e: KeyboardEvent) => e.key === "Escape" && onClose();
window.addEventListener("keydown", onKey);
return () => window.removeEventListener("keydown", onKey);
}, [open, onClose]);
if (!open) return null;
return (
<div className="fixed inset-0 z-50 flex items-center justify-center p-4">
<div className="absolute inset-0 bg-black/60" onClick={onClose} />
<div
className={`relative z-10 w-full ${wide ? "max-w-2xl" : "max-w-md"} max-h-[90vh] overflow-auto rounded-xl border border-border bg-surface shadow-2xl`}
role="dialog"
aria-modal="true"
>
<div className="flex items-center justify-between border-b border-border px-5 py-3">
<h2 className="text-sm font-bold text-text-primary">{title}</h2>
<button onClick={onClose} className="text-text-secondary hover:text-text-primary" aria-label="Close">
</button>
</div>
<div className="p-5">{children}</div>
</div>
</div>
);
}
+1
View File
@@ -2,3 +2,4 @@ export { Button } from "./Button";
export { Badge } from "./Badge";
export { Card, CardHeader, CardTitle } from "./Card";
export { Table, Thead, Tbody, Tr, Th, Td } from "./Table";
export { Modal } from "./Modal";
+110
View File
@@ -0,0 +1,110 @@
"use client";
import { useState } from "react";
import { useQueryClient } from "@tanstack/react-query";
import { api, WorkflowStep, InputParam } from "@/lib/api";
import { Button, Modal } from "@/components/ui";
const inputClass =
"w-full rounded-lg border border-border bg-surface-2 px-3 py-2 text-sm text-text-primary placeholder-text-secondary/50 focus:border-signal focus:outline-none focus:ring-1 focus:ring-signal";
export function EditStepModal({ open, step, onClose }: { open: boolean; step: WorkflowStep | null; onClose: () => void }) {
const qc = useQueryClient();
const [name, setName] = useState(step?.name ?? "");
const [interpreter, setInterpreter] = useState<"bash" | "powershell">(step?.interpreter ?? "bash");
const [script, setScript] = useState(step?.script ?? "");
const [outputs, setOutputs] = useState<string[]>(step?.declared_outputs ?? []);
const [inputs, setInputs] = useState<InputParam[]>(step?.declared_inputs ?? []);
const [newOut, setNewOut] = useState("");
const [busy, setBusy] = useState(false);
const [error, setError] = useState<string | null>(null);
// NOTE: because state is seeded from props, render the modal conditionally
// (parent mounts it only when opening) OR key it by step_id so it re-seeds.
const save = async () => {
setBusy(true); setError(null);
try {
const payload: Partial<WorkflowStep> = {
name: name.trim(), description: step?.description ?? "", interpreter, script,
declared_outputs: outputs, declared_inputs: inputs.filter((i) => i.name.trim() !== ""),
secret_refs: step?.secret_refs ?? [],
};
if (step) await api.updateStep(step.step_id, payload);
else await api.createStep(payload);
qc.invalidateQueries({ queryKey: ["steps"] });
onClose();
} catch (e) { setError((e as Error).message); } finally { setBusy(false); }
};
const del = async () => {
if (!step || !window.confirm("Delete this step? It will be removed from every workflow that uses it.")) return;
setBusy(true); setError(null);
try {
await api.deleteStep(step.step_id);
qc.invalidateQueries({ queryKey: ["steps"] });
qc.invalidateQueries({ queryKey: ["workflow"] });
onClose();
} catch (e) { setError((e as Error).message); } finally { setBusy(false); }
};
return (
<Modal open={open} onClose={onClose} title={step ? "Edit base step" : "New step"} wide>
<div className="space-y-4">
{error && <div className="rounded border border-danger/30 bg-danger/10 px-3 py-2 text-sm text-danger">{error}</div>}
<p className="text-xs text-text-secondary">Reusable steps are shared across all workflows. Editing here changes it everywhere.</p>
<div>
<label className="mb-1 block text-xs uppercase text-text-secondary">Name</label>
<input className={inputClass} value={name} onChange={(e) => setName(e.target.value)} />
</div>
<div>
<label className="mb-1 block text-xs uppercase text-text-secondary">Interpreter</label>
<select className={inputClass} value={interpreter} onChange={(e) => setInterpreter(e.target.value as "bash" | "powershell")}>
<option value="bash">bash</option>
<option value="powershell">powershell</option>
</select>
</div>
<div>
<label className="mb-1 block text-xs uppercase text-text-secondary">Script</label>
<textarea className={`${inputClass} h-40 font-mono text-xs`} value={script} onChange={(e) => setScript(e.target.value)} />
<p className="mt-1 text-xs text-text-secondary">Write <code className="text-signal">KEY=value</code> to <code className="text-signal">$WORKFLOW_ENV</code> to expose it to later steps.</p>
</div>
<div>
<label className="mb-1 block text-xs uppercase text-text-secondary">Outputs</label>
<div className="mb-2 flex flex-wrap gap-1">
{outputs.map((o) => (
<span key={o} className="flex items-center gap-1 rounded bg-signal px-2 py-0.5 font-mono text-[11px] text-signal-ink">
{o}<button onClick={() => setOutputs(outputs.filter((x) => x !== o))}></button>
</span>
))}
</div>
<div className="flex gap-2">
<input className={inputClass} placeholder="OUTPUT_NAME" value={newOut} onChange={(e) => setNewOut(e.target.value)} />
<Button variant="ghost" size="sm" onClick={() => { if (newOut.trim()) { setOutputs([...outputs, newOut.trim()]); setNewOut(""); } }}>Add</Button>
</div>
</div>
<div>
<label className="mb-1 block text-xs uppercase text-text-secondary">Inputs</label>
<div className="space-y-2">
{inputs.map((inp, i) => (
<div key={i} className="flex gap-2">
<input className={inputClass} placeholder="name" value={inp.name} onChange={(e) => setInputs(inputs.map((x, j) => j === i ? { ...x, name: e.target.value } : x))} />
<input className={inputClass} placeholder="default" value={inp.default} onChange={(e) => setInputs(inputs.map((x, j) => j === i ? { ...x, default: e.target.value } : x))} />
<input className={inputClass} placeholder="description" value={inp.description} onChange={(e) => setInputs(inputs.map((x, j) => j === i ? { ...x, description: e.target.value } : x))} />
<Button variant="ghost" size="sm" onClick={() => setInputs(inputs.filter((_, j) => j !== i))}></Button>
</div>
))}
</div>
<Button variant="ghost" size="sm" className="mt-2" onClick={() => setInputs([...inputs, { name: "", default: "", description: "" }])}>Add input</Button>
</div>
<div className="flex items-center justify-between pt-2">
{step ? <Button variant="danger" onClick={del} loading={busy}>Delete step</Button> : <span />}
<div className="flex gap-2">
<Button variant="ghost" onClick={onClose}>Cancel</Button>
<Button variant="primary" onClick={save} loading={busy} disabled={!name.trim()}>Save</Button>
</div>
</div>
</div>
</Modal>
);
}
@@ -0,0 +1,77 @@
"use client";
import { useState, useEffect } from "react";
import { useRouter } from "next/navigation";
import { useQuery } from "@tanstack/react-query";
import { api, Workflow } from "@/lib/api";
import { Button, Modal } from "@/components/ui";
const inputClass =
"w-full rounded-lg border border-border bg-surface-2 px-3 py-2 text-sm text-text-primary focus:border-signal focus:outline-none focus:ring-1 focus:ring-signal";
export function EditWorkflowModal({ open, workflow, onSaved, onClose }: { open: boolean; workflow: Workflow; onSaved: (w: Workflow) => void; onClose: () => void }) {
const router = useRouter();
const [name, setName] = useState(workflow.name);
const [targets, setTargets] = useState<string[]>(workflow.target_server_ids);
const [busy, setBusy] = useState(false);
const [error, setError] = useState<string | null>(null);
const { data: servers } = useQuery({ queryKey: ["servers"], queryFn: api.listServers });
useEffect(() => {
if (open) {
setName(workflow.name);
setTargets(workflow.target_server_ids);
}
}, [open, workflow]);
const toggle = (id: string) => setTargets((t) => (t.includes(id) ? t.filter((x) => x !== id) : [...t, id]));
const save = async () => {
setBusy(true); setError(null);
try {
const updated = await api.updateWorkflow(workflow.workflow_id, { ...workflow, name, target_server_ids: targets });
onSaved(updated); onClose();
} catch (e) { setError((e as Error).message); } finally { setBusy(false); }
};
const del = async () => {
if (!window.confirm("Delete this workflow? This cannot be undone.")) return;
setBusy(true); setError(null);
try { await api.deleteWorkflow(workflow.workflow_id); router.push("/workflows"); }
catch (e) { setError((e as Error).message); setBusy(false); }
};
return (
<Modal open={open} onClose={onClose} title="Edit workflow">
<div className="space-y-4">
{error && <div className="rounded border border-danger/30 bg-danger/10 px-3 py-2 text-sm text-danger">{error}</div>}
<div>
<label className="mb-1 block text-xs uppercase text-text-secondary">Name</label>
<input className={inputClass} value={name} onChange={(e) => setName(e.target.value)} />
</div>
<div>
<label className="mb-1 block text-xs uppercase text-text-secondary">Target servers</label>
<div className="flex flex-wrap gap-2">
{servers?.map((s) => {
const on = targets.includes(s.server_id);
return (
<label key={s.server_id} className={`flex cursor-pointer items-center gap-2 rounded-lg border px-2 py-1 text-sm ${on ? "border-signal bg-signal/10 text-text-primary" : "border-border text-text-secondary"}`}>
<input type="checkbox" className="accent-signal" checked={on} onChange={() => toggle(s.server_id)} />
{s.hostname}
</label>
);
})}
{servers && servers.length === 0 && <p className="text-xs text-text-secondary">No servers registered.</p>}
</div>
</div>
<div className="flex items-center justify-between pt-2">
<Button variant="danger" onClick={del} loading={busy}>Delete workflow</Button>
<div className="flex gap-2">
<Button variant="ghost" onClick={onClose}>Cancel</Button>
<Button variant="primary" onClick={save} loading={busy} disabled={!name.trim()}>Save</Button>
</div>
</div>
</div>
</Modal>
);
}
+136
View File
@@ -132,6 +132,73 @@ export interface ServerWithKeys extends Server {
keys: (Assignment & { key: Key })[];
}
export interface InputParam {
name: string;
default: string;
description: string;
}
export interface WorkflowStep {
step_id: string;
name: string;
description: string;
interpreter: "bash" | "powershell";
script: string;
declared_outputs: string[];
declared_inputs: InputParam[];
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[] };
inputs?: Record<string, 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<string, string>;
started_at?: string;
finished_at?: string;
}
export interface ServerRun {
server_id: string;
hostname: string;
status: string;
run_env: Record<string, string>;
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[];
}
class ApiError extends Error {
constructor(
public status: number,
@@ -324,4 +391,73 @@ export const api = {
body: JSON.stringify(body),
});
},
// Steps
listSteps(): Promise<WorkflowStep[]> {
return request<WorkflowStep[]>("/steps");
},
createStep(s: Partial<WorkflowStep>): Promise<WorkflowStep> {
return request<WorkflowStep>("/steps", {
method: "POST",
body: JSON.stringify(s),
});
},
updateStep(stepId: string, s: Partial<WorkflowStep>): Promise<WorkflowStep> {
return request<WorkflowStep>(`/steps/${stepId}`, {
method: "PUT",
body: JSON.stringify(s),
});
},
deleteStep(stepId: string): Promise<void> {
return request<void>(`/steps/${stepId}`, { method: "DELETE" });
},
// Workflows
listWorkflows(): Promise<Workflow[]> {
return request<Workflow[]>("/workflows");
},
getWorkflow(workflowId: string): Promise<Workflow> {
return request<Workflow>(`/workflows/${workflowId}`);
},
createWorkflow(w: Partial<Workflow>): Promise<Workflow> {
return request<Workflow>("/workflows", {
method: "POST",
body: JSON.stringify(w),
});
},
updateWorkflow(workflowId: string, w: Partial<Workflow>): Promise<Workflow> {
return request<Workflow>(`/workflows/${workflowId}`, {
method: "PUT",
body: JSON.stringify(w),
});
},
deleteWorkflow(workflowId: string): Promise<void> {
return request<void>(`/workflows/${workflowId}`, { method: "DELETE" });
},
runWorkflow(workflowId: string): Promise<{ run_id: string }> {
return request<{ run_id: string }>(`/workflows/${workflowId}/run`, {
method: "POST",
});
},
listRuns(workflowId: string): Promise<WorkflowRun[]> {
return request<WorkflowRun[]>(`/workflows/${workflowId}/runs`);
},
// Runs
getRun(runId: string): Promise<WorkflowRun> {
return request<WorkflowRun>(`/runs/${runId}`);
},
cancelRun(runId: string): Promise<void> {
return request<void>(`/runs/${runId}/cancel`, { method: "POST" });
},
};
+48 -6
View File
@@ -2,7 +2,11 @@
// The library attaches a global `Guacamole` object when loaded.
declare const Guacamole: any;
export function openConsole(container: HTMLElement, wsUrl: string, connectData = ""): { disconnect: () => void } {
export function openConsole(
container: HTMLElement,
wsUrl: string,
connectData = ""
): { disconnect: () => void; setScale: (scale: number) => void; resize: (width: number, height: number) => void } {
// Guacamole's WebSocketTunnel builds the socket URL as `wsUrl + "?" + data`,
// so wsUrl must NOT already contain a query string — pass params via connectData.
const tunnel = new Guacamole.WebSocketTunnel(wsUrl);
@@ -10,20 +14,58 @@ export function openConsole(container: HTMLElement, wsUrl: string, connectData =
container.innerHTML = "";
container.appendChild(client.getDisplay().getElement());
// Make the console focusable so keyboard capture is scoped to it (see below).
container.tabIndex = 0;
client.connect(connectData);
// Wire keyboard + mouse.
const mouse = new Guacamole.Mouse(client.getDisplay().getElement());
mouse.onmousedown = mouse.onmouseup = mouse.onmousemove = (state: any) =>
client.sendMouseState(state);
const keyboard = new Guacamole.Keyboard(document);
const display = client.getDisplay();
let scale = 1;
// Wire keyboard + mouse. The display element is rendered at `scale` of the
// remote's native resolution, but Guacamole.Mouse reports coordinates in
// element (on-screen) pixels. Divide by scale to map back to remote
// coordinates, otherwise the cursor is offset.
const mouse = new Guacamole.Mouse(display.getElement());
mouse.onmousedown = mouse.onmouseup = mouse.onmousemove = (state: any) => {
const s = new Guacamole.Mouse.State(
state.x / scale,
state.y / scale,
state.left,
state.middle,
state.right,
state.up,
state.down
);
client.sendMouseState(s);
};
// Scope keyboard capture to the container rather than `document`, so it only
// grabs keys while the console is focused and stops entirely once the element
// is removed (navigating away / disconnect). Attaching to `document` leaks the
// capture and swallows keystrokes in unrelated inputs.
const keyboard = new Guacamole.Keyboard(container);
keyboard.onkeydown = (k: number) => client.sendKeyEvent(1, k);
keyboard.onkeyup = (k: number) => client.sendKeyEvent(0, k);
// Guacamole.Mouse consumes the native mousedown, so clicking the console never
// moves DOM focus back to it. Refocus explicitly so keyboard capture resumes.
const refocus = () => container.focus();
container.addEventListener("mousedown", refocus);
container.focus();
return {
disconnect() {
container.removeEventListener("mousedown", refocus);
keyboard.onkeydown = null;
keyboard.onkeyup = null;
if (typeof keyboard.reset === "function") keyboard.reset();
client.disconnect();
},
setScale(s: number) {
scale = s;
display.scale(s);
},
resize(width: number, height: number) {
client.sendSize(width, height);
},
};
}
+4
View File
@@ -21,6 +21,10 @@ const config: Config = {
warning: "#f59e0b",
danger: "#ef4444",
"danger-hover": "#dc2626",
bash: "#3fb950",
pwsh: "#5b9bff",
signal: "#f5a524",
"signal-ink": "#241800",
},
},
},