diff --git a/go/aris_bridge/go.mod b/go/aris_bridge/go.mod index 65d666b2..3e7fa85e 100644 --- a/go/aris_bridge/go.mod +++ b/go/aris_bridge/go.mod @@ -1,3 +1,5 @@ module xworkmate/aris_bridge go 1.25.0 + +require github.com/gorilla/websocket v1.5.3 diff --git a/go/aris_bridge/go.sum b/go/aris_bridge/go.sum new file mode 100644 index 00000000..25a9fc4b --- /dev/null +++ b/go/aris_bridge/go.sum @@ -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= diff --git a/go/aris_bridge/main.go b/go/aris_bridge/main.go index 37408309..4cd61050 100644 --- a/go/aris_bridge/main.go +++ b/go/aris_bridge/main.go @@ -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 == "" { + 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 == "" { - 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 + } +} diff --git a/lib/app/app_controller_desktop.dart b/lib/app/app_controller_desktop.dart index 8bad8d42..cfc9c91f 100644 --- a/lib/app/app_controller_desktop.dart +++ b/lib/app/app_controller_desktop.dart @@ -2041,12 +2041,16 @@ class AppController extends ChangeNotifier { ); if (archived) { unawaited( - _gatewayAcpClient - .closeSession( + _enqueueThreadTurn(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 clearAssistantLocalState() async { await _flushAssistantThreadPersistence(); await _store.clearAssistantLocalState(); + await _store.saveAssistantThreadRecords(const []); _assistantThreadPersistQueue = Future.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, ); diff --git a/lib/runtime/agent_cli_bridge.dart b/lib/runtime/agent_cli_bridge.dart deleted file mode 100644 index b740d41e..00000000 --- a/lib/runtime/agent_cli_bridge.dart +++ /dev/null @@ -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 attachments; - final List selectedSkills; - final String aiGatewayBaseUrl; - final String aiGatewayApiKey; -} - -class AgentCliBridgeResult { - const AgentCliBridgeResult({ - required this.output, - required this.success, - required this.errorMessage, - this.events = const [], - }); - - final String output; - final bool success; - final String errorMessage; - final List events; -} - -abstract class AgentCliBridge { - Future run(AgentCliBridgeRequest request); -} - -class SubprocessCliBridge implements AgentCliBridge { - const SubprocessCliBridge({ - required this.command, - this.defaultArgs = const [], - }); - - final String command; - final List defaultArgs; - - @override - Future run(AgentCliBridgeRequest request) async { - try { - final process = await Process.start( - command, - [...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 run(AgentCliBridgeRequest request) async { - final socket = await WebSocket.connect(endpoint.toString()); - final requestId = DateTime.now().microsecondsSinceEpoch.toString(); - final completer = Completer(); - final events = []; - - socket.listen( - (raw) { - final json = jsonDecode(raw as String) as Map; - final method = json['method'] as String?; - if (method == 'multi_agent.event') { - final params = - (json['params'] as Map?)?.cast() ?? - const {}; - events.add(MultiAgentRunEvent.fromJson(params)); - return; - } - if (json['id']?.toString() == requestId && json['result'] is Map) { - final result = (json['result'] as Map).cast(); - 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(); - 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({ - 'jsonrpc': '2.0', - 'id': requestId, - 'method': 'session.start', - 'params': { - 'sessionId': request.sessionId, - 'taskPrompt': request.taskPrompt, - 'workingDirectory': request.workingDirectory, - 'attachments': request.attachments - .map( - (item) => { - 'name': item.name, - 'description': item.description, - 'path': item.path, - }, - ) - .toList(growable: false), - 'selectedSkills': request.selectedSkills, - 'aiGatewayBaseUrl': request.aiGatewayBaseUrl, - 'aiGatewayApiKey': request.aiGatewayApiKey, - }, - }), - ); - - return completer.future; - } -} diff --git a/lib/runtime/multi_agent_broker.dart b/lib/runtime/multi_agent_broker.dart deleted file mode 100644 index ed0d15e4..00000000 --- a/lib/runtime/multi_agent_broker.dart +++ /dev/null @@ -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 _sessions = - {}; - 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 start() async { - if (_server != null) { - return; - } - _server = await HttpServer.bind(InternetAddress.loopbackIPv4, 0); - unawaited(_listen()); - } - - Future stop() async { - final server = _server; - _server = null; - _sessions.clear(); - await server?.close(force: true); - } - - Future _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 _handleSocket(WebSocket socket) async { - await for (final raw in socket) { - try { - final json = jsonDecode(raw as String) as Map; - final method = json['method'] as String? ?? ''; - final id = json['id']; - final params = - (json['params'] as Map?)?.cast() ?? - const {}; - 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, - {'accepted': true, 'cancelled': true}, - ); - break; - case 'session.close': - final sessionId = params['sessionId']?.toString().trim() ?? ''; - if (sessionId.isNotEmpty) { - _sessions.remove(sessionId); - } - _writeResult( - socket, - id, - {'accepted': true, 'closed': true}, - ); - break; - default: - _writeError(socket, id, -32601, 'Method not found'); - } - } catch (error) { - _writeError(socket, null, -32000, error.toString()); - } - } - } - - Future _handleRunStart( - WebSocket socket, - Object? id, - Map 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 _handleSessionStart( - WebSocket socket, - Object? id, - Map 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: [], - ); - _sessions[sessionId] = state; - await _runSession(socket, id, state, params['taskPrompt'] as String? ?? ''); - } - - Future _handleSessionMessage( - WebSocket socket, - Object? id, - Map 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: [], - ), - ); - 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 _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, - {...result.toJson(), 'sessionId': state.sessionId}, - ); - } - - String _composeSessionPrompt(List 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 _parseAttachments(Object? raw) { - return ((raw as List?) ?? const []) - .whereType() - .map( - (item) => CollaborationAttachment( - name: item['name']?.toString() ?? '', - description: item['description']?.toString() ?? '', - path: item['path']?.toString() ?? '', - ), - ) - .toList(growable: false); - } - - List _parseSelectedSkills(Object? raw) { - return ((raw as List?) ?? const []) - .map((item) => item.toString()) - .toList(growable: false); - } - - void _emitEvent(WebSocket socket, MultiAgentRunEvent event) { - socket.add( - jsonEncode({ - 'jsonrpc': '2.0', - 'method': 'multi_agent.event', - 'params': event.toJson(), - }), - ); - } - - void _writeResult(WebSocket socket, Object? id, Map result) { - socket.add( - jsonEncode({ - 'jsonrpc': '2.0', - 'id': id, - 'result': result, - }), - ); - } - - void _writeError( - WebSocket socket, - Object? id, - int code, - String message, - ) { - socket.add( - jsonEncode({ - 'jsonrpc': '2.0', - 'id': id, - 'error': {'code': code, 'message': message}, - }), - ); - } -} - -class MultiAgentBrokerClient { - MultiAgentBrokerClient(this._uri); - - final Uri _uri; - - Stream runTask({ - required String taskPrompt, - required String workingDirectory, - required List attachments, - required List selectedSkills, - required String aiGatewayBaseUrl, - required String aiGatewayApiKey, - }) { - return _streamRequest( - method: 'run.start', - params: { - 'taskPrompt': taskPrompt, - 'workingDirectory': workingDirectory, - 'attachments': _encodeAttachments(attachments), - 'selectedSkills': selectedSkills, - 'aiGatewayBaseUrl': aiGatewayBaseUrl, - 'aiGatewayApiKey': aiGatewayApiKey, - }, - ); - } - - Stream startSession({ - required String sessionId, - required String taskPrompt, - required String workingDirectory, - required List attachments, - required List selectedSkills, - required String aiGatewayBaseUrl, - required String aiGatewayApiKey, - }) { - return _streamRequest( - method: 'session.start', - params: { - 'sessionId': sessionId, - 'taskPrompt': taskPrompt, - 'workingDirectory': workingDirectory, - 'attachments': _encodeAttachments(attachments), - 'selectedSkills': selectedSkills, - 'aiGatewayBaseUrl': aiGatewayBaseUrl, - 'aiGatewayApiKey': aiGatewayApiKey, - }, - ); - } - - Stream sendSessionMessage({ - required String sessionId, - required String taskPrompt, - required String workingDirectory, - required List attachments, - required List selectedSkills, - required String aiGatewayBaseUrl, - required String aiGatewayApiKey, - }) { - return _streamRequest( - method: 'session.message', - params: { - 'sessionId': sessionId, - 'taskPrompt': taskPrompt, - 'workingDirectory': workingDirectory, - 'attachments': _encodeAttachments(attachments), - 'selectedSkills': selectedSkills, - 'aiGatewayBaseUrl': aiGatewayBaseUrl, - 'aiGatewayApiKey': aiGatewayApiKey, - }, - ); - } - - Future cancelSession(String sessionId) async { - await _requestOnly( - method: 'session.cancel', - params: {'sessionId': sessionId}, - ); - } - - Future closeSession(String sessionId) async { - await _requestOnly( - method: 'session.close', - params: {'sessionId': sessionId}, - ); - } - - Stream _streamRequest({ - required String method, - required Map params, - }) async* { - final socket = await WebSocket.connect(_uri.toString()); - final controller = StreamController(); - final requestId = DateTime.now().microsecondsSinceEpoch.toString(); - - socket.listen( - (raw) { - final json = jsonDecode(raw as String) as Map; - final rpcMethod = json['method'] as String?; - if (rpcMethod == 'multi_agent.event') { - final eventParams = - (json['params'] as Map?)?.cast() ?? - const {}; - controller.add(MultiAgentRunEvent.fromJson(eventParams)); - return; - } - if (json['id']?.toString() == requestId && json['result'] is Map) { - final result = (json['result'] as Map).cast(); - 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(); - 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({ - 'jsonrpc': '2.0', - 'id': requestId, - 'method': method, - 'params': params, - }), - ); - - yield* controller.stream; - } - - Future _requestOnly({ - required String method, - required Map params, - }) async { - await for (final _ in _streamRequest(method: method, params: params)) { - return; - } - } - - List> _encodeAttachments( - List attachments, - ) { - return attachments - .map( - (item) => { - '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 attachments; - List selectedSkills; - String aiGatewayBaseUrl; - String aiGatewayApiKey; - final List history; -} diff --git a/test/runtime/agent_cli_bridge_suite.dart b/test/runtime/agent_cli_bridge_suite.dart deleted file mode 100644 index 6762d178..00000000 --- a/test/runtime/agent_cli_bridge_suite.dart +++ /dev/null @@ -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: [], - selectedSkills: ['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 runCollaboration({ - required String taskPrompt, - required String workingDirectory, - List attachments = const [], - List 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: [], - finalCode: 'ok', - finalScore: 7, - duration: Duration(milliseconds: 10), - iterations: 0, - ); - } -} diff --git a/test/runtime/agent_cli_bridge_test.dart b/test/runtime/agent_cli_bridge_test.dart deleted file mode 100644 index eb4ff509..00000000 --- a/test/runtime/agent_cli_bridge_test.dart +++ /dev/null @@ -1,7 +0,0 @@ -import '../test_suite_stub.dart' - if (dart.library.io) 'agent_cli_bridge_suite.dart' - as suite; - -void main() { - suite.main(); -} diff --git a/test/runtime/app_controller_assistant_flow_suite.dart b/test/runtime/app_controller_assistant_flow_suite.dart index bb7a0b5e..44eb124d 100644 --- a/test/runtime/app_controller_assistant_flow_suite.dart +++ b/test/runtime/app_controller_assistant_flow_suite.dart @@ -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({}); 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 _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, { @@ -277,6 +309,35 @@ class _FakeGatewayServer { } } + Future _serveAcpRpc(HttpRequest request) async { + final body = await utf8.decodeStream(request); + final envelope = (jsonDecode(body) as Map).cast(); + final id = envelope['id']; + final method = envelope['method']?.toString() ?? ''; + final response = { + 'jsonrpc': '2.0', + 'id': id, + 'result': method == 'acp.capabilities' + ? { + 'singleAgent': true, + 'multiAgent': true, + 'providers': ['claude', 'codex', 'gemini', 'opencode'], + 'capabilities': { + 'single_agent': true, + 'multi_agent': true, + 'providers': ['claude', 'codex', 'gemini', 'opencode'], + }, + } + : const {}, + }; + 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 _emitAssistantResult( WebSocket socket, { required String runId, @@ -336,6 +397,23 @@ class _FakeGatewayServer { } } +Future _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.delayed(const Duration(milliseconds: 100)); + } + } +} + Future _waitFor( bool Function() predicate, { Duration timeout = const Duration(seconds: 5), diff --git a/test/runtime/app_controller_execution_target_switch_suite.dart b/test/runtime/app_controller_execution_target_switch_suite.dart index abb06922..641efd3e 100644 --- a/test/runtime/app_controller_execution_target_switch_suite.dart +++ b/test/runtime/app_controller_execution_target_switch_suite.dart @@ -236,15 +236,21 @@ void main() { await controller.setAssistantExecutionTarget( AssistantExecutionTarget.local, ); + final expectedLocalProfile = + controller.settings.primaryLocalGatewayProfile; expect( gateway.connectedProfiles.last, isA() .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', diff --git a/test/runtime/gateway_acp_client_suite.dart b/test/runtime/gateway_acp_client_suite.dart new file mode 100644 index 00000000..0f9999c6 --- /dev/null +++ b/test/runtime/gateway_acp_client_suite.dart @@ -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 = []; + 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 [], + selectedSkills: const ['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 = []; + 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 [], + selectedSkills: const [], + 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 [], + selectedSkills: const ['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 rpcMethods = []; + + 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 close() async { + await _server.close(force: true); + } + + Future _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 _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 _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 notify(Map notification) async { + request.response.write('data: ${jsonEncode(notification)}\n\n'); + await request.response.flush(); + } + + Future respond(Map 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 _dispatch({ + required String method, + required Object? id, + required Map params, + required Future Function(Map notification) notify, + required Future Function(Map response) respond, + }) async { + switch (method) { + case 'acp.capabilities': + await respond( + _resultEnvelope( + id: id, + result: { + 'singleAgent': true, + 'multiAgent': true, + 'providers': ['codex', 'claude', 'gemini', 'opencode'], + 'capabilities': { + 'single_agent': true, + 'multi_agent': true, + 'providers': ['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: { + 'type': 'step', + 'title': 'Architect', + 'message': 'planning', + 'pending': false, + 'error': false, + 'data': {'seq': 1}, + }, + ), + ); + await respond( + _resultEnvelope( + id: id, + result: { + 'success': true, + 'summary': 'multi-agent done', + 'finalScore': 9, + 'iterations': 1, + }, + ), + ); + return; + } + final provider = params['provider']?.toString() ?? 'unknown'; + await notify( + _notificationEnvelope( + method: 'session.update', + params: { + 'sessionId': sessionId, + 'threadId': threadId, + 'turnId': 'turn-single', + 'type': 'delta', + 'delta': 'delta-single', + 'seq': 1, + 'mode': 'single-agent', + }, + ), + ); + await respond( + _resultEnvelope( + id: id, + result: { + 'success': true, + 'output': 'single-agent result ($provider)', + 'turnId': 'turn-single', + }, + ), + ); + return; + case 'session.cancel': + await respond( + _resultEnvelope( + id: id, + result: const { + 'accepted': true, + 'cancelled': true, + }, + ), + ); + return; + case 'session.close': + await respond( + _resultEnvelope( + id: id, + result: const {'accepted': true, 'closed': true}, + ), + ); + return; + default: + await respond({ + 'jsonrpc': '2.0', + 'id': id, + 'error': { + 'code': -32601, + 'message': 'method not found', + }, + }); + } + } + + Map _resultEnvelope({ + required Object? id, + required Map result, + }) { + return {'jsonrpc': '2.0', 'id': id, 'result': result}; + } + + Map _notificationEnvelope({ + required String method, + required Map params, + }) { + return { + 'jsonrpc': '2.0', + 'method': method, + 'params': params, + }; + } + + Map _decodeMap(Object raw) { + if (raw is String) { + final decoded = jsonDecode(raw); + return _asMap(decoded); + } + if (raw is List) { + final decoded = jsonDecode(utf8.decode(raw)); + return _asMap(decoded); + } + return _asMap(raw); + } + + Map _asMap(Object? raw) { + if (raw is Map) { + return raw; + } + if (raw is Map) { + return raw.cast(); + } + return const {}; + } +} diff --git a/test/runtime/multi_agent_broker_test.dart b/test/runtime/gateway_acp_client_test.dart similarity index 58% rename from test/runtime/multi_agent_broker_test.dart rename to test/runtime/gateway_acp_client_test.dart index a28c1008..a7029822 100644 --- a/test/runtime/multi_agent_broker_test.dart +++ b/test/runtime/gateway_acp_client_test.dart @@ -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() { diff --git a/test/runtime/multi_agent_broker_suite.dart b/test/runtime/multi_agent_broker_suite.dart deleted file mode 100644 index 0de84067..00000000 --- a/test/runtime/multi_agent_broker_suite.dart +++ /dev/null @@ -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 [], - selectedSkills: const ['aris'], - aiGatewayBaseUrl: '', - aiGatewayApiKey: '', - ) - .toList(); - final secondEvents = await client - .sendSessionMessage( - sessionId: 'session-1', - taskPrompt: 'second turn', - workingDirectory: '/tmp', - attachments: const [], - selectedSkills: const ['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 [], - selectedSkills: const ['aris'], - aiGatewayBaseUrl: '', - aiGatewayApiKey: '', - ) - .drain(); - await client.closeSession('session-2'); - await client - .sendSessionMessage( - sessionId: 'session-2', - taskPrompt: 'fresh turn', - workingDirectory: '/tmp', - attachments: const [], - selectedSkills: const ['aris'], - aiGatewayBaseUrl: '', - aiGatewayApiKey: '', - ) - .drain(); - - 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 prompts = []; - int abortCount = 0; - - @override - Future runCollaboration({ - required String taskPrompt, - required String workingDirectory, - List attachments = const [], - List 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: [], - finalCode: 'ok', - finalScore: 9, - duration: Duration(milliseconds: 10), - iterations: 0, - ); - } - - @override - Future abort() async { - abortCount += 1; - } -}