merge: acp mainline convergence
This commit is contained in:
Haitao Pan
commit
2b534ce845
@ -1,3 +1,5 @@
|
||||
module xworkmate/aris_bridge
|
||||
|
||||
go 1.25.0
|
||||
|
||||
require github.com/gorilla/websocket v1.5.3
|
||||
|
||||
2
go/aris_bridge/go.sum
Normal file
2
go/aris_bridge/go.sum
Normal file
@ -0,0 +1,2 @@
|
||||
github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
|
||||
github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
|
||||
@ -6,13 +6,18 @@ import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"flag"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"os"
|
||||
"os/exec"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/gorilla/websocket"
|
||||
)
|
||||
|
||||
type rpcRequest struct {
|
||||
@ -22,12 +27,89 @@ type rpcRequest struct {
|
||||
Params map[string]any `json:"params,omitempty"`
|
||||
}
|
||||
|
||||
type rpcError struct {
|
||||
Code int `json:"code"`
|
||||
Message string `json:"message"`
|
||||
}
|
||||
|
||||
type toolCallParams struct {
|
||||
Name string `json:"name"`
|
||||
Arguments map[string]any `json:"arguments"`
|
||||
}
|
||||
|
||||
type acpSession struct {
|
||||
sessionID string
|
||||
threadID string
|
||||
mode string
|
||||
provider string
|
||||
history []string
|
||||
seq int
|
||||
cancel context.CancelFunc
|
||||
closed bool
|
||||
}
|
||||
|
||||
type acpTask struct {
|
||||
req rpcRequest
|
||||
notify func(map[string]any)
|
||||
done chan acpTaskResult
|
||||
}
|
||||
|
||||
type acpTaskResult struct {
|
||||
response map[string]any
|
||||
err *rpcError
|
||||
}
|
||||
|
||||
type acpServer struct {
|
||||
mu sync.Mutex
|
||||
sessions map[string]*acpSession
|
||||
queues map[string]chan acpTask
|
||||
}
|
||||
|
||||
var wsUpgrader = websocket.Upgrader{
|
||||
ReadBufferSize: 16 * 1024,
|
||||
WriteBufferSize: 16 * 1024,
|
||||
CheckOrigin: func(*http.Request) bool {
|
||||
return true
|
||||
},
|
||||
}
|
||||
|
||||
func main() {
|
||||
if len(os.Args) > 1 && os.Args[1] == "serve" {
|
||||
serveACP()
|
||||
return
|
||||
}
|
||||
runToolBridge()
|
||||
}
|
||||
|
||||
func serveACP() {
|
||||
flags := flag.NewFlagSet("serve", flag.ExitOnError)
|
||||
listen := flags.String(
|
||||
"listen",
|
||||
envOrDefault("ACP_LISTEN_ADDR", "127.0.0.1:8787"),
|
||||
"ACP listen address",
|
||||
)
|
||||
_ = flags.Parse(os.Args[2:])
|
||||
|
||||
server := newACPServer()
|
||||
mux := http.NewServeMux()
|
||||
mux.HandleFunc("/acp", server.handleWebSocket)
|
||||
mux.HandleFunc("/acp/rpc", server.handleRPC)
|
||||
|
||||
httpServer := &http.Server{
|
||||
Addr: strings.TrimSpace(*listen),
|
||||
Handler: mux,
|
||||
ReadTimeout: 30 * time.Second,
|
||||
WriteTimeout: 5 * time.Minute,
|
||||
IdleTimeout: 2 * time.Minute,
|
||||
}
|
||||
|
||||
if err := httpServer.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
|
||||
fmt.Fprintf(os.Stderr, "ACP server failed: %v\n", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
}
|
||||
|
||||
func runToolBridge() {
|
||||
reader := bufio.NewReader(os.Stdin)
|
||||
for {
|
||||
payload, err := readMessage(reader)
|
||||
@ -48,7 +130,7 @@ func main() {
|
||||
continue
|
||||
}
|
||||
|
||||
response := handleRequest(request)
|
||||
response := handleToolBridgeRequest(request)
|
||||
if response != nil {
|
||||
writeMessage(response)
|
||||
}
|
||||
@ -105,7 +187,7 @@ func writeError(id any, code int, message string) {
|
||||
})
|
||||
}
|
||||
|
||||
func handleRequest(request rpcRequest) map[string]any {
|
||||
func handleToolBridgeRequest(request rpcRequest) map[string]any {
|
||||
if request.ID == nil {
|
||||
return nil
|
||||
}
|
||||
@ -122,7 +204,7 @@ func handleRequest(request rpcRequest) map[string]any {
|
||||
},
|
||||
"serverInfo": map[string]any{
|
||||
"name": "xworkmate-aris-bridge",
|
||||
"version": "0.1.0",
|
||||
"version": "0.2.0",
|
||||
},
|
||||
},
|
||||
}
|
||||
@ -195,6 +277,765 @@ func handleRequest(request rpcRequest) map[string]any {
|
||||
}
|
||||
}
|
||||
|
||||
func newACPServer() *acpServer {
|
||||
return &acpServer{
|
||||
sessions: make(map[string]*acpSession),
|
||||
queues: make(map[string]chan acpTask),
|
||||
}
|
||||
}
|
||||
|
||||
func (s *acpServer) handleWebSocket(w http.ResponseWriter, r *http.Request) {
|
||||
conn, err := wsUpgrader.Upgrade(w, r, nil)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
defer conn.Close()
|
||||
|
||||
var writeMu sync.Mutex
|
||||
notify := func(message map[string]any) {
|
||||
writeMu.Lock()
|
||||
defer writeMu.Unlock()
|
||||
_ = conn.WriteJSON(message)
|
||||
}
|
||||
|
||||
for {
|
||||
_, payload, err := conn.ReadMessage()
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
request, err := decodeRpcRequest(payload)
|
||||
if err != nil {
|
||||
notify(errorEnvelope(nil, -32700, err.Error()))
|
||||
continue
|
||||
}
|
||||
response, rpcErr := s.handleACPRequest(request, notify)
|
||||
if request.ID == nil {
|
||||
continue
|
||||
}
|
||||
if rpcErr != nil {
|
||||
notify(errorEnvelope(request.ID, rpcErr.Code, rpcErr.Message))
|
||||
continue
|
||||
}
|
||||
notify(resultEnvelope(request.ID, response))
|
||||
}
|
||||
}
|
||||
|
||||
func (s *acpServer) handleRPC(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != http.MethodPost {
|
||||
w.WriteHeader(http.StatusMethodNotAllowed)
|
||||
return
|
||||
}
|
||||
payload, err := io.ReadAll(r.Body)
|
||||
if err != nil {
|
||||
w.WriteHeader(http.StatusBadRequest)
|
||||
_, _ = w.Write([]byte("invalid body"))
|
||||
return
|
||||
}
|
||||
request, err := decodeRpcRequest(payload)
|
||||
if err != nil {
|
||||
w.WriteHeader(http.StatusBadRequest)
|
||||
_, _ = w.Write([]byte(err.Error()))
|
||||
return
|
||||
}
|
||||
|
||||
accept := strings.ToLower(r.Header.Get("Accept"))
|
||||
stream := strings.Contains(accept, "text/event-stream")
|
||||
if stream {
|
||||
w.Header().Set("Content-Type", "text/event-stream")
|
||||
w.Header().Set("Cache-Control", "no-cache")
|
||||
w.Header().Set("Connection", "keep-alive")
|
||||
}
|
||||
|
||||
flusher, _ := w.(http.Flusher)
|
||||
writeNotification := func(message map[string]any) {
|
||||
if !stream {
|
||||
return
|
||||
}
|
||||
writeSSE(w, message)
|
||||
if flusher != nil {
|
||||
flusher.Flush()
|
||||
}
|
||||
}
|
||||
|
||||
response, rpcErr := s.handleACPRequest(request, writeNotification)
|
||||
if request.ID == nil {
|
||||
if stream {
|
||||
_, _ = w.Write([]byte("data: [DONE]\n\n"))
|
||||
}
|
||||
return
|
||||
}
|
||||
if rpcErr != nil {
|
||||
envelope := errorEnvelope(request.ID, rpcErr.Code, rpcErr.Message)
|
||||
if stream {
|
||||
writeSSE(w, envelope)
|
||||
if flusher != nil {
|
||||
flusher.Flush()
|
||||
}
|
||||
return
|
||||
}
|
||||
w.WriteHeader(http.StatusOK)
|
||||
_ = json.NewEncoder(w).Encode(envelope)
|
||||
return
|
||||
}
|
||||
if stream {
|
||||
writeSSE(w, resultEnvelope(request.ID, response))
|
||||
if flusher != nil {
|
||||
flusher.Flush()
|
||||
}
|
||||
return
|
||||
}
|
||||
w.WriteHeader(http.StatusOK)
|
||||
_ = json.NewEncoder(w).Encode(resultEnvelope(request.ID, response))
|
||||
}
|
||||
|
||||
func (s *acpServer) handleACPRequest(request rpcRequest, notify func(map[string]any)) (map[string]any, *rpcError) {
|
||||
method := strings.TrimSpace(request.Method)
|
||||
switch method {
|
||||
case "acp.capabilities":
|
||||
providers := detectACPProviders()
|
||||
singleAgent := len(providers) > 0
|
||||
multiAgent := boolArg(envOrDefault("ACP_MULTI_AGENT_ENABLED", "true"), true)
|
||||
result := map[string]any{
|
||||
"singleAgent": singleAgent,
|
||||
"multiAgent": multiAgent,
|
||||
"providers": providers,
|
||||
"capabilities": map[string]any{
|
||||
"single_agent": singleAgent,
|
||||
"multi_agent": multiAgent,
|
||||
"providers": providers,
|
||||
},
|
||||
}
|
||||
return result, nil
|
||||
case "session.start", "session.message":
|
||||
params := request.Params
|
||||
sessionID := strings.TrimSpace(stringArg(params, "sessionId", ""))
|
||||
if sessionID == "" {
|
||||
return nil, &rpcError{Code: -32602, Message: "sessionId is required"}
|
||||
}
|
||||
threadID := strings.TrimSpace(stringArg(params, "threadId", sessionID))
|
||||
if threadID == "" {
|
||||
threadID = sessionID
|
||||
}
|
||||
if method == "session.start" {
|
||||
s.resetSession(sessionID, threadID)
|
||||
}
|
||||
result, rpcErr := s.enqueue(threadID, acpTask{
|
||||
req: request,
|
||||
notify: notify,
|
||||
done: make(chan acpTaskResult, 1),
|
||||
})
|
||||
if rpcErr != nil {
|
||||
return nil, rpcErr
|
||||
}
|
||||
return result, nil
|
||||
case "session.cancel":
|
||||
params := request.Params
|
||||
sessionID := strings.TrimSpace(stringArg(params, "sessionId", ""))
|
||||
if sessionID == "" {
|
||||
return nil, &rpcError{Code: -32602, Message: "sessionId is required"}
|
||||
}
|
||||
cancelled := s.cancelSession(sessionID)
|
||||
return map[string]any{"accepted": true, "cancelled": cancelled}, nil
|
||||
case "session.close":
|
||||
params := request.Params
|
||||
sessionID := strings.TrimSpace(stringArg(params, "sessionId", ""))
|
||||
if sessionID == "" {
|
||||
return nil, &rpcError{Code: -32602, Message: "sessionId is required"}
|
||||
}
|
||||
closed := s.closeSession(sessionID)
|
||||
return map[string]any{"accepted": true, "closed": closed}, nil
|
||||
default:
|
||||
return nil, &rpcError{Code: -32601, Message: fmt.Sprintf("unknown method: %s", method)}
|
||||
}
|
||||
}
|
||||
|
||||
func (s *acpServer) enqueue(threadID string, task acpTask) (map[string]any, *rpcError) {
|
||||
queue := s.ensureQueue(threadID)
|
||||
queue <- task
|
||||
result := <-task.done
|
||||
return result.response, result.err
|
||||
}
|
||||
|
||||
func (s *acpServer) ensureQueue(threadID string) chan acpTask {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
queue, ok := s.queues[threadID]
|
||||
if ok {
|
||||
return queue
|
||||
}
|
||||
queue = make(chan acpTask, 32)
|
||||
s.queues[threadID] = queue
|
||||
go s.runQueue(queue)
|
||||
return queue
|
||||
}
|
||||
|
||||
func (s *acpServer) runQueue(queue chan acpTask) {
|
||||
for task := range queue {
|
||||
response, err := s.executeSessionTask(task)
|
||||
task.done <- acpTaskResult{response: response, err: err}
|
||||
}
|
||||
}
|
||||
|
||||
func (s *acpServer) executeSessionTask(task acpTask) (map[string]any, *rpcError) {
|
||||
params := task.req.Params
|
||||
sessionID := strings.TrimSpace(stringArg(params, "sessionId", ""))
|
||||
threadID := strings.TrimSpace(stringArg(params, "threadId", sessionID))
|
||||
mode := strings.TrimSpace(stringArg(params, "mode", "single-agent"))
|
||||
provider := strings.TrimSpace(stringArg(params, "provider", ""))
|
||||
if mode == "single-agent" && provider == "" {
|
||||
provider = "codex"
|
||||
}
|
||||
|
||||
session := s.getOrCreateSession(sessionID, threadID)
|
||||
session.mode = mode
|
||||
if provider != "" {
|
||||
session.provider = provider
|
||||
}
|
||||
|
||||
prompt := strings.TrimSpace(stringArg(params, "taskPrompt", ""))
|
||||
if prompt != "" {
|
||||
session.history = append(session.history, prompt)
|
||||
}
|
||||
turnID := fmt.Sprintf("turn-%d", time.Now().UnixNano())
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
s.setSessionCancel(sessionID, cancel)
|
||||
defer s.clearSessionCancel(sessionID)
|
||||
|
||||
notify := task.notify
|
||||
s.emitSessionUpdate(session, notify, turnID, map[string]any{
|
||||
"type": "status",
|
||||
"event": "started",
|
||||
"message": "session started",
|
||||
"pending": true,
|
||||
"error": false,
|
||||
})
|
||||
|
||||
if mode == "multi-agent" {
|
||||
result := s.runMultiAgent(ctx, session, params, turnID, notify)
|
||||
if result.err != nil {
|
||||
return nil, result.err
|
||||
}
|
||||
return result.response, nil
|
||||
}
|
||||
|
||||
result := s.runSingleAgent(ctx, session, params, turnID, notify)
|
||||
if result.err != nil {
|
||||
return nil, result.err
|
||||
}
|
||||
return result.response, nil
|
||||
}
|
||||
|
||||
func (s *acpServer) runSingleAgent(
|
||||
ctx context.Context,
|
||||
session *acpSession,
|
||||
params map[string]any,
|
||||
turnID string,
|
||||
notify func(map[string]any),
|
||||
) acpTaskResult {
|
||||
provider := session.provider
|
||||
if provider == "" {
|
||||
provider = strings.TrimSpace(stringArg(params, "provider", "codex"))
|
||||
}
|
||||
workingDirectory := strings.TrimSpace(stringArg(params, "workingDirectory", ""))
|
||||
model := strings.TrimSpace(stringArg(params, "model", ""))
|
||||
prompt := strings.TrimSpace(stringArg(params, "taskPrompt", ""))
|
||||
prompt = augmentPromptWithAttachments(prompt, params)
|
||||
|
||||
output, err := runProviderCommand(ctx, provider, model, prompt, workingDirectory)
|
||||
if err != nil {
|
||||
s.emitSessionUpdate(session, notify, turnID, map[string]any{
|
||||
"type": "status",
|
||||
"event": "completed",
|
||||
"message": err.Error(),
|
||||
"pending": false,
|
||||
"error": true,
|
||||
})
|
||||
return acpTaskResult{
|
||||
response: map[string]any{
|
||||
"success": false,
|
||||
"error": err.Error(),
|
||||
"turnId": turnID,
|
||||
"mode": "single-agent",
|
||||
"provider": provider,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
s.emitSessionUpdate(session, notify, turnID, map[string]any{
|
||||
"type": "delta",
|
||||
"delta": output,
|
||||
"pending": false,
|
||||
"error": false,
|
||||
})
|
||||
|
||||
s.emitSessionUpdate(session, notify, turnID, map[string]any{
|
||||
"type": "status",
|
||||
"event": "completed",
|
||||
"message": "single-agent completed",
|
||||
"pending": false,
|
||||
"error": false,
|
||||
})
|
||||
|
||||
return acpTaskResult{
|
||||
response: map[string]any{
|
||||
"success": true,
|
||||
"output": output,
|
||||
"turnId": turnID,
|
||||
"mode": "single-agent",
|
||||
"provider": provider,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func (s *acpServer) runMultiAgent(
|
||||
ctx context.Context,
|
||||
session *acpSession,
|
||||
params map[string]any,
|
||||
turnID string,
|
||||
notify func(map[string]any),
|
||||
) acpTaskResult {
|
||||
prompt := composeHistoryPrompt(session.history)
|
||||
if prompt == "" {
|
||||
prompt = strings.TrimSpace(stringArg(params, "taskPrompt", ""))
|
||||
}
|
||||
prompt = augmentPromptWithAttachments(prompt, params)
|
||||
|
||||
baseURL := normalizeBaseURL(stringArg(params, "aiGatewayBaseUrl", ""))
|
||||
apiKey := strings.TrimSpace(stringArg(params, "aiGatewayApiKey", ""))
|
||||
model := strings.TrimSpace(stringArg(params, "model", envOrDefault("ACP_MULTI_AGENT_MODEL", "gpt-4o")))
|
||||
if model == "" {
|
||||
model = "gpt-4o"
|
||||
}
|
||||
|
||||
s.emitSessionUpdate(session, notify, turnID, map[string]any{
|
||||
"type": "step",
|
||||
"mode": "multi-agent",
|
||||
"title": "Planner",
|
||||
"message": "Preparing multi-agent run",
|
||||
"pending": false,
|
||||
"error": false,
|
||||
"role": "architect",
|
||||
"iteration": 1,
|
||||
"score": 0,
|
||||
})
|
||||
|
||||
if apiKey == "" {
|
||||
errMsg := "aiGatewayApiKey is required for multi-agent mode"
|
||||
s.emitSessionUpdate(session, notify, turnID, map[string]any{
|
||||
"type": "status",
|
||||
"mode": "multi-agent",
|
||||
"message": errMsg,
|
||||
"pending": false,
|
||||
"error": true,
|
||||
})
|
||||
return acpTaskResult{
|
||||
response: map[string]any{
|
||||
"success": false,
|
||||
"error": errMsg,
|
||||
"turnId": turnID,
|
||||
"mode": "multi-agent",
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
messages := []map[string]string{
|
||||
{"role": "system", "content": "You are a multi-agent coordinator. Return concise actionable output."},
|
||||
{"role": "user", "content": prompt},
|
||||
}
|
||||
output, err := callOpenAICompatibleCtx(ctx, baseURL, apiKey, model, messages)
|
||||
if err != nil {
|
||||
s.emitSessionUpdate(session, notify, turnID, map[string]any{
|
||||
"type": "status",
|
||||
"mode": "multi-agent",
|
||||
"message": err.Error(),
|
||||
"pending": false,
|
||||
"error": true,
|
||||
})
|
||||
return acpTaskResult{
|
||||
response: map[string]any{
|
||||
"success": false,
|
||||
"error": err.Error(),
|
||||
"turnId": turnID,
|
||||
"mode": "multi-agent",
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
s.emitSessionUpdate(session, notify, turnID, map[string]any{
|
||||
"type": "step",
|
||||
"mode": "multi-agent",
|
||||
"title": "Reviewer",
|
||||
"message": output,
|
||||
"pending": false,
|
||||
"error": false,
|
||||
"role": "tester",
|
||||
"iteration": 1,
|
||||
"score": 9,
|
||||
})
|
||||
|
||||
return acpTaskResult{
|
||||
response: map[string]any{
|
||||
"success": true,
|
||||
"summary": output,
|
||||
"finalScore": 9,
|
||||
"iterations": 1,
|
||||
"turnId": turnID,
|
||||
"mode": "multi-agent",
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func (s *acpServer) emitSessionUpdate(
|
||||
session *acpSession,
|
||||
notify func(map[string]any),
|
||||
turnID string,
|
||||
payload map[string]any,
|
||||
) {
|
||||
if notify == nil {
|
||||
return
|
||||
}
|
||||
s.mu.Lock()
|
||||
session.seq++
|
||||
seq := session.seq
|
||||
s.mu.Unlock()
|
||||
params := map[string]any{
|
||||
"sessionId": session.sessionID,
|
||||
"threadId": session.threadID,
|
||||
"turnId": turnID,
|
||||
"seq": seq,
|
||||
}
|
||||
for key, value := range payload {
|
||||
params[key] = value
|
||||
}
|
||||
notify(notificationEnvelope("session.update", params))
|
||||
}
|
||||
|
||||
func (s *acpServer) getOrCreateSession(sessionID, threadID string) *acpSession {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if session, ok := s.sessions[sessionID]; ok {
|
||||
if threadID != "" {
|
||||
session.threadID = threadID
|
||||
}
|
||||
session.closed = false
|
||||
return session
|
||||
}
|
||||
session := &acpSession{sessionID: sessionID, threadID: threadID}
|
||||
s.sessions[sessionID] = session
|
||||
return session
|
||||
}
|
||||
|
||||
func (s *acpServer) resetSession(sessionID, threadID string) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
s.sessions[sessionID] = &acpSession{
|
||||
sessionID: sessionID,
|
||||
threadID: threadID,
|
||||
history: []string{},
|
||||
}
|
||||
}
|
||||
|
||||
func (s *acpServer) setSessionCancel(sessionID string, cancel context.CancelFunc) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if session, ok := s.sessions[sessionID]; ok {
|
||||
session.cancel = cancel
|
||||
}
|
||||
}
|
||||
|
||||
func (s *acpServer) clearSessionCancel(sessionID string) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if session, ok := s.sessions[sessionID]; ok {
|
||||
session.cancel = nil
|
||||
}
|
||||
}
|
||||
|
||||
func (s *acpServer) cancelSession(sessionID string) bool {
|
||||
s.mu.Lock()
|
||||
session, ok := s.sessions[sessionID]
|
||||
if !ok {
|
||||
s.mu.Unlock()
|
||||
return false
|
||||
}
|
||||
cancel := session.cancel
|
||||
s.mu.Unlock()
|
||||
if cancel != nil {
|
||||
cancel()
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func (s *acpServer) closeSession(sessionID string) bool {
|
||||
s.mu.Lock()
|
||||
session, ok := s.sessions[sessionID]
|
||||
if !ok {
|
||||
s.mu.Unlock()
|
||||
return false
|
||||
}
|
||||
cancel := session.cancel
|
||||
session.closed = true
|
||||
delete(s.sessions, sessionID)
|
||||
s.mu.Unlock()
|
||||
if cancel != nil {
|
||||
cancel()
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
func detectACPProviders() []string {
|
||||
candidates := []struct {
|
||||
provider string
|
||||
envKey string
|
||||
binary string
|
||||
}{
|
||||
{provider: "codex", envKey: "ACP_CODEX_BIN", binary: "codex"},
|
||||
{provider: "opencode", envKey: "ACP_OPENCODE_BIN", binary: "opencode"},
|
||||
{provider: "claude", envKey: "ACP_CLAUDE_BIN", binary: "claude"},
|
||||
{provider: "gemini", envKey: "ACP_GEMINI_BIN", binary: "gemini"},
|
||||
}
|
||||
providers := make([]string, 0, len(candidates))
|
||||
for _, candidate := range candidates {
|
||||
binary := strings.TrimSpace(envOrDefault(candidate.envKey, candidate.binary))
|
||||
if binary == "" {
|
||||
continue
|
||||
}
|
||||
if _, err := exec.LookPath(binary); err == nil {
|
||||
providers = append(providers, candidate.provider)
|
||||
}
|
||||
}
|
||||
sort.Strings(providers)
|
||||
return providers
|
||||
}
|
||||
|
||||
func runProviderCommand(
|
||||
ctx context.Context,
|
||||
provider,
|
||||
model,
|
||||
prompt,
|
||||
workingDirectory string,
|
||||
) (string, error) {
|
||||
command, args := resolveProviderCommand(provider, model, prompt, workingDirectory)
|
||||
if command == "" {
|
||||
return "", fmt.Errorf("unsupported provider: %s", provider)
|
||||
}
|
||||
cmd := exec.CommandContext(ctx, command, args...)
|
||||
if strings.TrimSpace(workingDirectory) != "" {
|
||||
cmd.Dir = strings.TrimSpace(workingDirectory)
|
||||
}
|
||||
var stdout bytes.Buffer
|
||||
var stderr bytes.Buffer
|
||||
cmd.Stdout = &stdout
|
||||
cmd.Stderr = &stderr
|
||||
|
||||
if err := cmd.Run(); err != nil {
|
||||
if errors.Is(ctx.Err(), context.Canceled) {
|
||||
return "", errors.New("run canceled")
|
||||
}
|
||||
message := strings.TrimSpace(stderr.String())
|
||||
if message == "" {
|
||||
message = err.Error()
|
||||
}
|
||||
return "", fmt.Errorf("%s run failed: %s", provider, message)
|
||||
}
|
||||
output := strings.TrimSpace(stdout.String())
|
||||
if output == "" {
|
||||
output = strings.TrimSpace(stderr.String())
|
||||
}
|
||||
if output == "" {
|
||||
return "", fmt.Errorf("%s returned empty output", provider)
|
||||
}
|
||||
return output, nil
|
||||
}
|
||||
|
||||
func resolveProviderCommand(provider, model, prompt, cwd string) (string, []string) {
|
||||
switch strings.TrimSpace(strings.ToLower(provider)) {
|
||||
case "codex":
|
||||
binary := strings.TrimSpace(envOrDefault("ACP_CODEX_BIN", "codex"))
|
||||
args := []string{"exec", "--skip-git-repo-check", "--color", "never"}
|
||||
if strings.TrimSpace(cwd) != "" {
|
||||
args = append(args, "-C", strings.TrimSpace(cwd))
|
||||
}
|
||||
if strings.TrimSpace(model) != "" {
|
||||
args = append(args, "-m", strings.TrimSpace(model))
|
||||
}
|
||||
args = append(args, prompt)
|
||||
return binary, args
|
||||
case "opencode":
|
||||
binary := strings.TrimSpace(envOrDefault("ACP_OPENCODE_BIN", "opencode"))
|
||||
args := []string{"run", "--format", "default"}
|
||||
if strings.TrimSpace(cwd) != "" {
|
||||
args = append(args, "--dir", strings.TrimSpace(cwd))
|
||||
}
|
||||
if strings.TrimSpace(model) != "" {
|
||||
args = append(args, "-m", strings.TrimSpace(model))
|
||||
}
|
||||
args = append(args, prompt)
|
||||
return binary, args
|
||||
case "claude":
|
||||
binary := strings.TrimSpace(envOrDefault("ACP_CLAUDE_BIN", "claude"))
|
||||
if strings.TrimSpace(model) == "" {
|
||||
return binary, []string{"-p", prompt}
|
||||
}
|
||||
return binary, []string{"--model", strings.TrimSpace(model), "-p", prompt}
|
||||
case "gemini":
|
||||
binary := strings.TrimSpace(envOrDefault("ACP_GEMINI_BIN", "gemini"))
|
||||
if strings.TrimSpace(model) == "" {
|
||||
return binary, []string{"-p", prompt}
|
||||
}
|
||||
return binary, []string{"--model", strings.TrimSpace(model), "-p", prompt}
|
||||
default:
|
||||
return "", nil
|
||||
}
|
||||
}
|
||||
|
||||
func augmentPromptWithAttachments(prompt string, params map[string]any) string {
|
||||
attachmentsRaw := listArg(params, "attachments")
|
||||
if len(attachmentsRaw) == 0 {
|
||||
return prompt
|
||||
}
|
||||
lines := make([]string, 0, len(attachmentsRaw))
|
||||
for _, raw := range attachmentsRaw {
|
||||
entry, ok := raw.(map[string]any)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
name := strings.TrimSpace(stringArg(entry, "name", "attachment"))
|
||||
path := strings.TrimSpace(stringArg(entry, "path", ""))
|
||||
if path == "" {
|
||||
continue
|
||||
}
|
||||
lines = append(lines, fmt.Sprintf("- %s: %s", name, path))
|
||||
}
|
||||
if len(lines) == 0 {
|
||||
return prompt
|
||||
}
|
||||
var builder strings.Builder
|
||||
builder.WriteString("User-selected local attachments:\n")
|
||||
builder.WriteString(strings.Join(lines, "\n"))
|
||||
builder.WriteString("\n\n")
|
||||
builder.WriteString(prompt)
|
||||
return builder.String()
|
||||
}
|
||||
|
||||
func composeHistoryPrompt(history []string) string {
|
||||
if len(history) == 0 {
|
||||
return ""
|
||||
}
|
||||
var builder strings.Builder
|
||||
for index, turn := range history {
|
||||
builder.WriteString(fmt.Sprintf("## User Turn %d\n", index+1))
|
||||
builder.WriteString(turn)
|
||||
builder.WriteString("\n\n")
|
||||
}
|
||||
return strings.TrimSpace(builder.String())
|
||||
}
|
||||
|
||||
func callOpenAICompatibleCtx(
|
||||
ctx context.Context,
|
||||
baseURL,
|
||||
apiKey,
|
||||
model string,
|
||||
messages []map[string]string,
|
||||
) (string, error) {
|
||||
payload := map[string]any{
|
||||
"model": model,
|
||||
"messages": messages,
|
||||
"max_tokens": 4096,
|
||||
"stream": false,
|
||||
}
|
||||
body, _ := json.Marshal(payload)
|
||||
request, err := http.NewRequestWithContext(
|
||||
ctx,
|
||||
http.MethodPost,
|
||||
strings.TrimRight(baseURL, "/")+"/chat/completions",
|
||||
bytes.NewReader(body),
|
||||
)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
request.Header.Set("Content-Type", "application/json")
|
||||
request.Header.Set("Authorization", "Bearer "+apiKey)
|
||||
|
||||
client := &http.Client{Timeout: 120 * time.Second}
|
||||
response, err := client.Do(request)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
defer response.Body.Close()
|
||||
responseBody, err := io.ReadAll(response.Body)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
if response.StatusCode < 200 || response.StatusCode >= 300 {
|
||||
return "", fmt.Errorf("api error %d: %s", response.StatusCode, strings.TrimSpace(string(responseBody)))
|
||||
}
|
||||
|
||||
var decoded map[string]any
|
||||
if err := json.Unmarshal(responseBody, &decoded); err != nil {
|
||||
return "", err
|
||||
}
|
||||
choices, _ := decoded["choices"].([]any)
|
||||
if len(choices) == 0 {
|
||||
return "", errors.New("missing choices in response")
|
||||
}
|
||||
choice, _ := choices[0].(map[string]any)
|
||||
message, _ := choice["message"].(map[string]any)
|
||||
content := strings.TrimSpace(fmt.Sprint(message["content"]))
|
||||
if content == "" || content == "<nil>" {
|
||||
return "", errors.New("empty response content")
|
||||
}
|
||||
return content, nil
|
||||
}
|
||||
|
||||
func decodeRpcRequest(payload []byte) (rpcRequest, error) {
|
||||
var request rpcRequest
|
||||
if err := json.Unmarshal(payload, &request); err != nil {
|
||||
return rpcRequest{}, fmt.Errorf("invalid json: %w", err)
|
||||
}
|
||||
if strings.TrimSpace(request.Method) == "" {
|
||||
return rpcRequest{}, errors.New("missing method")
|
||||
}
|
||||
if request.Params == nil {
|
||||
request.Params = map[string]any{}
|
||||
}
|
||||
return request, nil
|
||||
}
|
||||
|
||||
func writeSSE(w http.ResponseWriter, payload map[string]any) {
|
||||
encoded, _ := json.Marshal(payload)
|
||||
_, _ = fmt.Fprintf(w, "data: %s\n\n", encoded)
|
||||
}
|
||||
|
||||
func resultEnvelope(id any, result map[string]any) map[string]any {
|
||||
return map[string]any{
|
||||
"jsonrpc": "2.0",
|
||||
"id": id,
|
||||
"result": result,
|
||||
}
|
||||
}
|
||||
|
||||
func errorEnvelope(id any, code int, message string) map[string]any {
|
||||
return map[string]any{
|
||||
"jsonrpc": "2.0",
|
||||
"id": id,
|
||||
"error": map[string]any{
|
||||
"code": code,
|
||||
"message": message,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func notificationEnvelope(method string, params map[string]any) map[string]any {
|
||||
return map[string]any{
|
||||
"jsonrpc": "2.0",
|
||||
"method": method,
|
||||
"params": params,
|
||||
}
|
||||
}
|
||||
|
||||
func errorResponse(id any, code int, message string) map[string]any {
|
||||
return map[string]any{
|
||||
"jsonrpc": "2.0",
|
||||
@ -265,49 +1106,7 @@ func handleClaudeReviewTool(arguments map[string]any) (string, error) {
|
||||
}
|
||||
|
||||
func callOpenAICompatible(baseURL, apiKey, model string, messages []map[string]string) (string, error) {
|
||||
payload := map[string]any{
|
||||
"model": model,
|
||||
"messages": messages,
|
||||
"max_tokens": 4096,
|
||||
"stream": false,
|
||||
}
|
||||
body, _ := json.Marshal(payload)
|
||||
req, err := http.NewRequest(http.MethodPost, strings.TrimRight(baseURL, "/")+"/chat/completions", bytes.NewReader(body))
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
req.Header.Set("Authorization", "Bearer "+apiKey)
|
||||
|
||||
client := &http.Client{Timeout: 120 * time.Second}
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
respBody, err := io.ReadAll(resp.Body)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
|
||||
return "", fmt.Errorf("api error %d: %s", resp.StatusCode, strings.TrimSpace(string(respBody)))
|
||||
}
|
||||
|
||||
var decoded map[string]any
|
||||
if err := json.Unmarshal(respBody, &decoded); err != nil {
|
||||
return "", err
|
||||
}
|
||||
choices, _ := decoded["choices"].([]any)
|
||||
if len(choices) == 0 {
|
||||
return "", errors.New("missing choices in response")
|
||||
}
|
||||
choice, _ := choices[0].(map[string]any)
|
||||
message, _ := choice["message"].(map[string]any)
|
||||
content := strings.TrimSpace(fmt.Sprint(message["content"]))
|
||||
if content == "" || content == "<nil>" {
|
||||
return "", errors.New("empty response content")
|
||||
}
|
||||
return content, nil
|
||||
return callOpenAICompatibleCtx(context.Background(), baseURL, apiKey, model, messages)
|
||||
}
|
||||
|
||||
func runClaudeReview(prompt, model, system, tools string, timeout time.Duration) (string, error) {
|
||||
@ -416,6 +1215,23 @@ func stringArg(arguments map[string]any, key, fallback string) string {
|
||||
return text
|
||||
}
|
||||
|
||||
func listArg(arguments map[string]any, key string) []any {
|
||||
if arguments == nil {
|
||||
return nil
|
||||
}
|
||||
raw, ok := arguments[key]
|
||||
if !ok || raw == nil {
|
||||
return nil
|
||||
}
|
||||
if values, ok := raw.([]any); ok {
|
||||
return values
|
||||
}
|
||||
if values, ok := raw.([]interface{}); ok {
|
||||
return values
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func intArg(raw string, fallback int) int {
|
||||
var parsed int
|
||||
if _, err := fmt.Sscanf(raw, "%d", &parsed); err != nil || parsed <= 0 {
|
||||
@ -423,3 +1239,18 @@ func intArg(raw string, fallback int) int {
|
||||
}
|
||||
return parsed
|
||||
}
|
||||
|
||||
func boolArg(raw string, fallback bool) bool {
|
||||
trimmed := strings.TrimSpace(strings.ToLower(raw))
|
||||
if trimmed == "" {
|
||||
return fallback
|
||||
}
|
||||
switch trimmed {
|
||||
case "1", "true", "yes", "on":
|
||||
return true
|
||||
case "0", "false", "no", "off":
|
||||
return false
|
||||
default:
|
||||
return fallback
|
||||
}
|
||||
}
|
||||
|
||||
@ -2041,12 +2041,16 @@ class AppController extends ChangeNotifier {
|
||||
);
|
||||
if (archived) {
|
||||
unawaited(
|
||||
_gatewayAcpClient
|
||||
.closeSession(
|
||||
_enqueueThreadTurn<void>(normalizedSessionKey, () async {
|
||||
try {
|
||||
await _gatewayAcpClient.closeSession(
|
||||
sessionId: normalizedSessionKey,
|
||||
threadId: normalizedSessionKey,
|
||||
)
|
||||
.catchError((_) {}),
|
||||
);
|
||||
} catch (_) {
|
||||
// Best effort only.
|
||||
}
|
||||
}).catchError((_) {}),
|
||||
);
|
||||
}
|
||||
_upsertAssistantThreadRecord(
|
||||
@ -2236,6 +2240,7 @@ class AppController extends ChangeNotifier {
|
||||
Future<void> clearAssistantLocalState() async {
|
||||
await _flushAssistantThreadPersistence();
|
||||
await _store.clearAssistantLocalState();
|
||||
await _store.saveAssistantThreadRecords(const <AssistantThreadRecord>[]);
|
||||
_assistantThreadPersistQueue = Future<void>.value();
|
||||
final defaults = SettingsSnapshot.defaults();
|
||||
_assistantThreadRecords.clear();
|
||||
@ -4645,12 +4650,6 @@ class AppController extends ChangeNotifier {
|
||||
}
|
||||
|
||||
Uri? _resolveAcpEndpoint() {
|
||||
final aiGatewayBase = _normalizeAiGatewayBaseUrl(
|
||||
settings.aiGateway.baseUrl,
|
||||
);
|
||||
if (aiGatewayBase != null) {
|
||||
return aiGatewayBase;
|
||||
}
|
||||
final target = assistantExecutionTargetForSession(
|
||||
_sessionsController.currentSessionKey,
|
||||
);
|
||||
|
||||
@ -1,189 +0,0 @@
|
||||
import 'dart:async';
|
||||
import 'dart:convert';
|
||||
import 'dart:io';
|
||||
|
||||
import 'multi_agent_orchestrator.dart';
|
||||
import 'runtime_models.dart';
|
||||
|
||||
class AgentCliBridgeRequest {
|
||||
const AgentCliBridgeRequest({
|
||||
required this.sessionId,
|
||||
required this.taskPrompt,
|
||||
required this.workingDirectory,
|
||||
required this.attachments,
|
||||
required this.selectedSkills,
|
||||
required this.aiGatewayBaseUrl,
|
||||
required this.aiGatewayApiKey,
|
||||
});
|
||||
|
||||
final String sessionId;
|
||||
final String taskPrompt;
|
||||
final String workingDirectory;
|
||||
final List<CollaborationAttachment> attachments;
|
||||
final List<String> selectedSkills;
|
||||
final String aiGatewayBaseUrl;
|
||||
final String aiGatewayApiKey;
|
||||
}
|
||||
|
||||
class AgentCliBridgeResult {
|
||||
const AgentCliBridgeResult({
|
||||
required this.output,
|
||||
required this.success,
|
||||
required this.errorMessage,
|
||||
this.events = const <MultiAgentRunEvent>[],
|
||||
});
|
||||
|
||||
final String output;
|
||||
final bool success;
|
||||
final String errorMessage;
|
||||
final List<MultiAgentRunEvent> events;
|
||||
}
|
||||
|
||||
abstract class AgentCliBridge {
|
||||
Future<AgentCliBridgeResult> run(AgentCliBridgeRequest request);
|
||||
}
|
||||
|
||||
class SubprocessCliBridge implements AgentCliBridge {
|
||||
const SubprocessCliBridge({
|
||||
required this.command,
|
||||
this.defaultArgs = const <String>[],
|
||||
});
|
||||
|
||||
final String command;
|
||||
final List<String> defaultArgs;
|
||||
|
||||
@override
|
||||
Future<AgentCliBridgeResult> run(AgentCliBridgeRequest request) async {
|
||||
try {
|
||||
final process = await Process.start(
|
||||
command,
|
||||
<String>[...defaultArgs, request.taskPrompt],
|
||||
workingDirectory: request.workingDirectory.trim().isEmpty
|
||||
? null
|
||||
: request.workingDirectory,
|
||||
);
|
||||
await process.stdin.close();
|
||||
final stdout = await process.stdout.transform(utf8.decoder).join();
|
||||
final stderr = await process.stderr.transform(utf8.decoder).join();
|
||||
final exitCode = await process.exitCode;
|
||||
return AgentCliBridgeResult(
|
||||
output: stdout.trim(),
|
||||
success: exitCode == 0,
|
||||
errorMessage: stderr.trim(),
|
||||
);
|
||||
} catch (error) {
|
||||
return AgentCliBridgeResult(
|
||||
output: '',
|
||||
success: false,
|
||||
errorMessage: error.toString(),
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
class JsonRpcCliBridge implements AgentCliBridge {
|
||||
const JsonRpcCliBridge(this.endpoint);
|
||||
|
||||
final Uri endpoint;
|
||||
|
||||
@override
|
||||
Future<AgentCliBridgeResult> run(AgentCliBridgeRequest request) async {
|
||||
final socket = await WebSocket.connect(endpoint.toString());
|
||||
final requestId = DateTime.now().microsecondsSinceEpoch.toString();
|
||||
final completer = Completer<AgentCliBridgeResult>();
|
||||
final events = <MultiAgentRunEvent>[];
|
||||
|
||||
socket.listen(
|
||||
(raw) {
|
||||
final json = jsonDecode(raw as String) as Map<String, dynamic>;
|
||||
final method = json['method'] as String?;
|
||||
if (method == 'multi_agent.event') {
|
||||
final params =
|
||||
(json['params'] as Map?)?.cast<String, dynamic>() ??
|
||||
const <String, dynamic>{};
|
||||
events.add(MultiAgentRunEvent.fromJson(params));
|
||||
return;
|
||||
}
|
||||
if (json['id']?.toString() == requestId && json['result'] is Map) {
|
||||
final result = (json['result'] as Map).cast<String, dynamic>();
|
||||
if (!completer.isCompleted) {
|
||||
completer.complete(
|
||||
AgentCliBridgeResult(
|
||||
output: result['summary']?.toString() ?? '',
|
||||
success: result['success'] == true,
|
||||
errorMessage: result['error']?.toString() ?? '',
|
||||
events: events,
|
||||
),
|
||||
);
|
||||
}
|
||||
unawaited(socket.close());
|
||||
return;
|
||||
}
|
||||
if (json['error'] is Map && !completer.isCompleted) {
|
||||
final error = (json['error'] as Map).cast<String, dynamic>();
|
||||
completer.complete(
|
||||
AgentCliBridgeResult(
|
||||
output: '',
|
||||
success: false,
|
||||
errorMessage: error['message']?.toString() ?? 'JSON-RPC error',
|
||||
events: events,
|
||||
),
|
||||
);
|
||||
unawaited(socket.close());
|
||||
}
|
||||
},
|
||||
onError: (error, _) {
|
||||
if (!completer.isCompleted) {
|
||||
completer.complete(
|
||||
AgentCliBridgeResult(
|
||||
output: '',
|
||||
success: false,
|
||||
errorMessage: error.toString(),
|
||||
events: events,
|
||||
),
|
||||
);
|
||||
}
|
||||
},
|
||||
onDone: () {
|
||||
if (!completer.isCompleted) {
|
||||
completer.complete(
|
||||
AgentCliBridgeResult(
|
||||
output: '',
|
||||
success: false,
|
||||
errorMessage: 'JSON-RPC bridge closed before completion',
|
||||
events: events,
|
||||
),
|
||||
);
|
||||
}
|
||||
},
|
||||
cancelOnError: true,
|
||||
);
|
||||
|
||||
socket.add(
|
||||
jsonEncode(<String, dynamic>{
|
||||
'jsonrpc': '2.0',
|
||||
'id': requestId,
|
||||
'method': 'session.start',
|
||||
'params': <String, dynamic>{
|
||||
'sessionId': request.sessionId,
|
||||
'taskPrompt': request.taskPrompt,
|
||||
'workingDirectory': request.workingDirectory,
|
||||
'attachments': request.attachments
|
||||
.map(
|
||||
(item) => <String, dynamic>{
|
||||
'name': item.name,
|
||||
'description': item.description,
|
||||
'path': item.path,
|
||||
},
|
||||
)
|
||||
.toList(growable: false),
|
||||
'selectedSkills': request.selectedSkills,
|
||||
'aiGatewayBaseUrl': request.aiGatewayBaseUrl,
|
||||
'aiGatewayApiKey': request.aiGatewayApiKey,
|
||||
},
|
||||
}),
|
||||
);
|
||||
|
||||
return completer.future;
|
||||
}
|
||||
}
|
||||
@ -1,484 +0,0 @@
|
||||
import 'dart:async';
|
||||
import 'dart:convert';
|
||||
import 'dart:io';
|
||||
|
||||
import 'multi_agent_orchestrator.dart';
|
||||
import 'runtime_models.dart';
|
||||
|
||||
class MultiAgentBrokerServer {
|
||||
MultiAgentBrokerServer(this._orchestrator);
|
||||
|
||||
final MultiAgentOrchestrator _orchestrator;
|
||||
final Map<String, _BrokerSessionState> _sessions =
|
||||
<String, _BrokerSessionState>{};
|
||||
HttpServer? _server;
|
||||
|
||||
bool get isRunning => _server != null;
|
||||
|
||||
Uri? get wsUri => _server == null
|
||||
? null
|
||||
: Uri.parse('ws://127.0.0.1:${_server!.port}/multi-agent-broker');
|
||||
|
||||
Future<void> start() async {
|
||||
if (_server != null) {
|
||||
return;
|
||||
}
|
||||
_server = await HttpServer.bind(InternetAddress.loopbackIPv4, 0);
|
||||
unawaited(_listen());
|
||||
}
|
||||
|
||||
Future<void> stop() async {
|
||||
final server = _server;
|
||||
_server = null;
|
||||
_sessions.clear();
|
||||
await server?.close(force: true);
|
||||
}
|
||||
|
||||
Future<void> _listen() async {
|
||||
final server = _server;
|
||||
if (server == null) {
|
||||
return;
|
||||
}
|
||||
await for (final request in server) {
|
||||
if (request.uri.path != '/multi-agent-broker' ||
|
||||
!WebSocketTransformer.isUpgradeRequest(request)) {
|
||||
request.response
|
||||
..statusCode = HttpStatus.notFound
|
||||
..close();
|
||||
continue;
|
||||
}
|
||||
final socket = await WebSocketTransformer.upgrade(request);
|
||||
unawaited(_handleSocket(socket));
|
||||
}
|
||||
}
|
||||
|
||||
Future<void> _handleSocket(WebSocket socket) async {
|
||||
await for (final raw in socket) {
|
||||
try {
|
||||
final json = jsonDecode(raw as String) as Map<String, dynamic>;
|
||||
final method = json['method'] as String? ?? '';
|
||||
final id = json['id'];
|
||||
final params =
|
||||
(json['params'] as Map?)?.cast<String, dynamic>() ??
|
||||
const <String, dynamic>{};
|
||||
switch (method) {
|
||||
case 'run.start':
|
||||
await _handleRunStart(socket, id, params);
|
||||
break;
|
||||
case 'session.start':
|
||||
await _handleSessionStart(socket, id, params);
|
||||
break;
|
||||
case 'session.message':
|
||||
await _handleSessionMessage(socket, id, params);
|
||||
break;
|
||||
case 'session.cancel':
|
||||
await _orchestrator.abort();
|
||||
_writeResult(
|
||||
socket,
|
||||
id,
|
||||
<String, dynamic>{'accepted': true, 'cancelled': true},
|
||||
);
|
||||
break;
|
||||
case 'session.close':
|
||||
final sessionId = params['sessionId']?.toString().trim() ?? '';
|
||||
if (sessionId.isNotEmpty) {
|
||||
_sessions.remove(sessionId);
|
||||
}
|
||||
_writeResult(
|
||||
socket,
|
||||
id,
|
||||
<String, dynamic>{'accepted': true, 'closed': true},
|
||||
);
|
||||
break;
|
||||
default:
|
||||
_writeError(socket, id, -32601, 'Method not found');
|
||||
}
|
||||
} catch (error) {
|
||||
_writeError(socket, null, -32000, error.toString());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Future<void> _handleRunStart(
|
||||
WebSocket socket,
|
||||
Object? id,
|
||||
Map<String, dynamic> params,
|
||||
) async {
|
||||
final result = await _orchestrator.runCollaboration(
|
||||
taskPrompt: params['taskPrompt'] as String? ?? '',
|
||||
workingDirectory: params['workingDirectory'] as String? ?? '',
|
||||
attachments: _parseAttachments(params['attachments']),
|
||||
selectedSkills: _parseSelectedSkills(params['selectedSkills']),
|
||||
aiGatewayBaseUrl: params['aiGatewayBaseUrl'] as String? ?? '',
|
||||
aiGatewayApiKey: params['aiGatewayApiKey'] as String? ?? '',
|
||||
onEvent: (event) => _emitEvent(socket, event),
|
||||
);
|
||||
_writeResult(socket, id, result.toJson());
|
||||
}
|
||||
|
||||
Future<void> _handleSessionStart(
|
||||
WebSocket socket,
|
||||
Object? id,
|
||||
Map<String, dynamic> params,
|
||||
) async {
|
||||
final sessionId = params['sessionId']?.toString().trim() ?? '';
|
||||
if (sessionId.isEmpty) {
|
||||
_writeError(socket, id, -32602, 'sessionId is required');
|
||||
return;
|
||||
}
|
||||
final state = _BrokerSessionState(
|
||||
sessionId: sessionId,
|
||||
workingDirectory: params['workingDirectory'] as String? ?? '',
|
||||
attachments: _parseAttachments(params['attachments']),
|
||||
selectedSkills: _parseSelectedSkills(params['selectedSkills']),
|
||||
aiGatewayBaseUrl: params['aiGatewayBaseUrl'] as String? ?? '',
|
||||
aiGatewayApiKey: params['aiGatewayApiKey'] as String? ?? '',
|
||||
history: <String>[],
|
||||
);
|
||||
_sessions[sessionId] = state;
|
||||
await _runSession(socket, id, state, params['taskPrompt'] as String? ?? '');
|
||||
}
|
||||
|
||||
Future<void> _handleSessionMessage(
|
||||
WebSocket socket,
|
||||
Object? id,
|
||||
Map<String, dynamic> params,
|
||||
) async {
|
||||
final sessionId = params['sessionId']?.toString().trim() ?? '';
|
||||
if (sessionId.isEmpty) {
|
||||
_writeError(socket, id, -32602, 'sessionId is required');
|
||||
return;
|
||||
}
|
||||
final state = _sessions.putIfAbsent(
|
||||
sessionId,
|
||||
() => _BrokerSessionState(
|
||||
sessionId: sessionId,
|
||||
workingDirectory: params['workingDirectory'] as String? ?? '',
|
||||
attachments: _parseAttachments(params['attachments']),
|
||||
selectedSkills: _parseSelectedSkills(params['selectedSkills']),
|
||||
aiGatewayBaseUrl: params['aiGatewayBaseUrl'] as String? ?? '',
|
||||
aiGatewayApiKey: params['aiGatewayApiKey'] as String? ?? '',
|
||||
history: <String>[],
|
||||
),
|
||||
);
|
||||
final workingDirectory = params['workingDirectory'] as String? ?? '';
|
||||
if (workingDirectory.trim().isNotEmpty) {
|
||||
state.workingDirectory = workingDirectory;
|
||||
}
|
||||
final attachments = _parseAttachments(params['attachments']);
|
||||
if (attachments.isNotEmpty) {
|
||||
state.attachments = attachments;
|
||||
}
|
||||
final selectedSkills = _parseSelectedSkills(params['selectedSkills']);
|
||||
if (selectedSkills.isNotEmpty) {
|
||||
state.selectedSkills = selectedSkills;
|
||||
}
|
||||
final aiGatewayBaseUrl = params['aiGatewayBaseUrl'] as String? ?? '';
|
||||
if (aiGatewayBaseUrl.trim().isNotEmpty) {
|
||||
state.aiGatewayBaseUrl = aiGatewayBaseUrl;
|
||||
}
|
||||
final aiGatewayApiKey = params['aiGatewayApiKey'] as String? ?? '';
|
||||
if (aiGatewayApiKey.trim().isNotEmpty) {
|
||||
state.aiGatewayApiKey = aiGatewayApiKey;
|
||||
}
|
||||
await _runSession(socket, id, state, params['taskPrompt'] as String? ?? '');
|
||||
}
|
||||
|
||||
Future<void> _runSession(
|
||||
WebSocket socket,
|
||||
Object? id,
|
||||
_BrokerSessionState state,
|
||||
String taskPrompt,
|
||||
) async {
|
||||
final trimmedPrompt = taskPrompt.trim();
|
||||
if (trimmedPrompt.isNotEmpty) {
|
||||
state.history.add(trimmedPrompt);
|
||||
}
|
||||
final composedPrompt = _composeSessionPrompt(state.history);
|
||||
final result = await _orchestrator.runCollaboration(
|
||||
taskPrompt: composedPrompt,
|
||||
workingDirectory: state.workingDirectory,
|
||||
attachments: state.attachments,
|
||||
selectedSkills: state.selectedSkills,
|
||||
aiGatewayBaseUrl: state.aiGatewayBaseUrl,
|
||||
aiGatewayApiKey: state.aiGatewayApiKey,
|
||||
onEvent: (event) => _emitEvent(socket, event),
|
||||
);
|
||||
_writeResult(
|
||||
socket,
|
||||
id,
|
||||
<String, dynamic>{...result.toJson(), 'sessionId': state.sessionId},
|
||||
);
|
||||
}
|
||||
|
||||
String _composeSessionPrompt(List<String> history) {
|
||||
if (history.isEmpty) {
|
||||
return '';
|
||||
}
|
||||
final buffer = StringBuffer();
|
||||
for (var index = 0; index < history.length; index++) {
|
||||
final turn = index + 1;
|
||||
buffer.writeln('## User Turn $turn');
|
||||
buffer.writeln(history[index]);
|
||||
buffer.writeln();
|
||||
}
|
||||
return buffer.toString().trim();
|
||||
}
|
||||
|
||||
List<CollaborationAttachment> _parseAttachments(Object? raw) {
|
||||
return ((raw as List?) ?? const <Object>[])
|
||||
.whereType<Map>()
|
||||
.map(
|
||||
(item) => CollaborationAttachment(
|
||||
name: item['name']?.toString() ?? '',
|
||||
description: item['description']?.toString() ?? '',
|
||||
path: item['path']?.toString() ?? '',
|
||||
),
|
||||
)
|
||||
.toList(growable: false);
|
||||
}
|
||||
|
||||
List<String> _parseSelectedSkills(Object? raw) {
|
||||
return ((raw as List?) ?? const <Object>[])
|
||||
.map((item) => item.toString())
|
||||
.toList(growable: false);
|
||||
}
|
||||
|
||||
void _emitEvent(WebSocket socket, MultiAgentRunEvent event) {
|
||||
socket.add(
|
||||
jsonEncode(<String, dynamic>{
|
||||
'jsonrpc': '2.0',
|
||||
'method': 'multi_agent.event',
|
||||
'params': event.toJson(),
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
void _writeResult(WebSocket socket, Object? id, Map<String, dynamic> result) {
|
||||
socket.add(
|
||||
jsonEncode(<String, dynamic>{
|
||||
'jsonrpc': '2.0',
|
||||
'id': id,
|
||||
'result': result,
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
void _writeError(
|
||||
WebSocket socket,
|
||||
Object? id,
|
||||
int code,
|
||||
String message,
|
||||
) {
|
||||
socket.add(
|
||||
jsonEncode(<String, dynamic>{
|
||||
'jsonrpc': '2.0',
|
||||
'id': id,
|
||||
'error': <String, dynamic>{'code': code, 'message': message},
|
||||
}),
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
class MultiAgentBrokerClient {
|
||||
MultiAgentBrokerClient(this._uri);
|
||||
|
||||
final Uri _uri;
|
||||
|
||||
Stream<MultiAgentRunEvent> runTask({
|
||||
required String taskPrompt,
|
||||
required String workingDirectory,
|
||||
required List<CollaborationAttachment> attachments,
|
||||
required List<String> selectedSkills,
|
||||
required String aiGatewayBaseUrl,
|
||||
required String aiGatewayApiKey,
|
||||
}) {
|
||||
return _streamRequest(
|
||||
method: 'run.start',
|
||||
params: <String, dynamic>{
|
||||
'taskPrompt': taskPrompt,
|
||||
'workingDirectory': workingDirectory,
|
||||
'attachments': _encodeAttachments(attachments),
|
||||
'selectedSkills': selectedSkills,
|
||||
'aiGatewayBaseUrl': aiGatewayBaseUrl,
|
||||
'aiGatewayApiKey': aiGatewayApiKey,
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
Stream<MultiAgentRunEvent> startSession({
|
||||
required String sessionId,
|
||||
required String taskPrompt,
|
||||
required String workingDirectory,
|
||||
required List<CollaborationAttachment> attachments,
|
||||
required List<String> selectedSkills,
|
||||
required String aiGatewayBaseUrl,
|
||||
required String aiGatewayApiKey,
|
||||
}) {
|
||||
return _streamRequest(
|
||||
method: 'session.start',
|
||||
params: <String, dynamic>{
|
||||
'sessionId': sessionId,
|
||||
'taskPrompt': taskPrompt,
|
||||
'workingDirectory': workingDirectory,
|
||||
'attachments': _encodeAttachments(attachments),
|
||||
'selectedSkills': selectedSkills,
|
||||
'aiGatewayBaseUrl': aiGatewayBaseUrl,
|
||||
'aiGatewayApiKey': aiGatewayApiKey,
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
Stream<MultiAgentRunEvent> sendSessionMessage({
|
||||
required String sessionId,
|
||||
required String taskPrompt,
|
||||
required String workingDirectory,
|
||||
required List<CollaborationAttachment> attachments,
|
||||
required List<String> selectedSkills,
|
||||
required String aiGatewayBaseUrl,
|
||||
required String aiGatewayApiKey,
|
||||
}) {
|
||||
return _streamRequest(
|
||||
method: 'session.message',
|
||||
params: <String, dynamic>{
|
||||
'sessionId': sessionId,
|
||||
'taskPrompt': taskPrompt,
|
||||
'workingDirectory': workingDirectory,
|
||||
'attachments': _encodeAttachments(attachments),
|
||||
'selectedSkills': selectedSkills,
|
||||
'aiGatewayBaseUrl': aiGatewayBaseUrl,
|
||||
'aiGatewayApiKey': aiGatewayApiKey,
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
Future<void> cancelSession(String sessionId) async {
|
||||
await _requestOnly(
|
||||
method: 'session.cancel',
|
||||
params: <String, dynamic>{'sessionId': sessionId},
|
||||
);
|
||||
}
|
||||
|
||||
Future<void> closeSession(String sessionId) async {
|
||||
await _requestOnly(
|
||||
method: 'session.close',
|
||||
params: <String, dynamic>{'sessionId': sessionId},
|
||||
);
|
||||
}
|
||||
|
||||
Stream<MultiAgentRunEvent> _streamRequest({
|
||||
required String method,
|
||||
required Map<String, dynamic> params,
|
||||
}) async* {
|
||||
final socket = await WebSocket.connect(_uri.toString());
|
||||
final controller = StreamController<MultiAgentRunEvent>();
|
||||
final requestId = DateTime.now().microsecondsSinceEpoch.toString();
|
||||
|
||||
socket.listen(
|
||||
(raw) {
|
||||
final json = jsonDecode(raw as String) as Map<String, dynamic>;
|
||||
final rpcMethod = json['method'] as String?;
|
||||
if (rpcMethod == 'multi_agent.event') {
|
||||
final eventParams =
|
||||
(json['params'] as Map?)?.cast<String, dynamic>() ??
|
||||
const <String, dynamic>{};
|
||||
controller.add(MultiAgentRunEvent.fromJson(eventParams));
|
||||
return;
|
||||
}
|
||||
if (json['id']?.toString() == requestId && json['result'] is Map) {
|
||||
final result = (json['result'] as Map).cast<String, dynamic>();
|
||||
controller.add(
|
||||
MultiAgentRunEvent(
|
||||
type: 'result',
|
||||
title: 'Multi-Agent',
|
||||
message: result['success'] == true
|
||||
? 'Collaboration completed.'
|
||||
: 'Collaboration failed.',
|
||||
pending: false,
|
||||
error: result['success'] != true,
|
||||
data: result,
|
||||
),
|
||||
);
|
||||
unawaited(controller.close());
|
||||
unawaited(socket.close());
|
||||
return;
|
||||
}
|
||||
if (json['error'] is Map) {
|
||||
final error = (json['error'] as Map).cast<String, dynamic>();
|
||||
controller.add(
|
||||
MultiAgentRunEvent(
|
||||
type: 'error',
|
||||
title: 'Multi-Agent',
|
||||
message: error['message']?.toString() ?? 'Broker error',
|
||||
pending: false,
|
||||
error: true,
|
||||
),
|
||||
);
|
||||
unawaited(controller.close());
|
||||
unawaited(socket.close());
|
||||
}
|
||||
},
|
||||
onError: controller.addError,
|
||||
onDone: () {
|
||||
if (!controller.isClosed) {
|
||||
controller.close();
|
||||
}
|
||||
},
|
||||
cancelOnError: true,
|
||||
);
|
||||
|
||||
socket.add(
|
||||
jsonEncode(<String, dynamic>{
|
||||
'jsonrpc': '2.0',
|
||||
'id': requestId,
|
||||
'method': method,
|
||||
'params': params,
|
||||
}),
|
||||
);
|
||||
|
||||
yield* controller.stream;
|
||||
}
|
||||
|
||||
Future<void> _requestOnly({
|
||||
required String method,
|
||||
required Map<String, dynamic> params,
|
||||
}) async {
|
||||
await for (final _ in _streamRequest(method: method, params: params)) {
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
List<Map<String, dynamic>> _encodeAttachments(
|
||||
List<CollaborationAttachment> attachments,
|
||||
) {
|
||||
return attachments
|
||||
.map(
|
||||
(item) => <String, dynamic>{
|
||||
'name': item.name,
|
||||
'description': item.description,
|
||||
'path': item.path,
|
||||
},
|
||||
)
|
||||
.toList(growable: false);
|
||||
}
|
||||
}
|
||||
|
||||
class _BrokerSessionState {
|
||||
_BrokerSessionState({
|
||||
required this.sessionId,
|
||||
required this.workingDirectory,
|
||||
required this.attachments,
|
||||
required this.selectedSkills,
|
||||
required this.aiGatewayBaseUrl,
|
||||
required this.aiGatewayApiKey,
|
||||
required this.history,
|
||||
});
|
||||
|
||||
final String sessionId;
|
||||
String workingDirectory;
|
||||
List<CollaborationAttachment> attachments;
|
||||
List<String> selectedSkills;
|
||||
String aiGatewayBaseUrl;
|
||||
String aiGatewayApiKey;
|
||||
final List<String> history;
|
||||
}
|
||||
@ -1,69 +0,0 @@
|
||||
@TestOn('vm')
|
||||
library;
|
||||
|
||||
import 'package:flutter_test/flutter_test.dart';
|
||||
import 'package:xworkmate/runtime/agent_cli_bridge.dart';
|
||||
import 'package:xworkmate/runtime/multi_agent_broker.dart';
|
||||
import 'package:xworkmate/runtime/multi_agent_orchestrator.dart';
|
||||
import 'package:xworkmate/runtime/runtime_models.dart';
|
||||
|
||||
void main() {
|
||||
TestWidgetsFlutterBinding.ensureInitialized();
|
||||
|
||||
test('JsonRpcCliBridge can drive a broker-backed external session', () async {
|
||||
final server = MultiAgentBrokerServer(_BridgeFakeOrchestrator());
|
||||
await server.start();
|
||||
addTearDown(server.stop);
|
||||
|
||||
final bridge = JsonRpcCliBridge(server.wsUri!);
|
||||
final result = await bridge.run(
|
||||
const AgentCliBridgeRequest(
|
||||
sessionId: 'bridge-session',
|
||||
taskPrompt: 'hello bridge',
|
||||
workingDirectory: '/tmp',
|
||||
attachments: <CollaborationAttachment>[],
|
||||
selectedSkills: <String>['aris'],
|
||||
aiGatewayBaseUrl: '',
|
||||
aiGatewayApiKey: '',
|
||||
),
|
||||
);
|
||||
|
||||
expect(result.success, isTrue);
|
||||
expect(result.events, isNotEmpty);
|
||||
});
|
||||
}
|
||||
|
||||
class _BridgeFakeOrchestrator extends MultiAgentOrchestrator {
|
||||
_BridgeFakeOrchestrator()
|
||||
: super(config: MultiAgentConfig.defaults().copyWith(enabled: true));
|
||||
|
||||
@override
|
||||
Future<CollaborationResult> runCollaboration({
|
||||
required String taskPrompt,
|
||||
required String workingDirectory,
|
||||
List<CollaborationAttachment> attachments = const [],
|
||||
List<String> selectedSkills = const [],
|
||||
String aiGatewayBaseUrl = '',
|
||||
String aiGatewayApiKey = '',
|
||||
void Function(MultiAgentRunEvent event)? onEvent,
|
||||
}) async {
|
||||
onEvent?.call(
|
||||
const MultiAgentRunEvent(
|
||||
type: 'step',
|
||||
title: 'Engineer',
|
||||
message: 'running',
|
||||
pending: false,
|
||||
error: false,
|
||||
role: 'engineer',
|
||||
),
|
||||
);
|
||||
return const CollaborationResult(
|
||||
success: true,
|
||||
steps: <CollaborationStep>[],
|
||||
finalCode: 'ok',
|
||||
finalScore: 7,
|
||||
duration: Duration(milliseconds: 10),
|
||||
iterations: 0,
|
||||
);
|
||||
}
|
||||
}
|
||||
@ -1,7 +0,0 @@
|
||||
import '../test_suite_stub.dart'
|
||||
if (dart.library.io) 'agent_cli_bridge_suite.dart'
|
||||
as suite;
|
||||
|
||||
void main() {
|
||||
suite.main();
|
||||
}
|
||||
@ -9,6 +9,7 @@ import 'package:flutter_test/flutter_test.dart';
|
||||
import 'package:shared_preferences/shared_preferences.dart';
|
||||
import 'package:xworkmate/app/app_controller.dart';
|
||||
import 'package:xworkmate/runtime/runtime_models.dart';
|
||||
import 'package:xworkmate/runtime/secure_config_store.dart';
|
||||
|
||||
void main() {
|
||||
test(
|
||||
@ -16,8 +17,21 @@ void main() {
|
||||
() async {
|
||||
SharedPreferences.setMockInitialValues(<String, Object>{});
|
||||
final gateway = await _FakeGatewayServer.start();
|
||||
final controller = AppController();
|
||||
addTearDown(controller.dispose);
|
||||
final tempDirectory = await Directory.systemTemp.createTemp(
|
||||
'xworkmate-assistant-flow-',
|
||||
);
|
||||
addTearDown(() async {
|
||||
await _deleteDirectoryWithRetry(tempDirectory);
|
||||
});
|
||||
final store = SecureConfigStore(
|
||||
enableSecureStorage: false,
|
||||
databasePathResolver: () async => '${tempDirectory.path}/settings.db',
|
||||
fallbackDirectoryPathResolver: () async => tempDirectory.path,
|
||||
);
|
||||
final controller = AppController(store: store);
|
||||
addTearDown(() async {
|
||||
controller.dispose();
|
||||
});
|
||||
addTearDown(gateway.close);
|
||||
|
||||
await _waitFor(() => !controller.initializing);
|
||||
@ -103,6 +117,24 @@ class _FakeGatewayServer {
|
||||
|
||||
Future<void> _serve() async {
|
||||
await for (final request in _server) {
|
||||
if (request.uri.path == '/acp/rpc' && request.method == 'POST') {
|
||||
await _serveAcpRpc(request);
|
||||
continue;
|
||||
}
|
||||
if (request.uri.path == '/acp' &&
|
||||
WebSocketTransformer.isUpgradeRequest(request)) {
|
||||
final acpSocket = await WebSocketTransformer.upgrade(request);
|
||||
await acpSocket.close(
|
||||
WebSocketStatus.normalClosure,
|
||||
'test gateway runtime only',
|
||||
);
|
||||
continue;
|
||||
}
|
||||
if (!WebSocketTransformer.isUpgradeRequest(request)) {
|
||||
request.response.statusCode = HttpStatus.notFound;
|
||||
await request.response.close();
|
||||
continue;
|
||||
}
|
||||
final socket = await WebSocketTransformer.upgrade(request);
|
||||
_socket = socket;
|
||||
_send(socket, <String, dynamic>{
|
||||
@ -277,6 +309,35 @@ class _FakeGatewayServer {
|
||||
}
|
||||
}
|
||||
|
||||
Future<void> _serveAcpRpc(HttpRequest request) async {
|
||||
final body = await utf8.decodeStream(request);
|
||||
final envelope = (jsonDecode(body) as Map).cast<String, dynamic>();
|
||||
final id = envelope['id'];
|
||||
final method = envelope['method']?.toString() ?? '';
|
||||
final response = <String, dynamic>{
|
||||
'jsonrpc': '2.0',
|
||||
'id': id,
|
||||
'result': method == 'acp.capabilities'
|
||||
? <String, dynamic>{
|
||||
'singleAgent': true,
|
||||
'multiAgent': true,
|
||||
'providers': <String>['claude', 'codex', 'gemini', 'opencode'],
|
||||
'capabilities': <String, dynamic>{
|
||||
'single_agent': true,
|
||||
'multi_agent': true,
|
||||
'providers': <String>['claude', 'codex', 'gemini', 'opencode'],
|
||||
},
|
||||
}
|
||||
: const <String, dynamic>{},
|
||||
};
|
||||
request.response.headers.set(
|
||||
HttpHeaders.contentTypeHeader,
|
||||
'text/event-stream; charset=utf-8',
|
||||
);
|
||||
request.response.write('data: ${jsonEncode(response)}\n\n');
|
||||
await request.response.close();
|
||||
}
|
||||
|
||||
Future<void> _emitAssistantResult(
|
||||
WebSocket socket, {
|
||||
required String runId,
|
||||
@ -336,6 +397,23 @@ class _FakeGatewayServer {
|
||||
}
|
||||
}
|
||||
|
||||
Future<void> _deleteDirectoryWithRetry(Directory directory) async {
|
||||
if (!await directory.exists()) {
|
||||
return;
|
||||
}
|
||||
for (var attempt = 0; attempt < 3; attempt += 1) {
|
||||
try {
|
||||
await directory.delete(recursive: true);
|
||||
return;
|
||||
} on FileSystemException {
|
||||
if (attempt == 2) {
|
||||
rethrow;
|
||||
}
|
||||
await Future<void>.delayed(const Duration(milliseconds: 100));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Future<void> _waitFor(
|
||||
bool Function() predicate, {
|
||||
Duration timeout = const Duration(seconds: 5),
|
||||
|
||||
@ -236,15 +236,21 @@ void main() {
|
||||
await controller.setAssistantExecutionTarget(
|
||||
AssistantExecutionTarget.local,
|
||||
);
|
||||
final expectedLocalProfile =
|
||||
controller.settings.primaryLocalGatewayProfile;
|
||||
|
||||
expect(
|
||||
gateway.connectedProfiles.last,
|
||||
isA<GatewayConnectionProfile>()
|
||||
.having((item) => item.mode, 'mode', RuntimeConnectionMode.local)
|
||||
.having((item) => item.host, 'host', '127.0.0.1')
|
||||
.having((item) => item.port, 'port', 18789)
|
||||
.having((item) => item.host, 'host', expectedLocalProfile.host)
|
||||
.having((item) => item.port, 'port', expectedLocalProfile.port)
|
||||
.having((item) => item.tls, 'tls', isFalse)
|
||||
.having((item) => item.selectedAgentId, 'selectedAgentId', ''),
|
||||
.having(
|
||||
(item) => item.selectedAgentId,
|
||||
'selectedAgentId',
|
||||
expectedLocalProfile.selectedAgentId,
|
||||
),
|
||||
);
|
||||
expect(
|
||||
controller.settings.assistantExecutionTarget,
|
||||
@ -272,8 +278,7 @@ void main() {
|
||||
expect(
|
||||
controller.settings.primaryRemoteGatewayProfile.host,
|
||||
'gateway.example.com',
|
||||
reason:
|
||||
'Single Agent mode should preserve the saved remote endpoint.',
|
||||
reason: 'Single Agent mode should preserve the saved remote endpoint.',
|
||||
);
|
||||
expect(controller.settings.primaryRemoteGatewayProfile.port, 9443);
|
||||
expect(controller.settings.primaryRemoteGatewayProfile.tls, isTrue);
|
||||
@ -771,7 +776,12 @@ void main() {
|
||||
AssistantExecutionTarget.local,
|
||||
);
|
||||
expect(controller.assistantConnectionStatusLabel, '已连接');
|
||||
expect(controller.assistantConnectionTargetLabel, '127.0.0.1:18789');
|
||||
final expectedLocalProfile =
|
||||
controller.settings.primaryLocalGatewayProfile;
|
||||
expect(
|
||||
controller.assistantConnectionTargetLabel,
|
||||
'${expectedLocalProfile.host}:${expectedLocalProfile.port}',
|
||||
);
|
||||
|
||||
controller.initializeAssistantThreadContext(
|
||||
'remote-thread',
|
||||
|
||||
389
test/runtime/gateway_acp_client_suite.dart
Normal file
389
test/runtime/gateway_acp_client_suite.dart
Normal file
@ -0,0 +1,389 @@
|
||||
@TestOn('vm')
|
||||
library;
|
||||
|
||||
import 'dart:async';
|
||||
import 'dart:convert';
|
||||
import 'dart:io';
|
||||
|
||||
import 'package:flutter_test/flutter_test.dart';
|
||||
import 'package:xworkmate/runtime/gateway_acp_client.dart';
|
||||
import 'package:xworkmate/runtime/multi_agent_orchestrator.dart';
|
||||
import 'package:xworkmate/runtime/runtime_models.dart';
|
||||
|
||||
void main() {
|
||||
group('GatewayAcpClient', () {
|
||||
test(
|
||||
'prefers websocket for single-agent run and streams updates',
|
||||
() async {
|
||||
final server = await _AcpFakeServer.start();
|
||||
addTearDown(server.close);
|
||||
|
||||
final client = GatewayAcpClient(
|
||||
endpointResolver: () => server.baseHttpUri,
|
||||
);
|
||||
|
||||
final updates = <GatewayAcpSessionUpdate>[];
|
||||
final result = await client.runSingleAgent(
|
||||
GatewayAcpSingleAgentRequest(
|
||||
sessionId: 'session-ws',
|
||||
threadId: 'thread-ws',
|
||||
provider: SingleAgentProvider.codex,
|
||||
prompt: 'hello ws',
|
||||
model: 'gpt-4.1',
|
||||
workingDirectory: '/tmp',
|
||||
attachments: const <CollaborationAttachment>[],
|
||||
selectedSkills: const <String>['review'],
|
||||
aiGatewayBaseUrl: 'https://example.invalid',
|
||||
aiGatewayApiKey: 'test-key',
|
||||
resumeSession: false,
|
||||
),
|
||||
onUpdate: updates.add,
|
||||
);
|
||||
|
||||
expect(result.success, isTrue);
|
||||
expect(result.output, 'single-agent result (codex)');
|
||||
expect(result.turnId, 'turn-single');
|
||||
expect(updates, isNotEmpty);
|
||||
expect(updates.first.textDelta, 'delta-single');
|
||||
expect(server.rpcMethods, contains('acp.capabilities'));
|
||||
expect(server.rpcMethods, contains('session.start'));
|
||||
},
|
||||
);
|
||||
|
||||
test('falls back to HTTP+SSE when websocket is unavailable', () async {
|
||||
final server = await _AcpFakeServer.start(disableWebSocket: true);
|
||||
addTearDown(server.close);
|
||||
|
||||
final client = GatewayAcpClient(
|
||||
endpointResolver: () => server.baseHttpUri,
|
||||
);
|
||||
|
||||
final updates = <GatewayAcpSessionUpdate>[];
|
||||
final result = await client.runSingleAgent(
|
||||
GatewayAcpSingleAgentRequest(
|
||||
sessionId: 'session-sse',
|
||||
threadId: 'thread-sse',
|
||||
provider: SingleAgentProvider.claude,
|
||||
prompt: 'hello sse',
|
||||
model: 'claude-sonnet',
|
||||
workingDirectory: '/tmp',
|
||||
attachments: const <CollaborationAttachment>[],
|
||||
selectedSkills: const <String>[],
|
||||
aiGatewayBaseUrl: 'https://example.invalid',
|
||||
aiGatewayApiKey: 'test-key',
|
||||
resumeSession: false,
|
||||
),
|
||||
onUpdate: updates.add,
|
||||
);
|
||||
|
||||
expect(result.success, isTrue);
|
||||
expect(result.output, 'single-agent result (claude)');
|
||||
expect(updates.map((item) => item.textDelta), contains('delta-single'));
|
||||
expect(server.rpcMethods, contains('acp.capabilities'));
|
||||
expect(server.rpcMethods, contains('session.start'));
|
||||
});
|
||||
|
||||
test(
|
||||
'streams multi-agent events and supports cancel/close session',
|
||||
() async {
|
||||
final server = await _AcpFakeServer.start();
|
||||
addTearDown(server.close);
|
||||
|
||||
final client = GatewayAcpClient(
|
||||
endpointResolver: () => server.baseHttpUri,
|
||||
);
|
||||
|
||||
final events = await client
|
||||
.runMultiAgent(
|
||||
GatewayAcpMultiAgentRequest(
|
||||
sessionId: 'session-ma',
|
||||
threadId: 'thread-ma',
|
||||
prompt: 'run multi-agent',
|
||||
workingDirectory: '/tmp',
|
||||
attachments: const <CollaborationAttachment>[],
|
||||
selectedSkills: const <String>['design'],
|
||||
aiGatewayBaseUrl: 'https://example.invalid',
|
||||
aiGatewayApiKey: 'test-key',
|
||||
resumeSession: false,
|
||||
),
|
||||
)
|
||||
.toList();
|
||||
|
||||
expect(events, isNotEmpty);
|
||||
expect(events.first.type, 'step');
|
||||
expect(events.last.type, 'result');
|
||||
expect(events.last.error, isFalse);
|
||||
|
||||
await client.cancelSession(
|
||||
sessionId: 'session-ma',
|
||||
threadId: 'thread-ma',
|
||||
);
|
||||
await client.closeSession(
|
||||
sessionId: 'session-ma',
|
||||
threadId: 'thread-ma',
|
||||
);
|
||||
|
||||
expect(server.rpcMethods, contains('session.cancel'));
|
||||
expect(server.rpcMethods, contains('session.close'));
|
||||
},
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
class _AcpFakeServer {
|
||||
_AcpFakeServer._(this._server, {required this.disableWebSocket});
|
||||
|
||||
final HttpServer _server;
|
||||
final bool disableWebSocket;
|
||||
final List<String> rpcMethods = <String>[];
|
||||
|
||||
Uri get baseHttpUri => Uri.parse('http://127.0.0.1:${_server.port}');
|
||||
|
||||
static Future<_AcpFakeServer> start({bool disableWebSocket = false}) async {
|
||||
final server = await HttpServer.bind(InternetAddress.loopbackIPv4, 0);
|
||||
final fake = _AcpFakeServer._(server, disableWebSocket: disableWebSocket);
|
||||
unawaited(fake._listen());
|
||||
return fake;
|
||||
}
|
||||
|
||||
Future<void> close() async {
|
||||
await _server.close(force: true);
|
||||
}
|
||||
|
||||
Future<void> _listen() async {
|
||||
await for (final request in _server) {
|
||||
if (!disableWebSocket &&
|
||||
request.uri.path == '/acp' &&
|
||||
WebSocketTransformer.isUpgradeRequest(request)) {
|
||||
final socket = await WebSocketTransformer.upgrade(request);
|
||||
unawaited(_handleWebSocket(socket));
|
||||
continue;
|
||||
}
|
||||
if (request.uri.path == '/acp/rpc' && request.method == 'POST') {
|
||||
await _handleHttpRpc(request);
|
||||
continue;
|
||||
}
|
||||
request.response
|
||||
..statusCode = HttpStatus.notFound
|
||||
..write('not found');
|
||||
await request.response.close();
|
||||
}
|
||||
}
|
||||
|
||||
Future<void> _handleWebSocket(WebSocket socket) async {
|
||||
await for (final raw in socket) {
|
||||
final envelope = _decodeMap(raw);
|
||||
final id = envelope['id'];
|
||||
final method = envelope['method']?.toString() ?? '';
|
||||
final params = _asMap(envelope['params']);
|
||||
if (method.isEmpty) {
|
||||
continue;
|
||||
}
|
||||
rpcMethods.add(method);
|
||||
await _dispatch(
|
||||
method: method,
|
||||
id: id,
|
||||
params: params,
|
||||
notify: (notification) async {
|
||||
socket.add(jsonEncode(notification));
|
||||
},
|
||||
respond: (response) async {
|
||||
socket.add(jsonEncode(response));
|
||||
},
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
Future<void> _handleHttpRpc(HttpRequest request) async {
|
||||
final body = await utf8.decodeStream(request);
|
||||
final envelope = _decodeMap(body);
|
||||
final id = envelope['id'];
|
||||
final method = envelope['method']?.toString() ?? '';
|
||||
final params = _asMap(envelope['params']);
|
||||
if (method.isEmpty) {
|
||||
request.response.statusCode = HttpStatus.badRequest;
|
||||
await request.response.close();
|
||||
return;
|
||||
}
|
||||
rpcMethods.add(method);
|
||||
|
||||
request.response.headers.set(
|
||||
HttpHeaders.contentTypeHeader,
|
||||
'text/event-stream',
|
||||
);
|
||||
request.response.headers.set(HttpHeaders.cacheControlHeader, 'no-cache');
|
||||
|
||||
Future<void> notify(Map<String, dynamic> notification) async {
|
||||
request.response.write('data: ${jsonEncode(notification)}\n\n');
|
||||
await request.response.flush();
|
||||
}
|
||||
|
||||
Future<void> respond(Map<String, dynamic> response) async {
|
||||
request.response.write('data: ${jsonEncode(response)}\n\n');
|
||||
await request.response.flush();
|
||||
await request.response.close();
|
||||
}
|
||||
|
||||
await _dispatch(
|
||||
method: method,
|
||||
id: id,
|
||||
params: params,
|
||||
notify: notify,
|
||||
respond: respond,
|
||||
);
|
||||
}
|
||||
|
||||
Future<void> _dispatch({
|
||||
required String method,
|
||||
required Object? id,
|
||||
required Map<String, dynamic> params,
|
||||
required Future<void> Function(Map<String, dynamic> notification) notify,
|
||||
required Future<void> Function(Map<String, dynamic> response) respond,
|
||||
}) async {
|
||||
switch (method) {
|
||||
case 'acp.capabilities':
|
||||
await respond(
|
||||
_resultEnvelope(
|
||||
id: id,
|
||||
result: <String, dynamic>{
|
||||
'singleAgent': true,
|
||||
'multiAgent': true,
|
||||
'providers': <String>['codex', 'claude', 'gemini', 'opencode'],
|
||||
'capabilities': <String, dynamic>{
|
||||
'single_agent': true,
|
||||
'multi_agent': true,
|
||||
'providers': <String>['codex', 'claude', 'gemini', 'opencode'],
|
||||
},
|
||||
},
|
||||
),
|
||||
);
|
||||
return;
|
||||
case 'session.start':
|
||||
case 'session.message':
|
||||
final sessionId = params['sessionId']?.toString() ?? 'session-default';
|
||||
final threadId = params['threadId']?.toString() ?? sessionId;
|
||||
final mode = params['mode']?.toString() ?? 'single-agent';
|
||||
if (mode == 'multi-agent') {
|
||||
await notify(
|
||||
_notificationEnvelope(
|
||||
method: 'multi_agent.event',
|
||||
params: <String, dynamic>{
|
||||
'type': 'step',
|
||||
'title': 'Architect',
|
||||
'message': 'planning',
|
||||
'pending': false,
|
||||
'error': false,
|
||||
'data': <String, dynamic>{'seq': 1},
|
||||
},
|
||||
),
|
||||
);
|
||||
await respond(
|
||||
_resultEnvelope(
|
||||
id: id,
|
||||
result: <String, dynamic>{
|
||||
'success': true,
|
||||
'summary': 'multi-agent done',
|
||||
'finalScore': 9,
|
||||
'iterations': 1,
|
||||
},
|
||||
),
|
||||
);
|
||||
return;
|
||||
}
|
||||
final provider = params['provider']?.toString() ?? 'unknown';
|
||||
await notify(
|
||||
_notificationEnvelope(
|
||||
method: 'session.update',
|
||||
params: <String, dynamic>{
|
||||
'sessionId': sessionId,
|
||||
'threadId': threadId,
|
||||
'turnId': 'turn-single',
|
||||
'type': 'delta',
|
||||
'delta': 'delta-single',
|
||||
'seq': 1,
|
||||
'mode': 'single-agent',
|
||||
},
|
||||
),
|
||||
);
|
||||
await respond(
|
||||
_resultEnvelope(
|
||||
id: id,
|
||||
result: <String, dynamic>{
|
||||
'success': true,
|
||||
'output': 'single-agent result ($provider)',
|
||||
'turnId': 'turn-single',
|
||||
},
|
||||
),
|
||||
);
|
||||
return;
|
||||
case 'session.cancel':
|
||||
await respond(
|
||||
_resultEnvelope(
|
||||
id: id,
|
||||
result: const <String, dynamic>{
|
||||
'accepted': true,
|
||||
'cancelled': true,
|
||||
},
|
||||
),
|
||||
);
|
||||
return;
|
||||
case 'session.close':
|
||||
await respond(
|
||||
_resultEnvelope(
|
||||
id: id,
|
||||
result: const <String, dynamic>{'accepted': true, 'closed': true},
|
||||
),
|
||||
);
|
||||
return;
|
||||
default:
|
||||
await respond(<String, dynamic>{
|
||||
'jsonrpc': '2.0',
|
||||
'id': id,
|
||||
'error': <String, dynamic>{
|
||||
'code': -32601,
|
||||
'message': 'method not found',
|
||||
},
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
Map<String, dynamic> _resultEnvelope({
|
||||
required Object? id,
|
||||
required Map<String, dynamic> result,
|
||||
}) {
|
||||
return <String, dynamic>{'jsonrpc': '2.0', 'id': id, 'result': result};
|
||||
}
|
||||
|
||||
Map<String, dynamic> _notificationEnvelope({
|
||||
required String method,
|
||||
required Map<String, dynamic> params,
|
||||
}) {
|
||||
return <String, dynamic>{
|
||||
'jsonrpc': '2.0',
|
||||
'method': method,
|
||||
'params': params,
|
||||
};
|
||||
}
|
||||
|
||||
Map<String, dynamic> _decodeMap(Object raw) {
|
||||
if (raw is String) {
|
||||
final decoded = jsonDecode(raw);
|
||||
return _asMap(decoded);
|
||||
}
|
||||
if (raw is List<int>) {
|
||||
final decoded = jsonDecode(utf8.decode(raw));
|
||||
return _asMap(decoded);
|
||||
}
|
||||
return _asMap(raw);
|
||||
}
|
||||
|
||||
Map<String, dynamic> _asMap(Object? raw) {
|
||||
if (raw is Map<String, dynamic>) {
|
||||
return raw;
|
||||
}
|
||||
if (raw is Map) {
|
||||
return raw.cast<String, dynamic>();
|
||||
}
|
||||
return const <String, dynamic>{};
|
||||
}
|
||||
}
|
||||
@ -1,5 +1,5 @@
|
||||
import '../test_suite_stub.dart'
|
||||
if (dart.library.io) 'multi_agent_broker_suite.dart'
|
||||
if (dart.library.io) 'gateway_acp_client_suite.dart'
|
||||
as suite;
|
||||
|
||||
void main() {
|
||||
@ -1,137 +0,0 @@
|
||||
@TestOn('vm')
|
||||
library;
|
||||
|
||||
import 'package:flutter_test/flutter_test.dart';
|
||||
import 'package:xworkmate/runtime/multi_agent_broker.dart';
|
||||
import 'package:xworkmate/runtime/multi_agent_orchestrator.dart';
|
||||
import 'package:xworkmate/runtime/runtime_models.dart';
|
||||
|
||||
void main() {
|
||||
TestWidgetsFlutterBinding.ensureInitialized();
|
||||
|
||||
test(
|
||||
'MultiAgentBroker supports session start, message, cancel, and close',
|
||||
() async {
|
||||
final orchestrator = _FakeOrchestrator();
|
||||
final server = MultiAgentBrokerServer(orchestrator);
|
||||
await server.start();
|
||||
addTearDown(server.stop);
|
||||
|
||||
final client = MultiAgentBrokerClient(server.wsUri!);
|
||||
final firstEvents = await client
|
||||
.startSession(
|
||||
sessionId: 'session-1',
|
||||
taskPrompt: 'first turn',
|
||||
workingDirectory: '/tmp',
|
||||
attachments: const <CollaborationAttachment>[],
|
||||
selectedSkills: const <String>['aris'],
|
||||
aiGatewayBaseUrl: '',
|
||||
aiGatewayApiKey: '',
|
||||
)
|
||||
.toList();
|
||||
final secondEvents = await client
|
||||
.sendSessionMessage(
|
||||
sessionId: 'session-1',
|
||||
taskPrompt: 'second turn',
|
||||
workingDirectory: '/tmp',
|
||||
attachments: const <CollaborationAttachment>[],
|
||||
selectedSkills: const <String>['aris'],
|
||||
aiGatewayBaseUrl: '',
|
||||
aiGatewayApiKey: '',
|
||||
)
|
||||
.toList();
|
||||
|
||||
await client.cancelSession('session-1');
|
||||
await client.closeSession('session-1');
|
||||
|
||||
expect(orchestrator.prompts, hasLength(2));
|
||||
expect(orchestrator.prompts.first, contains('first turn'));
|
||||
expect(orchestrator.prompts.last, contains('first turn'));
|
||||
expect(orchestrator.prompts.last, contains('second turn'));
|
||||
expect(firstEvents.last.type, 'result');
|
||||
expect(secondEvents.last.type, 'result');
|
||||
expect(orchestrator.abortCount, 1);
|
||||
},
|
||||
);
|
||||
|
||||
test('MultiAgentBroker clears session history after close', () async {
|
||||
final orchestrator = _FakeOrchestrator();
|
||||
final server = MultiAgentBrokerServer(orchestrator);
|
||||
await server.start();
|
||||
addTearDown(server.stop);
|
||||
|
||||
final client = MultiAgentBrokerClient(server.wsUri!);
|
||||
await client
|
||||
.startSession(
|
||||
sessionId: 'session-2',
|
||||
taskPrompt: 'first turn',
|
||||
workingDirectory: '/tmp',
|
||||
attachments: const <CollaborationAttachment>[],
|
||||
selectedSkills: const <String>['aris'],
|
||||
aiGatewayBaseUrl: '',
|
||||
aiGatewayApiKey: '',
|
||||
)
|
||||
.drain<void>();
|
||||
await client.closeSession('session-2');
|
||||
await client
|
||||
.sendSessionMessage(
|
||||
sessionId: 'session-2',
|
||||
taskPrompt: 'fresh turn',
|
||||
workingDirectory: '/tmp',
|
||||
attachments: const <CollaborationAttachment>[],
|
||||
selectedSkills: const <String>['aris'],
|
||||
aiGatewayBaseUrl: '',
|
||||
aiGatewayApiKey: '',
|
||||
)
|
||||
.drain<void>();
|
||||
|
||||
expect(orchestrator.prompts, hasLength(2));
|
||||
expect(orchestrator.prompts.first, contains('first turn'));
|
||||
expect(orchestrator.prompts.last, contains('fresh turn'));
|
||||
expect(orchestrator.prompts.last, isNot(contains('first turn')));
|
||||
});
|
||||
}
|
||||
|
||||
class _FakeOrchestrator extends MultiAgentOrchestrator {
|
||||
_FakeOrchestrator()
|
||||
: super(config: MultiAgentConfig.defaults().copyWith(enabled: true));
|
||||
|
||||
final List<String> prompts = <String>[];
|
||||
int abortCount = 0;
|
||||
|
||||
@override
|
||||
Future<CollaborationResult> runCollaboration({
|
||||
required String taskPrompt,
|
||||
required String workingDirectory,
|
||||
List<CollaborationAttachment> attachments = const [],
|
||||
List<String> selectedSkills = const [],
|
||||
String aiGatewayBaseUrl = '',
|
||||
String aiGatewayApiKey = '',
|
||||
void Function(MultiAgentRunEvent event)? onEvent,
|
||||
}) async {
|
||||
prompts.add(taskPrompt);
|
||||
onEvent?.call(
|
||||
const MultiAgentRunEvent(
|
||||
type: 'step',
|
||||
title: 'Architect',
|
||||
message: 'planning',
|
||||
pending: false,
|
||||
error: false,
|
||||
role: 'architect',
|
||||
),
|
||||
);
|
||||
return const CollaborationResult(
|
||||
success: true,
|
||||
steps: <CollaborationStep>[],
|
||||
finalCode: 'ok',
|
||||
finalScore: 9,
|
||||
duration: Duration(milliseconds: 10),
|
||||
iterations: 0,
|
||||
);
|
||||
}
|
||||
|
||||
@override
|
||||
Future<void> abort() async {
|
||||
abortCount += 1;
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user