1
import {
2
  now,
3
  toPositiveInteger,
4
  type PreparedExecutionEnvironment,
5
  type RunExecutionContext,
6
  type RunExecutionOutcome,
7
} from "@/worker/dispatch/shared/run-execution-context";
8
import { type RunLeaseControl } from "@/worker/dispatch/shared/run-lease";
9
import { executeSessionCommandStream, resolveExecStreamProcess, type CommandStreamResult } from "./command-stream";
10
import { createLogBatcher } from "./logging";
11
 
12
type RunStepExecutionContext = Pick<RunExecutionContext, "logs" | "runStore" | "scope" | "state">;
13
 
14
const toCommandFailureMessage = (stepName: string, result: CommandStreamResult): string => {
15
  if (result.terminalEvent === "complete") {
16
    if (typeof result.exitCode === "number") {
17
      return `Step "${stepName}" failed with exit code ${result.exitCode}.`;
18
    }
19
 
20
    return `Step "${stepName}" failed.`;
21
  }
22
 
23
  return result.errorMessage || result.stderr || `Step "${stepName}" failed.`;
24
};
25
 
26
export const executeRunSteps = async (
27
  context: RunStepExecutionContext,
28
  lease: RunLeaseControl,
29
  prepared: PreparedExecutionEnvironment,
30
): Promise<RunExecutionOutcome> => {
31
  if (lease.isCancellationRequested()) {
32
    return {
33
      kind: "canceled",
34
    };
35
  }
36
 
37
  await context.runStore.updateState({
38
    status: "running",
39
    startedAt: context.scope.startedAt,
40
    currentStep: null,
41
    finishedAt: null,
42
    exitCode: null,
43
    errorMessage: null,
44
  });
45
 
46
  const deadline = context.scope.startedAt + prepared.repoConfig.run.timeoutSeconds * 1000;
47
  if (prepared.repoConfig.run.steps.length === 0) {
48
    return {
49
      kind: "passed",
50
      exitCode: 0,
51
    };
52
  }
53
 
54
  let lastExitCode: number | null = null;
55
 
56
  for (const [index, step] of prepared.repoConfig.run.steps.entries()) {
57
    context.state.phase = "running_step";
58
    lease.throwIfOwnershipLost();
59
    if (lease.isCancellationRequested()) {
60
      return {
61
        kind: "canceled",
62
      };
63
    }
64
 
65
    const position = toPositiveInteger(index + 1);
66
    const stepStartedAt = now();
67
    const remainingMs = deadline - stepStartedAt;
68
    if (remainingMs <= 0) {
69
      throw new Error("Run exceeded timeout before the next step started.");
70
    }
71
 
72
    context.state.currentStepPosition = position;
73
    await context.runStore.updateState({
74
      status: "running",
75
      startedAt: context.scope.startedAt,
76
      currentStep: position,
77
      finishedAt: null,
78
      exitCode: null,
79
      errorMessage: null,
80
    });
81
    await context.runStore.updateStepState({
82
      position,
83
      status: "running",
84
      startedAt: stepStartedAt,
85
      finishedAt: null,
86
      exitCode: null,
87
    });
88
 
89
    const executionSession = context.state.session;
90
    if (!executionSession) {
91
      throw new Error(`Run ${context.scope.runId} lost its execution session.`);
92
    }
93
 
94
    const batcher = createLogBatcher(context);
95
 
96
    await lease.applyCancellationIfNeeded();
97
    lease.throwIfOwnershipLost();
98
 
99
    let commandResult: CommandStreamResult;
100
    try {
101
      commandResult = await executeSessionCommandStream(
102
        executionSession,
103
        step.run,
104
        {
105
          cwd: prepared.workingDirectory,
106
          timeout: remainingMs,
107
        },
108
        batcher,
109
        {
110
          onStart:
111
            "listProcesses" in executionSession
112
              ? async (event) => {
113
                  context.state.currentProcess = await resolveExecStreamProcess(executionSession, step.run, event.pid);
114
                }
115
              : undefined,
116
        },
117
      );
118
    } finally {
119
      context.state.currentProcess = null;
120
    }
121
    const stepFinishedAt = now();
122
    lease.throwIfOwnershipLost();
123
 
124
    if (lease.isCancellationRequested()) {
125
      await context.runStore.updateStepState({
126
        position,
127
        status: "failed",
128
        startedAt: stepStartedAt,
129
        finishedAt: stepFinishedAt,
130
        exitCode: commandResult.exitCode,
131
      });
132
      return {
133
        kind: "canceled",
134
      };
135
    }
136
 
137
    if (commandResult.terminalEvent !== "complete" || commandResult.exitCode !== 0) {
138
      await context.runStore.updateStepState({
139
        position,
140
        status: "failed",
141
        startedAt: stepStartedAt,
142
        finishedAt: stepFinishedAt,
143
        exitCode: commandResult.exitCode,
144
      });
145
      return {
146
        kind: "failed",
147
        exitCode: commandResult.exitCode,
148
        errorMessage: context.logs.redactMessage(toCommandFailureMessage(step.name, commandResult)),
149
      };
150
    }
151
 
152
    await context.runStore.updateStepState({
153
      position,
154
      status: "passed",
155
      startedAt: stepStartedAt,
156
      finishedAt: stepFinishedAt,
157
      exitCode: commandResult.exitCode,
158
    });
159
    lastExitCode = commandResult.exitCode;
160
  }
161
 
162
  return {
163
    kind: "passed",
164
    exitCode: lastExitCode,
165
  };
166
};