Remove the bidi Connect

examples/docker_agent/main.go

@@ -18,7 +18,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"io"
 	"log"
 	"net"
 	"os"
@@ -39,16 +38,8 @@ type server struct {
 	skillsExecutor *skills.Executor
 }
 
-func (s *server) Connect(stream grpc.BidiStreamingServer[proto.AgentMessage, proto.AgentMessage]) error {
-	incoming, err := stream.Recv()
-	if err == io.EOF {
-		return nil
-	}
-	if err != nil {
-		return err
-	}
-
-	start := incoming.GetStart()
+func (s *server) Connect(req *proto.AgentRequest, stream grpc.ServerStreamingServer[proto.AgentResponse]) error {
+	start := req.GetStart()
 	if start == nil {
 		return errors.New("missing start message")
 	}
@@ -131,10 +122,10 @@ func (s *server) Connect(stream grpc.BidiStreamingServer[proto.AgentMessage, pro
 						},
 					},
 				}
-				if err := stream.Send(&proto.AgentMessage{
-					ConversationId: incoming.ConversationId,
-					ExecId:         incoming.ExecId,
-					Type: &proto.AgentMessage_Outputs{
+				if err := stream.Send(&proto.AgentResponse{
+					ConversationId: req.ConversationId,
+					ExecId:         req.ExecId,
+					Type: &proto.AgentResponse_Outputs{
 						Outputs: &proto.AgentOutputs{
 							Messages: []*proto.Message{responseMsg},
 						},
@@ -153,10 +144,10 @@ func (s *server) Connect(stream grpc.BidiStreamingServer[proto.AgentMessage, pro
 	fmt.Println("DockerAgent finished execution turn.")
 
 	// Send AgentEnd
-	return stream.Send(&proto.AgentMessage{
-		ConversationId: incoming.ConversationId,
-		ExecId:         incoming.ExecId,
-		Type: &proto.AgentMessage_End{
+	return stream.Send(&proto.AgentResponse{
+		ConversationId: req.ConversationId,
+		ExecId:         req.ExecId,
+		Type: &proto.AgentResponse_End{
 			End: &proto.AgentEnd{},
 		},
 	})

examples/docker_agent/test/test_docker_agent.go

@@ -34,34 +34,28 @@ func main() {
 	defer conn.Close()
 
 	client := proto.NewAgentServiceClient(conn)
-	stream, err := client.Connect(context.Background())
-	if err != nil {
-		log.Fatalf("Failed to open stream: %v", err)
-	}
-
 	fmt.Println("Sending request to write a Dockerfile...")
-	// Send start message
-	err = stream.Send(&proto.AgentMessage{
+	req := &proto.AgentRequest{
 		ConversationId: "test-conversation-id",
 		ExecId:         "test-exec-id",
-		Type: &proto.AgentMessage_Start{
-			Start: &proto.AgentStart{
-				AgentId: "DockerAgent",
-				Messages: []*proto.Message{
-					{
-						Role: "user",
-						Content: &proto.Content{
-							Type: &proto.Content_Text{
-								Text: &proto.TextContent{Text: "Write a Dockerfile for a simple Node.js app."},
-							},
+		Start: &proto.AgentStart{
+			AgentId: "DockerAgent",
+			Messages: []*proto.Message{
+				{
+					Role: "user",
+					Content: &proto.Content{
+						Type: &proto.Content_Text{
+							Text: &proto.TextContent{Text: "Write a Dockerfile for a simple Node.js app."},
 						},
 					},
 				},
 			},
 		},
-	})
+	}
+
+	stream, err := client.Connect(context.Background(), req)
 	if err != nil {
-		log.Fatalf("Failed to send message: %v", err)
+		log.Fatalf("Failed to open stream: %v", err)
 	}
 
 	fmt.Println("Waiting for response...")

examples/remote_agent/main.go

@@ -18,7 +18,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"io"
 	"log"
 	"net"
 	"strings"
@@ -35,73 +34,64 @@ type server struct {
 	proto.UnimplementedAgentServiceServer
 }
 
-// Connect implements bidirectional streaming for agent processing.
-// This RPC is called by the gar controller when a session triggers this agent.
-func (s *server) Connect(stream grpc.BidiStreamingServer[proto.AgentMessage, proto.AgentMessage]) error {
-	for {
-		// Receive input content from gar controller
-		incoming, err := stream.Recv()
-		if err == io.EOF {
-			return nil
-		}
-		if err != nil {
-			return err
-		}
-
-		start := incoming.GetStart()
-		if start == nil {
-			continue // optionally wait for start
-		}
+// Connect implements server streaming for agent processing.
+// This RPC is called by the controller when a session triggers this agent.
+func (s *server) Connect(req *proto.AgentRequest, stream grpc.ServerStreamingServer[proto.AgentResponse]) error {
+	start := req.GetStart()
+	if start == nil {
+		return errors.New("start is required")
+	}
 
-		var messages []*proto.Message
-		for _, input := range start.Messages {
-			textContent := input.GetContent().GetText()
-			if textContent == nil {
-				continue
-			}
-			messages = append(messages, input)
-		}
-		if len(messages) == 0 {
-			return errors.New("no text inputs, cannot uppercase")
+	var messages []*proto.Message
+	for _, input := range start.Messages {
+		textContent := input.GetContent().GetText()
+		if textContent == nil {
+			continue
 		}
+		messages = append(messages, input)
+	}
+	if len(messages) == 0 {
+		return errors.New("no text inputs, cannot uppercase")
+	}
 
-		// We don't need to uppercase the whole history.
-		// Only uppercase the last text message.
-		lastMsg := messages[len(messages)-1]
-		responseMsg := &proto.Message{
-			Role: "assistant",
-			Content: &proto.Content{
-				Type: &proto.Content_Text{
-					Text: &proto.TextContent{
-						Text: strings.ToLower(lastMsg.GetContent().GetText().Text),
-					},
+	// We don't need to uppercase the whole history.
+	// Only uppercase the last text message.
+	lastMsg := messages[len(messages)-1]
+	responseMsg := &proto.Message{
+		Role: "assistant",
+		Content: &proto.Content{
+			Type: &proto.Content_Text{
+				Text: &proto.TextContent{
+					Text: strings.ToLower(lastMsg.GetContent().GetText().Text),
 				},
 			},
-		}
+		},
+	}
 
-		if err := stream.Send(&proto.AgentMessage{
-			ConversationId: incoming.ConversationId,
-			ExecId:         incoming.ExecId,
-			Type: &proto.AgentMessage_Outputs{
-				Outputs: &proto.AgentOutputs{
-					Messages: []*proto.Message{responseMsg},
-				},
+	if err := stream.Send(&proto.AgentResponse{
+		ConversationId: req.ConversationId,
+		ExecId:         req.ExecId,
+		Type: &proto.AgentResponse_Outputs{
+			Outputs: &proto.AgentOutputs{
+				Messages: []*proto.Message{responseMsg},
 			},
-		}); err != nil {
-			return err
-		}
+		},
+	}); err != nil {
+		return err
+	}
 
-		// Send AgentEnd to signal end of outputs.
-		if err := stream.Send(&proto.AgentMessage{
-			ConversationId: incoming.ConversationId,
-			ExecId:         incoming.ExecId,
-			Type: &proto.AgentMessage_End{
-				End: &proto.AgentEnd{},
-			},
-		}); err != nil {
-			return err
-		}
+	// Send AgentEnd to signal end of outputs.
+	if err := stream.Send(&proto.AgentResponse{
+		ConversationId: req.ConversationId,
+		ExecId:         req.ExecId,
+		Type: &proto.AgentResponse_End{
+			End: &proto.AgentEnd{},
+		},
+	}); err != nil {
+		return err
 	}
+
+	return nil
 }
 
 // HealthCheck checks if the agent is healthy.

internal/agent/remote.go

@@ -16,7 +16,6 @@ package agent
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"io"
 
@@ -75,24 +74,13 @@ func (a *RemoteAgent) Connect(ctx context.Context, conversationID string, execID
 	defer conn.Close()
 
 	client := proto.NewAgentServiceClient(conn)
-	stream, err := client.Connect(ctx)
-	if err != nil {
-		return fmt.Errorf("failed to create stream: %w", err)
-	}
-
-	if err := stream.Send(&proto.AgentMessage{
+	stream, err := client.Connect(ctx, &proto.AgentRequest{
 		ConversationId: conversationID,
 		ExecId:         execID,
-		Type: &proto.AgentMessage_Start{
-			Start: start,
-		},
-	}); err != nil {
-		return fmt.Errorf("failed to send content: %w", err)
-	}
-
-	// Close the send direction to signal we're done sending
-	if err := stream.CloseSend(); err != nil {
-		return fmt.Errorf("failed to close send: %w", err)
+		Start:          start,
+	})
+	if err != nil {
+		return fmt.Errorf("failed to create stream: %w", err)
 	}
 
 	// Receive outputs and call handler for each
@@ -107,9 +95,7 @@ func (a *RemoteAgent) Connect(ctx context.Context, conversationID string, execID
 		}
 
 		switch msg := resp.Type.(type) {
-		case *proto.AgentMessage_Start:
-			return errors.New("starting new executions from remote agents is not supported yet")
-		case *proto.AgentMessage_Outputs:
+		case *proto.AgentResponse_Outputs:
 			if resp.ConversationId != conversationID {
 				return fmt.Errorf("received outputs for different conversation id: %s != %s", resp.ConversationId, conversationID)
 			}
@@ -119,7 +105,7 @@ func (a *RemoteAgent) Connect(ctx context.Context, conversationID string, execID
 			if err := o(msg.Outputs); err != nil {
 				return fmt.Errorf("handler error: %w", err)
 			}
-		case *proto.AgentMessage_End:
+		case *proto.AgentResponse_End:
 			if resp.ConversationId != conversationID {
 				return fmt.Errorf("received end for different conversation id: %s != %s", resp.ConversationId, conversationID)
 			}