Files
goclaw/cmd/agent_chat_client.go
T
040b0f1944 fix(acp): Gemini ACP protocol fixes and multi-session architecture (#901)
* fix(cli): add missing X-GoClaw-User-Id header to gateway client

* feat(acp): comprehensive integration with Gemini ACP protocol

- Support nested JSON-RPC notification structures
- Add robust streaming text collection and mapping
- Increase handshake timeout to 60s for heavy model initialization
- Fix WebSocket user_id authentication and schema v47 compatibility
- Allow Google/GCP environment variables for ACP subprocesses

* refactor(acp): multi-session architecture with session tracing and temp session cleanup

- One shared Gemini process, multiple ACP sessions per process (one per goclaw conversation)
- resolveSession: per-key mutex prevents TOCTOU race on concurrent session creation
- Respawn detection via proc pointer comparison; session/load fallback after crash
- sessionReaper: purges ACP sessions idle >30min; temp- sessions purged immediately on completion
- WithGoclawSession context propagation: goclaw session key appears alongside ACP sid in all logs
- dispatchUpdate: Gemini agent_message_chunk protocol mapping normalized here
- session_test.go: full coverage of multi-session API (Initialize, NewSession, Prompt, Cancel, dispatch)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix(acp): align with ACP protocol standard (SDK v0.16.1)

- InitializeRequest: "capabilities" → "clientCapabilities" (standard field name)
- LoadSessionRequest: add mcpServers field (required by standard)
- mapStopReason: add standard stop reasons (max_tokens, cancelled)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix(acp): address review findings from PR #901

- B1: Restore RequiredSchemaVersion to 55 (was incorrectly set to 47)
- S1: Move X-GoClaw-User-Id header outside token check
- S2: Add env gate (ACP_GEMINI_E2E) to Gemini E2E test
- S3: Fix tab indentation in jsonrpc.go writeMessage
- Build: Extract Pdeathsig to platform-specific files for cross-platform build

* fix(acp): address S4-S7 review findings

S4/S5: Send session/cancel before purging sessions locally
- purgeSession() now cancels ACP session before deleting map entry
- sessionReaper() sends cancel notification for idle sessions
- Updated comments to reflect actual behavior

S6: Tighten GOOGLE_/GCP_ env var filtering
- Add GOOGLE_, GCP_ back to sensitiveEnvPrefixes
- Add allowedEnvExact for safe vars: GOOGLE_API_KEY,
  GOOGLE_APPLICATION_CREDENTIALS, GOOGLE_CLOUD_PROJECT, GCP_PROJECT

S7: Propagate ACP errors to callers
- Chat/ChatStream now return err alongside ChatResponse
- Enables upstream retry logic and error metrics

---------

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
Co-authored-by: viettranx <viettranx@gmail.com>
2026-04-18 12:17:23 +07:00

224 lines
5.4 KiB
Go

package cmd
import (
"bufio"
"encoding/json"
"fmt"
"os"
"strings"
"github.com/google/uuid"
"github.com/gorilla/websocket"
"github.com/nextlevelbuilder/goclaw/internal/config"
"github.com/nextlevelbuilder/goclaw/internal/sessions"
"github.com/nextlevelbuilder/goclaw/pkg/protocol"
)
func runClientMode(cfg *config.Config, addr, agentName, message, sessionKey string) {
wsURL := fmt.Sprintf("ws://%s/ws", addr)
conn, _, err := websocket.DefaultDialer.Dial(wsURL, nil)
if err != nil {
fmt.Fprintf(os.Stderr, "WebSocket connect failed: %v\n", err)
os.Exit(1)
}
defer conn.Close()
// Authenticate
if err := wsConnect(conn, cfg.Gateway.Token); err != nil {
fmt.Fprintf(os.Stderr, "Gateway auth failed: %v\n", err)
os.Exit(1)
}
agentCfg := cfg.ResolveAgent(agentName)
if message != "" {
// One-shot mode
resp, err := wsChatSend(conn, agentName, sessionKey, message)
if err != nil {
fmt.Fprintf(os.Stderr, "Error: %v\n", err)
os.Exit(1)
}
fmt.Println(resp)
return
}
// Interactive REPL
fmt.Fprintf(os.Stderr, "\nGoClaw Interactive Chat (agent: %s, model: %s)\n", agentName, agentCfg.Model)
fmt.Fprintf(os.Stderr, "Session: %s\n", sessionKey)
fmt.Fprintf(os.Stderr, "Type \"exit\" to quit, \"/new\" for new session\n\n")
scanner := bufio.NewScanner(os.Stdin)
for {
fmt.Fprint(os.Stderr, "You: ")
if !scanner.Scan() {
break
}
input := strings.TrimSpace(scanner.Text())
if input == "" {
continue
}
if input == "exit" || input == "quit" {
fmt.Fprintln(os.Stderr, "Goodbye!")
return
}
if input == "/new" {
sessionKey = sessions.BuildSessionKey(agentName, "cli", sessions.PeerDirect, uuid.NewString()[:8])
fmt.Fprintf(os.Stderr, "New session: %s\n\n", sessionKey)
continue
}
resp, err := wsChatSend(conn, agentName, sessionKey, input)
if err != nil {
fmt.Fprintf(os.Stderr, "Error: %v\n\n", err)
continue
}
fmt.Printf("\n%s\n\n", resp)
}
}
// wsConnect sends the connect RPC and waits for auth response.
func wsConnect(conn *websocket.Conn, token string) error {
params := map[string]string{}
userId := os.Getenv("GOCLAW_USER_ID")
if userId == "" { userId = "system" }
params["user_id"] = userId
if token != "" {
params["token"] = token
}
paramsJSON, _ := json.Marshal(params)
reqFrame := protocol.RequestFrame{
Type: protocol.FrameTypeRequest,
ID: "connect-1",
Method: protocol.MethodConnect,
Params: paramsJSON,
}
if err := conn.WriteJSON(reqFrame); err != nil {
return fmt.Errorf("send connect: %w", err)
}
var resp protocol.ResponseFrame
if err := conn.ReadJSON(&resp); err != nil {
return fmt.Errorf("read connect response: %w", err)
}
if !resp.OK {
if resp.Error != nil {
return fmt.Errorf("connect rejected: %s", resp.Error.Message)
}
return fmt.Errorf("connect rejected")
}
return nil
}
// wsChatSend sends a chat.send RPC and waits for the response,
// displaying events (tool calls, chunks) in real-time.
func wsChatSend(conn *websocket.Conn, agentID, sessionKey, message string) (string, error) {
reqID := uuid.NewString()[:8]
params, _ := json.Marshal(map[string]any{
"message": message,
"agentId": agentID,
"sessionKey": sessionKey,
"stream": true,
})
reqFrame := protocol.RequestFrame{
Type: protocol.FrameTypeRequest,
ID: reqID,
Method: protocol.MethodChatSend,
Params: params,
}
if err := conn.WriteJSON(reqFrame); err != nil {
return "", fmt.Errorf("send chat: %w", err)
}
// Read frames until we get our response
var finalContent string
for {
_, rawMsg, err := conn.ReadMessage()
if err != nil {
return "", fmt.Errorf("read: %w", err)
}
frameType, _ := protocol.ParseFrameType(rawMsg)
switch frameType {
case protocol.FrameTypeResponse:
var resp protocol.ResponseFrame
if err := json.Unmarshal(rawMsg, &resp); err != nil {
continue
}
if resp.ID != reqID {
continue // response for a different request
}
if !resp.OK {
if resp.Error != nil {
return "", fmt.Errorf("agent error: %s", resp.Error.Message)
}
return "", fmt.Errorf("agent error (unknown)")
}
// Extract content from payload
if payload, ok := resp.Payload.(map[string]any); ok {
if content, ok := payload["content"].(string); ok && content != "" {
finalContent = content
}
}
return finalContent, nil
case protocol.FrameTypeEvent:
var evt protocol.EventFrame
if err := json.Unmarshal(rawMsg, &evt); err != nil {
continue
}
handleCLIEvent(evt)
}
}
}
// handleCLIEvent displays agent events in the terminal.
func handleCLIEvent(evt protocol.EventFrame) {
payload, ok := evt.Payload.(map[string]any)
if !ok {
return
}
evtType, _ := payload["type"].(string)
switch evt.Event {
case protocol.EventAgent:
switch evtType {
case protocol.AgentEventToolCall:
if p, ok := payload["payload"].(map[string]any); ok {
name, _ := p["toolName"].(string)
if name == "" {
name, _ = p["name"].(string)
}
fmt.Fprintf(os.Stderr, " [tool] %s\n", name)
}
case protocol.AgentEventToolResult:
if p, ok := payload["payload"].(map[string]any); ok {
isErr, _ := p["is_error"].(bool)
name, _ := p["toolName"].(string)
if name == "" {
name, _ = p["name"].(string)
}
if isErr {
fmt.Fprintf(os.Stderr, " [tool] %s -> error\n", name)
}
}
}
case protocol.EventChat:
switch evtType {
case protocol.ChatEventChunk:
if content, ok := payload["content"].(string); ok {
fmt.Print(content)
}
}
}
}