Add workflow status API for direct linkspan monitoring

Linkspan now exposes GET /api/v1/status with real-time workflow state
(running/complete/failed), current step, and all captured outputs.
GET /api/v1/health provides a simple liveness check.

The workflow Engine tracks execution state with mutex-protected Status
struct. CS-Bridge polls /api/v1/status through the tunnel once the
tunnel URL is known, falling back to SSH log parsing when unreachable.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: yasithdev

internal/workflow/workflow.go

@@ -6,6 +6,7 @@ import (
 	"io"
 	"log"
 	"os"
+	"sync"
 	"text/template"
 
 	"gopkg.in/yaml.v3"
@@ -57,18 +58,60 @@ func parse(data []byte) (*WorkflowConfig, error) {
 	return &cfg, nil
 }
 
+// WorkflowState represents the current execution status.
+type WorkflowState string
+
+const (
+	StateIdle     WorkflowState = "idle"
+	StateRunning  WorkflowState = "running"
+	StateComplete WorkflowState = "complete"
+	StateFailed   WorkflowState = "failed"
+)
+
+// Status is a snapshot of the engine's current execution state,
+// safe to serialize as JSON for the /api/v1/status endpoint.
+type Status struct {
+	State       WorkflowState  `json:"state"`
+	CurrentStep int            `json:"currentStep"`
+	TotalSteps  int            `json:"totalSteps"`
+	StepName    string         `json:"stepName,omitempty"`
+	Error       string         `json:"error,omitempty"`
+	Outputs     map[string]any `json:"outputs"`
+}
+
 // Engine executes a workflow using its registry and collected variables.
 type Engine struct {
 	Registry *Registry
 	Vars     map[string]any
+
+	mu     sync.Mutex
+	status Status
 }
 
 // NewEngine creates an Engine with the given registry and initial variables.
 func NewEngine(reg *Registry, vars map[string]any) *Engine {
 	if vars == nil {
 		vars = make(map[string]any)
 	}
-	return &Engine{Registry: reg, Vars: vars}
+	return &Engine{
+		Registry: reg,
+		Vars:     vars,
+		status:   Status{State: StateIdle, Outputs: make(map[string]any)},
+	}
+}
+
+// Status returns a snapshot of the current workflow execution state.
+func (e *Engine) Status() Status {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	// Return a copy of outputs so the caller can't mutate engine state.
+	out := make(map[string]any, len(e.status.Outputs))
+	for k, v := range e.status.Outputs {
+		out[k] = v
+	}
+	s := e.status
+	s.Outputs = out
+	return s
 }
 
 // Run executes every step in the workflow sequentially.
@@ -77,41 +120,73 @@ func NewEngine(reg *Registry, vars map[string]any) *Engine {
 func (e *Engine) Run(wf *WorkflowConfig) error {
 	log.Printf("workflow: starting %q (%d steps)", wf.Name, len(wf.Steps))
 
+	e.mu.Lock()
+	e.status = Status{State: StateRunning, TotalSteps: len(wf.Steps), Outputs: make(map[string]any)}
+	e.mu.Unlock()
+
 	for i, step := range wf.Steps {
 		log.Printf("workflow: [%d/%d] %s (action=%s)", i+1, len(wf.Steps), step.Name, step.Action)
 
+		e.mu.Lock()
+		e.status.CurrentStep = i + 1
+		e.status.StepName = step.Name
+		e.mu.Unlock()
+
 		fn := e.Registry.Get(step.Action)
 		if fn == nil {
-			return fmt.Errorf("workflow step %d: unknown action %q", i+1, step.Action)
+			err := fmt.Errorf("workflow step %d: unknown action %q", i+1, step.Action)
+			e.mu.Lock()
+			e.status.State = StateFailed
+			e.status.Error = err.Error()
+			e.mu.Unlock()
+			return err
 		}
 
 		// Resolve {{.var}} templates in string params.
 		resolved, err := e.resolveParams(step.Params)
 		if err != nil {
-			return fmt.Errorf("workflow step %d (%s): resolving params: %w", i+1, step.Name, err)
+			err = fmt.Errorf("workflow step %d (%s): resolving params: %w", i+1, step.Name, err)
+			e.mu.Lock()
+			e.status.State = StateFailed
+			e.status.Error = err.Error()
+			e.mu.Unlock()
+			return err
 		}
 
 		result, err := fn(resolved)
 		if err != nil {
-			return fmt.Errorf("workflow step %d (%s): %w", i+1, step.Name, err)
+			err = fmt.Errorf("workflow step %d (%s): %w", i+1, step.Name, err)
+			e.mu.Lock()
+			e.status.State = StateFailed
+			e.status.Error = err.Error()
+			e.mu.Unlock()
+			return err
 		}
 
 		// Map action outputs to workflow variables.
 		if result != nil && step.Outputs != nil {
+			e.mu.Lock()
 			for field, varName := range step.Outputs {
 				val, ok := (*result)[field]
 				if !ok {
 					log.Printf("workflow: warning: step %d output field %q not found in result", i+1, field)
 					continue
 				}
 				e.Vars[varName] = val
+				e.status.Outputs[varName] = val
 				log.Printf("workflow: captured %s = %v", varName, val)
 			}
+			e.mu.Unlock()
 		}
 
 		log.Printf("workflow: [%d/%d] %s completed", i+1, len(wf.Steps), step.Name)
 	}
 
+	e.mu.Lock()
+	e.status.State = StateComplete
+	e.status.StepName = ""
+	e.mu.Unlock()
+
 	log.Printf("workflow: %q finished successfully", wf.Name)
 	return nil
 }

main.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"context"
+	"encoding/json"
 	"flag"
 	"fmt"
 	"log"
@@ -114,6 +115,27 @@ func main() {
 	api.HandleFunc("/tunnels/frp", tunnel.CreateFrpTunnelProxy).Methods("POST")
 	api.HandleFunc("/tunnels/frp/{id}", tunnel.TerminateFrpTunnel).Methods("DELETE")
 
+	// Health and workflow status
+	api.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		w.WriteHeader(http.StatusOK)
+		fmt.Fprintf(w, `{"status":"ok"}`)
+	}).Methods("GET")
+
+	// Workflow status — set up after engine creation below.
+	var workflowEngine *workflow.Engine
+	api.HandleFunc("/status", func(w http.ResponseWriter, r *http.Request) {
+		if workflowEngine == nil {
+			w.Header().Set("Content-Type", "application/json")
+			w.WriteHeader(http.StatusOK)
+			fmt.Fprintf(w, `{"state":"idle","currentStep":0,"totalSteps":0,"outputs":{}}`)
+			return
+		}
+		w.Header().Set("Content-Type", "application/json")
+		w.WriteHeader(http.StatusOK)
+		json.NewEncoder(w).Encode(workflowEngine.Status())
+	}).Methods("GET")
+
 	// Use the configured server host and port from CLI flags.
 	// Port 0 means "let the OS pick a free port".
 	serverPort := *serverPortFlag
@@ -153,14 +175,14 @@ func main() {
 		if err != nil {
 			log.Fatalf("workflow: %v", err)
 		}
-		engine := workflow.NewEngine(workflow.DefaultRegistry(), map[string]any{
+		workflowEngine = workflow.NewEngine(workflow.DefaultRegistry(), map[string]any{
 			"Timestamp":       time.Now().Unix(),
 			"ServerPort":      serverPort,
 			"ServerHost":      serverHost,
 			"TunnelAuthToken": *tunnelAuthToken,
 		})
 		go func() {
-			if err := engine.Run(wf); err != nil {
+			if err := workflowEngine.Run(wf); err != nil {
 				log.Printf("workflow: %v", err)
 			}
 		}()