返回 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 := max(time.Since(started), 0)
89 return &startupFailure{
90 Stage: strings.TrimSpace(stage),
91 Elapsed: elapsed,
92 Stderr: secrets.RedactCredentials(strings.TrimSpace(stderr)),
93 Err: err,
94 }
95 }
96
97 func formatElapsed(elapsed time.Duration) string {
98 if elapsed < time.Millisecond {
99 return elapsed.String()
100 }
101 return elapsed.Round(time.Millisecond).String()
102 }
103
104 // tailBuffer is a bounded ring holding the most recent stderr bytes. Writes
105 // never block the child; only the tail is ever surfaced, and only after
106 // credential redaction.
107 type tailBuffer struct {
108 mu sync.Mutex
109 limit int
110 buf []byte
111 }
112
113 func (b *tailBuffer) Write(p []byte) (int, error) {
114 b.mu.Lock()
115 defer b.mu.Unlock()
116 b.buf = append(b.buf, p...)
117 if b.limit > 0 && len(b.buf) > b.limit {
118 b.buf = append([]byte(nil), b.buf[len(b.buf)-b.limit:]...)
119 }
120 return len(p), nil
121 }
122
123 func (b *tailBuffer) String() string {
124 b.mu.Lock()
125 defer b.mu.Unlock()
126 return strings.TrimSpace(string(b.buf))
127 }
128
129 // process is one spawned sidecar OS process with its pipes and tracked
130 // process-tree handle.
131 type process struct {
132 pluginID string
133 cmd *exec.Cmd
134 job uintptr
135 stdin io.WriteCloser
136 stdout io.ReadCloser
137 stderr *tailBuffer
138
139 waitOnce sync.Once
140 waitDone chan struct{}
141 jobOnce sync.Once
142 }
143
144 // resolveRuntimeCommand expands ${REASONIX_PLUGIN_ROOT} and enforces the exec
145 // contract: the resolved command must be an absolute path to the extension
146 // executable itself, never a relative name (no PATH lookup — the package must
147 // know exactly what it runs) and never a shell.
148 func resolveRuntimeCommand(rt *pluginpkg.RuntimeSpec, root string) (string, error) {
149 command := strings.TrimSpace(pluginpkg.ExpandRuntimeCommand(rt.Command, root))
150 if command == "" {
151 return "", errors.New("runtime command is empty after expansion")
152 }
153 if !filepath.IsAbs(command) {
154 return "", fmt.Errorf("runtime command %q is not an absolute path after expansion (use %s to address the installed package)", rt.Command, pluginpkg.PluginRootEnvVar)
155 }
156 base := strings.ToLower(filepath.Base(command))
157 if shellExecutables[base] {
158 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)
159 }
160 return command, nil
161 }
162
163 // runtimeEnv builds the sidecar's environment: the UNFILTERED inherited
164 // process environment (full-trust contract — see the package doc), the
165 // manifest's env, and the well-known plugin identity variables. Later entries
166 // win over earlier ones for duplicate keys (exec.Cmd.Env semantics), so the
167 // manifest can tune but the identity variables cannot be forged by it.
168 func runtimeEnv(rt *pluginpkg.RuntimeSpec, pkg pluginpkg.Package, installed pluginpkg.InstalledPlugin) []string {
169 env := append([]string(nil), os.Environ()...)
170 for key, value := range rt.Env {
171 env = append(env, key+"="+value)
172 }
173 version := strings.TrimSpace(installed.Version)
174 if version == "" {
175 version = strings.TrimSpace(pkg.Manifest.Version)
176 }
177 env = append(env,
178 envPluginRoot+"="+pkg.Root,
179 envPluginName+"="+installed.Name,
180 envPluginVersion+"="+version,
181 )
182 return env
183 }
184
185 // startProcess spawns the sidecar process. It never goes through a shell:
186 // exec.Command takes the resolved executable and the argv vector directly.
187 func startProcess(pkg pluginpkg.Package, installed pluginpkg.InstalledPlugin) (*process, error) {
188 started := time.Now()
189 rt := pkg.Manifest.Runtime
190 if rt == nil {
191 return nil, fmt.Errorf("plugin %q declares no runtime", installed.Name)
192 }
193 command, err := resolveRuntimeCommand(rt, pkg.Root)
194 if err != nil {
195 return nil, newStartupFailure("resolve", started, "", err)
196 }
197 cmd := proc.Command(command, rt.Args...)
198 cmd.Env = runtimeEnv(rt, pkg, installed)
199 proc.HideWindow(cmd)
200
201 stdin, err := cmd.StdinPipe()
202 if err != nil {
203 return nil, newStartupFailure("pipes", started, "", err)
204 }
205 stdout, err := cmd.StdoutPipe()
206 if err != nil {
207 return nil, newStartupFailure("pipes", started, "", err)
208 }
209 stderr := &tailBuffer{limit: stderrTailBytes}
210 cmd.Stderr = stderr
211
212 job, err := proc.StartTracked(cmd)
213 if err != nil {
214 return nil, newStartupFailure("spawn", started, stderr.String(), err)
215 }
216 p := &process{
217 pluginID: installed.Name,
218 cmd: cmd,
219 job: job,
220 stdin: stdin,
221 stdout: stdout,
222 stderr: stderr,
223 waitDone: make(chan struct{}),
224 }
225 return p, nil
226 }
227
228 // wait blocks until the process exits, exactly once; later callers observe
229 // the same completed wait. Safe to abandon: the first caller owns cmd.Wait.
230 func (p *process) wait() {
231 p.waitOnce.Do(func() {
232 if p.cmd != nil && p.cmd.Process != nil {
233 _ = p.cmd.Wait()
234 }
235 p.finishJob()
236 close(p.waitDone)
237 })
238 }
239
240 // finishJob and kill share ownership of the Windows Job Object. Its numeric
241 // handle may be reused immediately after either path closes it.
242 func (p *process) finishJob() {
243 p.jobOnce.Do(func() { proc.FinishTracked(p.job) })
244 }
245
246 // kill terminates the whole process tree.
247 func (p *process) kill() {
248 if p.cmd == nil || p.cmd.Process == nil {
249 return
250 }
251 if p.job != 0 {
252 p.jobOnce.Do(func() { proc.KillTracked(p.cmd, p.job) })
253 return
254 }
255 proc.KillTracked(p.cmd, 0)
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