feat: add GitHub Copilot CLI and Sourcegraph Amp agent support (#130)
This commit is contained in:
Carlos Florêncio
@@ -0,0 +1,798 @@
|
||||
package adapter
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/kandev/kandev/internal/agentctl/server/adapter/transport/shared"
|
||||
"github.com/kandev/kandev/internal/agentctl/types"
|
||||
"github.com/kandev/kandev/internal/agentctl/types/streams"
|
||||
"github.com/kandev/kandev/internal/common/logger"
|
||||
"github.com/kandev/kandev/pkg/amp"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
// AmpAdapter implements AgentAdapter for Sourcegraph Amp CLI.
|
||||
// It uses the stream-json protocol over stdin/stdout, similar to Claude Code.
|
||||
type AmpAdapter struct {
|
||||
cfg *Config
|
||||
logger *logger.Logger
|
||||
|
||||
// Normalizer for converting tool data to NormalizedPayload
|
||||
normalizer *AmpNormalizer
|
||||
|
||||
// Subprocess stdin/stdout (set via Connect)
|
||||
stdin io.Writer
|
||||
stdout io.Reader
|
||||
|
||||
// Amp client for protocol communication
|
||||
client *amp.Client
|
||||
|
||||
// Context for managing goroutine lifecycle
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
|
||||
// Session state
|
||||
sessionID string // Thread ID in Amp terminology
|
||||
operationID string
|
||||
sessionStatusSent bool
|
||||
|
||||
// Track pending tool calls and their normalized payloads
|
||||
pendingToolCalls map[string]bool
|
||||
pendingToolPayloads map[string]*streams.NormalizedPayload
|
||||
|
||||
// Accumulate text for the complete event
|
||||
// This is needed because the manager's buffer may be cleared by session code
|
||||
// before events are processed, causing a race condition
|
||||
textAccumulator strings.Builder
|
||||
|
||||
// Agent info
|
||||
agentInfo *AgentInfo
|
||||
|
||||
// Update channel
|
||||
updatesCh chan AgentEvent
|
||||
|
||||
// Permission handler
|
||||
permissionHandler PermissionHandler
|
||||
|
||||
// Result completion signaling
|
||||
resultCh chan resultComplete
|
||||
|
||||
// Context window tracking
|
||||
mainModelName string
|
||||
mainModelContextWindow int64
|
||||
contextTokensUsed int64
|
||||
|
||||
// Track if complete event was already sent for current operation
|
||||
// (to avoid duplicates from both stop_reason=end_turn and result message)
|
||||
completeSent bool
|
||||
|
||||
// Synchronization
|
||||
mu sync.RWMutex
|
||||
closed bool
|
||||
}
|
||||
|
||||
// defaultAmpContextWindow is the fallback context window size.
|
||||
const defaultAmpContextWindow = 200000
|
||||
|
||||
// NewAmpAdapter creates a new Amp protocol adapter.
|
||||
func NewAmpAdapter(cfg *Config, log *logger.Logger) *AmpAdapter {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
|
||||
return &AmpAdapter{
|
||||
cfg: cfg,
|
||||
logger: log.WithFields(zap.String("adapter", "amp")),
|
||||
normalizer: NewAmpNormalizer(),
|
||||
ctx: ctx,
|
||||
cancel: cancel,
|
||||
updatesCh: make(chan AgentEvent, 100),
|
||||
mainModelContextWindow: defaultAmpContextWindow,
|
||||
pendingToolCalls: make(map[string]bool),
|
||||
pendingToolPayloads: make(map[string]*streams.NormalizedPayload),
|
||||
}
|
||||
}
|
||||
|
||||
// PrepareEnvironment performs protocol-specific setup before the agent process starts.
|
||||
// Amp reads configuration from ~/.config/amp/settings.json.
|
||||
func (a *AmpAdapter) PrepareEnvironment() (map[string]string, error) {
|
||||
// Amp MCP configuration is handled via config file
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
// PrepareCommandArgs returns extra command-line arguments for the agent process.
|
||||
// For Amp, no extra args are needed - configuration is via config files.
|
||||
func (a *AmpAdapter) PrepareCommandArgs() []string {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Connect wires up the stdin/stdout pipes from the running agent subprocess.
|
||||
func (a *AmpAdapter) Connect(stdin io.Writer, stdout io.Reader) error {
|
||||
a.mu.Lock()
|
||||
defer a.mu.Unlock()
|
||||
|
||||
if a.stdin != nil || a.stdout != nil {
|
||||
return fmt.Errorf("adapter already connected")
|
||||
}
|
||||
|
||||
a.stdin = stdin
|
||||
a.stdout = stdout
|
||||
return nil
|
||||
}
|
||||
|
||||
// Initialize establishes the Amp connection with the agent subprocess.
|
||||
func (a *AmpAdapter) Initialize(ctx context.Context) error {
|
||||
a.logger.Info("initializing Amp adapter",
|
||||
zap.String("workdir", a.cfg.WorkDir))
|
||||
|
||||
// Create Amp client
|
||||
a.client = amp.NewClient(a.stdin, a.stdout, a.logger)
|
||||
a.client.SetMessageHandler(a.handleMessage)
|
||||
|
||||
// Start reading from stdout
|
||||
a.client.Start(a.ctx)
|
||||
|
||||
// Store agent info
|
||||
a.agentInfo = &AgentInfo{
|
||||
Name: "amp",
|
||||
Version: "unknown",
|
||||
}
|
||||
|
||||
a.logger.Info("Amp adapter initialized")
|
||||
return nil
|
||||
}
|
||||
|
||||
// GetAgentInfo returns information about the connected agent.
|
||||
func (a *AmpAdapter) GetAgentInfo() *AgentInfo {
|
||||
return a.agentInfo
|
||||
}
|
||||
|
||||
// NewSession creates a new Amp session (thread).
|
||||
// Note: Amp sessions are created implicitly with the first prompt.
|
||||
func (a *AmpAdapter) NewSession(ctx context.Context, _ []types.McpServer) (string, error) {
|
||||
a.mu.Lock()
|
||||
defer a.mu.Unlock()
|
||||
|
||||
// Generate a placeholder session ID - Amp will return its thread ID
|
||||
sessionID := uuid.New().String()
|
||||
a.sessionID = sessionID
|
||||
|
||||
a.logger.Info("created new session placeholder", zap.String("session_id", sessionID))
|
||||
|
||||
return sessionID, nil
|
||||
}
|
||||
|
||||
// LoadSession resumes an existing Amp session (thread).
|
||||
// The thread ID will be passed to Amp via the process manager which handles
|
||||
// the "threads fork" and "threads continue" commands.
|
||||
func (a *AmpAdapter) LoadSession(ctx context.Context, sessionID string) error {
|
||||
a.mu.Lock()
|
||||
a.sessionID = sessionID
|
||||
a.mu.Unlock()
|
||||
|
||||
// Set thread ID in client for tracking
|
||||
a.client.SetThreadID(sessionID)
|
||||
|
||||
a.logger.Info("loaded session", zap.String("session_id", sessionID))
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// Prompt sends a prompt to Amp and waits for completion.
|
||||
func (a *AmpAdapter) Prompt(ctx context.Context, message string) error {
|
||||
a.mu.Lock()
|
||||
sessionID := a.sessionID
|
||||
operationID := uuid.New().String()
|
||||
a.operationID = operationID
|
||||
a.resultCh = make(chan resultComplete, 1)
|
||||
a.completeSent = false // Reset completion flag for new operation
|
||||
a.mu.Unlock()
|
||||
|
||||
if a.client == nil {
|
||||
return fmt.Errorf("adapter not initialized")
|
||||
}
|
||||
|
||||
a.logger.Info("sending prompt",
|
||||
zap.String("session_id", sessionID),
|
||||
zap.String("operation_id", operationID))
|
||||
|
||||
// Send user message
|
||||
if err := a.client.SendUserMessage(message); err != nil {
|
||||
a.mu.Lock()
|
||||
a.resultCh = nil
|
||||
a.mu.Unlock()
|
||||
return fmt.Errorf("failed to send user message: %w", err)
|
||||
}
|
||||
|
||||
// Wait for result or context cancellation
|
||||
a.mu.RLock()
|
||||
resultCh := a.resultCh
|
||||
a.mu.RUnlock()
|
||||
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
a.mu.Lock()
|
||||
a.resultCh = nil
|
||||
a.mu.Unlock()
|
||||
return ctx.Err()
|
||||
case result := <-resultCh:
|
||||
a.mu.Lock()
|
||||
a.resultCh = nil
|
||||
a.mu.Unlock()
|
||||
if !result.success && result.err != "" {
|
||||
return fmt.Errorf("prompt failed: %s", result.err)
|
||||
}
|
||||
a.logger.Info("prompt completed",
|
||||
zap.String("operation_id", operationID),
|
||||
zap.Bool("success", result.success))
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// Cancel interrupts the current operation.
|
||||
func (a *AmpAdapter) Cancel(ctx context.Context) error {
|
||||
// Amp doesn't have a cancel command in stream-json mode
|
||||
// Cancellation is handled by killing the process
|
||||
a.logger.Info("cancel requested (not supported in stream-json mode)")
|
||||
return nil
|
||||
}
|
||||
|
||||
// Updates returns the channel for agent events.
|
||||
func (a *AmpAdapter) Updates() <-chan AgentEvent {
|
||||
return a.updatesCh
|
||||
}
|
||||
|
||||
// GetSessionID returns the current session ID (thread ID).
|
||||
func (a *AmpAdapter) GetSessionID() string {
|
||||
a.mu.RLock()
|
||||
defer a.mu.RUnlock()
|
||||
return a.sessionID
|
||||
}
|
||||
|
||||
// GetOperationID returns the current operation ID.
|
||||
func (a *AmpAdapter) GetOperationID() string {
|
||||
a.mu.RLock()
|
||||
defer a.mu.RUnlock()
|
||||
return a.operationID
|
||||
}
|
||||
|
||||
// SetPermissionHandler sets the handler for permission requests.
|
||||
func (a *AmpAdapter) SetPermissionHandler(handler PermissionHandler) {
|
||||
a.mu.Lock()
|
||||
defer a.mu.Unlock()
|
||||
a.permissionHandler = handler
|
||||
}
|
||||
|
||||
// Close releases resources held by the adapter.
|
||||
func (a *AmpAdapter) Close() error {
|
||||
a.mu.Lock()
|
||||
defer a.mu.Unlock()
|
||||
|
||||
if a.closed {
|
||||
return nil
|
||||
}
|
||||
a.closed = true
|
||||
|
||||
a.logger.Info("closing Amp adapter")
|
||||
|
||||
// Cancel the context
|
||||
if a.cancel != nil {
|
||||
a.cancel()
|
||||
}
|
||||
|
||||
// Stop the client
|
||||
if a.client != nil {
|
||||
a.client.Stop()
|
||||
}
|
||||
|
||||
// Close update channel
|
||||
close(a.updatesCh)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// sendUpdate safely sends an event to the updates channel.
|
||||
func (a *AmpAdapter) sendUpdate(update AgentEvent) {
|
||||
shared.LogNormalizedEvent(shared.ProtocolAmp, a.cfg.AgentID, &update)
|
||||
select {
|
||||
case a.updatesCh <- update:
|
||||
default:
|
||||
a.logger.Warn("updates channel full, dropping event")
|
||||
}
|
||||
}
|
||||
|
||||
// handleMessage processes streaming messages from Amp.
|
||||
func (a *AmpAdapter) handleMessage(msg *amp.Message) {
|
||||
// Log raw event for debugging
|
||||
if rawData, err := json.Marshal(msg); err == nil {
|
||||
shared.LogRawEvent(shared.ProtocolAmp, a.cfg.AgentID, msg.Type, rawData)
|
||||
}
|
||||
|
||||
a.mu.RLock()
|
||||
sessionID := a.sessionID
|
||||
operationID := a.operationID
|
||||
a.mu.RUnlock()
|
||||
|
||||
// Update session ID from thread ID if provided
|
||||
if msg.ThreadID != "" {
|
||||
a.mu.Lock()
|
||||
a.sessionID = msg.ThreadID
|
||||
sessionID = msg.ThreadID
|
||||
a.mu.Unlock()
|
||||
}
|
||||
|
||||
switch msg.Type {
|
||||
case amp.MessageTypeSystem:
|
||||
a.handleSystemMessage(msg, sessionID)
|
||||
|
||||
case amp.MessageTypeAssistant:
|
||||
a.handleAssistantMessage(msg, sessionID, operationID)
|
||||
|
||||
case amp.MessageTypeUser:
|
||||
// Amp sends tool_result blocks in user messages (unlike Claude Code which sends them in assistant messages)
|
||||
a.handleUserMessage(msg, sessionID, operationID)
|
||||
|
||||
case amp.MessageTypeResult:
|
||||
a.handleResultMessage(msg, sessionID, operationID)
|
||||
|
||||
default:
|
||||
a.logger.Debug("unhandled message type", zap.String("type", msg.Type))
|
||||
}
|
||||
}
|
||||
|
||||
// handleSystemMessage processes system init messages.
|
||||
func (a *AmpAdapter) handleSystemMessage(msg *amp.Message, sessionID string) {
|
||||
a.logger.Info("received system message",
|
||||
zap.String("thread_id", msg.ThreadID))
|
||||
|
||||
a.mu.Lock()
|
||||
alreadySent := a.sessionStatusSent
|
||||
a.sessionStatusSent = true
|
||||
a.mu.Unlock()
|
||||
|
||||
// Only send session status once
|
||||
if alreadySent {
|
||||
return
|
||||
}
|
||||
|
||||
status := "new"
|
||||
if msg.ThreadID != "" && msg.ThreadID == sessionID {
|
||||
status = "resumed"
|
||||
}
|
||||
|
||||
a.sendUpdate(AgentEvent{
|
||||
Type: EventTypeSessionStatus,
|
||||
SessionID: sessionID,
|
||||
SessionStatus: status,
|
||||
Data: map[string]any{
|
||||
"session_status": status,
|
||||
"init": true,
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
// handleUserMessage processes user messages.
|
||||
// In Amp's protocol, tool_result blocks come in user messages (not assistant messages like Claude Code).
|
||||
func (a *AmpAdapter) handleUserMessage(msg *amp.Message, sessionID, operationID string) {
|
||||
if msg.Message == nil {
|
||||
return
|
||||
}
|
||||
|
||||
// Process content blocks - only interested in tool_result
|
||||
for _, block := range msg.Message.Content {
|
||||
if block.Type == amp.ContentTypeToolResult {
|
||||
a.handleToolResult(&block, sessionID, operationID)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// handleAssistantMessage processes assistant messages.
|
||||
func (a *AmpAdapter) handleAssistantMessage(msg *amp.Message, sessionID, operationID string) {
|
||||
if msg.Message == nil {
|
||||
return
|
||||
}
|
||||
|
||||
// Log content block types for debugging
|
||||
blockTypes := make([]string, 0, len(msg.Message.Content))
|
||||
for _, block := range msg.Message.Content {
|
||||
blockTypes = append(blockTypes, block.Type)
|
||||
}
|
||||
a.logger.Debug("processing assistant message",
|
||||
zap.Int("num_blocks", len(msg.Message.Content)),
|
||||
zap.Strings("block_types", blockTypes))
|
||||
|
||||
// Update agent version and track main model
|
||||
if msg.Message.Model != "" && a.agentInfo != nil {
|
||||
a.agentInfo.Version = msg.Message.Model
|
||||
|
||||
a.mu.Lock()
|
||||
if a.mainModelName == "" {
|
||||
a.mainModelName = msg.Message.Model
|
||||
a.logger.Debug("tracking main model", zap.String("model", msg.Message.Model))
|
||||
}
|
||||
a.mu.Unlock()
|
||||
}
|
||||
|
||||
// Process content blocks
|
||||
for _, block := range msg.Message.Content {
|
||||
switch block.Type {
|
||||
case amp.ContentTypeText:
|
||||
if block.Text != "" {
|
||||
// Accumulate text locally to include in complete event
|
||||
// This bypasses the race condition with the manager's buffer
|
||||
a.mu.Lock()
|
||||
a.textAccumulator.WriteString(block.Text)
|
||||
a.mu.Unlock()
|
||||
|
||||
a.sendUpdate(AgentEvent{
|
||||
Type: EventTypeMessageChunk,
|
||||
SessionID: sessionID,
|
||||
OperationID: operationID,
|
||||
Text: block.Text,
|
||||
})
|
||||
}
|
||||
|
||||
case amp.ContentTypeThinking:
|
||||
if block.Thinking != "" {
|
||||
a.sendUpdate(AgentEvent{
|
||||
Type: EventTypeReasoning,
|
||||
SessionID: sessionID,
|
||||
OperationID: operationID,
|
||||
ReasoningText: block.Thinking,
|
||||
})
|
||||
}
|
||||
|
||||
case amp.ContentTypeToolUse:
|
||||
a.handleToolUse(&block, sessionID, operationID)
|
||||
|
||||
case amp.ContentTypeToolResult:
|
||||
a.handleToolResult(&block, sessionID, operationID)
|
||||
}
|
||||
}
|
||||
|
||||
// Calculate and emit token usage
|
||||
if msg.Message.Usage != nil {
|
||||
usage := msg.Message.Usage
|
||||
contextUsed := usage.InputTokens + usage.OutputTokens +
|
||||
usage.CacheCreationInputTokens + usage.CacheReadInputTokens
|
||||
|
||||
a.mu.Lock()
|
||||
a.contextTokensUsed = contextUsed
|
||||
contextSize := a.mainModelContextWindow
|
||||
a.mu.Unlock()
|
||||
|
||||
remaining := contextSize - contextUsed
|
||||
if remaining < 0 {
|
||||
remaining = 0
|
||||
}
|
||||
|
||||
a.sendUpdate(AgentEvent{
|
||||
Type: EventTypeContextWindow,
|
||||
SessionID: sessionID,
|
||||
OperationID: operationID,
|
||||
ContextWindowSize: contextSize,
|
||||
ContextWindowUsed: contextUsed,
|
||||
ContextWindowRemaining: remaining,
|
||||
ContextEfficiency: float64(contextUsed) / float64(contextSize) * 100,
|
||||
})
|
||||
}
|
||||
|
||||
// Check for turn completion - Amp signals this via stop_reason in assistant messages
|
||||
// (unlike Claude Code which sends a separate "result" message)
|
||||
if msg.Message.StopReason == "end_turn" {
|
||||
a.handleTurnComplete(sessionID, operationID)
|
||||
}
|
||||
}
|
||||
|
||||
// handleToolUse processes a tool_use content block.
|
||||
func (a *AmpAdapter) handleToolUse(block *amp.ContentBlock, sessionID, operationID string) {
|
||||
// Generate normalized payload using the normalizer
|
||||
normalizedPayload := a.normalizer.NormalizeToolCall(block.Name, block.Input)
|
||||
|
||||
// Build human-readable title
|
||||
toolTitle := block.Name
|
||||
// Amp uses "cmd" for Bash commands, Claude Code uses "command"
|
||||
if cmd, ok := block.Input["cmd"].(string); ok && block.Name == AmpToolBash {
|
||||
toolTitle = cmd
|
||||
} else if cmd, ok := block.Input["command"].(string); ok && block.Name == AmpToolBash {
|
||||
toolTitle = cmd
|
||||
} else if path, ok := block.Input["path"].(string); ok {
|
||||
// Amp uses "path" for Read tool
|
||||
toolTitle = path
|
||||
} else if path, ok := block.Input["file_path"].(string); ok {
|
||||
// Claude Code uses "file_path"
|
||||
toolTitle = path
|
||||
}
|
||||
|
||||
a.logger.Info("tool_use block received",
|
||||
zap.String("tool_call_id", block.ID),
|
||||
zap.String("tool_name", block.Name),
|
||||
zap.String("title", toolTitle))
|
||||
|
||||
// Track pending tool call and cache the normalized payload for result handling
|
||||
a.mu.Lock()
|
||||
a.pendingToolCalls[block.ID] = true
|
||||
a.pendingToolPayloads[block.ID] = normalizedPayload
|
||||
a.mu.Unlock()
|
||||
|
||||
a.sendUpdate(AgentEvent{
|
||||
Type: EventTypeToolCall,
|
||||
SessionID: sessionID,
|
||||
OperationID: operationID,
|
||||
ToolCallID: block.ID,
|
||||
ToolName: block.Name,
|
||||
ToolTitle: toolTitle,
|
||||
ToolStatus: "running",
|
||||
NormalizedPayload: normalizedPayload,
|
||||
})
|
||||
}
|
||||
|
||||
// handleToolResult processes a tool_result content block.
|
||||
func (a *AmpAdapter) handleToolResult(block *amp.ContentBlock, sessionID, operationID string) {
|
||||
status := "complete"
|
||||
if block.IsError {
|
||||
status = "error"
|
||||
}
|
||||
|
||||
a.logger.Info("tool_result block received",
|
||||
zap.String("tool_call_id", block.ToolUseID),
|
||||
zap.String("status", status))
|
||||
|
||||
// Get cached payload and remove from pending
|
||||
a.mu.Lock()
|
||||
delete(a.pendingToolCalls, block.ToolUseID)
|
||||
cachedPayload := a.pendingToolPayloads[block.ToolUseID]
|
||||
delete(a.pendingToolPayloads, block.ToolUseID)
|
||||
a.mu.Unlock()
|
||||
|
||||
// Normalize the tool result content
|
||||
normalizedPayload := a.normalizer.NormalizeToolResult(cachedPayload, block.Content, block.IsError)
|
||||
|
||||
a.sendUpdate(AgentEvent{
|
||||
Type: EventTypeToolUpdate,
|
||||
SessionID: sessionID,
|
||||
OperationID: operationID,
|
||||
ToolCallID: block.ToolUseID,
|
||||
ToolStatus: status,
|
||||
NormalizedPayload: normalizedPayload,
|
||||
})
|
||||
}
|
||||
|
||||
// handleTurnComplete processes turn completion when stop_reason is "end_turn".
|
||||
// This is Amp's way of signaling turn completion (instead of a separate "result" message).
|
||||
func (a *AmpAdapter) handleTurnComplete(sessionID, operationID string) {
|
||||
// Check if we already sent completion for this operation
|
||||
// (Amp may send both stop_reason=end_turn AND a result message)
|
||||
a.mu.Lock()
|
||||
if a.completeSent {
|
||||
a.mu.Unlock()
|
||||
a.logger.Debug("ignoring duplicate turn complete event",
|
||||
zap.String("session_id", sessionID),
|
||||
zap.String("operation_id", operationID))
|
||||
return
|
||||
}
|
||||
a.completeSent = true
|
||||
|
||||
a.logger.Info("turn complete (stop_reason=end_turn)")
|
||||
|
||||
// Auto-complete any pending tool calls and get accumulated text
|
||||
pendingTools := make(map[string]*streams.NormalizedPayload, len(a.pendingToolCalls))
|
||||
for toolID := range a.pendingToolCalls {
|
||||
pendingTools[toolID] = a.pendingToolPayloads[toolID]
|
||||
}
|
||||
a.pendingToolCalls = make(map[string]bool)
|
||||
a.pendingToolPayloads = make(map[string]*streams.NormalizedPayload)
|
||||
contextSize := a.mainModelContextWindow
|
||||
contextUsed := a.contextTokensUsed
|
||||
// Clear the text accumulator (text was already sent via message_chunk events)
|
||||
a.textAccumulator.Reset()
|
||||
a.mu.Unlock()
|
||||
|
||||
for toolID, cachedPayload := range pendingTools {
|
||||
a.logger.Info("auto-completing pending tool call on turn complete",
|
||||
zap.String("tool_call_id", toolID))
|
||||
a.sendUpdate(AgentEvent{
|
||||
Type: EventTypeToolUpdate,
|
||||
SessionID: sessionID,
|
||||
OperationID: operationID,
|
||||
ToolCallID: toolID,
|
||||
ToolStatus: "complete",
|
||||
NormalizedPayload: cachedPayload,
|
||||
})
|
||||
}
|
||||
|
||||
// Emit final context window event
|
||||
if contextUsed > 0 {
|
||||
remaining := contextSize - contextUsed
|
||||
if remaining < 0 {
|
||||
remaining = 0
|
||||
}
|
||||
a.sendUpdate(AgentEvent{
|
||||
Type: EventTypeContextWindow,
|
||||
SessionID: sessionID,
|
||||
OperationID: operationID,
|
||||
ContextWindowSize: contextSize,
|
||||
ContextWindowUsed: contextUsed,
|
||||
ContextWindowRemaining: remaining,
|
||||
ContextEfficiency: float64(contextUsed) / float64(contextSize) * 100,
|
||||
})
|
||||
}
|
||||
|
||||
// Send completion event WITHOUT text - text was already sent via message_chunk events
|
||||
// Including text here would cause duplicate messages
|
||||
a.sendUpdate(AgentEvent{
|
||||
Type: EventTypeComplete,
|
||||
SessionID: sessionID,
|
||||
OperationID: operationID,
|
||||
Data: map[string]any{
|
||||
"stop_reason": "end_turn",
|
||||
},
|
||||
})
|
||||
|
||||
// Signal completion
|
||||
a.mu.RLock()
|
||||
resultCh := a.resultCh
|
||||
a.mu.RUnlock()
|
||||
|
||||
if resultCh != nil {
|
||||
select {
|
||||
case resultCh <- resultComplete{success: true}:
|
||||
a.logger.Debug("signaled prompt completion")
|
||||
default:
|
||||
a.logger.Warn("result channel full, dropping signal")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// handleResultMessage processes result (completion) messages.
|
||||
func (a *AmpAdapter) handleResultMessage(msg *amp.Message, sessionID, operationID string) {
|
||||
a.logger.Info("received result message",
|
||||
zap.Bool("is_error", msg.IsError),
|
||||
zap.Int("num_turns", msg.NumTurns),
|
||||
zap.String("subtype", msg.Subtype),
|
||||
zap.String("error", msg.Error))
|
||||
|
||||
// Update session ID from result if provided
|
||||
if resultData := msg.GetResultData(); resultData != nil {
|
||||
if resultData.ThreadID != "" {
|
||||
a.mu.Lock()
|
||||
a.sessionID = resultData.ThreadID
|
||||
sessionID = resultData.ThreadID
|
||||
a.mu.Unlock()
|
||||
}
|
||||
}
|
||||
|
||||
// Check if complete was already sent (by handleTurnComplete on stop_reason=end_turn)
|
||||
a.mu.Lock()
|
||||
alreadyCompleted := a.completeSent
|
||||
if !alreadyCompleted {
|
||||
a.completeSent = true
|
||||
}
|
||||
|
||||
// Auto-complete any pending tool calls
|
||||
pendingTools := make(map[string]*streams.NormalizedPayload, len(a.pendingToolCalls))
|
||||
for toolID := range a.pendingToolCalls {
|
||||
pendingTools[toolID] = a.pendingToolPayloads[toolID]
|
||||
}
|
||||
a.pendingToolCalls = make(map[string]bool)
|
||||
a.pendingToolPayloads = make(map[string]*streams.NormalizedPayload)
|
||||
a.mu.Unlock()
|
||||
|
||||
for toolID, cachedPayload := range pendingTools {
|
||||
a.logger.Info("auto-completing pending tool call on result",
|
||||
zap.String("tool_call_id", toolID))
|
||||
a.sendUpdate(AgentEvent{
|
||||
Type: EventTypeToolUpdate,
|
||||
SessionID: sessionID,
|
||||
OperationID: operationID,
|
||||
ToolCallID: toolID,
|
||||
ToolStatus: "complete",
|
||||
NormalizedPayload: cachedPayload,
|
||||
})
|
||||
}
|
||||
|
||||
// Extract context window from model_usage if available
|
||||
a.mu.Lock()
|
||||
modelName := a.mainModelName
|
||||
if msg.ModelUsage != nil && modelName != "" {
|
||||
if modelStats, ok := msg.ModelUsage[modelName]; ok && modelStats.ContextWindow != nil {
|
||||
a.mainModelContextWindow = *modelStats.ContextWindow
|
||||
a.logger.Debug("updated context window from model_usage",
|
||||
zap.String("model", modelName),
|
||||
zap.Int64("context_window", a.mainModelContextWindow))
|
||||
}
|
||||
}
|
||||
contextSize := a.mainModelContextWindow
|
||||
contextUsed := a.contextTokensUsed
|
||||
a.mu.Unlock()
|
||||
|
||||
// Emit final context window event
|
||||
if contextUsed > 0 {
|
||||
remaining := contextSize - contextUsed
|
||||
if remaining < 0 {
|
||||
remaining = 0
|
||||
}
|
||||
a.sendUpdate(AgentEvent{
|
||||
Type: EventTypeContextWindow,
|
||||
SessionID: sessionID,
|
||||
OperationID: operationID,
|
||||
ContextWindowSize: contextSize,
|
||||
ContextWindowUsed: contextUsed,
|
||||
ContextWindowRemaining: remaining,
|
||||
ContextEfficiency: float64(contextUsed) / float64(contextSize) * 100,
|
||||
})
|
||||
}
|
||||
|
||||
// Only send completion event if not already sent by handleTurnComplete
|
||||
if !alreadyCompleted {
|
||||
// Clear the text accumulator (text was already sent via message_chunk events)
|
||||
a.mu.Lock()
|
||||
a.textAccumulator.Reset()
|
||||
a.mu.Unlock()
|
||||
|
||||
// Send complete event WITHOUT text to avoid duplicates
|
||||
a.sendUpdate(AgentEvent{
|
||||
Type: EventTypeComplete,
|
||||
SessionID: sessionID,
|
||||
OperationID: operationID,
|
||||
Data: map[string]any{
|
||||
"cost_usd": msg.CostUSD,
|
||||
"duration_ms": msg.DurationMS,
|
||||
"num_turns": msg.NumTurns,
|
||||
"input_tokens": msg.TotalInputTokens,
|
||||
"output_tokens": msg.TotalOutputTokens,
|
||||
"is_error": msg.IsError,
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
// Extract error message if failed
|
||||
var errorMsg string
|
||||
if msg.IsError {
|
||||
errorMsg = "prompt failed"
|
||||
// Try the error field first (Amp sends errors in this field)
|
||||
if msg.Error != "" {
|
||||
errorMsg = msg.Error
|
||||
} else if errStr := msg.GetResultString(); errStr != "" {
|
||||
errorMsg = errStr
|
||||
} else if resultData := msg.GetResultData(); resultData != nil && resultData.Text != "" {
|
||||
errorMsg = resultData.Text
|
||||
}
|
||||
}
|
||||
|
||||
// Signal completion
|
||||
a.mu.RLock()
|
||||
resultCh := a.resultCh
|
||||
a.mu.RUnlock()
|
||||
|
||||
if resultCh != nil {
|
||||
select {
|
||||
case resultCh <- resultComplete{success: !msg.IsError, err: errorMsg}:
|
||||
a.logger.Debug("signaled prompt completion")
|
||||
default:
|
||||
a.logger.Warn("result channel full, dropping signal")
|
||||
}
|
||||
}
|
||||
|
||||
// Send error event if failed
|
||||
if msg.IsError {
|
||||
a.sendUpdate(AgentEvent{
|
||||
Type: EventTypeError,
|
||||
SessionID: sessionID,
|
||||
OperationID: operationID,
|
||||
Error: errorMsg,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// RequiresProcessKill returns false because Amp agents exit when stdin is closed.
|
||||
func (a *AmpAdapter) RequiresProcessKill() bool {
|
||||
return false
|
||||
}
|
||||
|
||||
// Verify interface implementation
|
||||
var _ AgentAdapter = (*AmpAdapter)(nil)
|
||||
@@ -0,0 +1,391 @@
|
||||
package adapter
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"strings"
|
||||
|
||||
"github.com/kandev/kandev/internal/agentctl/server/adapter/transport/shared"
|
||||
"github.com/kandev/kandev/internal/agentctl/types/streams"
|
||||
)
|
||||
|
||||
// Amp tool name constants (matching Claude Code tool names).
|
||||
const (
|
||||
AmpToolBash = "Bash"
|
||||
AmpToolRead = "Read"
|
||||
AmpToolWrite = "Write"
|
||||
AmpToolEdit = "Edit"
|
||||
AmpToolGlob = "Glob"
|
||||
AmpToolGrep = "Grep"
|
||||
AmpToolWebFetch = "WebFetch"
|
||||
AmpToolWebSearch = "WebSearch"
|
||||
AmpToolTask = "Task"
|
||||
AmpToolTaskCreate = "TaskCreate"
|
||||
AmpToolTaskUpdate = "TaskUpdate"
|
||||
AmpToolTaskList = "TaskList"
|
||||
AmpToolNotebookEdit = "NotebookEdit"
|
||||
AmpToolTodoWrite = "TodoWrite"
|
||||
)
|
||||
|
||||
// AmpNormalizer converts Amp protocol tool data to NormalizedPayload.
|
||||
type AmpNormalizer struct{}
|
||||
|
||||
// NewAmpNormalizer creates a new Amp normalizer.
|
||||
func NewAmpNormalizer() *AmpNormalizer {
|
||||
return &AmpNormalizer{}
|
||||
}
|
||||
|
||||
// NormalizeToolCall converts Amp tool call data to NormalizedPayload.
|
||||
func (n *AmpNormalizer) NormalizeToolCall(toolName string, args map[string]any) *streams.NormalizedPayload {
|
||||
switch toolName {
|
||||
case AmpToolEdit, AmpToolWrite, AmpToolNotebookEdit:
|
||||
return n.normalizeEdit(toolName, args)
|
||||
case AmpToolRead:
|
||||
return n.normalizeRead(args)
|
||||
case AmpToolGlob, AmpToolGrep:
|
||||
return n.normalizeCodeSearch(toolName, args)
|
||||
case AmpToolBash:
|
||||
return n.normalizeExecute(args)
|
||||
case AmpToolWebFetch, AmpToolWebSearch:
|
||||
return n.normalizeHttpRequest(toolName, args)
|
||||
case AmpToolTask:
|
||||
return n.normalizeSubagentTask(args)
|
||||
case AmpToolTaskCreate:
|
||||
return n.normalizeCreateTask(args)
|
||||
case AmpToolTaskUpdate, AmpToolTaskList:
|
||||
return n.normalizeManageTodos(toolName, args)
|
||||
case AmpToolTodoWrite:
|
||||
return n.normalizeManageTodos(toolName, args)
|
||||
default:
|
||||
return n.normalizeGeneric(toolName, args)
|
||||
}
|
||||
}
|
||||
|
||||
// NormalizeToolResult updates the payload with tool result data from Amp.
|
||||
// Amp's tool results are JSON-wrapped strings like "{\"output\":\"...\",\"exitCode\":0}".
|
||||
func (n *AmpNormalizer) NormalizeToolResult(cachedPayload *streams.NormalizedPayload, content any, isError bool) *streams.NormalizedPayload {
|
||||
if cachedPayload == nil {
|
||||
// No cached payload, create a generic one with the result
|
||||
output, _ := n.parseToolResultContent(content)
|
||||
payload := streams.NewGeneric("unknown", nil)
|
||||
if payload.Generic() != nil {
|
||||
payload.Generic().Output = output
|
||||
}
|
||||
return payload
|
||||
}
|
||||
|
||||
// Update the cached payload with the result
|
||||
switch cachedPayload.Kind() {
|
||||
case streams.ToolKindShellExec:
|
||||
if cachedPayload.ShellExec() != nil {
|
||||
output, exitCode := n.parseToolResultContent(content)
|
||||
cachedPayload.ShellExec().Output = &streams.ShellExecOutput{
|
||||
Stdout: output,
|
||||
ExitCode: exitCode,
|
||||
}
|
||||
if isError {
|
||||
cachedPayload.ShellExec().Output.Stderr = output
|
||||
cachedPayload.ShellExec().Output.Stdout = ""
|
||||
}
|
||||
}
|
||||
case streams.ToolKindHttpRequest:
|
||||
if cachedPayload.HttpRequest() != nil {
|
||||
output, _ := n.parseToolResultContent(content)
|
||||
cachedPayload.HttpRequest().Response = output
|
||||
}
|
||||
case streams.ToolKindGeneric:
|
||||
if cachedPayload.Generic() != nil {
|
||||
output, _ := n.parseToolResultContent(content)
|
||||
cachedPayload.Generic().Output = output
|
||||
}
|
||||
case streams.ToolKindReadFile:
|
||||
// Parse Amp's read result format: {absolutePath, content, isDirectory, directoryEntries}
|
||||
if cachedPayload.ReadFile() != nil {
|
||||
result := n.parseAmpReadResult(content)
|
||||
if result != nil {
|
||||
// If it's a directory, convert to code_search
|
||||
if result.IsDirectory {
|
||||
codeSearch := streams.NewCodeSearch(
|
||||
"", // query
|
||||
"", // pattern
|
||||
result.AbsolutePath, // path
|
||||
"*", // glob - all entries
|
||||
)
|
||||
if codeSearch.CodeSearch() != nil {
|
||||
codeSearch.CodeSearch().Output = &streams.CodeSearchOutput{
|
||||
Files: result.DirectoryEntries,
|
||||
FileCount: len(result.DirectoryEntries),
|
||||
}
|
||||
}
|
||||
return codeSearch
|
||||
}
|
||||
// Regular file read
|
||||
output := &streams.ReadFileOutput{
|
||||
Content: result.Content,
|
||||
}
|
||||
if result.Content != "" {
|
||||
output.LineCount = strings.Count(result.Content, "\n") + 1
|
||||
}
|
||||
cachedPayload.ReadFile().Output = output
|
||||
}
|
||||
}
|
||||
case streams.ToolKindCodeSearch:
|
||||
if cachedPayload.CodeSearch() != nil {
|
||||
output, _ := n.parseToolResultContent(content)
|
||||
shared.NormalizeCodeSearchResult(cachedPayload.CodeSearch(), output)
|
||||
}
|
||||
}
|
||||
|
||||
return cachedPayload
|
||||
}
|
||||
|
||||
// ampReadResult holds parsed read result data from Amp.
|
||||
type ampReadResult struct {
|
||||
AbsolutePath string
|
||||
Content string
|
||||
IsDirectory bool
|
||||
DirectoryEntries []string
|
||||
}
|
||||
|
||||
// parseAmpReadResult parses Amp's read result format.
|
||||
func (n *AmpNormalizer) parseAmpReadResult(content any) *ampReadResult {
|
||||
switch c := content.(type) {
|
||||
case string:
|
||||
var result ampReadResult
|
||||
if err := json.Unmarshal([]byte(c), &result); err == nil {
|
||||
return &result
|
||||
}
|
||||
// Not JSON, treat as raw file content
|
||||
return &ReadResult{Content: c}
|
||||
|
||||
case map[string]any:
|
||||
result := &ReadResult{}
|
||||
if v, ok := c["absolutePath"].(string); ok {
|
||||
result.AbsolutePath = v
|
||||
}
|
||||
if v, ok := c["content"].(string); ok {
|
||||
result.Content = v
|
||||
}
|
||||
if v, ok := c["isDirectory"].(bool); ok {
|
||||
result.IsDirectory = v
|
||||
}
|
||||
if v, ok := c["directoryEntries"].([]any); ok {
|
||||
for _, entry := range v {
|
||||
if s, ok := entry.(string); ok {
|
||||
result.DirectoryEntries = append(result.DirectoryEntries, s)
|
||||
}
|
||||
}
|
||||
}
|
||||
return result
|
||||
|
||||
default:
|
||||
return &ReadResult{}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
// parseToolResultContent extracts the output string and exit code from Amp's JSON-wrapped content.
|
||||
// Amp sends tool results in formats like:
|
||||
// - String: "{\"output\":\"actual content\",\"exitCode\":0}"
|
||||
// - Direct string (non-JSON)
|
||||
// - Map with output/exitCode keys
|
||||
// - Any other JSON structure (returned as-is)
|
||||
func (n *AmpNormalizer) parseToolResultContent(content any) (string, int) {
|
||||
switch c := content.(type) {
|
||||
case string:
|
||||
// Try to parse as JSON-wrapped content with {output, exitCode} structure
|
||||
var wrapped struct {
|
||||
Output string `json:"output"`
|
||||
ExitCode int `json:"exitCode"`
|
||||
}
|
||||
if err := json.Unmarshal([]byte(c), &wrapped); err == nil && wrapped.Output != "" {
|
||||
// Successfully parsed AND has output field
|
||||
return wrapped.Output, wrapped.ExitCode
|
||||
}
|
||||
// Either not JSON, or JSON without "output" field - use as-is
|
||||
return c, 0
|
||||
|
||||
case map[string]any:
|
||||
// Handle map format (already parsed JSON)
|
||||
output := shared.GetString(c, "output")
|
||||
exitCode := shared.GetInt(c, "exitCode")
|
||||
if output == "" {
|
||||
// Try alternative field names
|
||||
output = shared.GetString(c, "stdout")
|
||||
if stderr := shared.GetString(c, "stderr"); stderr != "" && output == "" {
|
||||
output = stderr
|
||||
}
|
||||
if ec := shared.GetInt(c, "exit_code"); ec != 0 {
|
||||
exitCode = ec
|
||||
}
|
||||
}
|
||||
return output, exitCode
|
||||
|
||||
case []any:
|
||||
// Handle array of content blocks (some tools return multiple blocks)
|
||||
var sb strings.Builder
|
||||
for _, item := range c {
|
||||
if str, ok := item.(string); ok {
|
||||
sb.WriteString(str)
|
||||
} else if m, ok := item.(map[string]any); ok {
|
||||
if text := shared.GetString(m, "text"); text != "" {
|
||||
sb.WriteString(text)
|
||||
}
|
||||
}
|
||||
}
|
||||
return sb.String(), 0
|
||||
|
||||
default:
|
||||
// Unknown format, try JSON marshaling for debugging
|
||||
if data, err := json.Marshal(content); err == nil {
|
||||
return string(data), 0
|
||||
}
|
||||
return "", 0
|
||||
}
|
||||
}
|
||||
|
||||
// normalizeEdit converts Amp Edit/Write tool data.
|
||||
func (n *AmpNormalizer) normalizeEdit(toolName string, args map[string]any) *streams.NormalizedPayload {
|
||||
filePath := shared.GetString(args, "file_path")
|
||||
oldString := shared.GetString(args, "old_string")
|
||||
newString := shared.GetString(args, "new_string")
|
||||
content := shared.GetString(args, "content")
|
||||
|
||||
var mutations []streams.FileMutation
|
||||
|
||||
if toolName == AmpToolWrite {
|
||||
// Write is a full file creation/replacement
|
||||
mutations = append(mutations, streams.FileMutation{
|
||||
Type: streams.MutationCreate,
|
||||
Content: content,
|
||||
})
|
||||
} else {
|
||||
// Edit is a patch operation
|
||||
mutation := streams.FileMutation{
|
||||
Type: streams.MutationPatch,
|
||||
OldContent: oldString,
|
||||
NewContent: newString,
|
||||
}
|
||||
|
||||
// Generate unified diff
|
||||
if oldString != "" && newString != "" {
|
||||
mutation.Diff = shared.GenerateUnifiedDiff(oldString, newString, filePath, 0)
|
||||
}
|
||||
|
||||
mutations = append(mutations, mutation)
|
||||
}
|
||||
|
||||
return streams.NewModifyFile(filePath, mutations)
|
||||
}
|
||||
|
||||
// normalizeRead converts Amp Read tool data.
|
||||
// Note: Amp uses "path" for the file path, while Claude Code uses "file_path".
|
||||
func (n *AmpNormalizer) normalizeRead(args map[string]any) *streams.NormalizedPayload {
|
||||
// Amp uses "path", Claude Code uses "file_path" - check both
|
||||
filePath := shared.GetString(args, "path")
|
||||
if filePath == "" {
|
||||
filePath = shared.GetString(args, "file_path")
|
||||
}
|
||||
offset := shared.GetInt(args, "offset")
|
||||
limit := shared.GetInt(args, "limit")
|
||||
|
||||
return streams.NewReadFile(filePath, offset, limit)
|
||||
}
|
||||
|
||||
// normalizeCodeSearch converts Amp Glob/Grep tool data.
|
||||
func (n *AmpNormalizer) normalizeCodeSearch(toolName string, args map[string]any) *streams.NormalizedPayload {
|
||||
path := shared.GetString(args, "path")
|
||||
pattern := shared.GetString(args, "pattern")
|
||||
|
||||
var query, glob string
|
||||
switch toolName {
|
||||
case AmpToolGlob:
|
||||
glob = pattern
|
||||
case AmpToolGrep:
|
||||
query = shared.GetString(args, "query")
|
||||
if query == "" {
|
||||
query = pattern
|
||||
}
|
||||
}
|
||||
|
||||
return streams.NewCodeSearch(query, pattern, path, glob)
|
||||
}
|
||||
|
||||
// normalizeExecute converts Amp Bash tool data.
|
||||
// Note: Amp uses "cmd" for the command field, while Claude Code uses "command".
|
||||
func (n *AmpNormalizer) normalizeExecute(args map[string]any) *streams.NormalizedPayload {
|
||||
// Amp uses "cmd", Claude Code uses "command" - check both
|
||||
command := shared.GetString(args, "cmd")
|
||||
if command == "" {
|
||||
command = shared.GetString(args, "command")
|
||||
}
|
||||
description := shared.GetString(args, "description")
|
||||
timeout := shared.GetInt(args, "timeout")
|
||||
background := shared.GetBool(args, "run_in_background")
|
||||
|
||||
return streams.NewShellExec(command, "", description, timeout, background)
|
||||
}
|
||||
|
||||
// normalizeHttpRequest converts Amp WebFetch/WebSearch tool data.
|
||||
func (n *AmpNormalizer) normalizeHttpRequest(toolName string, args map[string]any) *streams.NormalizedPayload {
|
||||
url := shared.GetString(args, "url")
|
||||
if url == "" {
|
||||
url = shared.GetString(args, "query")
|
||||
}
|
||||
|
||||
method := "GET"
|
||||
if toolName == AmpToolWebSearch {
|
||||
method = "SEARCH"
|
||||
}
|
||||
|
||||
return streams.NewHttpRequest(url, method)
|
||||
}
|
||||
|
||||
// normalizeSubagentTask converts Amp Task tool data (subagent invocation).
|
||||
func (n *AmpNormalizer) normalizeSubagentTask(args map[string]any) *streams.NormalizedPayload {
|
||||
description := shared.GetString(args, "description")
|
||||
prompt := shared.GetString(args, "prompt")
|
||||
subagentType := shared.GetString(args, "subagent_type")
|
||||
|
||||
return streams.NewSubagentTask(description, prompt, subagentType)
|
||||
}
|
||||
|
||||
// normalizeCreateTask converts Amp TaskCreate tool data.
|
||||
func (n *AmpNormalizer) normalizeCreateTask(args map[string]any) *streams.NormalizedPayload {
|
||||
title := shared.GetString(args, "subject")
|
||||
description := shared.GetString(args, "description")
|
||||
|
||||
return streams.NewCreateTask(title, description)
|
||||
}
|
||||
|
||||
// normalizeManageTodos converts Amp TaskUpdate/TaskList/TodoWrite tool data.
|
||||
func (n *AmpNormalizer) normalizeManageTodos(toolName string, args map[string]any) *streams.NormalizedPayload {
|
||||
operation := "update"
|
||||
switch toolName {
|
||||
case AmpToolTaskList:
|
||||
operation = "list"
|
||||
case AmpToolTodoWrite:
|
||||
operation = "write"
|
||||
}
|
||||
|
||||
var items []streams.TodoItem
|
||||
|
||||
// Extract items if present
|
||||
if rawItems, ok := args["items"].([]any); ok {
|
||||
for _, item := range rawItems {
|
||||
if itemMap, ok := item.(map[string]any); ok {
|
||||
items = append(items, streams.TodoItem{
|
||||
ID: shared.GetString(itemMap, "id"),
|
||||
Description: shared.GetString(itemMap, "description"),
|
||||
Status: shared.GetString(itemMap, "status"),
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return streams.NewManageTodos(operation, items)
|
||||
}
|
||||
|
||||
// normalizeGeneric wraps unknown tools as generic.
|
||||
func (n *AmpNormalizer) normalizeGeneric(toolName string, args map[string]any) *streams.NormalizedPayload {
|
||||
return streams.NewGeneric(toolName, args)
|
||||
}
|
||||
@@ -0,0 +1,805 @@
|
||||
package adapter
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"regexp"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/kandev/kandev/internal/agentctl/server/adapter/transport/shared"
|
||||
"github.com/kandev/kandev/internal/agentctl/types"
|
||||
"github.com/kandev/kandev/internal/agentctl/types/streams"
|
||||
"github.com/kandev/kandev/internal/common/logger"
|
||||
"github.com/kandev/kandev/pkg/copilot"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
// CopilotAdapter implements AgentAdapter for GitHub Copilot using the official SDK.
|
||||
//
|
||||
// The process manager spawns the Copilot CLI in server mode (--server), which
|
||||
// listens on a random TCP port and prints "listening on port <port>" to stdout.
|
||||
// Connect() captures the stdout pipe, and Initialize() scans it for the port,
|
||||
// then creates an SDK client that connects to the external server via CLIUrl.
|
||||
type CopilotAdapter struct {
|
||||
cfg *Config
|
||||
logger *logger.Logger
|
||||
|
||||
// Normalizer for converting tool data to NormalizedPayload
|
||||
normalizer *CopilotNormalizer
|
||||
|
||||
// stdout from the process manager (used to detect the listening port)
|
||||
stdout io.Reader
|
||||
|
||||
// Copilot SDK client (connects to externally managed CLI server via TCP)
|
||||
client *copilot.Client
|
||||
|
||||
// Context for managing goroutine lifecycle
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
|
||||
// Session state
|
||||
sessionID string
|
||||
operationID string
|
||||
|
||||
// Track pending tool calls and their normalized payloads
|
||||
pendingToolPayloads map[string]*streams.NormalizedPayload
|
||||
|
||||
// Accumulate text for the complete event
|
||||
textAccumulator strings.Builder
|
||||
|
||||
// Agent info
|
||||
agentInfo *AgentInfo
|
||||
|
||||
// Update channel
|
||||
updatesCh chan AgentEvent
|
||||
|
||||
// Permission handler
|
||||
permissionHandler PermissionHandler
|
||||
|
||||
// Result completion signaling
|
||||
resultCh chan resultComplete
|
||||
|
||||
// Context window tracking (from usage events)
|
||||
contextWindowSize int64
|
||||
contextTokensUsed int64
|
||||
|
||||
// Track if completion was already sent for current operation
|
||||
completeSent bool
|
||||
|
||||
// Track if we received streaming deltas for this operation
|
||||
// (to avoid double-accumulating from both delta and full message events)
|
||||
receivedDeltas bool
|
||||
|
||||
// Synchronization
|
||||
mu sync.RWMutex
|
||||
closed bool
|
||||
}
|
||||
|
||||
// resultComplete holds the result of a completed prompt
|
||||
type resultComplete struct {
|
||||
success bool
|
||||
err string
|
||||
}
|
||||
|
||||
// defaultCopilotContextWindow is the fallback context window size.
|
||||
const defaultCopilotContextWindow = 128000
|
||||
|
||||
// NewCopilotAdapter creates a new Copilot protocol adapter.
|
||||
func NewCopilotAdapter(cfg *Config, log *logger.Logger) *CopilotAdapter {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
|
||||
return &CopilotAdapter{
|
||||
cfg: cfg,
|
||||
logger: log.WithFields(zap.String("adapter", "copilot")),
|
||||
normalizer: NewCopilotNormalizer(),
|
||||
ctx: ctx,
|
||||
cancel: cancel,
|
||||
updatesCh: make(chan AgentEvent, 100),
|
||||
contextWindowSize: defaultCopilotContextWindow,
|
||||
pendingToolPayloads: make(map[string]*streams.NormalizedPayload),
|
||||
}
|
||||
}
|
||||
|
||||
// PrepareEnvironment performs protocol-specific setup before the agent process starts.
|
||||
// No special environment is needed; the process manager spawns the CLI in server mode.
|
||||
func (a *CopilotAdapter) PrepareEnvironment() (map[string]string, error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
// PrepareCommandArgs returns extra command-line arguments for the agent process.
|
||||
// For Copilot, no extra args are needed - the CLI is spawned in server mode.
|
||||
func (a *CopilotAdapter) PrepareCommandArgs() []string {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Connect stores the stdout pipe from the process manager.
|
||||
// The adapter reads stdout during Initialize to detect the TCP port
|
||||
// the Copilot CLI server is listening on.
|
||||
func (a *CopilotAdapter) Connect(stdin io.Writer, stdout io.Reader) error {
|
||||
a.mu.Lock()
|
||||
defer a.mu.Unlock()
|
||||
|
||||
if a.stdout != nil {
|
||||
return fmt.Errorf("adapter already connected")
|
||||
}
|
||||
|
||||
a.stdout = stdout
|
||||
a.logger.Info("Connect: stored stdout pipe for port detection")
|
||||
return nil
|
||||
}
|
||||
|
||||
// Initialize waits for the Copilot CLI server to print its listening port,
|
||||
// then creates an SDK client that connects to it via TCP.
|
||||
func (a *CopilotAdapter) Initialize(ctx context.Context) error {
|
||||
a.logger.Info("initializing Copilot adapter (server mode)",
|
||||
zap.String("workdir", a.cfg.WorkDir))
|
||||
|
||||
if a.stdout == nil {
|
||||
return fmt.Errorf("stdout not connected; call Connect() before Initialize()")
|
||||
}
|
||||
|
||||
// Scan stdout for "listening on port <port>" from the CLI server
|
||||
port, scanner, err := a.waitForPort(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to detect Copilot CLI server port: %w", err)
|
||||
}
|
||||
|
||||
cliURL := fmt.Sprintf("localhost:%d", port)
|
||||
a.logger.Info("detected Copilot CLI server",
|
||||
zap.Int("port", port),
|
||||
zap.String("cli_url", cliURL))
|
||||
|
||||
// Create SDK client pointing at the external server
|
||||
a.client = copilot.NewClient(copilot.ClientConfig{
|
||||
CLIUrl: cliURL,
|
||||
}, a.logger)
|
||||
|
||||
// Set event handler before starting
|
||||
a.client.SetEventHandler(a.handleEvent)
|
||||
|
||||
// Start the SDK client (connects to external server via TCP)
|
||||
if err := a.client.Start(ctx); err != nil {
|
||||
return fmt.Errorf("failed to start Copilot SDK client: %w", err)
|
||||
}
|
||||
|
||||
// Drain remaining stdout in background so the process doesn't block
|
||||
go func() {
|
||||
for scanner.Scan() {
|
||||
// discard
|
||||
}
|
||||
}()
|
||||
|
||||
// Store agent info
|
||||
a.agentInfo = &AgentInfo{
|
||||
Name: "copilot",
|
||||
Version: "sdk",
|
||||
}
|
||||
|
||||
a.logger.Info("Copilot adapter initialized (server mode)")
|
||||
return nil
|
||||
}
|
||||
|
||||
// portPattern matches "listening on port <number>" printed by the Copilot CLI in server mode.
|
||||
var portPattern = regexp.MustCompile(`listening on port (\d+)`)
|
||||
|
||||
// waitForPort scans stdout line-by-line until it finds the listening port.
|
||||
// Returns the detected port and the scanner (for background draining).
|
||||
func (a *CopilotAdapter) waitForPort(ctx context.Context) (int, *bufio.Scanner, error) {
|
||||
const timeout = 180 * time.Second
|
||||
|
||||
scanner := bufio.NewScanner(a.stdout)
|
||||
portCh := make(chan int, 1)
|
||||
errCh := make(chan error, 1)
|
||||
|
||||
go func() {
|
||||
var capturedLines []string
|
||||
for scanner.Scan() {
|
||||
line := scanner.Text()
|
||||
a.logger.Debug("CLI stdout", zap.String("line", line))
|
||||
|
||||
if len(capturedLines) < 64 {
|
||||
capturedLines = append(capturedLines, line)
|
||||
}
|
||||
|
||||
if m := portPattern.FindStringSubmatch(line); m != nil {
|
||||
port, err := strconv.Atoi(m[1])
|
||||
if err != nil {
|
||||
errCh <- fmt.Errorf("invalid port number %q: %w", m[1], err)
|
||||
return
|
||||
}
|
||||
portCh <- port
|
||||
return
|
||||
}
|
||||
}
|
||||
// EOF before finding port
|
||||
if err := scanner.Err(); err != nil {
|
||||
errCh <- fmt.Errorf("error reading stdout: %w", err)
|
||||
} else {
|
||||
var detail strings.Builder
|
||||
if len(capturedLines) > 0 {
|
||||
start := 0
|
||||
if len(capturedLines) > 12 {
|
||||
start = len(capturedLines) - 12
|
||||
}
|
||||
for _, l := range capturedLines[start:] {
|
||||
detail.WriteString("\n ")
|
||||
detail.WriteString(l)
|
||||
}
|
||||
}
|
||||
errCh <- fmt.Errorf("CLI exited before printing listening port%s", detail.String())
|
||||
}
|
||||
}()
|
||||
|
||||
timer := time.NewTimer(timeout)
|
||||
defer timer.Stop()
|
||||
|
||||
select {
|
||||
case port := <-portCh:
|
||||
return port, scanner, nil
|
||||
case err := <-errCh:
|
||||
return 0, nil, err
|
||||
case <-timer.C:
|
||||
return 0, nil, fmt.Errorf("timeout (%s) waiting for Copilot CLI to print listening port", timeout)
|
||||
case <-ctx.Done():
|
||||
return 0, nil, ctx.Err()
|
||||
}
|
||||
}
|
||||
|
||||
// GetAgentInfo returns information about the connected agent.
|
||||
func (a *CopilotAdapter) GetAgentInfo() *AgentInfo {
|
||||
return a.agentInfo
|
||||
}
|
||||
|
||||
// NewSession creates a new Copilot session via the SDK.
|
||||
func (a *CopilotAdapter) NewSession(ctx context.Context, _ []types.McpServer) (string, error) {
|
||||
if a.client == nil {
|
||||
return "", fmt.Errorf("adapter not initialized")
|
||||
}
|
||||
|
||||
sessionID, err := a.client.CreateSession(ctx)
|
||||
if err != nil {
|
||||
// If session creation fails, generate a placeholder ID
|
||||
sessionID = uuid.New().String()
|
||||
a.logger.Warn("failed to create SDK session, using placeholder",
|
||||
zap.Error(err),
|
||||
zap.String("session_id", sessionID))
|
||||
}
|
||||
|
||||
a.mu.Lock()
|
||||
a.sessionID = sessionID
|
||||
a.mu.Unlock()
|
||||
|
||||
a.logger.Info("created new session", zap.String("session_id", sessionID))
|
||||
|
||||
// Send session status event
|
||||
a.sendUpdate(AgentEvent{
|
||||
Type: EventTypeSessionStatus,
|
||||
SessionID: sessionID,
|
||||
SessionStatus: "new",
|
||||
Data: map[string]any{
|
||||
"session_status": "new",
|
||||
"init": true,
|
||||
},
|
||||
})
|
||||
|
||||
return sessionID, nil
|
||||
}
|
||||
|
||||
// LoadSession resumes an existing Copilot session via the SDK.
|
||||
func (a *CopilotAdapter) LoadSession(ctx context.Context, sessionID string) error {
|
||||
if a.client == nil {
|
||||
return fmt.Errorf("adapter not initialized")
|
||||
}
|
||||
|
||||
if err := a.client.ResumeSession(ctx, sessionID); err != nil {
|
||||
a.logger.Warn("failed to resume SDK session, continuing anyway",
|
||||
zap.Error(err),
|
||||
zap.String("session_id", sessionID))
|
||||
}
|
||||
|
||||
a.mu.Lock()
|
||||
a.sessionID = sessionID
|
||||
a.mu.Unlock()
|
||||
|
||||
a.logger.Info("loaded session", zap.String("session_id", sessionID))
|
||||
|
||||
// Send session status event
|
||||
a.sendUpdate(AgentEvent{
|
||||
Type: EventTypeSessionStatus,
|
||||
SessionID: sessionID,
|
||||
SessionStatus: "resumed",
|
||||
Data: map[string]any{
|
||||
"session_status": "resumed",
|
||||
"init": true,
|
||||
},
|
||||
})
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// Prompt sends a prompt to Copilot and waits for completion.
|
||||
func (a *CopilotAdapter) Prompt(ctx context.Context, message string) error {
|
||||
if a.client == nil {
|
||||
return fmt.Errorf("adapter not initialized")
|
||||
}
|
||||
|
||||
a.mu.Lock()
|
||||
sessionID := a.sessionID
|
||||
operationID := uuid.New().String()
|
||||
a.operationID = operationID
|
||||
a.resultCh = make(chan resultComplete, 1)
|
||||
a.completeSent = false // Reset completion flag for new operation
|
||||
a.receivedDeltas = false // Reset delta tracking for new operation
|
||||
a.mu.Unlock()
|
||||
|
||||
a.logger.Info("sending prompt via SDK",
|
||||
zap.String("session_id", sessionID),
|
||||
zap.String("operation_id", operationID))
|
||||
|
||||
// Send message (non-blocking, events come via handler)
|
||||
if _, err := a.client.Send(ctx, message); err != nil {
|
||||
a.mu.Lock()
|
||||
a.resultCh = nil
|
||||
a.mu.Unlock()
|
||||
return fmt.Errorf("failed to send prompt: %w", err)
|
||||
}
|
||||
|
||||
// Wait for idle (completion) or context cancellation
|
||||
a.mu.RLock()
|
||||
resultCh := a.resultCh
|
||||
a.mu.RUnlock()
|
||||
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
a.mu.Lock()
|
||||
a.resultCh = nil
|
||||
a.mu.Unlock()
|
||||
return ctx.Err()
|
||||
case result := <-resultCh:
|
||||
a.mu.Lock()
|
||||
a.resultCh = nil
|
||||
a.mu.Unlock()
|
||||
if !result.success && result.err != "" {
|
||||
return fmt.Errorf("prompt failed: %s", result.err)
|
||||
}
|
||||
a.logger.Info("prompt completed",
|
||||
zap.String("operation_id", operationID),
|
||||
zap.Bool("success", result.success))
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// Cancel interrupts the current operation.
|
||||
func (a *CopilotAdapter) Cancel(ctx context.Context) error {
|
||||
if a.client == nil {
|
||||
return fmt.Errorf("adapter not initialized")
|
||||
}
|
||||
|
||||
a.logger.Info("cancelling operation via SDK")
|
||||
return a.client.Abort(ctx)
|
||||
}
|
||||
|
||||
// Updates returns the channel for agent events.
|
||||
func (a *CopilotAdapter) Updates() <-chan AgentEvent {
|
||||
return a.updatesCh
|
||||
}
|
||||
|
||||
// GetSessionID returns the current session ID.
|
||||
func (a *CopilotAdapter) GetSessionID() string {
|
||||
a.mu.RLock()
|
||||
defer a.mu.RUnlock()
|
||||
return a.sessionID
|
||||
}
|
||||
|
||||
// GetOperationID returns the current operation ID.
|
||||
func (a *CopilotAdapter) GetOperationID() string {
|
||||
a.mu.RLock()
|
||||
defer a.mu.RUnlock()
|
||||
return a.operationID
|
||||
}
|
||||
|
||||
// SetPermissionHandler sets the handler for permission requests.
|
||||
func (a *CopilotAdapter) SetPermissionHandler(handler PermissionHandler) {
|
||||
a.mu.Lock()
|
||||
defer a.mu.Unlock()
|
||||
a.permissionHandler = handler
|
||||
}
|
||||
|
||||
// Close releases resources held by the adapter.
|
||||
func (a *CopilotAdapter) Close() error {
|
||||
a.mu.Lock()
|
||||
defer a.mu.Unlock()
|
||||
|
||||
if a.closed {
|
||||
return nil
|
||||
}
|
||||
a.closed = true
|
||||
|
||||
a.logger.Info("closing Copilot SDK adapter")
|
||||
|
||||
// Cancel the context
|
||||
if a.cancel != nil {
|
||||
a.cancel()
|
||||
}
|
||||
|
||||
// Stop the SDK client (disconnects from the CLI server)
|
||||
if a.client != nil {
|
||||
if err := a.client.Stop(); err != nil {
|
||||
a.logger.Warn("error stopping SDK client", zap.Error(err))
|
||||
}
|
||||
a.client = nil
|
||||
}
|
||||
|
||||
// Close update channel
|
||||
close(a.updatesCh)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// sendUpdate safely sends an event to the updates channel.
|
||||
func (a *CopilotAdapter) sendUpdate(update AgentEvent) {
|
||||
shared.LogNormalizedEvent(shared.ProtocolCopilot, a.cfg.AgentID, &update)
|
||||
select {
|
||||
case a.updatesCh <- update:
|
||||
default:
|
||||
a.logger.Warn("updates channel full, dropping event")
|
||||
}
|
||||
}
|
||||
|
||||
// handleEvent processes events from the Copilot SDK client.
|
||||
func (a *CopilotAdapter) handleEvent(evt copilot.SessionEvent) {
|
||||
// Log raw event for debugging
|
||||
if rawData, err := json.Marshal(evt); err == nil {
|
||||
shared.LogRawEvent(shared.ProtocolCopilot, a.cfg.AgentID, string(evt.Type), rawData)
|
||||
}
|
||||
|
||||
a.mu.RLock()
|
||||
sessionID := a.sessionID
|
||||
operationID := a.operationID
|
||||
a.mu.RUnlock()
|
||||
|
||||
// Use session ID from event data if provided
|
||||
if evt.Data.SessionID != nil && *evt.Data.SessionID != "" {
|
||||
sessionID = *evt.Data.SessionID
|
||||
}
|
||||
|
||||
switch evt.Type {
|
||||
case copilot.EventTypeSessionStart:
|
||||
a.logger.Info("session started", zap.String("session_id", sessionID))
|
||||
|
||||
case copilot.EventTypeSessionResume:
|
||||
a.logger.Info("session resumed", zap.String("session_id", sessionID))
|
||||
|
||||
case copilot.EventTypeAssistantMessage:
|
||||
// Full message (non-streaming) - only accumulate if we haven't received deltas
|
||||
// When streaming is used, deltas build up the text incrementally, and the final
|
||||
// assistant.message contains the full text again - skip to avoid duplication
|
||||
if evt.Data.Content != nil && *evt.Data.Content != "" {
|
||||
a.mu.Lock()
|
||||
alreadyReceivedDeltas := a.receivedDeltas
|
||||
if !alreadyReceivedDeltas {
|
||||
// Non-streaming mode: accumulate the full message
|
||||
a.textAccumulator.WriteString(*evt.Data.Content)
|
||||
}
|
||||
a.mu.Unlock()
|
||||
|
||||
// Only send message chunk if we haven't been streaming
|
||||
if !alreadyReceivedDeltas {
|
||||
a.sendUpdate(AgentEvent{
|
||||
Type: EventTypeMessageChunk,
|
||||
SessionID: sessionID,
|
||||
OperationID: operationID,
|
||||
Text: *evt.Data.Content,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
case copilot.EventTypeAssistantMessageDelta:
|
||||
// Streaming message delta
|
||||
if evt.Data.DeltaContent != nil && *evt.Data.DeltaContent != "" {
|
||||
text := *evt.Data.DeltaContent
|
||||
// Accumulate text locally to include in complete event
|
||||
a.mu.Lock()
|
||||
a.textAccumulator.WriteString(text)
|
||||
a.receivedDeltas = true // Mark that we've received streaming deltas
|
||||
a.mu.Unlock()
|
||||
|
||||
a.sendUpdate(AgentEvent{
|
||||
Type: EventTypeMessageChunk,
|
||||
SessionID: sessionID,
|
||||
OperationID: operationID,
|
||||
Text: text,
|
||||
})
|
||||
}
|
||||
|
||||
case copilot.EventTypeAssistantReasoning, copilot.EventTypeAssistantReasoningDelta:
|
||||
// Reasoning/thinking content
|
||||
var content string
|
||||
if evt.Data.Content != nil {
|
||||
content = *evt.Data.Content
|
||||
}
|
||||
if content == "" && evt.Data.DeltaContent != nil {
|
||||
content = *evt.Data.DeltaContent
|
||||
}
|
||||
if content != "" {
|
||||
a.sendUpdate(AgentEvent{
|
||||
Type: EventTypeReasoning,
|
||||
SessionID: sessionID,
|
||||
OperationID: operationID,
|
||||
ReasoningText: content,
|
||||
})
|
||||
}
|
||||
|
||||
case copilot.EventTypeToolStart:
|
||||
a.handleToolStart(evt, sessionID, operationID)
|
||||
|
||||
case copilot.EventTypeToolComplete:
|
||||
a.handleToolComplete(evt, sessionID, operationID)
|
||||
|
||||
case copilot.EventTypeToolProgress:
|
||||
// Tool progress update
|
||||
toolCallID := ""
|
||||
if evt.Data.ToolCallID != nil {
|
||||
toolCallID = *evt.Data.ToolCallID
|
||||
}
|
||||
a.sendUpdate(AgentEvent{
|
||||
Type: EventTypeToolUpdate,
|
||||
SessionID: sessionID,
|
||||
OperationID: operationID,
|
||||
ToolCallID: toolCallID,
|
||||
ToolStatus: "running",
|
||||
})
|
||||
|
||||
case copilot.EventTypeSessionIdle:
|
||||
a.handleSessionIdle(sessionID, operationID)
|
||||
|
||||
case copilot.EventTypeSessionError:
|
||||
a.handleSessionError(evt, sessionID, operationID)
|
||||
|
||||
case copilot.EventTypeSessionUsageInfo, copilot.EventTypeAssistantUsage:
|
||||
a.handleUsageInfo(evt, sessionID, operationID)
|
||||
|
||||
case copilot.EventTypeAssistantTurnStart:
|
||||
a.logger.Debug("turn started", zap.String("operation_id", operationID))
|
||||
|
||||
case copilot.EventTypeAssistantTurnEnd:
|
||||
a.logger.Debug("turn ended", zap.String("operation_id", operationID))
|
||||
|
||||
case copilot.EventTypeAbort:
|
||||
a.logger.Info("operation aborted")
|
||||
a.signalCompletion(false, "operation aborted")
|
||||
|
||||
default:
|
||||
a.logger.Debug("unhandled SDK event", zap.String("type", string(evt.Type)))
|
||||
}
|
||||
}
|
||||
|
||||
// handleToolStart processes tool execution start events.
|
||||
func (a *CopilotAdapter) handleToolStart(evt copilot.SessionEvent, sessionID, operationID string) {
|
||||
toolCallID := ""
|
||||
toolName := ""
|
||||
var toolArgs any
|
||||
|
||||
if evt.Data.ToolCallID != nil {
|
||||
toolCallID = *evt.Data.ToolCallID
|
||||
}
|
||||
if evt.Data.ToolName != nil {
|
||||
toolName = *evt.Data.ToolName
|
||||
}
|
||||
toolArgs = evt.Data.Arguments
|
||||
|
||||
// Generate normalized payload using the normalizer
|
||||
normalizedPayload := a.normalizer.NormalizeToolCall(toolName, toolArgs)
|
||||
|
||||
// Build human-readable title
|
||||
toolTitle := toolName
|
||||
if argsMap, ok := toolArgs.(map[string]any); ok && argsMap != nil {
|
||||
if cmd, ok := argsMap["command"].(string); ok && strings.ToLower(toolName) == "bash" {
|
||||
toolTitle = cmd
|
||||
} else if path, ok := argsMap["file_path"].(string); ok {
|
||||
toolTitle = fmt.Sprintf("%s: %s", toolName, path)
|
||||
}
|
||||
}
|
||||
|
||||
a.logger.Info("tool execution started",
|
||||
zap.String("tool_call_id", toolCallID),
|
||||
zap.String("tool_name", toolName))
|
||||
|
||||
// Track pending tool call and cache the normalized payload for result handling
|
||||
a.mu.Lock()
|
||||
a.pendingToolPayloads[toolCallID] = normalizedPayload
|
||||
a.mu.Unlock()
|
||||
|
||||
a.sendUpdate(AgentEvent{
|
||||
Type: EventTypeToolCall,
|
||||
SessionID: sessionID,
|
||||
OperationID: operationID,
|
||||
ToolCallID: toolCallID,
|
||||
ToolName: toolName,
|
||||
ToolTitle: toolTitle,
|
||||
ToolStatus: "running",
|
||||
NormalizedPayload: normalizedPayload,
|
||||
})
|
||||
}
|
||||
|
||||
// handleToolComplete processes tool execution complete events.
|
||||
func (a *CopilotAdapter) handleToolComplete(evt copilot.SessionEvent, sessionID, operationID string) {
|
||||
toolCallID := ""
|
||||
if evt.Data.ToolCallID != nil {
|
||||
toolCallID = *evt.Data.ToolCallID
|
||||
}
|
||||
|
||||
status := "complete"
|
||||
|
||||
a.logger.Info("tool execution completed",
|
||||
zap.String("tool_call_id", toolCallID),
|
||||
zap.String("status", status))
|
||||
|
||||
// Get cached payload and remove from pending
|
||||
a.mu.Lock()
|
||||
cachedPayload := a.pendingToolPayloads[toolCallID]
|
||||
delete(a.pendingToolPayloads, toolCallID)
|
||||
a.mu.Unlock()
|
||||
|
||||
// Normalize the tool result if we have result data
|
||||
var normalizedPayload *streams.NormalizedPayload
|
||||
if cachedPayload != nil {
|
||||
// Use the result from the event if available
|
||||
normalizedPayload = a.normalizer.NormalizeToolResult(cachedPayload, evt.Data.Result, false)
|
||||
}
|
||||
|
||||
a.sendUpdate(AgentEvent{
|
||||
Type: EventTypeToolUpdate,
|
||||
SessionID: sessionID,
|
||||
OperationID: operationID,
|
||||
ToolCallID: toolCallID,
|
||||
ToolStatus: status,
|
||||
NormalizedPayload: normalizedPayload,
|
||||
})
|
||||
}
|
||||
|
||||
// handleSessionIdle processes session idle (completion) events.
|
||||
func (a *CopilotAdapter) handleSessionIdle(sessionID, operationID string) {
|
||||
// Check if we already sent completion for this operation (SDK may send multiple idle events)
|
||||
a.mu.Lock()
|
||||
if a.completeSent {
|
||||
a.mu.Unlock()
|
||||
a.logger.Debug("ignoring duplicate session idle event",
|
||||
zap.String("session_id", sessionID),
|
||||
zap.String("operation_id", operationID))
|
||||
return
|
||||
}
|
||||
a.completeSent = true
|
||||
|
||||
// Clear the text accumulator - text was already sent via message_chunk events
|
||||
// so we should NOT include it in the complete event to avoid duplicates
|
||||
a.textAccumulator.Reset()
|
||||
|
||||
// Auto-complete any pending tool calls
|
||||
pendingTools := make(map[string]*streams.NormalizedPayload, len(a.pendingToolPayloads))
|
||||
for toolID, payload := range a.pendingToolPayloads {
|
||||
pendingTools[toolID] = payload
|
||||
}
|
||||
a.pendingToolPayloads = make(map[string]*streams.NormalizedPayload)
|
||||
a.mu.Unlock()
|
||||
|
||||
a.logger.Info("session idle (operation completed)",
|
||||
zap.String("session_id", sessionID),
|
||||
zap.String("operation_id", operationID))
|
||||
|
||||
// Auto-complete pending tool calls
|
||||
for toolID, cachedPayload := range pendingTools {
|
||||
a.logger.Info("auto-completing pending tool call on idle",
|
||||
zap.String("tool_call_id", toolID))
|
||||
a.sendUpdate(AgentEvent{
|
||||
Type: EventTypeToolUpdate,
|
||||
SessionID: sessionID,
|
||||
OperationID: operationID,
|
||||
ToolCallID: toolID,
|
||||
ToolStatus: "complete",
|
||||
NormalizedPayload: cachedPayload,
|
||||
})
|
||||
}
|
||||
|
||||
// Send completion event WITHOUT text - text was already sent via message_chunk events.
|
||||
// Including text here would cause duplicate messages.
|
||||
a.sendUpdate(AgentEvent{
|
||||
Type: EventTypeComplete,
|
||||
SessionID: sessionID,
|
||||
OperationID: operationID,
|
||||
})
|
||||
|
||||
// Signal completion
|
||||
a.signalCompletion(true, "")
|
||||
}
|
||||
|
||||
// handleSessionError processes session error events.
|
||||
func (a *CopilotAdapter) handleSessionError(evt copilot.SessionEvent, sessionID, operationID string) {
|
||||
errorMsg := "unknown error"
|
||||
if evt.Data.Message != nil {
|
||||
errorMsg = *evt.Data.Message
|
||||
}
|
||||
|
||||
errorType := ""
|
||||
if evt.Data.ErrorType != nil {
|
||||
errorType = *evt.Data.ErrorType
|
||||
}
|
||||
|
||||
a.logger.Error("session error",
|
||||
zap.String("error", errorMsg),
|
||||
zap.String("error_type", errorType))
|
||||
|
||||
a.sendUpdate(AgentEvent{
|
||||
Type: EventTypeError,
|
||||
SessionID: sessionID,
|
||||
OperationID: operationID,
|
||||
Error: errorMsg,
|
||||
})
|
||||
|
||||
// Signal failure
|
||||
a.signalCompletion(false, errorMsg)
|
||||
}
|
||||
|
||||
// handleUsageInfo processes usage info events.
|
||||
func (a *CopilotAdapter) handleUsageInfo(evt copilot.SessionEvent, sessionID, operationID string) {
|
||||
var inputTokens, outputTokens int64
|
||||
if evt.Data.InputTokens != nil {
|
||||
inputTokens = int64(*evt.Data.InputTokens)
|
||||
}
|
||||
if evt.Data.OutputTokens != nil {
|
||||
outputTokens = int64(*evt.Data.OutputTokens)
|
||||
}
|
||||
|
||||
// Update context tracking
|
||||
contextUsed := inputTokens + outputTokens
|
||||
|
||||
a.mu.Lock()
|
||||
a.contextTokensUsed = contextUsed
|
||||
contextSize := a.contextWindowSize
|
||||
a.mu.Unlock()
|
||||
|
||||
remaining := contextSize - contextUsed
|
||||
if remaining < 0 {
|
||||
remaining = 0
|
||||
}
|
||||
|
||||
a.sendUpdate(AgentEvent{
|
||||
Type: EventTypeContextWindow,
|
||||
SessionID: sessionID,
|
||||
OperationID: operationID,
|
||||
ContextWindowSize: contextSize,
|
||||
ContextWindowUsed: contextUsed,
|
||||
ContextWindowRemaining: remaining,
|
||||
ContextEfficiency: float64(contextUsed) / float64(contextSize) * 100,
|
||||
})
|
||||
}
|
||||
|
||||
// signalCompletion signals the result channel.
|
||||
func (a *CopilotAdapter) signalCompletion(success bool, errMsg string) {
|
||||
a.mu.RLock()
|
||||
resultCh := a.resultCh
|
||||
a.mu.RUnlock()
|
||||
|
||||
if resultCh != nil {
|
||||
select {
|
||||
case resultCh <- resultComplete{success: success, err: errMsg}:
|
||||
default:
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// RequiresProcessKill returns true because the Copilot CLI server doesn't exit on stdin close.
|
||||
// The CLI runs as an HTTP server that must be explicitly killed.
|
||||
func (a *CopilotAdapter) RequiresProcessKill() bool {
|
||||
return true
|
||||
}
|
||||
|
||||
// Verify interface implementation
|
||||
var _ AgentAdapter = (*CopilotAdapter)(nil)
|
||||
@@ -0,0 +1,381 @@
|
||||
package adapter
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"regexp"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"github.com/kandev/kandev/internal/agentctl/server/adapter/transport/shared"
|
||||
"github.com/kandev/kandev/internal/agentctl/types/streams"
|
||||
)
|
||||
|
||||
// exitedPattern matches the "<exited with exit code N>" suffix that Copilot appends to shell output.
|
||||
var exitedPattern = regexp.MustCompile(`\n?<exited with exit code (\d+)>$`)
|
||||
|
||||
// Copilot tool name constants.
|
||||
// Copilot typically uses lowercase snake_case tool names.
|
||||
const (
|
||||
CopilotToolBash = "bash"
|
||||
CopilotToolReadFile = "read_file"
|
||||
CopilotToolWriteFile = "write_file"
|
||||
CopilotToolEditFile = "edit_file"
|
||||
CopilotToolGlob = "glob"
|
||||
CopilotToolGrep = "grep"
|
||||
CopilotToolWebFetch = "web_fetch"
|
||||
CopilotToolWebSearch = "web_search"
|
||||
)
|
||||
|
||||
// CopilotNormalizer converts Copilot SDK tool data to NormalizedPayload.
|
||||
type CopilotNormalizer struct{}
|
||||
|
||||
// NewCopilotNormalizer creates a new Copilot normalizer.
|
||||
func NewCopilotNormalizer() *CopilotNormalizer {
|
||||
return &CopilotNormalizer{}
|
||||
}
|
||||
|
||||
// NormalizeToolCall converts Copilot tool call data to NormalizedPayload.
|
||||
func (n *CopilotNormalizer) NormalizeToolCall(toolName string, args any) *streams.NormalizedPayload {
|
||||
// Convert args to map if possible
|
||||
argsMap := n.toMap(args)
|
||||
|
||||
// Normalize tool name to lowercase for comparison
|
||||
toolNameLower := strings.ToLower(toolName)
|
||||
|
||||
switch toolNameLower {
|
||||
case CopilotToolBash, "shell", "execute", "run":
|
||||
return n.normalizeExecute(argsMap)
|
||||
case CopilotToolReadFile, "read":
|
||||
return n.normalizeRead(argsMap)
|
||||
case CopilotToolWriteFile, "write":
|
||||
return n.normalizeWrite(argsMap)
|
||||
case CopilotToolEditFile, "edit":
|
||||
return n.normalizeEdit(argsMap)
|
||||
case CopilotToolGlob, CopilotToolGrep, "search", "find":
|
||||
return n.normalizeCodeSearch(toolNameLower, argsMap)
|
||||
case CopilotToolWebFetch, CopilotToolWebSearch, "fetch", "http":
|
||||
return n.normalizeHttpRequest(toolNameLower, argsMap)
|
||||
default:
|
||||
return n.normalizeGeneric(toolName, argsMap)
|
||||
}
|
||||
}
|
||||
|
||||
// NormalizeToolResult updates the payload with tool result data.
|
||||
func (n *CopilotNormalizer) NormalizeToolResult(cachedPayload *streams.NormalizedPayload, result any, isError bool) *streams.NormalizedPayload {
|
||||
if cachedPayload == nil {
|
||||
// No cached payload, create a generic one with the result
|
||||
output := n.extractOutput(result)
|
||||
payload := streams.NewGeneric("unknown", nil)
|
||||
if payload.Generic() != nil {
|
||||
payload.Generic().Output = output
|
||||
}
|
||||
return payload
|
||||
}
|
||||
|
||||
// Update the cached payload with the result
|
||||
switch cachedPayload.Kind() {
|
||||
case streams.ToolKindShellExec:
|
||||
if cachedPayload.ShellExec() != nil {
|
||||
output, exitCode := n.parseShellResult(result)
|
||||
cachedPayload.ShellExec().Output = &streams.ShellExecOutput{
|
||||
Stdout: output,
|
||||
ExitCode: exitCode,
|
||||
}
|
||||
if isError {
|
||||
cachedPayload.ShellExec().Output.Stderr = output
|
||||
cachedPayload.ShellExec().Output.Stdout = ""
|
||||
}
|
||||
}
|
||||
case streams.ToolKindHttpRequest:
|
||||
if cachedPayload.HttpRequest() != nil {
|
||||
if output, ok := n.extractOutput(result).(string); ok {
|
||||
cachedPayload.HttpRequest().Response = output
|
||||
}
|
||||
}
|
||||
case streams.ToolKindGeneric:
|
||||
if cachedPayload.Generic() != nil {
|
||||
cachedPayload.Generic().Output = n.extractOutput(result)
|
||||
}
|
||||
case streams.ToolKindReadFile:
|
||||
if cachedPayload.ReadFile() != nil {
|
||||
if output, ok := n.extractOutput(result).(string); ok {
|
||||
cachedPayload.ReadFile().Output = &streams.ReadFileOutput{
|
||||
Content: output,
|
||||
}
|
||||
}
|
||||
}
|
||||
case streams.ToolKindCodeSearch:
|
||||
if cachedPayload.CodeSearch() != nil {
|
||||
if output, ok := n.extractOutput(result).(string); ok {
|
||||
// Parse file list from newline-separated output
|
||||
files := n.parseFileList(output)
|
||||
cachedPayload.CodeSearch().Output = &streams.CodeSearchOutput{
|
||||
Files: files,
|
||||
FileCount: len(files),
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return cachedPayload
|
||||
}
|
||||
|
||||
// toMap converts an argument to a map[string]any.
|
||||
func (n *CopilotNormalizer) toMap(args any) map[string]any {
|
||||
if args == nil {
|
||||
return nil
|
||||
}
|
||||
if m, ok := args.(map[string]any); ok {
|
||||
return m
|
||||
}
|
||||
// Try JSON round-trip for struct types
|
||||
if data, err := json.Marshal(args); err == nil {
|
||||
var m map[string]any
|
||||
if err := json.Unmarshal(data, &m); err == nil {
|
||||
return m
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// extractOutput extracts a string output from various result formats.
|
||||
func (n *CopilotNormalizer) extractOutput(result any) any {
|
||||
if result == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
switch r := result.(type) {
|
||||
case string:
|
||||
return r
|
||||
case map[string]any:
|
||||
// Try common output field names
|
||||
if output, ok := r["output"].(string); ok {
|
||||
return output
|
||||
}
|
||||
if stdout, ok := r["stdout"].(string); ok {
|
||||
return stdout
|
||||
}
|
||||
if content, ok := r["content"].(string); ok {
|
||||
return content
|
||||
}
|
||||
if text, ok := r["text"].(string); ok {
|
||||
return text
|
||||
}
|
||||
return r
|
||||
default:
|
||||
// Handle struct types (like Copilot SDK's *Result) via JSON round-trip
|
||||
if data, err := json.Marshal(result); err == nil {
|
||||
var m map[string]any
|
||||
if err := json.Unmarshal(data, &m); err == nil {
|
||||
// Try to extract content from the unmarshaled map
|
||||
if content, ok := m["content"].(string); ok {
|
||||
return content
|
||||
}
|
||||
if output, ok := m["output"].(string); ok {
|
||||
return output
|
||||
}
|
||||
if stdout, ok := m["stdout"].(string); ok {
|
||||
return stdout
|
||||
}
|
||||
if text, ok := m["text"].(string); ok {
|
||||
return text
|
||||
}
|
||||
}
|
||||
}
|
||||
return result
|
||||
}
|
||||
}
|
||||
|
||||
// parseShellResult extracts output and exit code from shell execution result.
|
||||
func (n *CopilotNormalizer) parseShellResult(result any) (string, int) {
|
||||
if result == nil {
|
||||
return "", 0
|
||||
}
|
||||
|
||||
var output string
|
||||
var exitCode int
|
||||
|
||||
switch r := result.(type) {
|
||||
case string:
|
||||
output = r
|
||||
case map[string]any:
|
||||
// Try various output field names used by different tools
|
||||
output = shared.GetString(r, "output")
|
||||
if output == "" {
|
||||
output = shared.GetString(r, "stdout")
|
||||
}
|
||||
if output == "" {
|
||||
output = shared.GetString(r, "content")
|
||||
}
|
||||
if output == "" {
|
||||
output = shared.GetString(r, "text")
|
||||
}
|
||||
exitCode = shared.GetInt(r, "exitCode")
|
||||
if exitCode == 0 {
|
||||
exitCode = shared.GetInt(r, "exit_code")
|
||||
}
|
||||
default:
|
||||
// Handle struct types (like Copilot SDK's *Result) via JSON round-trip
|
||||
if data, err := json.Marshal(result); err == nil {
|
||||
var m map[string]any
|
||||
if err := json.Unmarshal(data, &m); err == nil {
|
||||
// Try to extract content from the unmarshaled map
|
||||
if content, ok := m["content"].(string); ok {
|
||||
output = content
|
||||
} else if out, ok := m["output"].(string); ok {
|
||||
output = out
|
||||
} else if stdout, ok := m["stdout"].(string); ok {
|
||||
output = stdout
|
||||
} else {
|
||||
// Fallback: return the JSON string
|
||||
output = string(data)
|
||||
}
|
||||
} else {
|
||||
output = string(data)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Strip Copilot's "<exited with exit code N>" suffix and extract exit code
|
||||
output, exitCode = n.cleanShellOutput(output, exitCode)
|
||||
|
||||
return output, exitCode
|
||||
}
|
||||
|
||||
// cleanShellOutput strips Copilot's "<exited with exit code N>" suffix from shell output
|
||||
// and extracts the exit code if not already set.
|
||||
func (n *CopilotNormalizer) cleanShellOutput(output string, exitCode int) (string, int) {
|
||||
if matches := exitedPattern.FindStringSubmatch(output); matches != nil {
|
||||
// Strip the suffix
|
||||
output = exitedPattern.ReplaceAllString(output, "")
|
||||
// Extract exit code from the suffix if we don't already have one
|
||||
if exitCode == 0 && len(matches) > 1 {
|
||||
if code, err := strconv.Atoi(matches[1]); err == nil {
|
||||
exitCode = code
|
||||
}
|
||||
}
|
||||
}
|
||||
return strings.TrimSuffix(output, "\n"), exitCode
|
||||
}
|
||||
|
||||
// normalizeExecute converts shell execution tool data.
|
||||
func (n *CopilotNormalizer) normalizeExecute(args map[string]any) *streams.NormalizedPayload {
|
||||
command := shared.GetString(args, "command")
|
||||
if command == "" {
|
||||
command = shared.GetString(args, "cmd")
|
||||
}
|
||||
description := shared.GetString(args, "description")
|
||||
timeout := shared.GetInt(args, "timeout")
|
||||
background := shared.GetBool(args, "background")
|
||||
|
||||
return streams.NewShellExec(command, "", description, timeout, background)
|
||||
}
|
||||
|
||||
// normalizeRead converts file read tool data.
|
||||
func (n *CopilotNormalizer) normalizeRead(args map[string]any) *streams.NormalizedPayload {
|
||||
filePath := shared.GetString(args, "file_path")
|
||||
if filePath == "" {
|
||||
filePath = shared.GetString(args, "path")
|
||||
}
|
||||
offset := shared.GetInt(args, "offset")
|
||||
limit := shared.GetInt(args, "limit")
|
||||
|
||||
return streams.NewReadFile(filePath, offset, limit)
|
||||
}
|
||||
|
||||
// normalizeWrite converts file write tool data.
|
||||
func (n *CopilotNormalizer) normalizeWrite(args map[string]any) *streams.NormalizedPayload {
|
||||
filePath := shared.GetString(args, "file_path")
|
||||
if filePath == "" {
|
||||
filePath = shared.GetString(args, "path")
|
||||
}
|
||||
content := shared.GetString(args, "content")
|
||||
|
||||
return streams.NewModifyFile(filePath, []streams.FileMutation{
|
||||
{
|
||||
Type: streams.MutationCreate,
|
||||
Content: content,
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
// normalizeEdit converts file edit tool data.
|
||||
func (n *CopilotNormalizer) normalizeEdit(args map[string]any) *streams.NormalizedPayload {
|
||||
filePath := shared.GetString(args, "file_path")
|
||||
if filePath == "" {
|
||||
filePath = shared.GetString(args, "path")
|
||||
}
|
||||
oldContent := shared.GetString(args, "old_string")
|
||||
if oldContent == "" {
|
||||
oldContent = shared.GetString(args, "old_content")
|
||||
}
|
||||
newContent := shared.GetString(args, "new_string")
|
||||
if newContent == "" {
|
||||
newContent = shared.GetString(args, "new_content")
|
||||
}
|
||||
|
||||
mutation := streams.FileMutation{
|
||||
Type: streams.MutationPatch,
|
||||
OldContent: oldContent,
|
||||
NewContent: newContent,
|
||||
}
|
||||
|
||||
// Generate unified diff
|
||||
if oldContent != "" && newContent != "" {
|
||||
mutation.Diff = shared.GenerateUnifiedDiff(oldContent, newContent, filePath, 0)
|
||||
}
|
||||
|
||||
return streams.NewModifyFile(filePath, []streams.FileMutation{mutation})
|
||||
}
|
||||
|
||||
// normalizeCodeSearch converts code search tool data.
|
||||
func (n *CopilotNormalizer) normalizeCodeSearch(toolName string, args map[string]any) *streams.NormalizedPayload {
|
||||
path := shared.GetString(args, "path")
|
||||
if path == "" {
|
||||
path = shared.GetString(args, "directory")
|
||||
}
|
||||
pattern := shared.GetString(args, "pattern")
|
||||
query := shared.GetString(args, "query")
|
||||
glob := ""
|
||||
|
||||
if toolName == CopilotToolGlob {
|
||||
glob = pattern
|
||||
}
|
||||
|
||||
return streams.NewCodeSearch(query, pattern, path, glob)
|
||||
}
|
||||
|
||||
// normalizeHttpRequest converts HTTP request tool data.
|
||||
func (n *CopilotNormalizer) normalizeHttpRequest(toolName string, args map[string]any) *streams.NormalizedPayload {
|
||||
url := shared.GetString(args, "url")
|
||||
method := shared.GetString(args, "method")
|
||||
if method == "" {
|
||||
method = "GET"
|
||||
}
|
||||
if toolName == CopilotToolWebSearch {
|
||||
method = "SEARCH"
|
||||
}
|
||||
|
||||
return streams.NewHttpRequest(url, method)
|
||||
}
|
||||
|
||||
// normalizeGeneric wraps unknown tools as generic.
|
||||
func (n *CopilotNormalizer) normalizeGeneric(toolName string, args map[string]any) *streams.NormalizedPayload {
|
||||
return streams.NewGeneric(toolName, args)
|
||||
}
|
||||
|
||||
// parseFileList parses newline-separated file paths from grep/glob output.
|
||||
func (n *CopilotNormalizer) parseFileList(output string) []string {
|
||||
if output == "" {
|
||||
return nil
|
||||
}
|
||||
lines := strings.Split(output, "\n")
|
||||
var files []string
|
||||
for _, line := range lines {
|
||||
line = strings.TrimSpace(line)
|
||||
if line != "" {
|
||||
files = append(files, line)
|
||||
}
|
||||
}
|
||||
return files
|
||||
}
|
||||
@@ -32,6 +32,10 @@ func NewAdapter(protocol agent.Protocol, cfg *Config, log *logger.Logger) (Agent
|
||||
return newCodexAdapterWrapper(codex.NewAdapter(sharedCfg, log)), nil
|
||||
case agent.ProtocolOpenCode:
|
||||
return newOpenCodeAdapterWrapper(opencode.NewAdapter(sharedCfg, log)), nil
|
||||
case agent.ProtocolCopilot:
|
||||
return NewCopilotAdapter(cfg, log), nil
|
||||
case agent.ProtocolAmp:
|
||||
return NewAmpAdapter(cfg, log), nil
|
||||
default:
|
||||
return nil, fmt.Errorf("unsupported protocol: %s", protocol)
|
||||
}
|
||||
|
||||
@@ -34,6 +34,8 @@ const (
|
||||
ProtocolStreamJSON = "streamjson"
|
||||
ProtocolCodex = "codex"
|
||||
ProtocolOpenCode = "opencode"
|
||||
ProtocolAmp = "amp"
|
||||
ProtocolCopilot = "copilot"
|
||||
)
|
||||
|
||||
func resolveDebugLogDir() string {
|
||||
|
||||
@@ -125,6 +125,10 @@ func (m *Manager) CreateInstance(ctx context.Context, req *CreateRequest) (*Crea
|
||||
})
|
||||
}
|
||||
|
||||
m.logger.Info("CreateInstance: received request",
|
||||
zap.String("req_protocol", req.Protocol),
|
||||
zap.String("workspace_path", req.WorkspacePath))
|
||||
|
||||
overrides := &config.InstanceOverrides{
|
||||
Protocol: agent.Protocol(req.Protocol),
|
||||
AgentCommand: agentCmd,
|
||||
@@ -136,9 +140,15 @@ func (m *Manager) CreateInstance(ctx context.Context, req *CreateRequest) (*Crea
|
||||
SessionID: req.SessionID,
|
||||
}
|
||||
|
||||
m.logger.Info("CreateInstance: applying overrides",
|
||||
zap.String("override_protocol", string(overrides.Protocol)))
|
||||
|
||||
// Create instance config using the unified method
|
||||
instanceCfg := m.config.NewInstanceConfig(port, overrides)
|
||||
|
||||
m.logger.Info("CreateInstance: instance config created",
|
||||
zap.String("config_protocol", string(instanceCfg.Protocol)))
|
||||
|
||||
// Create process manager
|
||||
procMgr := process.NewManager(instanceCfg, m.logger)
|
||||
|
||||
@@ -220,17 +230,20 @@ func (m *Manager) ListInstances() []*InstanceInfo {
|
||||
|
||||
// StopInstance stops and removes an instance by ID.
|
||||
func (m *Manager) StopInstance(ctx context.Context, id string) error {
|
||||
// Get instance and remove from map under lock (quick operation)
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
|
||||
inst, ok := m.instances[id]
|
||||
if !ok {
|
||||
m.mu.Unlock()
|
||||
return fmt.Errorf("instance %s not found", id)
|
||||
}
|
||||
// Remove from map immediately so new instances aren't blocked
|
||||
delete(m.instances, id)
|
||||
m.mu.Unlock()
|
||||
|
||||
m.logger.Debug("stopping instance", zap.String("instance_id", id))
|
||||
|
||||
// Stop the process manager
|
||||
// Stop the process manager (potentially slow, done without lock)
|
||||
if inst.manager != nil {
|
||||
if err := inst.manager.Stop(ctx); err != nil {
|
||||
m.logger.Warn("error stopping process manager",
|
||||
@@ -243,7 +256,7 @@ func (m *Manager) StopInstance(ctx context.Context, id string) error {
|
||||
zap.String("instance_id", id),
|
||||
zap.Int("port", inst.Port))
|
||||
|
||||
// Shutdown HTTP server
|
||||
// Shutdown HTTP server (potentially slow, done without lock)
|
||||
if inst.server != nil {
|
||||
if err := inst.server.Shutdown(ctx); err != nil {
|
||||
m.logger.Warn("error shutting down HTTP server",
|
||||
@@ -256,11 +269,10 @@ func (m *Manager) StopInstance(ctx context.Context, id string) error {
|
||||
zap.String("instance_id", id),
|
||||
zap.Int("port", inst.Port))
|
||||
|
||||
// Release port
|
||||
// Release port (quick operation, re-acquire lock)
|
||||
m.mu.Lock()
|
||||
m.portAlloc.Release(inst.Port)
|
||||
|
||||
// Remove from instances map
|
||||
delete(m.instances, id)
|
||||
m.mu.Unlock()
|
||||
|
||||
m.logger.Info("StopInstance completed",
|
||||
zap.String("instance_id", id),
|
||||
|
||||
@@ -411,7 +411,7 @@ func (m *Manager) createAdapter() error {
|
||||
return fmt.Errorf("protocol not specified in configuration")
|
||||
}
|
||||
|
||||
// Use the adapter factory to create the appropriate adapter
|
||||
m.logger.Debug("creating adapter", zap.String("protocol", string(protocol)))
|
||||
adpt, err := adapter.NewAdapter(protocol, m.adapterCfg, m.logger)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to create adapter: %w", err)
|
||||
@@ -653,10 +653,35 @@ func (m *Manager) waitForExit() {
|
||||
|
||||
if err != nil {
|
||||
m.exitErr.Store(errorWrapper{err: err})
|
||||
exitCode := -1
|
||||
if exitErr, ok := err.(*exec.ExitError); ok {
|
||||
m.exitCode.Store(int32(exitErr.ExitCode()))
|
||||
exitCode = exitErr.ExitCode()
|
||||
m.exitCode.Store(int32(exitCode))
|
||||
}
|
||||
// Include recent stderr for better error diagnostics
|
||||
recentStderr := m.GetRecentStderr()
|
||||
m.logger.Error("agent process exited with error",
|
||||
zap.Error(err),
|
||||
zap.Int("exit_code", exitCode),
|
||||
zap.Strings("recent_stderr", recentStderr))
|
||||
|
||||
// Send error event to the updates channel so UI can display it
|
||||
errorMsg := fmt.Sprintf("Agent process exited with code %d", exitCode)
|
||||
if len(recentStderr) > 0 {
|
||||
errorMsg = fmt.Sprintf("%s: %s", errorMsg, strings.Join(recentStderr, "; "))
|
||||
}
|
||||
select {
|
||||
case m.updatesCh <- adapter.AgentEvent{
|
||||
Type: adapter.EventTypeError,
|
||||
Error: errorMsg,
|
||||
Data: map[string]any{
|
||||
"exit_code": exitCode,
|
||||
"recent_stderr": recentStderr,
|
||||
},
|
||||
}:
|
||||
default:
|
||||
m.logger.Warn("updates channel full, could not send exit error event")
|
||||
}
|
||||
m.logger.Info("agent process exited with error", zap.Error(err))
|
||||
} else {
|
||||
m.exitCode.Store(0)
|
||||
m.logger.Info("agent process exited successfully")
|
||||
|
||||
Reference in New Issue
Block a user