diff --git a/apps/backend/internal/agent/lifecycle/manager.go b/apps/backend/internal/agent/lifecycle/manager.go index 40a62f4cff..08151b4993 100644 --- a/apps/backend/internal/agent/lifecycle/manager.go +++ b/apps/backend/internal/agent/lifecycle/manager.go @@ -1235,11 +1235,6 @@ func (m *Manager) emitSessionStatusEvent(execution *AgentExecution, agentConfig }) } -// handlePermissionRequestEvent processes permission requests from the unified agent event stream -func (m *Manager) handlePermissionRequestEvent(execution *AgentExecution, event agentctl.AgentEvent) { - m.eventPublisher.PublishPermissionRequest(execution, event) -} - // handleAgentEvent processes incoming agent events from the agent func (m *Manager) handleAgentEvent(execution *AgentExecution, event agentctl.AgentEvent) { // Log all incoming events for debugging @@ -1284,6 +1279,8 @@ func (m *Manager) handleAgentEvent(execution *AgentExecution, event agentctl.Age execution.messageMu.Unlock() } } + // Return early - message_chunk is transformed to message_streaming, no need to publish raw event + return case "reasoning": // Stream thinking content like message chunks for real-time feedback @@ -1313,6 +1310,8 @@ func (m *Manager) handleAgentEvent(execution *AgentExecution, event agentctl.Age execution.messageMu.Unlock() } } + // Return early - reasoning is transformed to thinking_streaming, no need to publish raw event + return case "tool_call": // Tool call starting marks a step boundary - flush the accumulated message @@ -1443,8 +1442,10 @@ func (m *Manager) handleAgentEvent(execution *AgentExecution, event agentctl.Age zap.String("pending_id", event.PendingID), zap.String("title", event.PermissionTitle)) - // Handle permission request inline - m.handlePermissionRequestEvent(execution, event) + // Publish permission request to dedicated subject for orchestrator to handle + m.eventPublisher.PublishPermissionRequest(execution, event) + // Return early - permission requests don't need to be published as stream events + return case "context_window": m.logger.Debug("context window update received", diff --git a/apps/backend/internal/agentctl/server/adapter/adapter.go b/apps/backend/internal/agentctl/server/adapter/adapter.go index 07671e75d0..ffb155e123 100644 --- a/apps/backend/internal/agentctl/server/adapter/adapter.go +++ b/apps/backend/internal/agentctl/server/adapter/adapter.go @@ -36,16 +36,17 @@ type ( // Re-export agent event type constants from streams package. const ( - EventTypeMessageChunk = streams.EventTypeMessageChunk - EventTypeReasoning = streams.EventTypeReasoning - EventTypeToolCall = streams.EventTypeToolCall - EventTypeToolUpdate = streams.EventTypeToolUpdate - EventTypePlan = streams.EventTypePlan - EventTypeComplete = streams.EventTypeComplete - EventTypeError = streams.EventTypeError - EventTypePermissionRequest = streams.EventTypePermissionRequest - EventTypeContextWindow = streams.EventTypeContextWindow - EventTypeSessionStatus = streams.EventTypeSessionStatus + EventTypeMessageChunk = streams.EventTypeMessageChunk + EventTypeReasoning = streams.EventTypeReasoning + EventTypeToolCall = streams.EventTypeToolCall + EventTypeToolUpdate = streams.EventTypeToolUpdate + EventTypePlan = streams.EventTypePlan + EventTypeComplete = streams.EventTypeComplete + EventTypeError = streams.EventTypeError + EventTypePermissionRequest = streams.EventTypePermissionRequest + EventTypePermissionCancelled = streams.EventTypePermissionCancelled + EventTypeContextWindow = streams.EventTypeContextWindow + EventTypeSessionStatus = streams.EventTypeSessionStatus ) // StderrProvider provides access to recent stderr output for error context. 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 a8704fec7f..e734a13820 100644 --- a/apps/backend/internal/agentctl/server/adapter/transport/opencode/adapter.go +++ b/apps/backend/internal/agentctl/server/adapter/transport/opencode/adapter.go @@ -902,22 +902,14 @@ func (a *Adapter) handlePermissionAsked(props json.RawMessage, sessionID string) } } - // Emit permission request event - a.sendUpdate(AgentEvent{ - Type: streams.EventTypePermissionRequest, - SessionID: sessionID, - PendingID: requestID, - ToolCallID: toolCallID, - PermissionTitle: title, - PermissionOptions: []PermissionOption{ - {OptionID: "allow", Name: "Allow", Kind: "allow_once"}, - {OptionID: "deny", Name: "Deny", Kind: "reject_once"}, - }, - ActionType: permission, - ActionDetails: parsed.Metadata, - }) - - // If we have a handler, call it + // If we have a handler, call it to get user approval. + // The handler (process manager's handlePermissionRequest) will: + // 1. Send the permission_request notification to the frontend + // 2. Block waiting for user response + // 3. Return the response + // We pass PendingID so the handler uses OpenCode's ID (e.g., "per_xxx") + // instead of generating a new one - this ensures the frontend and backend + // use the same ID for response lookup. if handler != nil { go func() { req := &PermissionRequest{ @@ -926,6 +918,7 @@ func (a *Adapter) handlePermissionAsked(props json.RawMessage, sessionID string) Title: title, ActionType: permission, ActionDetails: parsed.Metadata, + PendingID: requestID, // Use OpenCode's ID so response lookup works Options: []PermissionOption{ {OptionID: "allow", Name: "Allow", Kind: "allow_once"}, {OptionID: "deny", Name: "Deny", Kind: "reject_once"}, diff --git a/apps/backend/internal/agentctl/server/adapter/transport/streamjson/adapter.go b/apps/backend/internal/agentctl/server/adapter/transport/streamjson/adapter.go index c455996b2b..92062c7e9d 100644 --- a/apps/backend/internal/agentctl/server/adapter/transport/streamjson/adapter.go +++ b/apps/backend/internal/agentctl/server/adapter/transport/streamjson/adapter.go @@ -394,7 +394,6 @@ func (a *Adapter) handleToolPermission(requestID string, req *claudecode.Control a.mu.RLock() handler := a.permissionHandler sessionID := a.sessionID - operationID := a.operationID a.mu.RUnlock() // Determine action type based on tool name @@ -425,6 +424,14 @@ func (a *Adapter) handleToolPermission(requestID string, req *claudecode.Control {OptionID: "deny", Name: "Deny", Kind: "reject_once"}, } + // Build permission request with Claude Code's requestID. + // The handler (process manager's handlePermissionRequest) will: + // 1. Send the permission_request notification to the frontend + // 2. Block waiting for user response + // 3. Return the response + // We pass PendingID so the handler uses Claude Code's requestID + // instead of generating a new one - this ensures the frontend and backend + // use the same ID for response lookup. permReq := &PermissionRequest{ SessionID: sessionID, ToolCallID: req.ToolUseID, @@ -432,6 +439,7 @@ func (a *Adapter) handleToolPermission(requestID string, req *claudecode.Control Options: options, ActionType: actionType, ActionDetails: req.Input, + PendingID: requestID, // Use Claude Code's requestID so response lookup works } // If no handler, auto-allow @@ -442,20 +450,7 @@ func (a *Adapter) handleToolPermission(requestID string, req *claudecode.Control return } - // Send permission notification through updates channel - a.sendUpdate(AgentEvent{ - Type: streams.EventTypePermissionRequest, - SessionID: sessionID, - OperationID: operationID, - ToolCallID: req.ToolUseID, - PendingID: requestID, - PermissionTitle: title, - PermissionOptions: options, - ActionType: actionType, - ActionDetails: req.Input, - }) - - // Call permission handler (blocking) + // Call permission handler (blocking) - it will send the notification ctx := context.Background() resp, err := handler(ctx, permReq) if err != nil { diff --git a/apps/backend/internal/agentctl/server/process/manager.go b/apps/backend/internal/agentctl/server/process/manager.go index 4f5e1df74e..30153b810f 100644 --- a/apps/backend/internal/agentctl/server/process/manager.go +++ b/apps/backend/internal/agentctl/server/process/manager.go @@ -711,8 +711,14 @@ func (m *Manager) GetProcessInfo() map[string]interface{} { // handlePermissionRequest handles permission requests from the agent // It stores the pending request and waits for a response from the backend func (m *Manager) handlePermissionRequest(ctx context.Context, req *adapter.PermissionRequest) (*adapter.PermissionResponse, error) { - // Generate a unique ID for this permission request - pendingID := fmt.Sprintf("%s-%s-%d", req.SessionID, req.ToolCallID, time.Now().UnixNano()) + // Use the adapter-provided pending ID if available, otherwise generate one. + // This ensures the ID sent to the frontend matches the one used for response lookup. + // For OpenCode (per_xxx) and Claude Code (requestID), the adapter passes its ID + // so we use the same ID throughout the permission flow. + pendingID := req.PendingID + if pendingID == "" { + pendingID = fmt.Sprintf("%s-%s-%d", req.SessionID, req.ToolCallID, time.Now().UnixNano()) + } m.logger.Info("handling permission request", zap.String("pending_id", pendingID), @@ -747,7 +753,6 @@ func (m *Manager) handlePermissionRequest(ctx context.Context, req *adapter.Perm }() // Send notification to backend via updates channel - // We use a custom notification type that the backend will recognize m.sendPermissionNotification(pending) // Wait for response indefinitely - user may close and reopen the page @@ -761,6 +766,8 @@ func (m *Manager) handlePermissionRequest(ctx context.Context, req *adapter.Perm case <-ctx.Done(): m.logger.Warn("permission request context cancelled", zap.String("pending_id", pendingID)) + // Send cancellation notification so the backend can update the permission message status + m.sendPermissionCancelledNotification(pending) return &adapter.PermissionResponse{Cancelled: true}, nil } } @@ -831,6 +838,29 @@ func (m *Manager) sendPermissionNotification(pending *PendingPermission) { } } +// sendPermissionCancelledNotification sends a notification that a permission request was cancelled. +// This happens when the context is cancelled (e.g., agent completes or user stops the task) +// before the user responds to the permission request. +func (m *Manager) sendPermissionCancelledNotification(pending *PendingPermission) { + event := adapter.AgentEvent{ + Type: adapter.EventTypePermissionCancelled, + SessionID: pending.Request.SessionID, + PendingID: pending.ID, + } + + m.logger.Info("sending permission cancelled notification", + zap.String("pending_id", pending.ID), + zap.String("session_id", pending.Request.SessionID)) + + select { + case m.updatesCh <- event: + // Sent successfully + default: + m.logger.Warn("updates channel full, dropping permission cancelled notification", + zap.String("pending_id", pending.ID)) + } +} + // RespondToPermission responds to a pending permission request func (m *Manager) RespondToPermission(pendingID string, optionID string, cancelled bool) error { m.permissionMu.RLock() diff --git a/apps/backend/internal/agentctl/server/process/procattr_linux.go b/apps/backend/internal/agentctl/server/process/procattr_linux.go new file mode 100644 index 0000000000..b4a49be881 --- /dev/null +++ b/apps/backend/internal/agentctl/server/process/procattr_linux.go @@ -0,0 +1,27 @@ +//go:build linux + +package process + +import ( + "os/exec" + "syscall" +) + +// setProcGroup configures the command to run in its own process group. +// This allows us to kill all child processes together. +// On Linux, we also set Pdeathsig to ensure the child is killed if the parent dies +// unexpectedly (SIGKILL, crash, etc.) without calling Stop(). +func setProcGroup(cmd *exec.Cmd) { + cmd.SysProcAttr = &syscall.SysProcAttr{ + Setpgid: true, + Pdeathsig: syscall.SIGTERM, + } +} + +// killProcessGroup kills the entire process group for the given PID. +// Returns nil if successful, or an error if the kill failed. +func killProcessGroup(pid int) error { + // Kill the entire process group by using negative PID + return syscall.Kill(-pid, syscall.SIGKILL) +} + diff --git a/apps/backend/internal/agentctl/server/process/procattr_unix.go b/apps/backend/internal/agentctl/server/process/procattr_unix.go index 7fdc90d8d5..29d1aadd22 100644 --- a/apps/backend/internal/agentctl/server/process/procattr_unix.go +++ b/apps/backend/internal/agentctl/server/process/procattr_unix.go @@ -1,4 +1,4 @@ -//go:build unix +//go:build unix && !linux package process @@ -9,6 +9,8 @@ import ( // setProcGroup configures the command to run in its own process group. // This allows us to kill all child processes together. +// Note: Pdeathsig is Linux-specific and not available on macOS/BSD. +// On these platforms, orphan cleanup relies on explicit Stop() calls. func setProcGroup(cmd *exec.Cmd) { cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} } diff --git a/apps/backend/internal/agentctl/types/streams/agent.go b/apps/backend/internal/agentctl/types/streams/agent.go index 9d2726853a..4076a72c8f 100644 --- a/apps/backend/internal/agentctl/types/streams/agent.go +++ b/apps/backend/internal/agentctl/types/streams/agent.go @@ -26,6 +26,11 @@ const ( // EventTypePermissionRequest indicates the agent is requesting permission. EventTypePermissionRequest = "permission_request" + // EventTypePermissionCancelled indicates a permission request was cancelled. + // This happens when the agent completes or the context is cancelled before + // the user responds to the permission request. + EventTypePermissionCancelled = "permission_cancelled" + // EventTypeSessionStatus indicates a session status update (resumed or new). EventTypeSessionStatus = "session_status" diff --git a/apps/backend/internal/agentctl/types/types.go b/apps/backend/internal/agentctl/types/types.go index 09f20b8b80..7bf86bb115 100644 --- a/apps/backend/internal/agentctl/types/types.go +++ b/apps/backend/internal/agentctl/types/types.go @@ -107,6 +107,13 @@ type PermissionRequest struct { Title string `json:"title"` Options []PermissionOption `json:"options"` + // PendingID is the unique identifier for this permission request. + // If set by the adapter (e.g., OpenCode's "per_xxx" ID or Claude Code's requestID), + // the process manager will use this ID instead of generating a new one. + // This ensures the frontend and backend use the same ID for permission response lookups. + // If empty, the process manager will generate a unique ID. + PendingID string `json:"pending_id,omitempty"` + // ActionType categorizes the action requiring approval. // Use ActionType* constants: "command", "file_write", "network", "mcp_tool", etc. ActionType string `json:"action_type,omitempty"` diff --git a/apps/backend/pkg/opencode/client.go b/apps/backend/pkg/opencode/client.go index 03ae848c08..9dd96f6c18 100644 --- a/apps/backend/pkg/opencode/client.go +++ b/apps/backend/pkg/opencode/client.go @@ -30,6 +30,10 @@ type Client struct { eventHandler EventHandler controlCh chan ControlEvent + // SSE connection tracking - prevents multiple concurrent connections + sseCancel context.CancelFunc + sseActive bool + mu sync.RWMutex closed bool } @@ -355,12 +359,38 @@ func (c *Client) ReplyPermission(ctx context.Context, requestID, reply string, m return nil } -// StartEventStream starts the SSE event stream and processes events +// StartEventStream starts the SSE event stream and processes events. +// It ensures only one SSE connection is active at a time to prevent duplicate events. 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", + zap.String("session_id", sessionID)) + return nil + } + c.sseActive = true + c.mu.Unlock() + url := c.baseURL + "/event?directory=" + c.directory - req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + // 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 { + c.mu.Lock() + c.sseActive = false + c.sseCancel = nil + c.mu.Unlock() + sseCancel() return fmt.Errorf("create event stream request: %w", err) } @@ -372,24 +402,45 @@ func (c *Client) StartEventStream(ctx context.Context, sessionID string) error { 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) } 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)) } + c.logger.Debug("SSE stream connected", + zap.String("session_id", sessionID)) + // Process events in background - go c.processEventStream(ctx, sessionID, resp.Body) + go c.processEventStream(sseCtx, sessionID, resp.Body) return nil } // processEventStream reads and processes SSE events func (c *Client) processEventStream(ctx context.Context, sessionID string, body io.ReadCloser) { - defer func() { _ = body.Close() }() + 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)) + }() scanner := bufio.NewScanner(body) // Increase buffer size for potentially large events @@ -544,7 +595,7 @@ func (c *Client) processControlEvent(event *SDKEventEnvelope, sessionID string) } } -// Close closes the client +// Close closes the client and terminates any active SSE connection func (c *Client) Close() { c.mu.Lock() defer c.mu.Unlock() @@ -554,5 +605,12 @@ func (c *Client) Close() { } c.closed = true + // Cancel any active SSE connection + if c.sseCancel != nil { + c.sseCancel() + c.sseCancel = nil + } + c.sseActive = false + close(c.controlCh) } diff --git a/apps/web/components/task/task-top-bar.tsx b/apps/web/components/task/task-top-bar.tsx index 3be0f3feae..1ac579dee3 100644 --- a/apps/web/components/task/task-top-bar.tsx +++ b/apps/web/components/task/task-top-bar.tsx @@ -100,6 +100,12 @@ const TaskTopBar = memo(function TaskTopBar({ // Use worktree branch if available, otherwise fall back to base branch const displayBranch = worktreeBranch || baseBranch; + // Check if the remote branch actually tracks the current branch (not just origin/main as base) + // Pull/Push badges should only show when remote_branch is origin/ + const currentBranch = gitStatus?.branch; + const remoteBranch = gitStatus?.remote_branch; + const hasMatchingUpstream = remoteBranch && currentBranch && remoteBranch === `origin/${currentBranch}`; + // Calculate total additions and deletions from uncommitted files. let uncommittedAdditions = 0; let uncommittedDeletions = 0; @@ -466,7 +472,8 @@ const TaskTopBar = memo(function TaskTopBar({ > Pull - {(gitStatus?.behind ?? 0) > 0 && ( + {/* Only show pull count when upstream matches current branch (origin/) */} + {hasMatchingUpstream && (gitStatus?.behind ?? 0) > 0 && ( ↓{gitStatus?.behind} @@ -479,7 +486,8 @@ const TaskTopBar = memo(function TaskTopBar({ > Push - {(gitStatus?.ahead ?? 0) > 0 && ( + {/* Only show push count when upstream matches current branch (origin/) */} + {hasMatchingUpstream && (gitStatus?.ahead ?? 0) > 0 && ( ↑{gitStatus?.ahead}