rdev/internal/adapter/codeagent/opencode/adapter.go

// Package opencode provides a CodeAgent implementation for the OpenCode server.
package opencode

import (
	"context"
	"encoding/json"
	"fmt"
	"strings"
	"sync"
	"time"

	"github.com/orchard9/rdev/internal/domain"
	"github.com/orchard9/rdev/internal/port"
)

// Adapter implements port.CodeAgent using OpenCode's HTTP server.
type Adapter struct {
	client *Client

	// Track active sessions for cancellation
	activeSessions map[string]context.CancelFunc
	sessionsMu     sync.Mutex
}

// NewAdapter creates a new OpenCode adapter.
func NewAdapter(cfg ClientConfig) *Adapter {
	return &Adapter{
		client:         NewClient(cfg),
		activeSessions: make(map[string]context.CancelFunc),
	}
}

// Ensure Adapter implements port.CodeAgent at compile time.
var _ port.CodeAgent = (*Adapter)(nil)

// Name returns a human-readable name for this agent.
func (a *Adapter) Name() string {
	return "OpenCode"
}

// Provider returns the agent provider identifier.
func (a *Adapter) Provider() domain.AgentProvider {
	return domain.AgentProviderOpenCode
}

// Execute runs an OpenCode command and streams events to the handler.
func (a *Adapter) Execute(ctx context.Context, req *domain.AgentRequest, handler domain.AgentEventHandler) (*domain.AgentResult, error) {
	if req.Prompt == "" {
		return nil, fmt.Errorf("prompt is required")
	}

	// Create cancellable context
	execCtx, cancel := context.WithCancel(ctx)
	defer cancel()

	// Apply timeout if specified
	if req.Timeout > 0 {
		var timeoutCancel context.CancelFunc
		execCtx, timeoutCancel = context.WithTimeout(execCtx, req.Timeout)
		defer timeoutCancel()
	}

	startTime := time.Now()

	// Get or create session
	sessionID := req.SessionID
	if sessionID == "" {
		session, err := a.client.CreateSession(execCtx, &CreateSessionRequest{
			Title: fmt.Sprintf("rdev-%d", time.Now().Unix()),
		})
		if err != nil {
			return &domain.AgentResult{
				ExitCode: 1,
				Error:    fmt.Errorf("create session: %w", err),
			}, nil
		}
		sessionID = session.ID
	}

	// Track session for potential cancellation
	a.sessionsMu.Lock()
	a.activeSessions[sessionID] = cancel
	a.sessionsMu.Unlock()

	defer func() {
		a.sessionsMu.Lock()
		delete(a.activeSessions, sessionID)
		a.sessionsMu.Unlock()
	}()

	// Emit session started event
	handler(domain.AgentEvent{
		Type:      domain.AgentEventOutput,
		Timestamp: time.Now(),
		Content:   "Session started",
		Metadata: map[string]any{
			"session_id": sessionID,
		},
	})

	// Subscribe to SSE events for real-time updates
	eventChan, err := a.client.SubscribeEvents(execCtx)
	if err != nil {
		// Non-fatal: we can still use sync API
		handler(domain.AgentEvent{
			Type:      domain.AgentEventError,
			Timestamp: time.Now(),
			Content:   fmt.Sprintf("SSE subscription failed: %v (using sync mode)", err),
		})
	}

	// Start event consumer if we have SSE
	var eventWg sync.WaitGroup
	if eventChan != nil {
		eventWg.Add(1)
		go func() {
			defer eventWg.Done()
			a.consumeEvents(execCtx, eventChan, handler)
		}()
	}

	// Build message request
	msgReq := &SendMessageRequest{
		Parts: []MessagePart{
			{Type: "text", Content: req.Prompt},
		},
	}

	// Set model if specified
	if req.Model != "" {
		msgReq.Model = req.Model
	}

	// Set tools if specified
	if len(req.AllowedTools) > 0 {
		msgReq.Tools = req.AllowedTools
	}

	// Send message (synchronous call that waits for response)
	resp, err := a.client.SendMessage(execCtx, sessionID, msgReq)
	if err != nil {
		// Check if cancelled
		if execCtx.Err() == context.Canceled {
			return &domain.AgentResult{
				SessionID:  sessionID,
				ExitCode:   1,
				DurationMs: time.Since(startTime).Milliseconds(),
				Error:      domain.ErrCommandCancelled,
			}, nil
		}

		return &domain.AgentResult{
			SessionID:  sessionID,
			ExitCode:   1,
			DurationMs: time.Since(startTime).Milliseconds(),
			Error:      fmt.Errorf("send message: %w", err),
		}, nil
	}

	// Process response parts
	var finalOutput strings.Builder
	var hasError bool
	var errorContent string
	for _, part := range resp.Parts {
		event := a.partToEvent(part)
		handler(event)

		if part.Type == "text" && part.Content != "" {
			finalOutput.WriteString(part.Content)
		}
		if part.Type == "error" {
			hasError = true
			errorContent = part.Content
		}
	}

	// Emit completion event
	duration := time.Since(startTime)
	completionStatus := "success"
	if hasError {
		completionStatus = "error"
	}
	handler(domain.AgentEvent{
		Type:      domain.AgentEventComplete,
		Timestamp: time.Now(),
		Metadata: map[string]any{
			"status":      completionStatus,
			"duration_ms": duration.Milliseconds(),
		},
	})

	// Wait for SSE consumer to finish (with timeout)
	if eventChan != nil {
		cancel() // Signal event consumer to stop
		waitDone := make(chan struct{})
		go func() {
			eventWg.Wait()
			close(waitDone)
		}()

		select {
		case <-waitDone:
		case <-time.After(2 * time.Second):
			// Event consumer didn't stop in time, continue anyway
		}
	}

	result := &domain.AgentResult{
		SessionID:   sessionID,
		ExitCode:    0,
		DurationMs:  duration.Milliseconds(),
		FinalOutput: finalOutput.String(),
	}

	// Set error state if error parts were found
	if hasError {
		result.ExitCode = 1
		if errorContent != "" {
			result.Error = fmt.Errorf("agent error: %s", errorContent)
		}
	}

	return result, nil
}

