返回 DeepSeek-Reasonix
process.go
根目录 / internal / extension / sidecar / process.go
1 package sidecar
2
3 import (
4 "errors"
5 "fmt"
6 "io"
7 "os"
8 "os/exec"
9 "path/filepath"
10 "strings"
11 "sync"
12 "time"
13
14 "reasonix/internal/pluginpkg"
15 "reasonix/internal/proc"
16 "reasonix/internal/secrets"
17 )
18
19 // Bounded-close budgets, mirrored from internal/plugin's stdio transport: a
20 // short stdin-EOF grace for protocol-aware sidecars, then a hard tree kill
21 // with a bounded reap so one wedged sidecar can never stall teardown.
22 const (
23 gracefulCloseWaitBudget = 750 * time.Millisecond
24 closeWaitBudget = 5 * time.Second
25 // stderrTailBytes bounds the ring of sidecar stderr retained for
26 // diagnostics, mirrored from internal/plugin's tailBuffer.
27 stderrTailBytes = 16 << 10
28 )
29
30 // pluginEnvVarPrefix is the well-known environment block every sidecar sees.
31 const (
32 envPluginRoot = "REASONIX_PLUGIN_ROOT"
33 envPluginName = "REASONIX_PLUGIN_NAME"
34 envPluginVersion = "REASONIX_PLUGIN_VERSION"
35 )
36
37 // shellExecutables are interpreter names a runtime command may not resolve
38 // to. The runtime contract is exec form: the command IS the extension
39 // executable and args are its argv. Routing through a shell would smuggle
40 // shell semantics (pipes, &&, $ expansion) into a contract that promises
41 // there are none.
42 var shellExecutables = map[string]bool{
43 "sh": true, "bash": true, "zsh": true, "fish": true, "dash": true, "ksh": true,
44 "cmd": true, "cmd.exe": true, "powershell": true, "powershell.exe": true,
45 "pwsh": true, "pwsh.exe": true,
46 }
47
48 // startupFailure carries bounded diagnostics for a sidecar that failed to
49 // start or hand shake: the stage, elapsed time, and the redacted stderr tail.
50 // The shape mirrors internal/plugin's startupFailure.
51 type startupFailure struct {
52 Stage string
53 Elapsed time.Duration
54 Stderr string
55 Err error
56 }
57
58 func (e *startupFailure) Error() string {
59 if e == nil {
60 return "extension sidecar startup failed"
61 }
62 stage := strings.TrimSpace(e.Stage)
63 if stage == "" {
64 stage = "unknown"
65 }
66 msg := fmt.Sprintf("extension sidecar startup %s failed after %s: %s", stage, formatElapsed(e.Elapsed), secrets.RedactError(e.Err))
67 if stderr := strings.TrimSpace(e.Stderr); stderr != "" {
68 msg += "; stderr: " + stderr
69 }
70 return msg
71 }
72
73 func (e *startupFailure) Unwrap() error {
74 if e == nil {
75 return nil
76 }
77 return e.Err
78 }
79
80 func newStartupFailure(stage string, started time.Time, stderr string, err error) error {
81 if err == nil {
82 return nil
83 }
84 var existing *startupFailure
85 if errors.As(err, &existing) {
86 return err
87 }
88 elapsed := time.Since(started)
89 if elapsed < 0 {
90 elapsed = 0
91 }
92 return &startupFailure{
93 Stage: strings.TrimSpace(stage),
94 Elapsed: elapsed,
95 Stderr: secrets.RedactCredentials(strings.TrimSpace(stderr)),
96 Err: err,
97 }
98 }
99
100 func formatElapsed(elapsed time.Duration) string {
101 if elapsed < time.Millisecond {
102 return elapsed.String()
103 }
104 return elapsed.Round(time.Millisecond).String()
105 }
106
107 // tailBuffer is a bounded ring holding the most recent stderr bytes. Writes
108 // never block the child; only the tail is ever surfaced, and only after
109 // credential redaction.
110 type tailBuffer struct {
111 mu sync.Mutex
112 limit int
113 buf []byte
114 }
115
116 func (b *tailBuffer) Write(p []byte) (int, error) {
117 b.mu.Lock()
118 defer b.mu.Unlock()
119 b.buf = append(b.buf, p...)
120 if b.limit > 0 && len(b.buf) > b.limit {
121 b.buf = append([]byte(nil), b.buf[len(b.buf)-b.limit:]...)
122 }
123 return len(p), nil
124 }
125
126 func (b *tailBuffer) String() string {
127 b.mu.Lock()
128 defer b.mu.Unlock()
129 return strings.TrimSpace(string(b.buf))
130 }
131
132 // process is one spawned sidecar OS process with its pipes and tracked
133 // process-tree handle.
134 type process struct {
135 pluginID string
136 cmd *exec.Cmd
137 job uintptr
138 stdin io.WriteCloser
139 stdout io.ReadCloser
140 stderr *tailBuffer
141
142 waitOnce sync.Once
143 waitDone chan struct{}
144 jobOnce sync.Once
145 }
146
147 // resolveRuntimeCommand expands ${REASONIX_PLUGIN_ROOT} and enforces the exec
148 // contract: the resolved command must be an absolute path to the extension
149 // executable itself, never a relative name (no PATH lookup — the package must
150 // know exactly what it runs) and never a shell.
151 func resolveRuntimeCommand(rt *pluginpkg.RuntimeSpec, root string) (string, error) {
152 command := strings.TrimSpace(pluginpkg.ExpandRuntimeCommand(rt.Command, root))
153 if command == "" {
154 return "", errors.New("runtime command is empty after expansion")
155 }
156 if !filepath.IsAbs(command) {
157 return "", fmt.Errorf("runtime command %q is not an absolute path after expansion (use %s to address the installed package)", rt.Command, pluginpkg.PluginRootEnvVar)
158 }
159 base := strings.ToLower(filepath.Base(command))
160 if shellExecutables[base] {
161 return "", fmt.Errorf("runtime command %q is a shell; the runtime contract is exec form (the command is the extension executable, args are its argv)", rt.Command)
162 }
163 return command, nil
164 }
165
166 // runtimeEnv builds the sidecar's environment: the UNFILTERED inherited
167 // process environment (full-trust contract — see the package doc), the
168 // manifest's env, and the well-known plugin identity variables. Later entries
169 // win over earlier ones for duplicate keys (exec.Cmd.Env semantics), so the
170 // manifest can tune but the identity variables cannot be forged by it.
171 func runtimeEnv(rt *pluginpkg.RuntimeSpec, pkg pluginpkg.Package, installed pluginpkg.InstalledPlugin) []string {
172 env := append([]string(nil), os.Environ()...)
173 for key, value := range rt.Env {
174 env = append(env, key+"="+value)
175 }
176 version := strings.TrimSpace(installed.Version)
177 if version == "" {
178 version = strings.TrimSpace(pkg.Manifest.Version)
179 }
180 env = append(env,
181 envPluginRoot+"="+pkg.Root,
182 envPluginName+"="+installed.Name,
183 envPluginVersion+"="+version,
184 )
185 return env
186 }
187
188 // startProcess spawns the sidecar process. It never goes through a shell:
189 // exec.Command takes the resolved executable and the argv vector directly.
190 func startProcess(pkg pluginpkg.Package, installed pluginpkg.InstalledPlugin) (*process, error) {
191 started := time.Now()
192 rt := pkg.Manifest.Runtime
193 if rt == nil {
194 return nil, fmt.Errorf("plugin %q declares no runtime", installed.Name)
195 }
196 command, err := resolveRuntimeCommand(rt, pkg.Root)
197 if err != nil {
198 return nil, newStartupFailure("resolve", started, "", err)
199 }
200 cmd := exec.Command(command, rt.Args...)
201 cmd.Env = runtimeEnv(rt, pkg, installed)
202 proc.HideWindow(cmd)
203
204 stdin, err := cmd.StdinPipe()
205 if err != nil {
206 return nil, newStartupFailure("pipes", started, "", err)
207 }
208 stdout, err := cmd.StdoutPipe()
209 if err != nil {
210 return nil, newStartupFailure("pipes", started, "", err)
211 }
212 stderr := &tailBuffer{limit: stderrTailBytes}
213 cmd.Stderr = stderr
214
215 job, err := proc.StartTracked(cmd)
216 if err != nil {
217 return nil, newStartupFailure("spawn", started, stderr.String(), err)
218 }
219 p := &process{
220 pluginID: installed.Name,
221 cmd: cmd,
222 job: job,
223 stdin: stdin,
224 stdout: stdout,
225 stderr: stderr,
226 waitDone: make(chan struct{}),
227 }
228 return p, nil
229 }
230
231 // wait blocks until the process exits, exactly once; later callers observe
232 // the same completed wait. Safe to abandon: the first caller owns cmd.Wait.
233 func (p *process) wait() {
234 p.waitOnce.Do(func() {
235 if p.cmd != nil && p.cmd.Process != nil {
236 _ = p.cmd.Wait()
237 }
238 p.finishJob()
239 close(p.waitDone)
240 })
241 }
242
243 // finishJob releases the Windows Job Object once the process is known to be
244 // gone (a no-op off Windows and after KillTracked already released it).
245 func (p *process) finishJob() {
246 p.jobOnce.Do(func() { proc.FinishTracked(p.job) })
247 }
248
249 // kill terminates the whole process tree.
250 func (p *process) kill() {
251 if p.cmd == nil || p.cmd.Process == nil {
252 return
253 }
254 proc.KillTracked(p.cmd, p.job)
255 p.finishJob()
256 }
257
258 // close stops the sidecar with the bounded sequence: close stdin, grant a
259 // short EOF grace for protocol-aware processes, kill the tree, and wait a
260 // bounded time for the reap. It is idempotent and never blocks longer than
261 // gracefulCloseWaitBudget + closeWaitBudget.
262 func (p *process) close() {
263 if p.stdin != nil {
264 _ = p.stdin.Close()
265 }
266 if p.cmd == nil || p.cmd.Process == nil {
267 return
268 }
269 if waitFinishedWithinBudget(p.wait, gracefulCloseWaitBudget) {
270 return
271 }
272 p.kill()
273 waitWithBudget(p.wait, closeWaitBudget)
274 }
275
276 func waitWithBudget(wait func(), budget time.Duration) {
277 _ = waitFinishedWithinBudget(wait, budget)
278 }
279
280 func waitFinishedWithinBudget(wait func(), budget time.Duration) bool {
281 done := make(chan struct{})
282 go func() { wait(); close(done) }()
283 select {
284 case <-done:
285 return true
286 case <-time.After(budget):
287 return false
288 }
289 }
290
290 lines GO