777 lines
20 KiB
Go
777 lines
20 KiB
Go
package cli_test
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"strings"
|
|
"sync"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
"go.uber.org/zap"
|
|
|
|
clipkg "iop/apps/node/internal/adapters/cli"
|
|
"iop/apps/node/internal/adapters/cli/internal/testutil"
|
|
noderuntime "iop/apps/node/internal/runtime"
|
|
"iop/packages/go/config"
|
|
)
|
|
|
|
// opencodeFakeServer is a minimal stand-in for the OpenCode HTTP/SSE server.
|
|
// Tests push SSE events through ch; the SSE handler forwards them as
|
|
// `data: <json>\n\n` lines until the channel closes.
|
|
type opencodeFakeServer struct {
|
|
mu sync.Mutex
|
|
events chan map[string]any
|
|
abortCalls atomic.Int32
|
|
promptCalls atomic.Int32
|
|
createCalls atomic.Int32
|
|
lastPromptBody []byte
|
|
createdID string
|
|
}
|
|
|
|
func newOpencodeFakeServer(t *testing.T, sessionID string) (*opencodeFakeServer, *httptest.Server) {
|
|
t.Helper()
|
|
s := &opencodeFakeServer{
|
|
events: make(chan map[string]any, 32),
|
|
createdID: sessionID,
|
|
}
|
|
mux := http.NewServeMux()
|
|
eventHandler := func(w http.ResponseWriter, r *http.Request) {
|
|
w.Header().Set("Content-Type", "text/event-stream")
|
|
w.Header().Set("Cache-Control", "no-cache")
|
|
w.WriteHeader(http.StatusOK)
|
|
flusher, _ := w.(http.Flusher)
|
|
if flusher != nil {
|
|
flusher.Flush()
|
|
}
|
|
for {
|
|
select {
|
|
case <-r.Context().Done():
|
|
return
|
|
case ev, ok := <-s.events:
|
|
if !ok {
|
|
return
|
|
}
|
|
b, err := json.Marshal(ev)
|
|
if err != nil {
|
|
continue
|
|
}
|
|
fmt.Fprintf(w, "data: %s\n\n", string(b))
|
|
if flusher != nil {
|
|
flusher.Flush()
|
|
}
|
|
}
|
|
}
|
|
}
|
|
mux.HandleFunc("/event", eventHandler)
|
|
mux.HandleFunc("/global/event", eventHandler)
|
|
mux.HandleFunc("/session", func(w http.ResponseWriter, r *http.Request) {
|
|
s.createCalls.Add(1)
|
|
w.Header().Set("Content-Type", "application/json")
|
|
_ = json.NewEncoder(w).Encode(map[string]any{"id": s.createdID})
|
|
})
|
|
mux.HandleFunc("/session/", func(w http.ResponseWriter, r *http.Request) {
|
|
// Routes: /session/{id}/prompt_async or /session/{id}/abort
|
|
path := strings.TrimPrefix(r.URL.Path, "/session/")
|
|
parts := strings.SplitN(path, "/", 2)
|
|
if len(parts) != 2 {
|
|
http.NotFound(w, r)
|
|
return
|
|
}
|
|
switch parts[1] {
|
|
case "prompt_async":
|
|
s.promptCalls.Add(1)
|
|
buf := make([]byte, 4096)
|
|
n, _ := r.Body.Read(buf)
|
|
s.mu.Lock()
|
|
s.lastPromptBody = append([]byte(nil), buf[:n]...)
|
|
s.mu.Unlock()
|
|
w.WriteHeader(http.StatusAccepted)
|
|
case "abort":
|
|
s.abortCalls.Add(1)
|
|
w.WriteHeader(http.StatusOK)
|
|
default:
|
|
http.NotFound(w, r)
|
|
}
|
|
})
|
|
mux.HandleFunc("/permission/", func(w http.ResponseWriter, r *http.Request) {
|
|
w.WriteHeader(http.StatusOK)
|
|
})
|
|
srv := httptest.NewServer(mux)
|
|
t.Cleanup(func() {
|
|
s.mu.Lock()
|
|
if s.events != nil {
|
|
close(s.events)
|
|
s.events = nil
|
|
}
|
|
s.mu.Unlock()
|
|
srv.Close()
|
|
})
|
|
return s, srv
|
|
}
|
|
|
|
func (s *opencodeFakeServer) push(ev map[string]any) {
|
|
s.mu.Lock()
|
|
ch := s.events
|
|
s.mu.Unlock()
|
|
if ch != nil {
|
|
ch <- ev
|
|
}
|
|
}
|
|
|
|
func opencodeGlobalEvent(eventType string, props map[string]any) map[string]any {
|
|
return map[string]any{
|
|
"directory": "/tmp/iop-test",
|
|
"payload": map[string]any{
|
|
"id": "evt_test",
|
|
"type": eventType,
|
|
"properties": props,
|
|
},
|
|
}
|
|
}
|
|
|
|
func opencodeSSEProfile(attachURL string, opts ...string) config.CLIProfileConf {
|
|
args := []string{"--attach", attachURL, "--title", "test"}
|
|
args = append(args, opts...)
|
|
return config.CLIProfileConf{
|
|
Command: "/bin/true",
|
|
Args: args,
|
|
Mode: "opencode-sse",
|
|
}
|
|
}
|
|
|
|
func TestCLIExecuteOpencodeSSE_StreamsTextDeltas(t *testing.T) {
|
|
fake, srv := newOpencodeFakeServer(t, "ses_1")
|
|
|
|
cfg := config.CLIConf{
|
|
Enabled: true,
|
|
Profiles: map[string]config.CLIProfileConf{
|
|
"opencode": opencodeSSEProfile(srv.URL, "--model", "prov/m1", "--dangerously-skip-permissions"),
|
|
},
|
|
}
|
|
c := clipkg.New(cfg, zap.NewNop())
|
|
sink := &testutil.FakeSink{}
|
|
|
|
done := make(chan error, 1)
|
|
go func() {
|
|
done <- c.Execute(context.Background(), noderuntime.ExecutionSpec{
|
|
RunID: "run-1",
|
|
Target: "opencode",
|
|
Input: map[string]any{"prompt": "hello"},
|
|
}, sink)
|
|
}()
|
|
|
|
// Wait until prompt_async has been received, then push events.
|
|
deadline := time.Now().Add(2 * time.Second)
|
|
for fake.promptCalls.Load() == 0 && time.Now().Before(deadline) {
|
|
time.Sleep(5 * time.Millisecond)
|
|
}
|
|
if fake.promptCalls.Load() == 0 {
|
|
t.Fatal("prompt_async was not called")
|
|
}
|
|
|
|
fake.push(map[string]any{
|
|
"type": "session.next.text.delta",
|
|
"properties": map[string]any{
|
|
"sessionID": "ses_1",
|
|
"delta": "Hello ",
|
|
},
|
|
})
|
|
fake.push(map[string]any{
|
|
"type": "session.next.text.delta",
|
|
"properties": map[string]any{
|
|
"sessionID": "ses_1",
|
|
"delta": "world",
|
|
},
|
|
})
|
|
fake.push(map[string]any{
|
|
"type": "session.next.step.ended",
|
|
"properties": map[string]any{
|
|
"sessionID": "ses_1",
|
|
"tokens": map[string]any{"input": 5.0, "output": 2.0},
|
|
},
|
|
})
|
|
fake.push(map[string]any{
|
|
"type": "session.idle",
|
|
"properties": map[string]any{
|
|
"sessionID": "ses_1",
|
|
},
|
|
})
|
|
|
|
select {
|
|
case err := <-done:
|
|
if err != nil {
|
|
t.Fatalf("execute: %v", err)
|
|
}
|
|
case <-time.After(3 * time.Second):
|
|
t.Fatal("execute did not return")
|
|
}
|
|
|
|
events := sink.Events()
|
|
if combined := testutil.CollectDeltas(events); combined != "Hello world" {
|
|
t.Fatalf("deltas: got %q, want %q", combined, "Hello world")
|
|
}
|
|
var complete *noderuntime.RuntimeEvent
|
|
for i := range events {
|
|
if events[i].Type == noderuntime.EventTypeComplete {
|
|
complete = &events[i]
|
|
}
|
|
}
|
|
if complete == nil {
|
|
t.Fatal("expected complete event")
|
|
}
|
|
if complete.Usage == nil || complete.Usage.InputTokens != 5 || complete.Usage.OutputTokens != 2 {
|
|
t.Fatalf("expected usage 5/2, got %+v", complete.Usage)
|
|
}
|
|
if fake.createCalls.Load() != 1 {
|
|
t.Errorf("expected 1 create session call, got %d", fake.createCalls.Load())
|
|
}
|
|
|
|
fake.mu.Lock()
|
|
body := append([]byte(nil), fake.lastPromptBody...)
|
|
fake.mu.Unlock()
|
|
var decoded struct {
|
|
Parts []struct {
|
|
Type string `json:"type"`
|
|
Text string `json:"text"`
|
|
} `json:"parts"`
|
|
Model struct {
|
|
ProviderID string `json:"providerID"`
|
|
ModelID string `json:"modelID"`
|
|
} `json:"model"`
|
|
}
|
|
if err := json.Unmarshal(body, &decoded); err != nil {
|
|
t.Fatalf("decode prompt body: %v (body=%q)", err, string(body))
|
|
}
|
|
if len(decoded.Parts) != 1 || decoded.Parts[0].Type != "text" || decoded.Parts[0].Text != "hello" {
|
|
t.Errorf("parts: got %+v", decoded.Parts)
|
|
}
|
|
if decoded.Model.ProviderID != "prov" || decoded.Model.ModelID != "m1" {
|
|
t.Errorf("model: got %+v", decoded.Model)
|
|
}
|
|
}
|
|
|
|
func TestCLIExecuteOpencodeSSE_GlobalMessagePartDeltaStreamsTextOnly(t *testing.T) {
|
|
fake, srv := newOpencodeFakeServer(t, "ses_global")
|
|
|
|
cfg := config.CLIConf{
|
|
Enabled: true,
|
|
Profiles: map[string]config.CLIProfileConf{
|
|
"opencode": opencodeSSEProfile(srv.URL),
|
|
},
|
|
}
|
|
c := clipkg.New(cfg, zap.NewNop())
|
|
sink := &testutil.FakeSink{}
|
|
|
|
done := make(chan error, 1)
|
|
go func() {
|
|
done <- c.Execute(context.Background(), noderuntime.ExecutionSpec{
|
|
RunID: "run-global",
|
|
Target: "opencode",
|
|
Input: map[string]any{"prompt": "hello"},
|
|
}, sink)
|
|
}()
|
|
|
|
deadline := time.Now().Add(2 * time.Second)
|
|
for fake.promptCalls.Load() == 0 && time.Now().Before(deadline) {
|
|
time.Sleep(5 * time.Millisecond)
|
|
}
|
|
if fake.promptCalls.Load() == 0 {
|
|
t.Fatal("prompt_async was not called")
|
|
}
|
|
|
|
fake.push(opencodeGlobalEvent("message.part.updated", map[string]any{
|
|
"sessionID": "ses_global",
|
|
"part": map[string]any{"id": "prt_reason", "type": "reasoning", "text": ""},
|
|
}))
|
|
fake.push(opencodeGlobalEvent("message.part.delta", map[string]any{
|
|
"sessionID": "ses_global",
|
|
"partID": "prt_reason",
|
|
"field": "text",
|
|
"delta": "thinking",
|
|
}))
|
|
fake.push(opencodeGlobalEvent("message.part.updated", map[string]any{
|
|
"sessionID": "ses_global",
|
|
"part": map[string]any{"id": "prt_text", "type": "text", "text": ""},
|
|
}))
|
|
fake.push(opencodeGlobalEvent("message.part.delta", map[string]any{
|
|
"sessionID": "ses_global",
|
|
"partID": "prt_text",
|
|
"field": "text",
|
|
"delta": "OK",
|
|
}))
|
|
fake.push(opencodeGlobalEvent("message.part.updated", map[string]any{
|
|
"sessionID": "ses_global",
|
|
"part": map[string]any{
|
|
"id": "prt_finish",
|
|
"type": "step-finish",
|
|
"tokens": map[string]any{"input": 7.0, "output": 1.0},
|
|
},
|
|
}))
|
|
fake.push(opencodeGlobalEvent("session.status", map[string]any{
|
|
"sessionID": "ses_global",
|
|
"status": map[string]any{"type": "idle"},
|
|
}))
|
|
|
|
select {
|
|
case err := <-done:
|
|
if err != nil {
|
|
t.Fatalf("execute: %v", err)
|
|
}
|
|
case <-time.After(3 * time.Second):
|
|
t.Fatal("execute did not return")
|
|
}
|
|
|
|
events := sink.Events()
|
|
if combined := testutil.CollectDeltas(events); combined != "OK" {
|
|
t.Fatalf("deltas: got %q, want %q", combined, "OK")
|
|
}
|
|
var complete *noderuntime.RuntimeEvent
|
|
for i := range events {
|
|
if events[i].Type == noderuntime.EventTypeComplete {
|
|
complete = &events[i]
|
|
}
|
|
}
|
|
if complete == nil {
|
|
t.Fatal("expected complete event")
|
|
}
|
|
if complete.Usage == nil || complete.Usage.InputTokens != 7 || complete.Usage.OutputTokens != 1 {
|
|
t.Fatalf("expected usage 7/1, got %+v", complete.Usage)
|
|
}
|
|
}
|
|
|
|
func TestCLIExecuteOpencodeSSE_PromptAsyncSendsModelObject(t *testing.T) {
|
|
fake, srv := newOpencodeFakeServer(t, "ses_m")
|
|
|
|
cfg := config.CLIConf{
|
|
Enabled: true,
|
|
Profiles: map[string]config.CLIProfileConf{
|
|
"opencode": opencodeSSEProfile(srv.URL, "--model", "ollama-dgx/qwen3.6:35b-a3b-bf16"),
|
|
},
|
|
}
|
|
c := clipkg.New(cfg, zap.NewNop())
|
|
sink := &testutil.FakeSink{}
|
|
|
|
done := make(chan error, 1)
|
|
go func() {
|
|
done <- c.Execute(context.Background(), noderuntime.ExecutionSpec{
|
|
RunID: "run-m",
|
|
Target: "opencode",
|
|
Input: map[string]any{"prompt": "hello"},
|
|
}, sink)
|
|
}()
|
|
|
|
deadline := time.Now().Add(2 * time.Second)
|
|
for fake.promptCalls.Load() == 0 && time.Now().Before(deadline) {
|
|
time.Sleep(5 * time.Millisecond)
|
|
}
|
|
if fake.promptCalls.Load() == 0 {
|
|
t.Fatal("prompt_async was not called")
|
|
}
|
|
fake.push(map[string]any{
|
|
"type": "session.idle",
|
|
"properties": map[string]any{"sessionID": "ses_m"},
|
|
})
|
|
select {
|
|
case err := <-done:
|
|
if err != nil {
|
|
t.Fatalf("execute: %v", err)
|
|
}
|
|
case <-time.After(3 * time.Second):
|
|
t.Fatal("execute did not return")
|
|
}
|
|
|
|
fake.mu.Lock()
|
|
body := append([]byte(nil), fake.lastPromptBody...)
|
|
fake.mu.Unlock()
|
|
var decoded struct {
|
|
Model map[string]any `json:"model"`
|
|
}
|
|
if err := json.Unmarshal(body, &decoded); err != nil {
|
|
t.Fatalf("decode body: %v (body=%q)", err, string(body))
|
|
}
|
|
if decoded.Model["providerID"] != "ollama-dgx" {
|
|
t.Errorf("providerID: got %v", decoded.Model["providerID"])
|
|
}
|
|
if decoded.Model["modelID"] != "qwen3.6:35b-a3b-bf16" {
|
|
t.Errorf("modelID: got %v", decoded.Model["modelID"])
|
|
}
|
|
}
|
|
|
|
func TestCLIExecuteOpencodeSSE_InvalidModelEmitsError(t *testing.T) {
|
|
_, srv := newOpencodeFakeServer(t, "ses_bad")
|
|
|
|
cfg := config.CLIConf{
|
|
Enabled: true,
|
|
Profiles: map[string]config.CLIProfileConf{
|
|
"opencode": opencodeSSEProfile(srv.URL, "--model", "missing-slash"),
|
|
},
|
|
}
|
|
c := clipkg.New(cfg, zap.NewNop())
|
|
sink := &testutil.FakeSink{}
|
|
|
|
err := c.Execute(context.Background(), noderuntime.ExecutionSpec{
|
|
RunID: "run-bad",
|
|
Target: "opencode",
|
|
Input: map[string]any{"prompt": "hi"},
|
|
}, sink)
|
|
if err == nil {
|
|
t.Fatal("expected error for invalid model, got nil")
|
|
}
|
|
if !strings.Contains(err.Error(), "missing provider/model separator") {
|
|
t.Fatalf("unexpected error: %v", err)
|
|
}
|
|
var sawErr bool
|
|
for _, e := range sink.Events() {
|
|
if e.Type == noderuntime.EventTypeError && strings.Contains(e.Error, "missing provider/model separator") {
|
|
sawErr = true
|
|
}
|
|
}
|
|
if !sawErr {
|
|
t.Fatal("expected error event with model parse failure")
|
|
}
|
|
}
|
|
|
|
func TestCLIExecuteOpencodeSSE_SessionStatusObjectIdleCompletes(t *testing.T) {
|
|
fake, srv := newOpencodeFakeServer(t, "ses_obj")
|
|
|
|
cfg := config.CLIConf{
|
|
Enabled: true,
|
|
Profiles: map[string]config.CLIProfileConf{
|
|
"opencode": opencodeSSEProfile(srv.URL),
|
|
},
|
|
}
|
|
c := clipkg.New(cfg, zap.NewNop())
|
|
sink := &testutil.FakeSink{}
|
|
|
|
done := make(chan error, 1)
|
|
go func() {
|
|
done <- c.Execute(context.Background(), noderuntime.ExecutionSpec{
|
|
RunID: "run-obj",
|
|
Target: "opencode",
|
|
Input: map[string]any{"prompt": "hi"},
|
|
}, sink)
|
|
}()
|
|
|
|
deadline := time.Now().Add(2 * time.Second)
|
|
for fake.promptCalls.Load() == 0 && time.Now().Before(deadline) {
|
|
time.Sleep(5 * time.Millisecond)
|
|
}
|
|
|
|
fake.push(map[string]any{
|
|
"type": "session.next.text.delta",
|
|
"properties": map[string]any{
|
|
"sessionID": "ses_obj",
|
|
"delta": "done",
|
|
},
|
|
})
|
|
fake.push(map[string]any{
|
|
"type": "session.status",
|
|
"properties": map[string]any{
|
|
"sessionID": "ses_obj",
|
|
"status": map[string]any{"type": "idle"},
|
|
},
|
|
})
|
|
|
|
select {
|
|
case err := <-done:
|
|
if err != nil {
|
|
t.Fatalf("execute: %v", err)
|
|
}
|
|
case <-time.After(3 * time.Second):
|
|
t.Fatal("execute did not return")
|
|
}
|
|
var sawComplete bool
|
|
for _, e := range sink.Events() {
|
|
if e.Type == noderuntime.EventTypeComplete {
|
|
sawComplete = true
|
|
}
|
|
}
|
|
if !sawComplete {
|
|
t.Fatal("expected complete event after object-idle session.status")
|
|
}
|
|
}
|
|
|
|
func TestCLIExecuteOpencodeSSE_RequireExistingWithoutSessionErrors(t *testing.T) {
|
|
_, srv := newOpencodeFakeServer(t, "ses_1")
|
|
|
|
cfg := config.CLIConf{
|
|
Enabled: true,
|
|
Profiles: map[string]config.CLIProfileConf{
|
|
"opencode": opencodeSSEProfile(srv.URL),
|
|
},
|
|
}
|
|
c := clipkg.New(cfg, zap.NewNop())
|
|
sink := &testutil.FakeSink{}
|
|
|
|
err := c.Execute(context.Background(), noderuntime.ExecutionSpec{
|
|
RunID: "run-x",
|
|
Target: "opencode",
|
|
SessionID: "abs",
|
|
SessionMode: noderuntime.SessionModeRequireExisting,
|
|
Input: map[string]any{"prompt": "hi"},
|
|
}, sink)
|
|
if err == nil {
|
|
t.Fatal("expected error when SessionModeRequireExisting and no cached session")
|
|
}
|
|
if !strings.Contains(err.Error(), "no persistent session") {
|
|
t.Fatalf("unexpected error: %v", err)
|
|
}
|
|
}
|
|
|
|
func TestCLIExecuteOpencodeSSE_SessionErrorEmitsRuntimeError(t *testing.T) {
|
|
fake, srv := newOpencodeFakeServer(t, "ses_1")
|
|
|
|
cfg := config.CLIConf{
|
|
Enabled: true,
|
|
Profiles: map[string]config.CLIProfileConf{
|
|
"opencode": opencodeSSEProfile(srv.URL),
|
|
},
|
|
}
|
|
c := clipkg.New(cfg, zap.NewNop())
|
|
sink := &testutil.FakeSink{}
|
|
|
|
done := make(chan error, 1)
|
|
go func() {
|
|
done <- c.Execute(context.Background(), noderuntime.ExecutionSpec{
|
|
RunID: "run-err",
|
|
Target: "opencode",
|
|
Input: map[string]any{"prompt": "hi"},
|
|
}, sink)
|
|
}()
|
|
|
|
deadline := time.Now().Add(2 * time.Second)
|
|
for fake.promptCalls.Load() == 0 && time.Now().Before(deadline) {
|
|
time.Sleep(5 * time.Millisecond)
|
|
}
|
|
|
|
fake.push(map[string]any{
|
|
"type": "session.error",
|
|
"properties": map[string]any{
|
|
"sessionID": "ses_1",
|
|
"error": map[string]any{
|
|
"name": "ModelError",
|
|
"data": map[string]any{"message": "Model not found: foo/bar"},
|
|
},
|
|
},
|
|
})
|
|
|
|
select {
|
|
case err := <-done:
|
|
if err == nil {
|
|
t.Fatal("expected error, got nil")
|
|
}
|
|
if !strings.Contains(err.Error(), "Model not found: foo/bar") {
|
|
t.Fatalf("unexpected error: %v", err)
|
|
}
|
|
case <-time.After(3 * time.Second):
|
|
t.Fatal("execute did not return")
|
|
}
|
|
|
|
var gotErr string
|
|
for _, e := range sink.Events() {
|
|
if e.Type == noderuntime.EventTypeError {
|
|
gotErr = e.Error
|
|
}
|
|
}
|
|
if gotErr != "Model not found: foo/bar" {
|
|
t.Fatalf("expected runtime error event, got %q", gotErr)
|
|
}
|
|
}
|
|
|
|
func TestCLIExecuteOpencodeSSE_ConsecutiveExecutesReuseSession(t *testing.T) {
|
|
fake, srv := newOpencodeFakeServer(t, "ses_reuse")
|
|
|
|
cfg := config.CLIConf{
|
|
Enabled: true,
|
|
Profiles: map[string]config.CLIProfileConf{
|
|
"opencode": opencodeSSEProfile(srv.URL),
|
|
},
|
|
}
|
|
c := clipkg.New(cfg, zap.NewNop())
|
|
|
|
waitPromptCalls := func(t *testing.T, want int32) {
|
|
t.Helper()
|
|
deadline := time.Now().Add(2 * time.Second)
|
|
for fake.promptCalls.Load() < want && time.Now().Before(deadline) {
|
|
time.Sleep(5 * time.Millisecond)
|
|
}
|
|
if fake.promptCalls.Load() < want {
|
|
t.Fatalf("prompt_async call count: got %d, want >= %d", fake.promptCalls.Load(), want)
|
|
}
|
|
}
|
|
|
|
// --- First Execute ---
|
|
sink1 := &testutil.FakeSink{}
|
|
done1 := make(chan error, 1)
|
|
go func() {
|
|
done1 <- c.Execute(context.Background(), noderuntime.ExecutionSpec{
|
|
RunID: "run-seq1",
|
|
Target: "opencode",
|
|
Input: map[string]any{"prompt": "first"},
|
|
}, sink1)
|
|
}()
|
|
|
|
waitPromptCalls(t, 1)
|
|
fake.push(map[string]any{
|
|
"type": "session.next.text.delta",
|
|
"properties": map[string]any{
|
|
"sessionID": "ses_reuse",
|
|
"delta": "IOP_E2E_OPENCODE_ONE",
|
|
},
|
|
})
|
|
fake.push(map[string]any{
|
|
"type": "session.idle",
|
|
"properties": map[string]any{"sessionID": "ses_reuse"},
|
|
})
|
|
|
|
select {
|
|
case err := <-done1:
|
|
if err != nil {
|
|
t.Fatalf("first execute: %v", err)
|
|
}
|
|
case <-time.After(3 * time.Second):
|
|
t.Fatal("first execute did not return")
|
|
}
|
|
|
|
if combined := testutil.CollectDeltas(sink1.Events()); combined != "IOP_E2E_OPENCODE_ONE" {
|
|
t.Errorf("run1 deltas: got %q, want %q", combined, "IOP_E2E_OPENCODE_ONE")
|
|
}
|
|
if fake.createCalls.Load() != 1 {
|
|
t.Errorf("after run1: expected 1 createSession call, got %d", fake.createCalls.Load())
|
|
}
|
|
|
|
// --- Second Execute on same CLI / same session ---
|
|
sink2 := &testutil.FakeSink{}
|
|
done2 := make(chan error, 1)
|
|
go func() {
|
|
done2 <- c.Execute(context.Background(), noderuntime.ExecutionSpec{
|
|
RunID: "run-seq2",
|
|
Target: "opencode",
|
|
Input: map[string]any{"prompt": "second"},
|
|
}, sink2)
|
|
}()
|
|
|
|
waitPromptCalls(t, 2)
|
|
fake.push(map[string]any{
|
|
"type": "session.next.text.delta",
|
|
"properties": map[string]any{
|
|
"sessionID": "ses_reuse",
|
|
"delta": "IOP_E2E_OPENCODE_TWO",
|
|
},
|
|
})
|
|
fake.push(map[string]any{
|
|
"type": "session.idle",
|
|
"properties": map[string]any{"sessionID": "ses_reuse"},
|
|
})
|
|
|
|
select {
|
|
case err := <-done2:
|
|
if err != nil {
|
|
t.Fatalf("second execute: %v", err)
|
|
}
|
|
case <-time.After(3 * time.Second):
|
|
t.Fatal("second execute did not return")
|
|
}
|
|
|
|
// Session was reused: still 1 create call, 2 prompt calls.
|
|
if fake.createCalls.Load() != 1 {
|
|
t.Errorf("after run2: expected 1 createSession call (reused), got %d", fake.createCalls.Load())
|
|
}
|
|
if fake.promptCalls.Load() != 2 {
|
|
t.Errorf("expected 2 prompt_async calls, got %d", fake.promptCalls.Load())
|
|
}
|
|
|
|
// run2 sink must have the second delta and a complete, not the first delta.
|
|
if combined := testutil.CollectDeltas(sink2.Events()); combined != "IOP_E2E_OPENCODE_TWO" {
|
|
t.Errorf("run2 deltas: got %q, want %q", combined, "IOP_E2E_OPENCODE_TWO")
|
|
}
|
|
for _, e := range sink2.Events() {
|
|
if e.Type == noderuntime.EventTypeDelta && e.Delta == "IOP_E2E_OPENCODE_ONE" {
|
|
t.Error("run2 sink must not contain run1 delta")
|
|
}
|
|
}
|
|
var run1Complete, run2Complete bool
|
|
for _, e := range sink1.Events() {
|
|
if e.Type == noderuntime.EventTypeComplete {
|
|
run1Complete = true
|
|
}
|
|
}
|
|
for _, e := range sink2.Events() {
|
|
if e.Type == noderuntime.EventTypeComplete {
|
|
run2Complete = true
|
|
}
|
|
}
|
|
if !run1Complete {
|
|
t.Error("run1: expected complete event")
|
|
}
|
|
if !run2Complete {
|
|
t.Error("run2: expected complete event")
|
|
}
|
|
}
|
|
|
|
func TestCLIExecuteOpencodeSSE_ContextCancelAbortsSession(t *testing.T) {
|
|
fake, srv := newOpencodeFakeServer(t, "ses_1")
|
|
|
|
cfg := config.CLIConf{
|
|
Enabled: true,
|
|
Profiles: map[string]config.CLIProfileConf{
|
|
"opencode": opencodeSSEProfile(srv.URL),
|
|
},
|
|
}
|
|
c := clipkg.New(cfg, zap.NewNop())
|
|
sink := &testutil.FakeSink{}
|
|
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
done := make(chan error, 1)
|
|
go func() {
|
|
done <- c.Execute(ctx, noderuntime.ExecutionSpec{
|
|
RunID: "run-cancel",
|
|
Target: "opencode",
|
|
Input: map[string]any{"prompt": "hi"},
|
|
}, sink)
|
|
}()
|
|
|
|
deadline := time.Now().Add(2 * time.Second)
|
|
for fake.promptCalls.Load() == 0 && time.Now().Before(deadline) {
|
|
time.Sleep(5 * time.Millisecond)
|
|
}
|
|
cancel()
|
|
|
|
select {
|
|
case err := <-done:
|
|
if err != noderuntime.ErrRunCancelled {
|
|
t.Fatalf("expected ErrRunCancelled, got %v", err)
|
|
}
|
|
case <-time.After(3 * time.Second):
|
|
t.Fatal("execute did not return after cancel")
|
|
}
|
|
|
|
var cancelEvent bool
|
|
var msg string
|
|
for _, e := range sink.Events() {
|
|
if e.Type == noderuntime.EventTypeCancelled {
|
|
cancelEvent = true
|
|
msg = e.Message
|
|
}
|
|
}
|
|
if !cancelEvent {
|
|
t.Fatal("expected cancelled event")
|
|
}
|
|
if msg != "user-cancel" {
|
|
t.Fatalf("expected user-cancel, got %q", msg)
|
|
}
|
|
|
|
// Allow a short window for the abort POST to arrive.
|
|
deadline = time.Now().Add(1 * time.Second)
|
|
for fake.abortCalls.Load() == 0 && time.Now().Before(deadline) {
|
|
time.Sleep(10 * time.Millisecond)
|
|
}
|
|
if fake.abortCalls.Load() == 0 {
|
|
t.Fatal("expected /session/{id}/abort to be called")
|
|
}
|
|
}
|