// Copyright 2026 Alibaba Group // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at // // http://www.apache.org/licenses/LICENSE-2.0 // // Unless required by applicable law or agreed to in writing, software // distributed under the License is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. package helpers import ( "context" "encoding/json" "fmt" "net/http" "net/http/httptest" "path/filepath" "strings" "testing" "time" ) func TestOpencodeForwarderUsesServerSessionAPI(t *testing.T) { dir := t.TempDir() storePath := filepath.Join(dir, "opencode-sessions.json") var calls []string var prompts []string ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { calls = append(calls, r.Method+" "+r.URL.Path) w.Header().Set("Content-Type", "application/json") switch { case r.Method == http.MethodGet && r.URL.Path == "/global/health": _, _ = w.Write([]byte(`{"healthy":true,"version":"test"}`)) case r.Method == http.MethodPost && r.URL.Path == "/session": _, _ = w.Write([]byte(`{"id":"ses_server"}`)) case r.Method == http.MethodPost && r.URL.Path == "/session/ses_server/message": var req struct { Parts []struct { Type string `json:"type"` Text string `json:"text"` } `json:"parts"` } if err := json.NewDecoder(r.Body).Decode(&req); err != nil { t.Fatalf("decode message request: %v", err) } if len(req.Parts) != 1 || req.Parts[0].Type != "text" { t.Fatalf("message parts = %#v, want one text part", req.Parts) } prompts = append(prompts, req.Parts[0].Text) _, _ = w.Write([]byte(`{"info":{"id":"msg_1"},"parts":[{"type":"text","text":"reply ` + req.Parts[0].Text + `"}]}`)) default: http.NotFound(w, r) } })) defer ts.Close() f := &opencodeForwarder{ bin: "opencode", timeout: 5 * time.Second, workDir: dir, sessions: newOpencodeSessions(storePath), server: &opencodeServer{baseURL: ts.URL, httpClient: ts.Client()}, } reply, err := f.forwardStream(context.Background(), "conv-1", "hi", func(string) { t.Fatal("opencode server mode must not stream deltas") }) if err != nil { t.Fatalf("first forward: %v", err) } if reply != "reply hi" { t.Fatalf("first reply = %q, want reply hi", reply) } if got := f.sessions.id("conv-1"); got != "ses_server" { t.Fatalf("captured session = %q, want ses_server", got) } reply, err = f.forwardStream(context.Background(), "conv-1", "again", nil) if err != nil { t.Fatalf("second forward: %v", err) } if reply != "reply again" { t.Fatalf("second reply = %q, want reply again", reply) } if got := strings.Join(calls, "\n"); strings.Count(got, "POST /session\n") != 1 && strings.Count(got, "POST /session") != 1 { t.Fatalf("POST /session should be called once, calls:\n%s", got) } if got := strings.Join(prompts, ","); got != "hi,again" { t.Fatalf("prompts = %q, want hi,again", got) } if f.canStream() { t.Fatal("opencode server mode should be one-shot for group chat replies") } } func TestOpencodeForwarderRecreatesMissingServerSession(t *testing.T) { dir := t.TempDir() storePath := filepath.Join(dir, "opencode-sessions.json") var created int ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") switch { case r.Method == http.MethodGet && r.URL.Path == "/global/health": _, _ = w.Write([]byte(`{"healthy":true,"version":"test"}`)) case r.Method == http.MethodPost && r.URL.Path == "/session": created++ _, _ = w.Write([]byte(`{"id":"ses_new"}`)) case r.Method == http.MethodPost && r.URL.Path == "/session/ses_stale/message": http.Error(w, `{"error":"missing"}`, http.StatusNotFound) case r.Method == http.MethodPost && r.URL.Path == "/session/ses_new/message": _, _ = w.Write([]byte(`{"parts":[{"type":"text","text":"fresh"}]}`)) default: http.NotFound(w, r) } })) defer ts.Close() f := &opencodeForwarder{ bin: "opencode", timeout: 5 * time.Second, workDir: dir, sessions: newOpencodeSessions(storePath), server: &opencodeServer{baseURL: ts.URL, httpClient: ts.Client()}, } f.sessions.set("conv-1", "ses_stale") reply, err := f.forwardStream(context.Background(), "conv-1", "again", nil) if err != nil { t.Fatalf("forward after missing session: %v", err) } if reply != "fresh" { t.Fatalf("reply = %q, want fresh", reply) } if got := f.sessions.id("conv-1"); got != "ses_new" { t.Fatalf("session after retry = %q, want ses_new", got) } if created != 1 { t.Fatalf("created sessions = %d, want 1", created) } } func TestOpencodeForwarderKeepsSessionWhenServerReturnsNoText(t *testing.T) { dir := t.TempDir() storePath := filepath.Join(dir, "opencode-sessions.json") ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") switch { case r.Method == http.MethodGet && r.URL.Path == "/global/health": _, _ = w.Write([]byte(`{"healthy":true,"version":"test"}`)) case r.Method == http.MethodPost && r.URL.Path == "/session/ses_keep/message": _, _ = w.Write([]byte(`{"parts":[]}`)) default: http.NotFound(w, r) } })) defer ts.Close() f := &opencodeForwarder{ bin: "opencode", timeout: 5 * time.Second, workDir: dir, sessions: newOpencodeSessions(storePath), server: &opencodeServer{baseURL: ts.URL, httpClient: ts.Client()}, } f.sessions.set("conv-1", "ses_keep") reply, err := f.forwardStream(context.Background(), "conv-1", "question", nil) if err != nil { t.Fatalf("forward with empty response: %v", err) } if reply != "(本地 agent 无文本输出)" { t.Fatalf("reply = %q, want no-text hint", reply) } if got := f.sessions.id("conv-1"); got != "ses_keep" { t.Fatalf("session after no-text response = %q, want ses_keep", got) } } func TestOpencodeForwarderOnlyReturnsTextParts(t *testing.T) { dir := t.TempDir() storePath := filepath.Join(dir, "opencode-sessions.json") ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") switch { case r.Method == http.MethodGet && r.URL.Path == "/global/health": _, _ = w.Write([]byte(`{"healthy":true,"version":"test"}`)) case r.Method == http.MethodPost && r.URL.Path == "/session": _, _ = w.Write([]byte(`{"id":"ses_parts"}`)) case r.Method == http.MethodPost && r.URL.Path == "/session/ses_parts/message": _, _ = w.Write([]byte(`{"parts":[{"type":"reasoning","text":"hidden reasoning"},{"type":"text","text":"visible"}]}`)) default: http.NotFound(w, r) } })) defer ts.Close() f := &opencodeForwarder{ bin: "opencode", timeout: 5 * time.Second, workDir: dir, sessions: newOpencodeSessions(storePath), server: &opencodeServer{baseURL: ts.URL, httpClient: ts.Client()}, } reply, err := f.forwardStream(context.Background(), "conv-1", "question", nil) if err != nil { t.Fatalf("forward: %v", err) } if reply != "visible" { t.Fatalf("reply = %q, want visible", reply) } } // TestOpencodeSessionsPersist verifies the convID→sessionID store survives a // restart and that reset is persisted. func TestOpencodeSessionsPersist(t *testing.T) { path := filepath.Join(t.TempDir(), "opencode-sessions.json") s := newOpencodeSessions(path) s.set("conv-1", "ses_a") s.set("conv-2", "ses_b") restarted := newOpencodeSessions(path) if got := restarted.id("conv-1"); got != "ses_a" { t.Errorf("after restart conv-1 = %q, want ses_a", got) } if got := restarted.id("conv-2"); got != "ses_b" { t.Errorf("after restart conv-2 = %q, want ses_b", got) } restarted.reset("conv-1") again := newOpencodeSessions(path) if got := again.id("conv-1"); got != "" { t.Errorf("after reset+restart conv-1 = %q, want empty", got) } if got := again.id("conv-2"); got != "ses_b" { t.Errorf("after reset+restart conv-2 = %q, want ses_b (untouched)", got) } } // TestNewOpencodeServerHasNoClientTimeout pins: the shared HTTP client // must not carry an overall deadline — long agent turns would be aborted // mid-flight. The per-turn ctx (f.timeout, default 0 = no limit) governs // only when the user explicitly sets --agent-timeout / DWS_AGENT_TIMEOUT_MS. func TestNewOpencodeServerHasNoClientTimeout(t *testing.T) { s := newOpencodeServer("opencode", nil, "") if s.httpClient == nil { t.Fatal("httpClient must be initialized") } if s.httpClient.Timeout != 0 { t.Fatalf("opencode http client Timeout = %s, want 0 (no client-level cap)", s.httpClient.Timeout) } } // TestOpencodeForwarderMessageGovernedByTurnCtx verifies a slow reply is not // cut by a fixed client timeout (the old 30s bug) and that the per-turn ctx is // the real governor instead. func TestOpencodeForwarderMessageGovernedByTurnCtx(t *testing.T) { dir := t.TempDir() const replyDelay = 200 * time.Millisecond ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") switch { case r.URL.Path == "/global/health": _, _ = w.Write([]byte(`{"healthy":true}`)) case r.URL.Path == "/session": _, _ = w.Write([]byte(`{"id":"ses_slow"}`)) case r.URL.Path == "/session/ses_slow/message": select { case <-time.After(replyDelay): case <-r.Context().Done(): return } _, _ = w.Write([]byte(`{"parts":[{"type":"text","text":"slow ok"}]}`)) default: http.NotFound(w, r) } })) defer ts.Close() newForwarder := func(turn time.Duration) *opencodeForwarder { return &opencodeForwarder{ bin: "opencode", timeout: turn, workDir: dir, sessions: newOpencodeSessions(filepath.Join(dir, "s.json")), // &http.Client{} mirrors the fixed production default (no overall cap). server: &opencodeServer{baseURL: ts.URL, httpClient: &http.Client{}}, } } // Generous turn budget: the slow reply must come back, not be cut by a cap. reply, err := newForwarder(5*time.Second).forwardStream(context.Background(), "conv-ok", "hi", nil) if err != nil { t.Fatalf("slow reply within the turn budget should succeed, got: %v", err) } if !strings.Contains(reply, "slow ok") { t.Fatalf("reply = %q, want it to contain slow ok", reply) } // Turn budget shorter than the reply delay: the per-turn ctx is what bounds // the round-trip, so this must error out (proving ctx, not a client cap, wins). if _, err := newForwarder(50*time.Millisecond).forwardStream(context.Background(), "conv-cut", "hi", nil); err == nil { t.Fatal("a reply slower than the turn ctx should fail") } } // TestOpencodeForwarderContextDeadlineExceededRepro reproduces the error seen // in the wild: user sends "run tech report", the opencode server processes for // longer than the user-configured turn timeout, and the context deadline kills // the POST /session/:id/message. The error must surface "context deadline // exceeded" so the higher-level formatter produces a meaningful reply. func TestOpencodeForwarderContextDeadlineExceededRepro(t *testing.T) { dir := t.TempDir() ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") switch { case r.URL.Path == "/global/health": _, _ = w.Write([]byte(`{"healthy":true}`)) case r.URL.Path == "/session": _, _ = w.Write([]byte(`{"id":"ses_1065c16edffeC46Oie6v9WY11l"}`)) case r.URL.Path == "/session/ses_1065c16edffeC46Oie6v9WY11l/message": // Simulate a long-running agent that never responds within the turn budget. // Fallback timeout ensures ts.Close() does not hang waiting on the handler. select { case <-r.Context().Done(): case <-time.After(1 * time.Second): } default: http.NotFound(w, r) } })) defer ts.Close() f := &opencodeForwarder{ bin: "opencode", timeout: 100 * time.Millisecond, // short turn budget to trigger deadline quickly workDir: dir, sessions: newOpencodeSessions(filepath.Join(dir, "s.json")), server: &opencodeServer{baseURL: ts.URL, httpClient: &http.Client{}}, } _, err := f.forwardStream(context.Background(), "conv-1", "run tech report", nil) if err == nil { t.Fatal("expected context deadline exceeded error, got nil") } errStr := err.Error() if !strings.Contains(errStr, "context deadline exceeded") { t.Fatalf("error = %q, want it to contain 'context deadline exceeded'", errStr) } if !strings.Contains(errStr, "/session/ses_1065c16edffeC46Oie6v9WY11l/message") { t.Fatalf("error = %q, want it to contain the session message path", errStr) } // Verify the higher-level format matches the screenshot. formatted := fmt.Sprintf("(opencode 调用失败:%v)", err) if !strings.Contains(formatted, "context deadline exceeded") { t.Fatalf("formatted reply = %q, want 'context deadline exceeded'", formatted) } if !strings.HasPrefix(formatted, "(opencode 调用失败:") { t.Fatalf("formatted reply = %q, want prefix '(opencode 调用失败:'", formatted) } } // opencodeForwarder must satisfy sessionClearer so /clear gets a real delete. var _ sessionClearer = (*opencodeForwarder)(nil) // TestOpencodeForwarderClearDeletesServerSession pins /clear semantics: it issues // a real DELETE /session/:id on the opencode server and drops the local mapping. func TestOpencodeForwarderClearDeletesServerSession(t *testing.T) { dir := t.TempDir() var deleted []string ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") switch { case r.URL.Path == "/global/health": _, _ = w.Write([]byte(`{"healthy":true}`)) case r.Method == http.MethodPost && r.URL.Path == "/session": _, _ = w.Write([]byte(`{"id":"ses_clear"}`)) case r.Method == http.MethodPost && r.URL.Path == "/session/ses_clear/message": _, _ = w.Write([]byte(`{"parts":[{"type":"text","text":"hi"}]}`)) case r.Method == http.MethodDelete && r.URL.Path == "/session/ses_clear": deleted = append(deleted, r.URL.Path) _, _ = w.Write([]byte(`{}`)) default: http.NotFound(w, r) } })) defer ts.Close() f := &opencodeForwarder{ bin: "opencode", timeout: 5 * time.Second, workDir: dir, sessions: newOpencodeSessions(filepath.Join(dir, "s.json")), server: &opencodeServer{baseURL: ts.URL, httpClient: ts.Client()}, } if _, err := f.forwardStream(context.Background(), "conv-1", "hi", nil); err != nil { t.Fatalf("forward: %v", err) } if got := f.sessions.id("conv-1"); got != "ses_clear" { t.Fatalf("session = %q, want ses_clear", got) } if err := f.clearSession(context.Background(), "conv-1"); err != nil { t.Fatalf("clearSession: %v", err) } if len(deleted) != 1 || deleted[0] != "/session/ses_clear" { t.Fatalf("server deletes = %v, want one DELETE /session/ses_clear", deleted) } if got := f.sessions.id("conv-1"); got != "" { t.Fatalf("session after /clear = %q, want empty", got) } } // TestOpencodeForwarderResetKeepsServerSession pins /new semantics: resetSession // only drops the local mapping and must NOT delete the server-side session, so // the old session stays resumable (the /new vs /clear distinction). func TestOpencodeForwarderResetKeepsServerSession(t *testing.T) { dir := t.TempDir() deletes := 0 ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") switch { case r.URL.Path == "/global/health": _, _ = w.Write([]byte(`{"healthy":true}`)) case r.Method == http.MethodPost && r.URL.Path == "/session": _, _ = w.Write([]byte(`{"id":"ses_keep"}`)) case r.Method == http.MethodPost && r.URL.Path == "/session/ses_keep/message": _, _ = w.Write([]byte(`{"parts":[{"type":"text","text":"hi"}]}`)) case r.Method == http.MethodDelete: deletes++ _, _ = w.Write([]byte(`{}`)) default: http.NotFound(w, r) } })) defer ts.Close() f := &opencodeForwarder{ bin: "opencode", timeout: 5 * time.Second, workDir: dir, sessions: newOpencodeSessions(filepath.Join(dir, "s.json")), server: &opencodeServer{baseURL: ts.URL, httpClient: ts.Client()}, } if _, err := f.forwardStream(context.Background(), "conv-1", "hi", nil); err != nil { t.Fatalf("forward: %v", err) } f.resetSession("conv-1") if deletes != 0 { t.Fatalf("resetSession issued %d DELETE(s), want 0 (old session must stay resumable)", deletes) } if got := f.sessions.id("conv-1"); got != "" { t.Fatalf("session after /new = %q, want empty", got) } }