feat: More verbose logging on workflow logs

server/internal/grpc/pb/vantage.pb.go

@@ -66,12 +66,19 @@ type ReportUpdatesResponse struct{}
 type ApplyUpdatesCmd struct{}
 
 type ServerCommand struct {
-	CommandId    string           `json:"command_id"`
-	GenerateKey  *GenerateKeyCmd  `json:"generate_key,omitempty"`
-	DeleteKey    *DeleteKeyCmd    `json:"delete_key,omitempty"`
-	UpdateAgent  *UpdateAgentCmd  `json:"update_agent,omitempty"`
-	ApplyUpdates *ApplyUpdatesCmd `json:"apply_updates,omitempty"`
-	RunStep      *RunStepCmd      `json:"run_step,omitempty"`
+	CommandId        string               `json:"command_id"`
+	GenerateKey      *GenerateKeyCmd      `json:"generate_key,omitempty"`
+	DeleteKey        *DeleteKeyCmd        `json:"delete_key,omitempty"`
+	UpdateAgent      *UpdateAgentCmd      `json:"update_agent,omitempty"`
+	ApplyUpdates     *ApplyUpdatesCmd     `json:"apply_updates,omitempty"`
+	RunStep          *RunStepCmd          `json:"run_step,omitempty"`
+	CleanupWorkspace *CleanupWorkspaceCmd `json:"cleanup_workspace,omitempty"`
+}
+
+// CleanupWorkspaceCmd tells the agent to recursively remove the run's working
+// directory once all steps on that server have finished.
+type CleanupWorkspaceCmd struct {
+	WorkspaceId string `json:"workspace_id"`
 }
 
 type DeleteKeyCmd struct {
@@ -113,6 +120,9 @@ type RunStepCmd struct {
 	Script         string            `json:"script"`
 	Env            map[string]string `json:"env,omitempty"`
 	TimeoutSeconds int               `json:"timeout_seconds,omitempty"`
+	// WorkspaceId names the per-run working directory the agent creates and uses
+	// as the step's cwd. Empty means run in the agent's default directory.
+	WorkspaceId string `json:"workspace_id,omitempty"`
 }
 
 type StepResult struct {

server/internal/services/dispatch.go

@@ -68,6 +68,19 @@ func DispatchRunStep(serverID, commandID string, cmd *pb.RunStepCmd) error {
 	return Dispatcher.dispatch(serverID, &pb.ServerCommand{CommandId: commandID, RunStep: cmd})
 }
 
+// DispatchCleanupWorkspace tells a server's agent to remove a run's working
+// directory. Best-effort and fire-and-forget: if the agent is gone the temp dir
+// is reclaimed by the OS on reboot anyway.
+func DispatchCleanupWorkspace(serverID, workspaceID string) {
+	if !Dispatcher.IsConnected(serverID) {
+		return
+	}
+	_ = Dispatcher.dispatch(serverID, &pb.ServerCommand{
+		CommandId:        uuid.New().String(),
+		CleanupWorkspace: &pb.CleanupWorkspaceCmd{WorkspaceId: workspaceID},
+	})
+}
+
 // KeyGenParams carries all options for a generate-key command.
 type KeyGenParams struct {
 	Label      string

server/internal/services/steplogs.go

@@ -1,6 +1,7 @@
 package services
 
 import (
+	"bytes"
 	"os"
 	"path/filepath"
 	"strings"
@@ -26,9 +27,15 @@ 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) {
+// logTS is the UTC timestamp prefix stamped on every log line. Stored in UTC
+// (RFC3339, millisecond precision); the UI renders it in the viewer's timezone.
+func logTS() string {
+	return time.Now().UTC().Format("2006-01-02T15:04:05.000") + "Z"
+}
+
+// AppendMarker writes a timestamped event 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, text string) (int64, error) {
 	path := ServerRunLogPath(runID, serverID)
 	if err := os.MkdirAll(filepath.Dir(path), 0700); err != nil {
 		return 0, err
@@ -39,7 +46,7 @@ func AppendMarker(runID, serverID, line string) (int64, error) {
 	}
 	defer f.Close()
 	off, _ := f.Seek(0, 2) // current end = offset before write
-	if _, err := f.WriteString(line); err != nil {
+	if _, err := f.WriteString("[" + logTS() + "] " + text + "\n"); err != nil {
 		return off, err
 	}
 	return off, nil
@@ -48,11 +55,10 @@ func AppendMarker(runID, serverID, line string) (int64, error) {
 // ---- streamed chunk writer, boundary-safe secret masking ----
 
 type stepLogWriter struct {
-	mu        sync.Mutex
-	f         *os.File
-	carry     []byte
-	secrets   []string
-	maxSecret int
+	mu      sync.Mutex
+	f       *os.File
+	carry   []byte // bytes of an as-yet-unterminated line
+	secrets []string
 }
 
 type stepLogRegistry struct {
@@ -71,13 +77,7 @@ func (r *stepLogRegistry) Open(commandID, path string, secrets []string) error {
 	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}
+	w := &stepLogWriter{f: f, secrets: secrets}
 	r.mu.Lock()
 	r.writers[commandID] = w
 	r.mu.Unlock()
@@ -90,8 +90,10 @@ func (r *stepLogRegistry) get(commandID string) *stepLogWriter {
 	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.
+// Append buffers chunks into whole lines, then writes each complete line with a
+// UTC timestamp prefix and secret masking applied. Buffering by line means a
+// secret split across a chunk boundary is always masked (the whole line is
+// assembled first) and every line carries its own timestamp.
 func (r *stepLogRegistry) Append(commandID string, data []byte) {
 	w := r.get(commandID)
 	if w == nil {
@@ -99,22 +101,27 @@ func (r *stepLogRegistry) Append(commandID string, data []byte) {
 	}
 	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
+	for {
+		i := bytes.IndexByte(buf, '\n')
+		if i < 0 {
+			break
+		}
+		w.writeLine(buf[:i])
+		buf = buf[i+1:]
 	}
-	flush := buf[:len(buf)-hold]
-	w.carry = append([]byte{}, buf[len(buf)-hold:]...)
-	_, _ = w.f.Write(maskBytes(flush, w.secrets))
+	w.carry = append([]byte{}, buf...)
 }
 
-// Close flushes the carry (masked) and closes the file.
+// writeLine emits one masked, timestamped log line. Caller holds w.mu.
+func (w *stepLogWriter) writeLine(line []byte) {
+	masked := maskBytes(line, w.secrets)
+	_, _ = w.f.WriteString("[" + logTS() + "] ")
+	_, _ = w.f.Write(masked)
+	_, _ = w.f.WriteString("\n")
+}
+
+// Close flushes any trailing partial line and closes the file.
 func (r *stepLogRegistry) Close(commandID string) {
 	r.mu.Lock()
 	w := r.writers[commandID]
@@ -126,7 +133,7 @@ func (r *stepLogRegistry) Close(commandID string) {
 	w.mu.Lock()
 	defer w.mu.Unlock()
 	if len(w.carry) > 0 {
-		_, _ = w.f.Write(maskBytes(w.carry, w.secrets))
+		w.writeLine(w.carry)
 		w.carry = nil
 	}
 	_ = w.f.Close()

server/internal/services/workflow_runner.go

@@ -162,16 +162,20 @@ func runServer(runID string, srvIdx int, steps []models.ResolvedStep, serverID s
 
 	if !Dispatcher.IsConnected(serverID) {
 		fin := time.Now()
+		_, _ = AppendMarker(runID, serverID, "agent not connected — server skipped")
 		setServerRun(runID, srvIdx, bson.M{"server_runs.$.status": "skipped", "server_runs.$.finished_at": fin})
 		return
 	}
 
+	_, _ = AppendMarker(runID, serverID, fmt.Sprintf("run started on %s — %d step(s), workspace vantage-run-%s", serverID, len(steps), runID))
+
 	runEnv := map[string]string{}
 	allSecrets := map[string]string{}
 	serverFailed := false
 
 	for i, step := range steps {
 		startStep(runID, serverID, i, "running")
+		stepStart := time.Now()
 		var res *pb.StepResult
 		attempts := 0
 		maxAttempts := 1
@@ -197,7 +201,7 @@ func runServer(runID string, srvIdx int, steps []models.ResolvedStep, serverID s
 
 		// Write the step marker to the server-run log and remember the offset so
 		// the UI can slice this step's output later.
-		marker := fmt.Sprintf("\n===== step %d: %s =====\n", step.Order, step.Name)
+		marker := fmt.Sprintf("===== step %d/%d: %s (%s) =====", step.Order+1, len(steps), step.Name, step.Interpreter)
 		offset, _ := AppendMarker(runID, serverID, marker)
 		logPath := ServerRunLogPath(runID, serverID)
 		secretsSlice := secretValues(secretVals)
@@ -205,6 +209,9 @@ func runServer(runID string, srvIdx int, steps []models.ResolvedStep, serverID s
 		commandID := uuid.New().String()
 		for attempts < maxAttempts {
 			attempts++
+			if attempts > 1 {
+				_, _ = AppendMarker(runID, serverID, fmt.Sprintf("retry %d/%d after failure", attempts-1, maxAttempts-1))
+			}
 			// Open a fresh writer per attempt; the agent's eof closes it, and the
 			// defensive Close below covers a missing result.
 			_ = StepLogs.Open(commandID, logPath, secretsSlice)
@@ -213,6 +220,7 @@ func runServer(runID string, srvIdx int, steps []models.ResolvedStep, serverID s
 				Script:         step.Script,
 				Env:            cmdEnv,
 				TimeoutSeconds: 0,
+				WorkspaceId:    runID,
 			})
 			StepLogs.Close(commandID) // idempotent; no-op if eof already closed it
 			if res != nil && res.ExitCode == 0 {
@@ -229,7 +237,7 @@ func runServer(runID string, srvIdx int, steps []models.ResolvedStep, serverID s
 				outEnv[k] = maskSecrets(v, allSecrets)
 			}
 		} else {
-			_, _ = AppendMarker(runID, serverID, "[vantage] agent did not return a result\n")
+			_, _ = AppendMarker(runID, serverID, "agent did not return a result")
 		}
 
 		status := "success"
@@ -238,25 +246,36 @@ func runServer(runID string, srvIdx int, steps []models.ResolvedStep, serverID s
 		}
 		finishStep(runID, serverID, i, status, attempts, exit, offset, outEnv)
 
+		dur := time.Since(stepStart).Round(time.Millisecond)
+		_, _ = AppendMarker(runID, serverID, fmt.Sprintf("step %d/%d %s — exit %d, %d attempt(s), %s",
+			step.Order+1, len(steps), status, exit, attempts, dur))
+
 		if exit != 0 {
 			switch step.OnFailure {
 			case "continue":
-				// keep going
+				_, _ = AppendMarker(runID, serverID, "on_failure=continue — proceeding to next step")
 			default: // "stop" or exhausted "retry"
 				serverFailed = true
 			}
 			if serverFailed {
+				_, _ = AppendMarker(runID, serverID, "stopping run — remaining steps skipped")
 				markRemainingSkipped(runID, serverID, i+1)
 				break
 			}
 		}
 	}
 
+	// Tell the agent to remove the run's working directory now that its steps are
+	// done (success or failure). Best-effort; the OS reclaims temp dirs anyway.
+	DispatchCleanupWorkspace(serverID, runID)
+
 	fin := time.Now()
 	status := "success"
 	if serverFailed {
 		status = "failed"
 	}
+	_, _ = AppendMarker(runID, serverID, fmt.Sprintf("run %s in %s — workspace removed",
+		status, fin.Sub(now).Round(time.Millisecond)))
 	// 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))