fix(workflows): mask output env in persisted logs, preserve cancelled status, run agent step async

agent/internal/sync/sync.go

@@ -14,6 +14,7 @@ import (
 	"path/filepath"
 	"runtime"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/mrhid6/vantage/agent/internal/config"
@@ -169,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 {
@@ -188,13 +199,15 @@ func connectAndHandleStream(ctx context.Context, cfg *config.Config) error {
 			go handleApplyUpdates(cfg, cmd)
 		}
 		if cmd.RunStep != nil {
-			res := agentexec.RunStep(cmd.RunStep)
-			res.CommandId = cmd.CommandId
-			_ = stream.Send(&pb.AgentMessage{
-				ServerId:   cfg.ServerID,
-				AgentToken: cfg.AgentToken,
-				StepResult: res,
-			})
+			go func(rc *pb.RunStepCmd, cid string) {
+				res := agentexec.RunStep(rc)
+				res.CommandId = cid
+				_ = send(&pb.AgentMessage{
+					ServerId:   cfg.ServerID,
+					AgentToken: cfg.AgentToken,
+					StepResult: res,
+				})
+			}(cmd.RunStep, cmd.CommandId)
 			continue
 		}
 	}