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>
This commit is contained in:
authored and GitHub committed 2026-04-18 12:17:23 +07:00
1 parent 5ee39c74af
commit 040b0f1944
14 files changed
+584 -196

No files matched your search

+3
View File
@@ -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
}
+1
View File
@@ -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)
}
+1 -1
View File
@@ -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)
+59
View File
@@ -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!")
}
}
+34 -1
View File
@@ -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
}
+2
View File
@@ -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
}
+102 -50
View File
@@ -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
}
+62 -23
View File
@@ -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,
})
}
+69 -37
View File
@@ -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")
}
}
+11
View File
@@ -0,0 +1,11 @@
//go:build linux
package acp
import "syscall"
func sysProcAttr() *syscall.SysProcAttr {
return &syscall.SysProcAttr{
Pdeathsig: syscall.SIGKILL,
}
}
+9
View File
@@ -0,0 +1,9 @@
//go:build !linux
package acp
import "syscall"
func sysProcAttr() *syscall.SysProcAttr {
return nil
}
+55 -36
View File
@@ -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"
}
+7 -7
View File
@@ -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)
}
}
+169 -41
View File
@@ -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"
}