diff --git a/cmd/agent_chat_client.go b/cmd/agent_chat_client.go index a5f8e51b..6a6d4c59 100644 --- a/cmd/agent_chat_client.go +++ b/cmd/agent_chat_client.go @@ -81,6 +81,9 @@ func runClientMode(cfg *config.Config, addr, agentName, message, sessionKey stri // 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 } diff --git a/cmd/gateway_http_client.go b/cmd/gateway_http_client.go index 1d86b20c..b1788a67 100644 --- a/cmd/gateway_http_client.go +++ b/cmd/gateway_http_client.go @@ -119,6 +119,7 @@ func gatewayHTTPDoRaw(method, path string, body any) ([]byte, int, error) { return nil, 0, err } req.Header.Set("Content-Type", "application/json") + req.Header.Set("X-GoClaw-User-Id", "system") if token := resolveGatewayToken(); token != "" { req.Header.Set("Authorization", "Bearer "+token) } diff --git a/internal/gateway/router.go b/internal/gateway/router.go index 6e994a8a..15eb6928 100644 --- a/internal/gateway/router.go +++ b/internal/gateway/router.go @@ -260,7 +260,7 @@ func (r *MethodRouter) handleConnect(ctx context.Context, client *Client, req *p if paired { client.role = permissions.RoleOperator client.authenticated = true - client.userID = params.UserID + client.userID = params.UserID client.pairedSenderID = params.SenderID client.pairedChannel = "browser" tid, errCode := r.resolveTenantHint(ctx, params.TenantHint, params.UserID) diff --git a/internal/providers/acp/acp_gemini_test.go b/internal/providers/acp/acp_gemini_test.go new file mode 100644 index 00000000..b4324a68 --- /dev/null +++ b/internal/providers/acp/acp_gemini_test.go @@ -0,0 +1,59 @@ +package acp + +import ( + "context" + "fmt" + "log/slog" + "os" + "testing" + "time" +) + +func TestGeminiProtocolMapping(t *testing.T) { + if os.Getenv("ACP_GEMINI_E2E") == "" { + t.Skip("set ACP_GEMINI_E2E=1 to run this test (requires gemini binary)") + } + // Enable debug logging to stderr + slog.SetDefault(slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelDebug}))) + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + pp := NewProcessPool("gemini", []string{"--acp", "--yolo"}, ".", 10*time.Minute) + defer pp.Close() + + proc, err := pp.GetOrSpawn(ctx, "test-session") + if err != nil { + t.Fatalf("Spawn failed: %v", err) + } + + sid, err := proc.NewSession(ctx) + if err != nil { + t.Fatalf("NewSession failed: %v", err) + } + + fmt.Println("-> Testing Gemini Protocol Mapping...") + + var collectedText string + _, err = proc.Prompt(ctx, sid, []ContentBlock{ + {Type: "text", Text: "Hello, reply with 'ACP_SUCCESS' and nothing else."}, + }, func(su SessionUpdate) { + if su.Message != nil { + for _, b := range su.Message.Content { + if b.Type == "text" { + fmt.Printf("<- MAPPED CHUNK: %s\n", b.Text) + collectedText += b.Text + } + } + } + }) + + if err != nil { + t.Fatalf("Prompt failed: %v", err) + } + + fmt.Printf("== COLLECTED TOTAL: %s ==\n", collectedText) + if collectedText == "" { + t.Error("No text collected via mapped Message field!") + } +} diff --git a/internal/providers/acp/helpers.go b/internal/providers/acp/helpers.go index fade9700..a89f6259 100644 --- a/internal/providers/acp/helpers.go +++ b/internal/providers/acp/helpers.go @@ -1,22 +1,50 @@ package acp import ( + "context" "io" "strings" "sync" ) +// contextKey is the unexported type for context values set by this package. +type contextKey string + +const goclawSessionCtxKey contextKey = "goclaw_session" + +// WithGoclawSession attaches the goclaw conversation session key to ctx so +// that ACP-level logs can correlate the ACP session ID with the goclaw session. +func WithGoclawSession(ctx context.Context, key string) context.Context { + return context.WithValue(ctx, goclawSessionCtxKey, key) +} + +// goclawSessionFromCtx extracts the goclaw session key injected by WithGoclawSession. +// Returns "" if not set. +func goclawSessionFromCtx(ctx context.Context) string { + v, _ := ctx.Value(goclawSessionCtxKey).(string) + return v +} + // sensitiveEnvPrefixes lists env var prefixes stripped from ACP subprocesses. var sensitiveEnvPrefixes = []string{ "GOCLAW", "CLAUDE", "ANTHROPIC", "OPENAI", "DATABASE", "POSTGRES", "MYSQL", "REDIS", "MONGO", - "AWS_", "GOOGLE_", "AZURE_", "GCP_", + "AWS_", "AZURE_", "GOOGLE_", "GCP_", "GITHUB_", "GH_", "GITLAB_", "BITBUCKET_", "DOCKER_", "REGISTRY_", "STRIPE_", "TWILIO_", "SENDGRID_", "SSH_", "GPG_", } +// allowedEnvExact lists env vars explicitly allowed even if they match sensitive prefixes. +// These are required for Google/GCP authentication in ACP subprocesses (e.g., Gemini). +var allowedEnvExact = map[string]bool{ + "GOOGLE_API_KEY": true, + "GOOGLE_APPLICATION_CREDENTIALS": true, + "GOOGLE_CLOUD_PROJECT": true, + "GCP_PROJECT": true, +} + // sensitiveEnvExact lists exact env var names stripped from ACP subprocesses. var sensitiveEnvExact = map[string]bool{ "DB_DSN": true, "PGPASSWORD": true, "PGUSER": true, "PGHOST": true, @@ -28,11 +56,16 @@ var sensitiveEnvExact = map[string]bool{ } // filterACPEnv strips sensitive env vars from the subprocess environment. +// Explicitly allowed vars (allowedEnvExact) pass through even if they match sensitive prefixes. func filterACPEnv(environ []string) []string { var filtered []string for _, e := range environ { key, _, _ := strings.Cut(e, "=") upper := strings.ToUpper(key) + if allowedEnvExact[upper] { + filtered = append(filtered, e) + continue + } if sensitiveEnvExact[upper] { continue } diff --git a/internal/providers/acp/jsonrpc.go b/internal/providers/acp/jsonrpc.go index 1386a054..8058099c 100644 --- a/internal/providers/acp/jsonrpc.go +++ b/internal/providers/acp/jsonrpc.go @@ -72,6 +72,7 @@ func (c *Conn) readLoop() { for scanner.Scan() { line := scanner.Bytes() + slog.Debug("acp.jsonrpc: < READ", "line", string(line)) if len(line) == 0 { continue } @@ -207,6 +208,7 @@ func (c *Conn) writeMessage(msg *jsonrpcMessage) error { c.mu.Lock() defer c.mu.Unlock() + slog.Debug("acp.jsonrpc: > WRITE", "data", string(data)) _, err = c.writer.Write(data) return err } diff --git a/internal/providers/acp/process.go b/internal/providers/acp/process.go index 1b0644d3..7e34de51 100644 --- a/internal/providers/acp/process.go +++ b/internal/providers/acp/process.go @@ -13,49 +13,107 @@ import ( ) // ACPProcess represents a running ACP agent subprocess. +// One process is shared across all sessions — each goclaw conversation +// creates its own ACP session (identified by session ID) on this process. type ACPProcess struct { - cmd *exec.Cmd - conn *Conn - sessionID string // ACP session ID from session/new + cmd *exec.Cmd + conn *Conn + agentCaps AgentCaps + workDir string lastActive time.Time - inUse atomic.Int32 // >0 means prompt active — reaper must skip + inUse atomic.Int32 // >0 means at least one prompt is active — reaper must skip mu sync.Mutex ctx context.Context cancel context.CancelFunc exited chan struct{} // closed when process exits - // updateFn is called for session/update notifications during a prompt. - updateFn func(SessionUpdate) - updateMu sync.Mutex + // updateFns routes session/update notifications to the correct active prompt. + updateFns map[string]func(SessionUpdate) + updateMu sync.Mutex } -// setUpdateFn sets the callback for session/update notifications (thread-safe). -func (p *ACPProcess) setUpdateFn(fn func(SessionUpdate)) { +// AgentCaps returns the capability flags reported by the agent during Initialize. +func (p *ACPProcess) AgentCaps() AgentCaps { + return p.agentCaps +} + +// registerUpdateFn registers a callback for session/update notifications on sessionID. +func (p *ACPProcess) registerUpdateFn(sid string, fn func(SessionUpdate)) { p.updateMu.Lock() - p.updateFn = fn - p.updateMu.Unlock() + defer p.updateMu.Unlock() + if p.updateFns == nil { + p.updateFns = make(map[string]func(SessionUpdate)) + } + p.updateFns[sid] = fn } -// dispatchUpdate routes a session/update to the active prompt callback. +// unregisterUpdateFn removes the callback for sessionID after a Prompt completes. +func (p *ACPProcess) unregisterUpdateFn(sid string) { + p.updateMu.Lock() + defer p.updateMu.Unlock() + delete(p.updateFns, sid) +} + +// dispatchUpdate routes a session/update notification to the registered callback. +// It also performs Gemini ACP protocol mapping: the "agent_message_chunk" update type +// carries content in Update.Content rather than the standard Message field; this is +// normalized here so all callers receive a consistent SessionUpdate. func (p *ACPProcess) dispatchUpdate(update SessionUpdate) { + // Gemini protocol mapping: agent_message_chunk → Message + if update.Update.SessionUpdate == "agent_message_chunk" && len(update.Update.Content) > 0 { + if update.Message == nil { + update.Message = &MessageUpdate{Role: "assistant"} + } + // Content may arrive as a single object {"type":"text","text":"..."} or an array + var single struct { + Type string `json:"type"` + Text string `json:"text"` + } + if err := json.Unmarshal(update.Update.Content, &single); err == nil && single.Type != "" { + update.Message.Content = append(update.Message.Content, ContentBlock{ + Type: single.Type, + Text: single.Text, + }) + } else { + var arr []struct { + Type string `json:"type"` + Text string `json:"text"` + } + if err := json.Unmarshal(update.Update.Content, &arr); err == nil { + for _, c := range arr { + update.Message.Content = append(update.Message.Content, ContentBlock{ + Type: c.Type, + Text: c.Text, + }) + } + } + } + } + p.updateMu.Lock() - fn := p.updateFn + fn, ok := p.updateFns[update.SessionID] p.updateMu.Unlock() + if !ok { + slog.Debug("acp: session/update with no registered callback", "sid", update.SessionID) + return + } if fn != nil { fn(update) } } -// ProcessPool manages a pool of ACP agent subprocesses keyed by session. +// ProcessPool manages a pool of ACP agent subprocesses. +// Typically a single shared process is used (poolKey = binary identifier), +// and multiple ACP sessions are multiplexed over it. type ProcessPool struct { - processes sync.Map // sessionKey → *ACPProcess - spawnMu sync.Map // sessionKey → *sync.Mutex — prevents concurrent spawn + processes sync.Map // poolKey → *ACPProcess + spawnMu sync.Map // poolKey → *sync.Mutex — prevents concurrent spawn agentBinary string agentArgs []string workDir string idleTTL time.Duration - mu sync.RWMutex // protects toolHandler + mu sync.RWMutex // protects toolHandler toolHandler RequestHandler done chan struct{} closeOnce sync.Once @@ -89,36 +147,37 @@ func (pp *ProcessPool) getToolHandler() RequestHandler { return pp.toolHandler } -// GetOrSpawn returns an existing process for the session key or spawns a new one. -// Uses per-key mutex to prevent concurrent spawn for the same session. -func (pp *ProcessPool) GetOrSpawn(ctx context.Context, sessionKey string) (*ACPProcess, error) { - // Acquire per-key spawn lock to prevent concurrent spawns - actual, _ := pp.spawnMu.LoadOrStore(sessionKey, &sync.Mutex{}) +// GetOrSpawn returns an existing process for the pool key or spawns a new one. +// Uses per-key mutex to prevent concurrent spawn for the same key. +func (pp *ProcessPool) GetOrSpawn(ctx context.Context, poolKey string) (*ACPProcess, error) { + actual, _ := pp.spawnMu.LoadOrStore(poolKey, &sync.Mutex{}) mu := actual.(*sync.Mutex) mu.Lock() defer mu.Unlock() - if val, ok := pp.processes.Load(sessionKey); ok { + if val, ok := pp.processes.Load(poolKey); ok { proc := val.(*ACPProcess) select { case <-proc.exited: - // Process crashed — remove and respawn - pp.processes.Delete(sessionKey) - slog.Info("acp: respawning crashed process", "session_key", sessionKey) + pp.processes.Delete(poolKey) + slog.Info("acp: respawning crashed process", "pool_key", poolKey) default: return proc, nil } } - return pp.spawn(ctx, sessionKey) + return pp.spawn(ctx, poolKey) } -// spawn creates a new ACP subprocess, initializes it, and creates a session. -func (pp *ProcessPool) spawn(ctx context.Context, sessionKey string) (*ACPProcess, error) { +// spawn creates a new ACP subprocess and performs the ACP initialize handshake. +// Session creation (session/new) is NOT done here — the provider handles that +// per-conversation via NewSession or LoadSession. +func (pp *ProcessPool) spawn(ctx context.Context, poolKey string) (*ACPProcess, error) { procCtx, cancel := context.WithCancel(context.Background()) cmd := exec.CommandContext(procCtx, pp.agentBinary, pp.agentArgs...) cmd.Dir = pp.workDir cmd.Env = filterACPEnv(os.Environ()) + cmd.SysProcAttr = sysProcAttr() stdinPipe, err := cmd.StdinPipe() if err != nil { @@ -130,9 +189,9 @@ func (pp *ProcessPool) spawn(ctx context.Context, sessionKey string) (*ACPProces cancel() return nil, fmt.Errorf("acp: stdout pipe: %w", err) } - // Capture stderr for diagnostics (capped via limitedWriter) cmd.Stderr = &limitedWriter{max: 4096} + slog.Info("acp: starting subprocess", "pool_key", poolKey, "binary", pp.agentBinary, "args", pp.agentArgs) if err := cmd.Start(); err != nil { cancel() return nil, fmt.Errorf("acp: start %s: %w", pp.agentBinary, err) @@ -144,14 +203,17 @@ func (pp *ProcessPool) spawn(ctx context.Context, sessionKey string) (*ACPProces ctx: procCtx, cancel: cancel, exited: make(chan struct{}), + workDir: pp.workDir, } - // Notification handler: route session/update to active prompt callback + // Notification handler: log all notifications and dispatch session/update to callers notifyHandler := func(method string, params json.RawMessage) { + slog.Info("acp: notification received", "method", method) + slog.Debug("acp: notification params", "method", method, "params", string(params)) if method == "session/update" { var update SessionUpdate if err := json.Unmarshal(params, &update); err != nil { - slog.Warn("acp: failed to parse session/update", "error", err) + slog.Warn("acp: session/update parse failed", "error", err) return } proc.dispatchUpdate(update) @@ -161,28 +223,23 @@ func (pp *ProcessPool) spawn(ctx context.Context, sessionKey string) (*ACPProces proc.conn = NewConn(stdinPipe, stdoutPipe, pp.getToolHandler(), notifyHandler) proc.conn.Start() - // Monitor process exit and log stderr stderrWriter := cmd.Stderr.(*limitedWriter) go func() { _ = cmd.Wait() if s := stderrWriter.String(); s != "" { - slog.Debug("acp: process stderr", "session_key", sessionKey, "stderr", s) + slog.Debug("acp: process stderr", "pool_key", poolKey, "stderr", s) } close(proc.exited) }() - // ACP handshake: initialize + session/new + slog.Info("acp: performing handshake (initialize)", "pool_key", poolKey) if err := proc.Initialize(ctx); err != nil { cancel() return nil, err } - if err := proc.NewSession(ctx); err != nil { - cancel() - return nil, err - } - pp.processes.Store(sessionKey, proc) - slog.Info("acp: process spawned", "session_key", sessionKey, "binary", pp.agentBinary) + pp.processes.Store(poolKey, proc) + slog.Info("acp: process spawned", "pool_key", poolKey, "binary", pp.agentBinary) return proc, nil } @@ -195,7 +252,6 @@ func (pp *ProcessPool) reapLoop() { case <-ticker.C: pp.processes.Range(func(key, value any) bool { proc := value.(*ACPProcess) - // Skip processes with active prompts if proc.inUse.Load() > 0 { return true } @@ -203,7 +259,7 @@ func (pp *ProcessPool) reapLoop() { idle := time.Since(proc.lastActive) > pp.idleTTL proc.mu.Unlock() if idle { - slog.Info("acp: reaping idle process", "session_key", key) + slog.Info("acp: reaping idle process", "pool_key", key) proc.cancel() pp.processes.Delete(key) } @@ -215,10 +271,8 @@ func (pp *ProcessPool) reapLoop() { } } -// processCloseTimeout is the per-process max wait during ProcessPool.Close -// before we log a warning and move on. Production default = 5s. Exposed as a -// package var so tests can override to a few ms, avoiding 5s of dead CI time -// when exercising the "process never exits" path. +// processCloseTimeout is the per-process max wait during ProcessPool.Close. +// Exposed as a package var so tests can shorten it. var processCloseTimeout = 5 * time.Second // Close shuts down all processes gracefully. @@ -228,11 +282,10 @@ func (pp *ProcessPool) Close() error { pp.processes.Range(func(key, value any) bool { proc := value.(*ACPProcess) proc.cancel() - // Wait briefly for process to exit select { case <-proc.exited: case <-time.After(processCloseTimeout): - slog.Warn("acp: process did not exit in time", "session_key", key) + slog.Warn("acp: process did not exit in time", "pool_key", key) } pp.processes.Delete(key) return true @@ -240,4 +293,3 @@ func (pp *ProcessPool) Close() error { }) return nil } - diff --git a/internal/providers/acp/session.go b/internal/providers/acp/session.go index ba0fa361..016d4303 100644 --- a/internal/providers/acp/session.go +++ b/internal/providers/acp/session.go @@ -3,39 +3,75 @@ package acp import ( "context" "fmt" + "log/slog" + "path/filepath" "time" ) // Initialize sends the ACP initialize request to establish capabilities. func (p *ACPProcess) Initialize(ctx context.Context) error { + ctx, cancel := context.WithTimeout(ctx, 60*time.Second) + defer cancel() req := InitializeRequest{ - ClientInfo: ClientInfo{Name: "goclaw", Version: "1.0"}, - Capabilities: ClientCaps{ - Fs: &FsCaps{ReadTextFile: true, WriteTextFile: true}, - Terminal: &TerminalCaps{Enabled: true}, - }, + ProtocolVersion: 1, + ClientInfo: ClientInfo{Name: "GoClaw", Version: "1.0"}, + Capabilities: ClientCaps{}, } var resp InitializeResponse if err := p.conn.Call(ctx, "initialize", req, &resp); err != nil { return fmt.Errorf("acp initialize: %w", err) } p.agentCaps = resp.Capabilities + slog.Info("acp: initialized", "agent", resp.AgentInfo.Name, "version", resp.AgentInfo.Version, "loadSession", resp.Capabilities.LoadSession) return nil } -// NewSession creates a new ACP session on this process. -func (p *ACPProcess) NewSession(ctx context.Context) error { - var resp NewSessionResponse - if err := p.conn.Call(ctx, "session/new", NewSessionRequest{}, &resp); err != nil { - return fmt.Errorf("acp session/new: %w", err) +// NewSession creates a new ACP session and returns its session ID. +func (p *ACPProcess) NewSession(ctx context.Context) (string, error) { + ctx, cancel := context.WithTimeout(ctx, 60*time.Second) + defer cancel() + + cwd := p.workDir + if cwd == "" { + cwd, _ = filepath.Abs(".") } - p.sessionID = resp.SessionID - return nil + + req := NewSessionRequest{ + Cwd: cwd, + McpServers: []string{}, + } + var resp NewSessionResponse + if err := p.conn.Call(ctx, "session/new", req, &resp); err != nil { + return "", fmt.Errorf("acp session/new: %w", err) + } + slog.Info("acp: session/new", "sid", resp.SessionID, "cwd", cwd) + return resp.SessionID, nil } -// Prompt sends user content and blocks until the agent responds. -// onUpdate is called for each session/update notification (streaming). -func (p *ACPProcess) Prompt(ctx context.Context, content []ContentBlock, onUpdate func(SessionUpdate)) (*PromptResponse, error) { +// LoadSession restores a previous ACP session by ID (used after process restart). +// Returns the session ID to use going forward (may equal the requested ID). +// Only call if AgentCaps().LoadSession is true. +func (p *ACPProcess) LoadSession(ctx context.Context, sessionID string) (string, error) { + ctx, cancel := context.WithTimeout(ctx, 60*time.Second) + defer cancel() + + cwd := p.workDir + if cwd == "" { + cwd, _ = filepath.Abs(".") + } + + req := LoadSessionRequest{SessionID: sessionID, Cwd: cwd} + var resp LoadSessionResponse + if err := p.conn.Call(ctx, "session/load", req, &resp); err != nil { + return "", fmt.Errorf("acp session/load: %w", err) + } + slog.Info("acp: session/load", "sid", resp.SessionID) + return resp.SessionID, nil +} + +// Prompt sends user content to sessionID and blocks until the agent completes, +// invoking onUpdate for each session/update notification received. +func (p *ACPProcess) Prompt(ctx context.Context, sessionID string, content []ContentBlock, onUpdate func(SessionUpdate)) (*PromptResponse, error) { p.inUse.Add(1) defer p.inUse.Add(-1) @@ -43,14 +79,16 @@ func (p *ACPProcess) Prompt(ctx context.Context, content []ContentBlock, onUpdat p.lastActive = time.Now() p.mu.Unlock() - // Install the update callback for this prompt - p.setUpdateFn(onUpdate) - defer p.setUpdateFn(nil) + p.registerUpdateFn(sessionID, onUpdate) + defer p.unregisterUpdateFn(sessionID) + goclawSession := goclawSessionFromCtx(ctx) + slog.Info("acp: session/prompt", "session", goclawSession, "sid", sessionID) req := PromptRequest{ - SessionID: p.sessionID, - Content: content, + SessionID: sessionID, + Prompt: content, } + var resp PromptResponse if err := p.conn.Call(ctx, "session/prompt", req, &resp); err != nil { return nil, fmt.Errorf("acp session/prompt: %w", err) @@ -60,12 +98,13 @@ func (p *ACPProcess) Prompt(ctx context.Context, content []ContentBlock, onUpdat p.lastActive = time.Now() p.mu.Unlock() + slog.Info("acp: session/prompt completed", "session", goclawSession, "sid", sessionID, "stopReason", resp.StopReason) return &resp, nil } -// Cancel sends a session/cancel notification for cooperative cancellation. -func (p *ACPProcess) Cancel() error { +// Cancel sends a session/cancel notification for the given session. +func (p *ACPProcess) Cancel(sessionID string) error { return p.conn.Notify("session/cancel", CancelNotification{ - SessionID: p.sessionID, + SessionID: sessionID, }) } diff --git a/internal/providers/acp/session_test.go b/internal/providers/acp/session_test.go index 8ba8f41f..3f048b5e 100644 --- a/internal/providers/acp/session_test.go +++ b/internal/providers/acp/session_test.go @@ -18,7 +18,7 @@ import ( // // clientW → serverR (test reads what the process sends) // serverW → clientR (test injects responses to the process) -func buildACPProcess(handler RequestHandler, notify NotifyHandler) ( +func buildACPProcess(handler RequestHandler, extraNotify NotifyHandler) ( proc *ACPProcess, serverW io.WriteCloser, serverR io.ReadCloser, @@ -26,15 +26,31 @@ func buildACPProcess(handler RequestHandler, notify NotifyHandler) ( clientR, sW := io.Pipe() // test writes here → conn reads sR, clientW := io.Pipe() // conn writes here → test reads - conn := NewConn(clientW, clientR, handler, notify) - conn.Start() - + // Create proc first so the notifyHandler closure can reference it. proc = &ACPProcess{ - conn: conn, exited: make(chan struct{}), ctx: context.Background(), cancel: func() {}, } + + // Mirror production spawn: always route session/update to proc.dispatchUpdate, + // then optionally forward to the test's extra notify hook. + internalNotify := func(method string, params json.RawMessage) { + if method == "session/update" { + var update SessionUpdate + if json.Unmarshal(params, &update) == nil { + proc.dispatchUpdate(update) + } + } + if extraNotify != nil { + extraNotify(method, params) + } + } + + conn := NewConn(clientW, clientR, handler, internalNotify) + conn.Start() + proc.conn = conn + return proc, sW, sR } @@ -119,7 +135,7 @@ func TestACPProcess_Initialize_Success(t *testing.T) { defer serverW.Close() defer serverR.Close() - respJSON := `{"agentInfo":{"name":"claude","version":"1.0"},"capabilities":{"loadSession":true}}` + respJSON := `{"agentInfo":{"name":"claude","version":"1.0"},"agentCapabilities":{"loadSession":true}}` done := replyTo(serverR, serverW, respJSON) ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) @@ -197,11 +213,12 @@ func TestACPProcess_NewSession_Success(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) defer cancel() - if err := proc.NewSession(ctx); err != nil { + sid, err := proc.NewSession(ctx) + if err != nil { t.Fatalf("NewSession error: %v", err) } - if proc.sessionID != "sess-xyz" { - t.Errorf("expected sessionID='sess-xyz', got %q", proc.sessionID) + if sid != "sess-xyz" { + t.Errorf("expected sessionID='sess-xyz', got %q", sid) } <-done } @@ -216,7 +233,7 @@ func TestACPProcess_NewSession_Error(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) defer cancel() - err := proc.NewSession(ctx) + _, err := proc.NewSession(ctx) if err == nil { t.Fatal("expected error from NewSession") } @@ -230,7 +247,6 @@ func TestACPProcess_NewSession_Error(t *testing.T) { func TestACPProcess_Prompt_Success(t *testing.T) { proc, serverW, serverR := buildACPProcess(nil, nil) - proc.sessionID = "sess-1" defer serverW.Close() defer serverR.Close() @@ -239,7 +255,7 @@ func TestACPProcess_Prompt_Success(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) defer cancel() - resp, err := proc.Prompt(ctx, []ContentBlock{{Type: "text", Text: "hi"}}, nil) + resp, err := proc.Prompt(ctx, "sess-1", []ContentBlock{{Type: "text", Text: "hi"}}, nil) if err != nil { t.Fatalf("Prompt error: %v", err) } @@ -260,23 +276,23 @@ func TestACPProcess_Prompt_WithUpdateCallback(t *testing.T) { } proc, serverW, serverR := buildACPProcess(nil, notifyHandler) - proc.sessionID = "sess-2" defer serverW.Close() defer serverR.Close() + const sid = "sess-2" + // Goroutine: send a notification, then a response go func() { - // Read the prompt request first buf := make([]byte, 32*1024) n, _ := serverR.Read(buf) var req jsonrpcMessage json.Unmarshal(buf[:n], &req) - // Send session/update notification + // Send session/update notification with matching session ID notif := jsonrpcMessage{ JSONRPC: "2.0", Method: "session/update", - Params: json.RawMessage(`{"kind":"message","stopReason":""}`), + Params: json.RawMessage(`{"sessionId":"sess-2","kind":"message","stopReason":""}`), } nd, _ := json.Marshal(notif) serverW.Write(append(nd, '\n')) @@ -295,21 +311,23 @@ func TestACPProcess_Prompt_WithUpdateCallback(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) defer cancel() - _, err := proc.Prompt(ctx, []ContentBlock{{Type: "text", Text: "hello"}}, func(u SessionUpdate) { + _, err := proc.Prompt(ctx, sid, []ContentBlock{{Type: "text", Text: "hello"}}, func(u SessionUpdate) { received = append(received, u) }) if err != nil { t.Fatalf("Prompt error: %v", err) } - // Drain notification channel + // Allow notification delivery time.Sleep(20 * time.Millisecond) - // The update callback + notify handler may both fire — just check no panic occurred + // Verify no panic and at least one update was dispatched + if len(received) == 0 { + t.Error("expected at least one session/update to be dispatched") + } } func TestACPProcess_Prompt_Error(t *testing.T) { proc, serverW, serverR := buildACPProcess(nil, nil) - proc.sessionID = "sess-3" defer serverW.Close() defer serverR.Close() @@ -318,7 +336,7 @@ func TestACPProcess_Prompt_Error(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) defer cancel() - _, err := proc.Prompt(ctx, []ContentBlock{{Type: "text", Text: "hi"}}, nil) + _, err := proc.Prompt(ctx, "sess-3", []ContentBlock{{Type: "text", Text: "hi"}}, nil) if err == nil { t.Fatal("expected error from Prompt") } @@ -329,15 +347,12 @@ func TestACPProcess_Prompt_Error(t *testing.T) { } func TestACPProcess_Prompt_SetsInUse(t *testing.T) { - // Verify inUse counter goes up during Prompt and back to 0 after proc, serverW, serverR := buildACPProcess(nil, nil) - proc.sessionID = "sess-4" defer serverW.Close() defer serverR.Close() started := make(chan struct{}) go func() { - // Read and wait a moment before responding buf := make([]byte, 32*1024) n, _ := serverR.Read(buf) var req jsonrpcMessage @@ -360,7 +375,7 @@ func TestACPProcess_Prompt_SetsInUse(t *testing.T) { defer close(done) ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) defer cancel() - proc.Prompt(ctx, []ContentBlock{{Type: "text", Text: "x"}}, nil) + proc.Prompt(ctx, "sess-4", []ContentBlock{{Type: "text", Text: "x"}}, nil) }() <-started @@ -378,7 +393,6 @@ func TestACPProcess_Prompt_SetsInUse(t *testing.T) { func TestACPProcess_Cancel_SendsNotification(t *testing.T) { proc, serverW, serverR := buildACPProcess(nil, nil) - proc.sessionID = "sess-cancel" defer serverW.Close() received := make(chan string, 1) @@ -390,7 +404,7 @@ func TestACPProcess_Cancel_SendsNotification(t *testing.T) { received <- msg.Method }() - err := proc.Cancel() + err := proc.Cancel("sess-cancel") if err != nil { t.Fatalf("Cancel error: %v", err) } @@ -405,30 +419,48 @@ func TestACPProcess_Cancel_SendsNotification(t *testing.T) { } } -// --- dispatchUpdate / setUpdateFn --- +// --- dispatchUpdate / registerUpdateFn --- func TestACPProcess_DispatchUpdate_NilFn(t *testing.T) { proc := &ACPProcess{} - // Should not panic with nil updateFn - proc.dispatchUpdate(SessionUpdate{Kind: "message"}) + // Should not panic with no registered fn for the session ID + proc.dispatchUpdate(SessionUpdate{SessionID: "x", Kind: "message"}) } func TestACPProcess_DispatchUpdate_CallsFn(t *testing.T) { proc := &ACPProcess{} called := false - proc.setUpdateFn(func(u SessionUpdate) { + proc.registerUpdateFn("sid-1", func(u SessionUpdate) { called = true }) - proc.dispatchUpdate(SessionUpdate{Kind: "message"}) + proc.dispatchUpdate(SessionUpdate{SessionID: "sid-1", Kind: "message"}) if !called { t.Error("expected updateFn to be called") } } -func TestACPProcess_SetUpdateFn_Nil(t *testing.T) { +func TestACPProcess_DispatchUpdate_RoutesCorrectSession(t *testing.T) { proc := &ACPProcess{} - proc.setUpdateFn(func(u SessionUpdate) {}) - proc.setUpdateFn(nil) - // Should not panic - proc.dispatchUpdate(SessionUpdate{Kind: "x"}) + var calledA, calledB bool + proc.registerUpdateFn("sid-A", func(u SessionUpdate) { calledA = true }) + proc.registerUpdateFn("sid-B", func(u SessionUpdate) { calledB = true }) + + proc.dispatchUpdate(SessionUpdate{SessionID: "sid-A"}) + if !calledA { + t.Error("expected fn for sid-A to be called") + } + if calledB { + t.Error("expected fn for sid-B NOT to be called") + } +} + +func TestACPProcess_UnregisterUpdateFn(t *testing.T) { + proc := &ACPProcess{} + called := false + proc.registerUpdateFn("sid-x", func(u SessionUpdate) { called = true }) + proc.unregisterUpdateFn("sid-x") + proc.dispatchUpdate(SessionUpdate{SessionID: "sid-x"}) + if called { + t.Error("expected fn to not be called after unregister") + } } diff --git a/internal/providers/acp/sysproc_linux.go b/internal/providers/acp/sysproc_linux.go new file mode 100644 index 00000000..6c29393c --- /dev/null +++ b/internal/providers/acp/sysproc_linux.go @@ -0,0 +1,11 @@ +//go:build linux + +package acp + +import "syscall" + +func sysProcAttr() *syscall.SysProcAttr { + return &syscall.SysProcAttr{ + Pdeathsig: syscall.SIGKILL, + } +} diff --git a/internal/providers/acp/sysproc_other.go b/internal/providers/acp/sysproc_other.go new file mode 100644 index 00000000..f9942b63 --- /dev/null +++ b/internal/providers/acp/sysproc_other.go @@ -0,0 +1,9 @@ +//go:build !linux + +package acp + +import "syscall" + +func sysProcAttr() *syscall.SysProcAttr { + return nil +} diff --git a/internal/providers/acp/types.go b/internal/providers/acp/types.go index 216bf5a4..9177e505 100644 --- a/internal/providers/acp/types.go +++ b/internal/providers/acp/types.go @@ -1,129 +1,148 @@ package acp -// ACP protocol types — client-side subset for GoClaw as ACP client. -// Covers: initialize, session lifecycle, content blocks, and agent→client requests. +import "encoding/json" -// --- Client → Agent Requests --- +// --- Client -> Agent Requests --- -// InitializeRequest starts the ACP handshake. type InitializeRequest struct { - ClientInfo ClientInfo `json:"clientInfo"` - Capabilities ClientCaps `json:"capabilities"` + ProtocolVersion int `json:"protocolVersion"` + ClientInfo ClientInfo `json:"clientInfo"` + Capabilities ClientCaps `json:"clientCapabilities"` } -// ClientInfo identifies the ACP client. type ClientInfo struct { Name string `json:"name"` Version string `json:"version"` } -// ClientCaps declares what the client can handle (fs, terminal, etc.). type ClientCaps struct { Fs *FsCaps `json:"fs,omitempty"` Terminal *TerminalCaps `json:"terminal,omitempty"` } -// FsCaps declares filesystem capabilities. type FsCaps struct { ReadTextFile bool `json:"readTextFile"` WriteTextFile bool `json:"writeTextFile"` } -// TerminalCaps declares terminal capabilities. type TerminalCaps struct { Enabled bool `json:"enabled"` } -// InitializeResponse carries the agent's identity and capabilities. type InitializeResponse struct { AgentInfo AgentInfo `json:"agentInfo"` - Capabilities AgentCaps `json:"capabilities"` + Capabilities AgentCaps `json:"agentCapabilities"` } -// AgentInfo identifies the ACP agent. type AgentInfo struct { Name string `json:"name"` Version string `json:"version"` } -// AgentCaps declares agent capabilities. type AgentCaps struct { LoadSession bool `json:"loadSession"` PromptCapabilities *PromptCaps `json:"promptCapabilities,omitempty"` SessionCapabilities *SessionCaps `json:"sessionCapabilities,omitempty"` + MCPCapabilities *MCPCaps `json:"mcpCapabilities,omitempty"` } -// PromptCaps describes what content types the agent accepts. type PromptCaps struct { Audio bool `json:"audio"` Image bool `json:"image"` EmbeddedContext bool `json:"embeddedContext"` } -// SessionCaps describes session-level capabilities. type SessionCaps struct{} +type MCPCaps struct { + HTTP bool `json:"http"` + SSE bool `json:"sse"` +} + // --- Session Methods --- -// NewSessionRequest creates a new ACP session. -type NewSessionRequest struct{} +type NewSessionRequest struct { + Cwd string `json:"cwd"` + McpServers []string `json:"mcpServers"` +} -// NewSessionResponse carries the new session ID. type NewSessionResponse struct { SessionID string `json:"sessionId"` } -// PromptRequest sends user content to the agent. +type LoadSessionRequest struct { + SessionID string `json:"sessionId"` + Cwd string `json:"cwd,omitempty"` + McpServers []string `json:"mcpServers"` +} + +type LoadSessionResponse struct { + SessionID string `json:"sessionId"` +} + type PromptRequest struct { SessionID string `json:"sessionId"` - Content []ContentBlock `json:"content"` + Prompt []ContentBlock `json:"prompt"` } -// PromptResponse is the final response after the agent completes. type PromptResponse struct { StopReason string `json:"stopReason,omitempty"` } -// CancelNotification requests cooperative cancellation. type CancelNotification struct { SessionID string `json:"sessionId"` } // --- Content Blocks --- -// ContentBlock represents a piece of content (text, image, audio). type ContentBlock struct { Type string `json:"type"` // "text", "image", "audio" Text string `json:"text,omitempty"` - Data string `json:"data,omitempty"` // base64 for image/audio + Data string `json:"data,omitempty"` MimeType string `json:"mimeType,omitempty"` } -// --- Agent → Client Notifications --- +// --- Agent -> Client Notifications --- -// SessionUpdate carries incremental updates during prompt execution. type SessionUpdate struct { - Kind string `json:"kind"` // "message", "toolCall", "plan" - StopReason string `json:"stopReason,omitempty"` // "endTurn", "cancelled" - Message *MessageUpdate `json:"message,omitempty"` - ToolCall *ToolCallUpdate `json:"toolCall,omitempty"` + SessionID string `json:"sessionId"` + StopReason string `json:"stopReason,omitempty"` + + Kind string `json:"kind,omitempty"` + Message *MessageUpdate `json:"message,omitempty"` + ToolCall *ToolCallUpdate `json:"toolCall,omitempty"` + + Update struct { + SessionUpdate string `json:"sessionUpdate"` + + Content json.RawMessage `json:"content,omitempty"` + + Entries []struct { + Content string `json:"content"` + Priority string `json:"priority"` + Status string `json:"status"` + } `json:"entries,omitempty"` + + ToolCallID string `json:"toolCallId,omitempty"` + Title string `json:"title,omitempty"` + Kind string `json:"kind,omitempty"` + Status string `json:"status,omitempty"` + } `json:"update"` } -// MessageUpdate carries an assistant text delta. type MessageUpdate struct { Role string `json:"role"` Content []ContentBlock `json:"content"` } -// ToolCallUpdate carries tool call progress. type ToolCallUpdate struct { ID string `json:"id"` Name string `json:"name"` - Status string `json:"status"` // "running", "completed" + Status string `json:"status"` Content []ContentBlock `json:"content,omitempty"` } -// --- Agent → Client Requests (fs/terminal/permission) --- +// --- Agent -> Client Requests --- type ReadTextFileRequest struct { Path string `json:"path"` @@ -185,5 +204,5 @@ type RequestPermissionRequest struct { } type RequestPermissionResponse struct { - Outcome string `json:"outcome"` // "approved", "denied", "cancelled" + Outcome string `json:"outcome"` // "proceed_always", "approved", "denied" } diff --git a/internal/providers/acp/types_test.go b/internal/providers/acp/types_test.go index 7ec2543c..293127f4 100644 --- a/internal/providers/acp/types_test.go +++ b/internal/providers/acp/types_test.go @@ -85,7 +85,7 @@ func TestNewSessionResponse_RoundTrip(t *testing.T) { func TestPromptRequest_RoundTrip(t *testing.T) { req := PromptRequest{ SessionID: "sess-1", - Content: []ContentBlock{ + Prompt: []ContentBlock{ {Type: "text", Text: "hello"}, {Type: "image", Data: "base64data", MimeType: "image/png"}, }, @@ -94,14 +94,14 @@ func TestPromptRequest_RoundTrip(t *testing.T) { if got.SessionID != "sess-1" { t.Errorf("SessionID: got %q", got.SessionID) } - if len(got.Content) != 2 { - t.Fatalf("expected 2 content blocks, got %d", len(got.Content)) + if len(got.Prompt) != 2 { + t.Fatalf("expected 2 content blocks, got %d", len(got.Prompt)) } - if got.Content[0].Type != "text" || got.Content[0].Text != "hello" { - t.Errorf("content[0]: got type=%q text=%q", got.Content[0].Type, got.Content[0].Text) + if got.Prompt[0].Type != "text" || got.Prompt[0].Text != "hello" { + t.Errorf("content[0]: got type=%q text=%q", got.Prompt[0].Type, got.Prompt[0].Text) } - if got.Content[1].Type != "image" || got.Content[1].Data != "base64data" { - t.Errorf("content[1]: got type=%q data=%q", got.Content[1].Type, got.Content[1].Data) + if got.Prompt[1].Type != "image" || got.Prompt[1].Data != "base64data" { + t.Errorf("content[1]: got type=%q data=%q", got.Prompt[1].Type, got.Prompt[1].Data) } } diff --git a/internal/providers/acp_provider.go b/internal/providers/acp_provider.go index 915c66c2..17380d5d 100644 --- a/internal/providers/acp_provider.go +++ b/internal/providers/acp_provider.go @@ -13,15 +13,28 @@ import ( "github.com/nextlevelbuilder/goclaw/internal/providers/acp" ) +// acpSessionEntry tracks a live ACP session for one goclaw conversation. +type acpSessionEntry struct { + id string // ACP session ID returned by session/new or session/load + proc *acp.ACPProcess // process that owns this session (for respawn detection) + lastUsed time.Time +} + // ACPProvider implements Provider by orchestrating ACP-compatible agent subprocesses. -// It delegates to a ProcessPool that manages agent lifecycle over JSON-RPC 2.0 stdio. +// One shared Gemini process is used; each goclaw conversation gets its own ACP session. type ACPProvider struct { - name string // provider name (default: "acp") + name string pool *acp.ProcessPool bridge *acp.ToolBridge defaultModel string - permMode string // permission mode for tool bridge - sessionMu sync.Map // sessionKey → *sync.Mutex + permMode string + poolKey string // key for the shared process in the pool (binary + args) + + acpSessions sync.Map // goclawSessionKey → *acpSessionEntry + sessionMu sync.Map // goclawSessionKey → *sync.Mutex (prevents concurrent session creation) + + done chan struct{} + closeOnce sync.Once } // ACPOption configures an ACPProvider. @@ -56,15 +69,22 @@ func WithACPPermMode(mode string) ACPOption { // NewACPProvider creates a provider that orchestrates ACP agents as subprocesses. func NewACPProvider(binary string, args []string, workDir string, idleTTL time.Duration, denyPatterns []*regexp.Regexp, opts ...ACPOption) *ACPProvider { + // Pool key identifies the shared process: binary + args combination + poolKey := binary + if len(args) > 0 { + poolKey += "|" + strings.Join(args, " ") + } + p := &ACPProvider{ name: "acp", defaultModel: "claude", + poolKey: poolKey, + done: make(chan struct{}), } for _, opt := range opts { opt(p) } - // Create tool bridge with workspace sandboxing, deny patterns, and permission mode var bridgeOpts []acp.ToolBridgeOption if len(denyPatterns) > 0 { bridgeOpts = append(bridgeOpts, acp.WithDenyPatterns(denyPatterns)) @@ -74,18 +94,82 @@ func NewACPProvider(binary string, args []string, workDir string, idleTTL time.D } p.bridge = acp.NewToolBridge(workDir, bridgeOpts...) - // Create process pool with tool bridge wired in p.pool = acp.NewProcessPool(binary, args, workDir, idleTTL) p.pool.SetToolHandler(p.bridge.Handle) + go p.sessionReaper() return p } +// sessionReaper removes ACP sessions idle for more than 30 minutes. +// Sends session/cancel to release resources on the agent side before purging locally. +func (p *ACPProvider) sessionReaper() { + const sessionIdleTTL = 30 * time.Minute + ticker := time.NewTicker(5 * time.Minute) + defer ticker.Stop() + for { + select { + case <-ticker.C: + p.acpSessions.Range(func(key, value any) bool { + entry := value.(*acpSessionEntry) + if time.Since(entry.lastUsed) > sessionIdleTTL { + slog.Info("acp: expiring idle session", "goclaw_session", key, "sid", entry.id) + if entry.proc != nil { + _ = entry.proc.Cancel(entry.id) + } + p.acpSessions.Delete(key) + } + return true + }) + case <-p.done: + return + } + } +} + +// resolveSession returns the ACP session ID for a goclaw session key. +// It creates a new session if none exists, or reloads it after a process respawn. +// A per-key mutex prevents concurrent creation races for the same session. +func (p *ACPProvider) resolveSession(ctx context.Context, proc *acp.ACPProcess, goclawKey string) (string, error) { + actual, _ := p.sessionMu.LoadOrStore(goclawKey, &sync.Mutex{}) + mu := actual.(*sync.Mutex) + mu.Lock() + defer mu.Unlock() + + if val, ok := p.acpSessions.Load(goclawKey); ok { + entry := val.(*acpSessionEntry) + if entry.proc == proc { + // Same process instance: session is still live, just update last-used + entry.lastUsed = time.Now() + return entry.id, nil + } + // Process was respawned — try to restore the session + slog.Info("acp: process respawned, attempting session restore", + "goclaw_session", goclawKey, "old_sid", entry.id) + if proc.AgentCaps().LoadSession { + sid, err := proc.LoadSession(ctx, entry.id) + if err == nil { + p.acpSessions.Store(goclawKey, &acpSessionEntry{id: sid, proc: proc, lastUsed: time.Now()}) + return sid, nil + } + slog.Warn("acp: session/load failed, creating new session", "old_sid", entry.id, "error", err) + } + // session/load not supported or failed — fall through to create new + } + + slog.Info("acp: creating new session", "goclaw_session", goclawKey, "pool_key", p.poolKey) + sid, err := proc.NewSession(ctx) + if err != nil { + return "", err + } + p.acpSessions.Store(goclawKey, &acpSessionEntry{id: sid, proc: proc, lastUsed: time.Now()}) + return sid, nil +} + func (p *ACPProvider) Name() string { return p.name } func (p *ACPProvider) DefaultModel() string { return p.defaultModel } // Capabilities implements CapabilitiesAware for pipeline code-path selection. -// ACP is subprocess-based (JSON-RPC 2.0 over stdio) — no HTTP adapter, capabilities only. func (p *ACPProvider) Capabilities() ProviderCapabilities { return ProviderCapabilities{ Streaming: true, @@ -103,37 +187,51 @@ func (p *ACPProvider) Capabilities() ProviderCapabilities { func (p *ACPProvider) Chat(ctx context.Context, req ChatRequest) (*ChatResponse, error) { sessionKey := extractStringOpt(req.Options, OptSessionKey) if sessionKey == "" { - return nil, fmt.Errorf("acp: session_key required in options") + sessionKey = fmt.Sprintf("temp-%d", time.Now().UnixNano()) } - unlock := p.lockSession(sessionKey) - defer unlock() - - proc, err := p.pool.GetOrSpawn(ctx, sessionKey) + proc, err := p.pool.GetOrSpawn(ctx, p.poolKey) if err != nil { return nil, fmt.Errorf("acp: spawn failed: %w", err) } + acpSessionID, err := p.resolveSession(ctx, proc, sessionKey) + if err != nil { + return nil, err + } + if strings.HasPrefix(sessionKey, "temp-") { + defer p.purgeSession(sessionKey) + } + content := extractACPContent(req) if len(content) == 0 { return nil, fmt.Errorf("acp: no user message in request") } - // Collect all text from session/update notifications + ctx = acp.WithGoclawSession(ctx, sessionKey) + var buf strings.Builder - promptResp, err := proc.Prompt(ctx, content, func(update acp.SessionUpdate) { + var updateCount int + promptResp, err := proc.Prompt(ctx, acpSessionID, content, func(update acp.SessionUpdate) { if update.Message != nil { for _, block := range update.Message.Content { if block.Type == "text" { buf.WriteString(block.Text) + updateCount++ } } } }) if err != nil { - return nil, fmt.Errorf("acp: prompt failed: %w", err) + slog.Error("acp: chat error", "session", sessionKey, "sid", acpSessionID, "error", err) + return &ChatResponse{ + Content: fmt.Sprintf("[ACP Error] %v", err), + FinishReason: "error", + }, err } + slog.Info("acp: chat completed", "session", sessionKey, "sid", acpSessionID, + "stopReason", mapStopReason(promptResp), "updates", updateCount, "contentLen", buf.Len()) return &ChatResponse{ Content: buf.String(), FinishReason: mapStopReason(promptResp), @@ -145,39 +243,52 @@ func (p *ACPProvider) Chat(ctx context.Context, req ChatRequest) (*ChatResponse, func (p *ACPProvider) ChatStream(ctx context.Context, req ChatRequest, onChunk func(StreamChunk)) (*ChatResponse, error) { sessionKey := extractStringOpt(req.Options, OptSessionKey) if sessionKey == "" { - return nil, fmt.Errorf("acp: session_key required in options") + sessionKey = fmt.Sprintf("temp-%d", time.Now().UnixNano()) } - unlock := p.lockSession(sessionKey) - defer unlock() - - proc, err := p.pool.GetOrSpawn(ctx, sessionKey) + proc, err := p.pool.GetOrSpawn(ctx, p.poolKey) if err != nil { return nil, fmt.Errorf("acp: spawn failed: %w", err) } + acpSessionID, err := p.resolveSession(ctx, proc, sessionKey) + if err != nil { + return nil, err + } + if strings.HasPrefix(sessionKey, "temp-") { + defer p.purgeSession(sessionKey) + } + content := extractACPContent(req) if len(content) == 0 { return nil, fmt.Errorf("acp: no user message in request") } - // Handle context cancellation → send session/cancel - cancelCtx, cancelFn := context.WithCancel(ctx) - defer cancelFn() + ctx = acp.WithGoclawSession(ctx, sessionKey) + + // done channel ensures the cancel goroutine exits cleanly on normal completion, + // preventing it from sending a spurious session/cancel after the prompt finishes. + done := make(chan struct{}) + defer close(done) go func() { - <-cancelCtx.Done() - if errors.Is(ctx.Err(), context.Canceled) { - _ = proc.Cancel() + select { + case <-ctx.Done(): + if errors.Is(ctx.Err(), context.Canceled) { + _ = proc.Cancel(acpSessionID) + } + case <-done: } }() var buf strings.Builder - promptResp, err := proc.Prompt(ctx, content, func(update acp.SessionUpdate) { + var updateCount int + promptResp, err := proc.Prompt(ctx, acpSessionID, content, func(update acp.SessionUpdate) { if update.Message != nil { for _, block := range update.Message.Content { if block.Type == "text" { onChunk(StreamChunk{Content: block.Text}) buf.WriteString(block.Text) + updateCount++ } } } @@ -186,10 +297,16 @@ func (p *ACPProvider) ChatStream(ctx context.Context, req ChatRequest, onChunk f } }) if err != nil { - return nil, fmt.Errorf("acp: prompt failed: %w", err) + slog.Error("acp: chat error", "session", sessionKey, "sid", acpSessionID, "error", err) + return &ChatResponse{ + Content: fmt.Sprintf("[ACP Error] %v", err), + FinishReason: "error", + }, err } onChunk(StreamChunk{Done: true}) + slog.Info("acp: chat stream completed", "session", sessionKey, "sid", acpSessionID, + "stopReason", mapStopReason(promptResp), "updates", updateCount, "contentLen", buf.Len()) return &ChatResponse{ Content: buf.String(), @@ -198,18 +315,28 @@ func (p *ACPProvider) ChatStream(ctx context.Context, req ChatRequest, onChunk f }, nil } -// Close shuts down all subprocesses and cleans up terminals. -func (p *ACPProvider) Close() error { - _ = p.bridge.Close() - return p.pool.Close() +// purgeSession removes a session entry from both tracking maps. +// Sends session/cancel to release resources on the agent side before purging locally. +// Used to immediately discard one-shot (temp-) sessions after completion. +func (p *ACPProvider) purgeSession(key string) { + if val, ok := p.acpSessions.Load(key); ok { + entry := val.(*acpSessionEntry) + if entry.proc != nil { + _ = entry.proc.Cancel(entry.id) + } + } + p.acpSessions.Delete(key) + p.sessionMu.Delete(key) + slog.Info("acp: purged temp session", "goclaw_session", key) } -// lockSession acquires a per-session mutex (same pattern as ClaudeCLIProvider). -func (p *ACPProvider) lockSession(sessionKey string) func() { - actual, _ := p.sessionMu.LoadOrStore(sessionKey, &sync.Mutex{}) - m := actual.(*sync.Mutex) - m.Lock() - return m.Unlock +// Close shuts down all subprocesses and cleans up terminals. +func (p *ACPProvider) Close() error { + p.closeOnce.Do(func() { + close(p.done) + }) + _ = p.bridge.Close() + return p.pool.Close() } // extractACPContent extracts user message + images from ChatRequest into ACP ContentBlocks. @@ -221,14 +348,13 @@ func extractACPContent(req ChatRequest) []acp.ContentBlock { var blocks []acp.ContentBlock - // Prepend system prompt to first user message (ACP agents don't have separate system prompt API) + // Prepend system prompt to user message (ACP agents have no separate system prompt API) text := userMsg if systemPrompt != "" { text = systemPrompt + "\n\n" + userMsg } blocks = append(blocks, acp.ContentBlock{Type: "text", Text: text}) - // Add images for _, img := range images { blocks = append(blocks, acp.ContentBlock{ Type: "image", @@ -246,8 +372,10 @@ func mapStopReason(resp *acp.PromptResponse) string { return "stop" } switch resp.StopReason { - case "maxContextLength": + case "max_tokens", "maxContextLength": return "length" + case "cancelled": + return "stop" default: return "stop" }