返回 DeepSeek-Reasonix
client_test.go
根目录 / internal / extension / sidecar / client_test.go
1 package sidecar
2
3 import (
4 "context"
5 "encoding/json"
6 "errors"
7 "strings"
8 "sync/atomic"
9 "testing"
10 "time"
11
12 "reasonix/internal/extension"
13 "reasonix/internal/extension/protocol"
14 "reasonix/internal/extension/rpcwire"
15 "reasonix/internal/pluginpkg"
16 )
17
18 func TestHandshakeSuccess(t *testing.T) {
19 client := startFakeClient(t, func(rt *pluginpkg.RuntimeSpec) {
20 rt.Intercepts = []string{"input.receive", "tool.before"}
21 rt.Env[fakeEnvInitResult] = `{"protocolVersion":"1","name":"fake-sidecar","version":"1.2.3","subscriptions":["input.receive"],"stateSchemaVersion":0}`
22 }, nil)
23 result := client.Handshake()
24 if result.Name != "fake-sidecar" || result.Version != "1.2.3" {
25 t.Fatalf("handshake identity = %q %q", result.Name, result.Version)
26 }
27 if len(result.Subscriptions) != 1 || result.Subscriptions[0] != "input.receive" {
28 t.Fatalf("subscriptions = %v", result.Subscriptions)
29 }
30 if client.Crashed() {
31 t.Fatal("client crashed during handshake")
32 }
33 }
34
35 func TestHandshakeProtocolVersionMismatch(t *testing.T) {
36 pkg, installed := fakeSidecarPackage(t, "fakeplugin", func(rt *pluginpkg.RuntimeSpec) {
37 rt.Env[fakeEnvInitResult] = `{"protocolVersion":"2","name":"fake-sidecar","version":"1.0.0","stateSchemaVersion":0}`
38 })
39 _, err := StartClient(context.Background(), ClientOptions{Package: pkg, Installed: installed, Session: testSessionContext()})
40 if err == nil {
41 t.Fatal("StartClient succeeded with protocol major 2")
42 }
43 if reason := protocolReason(t, err); reason != protocol.ErrUnsupportedVersion {
44 t.Fatalf("reason = %q, want %q", reason, protocol.ErrUnsupportedVersion)
45 }
46 }
47
48 func TestMapRequestErrorRedactsPeerMessages(t *testing.T) {
49 const secret = "sk-abcdef1234567890SECRETKEY"
50 structuredData, err := json.Marshal(protocol.ProtocolErrorData{
51 Reason: protocol.ErrProviderFailed,
52 Retryable: true,
53 })
54 if err != nil {
55 t.Fatalf("marshal protocol error data: %v", err)
56 }
57 tests := []struct {
58 name string
59 data json.RawMessage
60 }{
61 {name: "structured protocol error", data: structuredData},
62 {name: "unstructured transport error", data: json.RawMessage(`{"unexpected":true}`)},
63 }
64 for _, tt := range tests {
65 t.Run(tt.name, func(t *testing.T) {
66 mapped := mapRequestError(&rpcwire.ResponseError{
67 Code: protocol.DomainErrorCode,
68 Message: "provider rejected api_key=" + secret,
69 Data: tt.data,
70 })
71 if strings.Contains(mapped.Error(), secret) {
72 t.Fatalf("mapped error leaked peer credential: %q", mapped)
73 }
74 if !strings.Contains(mapped.Error(), "****") {
75 t.Fatalf("mapped error contains no redaction marker: %q", mapped)
76 }
77 })
78 }
79 }
80
81 // TestHandshakeCapabilityViolations pins the declaration contract: anything
82 // the sidecar activates beyond its manifest fails the handshake with
83 // capability_not_declared.
84 func TestHandshakeCapabilityViolations(t *testing.T) {
85 cases := []struct {
86 name string
87 configure func(rt *pluginpkg.RuntimeSpec)
88 initResult string
89 }{
90 {
91 name: "subscriptions superset",
92 configure: func(rt *pluginpkg.RuntimeSpec) { rt.Intercepts = []string{"input.receive"} },
93 initResult: `{"protocolVersion":"1","name":"fake","version":"1","subscriptions":["input.receive","tool.before"],"stateSchemaVersion":0}`,
94 },
95 {
96 name: "replaces superset",
97 configure: func(rt *pluginpkg.RuntimeSpec) { rt.Replaces = []string{"system_prompt"} },
98 initResult: `{"protocolVersion":"1","name":"fake","version":"1","replaces":["system_prompt","compaction"],"stateSchemaVersion":0}`,
99 },
100 {
101 name: "providers without capability",
102 configure: nil,
103 initResult: `{"protocolVersion":"1","name":"fake","version":"1","providers":[{"ref":"plugin/fakeplugin/openai/gpt-5"}],"stateSchemaVersion":0}`,
104 },
105 {
106 name: "provider ref outside plugin namespace",
107 configure: func(rt *pluginpkg.RuntimeSpec) { rt.Capabilities = []string{"providers"} },
108 initResult: `{"protocolVersion":"1","name":"fake","version":"1","providers":[{"ref":"plugin/other/openai/gpt-5"}],"stateSchemaVersion":0}`,
109 },
110 {
111 name: "ui actions without capability",
112 configure: nil,
113 initResult: `{"protocolVersion":"1","name":"fake","version":"1","uiActions":[{"actionId":"a1"}],"stateSchemaVersion":0}`,
114 },
115 }
116 for _, tc := range cases {
117 t.Run(tc.name, func(t *testing.T) {
118 pkg, installed := fakeSidecarPackage(t, "fakeplugin", func(rt *pluginpkg.RuntimeSpec) {
119 if tc.configure != nil {
120 tc.configure(rt)
121 }
122 rt.Env[fakeEnvInitResult] = tc.initResult
123 })
124 _, err := StartClient(context.Background(), ClientOptions{Package: pkg, Installed: installed, Session: testSessionContext()})
125 if err == nil {
126 t.Fatal("StartClient succeeded with an undeclared capability in use")
127 }
128 if reason := protocolReason(t, err); reason != protocol.ErrCapabilityNotDeclared {
129 t.Fatalf("reason = %q, want %q", reason, protocol.ErrCapabilityNotDeclared)
130 }
131 })
132 }
133 }
134
135 func TestHandshakeDeclaredProvidersAndUIAccepted(t *testing.T) {
136 client := startFakeClient(t, func(rt *pluginpkg.RuntimeSpec) {
137 rt.Capabilities = []string{"providers", "ui"}
138 rt.Env[fakeEnvInitResult] = `{"protocolVersion":"1","name":"fake","version":"1",` +
139 `"providers":[{"ref":"plugin/fakeplugin/openai/gpt-5"}],` +
140 `"uiActions":[{"actionId":"act1","label":"Act"}],"stateSchemaVersion":0}`
141 }, nil)
142 result := client.Handshake()
143 if len(result.Providers) != 1 || result.Providers[0].Ref != "plugin/fakeplugin/openai/gpt-5" {
144 t.Fatalf("providers = %+v", result.Providers)
145 }
146 if len(result.UIActions) != 1 || result.UIActions[0].ActionID != "act1" {
147 t.Fatalf("uiActions = %+v", result.UIActions)
148 }
149 }
150
151 // TestTrafficBeforeInitializedPoisons covers both E→H frame kinds arriving
152 // before extension/initialized: the connection is poisoned and the start
153 // fails with protocol_error.
154 func TestTrafficBeforeInitializedPoisons(t *testing.T) {
155 for _, mode := range []string{"early_request", "early_notify"} {
156 t.Run(mode, func(t *testing.T) {
157 pkg, installed := fakeSidecarPackage(t, "fakeplugin", func(rt *pluginpkg.RuntimeSpec) {
158 rt.Env[fakeEnvMode] = mode
159 })
160 _, err := StartClient(context.Background(), ClientOptions{Package: pkg, Installed: installed, Session: testSessionContext()})
161 if err == nil {
162 t.Fatal("StartClient succeeded despite pre-initialized traffic")
163 }
164 if reason := protocolReason(t, err); reason != protocol.ErrProtocolError {
165 t.Fatalf("reason = %q, want %q", reason, protocol.ErrProtocolError)
166 }
167 if !strings.Contains(err.Error(), "before extension/initialized") {
168 t.Fatalf("error %q does not name the gating rule", err)
169 }
170 })
171 }
172 }
173
174 func TestInterceptContinueAndNotifications(t *testing.T) {
175 client := startFakeClient(t, nil, nil)
176 result, err := client.Intercept(context.Background(), protocol.EventInputReceive, json.RawMessage(`{"text":"hi"}`), 5*time.Second)
177 if err != nil {
178 t.Fatalf("Intercept: %v", err)
179 }
180 if result.Decision != protocol.DecisionContinue {
181 t.Fatalf("decision = %q, want continue", result.Decision)
182 }
183 if err := client.NotifyEvent(protocol.EventSessionStart, json.RawMessage(`{"at":1}`)); err != nil {
184 t.Fatalf("NotifyEvent: %v", err)
185 }
186 if err := client.NotifyResourcesChanged([]string{"skills/x.md"}); err != nil {
187 t.Fatalf("NotifyResourcesChanged: %v", err)
188 }
189 }
190
191 func TestInterceptTimeout(t *testing.T) {
192 client := startFakeClient(t, func(rt *pluginpkg.RuntimeSpec) {
193 rt.Env[fakeEnvMode] = "stall_intercept"
194 }, nil)
195 start := time.Now()
196 _, err := client.Intercept(context.Background(), protocol.EventToolBefore, json.RawMessage(`{}`), 300*time.Millisecond)
197 if err == nil {
198 t.Fatal("Intercept succeeded against a stalled sidecar")
199 }
200 if reason := protocolReason(t, err); reason != protocol.ErrInterceptTimeout {
201 t.Fatalf("reason = %q, want %q", reason, protocol.ErrInterceptTimeout)
202 }
203 if elapsed := time.Since(start); elapsed > 5*time.Second {
204 t.Fatalf("intercept timeout took %s, want near 300ms", elapsed)
205 }
206 }
207
208 // TestShutdownBounded covers a sidecar that ignores extension/shutdown: the
209 // bounded close kills and reaps the tree within budget, and Shutdown is
210 // idempotent.
211 func TestShutdownBounded(t *testing.T) {
212 client := startFakeClient(t, func(rt *pluginpkg.RuntimeSpec) {
213 rt.Env[fakeEnvMode] = "ignore_shutdown"
214 }, nil)
215 start := time.Now()
216 if err := client.Shutdown(context.Background(), 300*time.Millisecond); err != nil {
217 t.Fatalf("Shutdown: %v", err)
218 }
219 elapsed := time.Since(start)
220 // 300ms request + 750ms EOF grace + kill + 5s reap must finish far below
221 // this ceiling.
222 if elapsed > 10*time.Second {
223 t.Fatalf("bounded shutdown took %s", elapsed)
224 }
225 if !client.Exited() {
226 t.Fatal("sidecar process still running after bounded shutdown")
227 }
228 // Idempotent: a second call returns immediately.
229 second := time.Now()
230 _ = client.Shutdown(context.Background(), 5*time.Second)
231 if time.Since(second) > time.Second {
232 t.Fatal("second Shutdown was not idempotent")
233 }
234 }
235
236 // TestCrashFailsPendingAndFastAfter kills the fake sidecar mid-intercept:
237 // the pending call errors, the crash callback fires exactly once, and later
238 // calls fail fast with provider_interrupted.
239 func TestCrashFailsPendingAndFastAfter(t *testing.T) {
240 var crashes atomic.Int32
241 client := startFakeClient(t, func(rt *pluginpkg.RuntimeSpec) {
242 rt.Env[fakeEnvMode] = "stall_intercept"
243 }, func(opts *ClientOptions) {
244 opts.OnCrash = func(error) { crashes.Add(1) }
245 })
246
247 pending := make(chan error, 1)
248 go func() {
249 _, err := client.Intercept(context.Background(), protocol.EventToolBefore, json.RawMessage(`{}`), 30*time.Second)
250 pending <- err
251 }()
252 waitFor(t, "the intercept to reach the sidecar", 5*time.Second, func() bool {
253 return strings.Contains(client.proc.stderr.String(), "intercept-stalled")
254 })
255
256 if err := client.proc.cmd.Process.Kill(); err != nil {
257 t.Fatalf("kill fake sidecar: %v", err)
258 }
259 select {
260 case err := <-pending:
261 if err == nil {
262 t.Fatal("pending Intercept succeeded after the sidecar was killed")
263 }
264 case <-time.After(5 * time.Second):
265 t.Fatal("pending Intercept did not error after the sidecar was killed")
266 }
267 waitFor(t, "crash detection", 5*time.Second, client.Crashed)
268 waitFor(t, "process reaping", 5*time.Second, client.Exited)
269 if got := crashes.Load(); got != 1 {
270 t.Fatalf("OnCrash fired %d times, want exactly 1", got)
271 }
272
273 // Later calls fail fast with the provider_interrupted family.
274 start := time.Now()
275 _, err := client.Intercept(context.Background(), protocol.EventToolBefore, json.RawMessage(`{}`), 30*time.Second)
276 if err == nil {
277 t.Fatal("Intercept succeeded after crash")
278 }
279 if reason := protocolReason(t, err); reason != protocol.ErrProviderInterrupted {
280 t.Fatalf("reason = %q, want %q", reason, protocol.ErrProviderInterrupted)
281 }
282 if elapsed := time.Since(start); elapsed > time.Second {
283 t.Fatalf("post-crash Intercept was not fail-fast (%s)", elapsed)
284 }
285 if got := crashes.Load(); got != 1 {
286 t.Fatalf("OnCrash fired %d times after fail-fast, want exactly 1", got)
287 }
288 }
289
290 func TestTimeoutFor(t *testing.T) {
291 client := startFakeClient(t, nil, nil)
292 fast := []extension.InterceptorPoint{
293 extension.PointInputReceive, extension.PointToolBefore,
294 extension.PointToolAfter, extension.PointPermissionDecision,
295 }
296 for _, point := range fast {
297 if got := client.TimeoutFor(point); got != fastInterceptTimeout {
298 t.Fatalf("TimeoutFor(%s) = %s, want %s", point, got, fastInterceptTimeout)
299 }
300 }
301 slow := []extension.InterceptorPoint{
302 extension.PointSessionStart, extension.PointSystemPromptBuild,
303 extension.PointContextPrepare, extension.PointCompactionPrepare,
304 }
305 for _, point := range slow {
306 if got := client.TimeoutFor(point); got != slowInterceptTimeout {
307 t.Fatalf("TimeoutFor(%s) = %s, want %s", point, got, slowInterceptTimeout)
308 }
309 }
310 }
311
312 func TestTimeoutForManifestOverrideClamped(t *testing.T) {
313 client := startFakeClient(t, func(rt *pluginpkg.RuntimeSpec) {
314 rt.TimeoutMillis = 250
315 }, nil)
316 if got := client.TimeoutFor(extension.PointInputReceive); got != 250*time.Millisecond {
317 t.Fatalf("TimeoutFor with manifest override = %s, want 250ms", got)
318 }
319 clamped := startFakeClient(t, func(rt *pluginpkg.RuntimeSpec) {
320 rt.TimeoutMillis = 10 * 60 * 1000 // 10 minutes
321 }, nil)
322 if got := clamped.TimeoutFor(extension.PointSessionStart); got != maxInterceptTimeout {
323 t.Fatalf("TimeoutFor beyond the ceiling = %s, want the 60s clamp", got)
324 }
325 }
326
327 func TestUIHandlerDefaultsToUnavailable(t *testing.T) {
328 client := startFakeClient(t, nil, nil)
329 _, err := client.ui.Publish(context.Background(), protocol.UIPublishParams{})
330 if err == nil {
331 t.Fatal("default UI handler accepted a publish")
332 }
333 if reason := protocolReason(t, err); reason != protocol.ErrUnknownMethod {
334 t.Fatalf("reason = %q, want %q", reason, protocol.ErrUnknownMethod)
335 }
336 }
337
338 // TestUIActionAndSubmitRoundTrip drives the host-initiated UI calls (stage 8)
339 // over the real wire: the fake sidecar echoes the action id and accepts the
340 // form submission.
341 func TestUIActionAndSubmitRoundTrip(t *testing.T) {
342 client := startFakeClient(t, nil, nil)
343 action, err := client.UIAction(context.Background(), protocol.UIActionParams{
344 ActionID: "act1", SessionID: "sess-test", Generation: 1, Args: map[string]string{"k": "v"},
345 })
346 if err != nil {
347 t.Fatalf("UIAction: %v", err)
348 }
349 if !action.Accepted || action.Message == "" {
350 t.Fatalf("UIAction result = %+v", action)
351 }
352 submit, err := client.UISubmit(context.Background(), protocol.UISubmitParams{
353 SurfaceID: "f1", SessionID: "sess-test", Generation: 1, Values: map[string]any{"name": "x"},
354 })
355 if err != nil {
356 t.Fatalf("UISubmit: %v", err)
357 }
358 if !submit.Accepted {
359 t.Fatalf("UISubmit result = %+v", submit)
360 }
361 }
362
363 func TestStartRejectsInvalidOptions(t *testing.T) {
364 pkg, installed := fakeSidecarPackage(t, "fakeplugin", nil)
365 if _, err := StartClient(context.Background(), ClientOptions{
366 Package: pkg,
367 Installed: installed,
368 Session: protocol.SessionContext{},
369 }); err == nil {
370 t.Fatal("StartClient accepted an empty session context")
371 }
372 pkgNoRuntime := pluginpkg.Package{Root: t.TempDir(), Manifest: pluginpkg.Manifest{Name: "x"}}
373 if _, err := StartClient(context.Background(), ClientOptions{
374 Package: pkgNoRuntime,
375 Installed: installed,
376 Session: testSessionContext(),
377 }); err == nil {
378 t.Fatal("StartClient accepted a package without a runtime")
379 }
380 }
381
382 // TestWriteStallKillsWedgedSidecar is the deterministic regression for the
383 // host-availability review finding: a sidecar that stays alive but stops
384 // reading stdin fills the pipe, and an unbounded write would hang the host
385 // forever. With the write-stall bound the call fails fast, the connection
386 // dies, and the process tree is killed.
387 func TestWriteStallKillsWedgedSidecar(t *testing.T) {
388 client := startFakeClient(t, func(rt *pluginpkg.RuntimeSpec) {
389 rt.Env[fakeEnvMode] = "wedge_after_init"
390 }, func(opts *ClientOptions) {
391 opts.WriteStallBound = 100 * time.Millisecond
392 })
393
394 // Bypass Intercept's externalization (payloads over 64 KiB offload to
395 // content refs) so the frame itself exceeds the OS pipe buffer.
396 big := json.RawMessage(`{"pad":"` + strings.Repeat("x", 1<<20) + `"}`)
397 start := time.Now()
398 _, err := client.conn.Request(context.Background(), string(protocol.MethodExtensionIntercept), json.RawMessage(big))
399 elapsed := time.Since(start)
400 var stall *rpcwire.WriteStallError
401 if !errors.As(err, &stall) {
402 t.Fatalf("error = %v, want WriteStallError", err)
403 }
404 if elapsed > 5*time.Second {
405 t.Fatalf("stall took %s to abort, want near 100ms", elapsed)
406 }
407
408 waitFor(t, "client marked crashed", 5*time.Second, client.Crashed)
409 waitFor(t, "wedged sidecar killed", 5*time.Second, client.Exited)
410 }
411
412 // TestWriteStallWatchdogOutlivesCallerTimeout: a per-call timeout shorter
413 // than the stall bound aborts only that call — the stall watchdog is an
414 // independent absolute bound that still fails the connection and kills the
415 // wedged sidecar afterwards. (Review finding: a 5s intercept ctx must not
416 // preempt the 10s stall watchdog.)
417 func TestWriteStallWatchdogOutlivesCallerTimeout(t *testing.T) {
418 client := startFakeClient(t, func(rt *pluginpkg.RuntimeSpec) {
419 rt.Env[fakeEnvMode] = "wedge_after_init"
420 }, func(opts *ClientOptions) {
421 opts.WriteStallBound = 300 * time.Millisecond
422 })
423
424 big := json.RawMessage(`{"pad":"` + strings.Repeat("x", 1<<20) + `"}`)
425 // Primer: a background-context request whose frame wedges the single
426 // writer for good (the fake never reads). Its write cannot be cancelled,
427 // so the stall watchdog has an active write to trip on.
428 go func() {
429 _, _ = client.conn.Request(context.Background(), string(protocol.MethodExtensionIntercept), big)
430 }()
431 // The timed call cancels fast — but the primer's write outlives it, and
432 // the 300ms stall watchdog still fires: the connection dies and the
433 // wedged process is killed. The caller timeout did not preempt it.
434 ctx, cancel := context.WithCancel(context.Background())
435 go func() {
436 time.Sleep(200 * time.Millisecond)
437 cancel()
438 }()
439 start := time.Now()
440 _, err := client.conn.Request(ctx, string(protocol.MethodExtensionIntercept), big)
441 elapsed := time.Since(start)
442 if !errors.Is(err, context.Canceled) {
443 t.Fatalf("error = %v, want context.Canceled", err)
444 }
445 if elapsed > 3*time.Second {
446 t.Fatalf("caller cancel took %s, want near 200ms", elapsed)
447 }
448
449 waitFor(t, "client marked crashed after caller cancel", 10*time.Second, client.Crashed)
450 waitFor(t, "wedged sidecar killed after caller cancel", 10*time.Second, client.Exited)
451 }
452
452 lines GO