// consumeEvents processes SSE events and dispatches them to the handler.
func (a *Adapter) consumeEvents(ctx context.Context, events <-chan SSEEvent, handler domain.AgentEventHandler) {
	for {
		select {
		case <-ctx.Done():
			return
		case event, ok := <-events:
			if !ok {
				return
			}
			agentEvent := a.sseToEvent(event)
			if agentEvent.Type != "" {
				handler(agentEvent)
			}
		}
	}
}

// sseToEvent converts an SSE event to a domain.AgentEvent.
func (a *Adapter) sseToEvent(sse SSEEvent) domain.AgentEvent {
	event := domain.AgentEvent{
		Timestamp: time.Now(),
		Metadata:  make(map[string]any),
	}

	// Parse event type
	switch sse.Event {
	case "server.connected":
		event.Type = domain.AgentEventOutput
		event.Content = "Connected to OpenCode server"
		return event

	case "message.created", "message.updated":
		event.Type = domain.AgentEventOutput
		// Try to parse data for content
		var data map[string]any
		if json.Unmarshal([]byte(sse.Data), &data) == nil {
			if content, ok := data["content"].(string); ok {
				event.Content = content
			}
		}
		return event

	case "tool.started":
		event.Type = domain.AgentEventToolUse
		var data map[string]any
		if json.Unmarshal([]byte(sse.Data), &data) == nil {
			if name, ok := data["name"].(string); ok {
				event.ToolName = name
				event.Content = name
			}
			if input, ok := data["input"].(map[string]any); ok {
				event.ToolInput = input
			}
		}
		return event

	case "tool.completed":
		event.Type = domain.AgentEventToolResult
		var data map[string]any
		if json.Unmarshal([]byte(sse.Data), &data) == nil {
			if output, ok := data["output"].(string); ok {
				event.Content = output
			}
		}
		return event

	case "session.completed":
		event.Type = domain.AgentEventComplete
		return event

	case "error":
		event.Type = domain.AgentEventError
		event.Content = sse.Data
		return event
	}

	// Unknown event type
	return domain.AgentEvent{}
}

// partToEvent converts a message part to a domain.AgentEvent.
func (a *Adapter) partToEvent(part MessagePart) domain.AgentEvent {
	event := domain.AgentEvent{
		Timestamp: time.Now(),
		Metadata:  make(map[string]any),
	}

	switch part.Type {
	case "text":
		event.Type = domain.AgentEventOutput
		event.Content = part.Content

	case "tool_use":
		event.Type = domain.AgentEventToolUse
		event.ToolName = part.Name
		event.Content = part.Name
		if input, ok := part.Input.(map[string]any); ok {
			event.ToolInput = input
		}

	case "tool_result":
		event.Type = domain.AgentEventToolResult
		event.Content = part.Content

	default:
		event.Type = domain.AgentEventOutput
		event.Content = part.Content
	}

	return event
}

// Cancel attempts to cancel a running session.
func (a *Adapter) Cancel(ctx context.Context, sessionID string) error {
	a.sessionsMu.Lock()
	cancel, exists := a.activeSessions[sessionID]
	if exists {
		// Call cancel while holding lock and delete to prevent double-cancel
		cancel()
		delete(a.activeSessions, sessionID)
	}
	a.sessionsMu.Unlock()

	// Abort on the server side regardless of local session state
	return a.client.AbortSession(ctx, sessionID)
}

// Capabilities returns what this agent supports.
func (a *Adapter) Capabilities() domain.AgentCapabilities {
	return domain.AgentCapabilities{
		Provider:                    domain.AgentProviderOpenCode,
		SupportsSessionContinuation: true,
		SupportsModelSelection:      true, // OpenCode supports multiple providers
		SupportsToolControl:         true,
		SupportedModels: []string{
			"claude-sonnet-4-20250514",
			"claude-opus-4-20250514",
			"gpt-4o",
			"gpt-4-turbo",
			"gemini-pro",
		},
		DefaultModel:      "claude-sonnet-4-20250514",
		MaxPromptLength:   0, // Unlimited
		SupportsStreaming: true,
	}
}

// DefaultAvailabilityTimeout is the maximum time to wait when checking agent availability.
// This timeout prevents blocking the caller when the server is slow or unresponsive.
const DefaultAvailabilityTimeout = 5 * time.Second

// Available checks if the OpenCode server is healthy.
func (a *Adapter) Available(ctx context.Context) bool {
	ctx, cancel := context.WithTimeout(ctx, DefaultAvailabilityTimeout)
	defer cancel()

	health, err := a.client.Health(ctx)
	if err != nil {
		return false
	}
	return health.Healthy
}