diff --git a/apps/backend/internal/agent/agents/opencode.go b/apps/backend/internal/agent/agents/opencode.go index 18598082d8..131f57f3ef 100644 --- a/apps/backend/internal/agent/agents/opencode.go +++ b/apps/backend/internal/agent/agents/opencode.go @@ -93,13 +93,13 @@ func (a *OpenCode) ListModels(ctx context.Context) (*ModelList, error) { } func (a *OpenCode) BuildCommand(opts CommandOptions) Command { - return Cmd("npx", "-y", "opencode-ai@1.1.25", "serve", "--hostname", "127.0.0.1", "--port", "0").Build() + return Cmd("npx", "-y", "opencode-ai@1.1.59", "serve", "--hostname", "127.0.0.1", "--port", "0").Build() } func (a *OpenCode) Runtime() *RuntimeConfig { canRecover := true return &RuntimeConfig{ - Cmd: Cmd("npx", "-y", "opencode-ai@1.1.25", "serve", "--hostname", "127.0.0.1", "--port", "0").Build(), + Cmd: Cmd("npx", "-y", "opencode-ai@1.1.59", "serve", "--hostname", "127.0.0.1", "--port", "0").Build(), WorkingDir: "{workspace}", Env: map[string]string{}, ResourceLimits: ResourceLimits{MemoryMB: 4096, CPUCores: 2.0, Timeout: time.Hour}, diff --git a/apps/backend/internal/agentctl/server/adapter/transport/opencode/adapter.go b/apps/backend/internal/agentctl/server/adapter/transport/opencode/adapter.go index 8a7f48b797..365a91fe31 100644 --- a/apps/backend/internal/agentctl/server/adapter/transport/opencode/adapter.go +++ b/apps/backend/internal/agentctl/server/adapter/transport/opencode/adapter.go @@ -86,6 +86,9 @@ type Adapter struct { // Result completion signaling resultCh chan resultComplete + // Model selection (parsed from AGENT_MODEL env var, format: "provider/model") + model *opencode.ModelSpec + // Token tracking totalTokens int modelContextWindow int @@ -140,11 +143,19 @@ func NewAdapter(cfg *shared.Config, log *logger.Logger) *Adapter { if agentID == "" { agentID = "opencode" // Fallback for unknown agents using this protocol } + // Parse model from AGENT_MODEL env var (format: "provider/model") + var modelSpec *opencode.ModelSpec + if modelStr := os.Getenv("AGENT_MODEL"); modelStr != "" { + modelSpec = parseModelSpec(modelStr) + log.Info("parsed model from AGENT_MODEL", zap.String("model", modelStr)) + } + return &Adapter{ cfg: cfg, logger: log.WithFields(zap.String("adapter", "opencode"), zap.String("agent_id", agentID)), agentID: agentID, normalizer: NewNormalizer(), + model: modelSpec, ctx: ctx, cancel: cancel, updatesCh: make(chan AgentEvent, 100), @@ -181,16 +192,23 @@ func (a *Adapter) PrepareEnvironment() (map[string]string, error) { "OPENCODE_SERVER_PASSWORD": a.password, } - // Set up permissions based on config - if a.cfg.AutoApprove { - env["OPENCODE_PERMISSION"] = `{"question":"deny"}` - } else { - env["OPENCODE_PERMISSION"] = `{"edit":"ask","bash":"ask","webfetch":"ask","doom_loop":"ask","external_directory":"ask","question":"deny"}` - } + // Set up permissions based on config with per-permission rules + env["OPENCODE_PERMISSION"] = buildPermissionJSON(a.cfg.AutoApprove) + + // Inject runtime config for auto-compaction + env["OPENCODE_CONFIG_CONTENT"] = `{"compaction":{"auto":true}}` return env, nil } +// buildPermissionJSON builds the OPENCODE_PERMISSION JSON value. +func buildPermissionJSON(autoApprove bool) string { + if autoApprove { + return `{"rules":[{"permission":"bash","grant":"always"},{"permission":"write","grant":"always"},{"permission":"read","grant":"always"},{"permission":"edit","grant":"always"},{"permission":"glob","grant":"always"},{"permission":"grep","grant":"always"},{"permission":"webfetch","grant":"always"}],"question":"deny"}` + } + return `{"rules":[{"permission":"bash","grant":"ask"},{"permission":"write","grant":"ask"},{"permission":"read","grant":"ask"},{"permission":"edit","grant":"ask"}],"question":"deny"}` +} + // PrepareCommandArgs returns extra command-line arguments for the agent process. // For OpenCode, no extra args are needed - MCP is configured via config file. func (a *Adapter) PrepareCommandArgs() []string { @@ -248,6 +266,9 @@ func (a *Adapter) Initialize(ctx context.Context) error { Version: "unknown", } + // Fetch providers and commands in the background — don't block initialization + go a.fetchPostInitData() + a.logger.Info("OpenCode adapter initialized") return nil @@ -325,12 +346,13 @@ func (a *Adapter) NewSession(ctx context.Context, _ []types.McpServer) (string, a.logger.Info("created new session", zap.String("session_id", sessionID)) - // Emit session status event for resume token storage + // Emit session status event a.sendUpdate(AgentEvent{ - Type: streams.EventTypeSessionStatus, - SessionID: sessionID, + Type: streams.EventTypeSessionStatus, + SessionID: sessionID, + SessionStatus: streams.SessionStatusNew, Data: map[string]any{ - "session_status": "active", + "session_status": streams.SessionStatusNew, "init": true, }, }) @@ -357,6 +379,17 @@ func (a *Adapter) LoadSession(ctx context.Context, sessionID string) error { zap.String("original_session_id", sessionID), zap.String("new_session_id", newSessionID)) + // Emit session status event for resumed session + a.sendUpdate(AgentEvent{ + Type: streams.EventTypeSessionStatus, + SessionID: newSessionID, + SessionStatus: streams.SessionStatusResumed, + Data: map[string]any{ + "session_status": streams.SessionStatusResumed, + "init": true, + }, + }) + return nil } @@ -397,15 +430,17 @@ func (a *Adapter) Prompt(ctx context.Context, message string, _ []v1.MessageAtta a.logger.Warn("failed to start event stream", zap.Error(err)) } - // Send prompt - // TODO: Support model selection from config - if err := client.SendPrompt(ctx, sessionID, message, nil, "", ""); err != nil { + // Send prompt with model selection (blocks until OpenCode finishes) + if err := client.SendPrompt(ctx, sessionID, message, a.model, "", ""); err != nil { a.mu.Lock() a.resultCh = nil a.mu.Unlock() return fmt.Errorf("failed to send prompt: %w", err) } + a.logger.Info("prompt HTTP request completed, waiting for SSE idle signal", + zap.String("session_id", sessionID)) + // Wait for result via control channel or context cancellation controlCh := client.ControlChannel() @@ -617,6 +652,17 @@ func (a *Adapter) handleSDKEvent(event *opencode.SDKEventEnvelope) { case opencode.SDKEventPermissionAsked: a.handlePermissionAsked(event.Properties, sessionID) + case opencode.SDKEventSessionCompacted: + a.handleSessionCompacted(event.Properties, sessionID) + + case opencode.SDKEventTodoUpdated: + // Log but don't emit — todo tracking is informational + a.logger.Debug("todo.updated event received") + + case opencode.SDKEventMessageRemoved, opencode.SDKEventMessagePartRemoved: + // Log message/part removal — these occur during compaction + a.logger.Debug("message removal event", zap.String("event_type", event.Type)) + case opencode.SDKEventSessionError: props, err := opencode.ParseSessionError(event.Properties) if err != nil { @@ -872,6 +918,27 @@ func deriveToolTitle(toolName string, input map[string]any) string { return "" } +// handleSessionCompacted processes session.compacted events from auto-compaction. +func (a *Adapter) handleSessionCompacted(props json.RawMessage, sessionID string) { + a.logger.Info("session compacted", zap.String("session_id", sessionID)) + + // Reset token tracking since compaction changes the context window usage + a.mu.Lock() + a.totalTokens = 0 + // Clear text part state since message IDs may have changed + a.textParts = make(map[string]*textPartState) + a.messageRoles = make(map[string]string) + a.mu.Unlock() + + // Emit a context window update showing reset usage + a.sendUpdate(AgentEvent{ + Type: streams.EventTypeContextWindow, + SessionID: sessionID, + ContextWindowUsed: 0, + ContextWindowSize: int64(a.modelContextWindow), + }) +} + // handleMessageUpdated processes message.updated events (for token tracking) func (a *Adapter) handleMessageUpdated(props json.RawMessage, sessionID string) { parsed, err := opencode.ParseMessageUpdated(props) @@ -996,6 +1063,20 @@ func (a *Adapter) handlePermissionAsked(props json.RawMessage, sessionID string) } } +// parseModelSpec parses a "providerID/modelID" string into a ModelSpec. +// If no "/" is found, the whole string is used as modelID with empty providerID. +func parseModelSpec(model string) *opencode.ModelSpec { + if idx := strings.Index(model, "/"); idx > 0 { + return &opencode.ModelSpec{ + ProviderID: model[:idx], + ModelID: model[idx+1:], + } + } + return &opencode.ModelSpec{ + ModelID: model, + } +} + // formatPermissionTitle builds a display title for a permission request. func formatPermissionTitle(permission string, metadata map[string]any) string { if metadata != nil { @@ -1028,6 +1109,87 @@ func (a *Adapter) RespondToPermission(ctx context.Context, requestID string, app return a.client.ReplyPermission(ctx, requestID, reply, msg) } +// fetchPostInitData fetches providers and commands after initialization. +// Runs in a goroutine with a timeout so it never blocks the main flow. +func (a *Adapter) fetchPostInitData() { + ctx, cancel := context.WithTimeout(a.ctx, 10*time.Second) + defer cancel() + + a.fetchContextWindow(ctx) + a.discoverCommands(ctx) +} + +// fetchContextWindow queries the /provider endpoint and updates the model context window. +func (a *Adapter) fetchContextWindow(ctx context.Context) { + providers, err := a.client.ListProviders(ctx) + if err != nil { + a.logger.Debug("failed to fetch providers for context window", zap.Error(err)) + return + } + + // Look up the context window for our configured model + a.mu.RLock() + model := a.model + a.mu.RUnlock() + + if model == nil { + return + } + + for _, p := range providers.All { + if p.ID != model.ProviderID { + continue + } + if modelInfo, ok := p.Models[model.ModelID]; ok && modelInfo.Limit.Context > 0 { + a.mu.Lock() + a.modelContextWindow = modelInfo.Limit.Context + a.mu.Unlock() + a.logger.Info("updated context window from provider", + zap.String("provider", model.ProviderID), + zap.String("model", model.ModelID), + zap.Int("context_window", modelInfo.Limit.Context)) + return + } + } +} + +// discoverCommands queries the /command endpoint and emits available commands. +func (a *Adapter) discoverCommands(ctx context.Context) { + commands, err := a.client.ListCommands(ctx) + if err != nil { + a.logger.Debug("failed to discover commands", zap.Error(err)) + return + } + + if len(commands) == 0 { + return + } + + a.mu.RLock() + sessionID := a.sessionID + a.mu.RUnlock() + + available := make([]streams.AvailableCommand, 0, len(commands)) + for _, cmd := range commands { + ac := streams.AvailableCommand{ + Name: cmd.Name, + Description: cmd.Description, + } + if cmd.Input != nil { + ac.InputHint = cmd.Input.Hint + } + available = append(available, ac) + } + + a.sendUpdate(AgentEvent{ + Type: streams.EventTypeAvailableCommands, + SessionID: sessionID, + AvailableCommands: available, + }) + + a.logger.Info("discovered slash commands", zap.Int("count", len(available))) +} + // openCodeConfigPath returns the path to the OpenCode config file. func openCodeConfigPath(homeDir string) (string, string, error) { if homeDir == "" { diff --git a/apps/backend/pkg/opencode/client.go b/apps/backend/pkg/opencode/client.go index 9dd96f6c18..c6aa580e58 100644 --- a/apps/backend/pkg/opencode/client.go +++ b/apps/backend/pkg/opencode/client.go @@ -18,6 +18,13 @@ import ( "github.com/kandev/kandev/internal/common/logger" ) +// SSE reconnection constants +const ( + sseReconnectBaseDelay = 3 * time.Second + sseReconnectMaxDelay = 30 * time.Second + sseMaxRetries = 20 +) + // Client manages HTTP communication with OpenCode server type Client struct { baseURL string @@ -31,8 +38,9 @@ type Client struct { controlCh chan ControlEvent // SSE connection tracking - prevents multiple concurrent connections - sseCancel context.CancelFunc - sseActive bool + sseCancel context.CancelFunc + sseActive bool + lastEventID string // Last-Event-ID for reconnection mu sync.RWMutex closed bool @@ -245,7 +253,8 @@ func (c *Client) ForkSession(ctx context.Context, sessionID string) (string, err return session.ID, nil } -// SendPrompt sends a prompt to the session +// SendPrompt sends a prompt to the session. +// This call blocks until OpenCode finishes processing the prompt. func (c *Client) SendPrompt(ctx context.Context, sessionID, prompt string, model *ModelSpec, agent, variant string) error { req := PromptRequest{ Model: model, @@ -261,25 +270,30 @@ func (c *Client) SendPrompt(ctx context.Context, sessionID, prompt string, model return fmt.Errorf("marshal prompt request: %w", err) } + c.logger.Debug("sending prompt HTTP request", + zap.String("session_id", sessionID), + zap.Int("body_len", len(body))) + path := fmt.Sprintf("/session/%s/message", sessionID) - // Use a dedicated client with long timeout for prompts - they can take minutes resp, err := c.doPromptRequest(ctx, http.MethodPost, path, strings.NewReader(string(body))) if err != nil { return fmt.Errorf("send prompt request: %w", err) } defer func() { _ = resp.Body.Close() }() - // Read and validate response respBody, err := io.ReadAll(resp.Body) if err != nil { return fmt.Errorf("read prompt response: %w", err) } + c.logger.Debug("prompt HTTP response received", + zap.Int("status", resp.StatusCode), + zap.Int("body_len", len(respBody))) + if resp.StatusCode != http.StatusOK { return fmt.Errorf("prompt failed: HTTP %d: %s", resp.StatusCode, string(respBody)) } - // Validate response structure trimmed := strings.TrimSpace(string(respBody)) if trimmed == "" { return fmt.Errorf("prompt returned empty response") @@ -290,14 +304,12 @@ func (c *Client) SendPrompt(ctx context.Context, sessionID, prompt string, model return fmt.Errorf("parse prompt response: %w", err) } - // Check for success response: { info, parts } if _, hasInfo := parsed["info"]; hasInfo { if _, hasParts := parsed["parts"]; hasParts { return nil } } - // Check for error response: { name, data } if name, ok := parsed["name"].(string); ok { message := "unknown error" if data, ok := parsed["data"].(map[string]any); ok { @@ -359,12 +371,101 @@ func (c *Client) ReplyPermission(ctx context.Context, requestID, reply string, m return nil } +// Summarize triggers compaction/summarization for a session. +func (c *Client) Summarize(ctx context.Context, sessionID string, model *ModelSpec) error { + payload := map[string]any{} + if model != nil { + payload["model"] = model + } + + body, err := json.Marshal(payload) + if err != nil { + return fmt.Errorf("marshal summarize request: %w", err) + } + + path := fmt.Sprintf("/session/%s/summarize", sessionID) + resp, err := c.doPromptRequest(ctx, http.MethodPost, path, strings.NewReader(string(body))) + if err != nil { + return fmt.Errorf("summarize request: %w", err) + } + defer func() { _ = resp.Body.Close() }() + + if resp.StatusCode != http.StatusOK { + respBody, _ := io.ReadAll(resp.Body) + return fmt.Errorf("summarize failed: HTTP %d: %s", resp.StatusCode, string(respBody)) + } + _, _ = io.ReadAll(resp.Body) + return nil +} + +// ListProviders fetches provider information including model context windows. +func (c *Client) ListProviders(ctx context.Context) (*ProviderListResponse, error) { + resp, err := c.doRequest(ctx, http.MethodGet, "/provider", nil) + if err != nil { + return nil, fmt.Errorf("list providers request: %w", err) + } + defer func() { _ = resp.Body.Close() }() + + if resp.StatusCode != http.StatusOK { + body, _ := io.ReadAll(resp.Body) + return nil, fmt.Errorf("list providers failed: HTTP %d: %s", resp.StatusCode, string(body)) + } + + var result ProviderListResponse + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return nil, fmt.Errorf("parse providers response: %w", err) + } + return &result, nil +} + +// ListCommands fetches available slash commands from the server. +func (c *Client) ListCommands(ctx context.Context) ([]CommandInfo, error) { + resp, err := c.doRequest(ctx, http.MethodGet, "/command", nil) + if err != nil { + return nil, fmt.Errorf("list commands request: %w", err) + } + defer func() { _ = resp.Body.Close() }() + + if resp.StatusCode != http.StatusOK { + body, _ := io.ReadAll(resp.Body) + return nil, fmt.Errorf("list commands failed: HTTP %d: %s", resp.StatusCode, string(body)) + } + + var result []CommandInfo + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return nil, fmt.Errorf("parse commands response: %w", err) + } + return result, nil +} + +// RunSessionCommand executes a slash command in a session. +func (c *Client) RunSessionCommand(ctx context.Context, sessionID, command string) error { + payload := map[string]string{"command": command} + body, err := json.Marshal(payload) + if err != nil { + return fmt.Errorf("marshal command request: %w", err) + } + + path := fmt.Sprintf("/session/%s/command", sessionID) + resp, err := c.doPromptRequest(ctx, http.MethodPost, path, strings.NewReader(string(body))) + if err != nil { + return fmt.Errorf("run command request: %w", err) + } + defer func() { _ = resp.Body.Close() }() + + if resp.StatusCode != http.StatusOK { + respBody, _ := io.ReadAll(resp.Body) + return fmt.Errorf("run command failed: HTTP %d: %s", resp.StatusCode, string(respBody)) + } + _, _ = io.ReadAll(resp.Body) + return nil +} + // StartEventStream starts the SSE event stream and processes events. // It ensures only one SSE connection is active at a time to prevent duplicate events. +// Supports automatic reconnection with Last-Event-ID for event resumption. func (c *Client) StartEventStream(ctx context.Context, sessionID string) error { c.mu.Lock() - // Check if already connected - prevent multiple concurrent SSE connections - // which would cause duplicate event processing if c.sseActive { c.mu.Unlock() c.logger.Debug("SSE stream already active, skipping duplicate connection", @@ -374,127 +475,173 @@ func (c *Client) StartEventStream(ctx context.Context, sessionID string) error { c.sseActive = true c.mu.Unlock() - url := c.baseURL + "/event?directory=" + c.directory - - // Create a cancellable context for this SSE connection sseCtx, sseCancel := context.WithCancel(ctx) - // Store the cancel function so we can close this connection later if needed c.mu.Lock() c.sseCancel = sseCancel c.mu.Unlock() - req, err := http.NewRequestWithContext(sseCtx, http.MethodGet, url, nil) - if err != nil { + // Connect and start processing with reconnect support + if err := c.connectSSE(sseCtx, sessionID); err != nil { c.mu.Lock() c.sseActive = false c.sseCancel = nil c.mu.Unlock() sseCancel() - return fmt.Errorf("create event stream request: %w", err) + return err + } + + return nil +} + +// connectSSE establishes the SSE connection and starts the event loop. +func (c *Client) connectSSE(ctx context.Context, sessionID string) error { + resp, err := c.dialSSE(ctx) + if err != nil { + return err + } + c.logger.Debug("SSE stream connected", zap.String("session_id", sessionID)) + go c.sseEventLoop(ctx, sessionID, resp.Body) + return nil +} + +// dialSSE creates and executes an SSE HTTP request, including Last-Event-ID if available. +func (c *Client) dialSSE(ctx context.Context) (*http.Response, error) { + url := c.baseURL + "/event?directory=" + c.directory + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + if err != nil { + return nil, fmt.Errorf("create event stream request: %w", err) } req.Header.Set("Authorization", c.buildAuthHeader()) req.Header.Set("X-OpenCode-Directory", c.directory) req.Header.Set("Accept", "text/event-stream") - // Use a client without timeout for SSE + c.mu.RLock() + lastID := c.lastEventID + c.mu.RUnlock() + if lastID != "" { + req.Header.Set("Last-Event-ID", lastID) + } + sseClient := &http.Client{} resp, err := sseClient.Do(req) if err != nil { - c.mu.Lock() - c.sseActive = false - c.sseCancel = nil - c.mu.Unlock() - sseCancel() - return fmt.Errorf("connect event stream: %w", err) + return nil, fmt.Errorf("connect event stream: %w", err) } if resp.StatusCode != http.StatusOK { body, _ := io.ReadAll(resp.Body) _ = resp.Body.Close() - c.mu.Lock() - c.sseActive = false - c.sseCancel = nil - c.mu.Unlock() - sseCancel() - return fmt.Errorf("event stream failed: HTTP %d: %s", resp.StatusCode, string(body)) + return nil, fmt.Errorf("event stream failed: HTTP %d: %s", resp.StatusCode, string(body)) } - c.logger.Debug("SSE stream connected", - zap.String("session_id", sessionID)) - - // Process events in background - go c.processEventStream(sseCtx, sessionID, resp.Body) - - return nil + return resp, nil } -// processEventStream reads and processes SSE events -func (c *Client) processEventStream(ctx context.Context, sessionID string, body io.ReadCloser) { +// sseEventLoop reads SSE events, dispatches them, and reconnects on disconnect. +func (c *Client) sseEventLoop(ctx context.Context, sessionID string, body io.ReadCloser) { defer func() { - _ = body.Close() - // Mark SSE as inactive when stream ends c.mu.Lock() c.sseActive = false c.sseCancel = nil c.mu.Unlock() - c.logger.Debug("SSE stream ended", zap.String("session_id", sessionID)) }() + for attempt := 0; ; attempt++ { + disconnectReason := c.readSSEStream(ctx, sessionID, body) + _ = body.Close() + + if disconnectReason == sseDisconnectCancelled || disconnectReason == sseDisconnectClosed { + return + } + + // Attempt reconnection with exponential backoff + if attempt >= sseMaxRetries { + c.logger.Warn("SSE max retries reached, giving up", + zap.Int("retries", attempt)) + c.sendControlEvent("disconnected") + return + } + + delay := sseBackoff(attempt) + c.logger.Info("SSE stream disconnected, reconnecting", + zap.Int("attempt", attempt+1), + zap.Duration("delay", delay)) + + select { + case <-ctx.Done(): + return + case <-time.After(delay): + } + + resp, err := c.dialSSE(ctx) + if err != nil { + c.logger.Warn("SSE reconnect failed", zap.Error(err), zap.Int("attempt", attempt+1)) + continue + } + + body = resp.Body + c.logger.Info("SSE stream reconnected", zap.Int("attempt", attempt+1)) + } +} + +// sseDisconnectReason indicates why the SSE stream ended. +type sseDisconnectReason int + +const ( + sseDisconnectError sseDisconnectReason = iota // network/scan error, should reconnect + sseDisconnectCancelled // context cancelled, no reconnect + sseDisconnectClosed // client closed, no reconnect +) + +// readSSEStream reads events from a single SSE connection until it ends. +func (c *Client) readSSEStream(ctx context.Context, sessionID string, body io.Reader) sseDisconnectReason { scanner := bufio.NewScanner(body) - // Increase buffer size for potentially large events buf := make([]byte, 0, 64*1024) scanner.Buffer(buf, 1024*1024) var dataBuffer strings.Builder + var currentID string for scanner.Scan() { select { case <-ctx.Done(): - return + return sseDisconnectCancelled default: } line := scanner.Text() - // SSE format: "data: {...}" + // Track event ID for reconnection + if strings.HasPrefix(line, "id: ") { + currentID = strings.TrimPrefix(line, "id: ") + continue + } + if strings.HasPrefix(line, "data: ") { dataBuffer.WriteString(strings.TrimPrefix(line, "data: ")) continue } - // Empty line signals end of event if line == "" && dataBuffer.Len() > 0 { data := strings.TrimSpace(dataBuffer.String()) dataBuffer.Reset() + // Update Last-Event-ID + if currentID != "" { + c.mu.Lock() + c.lastEventID = currentID + c.mu.Unlock() + currentID = "" + } + if data == "" { continue } - event, err := ParseSDKEvent([]byte(data)) - if err != nil { - c.logger.Warn("failed to parse SDK event", zap.Error(err)) - continue - } - - // Filter events for this session - if !c.eventMatchesSession(event, sessionID) { - continue - } - - // Process control events - c.processControlEvent(event, sessionID) - - // Call event handler - c.mu.RLock() - handler := c.eventHandler - c.mu.RUnlock() - - if handler != nil { - handler(event) - } + c.dispatchSSEEvent(data, sessionID) } } @@ -502,18 +649,69 @@ func (c *Client) processEventStream(ctx context.Context, sessionID string, body c.logger.Error("event stream error", zap.Error(err)) } - // Notify disconnection (check if closed first to avoid panic) + c.mu.RLock() + closed := c.closed + c.mu.RUnlock() + if closed { + return sseDisconnectClosed + } + return sseDisconnectError +} + +// dispatchSSEEvent parses and dispatches a single SSE event. +func (c *Client) dispatchSSEEvent(data, sessionID string) { + event, err := ParseSDKEvent([]byte(data)) + if err != nil { + c.logger.Warn("failed to parse SDK event", zap.Error(err)) + return + } + + c.logger.Debug("SSE event received", + zap.String("type", event.Type), + zap.String("session_id", sessionID)) + + if !c.eventMatchesSession(event, sessionID) { + c.logger.Debug("SSE event filtered (session mismatch)", zap.String("type", event.Type)) + return + } + + c.processControlEvent(event, sessionID) + + c.mu.RLock() + handler := c.eventHandler + c.mu.RUnlock() + + if handler != nil { + handler(event) + } +} + +// sendControlEvent sends a control event if the client is not closed. +func (c *Client) sendControlEvent(eventType string) { c.mu.RLock() closed := c.closed c.mu.RUnlock() if !closed { select { - case c.controlCh <- ControlEvent{Type: "disconnected"}: + case c.controlCh <- ControlEvent{Type: eventType}: default: } } } +// sseBackoff computes exponential backoff delay for SSE reconnection. +func sseBackoff(attempt int) time.Duration { + delay := sseReconnectBaseDelay + for i := 0; i < attempt; i++ { + delay *= 2 + if delay > sseReconnectMaxDelay { + delay = sseReconnectMaxDelay + break + } + } + return delay +} + // eventMatchesSession checks if an event belongs to the specified session func (c *Client) eventMatchesSession(event *SDKEventEnvelope, sessionID string) bool { // Try to extract sessionID from properties diff --git a/apps/backend/pkg/opencode/types.go b/apps/backend/pkg/opencode/types.go index ade4736777..5dcce152a9 100644 --- a/apps/backend/pkg/opencode/types.go +++ b/apps/backend/pkg/opencode/types.go @@ -253,6 +253,19 @@ type SDKTodo struct { Priority string `json:"priority"` } +// CommandInfo represents a slash command from GET /command. +type CommandInfo struct { + Name string `json:"name"` + Description string `json:"description,omitempty"` + // Some commands have structured input + Input *CommandInputInfo `json:"input,omitempty"` +} + +// CommandInputInfo describes command input expectations. +type CommandInputInfo struct { + Hint string `json:"hint,omitempty"` +} + // ProviderListResponse from GET /provider type ProviderListResponse struct { All []ProviderInfo `json:"all"` @@ -274,6 +287,29 @@ type ProviderModelLimit struct { Context int `json:"context"` } +// SessionCompactedProperties for session.compacted events +type SessionCompactedProperties struct { + SessionID string `json:"sessionID"` +} + +// ParseSessionCompacted parses session.compacted properties +func ParseSessionCompacted(data json.RawMessage) (*SessionCompactedProperties, error) { + var props SessionCompactedProperties + if err := json.Unmarshal(data, &props); err != nil { + return nil, err + } + return &props, nil +} + +// ParseTodoUpdated parses todo.updated properties +func ParseTodoUpdated(data json.RawMessage) (*TodoUpdatedProperties, error) { + var props TodoUpdatedProperties + if err := json.Unmarshal(data, &props); err != nil { + return nil, err + } + return &props, nil +} + // ParseSDKEvent parses an SDK event from JSON func ParseSDKEvent(data []byte) (*SDKEventEnvelope, error) { var env SDKEventEnvelope