返回 DeepSeek-Reasonix
runner.go
根目录 / internal / shellrun / runner.go
1 // Package shellrun provides a shared foreground shell runner used by the model
2 // shell tool and the user !command path. It classifies exits, collects a bounded
3 // output tail, and keeps combined stdout/stderr model-visible output intact.
4 package shellrun
5
6 import (
7 "bytes"
8 "context"
9 "errors"
10 "fmt"
11 "io"
12 "os/exec"
13 "strings"
14 "sync"
15 "time"
16
17 "reasonix/internal/proc"
18 "reasonix/internal/tool"
19 )
20
21 // DefaultWaitDelay mirrors the shell tool's child-process wait grace.
22 const DefaultWaitDelay = 5 * time.Second
23
24 const (
25 // combinedOutputMaxBytes bounds the foreground output retained in memory.
26 // Tool-result truncation happens only after the process exits, so it cannot
27 // protect the host from a command that prints forever (#6473, #6528).
28 combinedOutputMaxBytes = 10 << 20
29 // Keep the final diagnostics as well as the command's opening context after
30 // the cap is crossed. Build and test failures are commonly printed last.
31 combinedOutputTailBytes = 64 << 10
32 combinedOutputTruncated = "\n\n...[shell output truncated at 10 MiB; showing the final 64 KiB]...\n\n"
33 // Live progress crosses async UI queues and append-only reducers before the
34 // final bounded result replaces it. Keep that transient path small too, or a
35 // never-ending command can still exhaust memory while Combined stays bounded.
36 progressOutputMaxBytes = 64 << 10
37 progressOutputTruncated = "\n\n...[live shell output capped at 64 KiB; final diagnostics will appear when the command exits]...\n\n"
38 )
39
40 var errForegroundTimeout = errors.New("shell foreground timeout")
41
42 // Request describes one foreground shell launch. Argv must already include the
43 // interpreter and any sandbox wrapping; Command is only for diagnostics.
44 type Request struct {
45 Argv []string
46 Dir string
47 Env []string
48 Timeout time.Duration
49 WaitDelay time.Duration
50 CommandPreview string
51 ShellKind string
52 ShellPath string
53 Source string
54 Track bool
55 PreserveWaitDelay bool
56 // Progress receives live combined output chunks (optional).
57 Progress func(chunk string)
58 // Run is optional; tests inject a process runner. When nil, proc.RunCommand.
59 Run func(ctx context.Context, cmd *exec.Cmd, opts proc.RunOptions) (*proc.TrackedCommand, error)
60 }
61
62 // Result is the structured outcome of a foreground run.
63 type Result struct {
64 Combined string
65 // OutputTail is the bounded tail of combined output, populated only when the
66 // run did not complete successfully. Stdout and stderr share one pipe so the
67 // model-visible ordering is preserved, which makes a stderr-only tail
68 // impossible; in practice the last bytes before a failure are the diagnosis.
69 OutputTail string
70 ExitCode *int
71 Started bool
72 State string
73 FailurePhase string
74 Err error
75 Tracked *proc.TrackedCommand
76 Cmd *exec.Cmd
77 }
78
79 // RunForeground starts the process, captures combined stdout/stderr with a
80 // lock-safe collector, and classifies timeout / cancel / launch / execution
81 // failures. Combined output is always returned so callers can feed the model.
82 func RunForeground(ctx context.Context, req Request) Result {
83 if len(req.Argv) == 0 {
84 return Result{
85 State: tool.ShellStateFailed,
86 FailurePhase: tool.ShellPhaseLaunch,
87 Err: fmt.Errorf("empty argv"),
88 }
89 }
90 waitDelay := req.WaitDelay
91 if waitDelay <= 0 {
92 waitDelay = DefaultWaitDelay
93 }
94 runCtx := ctx
95 var cancel context.CancelFunc
96 if req.Timeout > 0 {
97 runCtx, cancel = context.WithTimeoutCause(ctx, req.Timeout, errForegroundTimeout)
98 defer cancel()
99 }
100
101 cmd := proc.CommandContext(runCtx, req.Argv[0], req.Argv[1:]...)
102 cmd.Dir = req.Dir
103 cmd.Env = req.Env
104 cmd.WaitDelay = waitDelay
105
106 collector := newOutputCollector(combinedOutputMaxBytes, tool.OutputTailMaxBytes)
107 var writers []io.Writer
108 writers = append(writers, collector.combined, collector.tail)
109 var progress *progressWriter
110 if req.Progress != nil {
111 progress = newProgressWriter(req.Progress, progressOutputMaxBytes, progressOutputTruncated)
112 writers = append(writers, progress)
113 }
114 // Stdout and Stderr must stay the *same* writer value: os/exec then hands the
115 // child a single pipe, so the two streams interleave in the order the child
116 // wrote them and only one copy goroutine calls Progress. Two MultiWriters
117 // would mean two pipes, and combined output would be reordered per stream.
118 // The bounded tail therefore covers combined output rather than stderr only;
119 // failing commands routinely report on stdout, so the tail stays useful.
120 w := io.MultiWriter(writers...)
121 cmd.Stdout = w
122 cmd.Stderr = w
123
124 run := req.Run
125 if run == nil {
126 run = proc.RunCommand
127 }
128 source := req.Source
129 if source == "" {
130 source = "shellrun"
131 }
132 tracked, err := run(runCtx, cmd, proc.RunOptions{
133 Track: req.Track,
134 CancelWaitGrace: waitDelay + time.Second,
135 Source: source,
136 ShellKind: req.ShellKind,
137 ShellPath: req.ShellPath,
138 CommandPreview: req.CommandPreview,
139 })
140
141 if progress != nil {
142 progress.Flush()
143 }
144 out := Result{
145 Combined: collector.combined.String(),
146 OutputTail: collector.tailString(),
147 Started: processStarted(cmd, err),
148 Tracked: tracked,
149 Cmd: cmd,
150 }
151
152 return classifyForegroundResult(runCtx, req, out, err)
153 }
154
155 func classifyForegroundResult(runCtx context.Context, req Request, out Result, err error) Result {
156 if req.PreserveWaitDelay && runCtx.Err() == nil && errors.Is(err, exec.ErrWaitDelay) {
157 err = nil
158 }
159
160 // Timeout takes precedence when the tool-local deadline fired.
161 if errors.Is(context.Cause(runCtx), errForegroundTimeout) {
162 out.State = tool.ShellStateTimedOut
163 out.FailurePhase = tool.ShellPhaseTimeout
164 out.ExitCode = exitCodeFromErr(err)
165 out.Err = fmt.Errorf("command timed out (> %s)", req.Timeout)
166 return out
167 }
168 // Parent cancellation (user stop / session cancel).
169 if err != nil && (errors.Is(err, context.Canceled) || errors.Is(runCtx.Err(), context.Canceled) || isCanceledWait(err)) {
170 out.State = tool.ShellStateCancelled
171 out.FailurePhase = tool.ShellPhaseCancellation
172 out.ExitCode = exitCodeFromErr(err)
173 if cause := context.Cause(runCtx); cause != nil {
174 out.Err = cause
175 } else {
176 out.Err = err
177 }
178 return out
179 }
180 if err == nil {
181 code := 0
182 out.ExitCode = &code
183 out.State = tool.ShellStateCompleted
184 // The tail exists to explain a failure. Dropping it on success keeps
185 // successful runs from persisting up to 16 KiB of ordinary stdout into
186 // every session record and tool card.
187 out.OutputTail = ""
188 return out
189 }
190 if code := exitCodeFromErr(err); code != nil {
191 out.ExitCode = code
192 out.Started = true
193 out.State = tool.ShellStateFailed
194 out.FailurePhase = tool.ShellPhaseExecution
195 out.Err = fmt.Errorf("command exited: %w", err)
196 if diagnostic := WindowsRuntimeDiagnostic(out.Combined); diagnostic != "" {
197 out.Err = fmt.Errorf("%s: %w", diagnostic, out.Err)
198 }
199 return out
200 }
201 // Process never produced an exit status — launch / dependency style failure.
202 out.State = tool.ShellStateFailed
203 if out.Started {
204 out.FailurePhase = tool.ShellPhaseExecution
205 } else {
206 out.FailurePhase = tool.ShellPhaseLaunch
207 }
208 out.Err = err
209 return out
210 }
211
212 func processStarted(cmd *exec.Cmd, err error) bool {
213 if cmd != nil && cmd.Process != nil {
214 return true
215 }
216 // ExitError means the process ran.
217 var ee *exec.ExitError
218 return errors.As(err, &ee)
219 }
220
221 func exitCodeFromErr(err error) *int {
222 if err == nil {
223 code := 0
224 return &code
225 }
226 var ee *exec.ExitError
227 if errors.As(err, &ee) {
228 code := ee.ExitCode()
229 return &code
230 }
231 return nil
232 }
233
234 func isCanceledWait(err error) bool {
235 var c proc.CanceledWaitError
236 return errors.As(err, &c)
237 }
238
239 // outputCollector owns the combined buffer and a bounded tail ring. Writes stay
240 // serialized behind one mutex so a caller that does wire two pipes cannot race
241 // on the Buffer.
242 type outputCollector struct {
243 mu sync.Mutex
244 combined *boundedBuffer
245 tail *tailWriter
246 }
247
248 func newOutputCollector(combinedLimit, tailLimit int) *outputCollector {
249 c := &outputCollector{}
250 c.combined = &boundedBuffer{
251 mu: &c.mu,
252 limit: combinedLimit,
253 tailLimit: combinedOutputTailBytes,
254 marker: combinedOutputTruncated,
255 }
256 c.tail = &tailWriter{mu: &c.mu, limit: tailLimit}
257 return c
258 }
259
260 func (c *outputCollector) tailString() string {
261 c.mu.Lock()
262 defer c.mu.Unlock()
263 return string(completeTail(c.tail.buf))
264 }
265
266 // boundedBuffer keeps complete output up to limit. Once output crosses the
267 // limit it retains a head plus a rolling tail separated by marker. Write always
268 // reports the full input consumed so a safety cap never changes child-process
269 // behavior into an artificial short-write failure.
270 type boundedBuffer struct {
271 mu *sync.Mutex
272 buf bytes.Buffer
273 tail []byte
274 limit int
275 tailLimit int
276 marker string
277 truncated bool
278 }
279
280 func (b *boundedBuffer) Write(p []byte) (int, error) {
281 b.mu.Lock()
282 defer b.mu.Unlock()
283 if len(p) == 0 {
284 return 0, nil
285 }
286 if !b.truncated && (b.limit <= 0 || b.buf.Len()+len(p) <= b.limit) {
287 _, err := b.buf.Write(p)
288 return len(p), err
289 }
290 if !b.truncated {
291 b.truncated = true
292 headLimit := max(0, b.limit-b.tailLimit-len(b.marker))
293 previous := b.buf.Bytes()
294 b.tail = appendBoundedTail(b.tail, previous, b.tailLimit)
295 if b.buf.Len() > headLimit {
296 b.buf.Truncate(headLimit)
297 } else if remaining := headLimit - b.buf.Len(); remaining > 0 {
298 b.buf.Write(p[:min(remaining, len(p))])
299 }
300 }
301 b.tail = appendBoundedTail(b.tail, p, b.tailLimit)
302 return len(p), nil
303 }
304
305 func (b *boundedBuffer) String() string {
306 b.mu.Lock()
307 defer b.mu.Unlock()
308 if !b.truncated {
309 return b.buf.String()
310 }
311 var out strings.Builder
312 out.Grow(b.buf.Len() + len(b.marker) + len(b.tail))
313 out.Write(completePrefix(b.buf.Bytes()))
314 out.WriteString(b.marker)
315 out.Write(completeTail(b.tail))
316 return out.String()
317 }
318
319 func appendBoundedTail(dst, p []byte, limit int) []byte {
320 if limit <= 0 || len(p) >= limit {
321 if limit <= 0 {
322 return nil
323 }
324 return append(dst[:0], p[len(p)-limit:]...)
325 }
326 if overflow := len(dst) + len(p) - limit; overflow > 0 {
327 copy(dst, dst[overflow:])
328 dst = dst[:len(dst)-overflow]
329 }
330 return append(dst, p...)
331 }
332
333 type tailWriter struct {
334 mu *sync.Mutex
335 limit int
336 buf []byte
337 }
338
339 func (w *tailWriter) Write(p []byte) (int, error) {
340 w.mu.Lock()
341 defer w.mu.Unlock()
342 w.buf = append(w.buf, p...)
343 if w.limit > 0 && len(w.buf) > w.limit {
344 w.buf = append([]byte(nil), w.buf[len(w.buf)-w.limit:]...)
345 }
346 return len(p), nil
347 }
348
349 type progressWriter struct {
350 mu sync.Mutex
351 emit func(string)
352 limit int
353 forwarded int
354 marker string
355 truncated bool
356 pending []byte
357 }
358
359 func newProgressWriter(emit func(string), limit int, marker string) *progressWriter {
360 return &progressWriter{emit: emit, limit: max(0, limit), marker: marker}
361 }
362
363 func (w *progressWriter) Write(p []byte) (int, error) {
364 if len(p) == 0 {
365 return 0, nil
366 }
367 w.mu.Lock()
368 defer w.mu.Unlock()
369 if w.emit == nil || w.truncated {
370 return len(p), nil
371 }
372 w.writeUTF8(p, false)
373 return len(p), nil
374 }
375
375 lines GO