diff --git a/apps/backend/cmd/kandev/main.go b/apps/backend/cmd/kandev/main.go index e41fed4233..3e5390affe 100644 --- a/apps/backend/cmd/kandev/main.go +++ b/apps/backend/cmd/kandev/main.go @@ -591,7 +591,11 @@ func main() { if metadata, ok := data["metadata"]; ok && metadata != nil { payload["metadata"] = metadata } - notification, _ := ws.NewNotification(ws.ActionCommentAdded, payload) + notification, err := ws.NewNotification(ws.ActionCommentAdded, payload) + if err != nil { + log.Error("Failed to create comment.added notification", zap.Error(err)) + return nil + } gateway.Hub.BroadcastToTask(taskID, notification) return nil }) @@ -601,6 +605,41 @@ func main() { log.Info("Subscribed to comment.added events for WebSocket broadcasting") } + // Subscribe to comment.updated events and broadcast to WebSocket subscribers + _, err = eventBus.Subscribe(events.CommentUpdated, func(ctx context.Context, event *bus.Event) error { + data := event.Data + taskID, _ := data["task_id"].(string) + if taskID == "" { + log.Warn("comment.updated event has no task_id, skipping") + return nil + } + payload := map[string]interface{}{ + "comment_id": data["comment_id"], + "task_id": taskID, + "author_type": data["author_type"], + "author_id": data["author_id"], + "content": data["content"], + "type": data["type"], + "requests_input": data["requests_input"], + "created_at": data["created_at"], + } + if metadata, ok := data["metadata"]; ok && metadata != nil { + payload["metadata"] = metadata + } + notification, err := ws.NewNotification(ws.ActionCommentUpdated, payload) + if err != nil { + log.Error("Failed to create comment.updated notification", zap.Error(err)) + return nil + } + gateway.Hub.BroadcastToTask(taskID, notification) + return nil + }) + if err != nil { + log.Error("Failed to subscribe to comment.updated events", zap.Error(err)) + } else { + log.Info("Subscribed to comment.updated events for WebSocket broadcasting") + } + // Subscribe to git status events and broadcast to WebSocket subscribers _, err = eventBus.Subscribe(events.BuildGitStatusWildcardSubject(), func(ctx context.Context, event *bus.Event) error { data := event.Data @@ -996,6 +1035,11 @@ func (a *commentCreatorAdapter) CreateToolCallComment(ctx context.Context, taskI return err } +// UpdateToolCallComment updates a tool call comment's status +func (a *commentCreatorAdapter) UpdateToolCallComment(ctx context.Context, taskID, toolCallID, status, result string) error { + return a.svc.UpdateToolCallComment(ctx, taskID, toolCallID, status, result) +} + // mapACPTypeToCommentType maps ACP message types to comment types func mapACPTypeToCommentType(msgType protocol.MessageType) string { switch msgType { diff --git a/apps/backend/internal/agent/agentctl/acp.go b/apps/backend/internal/agent/agentctl/acp.go index eeef7d61c1..ce20759723 100644 --- a/apps/backend/internal/agent/agentctl/acp.go +++ b/apps/backend/internal/agent/agentctl/acp.go @@ -152,17 +152,31 @@ func (c *Client) Prompt(ctx context.Context, text string) (*PromptResponse, erro // SessionUpdate represents a session update from the agent. // This matches adapter.SessionUpdate that agentctl sends over WebSocket. type SessionUpdate struct { - Type string `json:"type"` - SessionID string `json:"session_id,omitempty"` - Text string `json:"text,omitempty"` + Type string `json:"type"` + SessionID string `json:"session_id,omitempty"` + OperationID string `json:"operation_id,omitempty"` // Turn/operation ID for cancellation + + // Message fields + Text string `json:"text,omitempty"` + + // Reasoning fields (for "reasoning" type) + ReasoningText string `json:"reasoning_text,omitempty"` + ReasoningSummary string `json:"reasoning_summary,omitempty"` + + // Tool call fields ToolCallID string `json:"tool_call_id,omitempty"` ToolName string `json:"tool_name,omitempty"` ToolTitle string `json:"tool_title,omitempty"` ToolStatus string `json:"tool_status,omitempty"` ToolArgs map[string]interface{} `json:"tool_args,omitempty"` ToolResult interface{} `json:"tool_result,omitempty"` - Error string `json:"error,omitempty"` - Data map[string]interface{} `json:"data,omitempty"` + + // Diff for file changes + Diff string `json:"diff,omitempty"` + + // Error and raw data + Error string `json:"error,omitempty"` + Data map[string]interface{} `json:"data,omitempty"` } // PermissionOption represents a permission choice diff --git a/apps/backend/internal/agent/agentctl/standalone.go b/apps/backend/internal/agent/agentctl/standalone.go index 4894cee0fd..841c5176d7 100644 --- a/apps/backend/internal/agent/agentctl/standalone.go +++ b/apps/backend/internal/agent/agentctl/standalone.go @@ -27,7 +27,8 @@ type CreateInstanceRequest struct { ID string `json:"id,omitempty"` WorkspacePath string `json:"workspace_path"` AgentCommand string `json:"agent_command,omitempty"` - Protocol string `json:"protocol,omitempty"` // Protocol adapter to use (acp, rest, mcp) + Protocol string `json:"protocol,omitempty"` // Protocol adapter to use (acp, rest, mcp) + WorkspaceFlag string `json:"workspace_flag,omitempty"` // CLI flag for workspace path (e.g., "--workspace-root") Env map[string]string `json:"env,omitempty"` AutoStart bool `json:"auto_start,omitempty"` } diff --git a/apps/backend/internal/agent/lifecycle/manager.go b/apps/backend/internal/agent/lifecycle/manager.go index fa7bbabaaa..593e3a7ce8 100644 --- a/apps/backend/internal/agent/lifecycle/manager.go +++ b/apps/backend/internal/agent/lifecycle/manager.go @@ -51,9 +51,11 @@ type AgentInstance struct { standaloneInstanceID string // Instance ID in standalone agentctl standalonePort int // Port of the standalone instance - // Message buffer for accumulating agent response during a prompt - messageBuffer strings.Builder - messageMu sync.Mutex + // Buffers for accumulating agent response during a prompt + messageBuffer strings.Builder + reasoningBuffer strings.Builder + summaryBuffer strings.Builder + messageMu sync.Mutex } // GetAgentCtlClient returns the agentctl client for this instance @@ -583,7 +585,8 @@ func (m *Manager) launchStandalone(ctx context.Context, instanceID string, req * ID: instanceID, WorkspacePath: req.WorkspacePath, AgentCommand: agentCommand, - Protocol: string(agentConfig.Protocol), // Pass protocol from registry + Protocol: string(agentConfig.Protocol), // Pass protocol from registry + WorkspaceFlag: agentConfig.WorkspaceFlag, // CLI flag for workspace path Env: env, AutoStart: false, // We'll start via ACP initialize flow } @@ -816,7 +819,9 @@ func (m *Manager) initializeACPSession(ctx context.Context, instance *AgentInsta m.publishSessionInfo(instance, sessionID) // 3. Set up updates stream to receive session notifications - go m.handleUpdatesStream(instance) + // Use a ready channel to signal when the stream is connected + updatesReady := make(chan struct{}) + go m.handleUpdatesStream(instance, updatesReady) // 4. Set up permission stream to receive permission requests go m.handlePermissionStream(instance) @@ -827,6 +832,15 @@ func (m *Manager) initializeACPSession(ctx context.Context, instance *AgentInsta // 6. Set up file changes stream to track filesystem changes go m.handleFileChangesStream(instance) + // Wait for the updates stream to connect before sending prompt + // This prevents race conditions where notifications are sent before streams are ready + select { + case <-updatesReady: + m.logger.Debug("updates stream ready") + case <-time.After(5 * time.Second): + m.logger.Warn("timeout waiting for updates stream to connect, proceeding anyway") + } + // 7. Send the task prompt if provided if taskDescription != "" { m.logger.Info("sending ACP prompt", @@ -834,9 +848,11 @@ func (m *Manager) initializeACPSession(ctx context.Context, instance *AgentInsta zap.String("session_id", sessionID), zap.String("task_description", taskDescription)) - // Clear message buffer before starting prompt + // Clear buffers before starting prompt instance.messageMu.Lock() instance.messageBuffer.Reset() + instance.reasoningBuffer.Reset() + instance.summaryBuffer.Reset() instance.messageMu.Unlock() // Prompt is SYNCHRONOUS - it blocks until the agent completes the task @@ -852,10 +868,12 @@ func (m *Manager) initializeACPSession(ctx context.Context, instance *AgentInsta return fmt.Errorf("prompt failed: %w", err) } - // Extract accumulated message from buffer + // Extract accumulated content from buffers instance.messageMu.Lock() agentMessage := instance.messageBuffer.String() instance.messageBuffer.Reset() + instance.reasoningBuffer.Reset() + instance.summaryBuffer.Reset() instance.messageMu.Unlock() stopReason := "" @@ -867,7 +885,7 @@ func (m *Manager) initializeACPSession(ctx context.Context, instance *AgentInsta zap.String("stop_reason", stopReason)) // Publish prompt_complete event with the agent's response (for saving as comment) - m.publishPromptComplete(instance, agentMessage) + m.publishPromptComplete(instance, agentMessage, "", "") // Prompt completed - mark agent as READY for follow-up prompts m.logger.Info("agent ready for follow-up prompts", @@ -890,13 +908,22 @@ func (m *Manager) initializeACPSession(ctx context.Context, instance *AgentInsta return nil } -// handleUpdatesStream handles streaming session notifications from the agent -func (m *Manager) handleUpdatesStream(instance *AgentInstance) { +// handleUpdatesStream handles streaming session notifications from the agent. +// If ready is non-nil, it will be closed when the stream connection is established (or fails). +func (m *Manager) handleUpdatesStream(instance *AgentInstance, ready chan<- struct{}) { ctx := context.Background() err := instance.agentctl.StreamUpdates(ctx, func(update agentctl.SessionUpdate) { m.handleSessionUpdate(instance, update) }) + + // Signal that the stream connection attempt is complete (success or failure) + // StreamUpdates returns immediately after establishing the WebSocket connection + // and starting the read goroutine, so this signals that we're ready to receive updates + if ready != nil { + close(ready) + } + if err != nil { m.logger.Error("failed to connect to updates stream", zap.String("instance_id", instance.ID), @@ -969,7 +996,7 @@ func (m *Manager) reconnectStreams(instance *AgentInstance) { } // Reconnect to WebSocket streams - go m.handleUpdatesStream(instance) + go m.handleUpdatesStream(instance, nil) go m.handlePermissionStream(instance) go m.handleGitStatusStream(instance) go m.handleFileChangesStream(instance) @@ -1040,16 +1067,9 @@ func (m *Manager) publishPermissionRequest(instance *AgentInstance, notification // handleSessionUpdate processes incoming session updates from the agent func (m *Manager) handleSessionUpdate(instance *AgentInstance, update agentctl.SessionUpdate) { - m.logger.Debug("received session update", - zap.String("instance_id", instance.ID), - zap.String("session_id", update.SessionID), - zap.String("type", update.Type)) - // Handle different update types based on the Type field switch update.Type { case "message_chunk": - m.logger.Debug("agent message chunk", - zap.String("instance_id", instance.ID)) m.updateInstanceProgress(instance.ID, 50) // Accumulate message content for saving as comment when a step completes @@ -1059,6 +1079,17 @@ func (m *Manager) handleSessionUpdate(instance *AgentInstance, update agentctl.S instance.messageMu.Unlock() } + case "reasoning": + // Accumulate reasoning/thinking content + instance.messageMu.Lock() + if update.ReasoningText != "" { + instance.reasoningBuffer.WriteString(update.ReasoningText) + } + if update.ReasoningSummary != "" { + instance.summaryBuffer.WriteString(update.ReasoningSummary) + } + instance.messageMu.Unlock() + case "tool_call": // Tool call starting marks a step boundary - flush the accumulated message as a comment // This way, each agent response before a tool use becomes a separate comment @@ -1075,20 +1106,12 @@ func (m *Manager) handleSessionUpdate(instance *AgentInstance, update agentctl.S m.publishToolCall(instance, update.ToolCallID, update.ToolTitle, update.ToolStatus, update.ToolArgs) case "tool_update": - m.logger.Debug("tool call update", - zap.String("instance_id", instance.ID), - zap.String("tool_call_id", update.ToolCallID)) - // Check if tool call completed switch update.ToolStatus { case "complete", "completed": - m.logger.Info("tool call completed", - zap.String("instance_id", instance.ID)) m.updateInstanceProgress(instance.ID, 80) m.publishToolCallCompleteFromUpdate(instance, update) - case "error": - m.logger.Error("tool call error", - zap.String("instance_id", instance.ID)) + case "error", "failed": m.publishToolCallCompleteFromUpdate(instance, update) } @@ -1100,6 +1123,21 @@ func (m *Manager) handleSessionUpdate(instance *AgentInstance, update agentctl.S m.logger.Error("agent error", zap.String("instance_id", instance.ID), zap.String("error", update.Error)) + + case "complete": + m.logger.Info("agent turn complete", + zap.String("instance_id", instance.ID), + zap.String("session_id", update.SessionID)) + + // Flush accumulated message buffer as a comment + m.flushMessageBufferAsComment(instance) + + // Mark agent as READY for follow-up prompts + if err := m.MarkReady(instance.ID); err != nil { + m.logger.Error("failed to mark instance as ready after complete", + zap.String("instance_id", instance.ID), + zap.Error(err)) + } } // Publish session update to event bus for WebSocket streaming @@ -1320,7 +1358,7 @@ func (m *Manager) publishFileChange(instance *AgentInstance, notification *agent // publishPromptComplete publishes a prompt_complete event when an agent finishes responding // This is used to save the agent's response as a comment on the task -func (m *Manager) publishPromptComplete(instance *AgentInstance, agentMessage string) { +func (m *Manager) publishPromptComplete(instance *AgentInstance, agentMessage, reasoning, summary string) { if m.eventBus == nil { return } @@ -1337,6 +1375,12 @@ func (m *Manager) publishPromptComplete(instance *AgentInstance, agentMessage st "task_id": instance.TaskID, "agent_message": agentMessage, } + if reasoning != "" { + data["reasoning"] = reasoning + } + if summary != "" { + data["summary"] = summary + } event := bus.NewEvent(events.PromptComplete, "agent-manager", data) // Publish on task-specific subject so orchestrator can subscribe @@ -1347,10 +1391,6 @@ func (m *Manager) publishPromptComplete(instance *AgentInstance, agentMessage st zap.String("instance_id", instance.ID), zap.String("task_id", instance.TaskID), zap.Error(err)) - } else { - m.logger.Info("published prompt_complete event", - zap.String("task_id", instance.TaskID), - zap.Int("message_length", len(agentMessage))) } } @@ -1360,7 +1400,11 @@ func (m *Manager) publishPromptComplete(instance *AgentInstance, agentMessage st func (m *Manager) flushMessageBufferAsComment(instance *AgentInstance) { instance.messageMu.Lock() agentMessage := instance.messageBuffer.String() + reasoning := instance.reasoningBuffer.String() + summary := instance.summaryBuffer.String() instance.messageBuffer.Reset() + instance.reasoningBuffer.Reset() + instance.summaryBuffer.Reset() instance.messageMu.Unlock() // Only publish if there's actual content (ignore whitespace-only) @@ -1369,13 +1413,8 @@ func (m *Manager) flushMessageBufferAsComment(instance *AgentInstance) { return } - m.logger.Debug("flushing message buffer as step comment", - zap.String("instance_id", instance.ID), - zap.String("task_id", instance.TaskID), - zap.Int("message_length", len(agentMessage))) - // Reuse the prompt_complete event type - the orchestrator handles it the same way - m.publishPromptComplete(instance, agentMessage) + m.publishPromptComplete(instance, agentMessage, reasoning, summary) } // publishToolCall publishes a tool call start event (to be saved as a comment) @@ -1660,9 +1699,11 @@ func (m *Manager) PromptAgent(ctx context.Context, instanceID string, prompt str // Set status to RUNNING while processing instance.Status = v1.AgentStatusRunning - // Clear message buffer before starting new prompt + // Clear buffers before starting new prompt instance.messageMu.Lock() instance.messageBuffer.Reset() + instance.reasoningBuffer.Reset() + instance.summaryBuffer.Reset() instance.messageMu.Unlock() m.mu.Unlock() @@ -1677,10 +1718,12 @@ func (m *Manager) PromptAgent(ctx context.Context, instanceID string, prompt str return nil, err } - // Extract accumulated message from buffer + // Extract accumulated content from buffers instance.messageMu.Lock() agentMessage := instance.messageBuffer.String() instance.messageBuffer.Reset() + instance.reasoningBuffer.Reset() + instance.summaryBuffer.Reset() instance.messageMu.Unlock() result := &PromptResult{ @@ -1689,7 +1732,7 @@ func (m *Manager) PromptAgent(ctx context.Context, instanceID string, prompt str } // Publish prompt_complete event with the agent's response (for saving as comment) - m.publishPromptComplete(instance, agentMessage) + m.publishPromptComplete(instance, agentMessage, "", "") // Prompt completed - mark as READY for next prompt if err := m.MarkReady(instanceID); err != nil { diff --git a/apps/backend/internal/agent/registry/defaults.go b/apps/backend/internal/agent/registry/defaults.go index e74583bccc..b21276ad12 100644 --- a/apps/backend/internal/agent/registry/defaults.go +++ b/apps/backend/internal/agent/registry/defaults.go @@ -6,14 +6,14 @@ import "github.com/kandev/kandev/pkg/agent" func DefaultAgents() []*AgentTypeConfig { return []*AgentTypeConfig{ { - ID: "auggie-agent", - Name: "Augment Coding Agent", - Description: "Auggie CLI-powered autonomous coding agent. Requires AUGMENT_SESSION_AUTH for authentication.", - Image: "kandev/augment-agent", - Tag: "latest", - Cmd: []string{"auggie", "--acp"}, - WorkingDir: "/workspace", - RequiredEnv: []string{"AUGMENT_SESSION_AUTH"}, + ID: "auggie-agent", + Name: "Augment Coding Agent", + Description: "Auggie CLI-powered autonomous coding agent. Requires AUGMENT_SESSION_AUTH for authentication.", + Image: "kandev/augment-agent", + Tag: "latest", + Cmd: []string{"auggie", "--acp"}, + WorkingDir: "/workspace", + RequiredEnv: []string{"AUGMENT_SESSION_AUTH"}, Env: map[string]string{ "AGENTCTL_AUTO_APPROVE_PERMISSIONS": "true", }, @@ -26,10 +26,11 @@ func DefaultAgents() []*AgentTypeConfig { CPUCores: 2.0, TimeoutSeconds: 3600, }, - Capabilities: []string{"code_generation", "code_review", "refactoring", "testing", "shell_execution"}, - Enabled: true, - ModelFlag: "--model", - Protocol: agent.ProtocolACP, + Capabilities: []string{"code_generation", "code_review", "refactoring", "testing", "shell_execution"}, + Enabled: true, + ModelFlag: "--model", + WorkspaceFlag: "--workspace-root", + Protocol: agent.ProtocolACP, }, { ID: "claude-agent", @@ -57,10 +58,10 @@ func DefaultAgents() []*AgentTypeConfig { { ID: "codex-agent", Name: "OpenAI Codex Agent", - Description: "OpenAI Codex CLI-powered autonomous coding agent.", + Description: "OpenAI Codex CLI-powered autonomous coding agent using the Codex app-server protocol.", Image: "kandev/codex-agent", Tag: "latest", - Cmd: []string{"codex", "--acp"}, + Cmd: []string{"codex", "app-server"}, WorkingDir: "/workspace", RequiredEnv: []string{}, Env: map[string]string{}, @@ -75,7 +76,7 @@ func DefaultAgents() []*AgentTypeConfig { Capabilities: []string{"code_generation", "code_review", "refactoring", "testing", "shell_execution"}, Enabled: true, ModelFlag: "--model", - Protocol: agent.ProtocolACP, + Protocol: agent.ProtocolCodex, }, { ID: "gemini-agent", diff --git a/apps/backend/internal/agent/registry/registry.go b/apps/backend/internal/agent/registry/registry.go index 38d8b4b2cb..6cc5e5cea5 100644 --- a/apps/backend/internal/agent/registry/registry.go +++ b/apps/backend/internal/agent/registry/registry.go @@ -32,6 +32,7 @@ type AgentTypeConfig struct { Capabilities []string `json:"capabilities"` // What the agent can do Enabled bool `json:"enabled"` ModelFlag string `json:"model_flag,omitempty"` // CLI flag for model selection (e.g., "--model") + WorkspaceFlag string `json:"workspace_flag,omitempty"` // CLI flag for workspace path (e.g., "--workspace-root"), empty means use cwd only // Protocol configuration Protocol agent.Protocol `json:"protocol,omitempty"` // Communication protocol: "acp", "rest", "mcp" diff --git a/apps/backend/internal/agentctl/adapter/acp_adapter.go b/apps/backend/internal/agentctl/adapter/acp_adapter.go index 1eb657d41b..dfc8fb5c63 100644 --- a/apps/backend/internal/agentctl/adapter/acp_adapter.go +++ b/apps/backend/internal/agentctl/adapter/acp_adapter.go @@ -202,6 +202,13 @@ func (a *ACPAdapter) GetSessionID() string { return a.sessionID } +// GetOperationID returns the current operation/turn ID. +// ACP protocol doesn't have explicit turn/operation IDs, so this returns empty string. +func (a *ACPAdapter) GetOperationID() string { + // ACP doesn't have explicit operation/turn IDs + return "" +} + // SetPermissionHandler sets the handler for permission requests. func (a *ACPAdapter) SetPermissionHandler(handler PermissionHandler) { a.mu.Lock() @@ -253,16 +260,34 @@ func (a *ACPAdapter) handleACPUpdate(n acp.SessionNotification) { func (a *ACPAdapter) convertNotification(n acp.SessionNotification) *SessionUpdate { u := n.Update + // Log what update types we're receiving + a.logger.Debug("ACP notification received", + zap.Bool("has_tool_call", u.ToolCall != nil), + zap.Bool("has_tool_call_update", u.ToolCallUpdate != nil), + zap.Bool("has_message_chunk", u.AgentMessageChunk != nil), + zap.Bool("has_thought_chunk", u.AgentThoughtChunk != nil), + zap.Bool("has_plan", u.Plan != nil)) + switch { case u.AgentMessageChunk != nil: if u.AgentMessageChunk.Content.Text != nil { return &SessionUpdate{ - Type: "message_chunk", + Type: UpdateTypeMessageChunk, SessionID: string(n.SessionId), Text: u.AgentMessageChunk.Content.Text.Text, } } + case u.AgentThoughtChunk != nil: + // Agent thinking/reasoning - map to the new reasoning type + if u.AgentThoughtChunk.Content.Text != nil { + return &SessionUpdate{ + Type: UpdateTypeReasoning, + SessionID: string(n.SessionId), + ReasoningText: u.AgentThoughtChunk.Content.Text.Text, + } + } + case u.ToolCall != nil: // Extract rich tool call information args := map[string]interface{}{} @@ -293,13 +318,19 @@ func (a *ACPAdapter) convertNotification(n acp.SessionNotification) *SessionUpda args["raw_input"] = u.ToolCall.RawInput } + // Normalize status - if empty, default to "running" for tool_call start + status := string(u.ToolCall.Status) + if status == "" { + status = "running" + } + return &SessionUpdate{ - Type: "tool_call", + Type: UpdateTypeToolCall, SessionID: string(n.SessionId), ToolCallID: string(u.ToolCall.ToolCallId), ToolName: string(u.ToolCall.Kind), // Kind is effectively the tool name ToolTitle: u.ToolCall.Title, - ToolStatus: string(u.ToolCall.Status), + ToolStatus: status, ToolArgs: args, } @@ -308,8 +339,12 @@ func (a *ACPAdapter) convertNotification(n acp.SessionNotification) *SessionUpda if u.ToolCallUpdate.Status != nil { status = string(*u.ToolCallUpdate.Status) } + // Normalize status - "completed" → "complete" for frontend consistency + if status == "completed" { + status = "complete" + } return &SessionUpdate{ - Type: "tool_update", + Type: UpdateTypeToolUpdate, SessionID: string(n.SessionId), ToolCallID: string(u.ToolCallUpdate.ToolCallId), ToolStatus: status, @@ -325,7 +360,7 @@ func (a *ACPAdapter) convertNotification(n acp.SessionNotification) *SessionUpda } } return &SessionUpdate{ - Type: "plan", + Type: UpdateTypePlan, SessionID: string(n.SessionId), PlanEntries: entries, } diff --git a/apps/backend/internal/agentctl/adapter/adapter.go b/apps/backend/internal/agentctl/adapter/adapter.go index 7cfc242358..55948b9945 100644 --- a/apps/backend/internal/agentctl/adapter/adapter.go +++ b/apps/backend/internal/agentctl/adapter/adapter.go @@ -18,18 +18,39 @@ type ( PermissionHandler = types.PermissionHandler ) +// SessionUpdate type constants +const ( + UpdateTypeMessageChunk = "message_chunk" // Agent text streaming + UpdateTypeReasoning = "reasoning" // Chain-of-thought/thinking + UpdateTypeToolCall = "tool_call" // Tool invocation started + UpdateTypeToolUpdate = "tool_update" // Tool status update + UpdateTypePlan = "plan" // Agent plan updates + UpdateTypeComplete = "complete" // Turn/session complete + UpdateTypeError = "error" // Error occurred +) + // SessionUpdate is a protocol-agnostic update from the agent. // All protocol adapters normalize their updates to this format. type SessionUpdate struct { - // Type identifies the update type: "message_chunk", "tool_call", "tool_update", "plan", "complete", "error" + // Type identifies the update type. Use UpdateType* constants. Type string `json:"type"` // SessionID is the current session identifier SessionID string `json:"session_id,omitempty"` + // OperationID identifies the current in-flight operation (turn, prompt, etc.) + // Used to target specific operations for cancellation or status updates. + // For Codex this is the turn ID, for other protocols it may be empty. + OperationID string `json:"operation_id,omitempty"` + // Message fields (for "message_chunk" type) Text string `json:"text,omitempty"` + // Reasoning fields (for "reasoning" type) + // Used for chain-of-thought, thinking traces, etc. + ReasoningText string `json:"reasoning_text,omitempty"` // Full reasoning content + ReasoningSummary string `json:"reasoning_summary,omitempty"` // Summarized version (if available) + // Tool call fields (for "tool_call" and "tool_update" types) ToolCallID string `json:"tool_call_id,omitempty"` ToolName string `json:"tool_name,omitempty"` @@ -38,6 +59,10 @@ type SessionUpdate struct { ToolArgs map[string]interface{} `json:"tool_args,omitempty"` ToolResult interface{} `json:"tool_result,omitempty"` + // Diff contains unified diff content for file changes. + // Populated when tools modify files, providing the aggregated diff. + Diff string `json:"diff,omitempty"` + // Plan fields (for "plan" type) PlanEntries []PlanEntry `json:"plan_entries,omitempty"` @@ -94,6 +119,11 @@ type AgentAdapter interface { // GetSessionID returns the current session ID. GetSessionID() string + // GetOperationID returns the current operation/turn ID. + // Returns empty string if no operation is in progress or not supported by the protocol. + // For Codex this is the turn ID, for ACP this may be empty. + GetOperationID() string + // SetPermissionHandler sets the handler for permission requests. SetPermissionHandler(handler PermissionHandler) diff --git a/apps/backend/internal/agentctl/adapter/codex_adapter.go b/apps/backend/internal/agentctl/adapter/codex_adapter.go new file mode 100644 index 0000000000..a5c0acb2a6 --- /dev/null +++ b/apps/backend/internal/agentctl/adapter/codex_adapter.go @@ -0,0 +1,675 @@ +package adapter + +import ( + "context" + "encoding/json" + "fmt" + "io" + "strings" + "sync" + + "github.com/kandev/kandev/internal/agentctl/types" + "github.com/kandev/kandev/internal/common/logger" + "github.com/kandev/kandev/pkg/codex" + "go.uber.org/zap" +) + +// CodexAdapter implements AgentAdapter for agents using the OpenAI Codex app-server protocol. +// Codex uses a JSON-RPC 2.0 variant over stdio (omitting the jsonrpc field). +type CodexAdapter struct { + cfg *Config + logger *logger.Logger + + // Subprocess stdin/stdout (managed externally) + stdin io.Writer + stdout io.Reader + + // Codex client for JSON-RPC communication + client *codex.Client + + // Context for managing goroutine lifecycle + ctx context.Context + cancel context.CancelFunc + + // Session state - Thread maps to Session, Turn maps to Operation + threadID string // session ID + turnID string // operation ID + + // Agent info (populated after Initialize) + agentInfo *AgentInfo + + // Update channel + updatesCh chan SessionUpdate + + // Permission handler + permissionHandler PermissionHandler + + // Accumulators for streaming content + messageBuffer string + reasoningBuffer string + summaryBuffer string + + // Synchronization + mu sync.RWMutex + closed bool +} + +// NewCodexAdapter creates a new Codex protocol adapter. +// stdin and stdout are the subprocess's stdin/stdout pipes (managed by process.Manager). +func NewCodexAdapter(stdin io.Writer, stdout io.Reader, cfg *Config, log *logger.Logger) *CodexAdapter { + ctx, cancel := context.WithCancel(context.Background()) + return &CodexAdapter{ + cfg: cfg, + logger: log.WithFields(zap.String("adapter", "codex")), + stdin: stdin, + stdout: stdout, + ctx: ctx, + cancel: cancel, + updatesCh: make(chan SessionUpdate, 100), + } +} + +// Initialize establishes the Codex connection with the agent subprocess. +func (a *CodexAdapter) Initialize(ctx context.Context) error { + a.logger.Info("initializing Codex adapter", + zap.String("workdir", a.cfg.WorkDir)) + + // Create Codex client + a.client = codex.NewClient(a.stdin, a.stdout, a.logger) + a.client.SetNotificationHandler(a.handleNotification) + a.client.SetRequestHandler(a.handleRequest) + + // Start reading from stdout with the adapter's context + // The readLoop needs to stay alive for the entire lifecycle of the adapter, + // not just the initialize HTTP request. It will be cancelled when Close() is called. + a.client.Start(a.ctx) + + // Perform Codex initialize handshake + resp, err := a.client.Call(ctx, codex.MethodInitialize, &codex.InitializeParams{ + ClientInfo: &codex.ClientInfo{ + Name: "kandev-agentctl", + Title: "Kandev Agent Controller", + Version: "1.0.0", + }, + }) + if err != nil { + return fmt.Errorf("Codex initialize handshake failed: %w", err) + } + if resp.Error != nil { + return fmt.Errorf("Codex initialize error: %s", resp.Error.Message) + } + + // Parse initialize result + var initResult codex.InitializeResult + if resp.Result != nil { + if err := json.Unmarshal(resp.Result, &initResult); err != nil { + a.logger.Warn("failed to parse initialize result", zap.Error(err)) + } + } + + // Send initialized notification + if err := a.client.Notify(codex.MethodInitialized, nil); err != nil { + return fmt.Errorf("failed to send initialized notification: %w", err) + } + + // Store agent info + a.agentInfo = &AgentInfo{ + Name: "codex", + Version: initResult.UserAgent, + } + + a.logger.Info("Codex adapter initialized", + zap.String("user_agent", initResult.UserAgent)) + + return nil +} + +// GetAgentInfo returns information about the connected agent. +func (a *CodexAdapter) GetAgentInfo() *AgentInfo { + return a.agentInfo +} + +// NewSession creates a new Codex thread (session). +func (a *CodexAdapter) NewSession(ctx context.Context) (string, error) { + a.mu.Lock() + defer a.mu.Unlock() + + if a.client == nil { + return "", fmt.Errorf("adapter not initialized") + } + + resp, err := a.client.Call(ctx, codex.MethodThreadStart, &codex.ThreadStartParams{ + Cwd: a.cfg.WorkDir, + ApprovalPolicy: "never", // Valid values: untrusted, on-failure, on-request, never + Sandbox: "danger-full-access", // Disable sandbox to avoid landlock permission issues + }) + if err != nil { + return "", fmt.Errorf("failed to start thread: %w", err) + } + if resp.Error != nil { + return "", fmt.Errorf("thread start error: %s", resp.Error.Message) + } + + var result codex.ThreadStartResult + if err := json.Unmarshal(resp.Result, &result); err != nil { + return "", fmt.Errorf("failed to parse thread start result: %w", err) + } + + a.threadID = result.Thread.ID + a.logger.Info("created new thread", zap.String("thread_id", a.threadID)) + + return a.threadID, nil +} + +// LoadSession resumes an existing Codex thread. +func (a *CodexAdapter) LoadSession(ctx context.Context, sessionID string) error { + a.mu.Lock() + defer a.mu.Unlock() + + if a.client == nil { + return fmt.Errorf("adapter not initialized") + } + + resp, err := a.client.Call(ctx, codex.MethodThreadResume, &codex.ThreadResumeParams{ + ThreadID: sessionID, + }) + if err != nil { + return fmt.Errorf("failed to resume thread: %w", err) + } + if resp.Error != nil { + return fmt.Errorf("thread resume error: %s", resp.Error.Message) + } + + a.threadID = sessionID + a.logger.Info("resumed thread", zap.String("thread_id", a.threadID)) + + return nil +} + +// Prompt sends a prompt to the agent, starting a new turn. +func (a *CodexAdapter) Prompt(ctx context.Context, message string) error { + a.mu.Lock() + threadID := a.threadID + // Reset accumulators for new turn + a.messageBuffer = "" + a.reasoningBuffer = "" + a.summaryBuffer = "" + a.mu.Unlock() + + if a.client == nil { + return fmt.Errorf("adapter not initialized") + } + + a.logger.Info("sending prompt", zap.String("thread_id", threadID)) + + params := &codex.TurnStartParams{ + ThreadID: threadID, + Input: []codex.UserInput{ + {Type: "text", Text: message}, + }, + } + + resp, err := a.client.Call(ctx, codex.MethodTurnStart, params) + if err != nil { + return fmt.Errorf("failed to start turn: %w", err) + } + + if resp.Error != nil { + return fmt.Errorf("turn start error: %s", resp.Error.Message) + } + + var result codex.TurnStartResult + if err := json.Unmarshal(resp.Result, &result); err != nil { + a.logger.Warn("failed to parse turn start result", zap.Error(err), zap.String("raw", string(resp.Result))) + } + + turnID := "" + if result.Turn != nil { + turnID = result.Turn.ID + } + + a.mu.Lock() + a.turnID = turnID + a.mu.Unlock() + + if result.Turn != nil { + a.logger.Info("started turn", zap.String("turn_id", turnID), zap.String("status", result.Turn.Status)) + } else { + a.logger.Info("started turn", zap.String("turn_id", turnID)) + } + + return nil +} + +// Cancel interrupts the current turn. +func (a *CodexAdapter) Cancel(ctx context.Context) error { + a.mu.RLock() + threadID := a.threadID + turnID := a.turnID + a.mu.RUnlock() + + if a.client == nil { + return fmt.Errorf("adapter not initialized") + } + + a.logger.Info("cancelling turn", zap.String("thread_id", threadID), zap.String("turn_id", turnID)) + + // Codex uses turn/interrupt to cancel + _, err := a.client.Call(ctx, codex.MethodTurnInterrupt, map[string]string{ + "threadId": threadID, + "turnId": turnID, + }) + return err +} + +// Updates returns the channel for session updates. +func (a *CodexAdapter) Updates() <-chan SessionUpdate { + return a.updatesCh +} + +// GetSessionID returns the current thread ID (session). +func (a *CodexAdapter) GetSessionID() string { + a.mu.RLock() + defer a.mu.RUnlock() + return a.threadID +} + +// GetOperationID returns the current turn ID (operation). +func (a *CodexAdapter) GetOperationID() string { + a.mu.RLock() + defer a.mu.RUnlock() + return a.turnID +} + +// SetPermissionHandler sets the handler for permission requests. +func (a *CodexAdapter) SetPermissionHandler(handler PermissionHandler) { + a.mu.Lock() + defer a.mu.Unlock() + a.permissionHandler = handler +} + +// Close releases resources held by the adapter. +func (a *CodexAdapter) Close() error { + a.mu.Lock() + defer a.mu.Unlock() + + if a.closed { + return nil + } + a.closed = true + + a.logger.Info("closing Codex adapter") + + // Cancel the context to stop the read loop goroutine + 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 update to the updates channel. +func (a *CodexAdapter) sendUpdate(update SessionUpdate) { + select { + case a.updatesCh <- update: + default: + a.logger.Warn("updates channel full, dropping notification") + } +} + +// handleNotification processes Codex notifications and emits SessionUpdates. +func (a *CodexAdapter) handleNotification(method string, params json.RawMessage) { + a.mu.RLock() + threadID := a.threadID + turnID := a.turnID + a.mu.RUnlock() + + switch method { + // Standard notifications + case codex.NotifyItemAgentMessageDelta: + var p codex.AgentMessageDeltaParams + if err := json.Unmarshal(params, &p); err != nil { + a.logger.Warn("failed to parse agent message delta", zap.Error(err)) + return + } + a.mu.Lock() + a.messageBuffer += p.Delta + a.mu.Unlock() + a.sendUpdate(SessionUpdate{ + Type: UpdateTypeMessageChunk, + SessionID: threadID, + OperationID: turnID, + Text: p.Delta, + }) + + case codex.NotifyItemReasoningTextDelta: + var p codex.ReasoningDeltaParams + if err := json.Unmarshal(params, &p); err != nil { + a.logger.Warn("failed to parse reasoning text delta", zap.Error(err)) + return + } + a.mu.Lock() + a.reasoningBuffer += p.Delta + a.mu.Unlock() + a.sendUpdate(SessionUpdate{ + Type: UpdateTypeReasoning, + SessionID: threadID, + OperationID: turnID, + ReasoningText: p.Delta, + }) + + case codex.NotifyItemReasoningSummaryDelta: + var p codex.ReasoningDeltaParams + if err := json.Unmarshal(params, &p); err != nil { + a.logger.Warn("failed to parse reasoning summary delta", zap.Error(err)) + return + } + a.mu.Lock() + a.summaryBuffer += p.Delta + a.mu.Unlock() + a.sendUpdate(SessionUpdate{ + Type: UpdateTypeReasoning, + SessionID: threadID, + OperationID: turnID, + ReasoningSummary: p.Delta, + }) + + case codex.NotifyTurnCompleted: + var p codex.TurnCompletedParams + if err := json.Unmarshal(params, &p); err != nil { + a.logger.Warn("failed to parse turn completed", zap.Error(err)) + return + } + update := SessionUpdate{ + Type: UpdateTypeComplete, + SessionID: threadID, + OperationID: p.TurnID, + } + if !p.Success && p.Error != "" { + update.Type = UpdateTypeError + update.Error = p.Error + } + a.sendUpdate(update) + + case codex.NotifyTurnDiffUpdated: + var p codex.TurnDiffUpdatedParams + if err := json.Unmarshal(params, &p); err != nil { + a.logger.Warn("failed to parse turn diff updated", zap.Error(err)) + return + } + a.sendUpdate(SessionUpdate{ + Type: UpdateTypeMessageChunk, + SessionID: threadID, + OperationID: p.TurnID, + Diff: p.Diff, + }) + + case codex.NotifyTurnPlanUpdated: + var p codex.TurnPlanUpdatedParams + if err := json.Unmarshal(params, &p); err != nil { + a.logger.Warn("failed to parse turn plan updated", zap.Error(err)) + return + } + entries := make([]PlanEntry, len(p.Plan)) + for i, e := range p.Plan { + entries[i] = PlanEntry{ + Description: e.Description, + Status: e.Status, + } + } + a.sendUpdate(SessionUpdate{ + Type: UpdateTypePlan, + SessionID: threadID, + OperationID: p.TurnID, + PlanEntries: entries, + }) + + case codex.NotifyError: + var p codex.ErrorParams + if err := json.Unmarshal(params, &p); err != nil { + a.logger.Warn("failed to parse error notification", zap.Error(err)) + return + } + a.sendUpdate(SessionUpdate{ + Type: UpdateTypeError, + SessionID: threadID, + Error: p.Message, + }) + + case codex.NotifyItemStarted: + var p codex.ItemStartedParams + if err := json.Unmarshal(params, &p); err != nil { + a.logger.Warn("failed to parse item started", zap.Error(err)) + return + } + if p.Item == nil { + return + } + // Map Codex item types to tool call updates + // Item types: "userMessage", "agentMessage", "commandExecution", "fileChange", "reasoning" + switch p.Item.Type { + case "commandExecution": + a.sendUpdate(SessionUpdate{ + Type: UpdateTypeToolCall, + SessionID: threadID, + OperationID: turnID, + ToolCallID: p.Item.ID, + ToolName: "commandExecution", + ToolTitle: p.Item.Command, + ToolStatus: "running", + ToolArgs: map[string]interface{}{ + "command": p.Item.Command, + "cwd": p.Item.Cwd, + }, + }) + case "fileChange": + // Build title from file paths + var title string + if len(p.Item.Changes) > 0 { + title = p.Item.Changes[0].Path + if len(p.Item.Changes) > 1 { + title += fmt.Sprintf(" (+%d more)", len(p.Item.Changes)-1) + } + } + a.sendUpdate(SessionUpdate{ + Type: UpdateTypeToolCall, + SessionID: threadID, + OperationID: turnID, + ToolCallID: p.Item.ID, + ToolName: "fileChange", + ToolTitle: title, + ToolStatus: "running", + }) + } + + case codex.NotifyItemCompleted: + var p codex.ItemCompletedParams + if err := json.Unmarshal(params, &p); err != nil { + a.logger.Warn("failed to parse item completed", zap.Error(err)) + return + } + if p.Item == nil { + return + } + // Only send updates for tool-like items + if p.Item.Type == "commandExecution" || p.Item.Type == "fileChange" { + status := "complete" + if p.Item.Status == "failed" { + status = "error" + } + update := SessionUpdate{ + Type: UpdateTypeToolUpdate, + SessionID: threadID, + OperationID: turnID, + ToolCallID: p.Item.ID, + ToolStatus: status, + } + // Include output for commands + if p.Item.Type == "commandExecution" && p.Item.AggregatedOutput != "" { + update.ToolResult = p.Item.AggregatedOutput + } + // Include diff for file changes + if p.Item.Type == "fileChange" && len(p.Item.Changes) > 0 { + var diffs []string + for _, c := range p.Item.Changes { + if c.Diff != "" { + diffs = append(diffs, c.Diff) + } + } + if len(diffs) > 0 { + update.Diff = strings.Join(diffs, "\n") + } + } + a.sendUpdate(update) + } + + case codex.NotifyItemCmdExecOutputDelta: + var p codex.CommandOutputDeltaParams + if err := json.Unmarshal(params, &p); err != nil { + a.logger.Warn("failed to parse command output delta", zap.Error(err)) + return + } + a.sendUpdate(SessionUpdate{ + Type: UpdateTypeToolUpdate, + SessionID: threadID, + OperationID: turnID, + ToolCallID: p.ItemID, + ToolResult: p.Delta, + }) + + default: + // Log unhandled notifications at debug level + a.logger.Debug("unhandled notification", zap.String("method", method)) + } +} + +// handleRequest processes Codex requests (approval requests) and calls permissionHandler. +func (a *CodexAdapter) handleRequest(id interface{}, method string, params json.RawMessage) { + a.mu.RLock() + handler := a.permissionHandler + threadID := a.threadID + a.mu.RUnlock() + + switch method { + case codex.NotifyItemCmdExecRequestApproval: + var p codex.CommandApprovalParams + if err := json.Unmarshal(params, &p); err != nil { + a.logger.Warn("failed to parse command approval request", zap.Error(err)) + a.client.SendResponse(id, nil, &codex.Error{Code: codex.InvalidParams, Message: "invalid params"}) + return + } + a.handleApprovalRequest(id, handler, threadID, p.ItemID, types.ActionTypeCommand, p.Command, map[string]interface{}{ + "command": p.Command, + "cwd": p.Cwd, + "reasoning": p.Reasoning, + }, p.Options) + + case codex.NotifyItemFileChangeRequestApproval: + var p codex.FileChangeApprovalParams + if err := json.Unmarshal(params, &p); err != nil { + a.logger.Warn("failed to parse file change approval request", zap.Error(err)) + a.client.SendResponse(id, nil, &codex.Error{Code: codex.InvalidParams, Message: "invalid params"}) + return + } + a.handleApprovalRequest(id, handler, threadID, p.ItemID, types.ActionTypeFileWrite, p.Path, map[string]interface{}{ + "path": p.Path, + "diff": p.Diff, + "reasoning": p.Reasoning, + }, p.Options) + + default: + a.logger.Warn("unhandled request", zap.String("method", method)) + a.client.SendResponse(id, nil, &codex.Error{Code: codex.MethodNotFound, Message: "method not found"}) + } +} + +// handleApprovalRequest handles permission request logic for both command and file change approvals. +func (a *CodexAdapter) handleApprovalRequest( + id interface{}, + handler PermissionHandler, + sessionID string, + itemID string, + actionType string, + title string, + details map[string]interface{}, + optionStrings []string, +) { + // Build permission options from Codex options + options := make([]PermissionOption, len(optionStrings)) + for i, opt := range optionStrings { + kind := "allow_once" + if opt == "approveAlways" { + kind = "allow_always" + } else if opt == "reject" { + kind = "reject_once" + } + options[i] = PermissionOption{ + OptionID: opt, + Name: opt, + Kind: kind, + } + } + + // If no options provided, use defaults + if len(options) == 0 { + options = []PermissionOption{ + {OptionID: "approve", Name: "Approve", Kind: "allow_once"}, + {OptionID: "reject", Name: "Reject", Kind: "reject_once"}, + } + } + + req := &PermissionRequest{ + SessionID: sessionID, + ToolCallID: itemID, + Title: title, + Options: options, + ActionType: actionType, + ActionDetails: details, + } + + if handler == nil { + // Auto-approve if no handler + a.logger.Info("auto-approving request (no handler)", zap.String("item_id", itemID)) + a.client.SendResponse(id, &codex.ApprovalResponse{ + ItemID: itemID, + Decision: "approve", + }, nil) + return + } + + // Call the permission handler + ctx := context.Background() + resp, err := handler(ctx, req) + if err != nil { + a.logger.Error("permission handler error", zap.Error(err)) + a.client.SendResponse(id, &codex.ApprovalResponse{ + ItemID: itemID, + Decision: "reject", + }, nil) + return + } + + decision := "approve" + if resp.Cancelled { + decision = "reject" + } else if resp.OptionID != "" { + decision = resp.OptionID + } + + a.client.SendResponse(id, &codex.ApprovalResponse{ + ItemID: itemID, + Decision: decision, + }, nil) +} + +// Verify interface implementation +var _ AgentAdapter = (*CodexAdapter)(nil) + diff --git a/apps/backend/internal/agentctl/api/acp.go b/apps/backend/internal/agentctl/api/acp.go index dac1e5024e..0e23a510be 100644 --- a/apps/backend/internal/agentctl/api/acp.go +++ b/apps/backend/internal/agentctl/api/acp.go @@ -218,13 +218,14 @@ func (s *Server) handleACPStreamWS(c *gin.Context) { } if err := conn.WriteMessage(websocket.TextMessage, data); err != nil { - s.logger.Debug("WebSocket write error", zap.Error(err)) return } } } } + + // PermissionRespondRequest is a request to respond to a permission request type PermissionRespondRequest struct { PendingID string `json:"pending_id" binding:"required"` diff --git a/apps/backend/internal/agentctl/config/config.go b/apps/backend/internal/agentctl/config/config.go index f1dc01ed4c..fffb9c2f59 100644 --- a/apps/backend/internal/agentctl/config/config.go +++ b/apps/backend/internal/agentctl/config/config.go @@ -46,12 +46,11 @@ type Config struct { // Load loads configuration from environment variables func Load() *Config { workDir := getEnv("AGENTCTL_WORKDIR", "/workspace") - defaultCmd := "auggie --acp --workspace-root " + workDir cfg := &Config{ Port: getEnvInt("AGENTCTL_PORT", 9999), Protocol: agent.Protocol(getEnv("AGENTCTL_PROTOCOL", string(agent.ProtocolACP))), - AgentCommand: getEnv("AGENTCTL_AGENT_COMMAND", defaultCmd), + AgentCommand: getEnv("AGENTCTL_AGENT_COMMAND", "auggie --acp"), WorkDir: workDir, AutoStart: getEnvBool("AGENTCTL_AUTO_START", false), AutoApprovePermissions: getEnvBool("AGENTCTL_AUTO_APPROVE_PERMISSIONS", false), diff --git a/apps/backend/internal/agentctl/instance/instance.go b/apps/backend/internal/agentctl/instance/instance.go index f508f8db45..566adb40b2 100644 --- a/apps/backend/internal/agentctl/instance/instance.go +++ b/apps/backend/internal/agentctl/instance/instance.go @@ -52,6 +52,13 @@ type CreateRequest struct { // AgentCommand is an optional command to start the agent. If empty, a default is used. AgentCommand string `json:"agent_command,omitempty"` + // Protocol is the protocol adapter to use (acp, codex). If empty, default is used. + Protocol string `json:"protocol,omitempty"` + + // WorkspaceFlag is the CLI flag for workspace path (e.g., "--workspace-root"). + // If empty, only cwd is used for workspace path. + WorkspaceFlag string `json:"workspace_flag,omitempty"` + // Env contains optional environment variables to pass to the agent process. Env map[string]string `json:"env,omitempty"` diff --git a/apps/backend/internal/agentctl/instance/manager.go b/apps/backend/internal/agentctl/instance/manager.go index 0ee6c88e0f..f575d4ca4c 100644 --- a/apps/backend/internal/agentctl/instance/manager.go +++ b/apps/backend/internal/agentctl/instance/manager.go @@ -14,6 +14,7 @@ import ( "github.com/kandev/kandev/internal/agentctl/config" "github.com/kandev/kandev/internal/agentctl/process" "github.com/kandev/kandev/internal/common/logger" + "github.com/kandev/kandev/pkg/agent" "go.uber.org/zap" ) @@ -81,15 +82,25 @@ func (m *Manager) CreateInstance(ctx context.Context, req *CreateRequest) (*Crea agentCmd = m.config.DefaultAgentCommand } - // Ensure auggie has the correct workspace-root for standalone mode - // The workspace path is the actual worktree path on the host, not /workspace - if req.WorkspacePath != "" && !strings.Contains(agentCmd, "--workspace-root") { - agentCmd = agentCmd + " --workspace-root " + req.WorkspacePath + // Determine protocol + protocol := agent.Protocol(req.Protocol) + if protocol == "" { + protocol = m.config.DefaultProtocol + } + + // Add workspace path flag if configured + // Note: We also set cmd.Dir to the workspace path, so this is mainly for + // agents that need an explicit flag rather than relying on cwd + if req.WorkspacePath != "" && req.WorkspaceFlag != "" { + if !strings.Contains(agentCmd, req.WorkspaceFlag) { + agentCmd = agentCmd + " " + req.WorkspaceFlag + " " + req.WorkspacePath + } } // Create config for the process manager instanceCfg := &config.Config{ Port: port, + Protocol: protocol, AgentCommand: agentCmd, WorkDir: req.WorkspacePath, AutoStart: req.AutoStart, diff --git a/apps/backend/internal/agentctl/process/manager.go b/apps/backend/internal/agentctl/process/manager.go index 51ee114f24..37f379764b 100644 --- a/apps/backend/internal/agentctl/process/manager.go +++ b/apps/backend/internal/agentctl/process/manager.go @@ -182,6 +182,10 @@ func (m *Manager) Start(ctx context.Context) error { } // Start the process + m.logger.Info("starting agent process", + zap.Strings("args", m.cfg.AgentArgs), + zap.String("workdir", m.cfg.WorkDir), + zap.Int("env_count", len(m.cfg.AgentEnv))) if err := m.cmd.Start(); err != nil { m.status.Store(StatusError) return fmt.Errorf("failed to start agent: %w", err) @@ -229,6 +233,8 @@ func (m *Manager) createAdapter() error { switch protocol { case agent.ProtocolACP: m.adapter = adapter.NewACPAdapter(m.stdin, m.stdout, adapterCfg, m.logger) + case agent.ProtocolCodex: + m.adapter = adapter.NewCodexAdapter(m.stdin, m.stdout, adapterCfg, m.logger) default: return fmt.Errorf("unsupported protocol: %s", protocol) } diff --git a/apps/backend/internal/agentctl/types/types.go b/apps/backend/internal/agentctl/types/types.go index 8d63e7b3ff..52dc51fd04 100644 --- a/apps/backend/internal/agentctl/types/types.go +++ b/apps/backend/internal/agentctl/types/types.go @@ -4,12 +4,32 @@ package types import "context" +// Permission action types - categorize the kind of action requiring approval +const ( + ActionTypeCommand = "command" // Shell command execution + ActionTypeFileWrite = "file_write" // File modification/creation + ActionTypeFileRead = "file_read" // File read (for sensitive files) + ActionTypeNetwork = "network" // Network access + ActionTypeMCPTool = "mcp_tool" // MCP tool invocation + ActionTypeOther = "other" // Other/unknown action type +) + // PermissionRequest represents a permission request from the agent type PermissionRequest struct { SessionID string `json:"session_id"` ToolCallID string `json:"tool_call_id"` Title string `json:"title"` Options []PermissionOption `json:"options"` + + // ActionType categorizes the action requiring approval. + // Use ActionType* constants: "command", "file_write", "network", "mcp_tool", etc. + ActionType string `json:"action_type,omitempty"` + + // ActionDetails contains structured details about the action. + // For commands: {"command": ["ls", "-la"], "cwd": "/path"} + // For files: {"path": "/file.go", "diff": "..."} + // For MCP tools: {"server": "...", "tool": "...", "arguments": {...}} + ActionDetails map[string]interface{} `json:"action_details,omitempty"` } // PermissionOption represents a permission choice @@ -17,12 +37,20 @@ type PermissionOption struct { OptionID string `json:"option_id"` Name string `json:"name"` Kind string `json:"kind"` // allow_once, allow_always, reject_once, reject_always + + // Metadata contains protocol-specific option data. + // For Codex: {"for_session": true} for session-wide approvals + Metadata map[string]interface{} `json:"metadata,omitempty"` } // PermissionResponse is the user's response to a permission request type PermissionResponse struct { OptionID string `json:"option_id,omitempty"` Cancelled bool `json:"cancelled,omitempty"` + + // ResponseMetadata contains protocol-specific response data. + // For Codex: {"accept_settings": {"for_session": true}} + ResponseMetadata map[string]interface{} `json:"response_metadata,omitempty"` } // PermissionHandler is called when the agent requests permission for an action diff --git a/apps/backend/internal/common/stringutil/stringutil.go b/apps/backend/internal/common/stringutil/stringutil.go new file mode 100644 index 0000000000..2b3e5fd6d9 --- /dev/null +++ b/apps/backend/internal/common/stringutil/stringutil.go @@ -0,0 +1,26 @@ +// Package stringutil provides common string utility functions. +package stringutil + +// TruncateString truncates a string to a maximum length. +// If the string is shorter than maxLen, it returns the original string. +// If the string is longer, it returns the first maxLen characters. +func TruncateString(s string, maxLen int) string { + if len(s) <= maxLen { + return s + } + return s[:maxLen] +} + +// TruncateStringWithEllipsis truncates a string to a maximum length and adds "..." suffix. +// If the string is shorter than maxLen, it returns the original string. +// If the string is longer, it returns the first (maxLen-3) characters followed by "...". +func TruncateStringWithEllipsis(s string, maxLen int) string { + if maxLen < 4 { + return TruncateString(s, maxLen) + } + if len(s) <= maxLen { + return s + } + return s[:maxLen-3] + "..." +} + diff --git a/apps/backend/internal/events/types.go b/apps/backend/internal/events/types.go index a485ffb1c1..5c630d0764 100644 --- a/apps/backend/internal/events/types.go +++ b/apps/backend/internal/events/types.go @@ -32,7 +32,8 @@ const ( // Event types for comments const ( - CommentAdded = "comment.added" + CommentAdded = "comment.added" + CommentUpdated = "comment.updated" ) // Event types for repositories diff --git a/apps/backend/internal/gateway/websocket/client.go b/apps/backend/internal/gateway/websocket/client.go index 4e09975fc5..284dad2cb2 100644 --- a/apps/backend/internal/gateway/websocket/client.go +++ b/apps/backend/internal/gateway/websocket/client.go @@ -70,7 +70,10 @@ func (c *Client) ReadPump(ctx context.Context) { for { _, message, err := c.conn.ReadMessage() if err != nil { - if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) { + // CloseGoingAway (1001): Client navigating away + // CloseNoStatusReceived (1005): Client closed without status (normal browser close) + // CloseAbnormalClosure (1006): Abnormal close (network drop) + if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseNoStatusReceived, websocket.CloseAbnormalClosure) { c.logger.Error("WebSocket read error", zap.Error(err)) } break diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index eb7b6e44fa..217a837adb 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -51,6 +51,7 @@ type InputRequestHandler func(ctx context.Context, taskID, agentID, message stri type CommentCreator interface { CreateAgentComment(ctx context.Context, taskID, content, agentSessionID string) error CreateToolCallComment(ctx context.Context, taskID, toolCallID, title, status, agentSessionID string, args map[string]interface{}) error + UpdateToolCallComment(ctx context.Context, taskID, toolCallID, status, result string) error } // Service is the main orchestrator service @@ -145,6 +146,7 @@ func NewService( OnACPMessage: s.handleACPMessage, OnPromptComplete: s.handlePromptComplete, OnToolCallStarted: s.handleToolCallStarted, + OnToolCallComplete: s.handleToolCallComplete, OnGitStatusUpdated: s.handleGitStatusUpdated, } s.watcher = watcher.NewWatcher(eventBus, handlers, log) @@ -672,3 +674,16 @@ func (s *Service) handleToolCallStarted(ctx context.Context, data watcher.ToolCa } } } + +// handleToolCallComplete handles tool call complete events and updates the comment +func (s *Service) handleToolCallComplete(ctx context.Context, data watcher.ToolCallCompleteData) { + // Update tool call comment status if we have a comment creator + if s.commentCreator != nil { + if err := s.commentCreator.UpdateToolCallComment(ctx, data.TaskID, data.ToolCallID, data.Status, data.Result); err != nil { + s.logger.Error("failed to update tool call comment", + zap.String("task_id", data.TaskID), + zap.String("tool_call_id", data.ToolCallID), + zap.Error(err)) + } + } +} diff --git a/apps/backend/internal/orchestrator/watcher/watcher.go b/apps/backend/internal/orchestrator/watcher/watcher.go index 80dd01fa92..6a98174831 100644 --- a/apps/backend/internal/orchestrator/watcher/watcher.go +++ b/apps/backend/internal/orchestrator/watcher/watcher.go @@ -50,6 +50,15 @@ type ToolCallData struct { Args map[string]interface{} `json:"args,omitempty"` // Tool call arguments (kind, path, locations, raw_input) } +// ToolCallCompleteData contains data from tool_call_complete events +type ToolCallCompleteData struct { + TaskID string `json:"task_id"` + AgentID string `json:"agent_id"` + ToolCallID string `json:"tool_call_id"` + Status string `json:"status"` // "complete" or "error" + Result string `json:"result,omitempty"` +} + // GitStatusData contains data from git status events type GitStatusData struct { TaskID string `json:"task_id"` @@ -89,7 +98,8 @@ type EventHandlers struct { OnPromptComplete func(ctx context.Context, data PromptCompleteData) // Tool call events - OnToolCallStarted func(ctx context.Context, data ToolCallData) + OnToolCallStarted func(ctx context.Context, data ToolCallData) + OnToolCallComplete func(ctx context.Context, data ToolCallCompleteData) // Git status events OnGitStatusUpdated func(ctx context.Context, data GitStatusData) @@ -407,27 +417,37 @@ func (w *Watcher) createPromptCompleteHandler() bus.EventHandler { // subscribeToToolCallEvents subscribes to tool call events func (w *Watcher) subscribeToToolCallEvents() error { - if w.handlers.OnToolCallStarted == nil { - return nil + // Subscribe to tool call started events + if w.handlers.OnToolCallStarted != nil { + subject := events.ToolCallStarted + ".*" + sub, err := w.eventBus.Subscribe(subject, w.createToolCallStartedHandler()) + if err != nil { + w.logger.Error("Failed to subscribe to tool call started events", + zap.String("subject", subject), + zap.Error(err)) + return err + } + w.subscriptions = append(w.subscriptions, sub) } - // Use wildcard to subscribe to all tool call events (tool_call.started.*) - subject := events.ToolCallStarted + ".*" - - // Use regular subscription (each instance needs all messages for comment creation) - sub, err := w.eventBus.Subscribe(subject, w.createToolCallHandler()) - if err != nil { - w.logger.Error("Failed to subscribe to tool call events", - zap.String("subject", subject), - zap.Error(err)) - return err + // Subscribe to tool call complete events + if w.handlers.OnToolCallComplete != nil { + subject := events.ToolCallComplete + ".*" + sub, err := w.eventBus.Subscribe(subject, w.createToolCallCompleteHandler()) + if err != nil { + w.logger.Error("Failed to subscribe to tool call complete events", + zap.String("subject", subject), + zap.Error(err)) + return err + } + w.subscriptions = append(w.subscriptions, sub) } - w.subscriptions = append(w.subscriptions, sub) + return nil } -// createToolCallHandler creates a bus.EventHandler for tool call events -func (w *Watcher) createToolCallHandler() bus.EventHandler { +// createToolCallStartedHandler creates a bus.EventHandler for tool call started events +func (w *Watcher) createToolCallStartedHandler() bus.EventHandler { return func(ctx context.Context, event *bus.Event) error { var data ToolCallData if err := w.parseEventData(event.Data, &data); err != nil { @@ -438,7 +458,7 @@ func (w *Watcher) createToolCallHandler() bus.EventHandler { return nil // Don't return error to continue processing other events } - w.logger.Debug("Handling tool call event", + w.logger.Debug("Handling tool call started event", zap.String("task_id", data.TaskID), zap.String("tool_call_id", data.ToolCallID), zap.String("title", data.Title)) @@ -448,6 +468,28 @@ func (w *Watcher) createToolCallHandler() bus.EventHandler { } } +// createToolCallCompleteHandler creates a bus.EventHandler for tool call complete events +func (w *Watcher) createToolCallCompleteHandler() bus.EventHandler { + return func(ctx context.Context, event *bus.Event) error { + var data ToolCallCompleteData + if err := w.parseEventData(event.Data, &data); err != nil { + w.logger.Error("Failed to parse tool call complete event data", + zap.String("event_type", event.Type), + zap.String("event_id", event.ID), + zap.Error(err)) + return nil + } + + w.logger.Debug("Handling tool call complete event", + zap.String("task_id", data.TaskID), + zap.String("tool_call_id", data.ToolCallID), + zap.String("status", data.Status)) + + w.handlers.OnToolCallComplete(ctx, data) + return nil + } +} + // subscribeToGitStatusEvents subscribes to git status events func (w *Watcher) subscribeToGitStatusEvents() error { if w.handlers.OnGitStatusUpdated == nil { diff --git a/apps/backend/internal/task/repository/interface.go b/apps/backend/internal/task/repository/interface.go index 569a1d417a..aa05e35c81 100644 --- a/apps/backend/internal/task/repository/interface.go +++ b/apps/backend/internal/task/repository/interface.go @@ -45,6 +45,8 @@ type Repository interface { // Comment operations CreateComment(ctx context.Context, comment *models.Comment) error GetComment(ctx context.Context, id string) (*models.Comment, error) + GetCommentByToolCallID(ctx context.Context, taskID, toolCallID string) (*models.Comment, error) + UpdateComment(ctx context.Context, comment *models.Comment) error ListComments(ctx context.Context, taskID string) ([]*models.Comment, error) DeleteComment(ctx context.Context, id string) error diff --git a/apps/backend/internal/task/repository/memory.go b/apps/backend/internal/task/repository/memory.go index 72eebbcdde..e9593681a0 100644 --- a/apps/backend/internal/task/repository/memory.go +++ b/apps/backend/internal/task/repository/memory.go @@ -700,6 +700,33 @@ func (r *MemoryRepository) DeleteComment(ctx context.Context, id string) error { return nil } +// GetCommentByToolCallID retrieves a comment by task ID and tool_call_id in metadata +func (r *MemoryRepository) GetCommentByToolCallID(ctx context.Context, taskID, toolCallID string) (*models.Comment, error) { + r.mu.RLock() + defer r.mu.RUnlock() + + for _, comment := range r.comments { + if comment.TaskID == taskID && comment.Metadata != nil { + if tcID, ok := comment.Metadata["tool_call_id"].(string); ok && tcID == toolCallID { + return comment, nil + } + } + } + return nil, fmt.Errorf("comment not found for tool_call_id: %s", toolCallID) +} + +// UpdateComment updates an existing comment +func (r *MemoryRepository) UpdateComment(ctx context.Context, comment *models.Comment) error { + r.mu.Lock() + defer r.mu.Unlock() + + if _, ok := r.comments[comment.ID]; !ok { + return fmt.Errorf("comment not found: %s", comment.ID) + } + r.comments[comment.ID] = comment + return nil +} + // AgentSession operations // CreateAgentSession creates a new agent session diff --git a/apps/backend/internal/task/repository/sqlite.go b/apps/backend/internal/task/repository/sqlite.go index d8e0856bb5..789781c667 100644 --- a/apps/backend/internal/task/repository/sqlite.go +++ b/apps/backend/internal/task/repository/sqlite.go @@ -1361,6 +1361,55 @@ func (r *SQLiteRepository) DeleteComment(ctx context.Context, id string) error { return nil } +// GetCommentByToolCallID retrieves a comment by task ID and tool_call_id in metadata +func (r *SQLiteRepository) GetCommentByToolCallID(ctx context.Context, taskID, toolCallID string) (*models.Comment, error) { + comment := &models.Comment{} + var requestsInput int + var commentType string + var metadataJSON string + err := r.db.QueryRowContext(ctx, ` + SELECT id, task_id, author_type, author_id, content, requests_input, acp_session_id, type, metadata, created_at + FROM task_comments WHERE task_id = ? AND json_extract(metadata, '$.tool_call_id') = ? + `, taskID, toolCallID).Scan(&comment.ID, &comment.TaskID, &comment.AuthorType, &comment.AuthorID, + &comment.Content, &requestsInput, &comment.ACPSessionID, &commentType, &metadataJSON, &comment.CreatedAt) + if err != nil { + return nil, err + } + comment.RequestsInput = requestsInput == 1 + comment.Type = models.CommentType(commentType) + if metadataJSON != "" { + _ = json.Unmarshal([]byte(metadataJSON), &comment.Metadata) + } + return comment, nil +} + +// UpdateComment updates an existing comment +func (r *SQLiteRepository) UpdateComment(ctx context.Context, comment *models.Comment) error { + metadataJSON, err := json.Marshal(comment.Metadata) + if err != nil { + return fmt.Errorf("failed to marshal metadata: %w", err) + } + + requestsInput := 0 + if comment.RequestsInput { + requestsInput = 1 + } + + result, err := r.db.ExecContext(ctx, ` + UPDATE task_comments SET content = ?, requests_input = ?, type = ?, metadata = ? + WHERE id = ? + `, comment.Content, requestsInput, string(comment.Type), string(metadataJSON), comment.ID) + if err != nil { + return err + } + + rows, _ := result.RowsAffected() + if rows == 0 { + return fmt.Errorf("comment not found: %s", comment.ID) + } + return nil +} + // Repository operations // CreateRepository creates a new repository diff --git a/apps/backend/internal/task/service/service.go b/apps/backend/internal/task/service/service.go index 8114550e6d..2a395560f1 100644 --- a/apps/backend/internal/task/service/service.go +++ b/apps/backend/internal/task/service/service.go @@ -1300,9 +1300,51 @@ func (s *Service) DeleteComment(ctx context.Context, id string) error { return nil } +// UpdateToolCallComment updates a tool call comment's status +func (s *Service) UpdateToolCallComment(ctx context.Context, taskID, toolCallID, status, result string) error { + // Find the comment by tool_call_id + comment, err := s.repo.GetCommentByToolCallID(ctx, taskID, toolCallID) + if err != nil { + s.logger.Warn("tool call comment not found for update", + zap.String("task_id", taskID), + zap.String("tool_call_id", toolCallID), + zap.Error(err)) + return err + } + + // Update the metadata status + if comment.Metadata == nil { + comment.Metadata = make(map[string]interface{}) + } + comment.Metadata["status"] = status + if result != "" { + comment.Metadata["result"] = result + } + + // Save the updated comment + if err := s.repo.UpdateComment(ctx, comment); err != nil { + s.logger.Error("failed to update tool call comment", + zap.String("comment_id", comment.ID), + zap.String("tool_call_id", toolCallID), + zap.Error(err)) + return err + } + + // Publish comment.updated event + s.publishCommentEvent(ctx, events.CommentUpdated, comment) + + s.logger.Info("tool call comment updated", + zap.String("comment_id", comment.ID), + zap.String("tool_call_id", toolCallID), + zap.String("status", status)) + + return nil +} + // publishCommentEvent publishes comment events to the event bus func (s *Service) publishCommentEvent(ctx context.Context, eventType string, comment *models.Comment) { if s.eventBus == nil { + s.logger.Warn("publishCommentEvent: eventBus is nil, skipping") return } diff --git a/apps/backend/pkg/agent/protocol.go b/apps/backend/pkg/agent/protocol.go index 7c9bea955d..9cd6bacf63 100644 --- a/apps/backend/pkg/agent/protocol.go +++ b/apps/backend/pkg/agent/protocol.go @@ -11,6 +11,10 @@ const ( ProtocolREST Protocol = "rest" // ProtocolMCP is the Model Context Protocol. ProtocolMCP Protocol = "mcp" + // ProtocolCodex is the OpenAI Codex app-server protocol (JSON-RPC variant over stdin/stdout). + // Codex uses a similar structure to ACP but with different message formats and + // a Thread/Turn model instead of Session/Prompt. + ProtocolCodex Protocol = "codex" ) // String returns the string representation of the protocol. @@ -21,7 +25,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: + case ProtocolACP, ProtocolREST, ProtocolMCP, ProtocolCodex: return true default: return false diff --git a/apps/backend/pkg/codex/client.go b/apps/backend/pkg/codex/client.go new file mode 100644 index 0000000000..cafe761866 --- /dev/null +++ b/apps/backend/pkg/codex/client.go @@ -0,0 +1,249 @@ +package codex + +import ( + "bufio" + "context" + "encoding/json" + "fmt" + "io" + "sync" + "sync/atomic" + + "github.com/kandev/kandev/internal/common/logger" + "go.uber.org/zap" +) + +// Client handles Codex JSON-RPC communication over stdin/stdout streams. +// Unlike standard JSON-RPC 2.0, Codex omits the "jsonrpc":"2.0" field. +type Client struct { + stdin io.Writer + stdout io.Reader + + requestID atomic.Int64 + pending map[interface{}]chan *Response + mu sync.Mutex + + onNotification func(method string, params json.RawMessage) + onRequest func(id interface{}, method string, params json.RawMessage) + + logger *logger.Logger + done chan struct{} +} + +// NewClient creates a new Codex JSON-RPC client +func NewClient(stdin io.Writer, stdout io.Reader, log *logger.Logger) *Client { + return &Client{ + stdin: stdin, + stdout: stdout, + pending: make(map[interface{}]chan *Response), + logger: log.WithFields(zap.String("component", "codex-client")), + done: make(chan struct{}), + } +} + +// SetNotificationHandler sets the handler for incoming notifications +func (c *Client) SetNotificationHandler(handler func(method string, params json.RawMessage)) { + c.onNotification = handler +} + +// SetRequestHandler sets the handler for incoming requests from the agent +func (c *Client) SetRequestHandler(handler func(id interface{}, method string, params json.RawMessage)) { + c.onRequest = handler +} + +// SendResponse sends a response to an agent request +func (c *Client) SendResponse(id interface{}, result interface{}, err *Error) error { + var resultJSON json.RawMessage + if result != nil && err == nil { + var marshalErr error + resultJSON, marshalErr = json.Marshal(result) + if marshalErr != nil { + return fmt.Errorf("failed to marshal result: %w", marshalErr) + } + } + resp := &Response{ID: id, Result: resultJSON, Error: err} + return c.send(resp) +} + +// Start begins reading responses from stdout +func (c *Client) Start(ctx context.Context) { + go c.readLoop(ctx) +} + +// Stop stops the client +func (c *Client) Stop() { + close(c.done) +} + +// Call sends a request and waits for a response +func (c *Client) Call(ctx context.Context, method string, params interface{}) (*Response, error) { + id := c.requestID.Add(1) + + var paramsJSON json.RawMessage + if params != nil { + var err error + paramsJSON, err = json.Marshal(params) + if err != nil { + return nil, fmt.Errorf("failed to marshal params: %w", err) + } + } + + req := &Request{ID: id, Method: method, Params: paramsJSON} + + respCh := make(chan *Response, 1) + c.mu.Lock() + c.pending[id] = respCh + c.mu.Unlock() + + defer func() { + c.mu.Lock() + delete(c.pending, id) + c.mu.Unlock() + }() + + if err := c.send(req); err != nil { + return nil, err + } + + select { + case resp := <-respCh: + return resp, nil + case <-ctx.Done(): + return nil, ctx.Err() + case <-c.done: + return nil, fmt.Errorf("client closed") + } +} + +// Notify sends a notification (no response expected) +func (c *Client) Notify(method string, params interface{}) error { + var paramsJSON json.RawMessage + if params != nil { + var err error + paramsJSON, err = json.Marshal(params) + if err != nil { + return fmt.Errorf("failed to marshal params: %w", err) + } + } + notif := &Notification{Method: method, Params: paramsJSON} + return c.send(notif) +} + +func (c *Client) send(msg interface{}) error { + data, err := json.Marshal(msg) + if err != nil { + return fmt.Errorf("failed to marshal message: %w", err) + } + data = append(data, '\n') + _, err = c.stdin.Write(data) + if err != nil { + return fmt.Errorf("failed to write message: %w", err) + } + c.logger.Debug("codex: sent message", zap.String("data", string(data))) + return nil +} + +func (c *Client) readLoop(ctx context.Context) { + scanner := bufio.NewScanner(c.stdout) + buf := make([]byte, 0, 64*1024) + scanner.Buffer(buf, 1024*1024) + + for scanner.Scan() { + select { + case <-ctx.Done(): + return + case <-c.done: + return + default: + } + + line := scanner.Bytes() + if len(line) == 0 { + continue + } + + var msg struct { + ID interface{} `json:"id"` + Method string `json:"method"` + Result json.RawMessage `json:"result"` + Error *Error `json:"error"` + Params json.RawMessage `json:"params"` + } + if err := json.Unmarshal(line, &msg); err != nil { + c.logger.Warn("failed to parse message", zap.Error(err)) + continue + } + + hasID := msg.ID != nil + hasMethod := msg.Method != "" + hasResult := msg.Result != nil + hasError := msg.Error != nil + + if hasID && !hasMethod && (hasResult || hasError) { + c.handleResponse(&Response{ID: msg.ID, Result: msg.Result, Error: msg.Error}) + } else if hasID && hasMethod { + c.handleRequest(msg.ID, msg.Method, msg.Params) + } else if hasMethod && !hasID { + c.handleNotification(msg.Method, msg.Params) + } + } + + if err := scanner.Err(); err != nil { + c.logger.Error("read loop error", zap.Error(err)) + } +} + +func (c *Client) handleResponse(resp *Response) { + id := normalizeID(resp.ID) + c.mu.Lock() + ch, ok := c.pending[id] + c.mu.Unlock() + if ok { + ch <- resp + } else { + c.logger.Warn("received response for unknown request", zap.Any("id", resp.ID)) + } +} + +func normalizeID(id interface{}) interface{} { + switch v := id.(type) { + case float64: + return int64(v) + case json.Number: + if i, err := v.Int64(); err == nil { + return i + } + } + return id +} + +func (c *Client) handleNotification(method string, params json.RawMessage) { + if c.onNotification != nil { + c.onNotification(method, params) + } +} + +func (c *Client) handleRequest(id interface{}, method string, params json.RawMessage) { + if c.onRequest != nil { + c.onRequest(id, method, params) + } else { + c.logger.Warn("received request but no handler registered", zap.String("method", method)) + c.SendResponse(id, nil, &Error{Code: MethodNotFound, Message: "Method not found"}) + } +} + +// extractMethod extracts the method field from a JSON message for logging +func extractMethod(data []byte) string { + var msg struct { + Method string `json:"method"` + ID int `json:"id"` + } + if err := json.Unmarshal(data, &msg); err != nil { + return "parse-error" + } + if msg.Method != "" { + return msg.Method + } + return "response" +} + diff --git a/apps/backend/pkg/codex/types.go b/apps/backend/pkg/codex/types.go new file mode 100644 index 0000000000..affd846b51 --- /dev/null +++ b/apps/backend/pkg/codex/types.go @@ -0,0 +1,301 @@ +// Package codex provides types and client for the OpenAI Codex app-server protocol. +// Codex uses a JSON-RPC 2.0 variant over stdio, but omits the "jsonrpc":"2.0" header. +package codex + +import "encoding/json" + +// Request represents a Codex JSON-RPC request (without jsonrpc field) +type Request struct { + ID interface{} `json:"id,omitempty"` + Method string `json:"method"` + Params json.RawMessage `json:"params,omitempty"` +} + +// Response represents a Codex JSON-RPC response +type Response struct { + ID interface{} `json:"id"` + Result json.RawMessage `json:"result,omitempty"` + Error *Error `json:"error,omitempty"` +} + +// Notification represents a Codex notification (no id field) +type Notification struct { + Method string `json:"method"` + Params json.RawMessage `json:"params,omitempty"` +} + +// Error represents a JSON-RPC error +type Error struct { + Code int `json:"code"` + Message string `json:"message"` + Data any `json:"data,omitempty"` +} + +// Standard error codes +const ( + ParseError = -32700 + InvalidRequest = -32600 + MethodNotFound = -32601 + InvalidParams = -32602 + InternalError = -32603 +) + +// Codex method names +const ( + MethodInitialize = "initialize" + MethodInitialized = "initialized" // Notification + MethodThreadStart = "thread/start" + MethodThreadResume = "thread/resume" + MethodThreadFork = "thread/fork" + MethodThreadList = "thread/list" + MethodThreadArchive = "thread/archive" + MethodTurnStart = "turn/start" + MethodTurnInterrupt = "turn/interrupt" + MethodCommandExec = "command/exec" + MethodModelList = "model/list" + MethodSkillsList = "skills/list" + MethodAccountRead = "account/read" + MethodAccountLogout = "account/logout" + MethodReviewStart = "review/start" + MethodConfigRead = "config/read" +) + +// Codex notification methods (server → client) +const ( + NotifyThreadStarted = "thread/started" + NotifyTurnStarted = "turn/started" + NotifyTurnCompleted = "turn/completed" + NotifyTurnDiffUpdated = "turn/diff/updated" + NotifyTurnPlanUpdated = "turn/plan/updated" + NotifyItemStarted = "item/started" + NotifyItemCompleted = "item/completed" + NotifyItemAgentMessageDelta = "item/agentMessage/delta" + NotifyItemReasoningSummaryDelta = "item/reasoning/summaryTextDelta" + NotifyItemReasoningTextDelta = "item/reasoning/textDelta" + NotifyItemCmdExecOutputDelta = "item/commandExecution/outputDelta" + NotifyItemCmdExecRequestApproval = "item/commandExecution/requestApproval" + NotifyItemFileChangeRequestApproval = "item/fileChange/requestApproval" + NotifyAccountUpdated = "account/updated" + NotifyAccountLoginCompleted = "account/login/completed" + NotifyError = "error" +) + +// InitializeParams for initialize request +type InitializeParams struct { + ClientInfo *ClientInfo `json:"clientInfo"` +} + +// ClientInfo identifies the client +type ClientInfo struct { + Name string `json:"name"` + Title string `json:"title,omitempty"` + Version string `json:"version"` +} + +// InitializeResult from initialize +type InitializeResult struct { + UserAgent string `json:"userAgent,omitempty"` +} + +// ThreadStartParams for thread/start +type ThreadStartParams struct { + Model string `json:"model,omitempty"` + Cwd string `json:"cwd,omitempty"` + ApprovalPolicy string `json:"approvalPolicy,omitempty"` // "untrusted", "on-failure", "on-request", "never" + Sandbox string `json:"sandbox,omitempty"` // "workspaceWrite", "readOnly", etc. + SandboxPolicy *SandboxPolicy `json:"sandboxPolicy,omitempty"` +} + +// SandboxPolicy configures sandbox behavior +type SandboxPolicy struct { + Type string `json:"type"` // "workspaceWrite", "readOnly", "dangerFullAccess", "externalSandbox" + WritableRoots []string `json:"writableRoots,omitempty"` + NetworkAccess bool `json:"networkAccess,omitempty"` +} + +// Thread represents a Codex thread (conversation) +type Thread struct { + ID string `json:"id"` + Preview string `json:"preview,omitempty"` + ModelProvider string `json:"modelProvider,omitempty"` + CreatedAt int64 `json:"createdAt,omitempty"` +} + +// ThreadStartResult from thread/start +type ThreadStartResult struct { + Thread *Thread `json:"thread"` +} + +// ThreadResumeParams for thread/resume +type ThreadResumeParams struct { + ThreadID string `json:"threadId"` +} + +// ThreadResumeResult from thread/resume +type ThreadResumeResult struct { + Thread *Thread `json:"thread"` +} + +// UserInput represents input to a turn +type UserInput struct { + Type string `json:"type"` // "text", "image", "localImage", "skill" + Text string `json:"text,omitempty"` + URL string `json:"url,omitempty"` + Path string `json:"path,omitempty"` + Name string `json:"name,omitempty"` +} + +// TurnStartParams for turn/start +type TurnStartParams struct { + ThreadID string `json:"threadId"` + Input []UserInput `json:"input"` +} + +// Turn represents a Codex turn within a thread +type Turn struct { + ID string `json:"id"` + Status string `json:"status"` // "inProgress", "completed", "failed" + Items []Item `json:"items"` + Error *Error `json:"error,omitempty"` +} + +// TurnStartResult from turn/start +type TurnStartResult struct { + Turn *Turn `json:"turn"` +} + +// Item represents a Codex item (message, command, file change, etc.) +type Item struct { + ID string `json:"id"` + Type string `json:"type"` // "userMessage", "agentMessage", "commandExecution", "fileChange", "reasoning" + Status string `json:"status"` // "inProgress", "completed", "failed" + + // For commandExecution type + Command string `json:"command,omitempty"` + Cwd string `json:"cwd,omitempty"` + AggregatedOutput string `json:"aggregatedOutput,omitempty"` + ExitCode *int `json:"exitCode,omitempty"` + DurationMs *int `json:"durationMs,omitempty"` + + // For fileChange type + Changes []FileChange `json:"changes,omitempty"` + + // For reasoning type + Summary []string `json:"summary,omitempty"` + Content []string `json:"content,omitempty"` +} + +// FileChange represents a file change in a fileChange item +type FileChange struct { + Path string `json:"path"` + Kind FileChangeKind `json:"kind"` + Diff string `json:"diff,omitempty"` +} + +// FileChangeKind represents the type of file change +type FileChangeKind struct { + Type string `json:"type"` // "add", "modify", "delete" +} + +// ItemStartedParams for item/started notification +type ItemStartedParams struct { + ThreadID string `json:"threadId"` + TurnID string `json:"turnId"` + Item *Item `json:"item"` +} + +// ItemCompletedParams for item/completed notification +type ItemCompletedParams struct { + ThreadID string `json:"threadId"` + TurnID string `json:"turnId"` + Item *Item `json:"item"` +} + +// AgentMessageDeltaParams for item/agentMessage/delta +type AgentMessageDeltaParams struct { + ThreadID string `json:"threadId"` + TurnID string `json:"turnId"` + ItemID string `json:"itemId"` + Delta string `json:"delta"` +} + +// ReasoningDeltaParams for item/reasoning/textDelta and summaryTextDelta +type ReasoningDeltaParams struct { + ThreadID string `json:"threadId"` + TurnID string `json:"turnId"` + ItemID string `json:"itemId"` + Delta string `json:"delta"` +} + +// CommandOutputDeltaParams for item/commandExecution/outputDelta +type CommandOutputDeltaParams struct { + ThreadID string `json:"threadId"` + TurnID string `json:"turnId"` + ItemID string `json:"itemId"` + Delta string `json:"delta"` +} + +// CommandApprovalParams for item/commandExecution/requestApproval +type CommandApprovalParams struct { + ThreadID string `json:"threadId"` + TurnID string `json:"turnId"` + ItemID string `json:"itemId"` + Command string `json:"command"` + Cwd string `json:"cwd,omitempty"` + Reasoning string `json:"reasoning,omitempty"` + Options []string `json:"options,omitempty"` // e.g., ["approve", "reject", "approveAlways"] +} + +// FileChangeApprovalParams for item/fileChange/requestApproval +type FileChangeApprovalParams struct { + ThreadID string `json:"threadId"` + TurnID string `json:"turnId"` + ItemID string `json:"itemId"` + Path string `json:"path"` + Diff string `json:"diff,omitempty"` + Reasoning string `json:"reasoning,omitempty"` + Options []string `json:"options,omitempty"` +} + +// TurnCompletedParams for turn/completed notification +type TurnCompletedParams struct { + ThreadID string `json:"threadId"` + TurnID string `json:"turnId"` + Success bool `json:"success"` + Error string `json:"error,omitempty"` +} + +// TurnDiffUpdatedParams for turn/diff/updated notification +type TurnDiffUpdatedParams struct { + ThreadID string `json:"threadId"` + TurnID string `json:"turnId"` + Diff string `json:"diff"` +} + +// TurnPlanUpdatedParams for turn/plan/updated notification +type TurnPlanUpdatedParams struct { + ThreadID string `json:"threadId"` + TurnID string `json:"turnId"` + Plan []PlanEntry `json:"plan"` +} + +// PlanEntry represents a single plan item +type PlanEntry struct { + Description string `json:"description"` + Status string `json:"status"` // "pending", "in_progress", "completed", "failed" +} + +// ApprovalResponse for responding to approval requests +type ApprovalResponse struct { + ThreadID string `json:"threadId"` + TurnID string `json:"turnId"` + ItemID string `json:"itemId"` + Decision string `json:"decision"` // "approve", "reject", "approveAlways" +} + +// ErrorParams for error notification +type ErrorParams struct { + Code int `json:"code,omitempty"` + Message string `json:"message"` + Data any `json:"data,omitempty"` +} diff --git a/apps/backend/pkg/websocket/actions.go b/apps/backend/pkg/websocket/actions.go index e341704f60..03e88e7c95 100644 --- a/apps/backend/pkg/websocket/actions.go +++ b/apps/backend/pkg/websocket/actions.go @@ -119,6 +119,7 @@ const ( ActionColumnUpdated = "column.updated" ActionColumnDeleted = "column.deleted" ActionCommentAdded = "comment.added" + ActionCommentUpdated = "comment.updated" ActionInputRequested = "input.requested" ActionRepositoryCreated = "repository.created" ActionRepositoryUpdated = "repository.updated" diff --git a/apps/web/lib/state/store.ts b/apps/web/lib/state/store.ts index 07e5270112..6f57e7fb74 100644 --- a/apps/web/lib/state/store.ts +++ b/apps/web/lib/state/store.ts @@ -190,6 +190,7 @@ export type AppState = { setComments: (taskId: string, comments: Comment[]) => void; setCommentsTaskId: (taskId: string | null) => void; addComment: (comment: Comment) => void; + updateComment: (comment: Comment) => void; setCommentsLoading: (loading: boolean) => void; setGitStatus: (taskId: string, gitStatus: Omit) => void; clearGitStatus: () => void; @@ -252,6 +253,7 @@ const defaultState: AppState = { setComments: () => undefined, setCommentsTaskId: () => undefined, addComment: () => undefined, + updateComment: () => undefined, setCommentsLoading: () => undefined, setGitStatus: () => undefined, clearGitStatus: () => undefined, @@ -283,6 +285,7 @@ function mergeInitialState( | 'setComments' | 'setCommentsTaskId' | 'addComment' + | 'updateComment' | 'setCommentsLoading' | 'setGitStatus' | 'clearGitStatus' @@ -459,6 +462,16 @@ export function createAppStore(initialState?: Partial) { } } }), + updateComment: (comment) => + set((draft) => { + // Only update if this comment is for the current task + if (draft.comments.taskId === comment.task_id && draft.comments.items) { + const index = draft.comments.items.findIndex((c) => c.id === comment.id); + if (index !== -1) { + draft.comments.items[index] = comment; + } + } + }), setCommentsLoading: (loading) => set((draft) => { draft.comments.isLoading = loading; diff --git a/apps/web/lib/types/backend.ts b/apps/web/lib/types/backend.ts index 62c3f32497..670c3ffe15 100644 --- a/apps/web/lib/types/backend.ts +++ b/apps/web/lib/types/backend.ts @@ -19,6 +19,7 @@ export type BackendMessageType = | 'column.updated' | 'column.deleted' | 'comment.added' + | 'comment.updated' | 'executor.created' | 'executor.updated' | 'executor.deleted' @@ -234,6 +235,7 @@ export type BackendMessageMap = { 'column.updated': BackendMessage<'column.updated', ColumnPayload>; 'column.deleted': BackendMessage<'column.deleted', ColumnPayload>; 'comment.added': BackendMessage<'comment.added', CommentAddedPayload>; + 'comment.updated': BackendMessage<'comment.updated', CommentAddedPayload>; 'executor.created': BackendMessage<'executor.created', ExecutorPayload>; 'executor.updated': BackendMessage<'executor.updated', ExecutorPayload>; 'executor.deleted': BackendMessage<'executor.deleted', ExecutorPayload>; diff --git a/apps/web/lib/ws/handlers/comments.ts b/apps/web/lib/ws/handlers/comments.ts index a30303dab5..570795e92e 100644 --- a/apps/web/lib/ws/handlers/comments.ts +++ b/apps/web/lib/ws/handlers/comments.ts @@ -25,5 +25,22 @@ export function registerCommentsHandlers(store: StoreApi): WsHandlers }); console.log('[WS] addComment called'); }, + 'comment.updated': (message) => { + const payload = message.payload; + console.log('[WS] comment.updated payload:', JSON.stringify(payload, null, 2)); + const state = store.getState(); + state.updateComment({ + id: payload.comment_id, + task_id: payload.task_id, + author_type: payload.author_type, + author_id: payload.author_id, + content: payload.content, + type: (payload.type as 'message' | 'content' | 'tool_call' | 'progress' | 'error' | 'status') || 'message', + metadata: payload.metadata, + requests_input: payload.requests_input, + created_at: payload.created_at, + }); + console.log('[WS] updateComment called'); + }, }; }