diff --git a/apps/backend/cmd/kandev/adapters.go b/apps/backend/cmd/kandev/adapters.go index 6ba316e53c..03f4171df0 100644 --- a/apps/backend/cmd/kandev/adapters.go +++ b/apps/backend/cmd/kandev/adapters.go @@ -265,39 +265,6 @@ func (a *messageCreatorAdapter) CreateUserMessage(ctx context.Context, taskID, c return err } -// toolKindToMessageType maps the normalized tool kind to a frontend message type. -// This allows the frontend to use specialized renderers for different tool categories. -// Returns the message type and whether a warning should be logged (for unknown kinds). -func toolKindToMessageType(kind streams.ToolKind) (string, bool) { - switch kind { - case streams.ToolKindReadFile: - return "tool_read", false - case streams.ToolKindCodeSearch: - return "tool_search", false - case streams.ToolKindModifyFile: - return "tool_edit", false - case streams.ToolKindShellExec: - return "tool_execute", false - case streams.ToolKindHttpRequest: - return "tool_call", false - case streams.ToolKindGeneric: - return "tool_call", false - case streams.ToolKindCreateTask: - return "tool_call", false - case streams.ToolKindSubagentTask: - return "tool_call", false - case streams.ToolKindShowPlan: - return "tool_call", false - case streams.ToolKindManageTodos: - return "todo", false - case streams.ToolKindMisc: - return "tool_call", false - default: - // Unknown kind - this shouldn't happen, log a warning - return "tool_call", true - } -} - // CreateToolCallMessage creates a message for a tool call func (a *messageCreatorAdapter) CreateToolCallMessage(ctx context.Context, taskID, toolCallID, parentToolCallID, title, status, agentSessionID, turnID string, normalized *streams.NormalizedPayload) error { metadata := map[string]interface{}{ @@ -317,14 +284,7 @@ func (a *messageCreatorAdapter) CreateToolCallMessage(ctx context.Context, taskI // Determine message type from the normalized tool kind msgType := "tool_call" if normalized != nil { - var warn bool - msgType, warn = toolKindToMessageType(normalized.Kind()) - if warn { - a.logger.Warn("unknown tool kind falling back to tool_call", - zap.String("kind", string(normalized.Kind())), - zap.String("tool_call_id", toolCallID), - zap.String("title", title)) - } + msgType = normalized.Kind().ToMessageType() } _, err := a.svc.CreateMessage(ctx, &taskservice.CreateMessageRequest{ diff --git a/apps/backend/docs/add_agent.md b/apps/backend/docs/add_agent.md index 81724e842f..7bd89b3bb8 100644 --- a/apps/backend/docs/add_agent.md +++ b/apps/backend/docs/add_agent.md @@ -215,7 +215,7 @@ Add the agent definition: "model_flag": "--model {model}", "workspace_flag": "--workspace", "session_config": { - "resume_via_acp": false, + "native_session_resume": false, "resume_flag": "--resume", "can_recover": true, "session_dir_template": "{home}/.gemini/sessions", @@ -352,7 +352,7 @@ a.sendUpdate(AgentEvent{ Configure session handling in `session_config`: -- `resume_via_acp`: If true, sessions resume via ACP protocol. If false, use CLI flag. +- `native_session_resume`: If true, sessions resume via the adapter's native session loading (works across protocols). If false, use CLI flag or context injection. - `resume_flag`: CLI flag for resuming (e.g., `--resume`) - `can_recover`: Whether sessions can survive backend restarts - `session_dir_template`: Where session data is stored on host @@ -493,6 +493,6 @@ Before releasing a new agent integration: ### Session resume not working - For stream-based protocols: ensure adapter emits `session_status` event with agent's native session ID - Check that `resume_flag` is set in agent config (e.g., `"--resume"`) -- Verify `resume_via_acp` is `false` for CLI-based resume +- Verify `native_session_resume` is `false` for CLI-based resume - Check `executor_running` table has the correct `resume_token` value - For Claude Code: the session ID from `system` message should be stored as resume token diff --git a/apps/backend/docs/opencode_gemini_plan.md b/apps/backend/docs/opencode_gemini_plan.md index f0c8ba2455..ee4c2fa27d 100644 --- a/apps/backend/docs/opencode_gemini_plan.md +++ b/apps/backend/docs/opencode_gemini_plan.md @@ -514,7 +514,7 @@ Add these entries to the `agents` array: "model_flag": "", "workspace_flag": "", "session_config": { - "resume_via_acp": false, + "native_session_resume": false, "resume_flag": "", "can_recover": true, "session_dir_template": "{home}/.opencode", @@ -626,7 +626,7 @@ Add these entries to the `agents` array: "model_flag": "--model {model}", "workspace_flag": "", "session_config": { - "resume_via_acp": true, + "native_session_resume": true, "resume_flag": "", "can_recover": true, "session_dir_template": "{home}/.gemini", diff --git a/apps/backend/go.mod b/apps/backend/go.mod index d76a5e5e2e..19a933e3bb 100644 --- a/apps/backend/go.mod +++ b/apps/backend/go.mod @@ -39,12 +39,14 @@ require ( github.com/felixge/httpsnoop v1.0.4 // indirect github.com/gabriel-vasile/mimetype v1.4.2 // indirect github.com/gin-contrib/sse v0.1.0 // indirect + github.com/github/copilot-sdk/go v0.1.18 // indirect github.com/go-logr/logr v1.4.3 // indirect github.com/go-logr/stdr v1.2.2 // indirect github.com/go-playground/locales v0.14.1 // indirect github.com/go-playground/universal-translator v0.18.1 // indirect github.com/go-playground/validator/v10 v10.14.0 // indirect github.com/goccy/go-json v0.10.2 // indirect + github.com/google/jsonschema-go v0.4.2 // indirect github.com/hashicorp/hcl v1.0.0 // indirect github.com/invopop/jsonschema v0.13.0 // indirect github.com/json-iterator/go v1.1.12 // indirect diff --git a/apps/backend/go.sum b/apps/backend/go.sum index a58b499610..d79ef832fe 100644 --- a/apps/backend/go.sum +++ b/apps/backend/go.sum @@ -103,6 +103,8 @@ github.com/gin-contrib/sse v0.1.0 h1:Y/yl/+YNO8GZSjAhjMsSuLt29uWRFHdHYUb5lYOV9qE github.com/gin-contrib/sse v0.1.0/go.mod h1:RHrZQHXnP2xjPF+u1gW/2HnVO7nvIa9PG3Gm+fLHvGI= github.com/gin-gonic/gin v1.9.1 h1:4idEAncQnU5cB7BeOkPtxjfCSye0AAm1R0RVIqJ+Jmg= github.com/gin-gonic/gin v1.9.1/go.mod h1:hPrL7YrpYKXt5YId3A/Tnip5kqbEAP+KLuI3SUcPTeU= +github.com/github/copilot-sdk/go v0.1.18 h1:S1ocOfTKxiNGtj+/qp4z+RZeOr9hniqy3UqIIYZxsuQ= +github.com/github/copilot-sdk/go v0.1.18/go.mod h1:0SYT+64k347IDT0Trn4JHVFlUhPtGSE6ab479tU/+tY= github.com/go-gl/glfw v0.0.0-20190409004039-e6da0acd62b1/go.mod h1:vR7hzQXu2zJy9AVAgeJqvqgH9Q5CA+iKCZ2gyEVpxRU= github.com/go-gl/glfw/v3.3/glfw v0.0.0-20191125211704-12ad95a8df72/go.mod h1:tQ2UAYgL5IevRw8kRxooKSPJfGvJ9fJQFa0TUsXzTg8= github.com/go-gl/glfw/v3.3/glfw v0.0.0-20200222043503-6f7a984d4dc4/go.mod h1:tQ2UAYgL5IevRw8kRxooKSPJfGvJ9fJQFa0TUsXzTg8= @@ -160,6 +162,8 @@ github.com/google/go-cmp v0.5.4/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/ github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= +github.com/google/jsonschema-go v0.4.2 h1:tmrUohrwoLZZS/P3x7ex0WAVknEkBZM46iALbcqoRA8= +github.com/google/jsonschema-go v0.4.2/go.mod h1:r5quNTdLOYEz95Ru18zA0ydNbBuYoo9tgaYcxEYhJVE= github.com/google/martian v2.1.0+incompatible/go.mod h1:9I4somxYTbIHy5NJKHRl3wXiIaQGbYVAs8BPL6v8lEs= github.com/google/martian/v3 v3.0.0/go.mod h1:y5Zk1BBys9G+gd6Jrk0W3cC1+ELVxBWuIGO+w/tUAp0= github.com/google/martian/v3 v3.1.0/go.mod h1:y5Zk1BBys9G+gd6Jrk0W3cC1+ELVxBWuIGO+w/tUAp0= diff --git a/apps/backend/internal/agent/lifecycle/command.go b/apps/backend/internal/agent/lifecycle/command.go index a41d6c8b45..c01a82f6d2 100644 --- a/apps/backend/internal/agent/lifecycle/command.go +++ b/apps/backend/internal/agent/lifecycle/command.go @@ -20,7 +20,7 @@ func NewCommandBuilder() *CommandBuilder { // CommandOptions contains options for building a command type CommandOptions struct { Model string // Model to use (appended via ModelFlag if set) - SessionID string // Session ID to resume (appended via SessionConfig.ResumeFlag if not ResumeViaACP) + SessionID string // Session ID to resume (appended via SessionConfig.ResumeFlag if not NativeSessionResume) AutoApprove bool // If true, skip permission flags (auto-approve all tool calls) // PermissionValues contains the values for each permission setting @@ -47,10 +47,10 @@ func (cb *CommandBuilder) BuildCommand(agentConfig *registry.AgentTypeConfig, op // Append session resume flag if: // 1. Session ID is provided - // 2. Agent does NOT use ACP for session resumption (uses CLI flag instead) + // 2. Agent does NOT use native session loading (uses CLI flag instead) // 3. Agent has a ResumeFlag configured // 4. Agent supports session recovery (CanRecover is not false) - if opts.SessionID != "" && !agentConfig.SessionConfig.ResumeViaACP && agentConfig.SessionConfig.ResumeFlag != "" && agentConfig.SessionConfig.SupportsRecovery() { + if opts.SessionID != "" && !agentConfig.SessionConfig.NativeSessionResume && agentConfig.SessionConfig.ResumeFlag != "" && agentConfig.SessionConfig.SupportsRecovery() { cmd = append(cmd, agentConfig.SessionConfig.ResumeFlag, opts.SessionID) } diff --git a/apps/backend/internal/agent/lifecycle/manager.go b/apps/backend/internal/agent/lifecycle/manager.go index e3b08a5bbb..40a62f4cff 100644 --- a/apps/backend/internal/agent/lifecycle/manager.go +++ b/apps/backend/internal/agent/lifecycle/manager.go @@ -817,6 +817,10 @@ func (m *Manager) Launch(ctx context.Context, req *LaunchRequest) (*AgentExecuti execution.ACPSessionID = req.ACPSessionID } + // Set flag indicating if agent adapter reports completion via stream + // When true, session manager should NOT emit a complete event after prompt returns + execution.ReportsStatusViaStream = agentConfig.SessionConfig.ReportsStatusViaStream + // Build agent command string for later use with StartAgentProcess model := "" autoApprove := false @@ -1212,7 +1216,7 @@ func (m *Manager) emitSessionStatusEvent(execution *AgentExecution, agentConfig wasResumed := false if providedACPSessionID != "" { - if agentConfig.SessionConfig.ResumeViaACP { + if agentConfig.SessionConfig.NativeSessionResume { wasResumed = true } else if agentConfig.SessionConfig.ResumeFlag != "" && agentConfig.SessionConfig.SupportsRecovery() { wasResumed = true @@ -1238,6 +1242,13 @@ func (m *Manager) handlePermissionRequestEvent(execution *AgentExecution, event // handleAgentEvent processes incoming agent events from the agent func (m *Manager) handleAgentEvent(execution *AgentExecution, event agentctl.AgentEvent) { + // Log all incoming events for debugging + m.logger.Debug("handleAgentEvent entry", + zap.String("execution_id", execution.ID), + zap.String("event_type", event.Type), + zap.String("operation_id", event.OperationID), + zap.Int("text_length", len(event.Text))) + // Handle different event types based on the Type field switch event.Type { case "message_chunk": @@ -1245,6 +1256,12 @@ func (m *Manager) handleAgentEvent(execution *AgentExecution, event agentctl.Age if event.Text != "" { execution.messageMu.Lock() execution.messageBuffer.WriteString(event.Text) + bufferLenAfterWrite := execution.messageBuffer.Len() + m.logger.Debug("message_chunk written to buffer", + zap.String("execution_id", execution.ID), + zap.String("operation_id", event.OperationID), + zap.Int("text_length", len(event.Text)), + zap.Int("buffer_length_after", bufferLenAfterWrite)) // Check if buffer contains newlines - if so, flush up to the last newline bufContent := execution.messageBuffer.String() @@ -1349,17 +1366,65 @@ func (m *Manager) handleAgentEvent(execution *AgentExecution, event agentctl.Age zap.Any("data", event.Data)) case "complete": + // Check buffer content BEFORE any processing + execution.messageMu.Lock() + bufferContentBeforeFlush := execution.messageBuffer.String() + streamingWasUsed := execution.streamingUsedThisTurn + currentMsgID := execution.currentMessageID + execution.messageMu.Unlock() + + // Log preview of buffer content for debugging (first 100 chars) + bufferPreview := bufferContentBeforeFlush + if len(bufferPreview) > 100 { + bufferPreview = bufferPreview[:100] + "..." + } + m.logger.Info("agent turn complete", zap.String("execution_id", execution.ID), - zap.String("session_id", event.SessionID)) + zap.String("operation_id", event.OperationID), + zap.String("session_id", event.SessionID), + zap.Bool("streaming_was_used", streamingWasUsed), + zap.String("current_msg_id", currentMsgID), + zap.String("event_text", event.Text), + zap.Int("buffer_length", len(bufferContentBeforeFlush)), + zap.String("buffer_preview", bufferPreview)) - // Flush accumulated message buffer and include in the event - // The orchestrator will save this as an agent message - if flushedText := m.flushMessageBuffer(execution); flushedText != "" { - event.Text = flushedText + // Get the agent message text: + // 1. If streaming was used, flush the buffer (which publishes final chunk) and ignore event.Text + // 2. Otherwise, prefer text from event (adapter-provided, e.g., Amp) or flush buffer + var agentText string + if streamingWasUsed { + // Streaming messages were created - flush buffer to publish final chunk + // and return empty so we don't create duplicate messages + m.flushMessageBuffer(execution) + agentText = "" + } else if event.Text != "" { + // No streaming used, use adapter-provided text + agentText = event.Text + // Still clear the buffer to avoid stale data + m.flushMessageBuffer(execution) + } else { + // No streaming used, no adapter text - flush buffer for content + agentText = m.flushMessageBuffer(execution) + } + + m.logger.Info("complete event text resolution", + zap.String("execution_id", execution.ID), + zap.String("operation_id", event.OperationID), + zap.Int("agent_text_length", len(agentText)), + zap.Bool("will_save_message", agentText != "")) + + // Reset the streaming flag for next turn + execution.messageMu.Lock() + execution.streamingUsedThisTurn = false + execution.messageMu.Unlock() + + // Include agent text in the event for the orchestrator to save as an agent message + if agentText != "" { + event.Text = agentText // Store agent message to session history for context injection if m.historyManager != nil && execution.SessionID != "" { - if err := m.historyManager.AppendAgentMessage(execution.SessionID, flushedText); err != nil { + if err := m.historyManager.AppendAgentMessage(execution.SessionID, agentText); err != nil { m.logger.Warn("failed to store agent message to history", zap.Error(err)) } } @@ -1478,6 +1543,9 @@ func (m *Manager) publishStreamingMessage(execution *AgentExecution, content str messageID = uuid.New().String() execution.currentMessageID = messageID } + + // Mark that streaming was used this turn (for duplicate prevention on complete) + execution.streamingUsedThisTurn = true execution.messageMu.Unlock() event := AgentStreamEventData{ @@ -2471,7 +2539,7 @@ func (m *Manager) buildPassthroughCommand(agentConfig *registry.AgentTypeConfig, // 2. Agent uses CLI-based resume (not ACP) // 3. Agent has a ResumeFlag configured if acpSessionID != "" && - !agentConfig.SessionConfig.ResumeViaACP && + !agentConfig.SessionConfig.NativeSessionResume && agentConfig.SessionConfig.ResumeFlag != "" { cmd = append(cmd, agentConfig.SessionConfig.ResumeFlag, acpSessionID) } else if initialPrompt != "" { diff --git a/apps/backend/internal/agent/lifecycle/runtime_standalone.go b/apps/backend/internal/agent/lifecycle/runtime_standalone.go index f3148d493e..02ee6f57ec 100644 --- a/apps/backend/internal/agent/lifecycle/runtime_standalone.go +++ b/apps/backend/internal/agent/lifecycle/runtime_standalone.go @@ -111,6 +111,11 @@ func (r *StandaloneRuntime) CreateInstance(ctx context.Context, req *RuntimeCrea SessionID: req.SessionID, } + r.logger.Info("CreateInstance: sending request to agentctl", + zap.String("instance_id", req.InstanceID), + zap.String("req_protocol", req.Protocol), + zap.String("createReq_protocol", createReq.Protocol)) + resp, err := r.ctl.CreateInstance(ctx, createReq) if err != nil { return nil, fmt.Errorf("failed to create standalone instance: %w", err) diff --git a/apps/backend/internal/agent/lifecycle/session.go b/apps/backend/internal/agent/lifecycle/session.go index c4b1ed9759..8209ce2d8c 100644 --- a/apps/backend/internal/agent/lifecycle/session.go +++ b/apps/backend/internal/agent/lifecycle/session.go @@ -57,8 +57,8 @@ type InitializeResult struct { // It handles the initialize handshake and session creation/loading based on config. // // Session behavior: -// - If agentConfig.SessionConfig.ResumeViaACP is true AND existingSessionID is provided: use session/load -// - If agentConfig.SessionConfig.ResumeViaACP is false (CLI handles resume): always use session/new +// - If agentConfig.SessionConfig.NativeSessionResume is true AND existingSessionID is provided: use session/load +// - If agentConfig.SessionConfig.NativeSessionResume is false (CLI handles resume): always use session/new // - Otherwise: use session/new func (sm *SessionManager) InitializeSession( ctx context.Context, @@ -71,7 +71,7 @@ func (sm *SessionManager) InitializeSession( sm.logger.Info("initializing ACP session", zap.String("agent_type", agentConfig.ID), zap.String("workspace_path", workspacePath), - zap.Bool("resume_via_acp", agentConfig.SessionConfig.ResumeViaACP), + zap.Bool("native_session_resume", agentConfig.SessionConfig.NativeSessionResume), zap.String("existing_session_id", existingSessionID)) // Step 1: Send initialize request @@ -119,7 +119,7 @@ func (sm *SessionManager) createOrLoadSession( workspacePath string, mcpServers []agentctltypes.McpServer, ) (string, error) { - if agentConfig.SessionConfig.ResumeViaACP && existingSessionID != "" { + if agentConfig.SessionConfig.NativeSessionResume && existingSessionID != "" { sessionID, err := sm.loadSession(ctx, client, agentConfig, existingSessionID) if err != nil { // If session/load fails with "Method not found" or "LoadSession capability is false", @@ -142,7 +142,7 @@ func (sm *SessionManager) createOrLoadSession( // shouldInjectResumeContext determines if we should inject resume context for this session. // Returns true if: // 1. The agent uses ACP protocol -// 2. The agent doesn't support session/load (ResumeViaACP is false) +// 2. The agent doesn't support native session loading (NativeSessionResume is false) // 3. We have a task session ID (for history lookup) // 4. There's existing history for this session func (sm *SessionManager) shouldInjectResumeContext(agentConfig *registry.AgentTypeConfig, taskSessionID string) bool { @@ -155,8 +155,8 @@ func (sm *SessionManager) shouldInjectResumeContext(agentConfig *registry.AgentT return false } - // If agent supports session/load via ACP, don't inject (it will restore context natively) - if agentConfig.SessionConfig.ResumeViaACP { + // If agent supports native session loading, don't inject (it will restore context natively) + if agentConfig.SessionConfig.NativeSessionResume { return false } @@ -273,7 +273,7 @@ func (sm *SessionManager) InitializeAndPrompt( zap.String("agentctl_url", execution.agentctl.BaseURL()), zap.String("agent_type", agentConfig.ID), zap.String("existing_acp_session_id", execution.ACPSessionID), - zap.Bool("resume_via_acp", agentConfig.SessionConfig.ResumeViaACP)) + zap.Bool("native_session_resume", agentConfig.SessionConfig.NativeSessionResume)) // Use InitializeSession for configuration-driven session initialization result, err := sm.InitializeSession( @@ -464,58 +464,81 @@ func (sm *SessionManager) SendPrompt( } // Extract accumulated content from buffers and handle streaming message completion - execution.messageMu.Lock() - agentMessage := execution.messageBuffer.String() - execution.messageBuffer.Reset() - execution.thinkingBuffer.Reset() - currentMsgID := execution.currentMessageID - execution.currentMessageID = "" // Clear for next turn - execution.currentThinkingID = "" // Clear for next turn - execution.messageMu.Unlock() - + // NOTE: When ReportsStatusViaStream is true, the adapter's complete event (via handleAgentEvent) + // will handle the buffer content. We should NOT clear the buffer here to avoid a race condition + // where the buffer is cleared before handleAgentEvent processes the complete event. stopReason := "" if resp != nil { stopReason = string(resp.StopReason) } - trimmedMessage := strings.TrimSpace(agentMessage) + var agentMessage string - sm.logger.Info("ACP prompt completed", - zap.String("execution_id", execution.ID), - zap.String("stop_reason", stopReason), - zap.Int("message_length", len(trimmedMessage)), - zap.Bool("has_streaming_msg", currentMsgID != "")) + if execution.ReportsStatusViaStream { + // Agent adapter handles completion via stream - let handleAgentEvent deal with the buffer + // Just log for debugging. The buffer content will be handled by handleAgentEvent when + // the complete event arrives from the adapter. + execution.messageMu.Lock() + bufLen := execution.messageBuffer.Len() + // Peek at the message for the return value, but don't clear the buffer + agentMessage = execution.messageBuffer.String() + execution.messageMu.Unlock() - // Handle remaining content from the buffer - if sm.eventPublisher != nil { - // If there's remaining content and an active streaming message, - // append it to the streaming message instead of creating a new one - if trimmedMessage != "" && currentMsgID != "" { - // Publish streaming append with the remaining content - sm.eventPublisher.PublishAgentStreamEventPayload(&AgentStreamEventPayload{ - Type: "agent/event", - Timestamp: time.Now().UTC().Format(time.RFC3339), - AgentID: execution.AgentProfileID, - TaskID: execution.TaskID, - SessionID: execution.SessionID, - Data: &AgentStreamEventData{ - Type: "message_streaming", - MessageID: currentMsgID, - IsAppend: true, - Text: trimmedMessage, - }, - }) - // Content was handled via streaming, publish complete with empty text - trimmedMessage = "" + sm.logger.Info("ACP prompt completed (stream-based completion)", + zap.String("execution_id", execution.ID), + zap.String("stop_reason", stopReason), + zap.Int("buffer_length", bufLen), + zap.Bool("reports_status_via_stream", true)) + } else { + // Agent adapter does NOT handle completion via stream - we need to handle the buffer here + execution.messageMu.Lock() + agentMessage = execution.messageBuffer.String() + execution.messageBuffer.Reset() + execution.thinkingBuffer.Reset() + currentMsgID := execution.currentMessageID + execution.currentMessageID = "" // Clear for next turn + execution.currentThinkingID = "" // Clear for next turn + execution.messageMu.Unlock() + + trimmedMessage := strings.TrimSpace(agentMessage) + + sm.logger.Info("ACP prompt completed", + zap.String("execution_id", execution.ID), + zap.String("stop_reason", stopReason), + zap.Int("message_length", len(trimmedMessage)), + zap.Bool("has_streaming_msg", currentMsgID != "")) + + // Handle remaining content from the buffer + if sm.eventPublisher != nil { + // If there's remaining content and an active streaming message, + // append it to the streaming message instead of creating a new one + if trimmedMessage != "" && currentMsgID != "" { + // Publish streaming append with the remaining content + sm.eventPublisher.PublishAgentStreamEventPayload(&AgentStreamEventPayload{ + Type: "agent/event", + Timestamp: time.Now().UTC().Format(time.RFC3339), + AgentID: execution.AgentProfileID, + TaskID: execution.TaskID, + SessionID: execution.SessionID, + Data: &AgentStreamEventData{ + Type: "message_streaming", + MessageID: currentMsgID, + IsAppend: true, + Text: trimmedMessage, + }, + }) + // Content was handled via streaming, publish complete with empty text + trimmedMessage = "" + } + + // Publish complete event (with remaining text only if no streaming message was active) + completeEvent := streams.AgentEvent{ + Type: streams.EventTypeComplete, + SessionID: execution.ACPSessionID, + Text: trimmedMessage, + } + sm.eventPublisher.PublishAgentStreamEvent(execution, completeEvent) } - - // Publish complete event (with remaining text only if no streaming message was active) - completeEvent := streams.AgentEvent{ - Type: streams.EventTypeComplete, - SessionID: execution.ACPSessionID, - Text: trimmedMessage, - } - sm.eventPublisher.PublishAgentStreamEvent(execution, completeEvent) } // Mark as READY for next prompt diff --git a/apps/backend/internal/agent/lifecycle/types.go b/apps/backend/internal/agent/lifecycle/types.go index 704940c04b..9320cafbe0 100644 --- a/apps/backend/internal/agent/lifecycle/types.go +++ b/apps/backend/internal/agent/lifecycle/types.go @@ -59,11 +59,21 @@ type AgentExecution struct { currentMessageID string currentThinkingID string + // Track whether streaming messages were created during this turn + // This is set true when publishStreamingMessage is called and cleared on complete + // Used to avoid duplicate messages when both streaming and complete event have text + streamingUsedThisTurn bool + // Resume context tracking for fork_session pattern (ACP agents that don't support session/load) // needsResumeContext is set to true when the session has history that should be injected // resumeContextInjected is set to true after context has been injected into a prompt needsResumeContext bool resumeContextInjected bool + + // ReportsStatusViaStream indicates whether the agent adapter handles completion events + // via the stream protocol. When true, the session manager should NOT emit a complete + // event after prompt returns, as the adapter already sent one via the stream. + ReportsStatusViaStream bool } // GetAgentCtlClient returns the agentctl client for this execution diff --git a/apps/backend/internal/agent/registry/agents.json b/apps/backend/internal/agent/registry/agents.json index 1fa9be1672..7d4ab06d13 100644 --- a/apps/backend/internal/agent/registry/agents.json +++ b/apps/backend/internal/agent/registry/agents.json @@ -59,7 +59,7 @@ "model_flag": "--model {model}", "workspace_flag": "--workspace-root", "session_config": { - "resume_via_acp": false, + "native_session_resume": false, "resume_flag": "--resume", "can_recover": false, "session_dir_template": "{home}/.augment/sessions", @@ -225,7 +225,7 @@ "model_flag": "-c model=\"{model}\"", "workspace_flag": "", "session_config": { - "resume_via_acp": true, + "native_session_resume": true, "resume_flag": "", "can_recover": true, "session_dir_template": "", @@ -364,7 +364,7 @@ "model_flag": "--model {model}", "workspace_flag": "", "session_config": { - "resume_via_acp": false, + "native_session_resume": false, "resume_flag": "--resume", "can_recover": true, "session_dir_template": "{home}/.claude", @@ -489,7 +489,7 @@ "model_flag": "--model {model}", "workspace_flag": "", "session_config": { - "resume_via_acp": false, + "native_session_resume": false, "resume_flag": "", "can_recover": false, "session_dir_template": "{home}/.gemini", @@ -629,7 +629,7 @@ "model_flag": "", "workspace_flag": "", "session_config": { - "resume_via_acp": true, + "native_session_resume": true, "resume_flag": "", "can_recover": true, "session_dir_template": "{home}/.opencode", @@ -728,6 +728,349 @@ } ] } + }, + { + "id": "copilot", + "name": "GitHub Copilot Agent", + "display_name": "Copilot", + "description": "GitHub Copilot CLI-powered autonomous coding agent using the Copilot SDK protocol.", + "enabled": true, + "image": "", + "tag": "", + "cmd": [ + "npx", + "-y", + "@github/copilot@0.0.375", + "--server", + "--log-level", + "error" + ], + "passthrough_config": { + "supported": true, + "label": "CLI Passthrough", + "description": "Show terminal directly instead of chat interface", + "passthrough_cmd": [ + "npx", + "-y", + "@github/copilot" + ], + "model_flag": "--model {model}", + "idle_timeout_ms": 3000, + "buffer_max_bytes": 2097152, + "resume_flag": "--resume" + }, + "working_dir": "{workspace}", + "required_env": [], + "env": {}, + "mounts": [], + "resource_limits": { + "memory_mb": 4096, + "cpu_cores": 2.0, + "timeout_seconds": 3600 + }, + "capabilities": [ + "code_generation", + "code_review", + "refactoring", + "testing", + "shell_execution" + ], + "protocol": "copilot", + "protocol_config": {}, + "model_flag": "--model {model}", + "workspace_flag": "", + "session_config": { + "native_session_resume": true, + "resume_flag": "--resume", + "can_recover": true, + "session_dir_template": "{home}/.copilot", + "session_dir_target": "", + "reports_status_via_stream": true + }, + "permission_config": { + "permission_flag": "", + "tools_requiring_permission": [ + "bash", + "write_file", + "edit_file" + ] + }, + "permission_settings": { + "auto_approve": { + "supported": true, + "default": true, + "label": "Auto-approve", + "description": "Automatically approve tool calls", + "apply_method": "cli_flag", + "cli_flag": "--allow-all-tools" + } + }, + "discovery": { + "supports_mcp": true, + "mcp_config_path": { + "linux": [ + "~/.copilot/mcp-config.json" + ], + "windows": [], + "macos": [ + "~/.copilot/mcp-config.json" + ] + }, + "installation_path": { + "linux": [], + "windows": [], + "macos": [ + "~/.copilot/pkg" + ] + }, + "discovery_capabilities": { + "supports_session_resume": true, + "supports_shell": false, + "supports_workspace_only": false + } + }, + "model_config": { + "default_model": "gpt-4.1", + "available_models": [ + { + "id": "gpt-4.1", + "name": "GPT-4.1", + "description": "OpenAI GPT-4.1", + "provider": "openai", + "is_default": true + }, + { + "id": "claude-sonnet-4", + "name": "Claude Sonnet 4", + "description": "Anthropic Claude Sonnet 4", + "provider": "anthropic", + "is_default": false + }, + { + "id": "claude-sonnet-4.5", + "name": "Claude Sonnet 4.5", + "description": "Anthropic Claude Sonnet 4.5", + "provider": "anthropic", + "is_default": false + }, + { + "id": "claude-opus-4.5", + "name": "Claude Opus 4.5", + "description": "Anthropic Claude Opus 4.5", + "provider": "anthropic", + "is_default": false + }, + { + "id": "claude-haiku-4.5", + "name": "Claude Haiku 4.5", + "description": "Anthropic Claude Haiku 4.5", + "provider": "anthropic", + "is_default": false + }, + { + "id": "gpt-5.2-codex", + "name": "GPT-5.2 Codex", + "description": "OpenAI GPT-5.2 Codex", + "provider": "openai", + "is_default": false + }, + { + "id": "gpt-5.2", + "name": "GPT-5.2", + "description": "OpenAI GPT-5.2", + "provider": "openai", + "is_default": false + }, + { + "id": "gpt-5.1-codex-max", + "name": "GPT-5.1 Codex Max", + "description": "OpenAI GPT-5.1 Codex Max", + "provider": "openai", + "is_default": false + }, + { + "id": "gpt-5.1-codex", + "name": "GPT-5.1 Codex", + "description": "OpenAI GPT-5.1 Codex", + "provider": "openai", + "is_default": false + }, + { + "id": "gpt-5.1", + "name": "GPT-5.1", + "description": "OpenAI GPT-5.1", + "provider": "openai", + "is_default": false + }, + { + "id": "gpt-5", + "name": "GPT-5", + "description": "OpenAI GPT-5", + "provider": "openai", + "is_default": false + }, + { + "id": "gpt-5.1-codex-mini", + "name": "GPT-5.1 Codex Mini", + "description": "OpenAI GPT-5.1 Codex Mini", + "provider": "openai", + "is_default": false + }, + { + "id": "gpt-5-mini", + "name": "GPT-5 Mini", + "description": "OpenAI GPT-5 Mini", + "provider": "openai", + "is_default": false + }, + { + "id": "gemini-3-pro-preview", + "name": "Gemini 3 Pro Preview", + "description": "Google Gemini 3 Pro Preview", + "provider": "google", + "is_default": false + } + ] + } + }, + { + "id": "amp", + "name": "Sourcegraph Amp Agent", + "display_name": "Amp", + "description": "Sourcegraph Amp CLI-powered autonomous coding agent using stream-json protocol.", + "enabled": true, + "image": "", + "tag": "", + "cmd": [ + "npx", + "-y", + "@sourcegraph/amp@latest", + "--execute", + "--stream-json", + "--stream-json-input" + ], + "passthrough_config": { + "supported": true, + "label": "CLI Passthrough", + "description": "Show terminal directly instead of chat interface", + "passthrough_cmd": [ + "npx", + "-y", + "@sourcegraph/amp@latest" + ], + "model_flag": "-m {model}", + "idle_timeout_ms": 3000, + "buffer_max_bytes": 2097152, + "resume_flag": "" + }, + "working_dir": "{workspace}", + "required_env": [], + "env": {}, + "mounts": [], + "resource_limits": { + "memory_mb": 4096, + "cpu_cores": 2.0, + "timeout_seconds": 3600 + }, + "capabilities": [ + "code_generation", + "code_review", + "refactoring", + "testing", + "shell_execution" + ], + "protocol": "amp", + "protocol_config": {}, + "model_flag": "-m {model}", + "workspace_flag": "", + "session_config": { + "native_session_resume": false, + "resume_flag": "", + "can_recover": true, + "session_dir_template": "{home}/.config/amp", + "session_dir_target": "", + "reports_status_via_stream": true, + "fork_session_cmd": [ + "npx", + "-y", + "@sourcegraph/amp@latest", + "threads", + "fork" + ], + "continue_session_cmd": [ + "npx", + "-y", + "@sourcegraph/amp@latest", + "threads", + "continue", + "--execute", + "--stream-json", + "--stream-json-input" + ] + }, + "permission_config": { + "permission_flag": "", + "tools_requiring_permission": [ + "bash", + "write_file", + "edit_file" + ] + }, + "permission_settings": { + "auto_approve": { + "supported": true, + "default": true, + "label": "Auto-approve (Dangerously Allow All)", + "description": "Automatically approve all tool calls including shell commands", + "apply_method": "cli_flag", + "cli_flag": "--dangerously-allow-all" + } + }, + "discovery": { + "supports_mcp": true, + "mcp_config_path": { + "linux": [ + "~/.config/amp/settings.json" + ], + "windows": [], + "macos": [ + "~/.amp/bin" + ] + }, + "installation_path": { + "linux": [ + "~/.amp/bin" + ], + "windows": [], + "macos": [ + "~/.amp/bin" + ] + }, + "discovery_capabilities": { + "supports_session_resume": true, + "supports_shell": false, + "supports_workspace_only": false + } + }, + "model_config": { + "default_model": "smart", + "available_models": [ + { + "id": "smart", + "name": "Smart Mode", + "description": "State-of-the-art models for maximum capability and autonomy", + "provider": "amp", + "is_default": true + }, + { + "id": "deep", + "name": "Deep Mode", + "description": "Deep reasoning with extended thinking on complex problems", + "provider": "amp", + "is_default": false + } + ] + } } ] } diff --git a/apps/backend/internal/agent/registry/registry.go b/apps/backend/internal/agent/registry/registry.go index 14e9048c89..bfea6eb829 100644 --- a/apps/backend/internal/agent/registry/registry.go +++ b/apps/backend/internal/agent/registry/registry.go @@ -72,12 +72,14 @@ type AgentTypeConfig struct { // SessionConfig defines how session resumption is handled for an agent type type SessionConfig struct { - // ResumeViaACP indicates whether session resumption should use the ACP session/load method. - // If false, session resumption is handled via CLI flags (ResumeFlag). - ResumeViaACP bool `json:"resume_via_acp"` + // NativeSessionResume indicates whether session resumption should use the adapter's + // native session loading mechanism (e.g., LoadSession method). This works across + // different protocols (ACP, Copilot SDK, Codex, OpenCode, etc.). + // If false, session resumption is handled via CLI flags (ResumeFlag) or context injection. + NativeSessionResume bool `json:"native_session_resume"` - // ResumeFlag is the CLI flag used for session resumption when ResumeViaACP is false. - // Example: "--resume" for auggie, empty for agents that use ACP session/load. + // ResumeFlag is the CLI flag used for session resumption when NativeSessionResume is false. + // Example: "--resume" for auggie, empty for agents that use native session loading. ResumeFlag string `json:"resume_flag,omitempty"` // CanRecover indicates whether this agent supports session recovery after a backend restart. @@ -510,6 +512,10 @@ func parseProtocol(p string) agent.Protocol { return agent.ProtocolClaudeCode case "opencode": return agent.ProtocolOpenCode + case "copilot": + return agent.ProtocolCopilot + case "amp": + return agent.ProtocolAmp default: return agent.ProtocolACP // Default to ACP } diff --git a/apps/backend/internal/agentctl/server/adapter/amp_adapter.go b/apps/backend/internal/agentctl/server/adapter/amp_adapter.go new file mode 100644 index 0000000000..e776bf709c --- /dev/null +++ b/apps/backend/internal/agentctl/server/adapter/amp_adapter.go @@ -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) diff --git a/apps/backend/internal/agentctl/server/adapter/amp_normalizer.go b/apps/backend/internal/agentctl/server/adapter/amp_normalizer.go new file mode 100644 index 0000000000..78df570b67 --- /dev/null +++ b/apps/backend/internal/agentctl/server/adapter/amp_normalizer.go @@ -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) +} diff --git a/apps/backend/internal/agentctl/server/adapter/copilot_adapter.go b/apps/backend/internal/agentctl/server/adapter/copilot_adapter.go new file mode 100644 index 0000000000..a2f115f4e9 --- /dev/null +++ b/apps/backend/internal/agentctl/server/adapter/copilot_adapter.go @@ -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 " 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 " 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 " 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) diff --git a/apps/backend/internal/agentctl/server/adapter/copilot_normalizer.go b/apps/backend/internal/agentctl/server/adapter/copilot_normalizer.go new file mode 100644 index 0000000000..cd07d5f1df --- /dev/null +++ b/apps/backend/internal/agentctl/server/adapter/copilot_normalizer.go @@ -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 "" suffix that Copilot appends to shell output. +var exitedPattern = regexp.MustCompile(`\n?$`) + +// 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 "" suffix and extract exit code + output, exitCode = n.cleanShellOutput(output, exitCode) + + return output, exitCode +} + +// cleanShellOutput strips Copilot's "" 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 +} diff --git a/apps/backend/internal/agentctl/server/adapter/factory.go b/apps/backend/internal/agentctl/server/adapter/factory.go index 8d54c8cb94..1723dba7b3 100644 --- a/apps/backend/internal/agentctl/server/adapter/factory.go +++ b/apps/backend/internal/agentctl/server/adapter/factory.go @@ -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) } diff --git a/apps/backend/internal/agentctl/server/adapter/transport/shared/debug.go b/apps/backend/internal/agentctl/server/adapter/transport/shared/debug.go index 65a97abb22..a641c8b644 100644 --- a/apps/backend/internal/agentctl/server/adapter/transport/shared/debug.go +++ b/apps/backend/internal/agentctl/server/adapter/transport/shared/debug.go @@ -34,6 +34,8 @@ const ( ProtocolStreamJSON = "streamjson" ProtocolCodex = "codex" ProtocolOpenCode = "opencode" + ProtocolAmp = "amp" + ProtocolCopilot = "copilot" ) func resolveDebugLogDir() string { diff --git a/apps/backend/internal/agentctl/server/instance/manager.go b/apps/backend/internal/agentctl/server/instance/manager.go index a074cd20ad..59495bc5bb 100644 --- a/apps/backend/internal/agentctl/server/instance/manager.go +++ b/apps/backend/internal/agentctl/server/instance/manager.go @@ -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), diff --git a/apps/backend/internal/agentctl/server/process/manager.go b/apps/backend/internal/agentctl/server/process/manager.go index 1617883cb8..4f5e1df74e 100644 --- a/apps/backend/internal/agentctl/server/process/manager.go +++ b/apps/backend/internal/agentctl/server/process/manager.go @@ -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") diff --git a/apps/backend/internal/agentctl/types/streams/tool_payload.go b/apps/backend/internal/agentctl/types/streams/tool_payload.go index 7c6c040ab7..1e417312ec 100644 --- a/apps/backend/internal/agentctl/types/streams/tool_payload.go +++ b/apps/backend/internal/agentctl/types/streams/tool_payload.go @@ -19,6 +19,25 @@ const ( ToolKindMisc ToolKind = "misc" ) +// ToMessageType maps the ToolKind to a frontend message type. +// This allows the frontend to use specialized renderers for different tool categories. +func (k ToolKind) ToMessageType() string { + switch k { + case ToolKindReadFile: + return "tool_read" + case ToolKindCodeSearch: + return "tool_search" + case ToolKindModifyFile: + return "tool_edit" + case ToolKindShellExec: + return "tool_execute" + case ToolKindManageTodos: + return "todo" + default: + return "tool_call" + } +} + // NormalizedPayload is the normalized tool data (discriminated union). // Exactly one of the kind-specific fields will be set based on Kind. // Fields are unexported to enforce use of factory functions (NewReadFile, NewShellExec, etc.) diff --git a/apps/backend/internal/orchestrator/event_handlers.go b/apps/backend/internal/orchestrator/event_handlers.go index 22b59127a2..6fc558a3de 100644 --- a/apps/backend/internal/orchestrator/event_handlers.go +++ b/apps/backend/internal/orchestrator/event_handlers.go @@ -21,20 +21,7 @@ func toolKindToMessageType(normalized *streams.NormalizedPayload) string { if normalized == nil { return "tool_call" } - switch normalized.Kind() { - case streams.ToolKindReadFile: - return "tool_read" - case streams.ToolKindCodeSearch: - return "tool_search" - case streams.ToolKindModifyFile: - return "tool_edit" - case streams.ToolKindShellExec: - return "tool_execute" - case streams.ToolKindManageTodos: - return "todo" - default: - return "tool_call" - } + return normalized.Kind().ToMessageType() } // Event handlers @@ -548,6 +535,11 @@ func (s *Service) handleAgentStreamEvent(ctx context.Context, payload *lifecycle s.handleToolUpdateEvent(ctx, payload) case "complete": + s.logger.Debug("orchestrator received complete event", + zap.String("task_id", taskID), + zap.String("session_id", sessionID), + zap.Int("text_length", len(payload.Data.Text)), + zap.Bool("has_text", payload.Data.Text != "")) s.saveAgentTextIfPresent(ctx, payload) s.completeTurnForSession(ctx, sessionID) s.setSessionWaitingForInput(ctx, taskID, sessionID) diff --git a/apps/backend/internal/task/service/service.go b/apps/backend/internal/task/service/service.go index 7c24799b65..da9f03f2db 100644 --- a/apps/backend/internal/task/service/service.go +++ b/apps/backend/internal/task/service/service.go @@ -2105,8 +2105,21 @@ func (s *Service) UpdateToolCallMessageWithCreate(ctx context.Context, sessionID } // Add normalized tool data to metadata for frontend consumption + // and update message type if the normalized kind changed (e.g., read_file -> code_search for directories) if normalized != nil { message.Metadata["normalized"] = normalized + + // Update message type if the normalized kind changed + // This handles cases like Read on a directory converting to code_search + newMsgType := models.MessageType(normalized.Kind().ToMessageType()) + if newMsgType != message.Type { + s.logger.Debug("updating message type based on normalized kind", + zap.String("message_id", message.ID), + zap.String("old_type", string(message.Type)), + zap.String("new_type", string(newMsgType)), + zap.String("normalized_kind", string(normalized.Kind()))) + message.Type = newMsgType + } } // Update title/content if provided and different from current @@ -2315,7 +2328,8 @@ func (s *Service) CompleteTurn(ctx context.Context, turnID string) error { turn, err := s.repo.GetTurn(ctx, turnID) if err != nil { s.logger.Warn("failed to refetch completed turn", zap.String("turn_id", turnID), zap.Error(err)) - // Still publish with the old turn data + // Turn was likely deleted (task deletion race), skip publishing + return nil } s.publishTurnEvent(events.TurnCompleted, turn) @@ -2351,6 +2365,10 @@ func (s *Service) publishTurnEvent(eventType string, turn *models.Turn) { if s.eventBus == nil { return } + if turn == nil { + s.logger.Warn("publishTurnEvent: turn is nil, skipping", zap.String("event_type", eventType)) + return + } _ = s.eventBus.Publish(context.Background(), eventType, bus.NewEvent(eventType, "task-service", map[string]interface{}{ "id": turn.ID, "session_id": turn.TaskSessionID, diff --git a/apps/backend/pkg/agent/protocol.go b/apps/backend/pkg/agent/protocol.go index ba9cd73639..00256e176d 100644 --- a/apps/backend/pkg/agent/protocol.go +++ b/apps/backend/pkg/agent/protocol.go @@ -23,6 +23,13 @@ const ( // OpenCode spawns an HTTP server and communicates via REST API calls // with Server-Sent Events (SSE) for streaming responses. ProtocolOpenCode Protocol = "opencode" + // ProtocolCopilot is the GitHub Copilot SDK protocol. + // Uses the official Go SDK which internally communicates via JSON-RPC with the Copilot CLI. + ProtocolCopilot Protocol = "copilot" + // ProtocolAmp is the Sourcegraph Amp protocol (stream-json over stdin/stdout). + // Amp uses a streaming JSON format similar to Claude Code, with thread-based + // session management for multi-turn conversations. + ProtocolAmp Protocol = "amp" ) // String returns the string representation of the protocol. @@ -33,7 +40,7 @@ func (p Protocol) String() string { // IsValid returns true if the protocol is a known valid protocol. func (p Protocol) IsValid() bool { switch p { - case ProtocolACP, ProtocolREST, ProtocolMCP, ProtocolCodex, ProtocolClaudeCode, ProtocolOpenCode: + case ProtocolACP, ProtocolREST, ProtocolMCP, ProtocolCodex, ProtocolClaudeCode, ProtocolOpenCode, ProtocolCopilot, ProtocolAmp: return true default: return false diff --git a/apps/backend/pkg/amp/client.go b/apps/backend/pkg/amp/client.go new file mode 100644 index 0000000000..43930a8eea --- /dev/null +++ b/apps/backend/pkg/amp/client.go @@ -0,0 +1,288 @@ +// Package amp provides a client wrapper for the Sourcegraph Amp CLI. +// This package handles communication with the Amp CLI via stream-json protocol. +package amp + +import ( + "bufio" + "context" + "encoding/json" + "fmt" + "io" + "sync" + + "github.com/kandev/kandev/internal/common/logger" + "go.uber.org/zap" +) + +// Message types for Amp stream-json protocol. +const ( + MessageTypeSystem = "system" + MessageTypeAssistant = "assistant" + MessageTypeResult = "result" + MessageTypeUser = "user" +) + +// Content block types. +const ( + ContentTypeText = "text" + ContentTypeThinking = "thinking" + ContentTypeToolUse = "tool_use" + ContentTypeToolResult = "tool_result" +) + +// Client wraps communication with Amp CLI via stdin/stdout. +type Client struct { + stdin io.Writer + stdout io.Reader + logger *logger.Logger + + scanner *bufio.Scanner + + // Message handler + messageHandler func(*Message) + handlerMu sync.RWMutex + + // Context for lifecycle + ctx context.Context + cancel context.CancelFunc + wg sync.WaitGroup + + // State + threadID string + mu sync.RWMutex +} + +// Message represents a message from Amp. +type Message struct { + Type string `json:"type"` + ThreadID string `json:"thread_id,omitempty"` + Message *Content `json:"message,omitempty"` + + // Result fields (when Type == "result") + IsError bool `json:"is_error,omitempty"` + Error string `json:"error,omitempty"` // Error message when is_error is true + Subtype string `json:"subtype,omitempty"` // e.g., "error_during_execution" + CostUSD float64 `json:"cost_usd,omitempty"` + DurationMS int64 `json:"duration_ms,omitempty"` + NumTurns int `json:"num_turns,omitempty"` + TotalInputTokens int64 `json:"total_input_tokens,omitempty"` + TotalOutputTokens int64 `json:"total_output_tokens,omitempty"` + Result json.RawMessage `json:"result,omitempty"` + ModelUsage map[string]*Usage `json:"model_usage,omitempty"` +} + +// Content represents the message content. +type Content struct { + Model string `json:"model,omitempty"` + Content []ContentBlock `json:"content,omitempty"` + Usage *TokenUsage `json:"usage,omitempty"` + StopReason string `json:"stop_reason,omitempty"` + StopSequence string `json:"stop_sequence,omitempty"` +} + +// ContentBlock represents a content block (text, tool_use, etc.). +type ContentBlock struct { + Type string `json:"type"` + Text string `json:"text,omitempty"` + Thinking string `json:"thinking,omitempty"` + ID string `json:"id,omitempty"` + Name string `json:"name,omitempty"` + Input map[string]any `json:"input,omitempty"` + ToolUseID string `json:"tool_use_id,omitempty"` + Content any `json:"content,omitempty"` + IsError bool `json:"is_error,omitempty"` +} + +// TokenUsage represents token usage information. +type TokenUsage struct { + InputTokens int64 `json:"input_tokens,omitempty"` + OutputTokens int64 `json:"output_tokens,omitempty"` + CacheCreationInputTokens int64 `json:"cache_creation_input_tokens,omitempty"` + CacheReadInputTokens int64 `json:"cache_read_input_tokens,omitempty"` +} + +// Usage represents model-specific usage stats. +type Usage struct { + InputTokens int64 `json:"input_tokens,omitempty"` + OutputTokens int64 `json:"output_tokens,omitempty"` + ContextWindow *int64 `json:"context_window,omitempty"` +} + +// ResultData represents the result data structure. +type ResultData struct { + SessionID string `json:"session_id,omitempty"` + ThreadID string `json:"thread_id,omitempty"` + Text string `json:"text,omitempty"` +} + +// UserMessage represents a user message to send (Claude Code-compatible format). +type UserMessage struct { + Type string `json:"type"` // "user" + Message UserMessageBody `json:"message"` +} + +// UserMessageBody is the nested message body for user messages. +type UserMessageBody struct { + Role string `json:"role"` // "user" + Content []UserContentBlock `json:"content"` +} + +// UserContentBlock represents a content block in user messages. +type UserContentBlock struct { + Type string `json:"type"` // "text" + Text string `json:"text"` +} + +// NewClient creates a new Amp client. +func NewClient(stdin io.Writer, stdout io.Reader, log *logger.Logger) *Client { + ctx, cancel := context.WithCancel(context.Background()) + return &Client{ + stdin: stdin, + stdout: stdout, + logger: log.WithFields(zap.String("component", "amp-client")), + ctx: ctx, + cancel: cancel, + } +} + +// SetMessageHandler sets the handler for incoming messages. +func (c *Client) SetMessageHandler(handler func(*Message)) { + c.handlerMu.Lock() + defer c.handlerMu.Unlock() + c.messageHandler = handler +} + +// Start begins reading from stdout and dispatching messages. +func (c *Client) Start(ctx context.Context) { + c.scanner = bufio.NewScanner(c.stdout) + // Allow larger buffer for JSON messages + c.scanner.Buffer(make([]byte, 64*1024), 1024*1024) + + c.wg.Add(1) + go c.readLoop(ctx) +} + +// Stop stops the client and waits for goroutines to finish. +func (c *Client) Stop() { + c.cancel() + c.wg.Wait() +} + +// readLoop reads JSON lines from stdout and dispatches them. +func (c *Client) readLoop(ctx context.Context) { + defer c.wg.Done() + + for { + select { + case <-ctx.Done(): + return + case <-c.ctx.Done(): + return + default: + } + + if !c.scanner.Scan() { + if err := c.scanner.Err(); err != nil { + c.logger.Error("scanner error", zap.Error(err)) + } + return + } + + line := c.scanner.Text() + if line == "" { + continue + } + + c.handleLine(line) + } +} + +// handleLine parses and dispatches a JSON line. +func (c *Client) handleLine(line string) { + var msg Message + if err := json.Unmarshal([]byte(line), &msg); err != nil { + c.logger.Debug("failed to parse message", zap.Error(err), zap.String("line", line)) + return + } + + // Update thread ID if provided + if msg.ThreadID != "" { + c.mu.Lock() + c.threadID = msg.ThreadID + c.mu.Unlock() + } + + c.handlerMu.RLock() + handler := c.messageHandler + c.handlerMu.RUnlock() + + if handler != nil { + handler(&msg) + } +} + +// SendUserMessage sends a user message to Amp. +func (c *Client) SendUserMessage(message string) error { + msg := &UserMessage{ + Type: "user", + Message: UserMessageBody{ + Role: "user", + Content: []UserContentBlock{ + {Type: "text", Text: message}, + }, + }, + } + + data, err := json.Marshal(msg) + if err != nil { + return fmt.Errorf("failed to marshal message: %w", err) + } + + if _, err := c.stdin.Write(append(data, '\n')); err != nil { + return fmt.Errorf("failed to write message: %w", err) + } + + return nil +} + +// GetThreadID returns the current thread ID. +func (c *Client) GetThreadID() string { + c.mu.RLock() + defer c.mu.RUnlock() + return c.threadID +} + +// SetThreadID sets the current thread ID. +func (c *Client) SetThreadID(threadID string) { + c.mu.Lock() + defer c.mu.Unlock() + c.threadID = threadID +} + +// GetResultData extracts ResultData from a result message. +func (m *Message) GetResultData() *ResultData { + if len(m.Result) == 0 { + return nil + } + + var data ResultData + if err := json.Unmarshal(m.Result, &data); err != nil { + return nil + } + + return &data +} + +// GetResultString extracts a string result (for error messages). +func (m *Message) GetResultString() string { + if len(m.Result) == 0 { + return "" + } + + var s string + if err := json.Unmarshal(m.Result, &s); err != nil { + return "" + } + + return s +} diff --git a/apps/backend/pkg/copilot/client.go b/apps/backend/pkg/copilot/client.go new file mode 100644 index 0000000000..79e8994306 --- /dev/null +++ b/apps/backend/pkg/copilot/client.go @@ -0,0 +1,350 @@ +// Package copilot provides integration with the GitHub Copilot SDK. +// This is a thin wrapper around github.com/github/copilot-sdk/go that +// provides a consistent interface for the kandev agent system. +// +// When CLIUrl is configured, the SDK connects to an externally managed +// Copilot CLI server via TCP (JSON-RPC). Otherwise, the SDK spawns and +// manages the CLI process internally via stdio. +package copilot + +import ( + "context" + "fmt" + "sync" + "time" + + "github.com/github/copilot-sdk/go" + "github.com/kandev/kandev/internal/common/logger" + "go.uber.org/zap" +) + +// Re-export SDK types for convenience +type ( + SessionEvent = copilot.SessionEvent + SessionEventType = copilot.SessionEventType + SessionConfig = copilot.SessionConfig + MessageOptions = copilot.MessageOptions + Data = copilot.Data +) + +// Re-export event type constants +const ( + EventTypeSessionStart = copilot.SessionStart + EventTypeSessionResume = copilot.SessionResume + EventTypeSessionIdle = copilot.SessionIdle + EventTypeSessionError = copilot.SessionError + EventTypeSessionUsageInfo = copilot.SessionUsageInfo + EventTypeAssistantMessage = copilot.AssistantMessage + EventTypeAssistantMessageDelta = copilot.AssistantMessageDelta + EventTypeAssistantReasoning = copilot.AssistantReasoning + EventTypeAssistantReasoningDelta = copilot.AssistantReasoningDelta + EventTypeAssistantTurnStart = copilot.AssistantTurnStart + EventTypeAssistantTurnEnd = copilot.AssistantTurnEnd + EventTypeAssistantUsage = copilot.AssistantUsage + EventTypeToolStart = copilot.ToolExecutionStart + EventTypeToolComplete = copilot.ToolExecutionComplete + EventTypeToolProgress = copilot.ToolExecutionProgress + EventTypeAbort = copilot.Abort +) + +// Client wraps the Copilot SDK client with additional functionality. +type Client struct { + sdkClient *copilot.Client + session *copilot.Session + logger *logger.Logger + + // Configuration + cliURL string + model string + + // Event handler + eventHandler func(SessionEvent) + unsubscribe func() + handlerMu sync.RWMutex + + // State + sessionID string + mu sync.RWMutex + started bool +} + +// ClientConfig holds configuration for creating a Client. +type ClientConfig struct { + // CLIUrl is the address of an externally managed Copilot CLI server (e.g. "localhost:12345"). + // When set, the SDK connects via TCP instead of spawning its own process. + CLIUrl string + Model string +} + +// NewClient creates a new Copilot client wrapper. +func NewClient(cfg ClientConfig, log *logger.Logger) *Client { + if cfg.Model == "" { + cfg.Model = "gpt-4.1" + } + + return &Client{ + cliURL: cfg.CLIUrl, + model: cfg.Model, + logger: log.WithFields(zap.String("component", "copilot-sdk-client")), + } +} + +// SetEventHandler sets the handler for session events. +func (c *Client) SetEventHandler(handler func(SessionEvent)) { + c.handlerMu.Lock() + defer c.handlerMu.Unlock() + c.eventHandler = handler +} + +// Start initializes the Copilot SDK client. +// When CLIUrl is configured, the SDK connects to an external CLI server via TCP. +// Otherwise, the SDK spawns and manages the CLI process internally via stdio. +// The actual connection is deferred to the first CreateSession call via AutoStart. +func (c *Client) Start(ctx context.Context) error { + c.mu.Lock() + defer c.mu.Unlock() + + if c.started { + return fmt.Errorf("client already started") + } + + c.logger.Info("starting Copilot SDK client", + zap.String("model", c.model), + zap.String("cli_url", c.cliURL)) + + if c.cliURL != "" { + // Connect to externally managed CLI server via TCP + c.sdkClient = copilot.NewClient(&copilot.ClientOptions{ + CLIUrl: c.cliURL, + LogLevel: "error", + }) + } else { + // SDK spawns and manages the CLI process internally (stdio) + c.sdkClient = copilot.NewClient(nil) + } + + // SDK AutoStart (default: true) defers the actual connection + // to the first CreateSession call, so no explicit Start() needed. + + c.started = true + c.logger.Info("Copilot SDK client initialized") + + return nil +} + +// Stop shuts down the Copilot SDK client. +func (c *Client) Stop() error { + c.mu.Lock() + defer c.mu.Unlock() + + if !c.started { + return nil + } + + c.logger.Info("stopping Copilot SDK client") + + // Unsubscribe from events + c.handlerMu.Lock() + if c.unsubscribe != nil { + c.unsubscribe() + c.unsubscribe = nil + } + c.handlerMu.Unlock() + + // Destroy session if active + if c.session != nil { + if err := c.session.Destroy(); err != nil { + c.logger.Warn("error destroying session", zap.Error(err)) + } + c.session = nil + } + + // Stop the SDK client + if c.sdkClient != nil { + errs := c.sdkClient.Stop() + for _, err := range errs { + c.logger.Warn("error stopping SDK client", zap.Error(err)) + } + c.sdkClient = nil + } + + c.started = false + return nil +} + +// CreateSession creates a new Copilot session. +func (c *Client) CreateSession(ctx context.Context) (string, error) { + c.mu.Lock() + defer c.mu.Unlock() + + if !c.started { + return "", fmt.Errorf("client not started") + } + + // Unsubscribe from previous session events + c.handlerMu.Lock() + if c.unsubscribe != nil { + c.unsubscribe() + c.unsubscribe = nil + } + c.handlerMu.Unlock() + + // Destroy existing session if any + if c.session != nil { + if err := c.session.Destroy(); err != nil { + c.logger.Warn("error destroying previous session", zap.Error(err)) + } + c.session = nil + } + + c.logger.Info("creating new session", zap.String("model", c.model)) + + // Create session with configuration + session, err := c.sdkClient.CreateSession(&copilot.SessionConfig{ + Model: c.model, + Streaming: true, + }) + if err != nil { + return "", fmt.Errorf("failed to create session: %w", err) + } + + c.session = session + c.sessionID = session.SessionID + + // Register event handler + c.handlerMu.Lock() + if c.eventHandler != nil { + c.unsubscribe = session.On(c.eventHandler) + } + c.handlerMu.Unlock() + + c.logger.Info("session created", zap.String("session_id", c.sessionID)) + + return c.sessionID, nil +} + +// ResumeSession resumes an existing session with streaming enabled. +func (c *Client) ResumeSession(ctx context.Context, sessionID string) error { + c.mu.Lock() + defer c.mu.Unlock() + + if !c.started { + return fmt.Errorf("client not started") + } + + // Unsubscribe from previous session events + c.handlerMu.Lock() + if c.unsubscribe != nil { + c.unsubscribe() + c.unsubscribe = nil + } + c.handlerMu.Unlock() + + // Destroy existing session if any + if c.session != nil { + if err := c.session.Destroy(); err != nil { + c.logger.Warn("error destroying previous session", zap.Error(err)) + } + c.session = nil + } + + c.logger.Info("resuming session", zap.String("session_id", sessionID)) + + // Use ResumeSessionWithOptions to enable streaming on resumed sessions + session, err := c.sdkClient.ResumeSessionWithOptions(sessionID, &copilot.ResumeSessionConfig{ + Streaming: true, + }) + if err != nil { + return fmt.Errorf("failed to resume session: %w", err) + } + + c.session = session + c.sessionID = sessionID + + // Register event handler + c.handlerMu.Lock() + if c.eventHandler != nil { + c.unsubscribe = session.On(c.eventHandler) + } + c.handlerMu.Unlock() + + c.logger.Info("session resumed", zap.String("session_id", sessionID)) + + return nil +} + +// Send sends a message to the current session (non-blocking). +func (c *Client) Send(ctx context.Context, message string) (string, error) { + c.mu.RLock() + session := c.session + c.mu.RUnlock() + + if session == nil { + return "", fmt.Errorf("no active session") + } + + c.logger.Info("sending message to session") + + messageID, err := session.Send(copilot.MessageOptions{ + Prompt: message, + }) + if err != nil { + return "", fmt.Errorf("failed to send message: %w", err) + } + + return messageID, nil +} + +// SendAndWait sends a message and waits for completion. +func (c *Client) SendAndWait(ctx context.Context, message string, timeout time.Duration) (*SessionEvent, error) { + c.mu.RLock() + session := c.session + c.mu.RUnlock() + + if session == nil { + return nil, fmt.Errorf("no active session") + } + + if timeout == 0 { + timeout = 5 * time.Minute + } + + c.logger.Info("sending message and waiting for completion") + + result, err := session.SendAndWait(copilot.MessageOptions{ + Prompt: message, + }, timeout) + if err != nil { + return nil, fmt.Errorf("failed to send message: %w", err) + } + + return result, nil +} + +// Abort cancels the current operation. +func (c *Client) Abort(ctx context.Context) error { + c.mu.RLock() + session := c.session + c.mu.RUnlock() + + if session == nil { + return nil + } + + c.logger.Info("aborting current operation") + return session.Abort() +} + +// GetSessionID returns the current session ID. +func (c *Client) GetSessionID() string { + c.mu.RLock() + defer c.mu.RUnlock() + return c.sessionID +} + +// IsStarted returns whether the client has been started. +func (c *Client) IsStarted() bool { + c.mu.RLock() + defer c.mu.RUnlock() + return c.started +} diff --git a/apps/web/components/task/chat/messages/chat-message.tsx b/apps/web/components/task/chat/messages/chat-message.tsx index 4d8ea7df3e..2f7bba90b8 100644 --- a/apps/web/components/task/chat/messages/chat-message.tsx +++ b/apps/web/components/task/chat/messages/chat-message.tsx @@ -194,6 +194,38 @@ export const ChatMessage = memo(function ChatMessage({ comment, label, className {children}

), + table: ({ children }) => ( +
+ + {children} +
+
+ ), + thead: ({ children }) => ( + + {children} + + ), + tbody: ({ children }) => ( + + {children} + + ), + tr: ({ children }) => ( + + {children} + + ), + th: ({ children }) => ( + + {children} + + ), + td: ({ children }) => ( + + {children} + + ), }} > {comment.content || '(empty)'} diff --git a/apps/web/components/task/chat/messages/tool-call-message.tsx b/apps/web/components/task/chat/messages/tool-call-message.tsx index d10efe9e99..cee90087f2 100644 --- a/apps/web/components/task/chat/messages/tool-call-message.tsx +++ b/apps/web/components/task/chat/messages/tool-call-message.tsx @@ -1,10 +1,8 @@ 'use client'; -import { useCallback, useEffect, useState, memo } from 'react'; +import { useCallback, useState, useRef, memo } from 'react'; import { IconCheck, - IconChevronDown, - IconChevronRight, IconCode, IconEdit, IconEye, @@ -14,11 +12,12 @@ import { IconX, } from '@tabler/icons-react'; import { GridSpinner } from '@/components/grid-spinner'; -import { cn, toRelativePath, transformPathsInText } from '@/lib/utils'; +import { cn, transformPathsInText } from '@/lib/utils'; import { getWebSocketClient } from '@/lib/ws/connection'; import type { Message } from '@/lib/types/http'; import type { ToolCallMetadata } from '@/components/task/chat/types'; import { PermissionActionRow } from './permission-action-row'; +import { ExpandableRow } from './expandable-row'; type PermissionOption = { option_id: string; @@ -59,28 +58,6 @@ function getToolIcon(toolName: string | undefined, className: string) { return ; } -function getStatusIcon(status?: string, permissionStatus?: string) { - // If there's a permission status, show that instead - if (permissionStatus === 'approved') { - return ; - } - if (permissionStatus === 'rejected') { - return ; - } - - switch (status) { - case 'complete': - return ; - case 'error': - return ; - case 'running': - case 'in_progress': - return ; - default: - return null; - } -} - type ToolCallMessageProps = { comment: Message; permissionMessage?: Message; @@ -95,38 +72,40 @@ export const ToolCallMessage = memo(function ToolCallMessage({ comment, permissi const rawTitle = metadata?.title ?? comment.content ?? 'Tool call'; const title = transformPathsInText(rawTitle, worktreePath); const status = metadata?.status; - const args = metadata?.args; - const result = metadata?.result; + + // Get output from normalized payload (generic tools) + const normalizedGeneric = metadata?.normalized?.generic; + const output = normalizedGeneric?.output ?? metadata?.result; // Permission state const hasPermission = !!permissionMessage; const permissionStatus = permissionMetadata?.status; const isPermissionPending = hasPermission && permissionStatus !== 'approved' && permissionStatus !== 'rejected'; - // Initialize isExpanded based on status - auto-expand if running - const [isExpanded, setIsExpanded] = useState(() => { - return status === 'running'; - }); - const [isManuallyToggled, setIsManuallyToggled] = useState(false); + // Expand state management + const [manualExpandState, setManualExpandState] = useState(null); + const prevStatusRef = useRef(status); const [isResponding, setIsResponding] = useState(false); - // Auto-collapse when complete (unless manually overridden) - useEffect(() => { - // Permission requests always stay expanded - if (isPermissionPending) { - setIsExpanded(true); - return; + // Reset manual state when status changes + if (prevStatusRef.current !== status) { + prevStatusRef.current = status; + if (manualExpandState !== null) { + setManualExpandState(null); } + } - // Only auto-collapse if user hasn't manually toggled - if (!isManuallyToggled) { - if (status === 'running') { - setIsExpanded(true); - } else if (status === 'complete' || status === 'error') { - setIsExpanded(false); - } - } - }, [status, isManuallyToggled, isPermissionPending]); + // Auto-expand when running or permission pending + const autoExpanded = status === 'running' || isPermissionPending; + const isExpanded = manualExpandState ?? autoExpanded; + + // Determine if we have expandable content + const hasOutput = output && (typeof output === 'string' ? output.length > 0 : Object.keys(output).length > 0); + const hasExpandableContent = hasOutput || isPermissionPending; + + const handleToggle = useCallback(() => { + setManualExpandState((prev) => !(prev ?? autoExpanded)); + }, [autoExpanded]); const handleRespond = useCallback( async (optionId: string, cancelled: boolean = false) => { @@ -175,123 +154,88 @@ export const ToolCallMessage = memo(function ToolCallMessage({ comment, permissi } }, [permissionMetadata, handleRespond]); - const hasDetails = args && Object.keys(args).length > 0; - - let filePath: string | undefined; - const rawPath = args?.path ?? args?.file ?? args?.file_path; - if (typeof rawPath === 'string') { - filePath = rawPath; - } - if (!filePath && Array.isArray(args?.locations) && args.locations.length > 0) { - const firstLoc = args.locations[0] as { path?: string } | undefined; - if (firstLoc?.path) { - filePath = firstLoc.path; - } - } - const isSuccess = status === 'complete' && !permissionStatus; - const handleToggleExpand = () => { - if (hasDetails) { - setIsExpanded(!isExpanded); - setIsManuallyToggled(true); + const getStatusIcon = () => { + // If there's a permission status, show that instead + if (permissionStatus === 'approved') { + return ; + } + if (permissionStatus === 'rejected') { + return ; + } + + switch (status) { + case 'complete': + return ; + case 'error': + return ; + case 'running': + return ; + default: + return null; } }; + // Format output for display - beautify JSON if detected + const formatOutput = (value: unknown): { content: string; isJson: boolean } => { + if (typeof value === 'string') { + // Try to parse as JSON and beautify + try { + const parsed = JSON.parse(value); + return { content: JSON.stringify(parsed, null, 2), isJson: true }; + } catch { + // Not valid JSON, return as-is + return { content: value, isJson: false }; + } + } + // Already an object, stringify with formatting + return { content: JSON.stringify(value, null, 2), isJson: true }; + }; + + const formattedOutput = hasOutput ? formatOutput(output) : null; + return ( -
-
- {/* Icon with hover-to-show chevron */} -
- {/* Show tool icon by default, chevron on hover if expandable */} -
+ - {getToolIcon(toolName, cn( - 'h-4 w-4', - isPermissionPending ? 'text-amber-600 dark:text-amber-400' : 'text-muted-foreground' - ))} -
- {hasDetails && ( - isExpanded - ? - : - )} + {title} + {!isSuccess && getStatusIcon()} +
+ } + hasExpandableContent={!!hasExpandableContent} + isExpanded={isExpanded} + onToggle={handleToggle} + > +
+ {/* Output display */} + {formattedOutput && ( + 
+            {formattedOutput.content}
+
+ )} - {/* Content */} -
-
- - - {title} - - {/* Status indicator - only show if not success */} - {!isSuccess && getStatusIcon(status, permissionStatus)} - - {filePath && ( - - {toRelativePath(filePath, worktreePath)} - - )} -
- - {/* Expanded Details */} - {isExpanded && hasDetails && ( -
- {args && Object.entries(args).map(([key, value]) => { - const strValue = typeof value === 'string' ? value : JSON.stringify(value, null, 2); - const isLongValue = strValue.length > 100 || strValue.includes('\n'); - - return ( -
- {key}: - {isLongValue ? ( - 
-                        {strValue}
-
- ) : ( - {strValue} - )} -
- ); - })} - {result && ( -
- Result: - 
-                    {result}
-
-
- )} -
- )} - - {/* Inline permission approval UI */} - {isPermissionPending && ( -
- -
- )} -
+ {/* Inline permission approval UI */} + {isPermissionPending && ( + + )}
-
+ ); }); diff --git a/apps/web/components/task/task-layout.tsx b/apps/web/components/task/task-layout.tsx index e23b2c8f64..0bc7fdc527 100644 --- a/apps/web/components/task/task-layout.tsx +++ b/apps/web/components/task/task-layout.tsx @@ -100,7 +100,7 @@ export const TaskLayout = memo(function TaskLayout({ }); const previewPanelIds = ['chat', 'preview']; - const previewLayoutKey = `task-layout-preview-v2:${sessionKey}`; + const previewLayoutKey = 'task-layout-preview-v2'; const { defaultLayout: defaultPreviewLayout, onLayoutChanged: onPreviewLayoutChange } = useDefaultLayout({ id: previewLayoutKey, diff --git a/apps/web/components/task/task-right-panel.tsx b/apps/web/components/task/task-right-panel.tsx index 6c9eb6e0ba..246c7dd116 100644 --- a/apps/web/components/task/task-right-panel.tsx +++ b/apps/web/components/task/task-right-panel.tsx @@ -32,7 +32,7 @@ const DEFAULT_RIGHT_LAYOUT: Record = { top: 55, bottom: 45 }; const TaskRightPanel = memo(function TaskRightPanel({ topPanel, sessionId = null }: TaskRightPanelProps) { const rightPanelIds = ['top', 'bottom']; - const rightLayoutKey = `task-layout-right-v2:${sessionId ?? 'default'}`; + const rightLayoutKey = 'task-layout-right-v2'; const { defaultLayout: rightLayout, onLayoutChanged: onRightLayoutChange } = useDefaultLayout({ id: rightLayoutKey, panelIds: rightPanelIds,