feat(server): stream step logs to files, drop log bodies from run docs

server/internal/services/workflow_runner.go

@@ -195,40 +195,48 @@ func runServer(runID string, srvIdx int, steps []models.ResolvedStep, serverID s
 			cmdEnv[k] = v
 		}
 
+		// 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)
+		offset, _ := AppendMarker(runID, serverID, marker)
+		logPath := ServerRunLogPath(runID, serverID)
+		secretsSlice := secretValues(secretVals)
+
+		commandID := uuid.New().String()
 		for attempts < maxAttempts {
 			attempts++
-			res = dispatchAndWait(serverID, &pb.RunStepCmd{
+			// 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)
+			res = dispatchAndWait(serverID, commandID, &pb.RunStepCmd{
 				Interpreter:    step.Interpreter,
 				Script:         step.Script,
 				Env:            cmdEnv,
 				TimeoutSeconds: 0,
 			})
+			StepLogs.Close(commandID) // idempotent; no-op if eof already closed it
 			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
+		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"
+			_, _ = AppendMarker(runID, serverID, "[vantage] agent did not return a result\n")
 		}
 
 		status := "success"
 		if exit != 0 {
 			status = "failed"
 		}
-		finishStep(runID, serverID, i, status, attempts, exit, stdout, stderr, outEnv)
+		finishStep(runID, serverID, i, status, attempts, exit, offset, outEnv)
 
 		if exit != 0 {
 			switch step.OnFailure {
@@ -264,8 +272,7 @@ func runServer(runID string, srvIdx int, steps []models.ResolvedStep, serverID s
 
 // 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()
+func dispatchAndWait(serverID, commandID string, cmd *pb.RunStepCmd) *pb.StepResult {
 	ch := StepResults.Await(commandID)
 	if err := DispatchRunStep(serverID, commandID, cmd); err != nil {
 		StepResults.Cancel(commandID)
@@ -336,19 +343,27 @@ func startStep(runID, serverID string, order int, status string) {
 	})
 }
 
-func finishStep(runID, serverID string, order int, status string, attempts, exit int, stdout, stderr string, outEnv map[string]string) {
+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].stdout":      stdout,
-		"server_runs.$[s].steps.$[t].stderr":      stderr,
+		"server_runs.$[s].steps.$[t].log_offset":  logOffset,
 		"server_runs.$[s].steps.$[t].output_env":  outEnv,
 		"server_runs.$[s].steps.$[t].finished_at": now,
 	})
 }
 
+// secretValues returns just the values of a secret map, for masking log output.
+func secretValues(m map[string]string) []string {
+	out := make([]string, 0, len(m))
+	for _, v := range m {
+		out = append(out, v)
+	}
+	return out
+}
+
 func markRemainingSkipped(runID, serverID string, fromOrder int) {
 	ctx, cancel := wfCtx()
 	defer cancel()