fix: address issues with processing large events

Signed-off-by: Donnie Adams <[email protected]>

Author: thedadams

client.go

@@ -171,7 +171,7 @@ func (c *Client) runBasicCommand(ctx context.Context, command, requestPath, tool
 	if run.url != "" {
 		var m any
 		if content != "" || toolPath != "" {
-			m = map[string]any{"input": content, "file": toolPath}
+			m = map[string]any{"content": content, "file": toolPath}
 		}
 		err = run.request(ctx, m)
 	} else {

client_test.go

@@ -567,6 +567,57 @@ func TestFileChat(t *testing.T) {
 	}
 }
 
+func TestToolWithGlobalTools(t *testing.T) {
+	var runStartSeen, callStartSeen, callFinishSeen, callProgressSeen, runFinishSeen bool
+	wd, err := os.Getwd()
+	if err != nil {
+		t.Fatalf("Error getting current working directory: %v", err)
+	}
+
+	var eventContent string
+
+	run, err := client.Run(context.Background(), wd+"/test/global-tools.gpt", Opts{DisableCache: true, IncludeEvents: true})
+	if err != nil {
+		t.Errorf("Error executing tool: %v", err)
+	}
+
+	for e := range run.Events() {
+		if e.Type == EventTypeRunStart {
+			runStartSeen = true
+		} else if e.Type == EventTypeCallStart {
+			callStartSeen = true
+		} else if e.Type == EventTypeCallFinish {
+			callFinishSeen = true
+		} else if e.Type == EventTypeRunFinish {
+			runFinishSeen = true
+		} else if e.Type == EventTypeCallProgress {
+			callProgressSeen = true
+		}
+		eventContent += e.Content
+	}
+
+	out, err := run.Text()
+	if err != nil {
+		t.Errorf("Error reading output: %v", err)
+	}
+
+	if !strings.Contains(eventContent, "Hello") {
+		t.Errorf("Unexpected event output: %s", eventContent)
+	}
+
+	if !strings.Contains(out, "Hello!") {
+		t.Errorf("Unexpected output: %s", out)
+	}
+
+	if len(run.ErrorOutput()) == 0 {
+		t.Error("No stderr output")
+	}
+
+	if !runStartSeen || !callStartSeen || !callFinishSeen || !runFinishSeen || !callProgressSeen {
+		t.Errorf("Missing events: %t %t %t %t %t", runStartSeen, callStartSeen, callFinishSeen, runFinishSeen, callProgressSeen)
+	}
+}
+
 func TestGetCommand(t *testing.T) {
 	currentEnvVar := os.Getenv("GPTSCRIPT_BIN")
 	t.Cleanup(func() {

run.go

@@ -215,34 +215,48 @@ func (r *Run) exec(ctx context.Context, extraArgs ...string) error {
 func (r *Run) readEvents(ctx context.Context, events io.Reader) {
 	defer close(r.events)
 
-	scan := bufio.NewScanner(events)
-	for scan.Scan() {
-		if !r.opts.IncludeEvents {
-			continue
-		}
+	var (
+		n    int
+		err  error
+		frag []byte
 
-		line := scan.Bytes()
-		if len(line) == 0 {
-			continue
+		b = make([]byte, 64*1024)
+	)
+	for ; ; n, err = events.Read(b) {
+		if n == 0 && err != nil {
+			break
 		}
 
-		var event Event
-		if err := json.Unmarshal(line, &event); err != nil {
-			slog.Debug("failed to unmarshal event", "error", err, "event", string(line))
+		if !r.opts.IncludeEvents {
 			continue
 		}
 
-		select {
-		case <-ctx.Done():
-			go func() {
-				for scan.Scan() {
-					// Drain any remaining events
-				}
-			}()
-			return
-		case r.events <- event:
+		for _, line := range bytes.Split(append(frag, b[:n]...), []byte("\n\n")) {
+			if line = bytes.TrimSpace(line); len(line) == 0 {
+				frag = frag[:0]
+				continue
+			}
+
+			var event Event
+			if err := json.Unmarshal(line, &event); err != nil {
+				slog.Debug("failed to unmarshal event", "error", err, "event", string(b))
+				frag = line[:]
+				continue
+			}
+
+			select {
+			case <-ctx.Done():
+				return
+			case r.events <- event:
+				frag = frag[:0]
+			}
 		}
 	}
+
+	if err != nil && !errors.Is(err, io.EOF) {
+		slog.Debug("failed to read events", "error", err)
+		r.err = fmt.Errorf("failed to read events: %w", err)
+	}
 }
 
 func (r *Run) readAllOutput() error {
@@ -367,16 +381,14 @@ func (r *Run) request(ctx context.Context, payload any) (err error) {
 	go r.readEvents(cancelCtx, eventsRead)
 
 	go func() {
+		var (
+			n    int
+			frag []byte
+			buf  = make([]byte, 64*1024)
+		)
 		bufferedStdout := bufio.NewWriter(stdoutWriter)
 		bufferedStderr := bufio.NewWriter(stderrWriter)
-		scan := bufio.NewScanner(resp.Body)
 		defer func() {
-			go func() {
-				for scan.Scan() {
-					// Drain any remaining events
-				}
-			}()
-
 			eventsWrite.Close()
 
 			bufferedStderr.Flush()
@@ -390,38 +402,59 @@ func (r *Run) request(ctx context.Context, payload any) (err error) {
 			resp.Body.Close()
 		}()
 
-		for scan.Scan() {
-			line := bytes.TrimSpace(bytes.TrimPrefix(scan.Bytes(), []byte("data: ")))
-			if len(line) == 0 {
-				continue
-			}
-			if bytes.Equal(line, []byte("[DONE]")) {
-				return
+		for ; ; n, err = resp.Body.Read(buf) {
+			if n == 0 && err != nil {
+				break
 			}
 
-			if bytes.HasPrefix(line, []byte(`{"stdout":`)) {
-				_, err = bufferedStdout.Write(bytes.TrimSuffix(bytes.TrimPrefix(line, []byte(`{"stdout":`)), []byte("}")))
-				if err != nil {
-					r.state = Error
-					r.err = fmt.Errorf("failed to write stdout: %w", err)
-					return
+			for _, line := range bytes.Split(bytes.TrimSpace(append(frag, buf[:n]...)), []byte("\n\n")) {
+				line = bytes.TrimSpace(bytes.TrimPrefix(line, []byte("data: ")))
+				if len(line) == 0 {
+					frag = frag[:0]
+					continue
 				}
-			} else if bytes.HasPrefix(line, []byte(`{"stderr":`)) {
-				_, err = bufferedStderr.Write(bytes.TrimSuffix(bytes.TrimPrefix(line, []byte(`{"stderr":`)), []byte("}")))
-				if err != nil {
-					r.state = Error
-					r.err = fmt.Errorf("failed to write stderr: %w", err)
+				if bytes.Equal(line, []byte("[DONE]")) {
 					return
 				}
-			} else {
-				_, err = eventsWrite.Write(append(line, '\n'))
-				if err != nil {
-					r.state = Error
-					r.err = fmt.Errorf("failed to write events: %w", err)
-					return
+
+				// Is this a JSON object?
+				if err := json.Unmarshal(line, &[]map[string]any{make(map[string]any)}[0]); err != nil {
+					// If not, then wait until we get the rest of the output.
+					frag = line[:]
+					continue
+				}
+
+				frag = frag[:0]
+
+				if bytes.HasPrefix(line, []byte(`{"stdout":`)) {
+					_, err = bufferedStdout.Write(bytes.TrimSuffix(bytes.TrimPrefix(line, []byte(`{"stdout":`)), []byte("}")))
+					if err != nil {
+						r.state = Error
+						r.err = fmt.Errorf("failed to write stdout: %w", err)
+						return
+					}
+				} else if bytes.HasPrefix(line, []byte(`{"stderr":`)) {
+					_, err = bufferedStderr.Write(bytes.TrimSuffix(bytes.TrimPrefix(line, []byte(`{"stderr":`)), []byte("}")))
+					if err != nil {
+						r.state = Error
+						r.err = fmt.Errorf("failed to write stderr: %w", err)
+						return
+					}
+				} else {
+					_, err = eventsWrite.Write(append(line, '\n', '\n'))
+					if err != nil {
+						r.state = Error
+						r.err = fmt.Errorf("failed to write events: %w", err)
+						return
+					}
 				}
 			}
 		}
+
+		if err != nil && !errors.Is(err, io.EOF) {
+			slog.Debug("failed to read events from response", "error", err)
+			r.err = fmt.Errorf("failed to read events: %w", err)
+		}
 	}()
 
 	r.wait = func() error {

test/global-tools.gpt

@@ -0,0 +1,19 @@
+!title
+
+Runbook 3
+
+---
+Name: tool_1
+Global Tools: sys.workspace.ls, sys.workspace.read, sys.workspace.write, github.com/gptscript-ai/knowledge, github.com/drpebcak/duckdb, github.com/gptscript-ai/browser, github.com/gptscript-ai/browser-search/google, github.com/gptscript-ai/browser-search/google-question-answerer
+
+Hi
+
+---
+Name: tool_2
+
+What time is it?
+
+---
+Name: tool_3
+
+Give me a paragraph of lorem ipsum