dreampiler/kandev
1
0
Fork 0
Code Issues 43 Pull Requests Actions 2 Packages Projects Releases Wiki Activity

refactor: unify permission requests into agent event stream (#31)

Browse Source
This commit is contained in:
João Salavisa
2026-01-17 22:21:10 +00:00
committed by GitHub
parent dce2c6fe40
commit dd68da4275
45 changed files with 1507 additions and 1225 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/backend/cmd/kandev/main.go
+37 -69
View File
@@ -322,13 +322,13 @@ func main() {
var containerMgr *lifecycle.ContainerManager
switch cfg.Agent.Runtime {
case "standalone":
standaloneCtl := agentctl.NewStandaloneCtl(
controlClient := agentctl.NewControlClient(
cfg.Agent.StandaloneHost,
cfg.Agent.StandalonePort,
log,
)
agentRuntime = lifecycle.NewStandaloneRuntime(
standaloneCtl,
controlClient,
cfg.Agent.StandaloneHost,
cfg.Agent.StandalonePort,
log,
@@ -421,7 +421,8 @@ func main() {
taskSvc.SetExecutionStopper(orchestratorSvc)
// Set up comment creator for saving agent responses as comments
orchestratorSvc.SetMessageCreator(&messageCreatorAdapter{svc: taskSvc})
msgCreator := &messageCreatorAdapter{svc: taskSvc}
orchestratorSvc.SetMessageCreator(msgCreator)
// ============================================
// WEBSOCKET GATEWAY (All communication via WebSocket)
@@ -554,44 +555,8 @@ func main() {
})
log.Info("Historical logs provider configured for task subscriptions (using messages and git status)")
// Set up pending permissions provider for task subscriptions
gateway.Hub.SetPendingPermissionsProvider(func(ctx context.Context, taskID string) ([]*ws.Message, error) {
if lifecycleMgr == nil {
return nil, nil
}
session, err := taskRepo.GetActiveTaskSessionByTaskID(ctx, taskID)
if err != nil || session == nil || session.ID == "" {
return nil, nil
}
permissions, err := lifecycleMgr.GetPendingPermissionsForSession(ctx, session.ID)
if err != nil {
return nil, err
}
result := make([]*ws.Message, 0, len(permissions))
for _, p := range permissions {
// Build payload matching the permission.requested format
payload := map[string]interface{}{
"type": "permission_request",
"task_id": taskID,
"pending_id": p.PendingID,
"session_id": p.SessionID,
"tool_call_id": p.ToolCallID,
"title": p.Title,
"options": p.Options,
"action_type": p.ActionType,
"action_details": p.ActionDetails,
"created_at": p.CreatedAt.Format(time.RFC3339),
}
notification, err := ws.NewNotification(ws.ActionPermissionRequested, payload)
if err != nil {
continue
}
result = append(result, notification)
}
return result, nil
})
log.Info("Pending permissions provider configured for task subscriptions")
// NOTE: Pending permissions are now stored as messages and retrieved via the historical logs provider.
// The separate pending permissions provider is no longer needed.
// NOTE: We no longer create comments for each ACP message chunk.
// Instead, the lifecycle manager accumulates message chunks and:
@@ -600,35 +565,9 @@ func main() {
// This is handled via the prompt.complete and step.complete events.
// Wire ACP handler to broadcast to WebSocket clients as notifications
// NOTE: Permission requests are now handled via message creation in the lifecycle manager
// and broadcast automatically via the session.message.added event
orchestratorSvc.RegisterACPHandler(func(taskID string, msg *protocol.Message) {
// Check for permission_request events (type is in msg.Type, payload in msg.Data)
if msg.Type == "permission_request" {
sessionID := msg.SessionID
if sessionID == "" {
log.Warn("permission_request missing session_id, skipping",
zap.String("task_id", taskID))
return
}
// Build payload for frontend with all necessary fields
payload := map[string]interface{}{
"type": "permission_request",
"task_id": taskID,
"agent_instance_id": msg.AgentID,
"timestamp": msg.Timestamp,
}
// Copy all fields from msg.Data (pending_id, options, action_type, etc.)
for k, v := range msg.Data {
payload[k] = v
}
// Broadcast permission.requested notification to task subscribers
notification, _ := ws.NewNotification(ws.ActionPermissionRequested, payload)
gateway.Hub.BroadcastToSession(sessionID, notification)
log.Info("broadcasted permission.requested notification",
zap.String("task_id", taskID),
zap.Any("pending_id", payload["pending_id"]))
return
}
// Only broadcast message types that the frontend handles
// Skip internal message types like session/update
switch msg.Type {
@@ -1311,3 +1250,32 @@ func (a *messageCreatorAdapter) CreateSessionMessage(ctx context.Context, taskID
})
return err
}
// CreatePermissionRequestMessage creates a message for a permission request
func (a *messageCreatorAdapter) CreatePermissionRequestMessage(ctx context.Context, taskID, sessionID, pendingID, toolCallID, title string, options []map[string]interface{}, actionType string, actionDetails map[string]interface{}) (string, error) {
metadata := map[string]interface{}{
"pending_id": pendingID,
"tool_call_id": toolCallID,
"options": options,
"action_type": actionType,
"action_details": actionDetails,
}
msg, err := a.svc.CreateMessage(ctx, &taskservice.CreateMessageRequest{
TaskSessionID: sessionID,
TaskID: taskID,
Content: title,
AuthorType: "agent",
Type: "permission_request",
Metadata: metadata,
})
if err != nil {
return "", err
}
return msg.ID, nil
}
// UpdatePermissionMessage updates a permission message's status
func (a *messageCreatorAdapter) UpdatePermissionMessage(ctx context.Context, sessionID, pendingID, status string) error {
return a.svc.UpdatePermissionMessage(ctx, sessionID, pendingID, status)
}
apps/backend/internal/agent/lifecycle/container.go
+3 -3
View File
@@ -78,8 +78,8 @@ func (cm *ContainerManager) LaunchContainer(ctx context.Context, config Containe
containerIP = "127.0.0.1"
}
// Create StandaloneCtl to communicate with the container's ControlServer
ctl := agentctl.NewStandaloneCtl(containerIP, AgentCtlPort, cm.logger)
// Create ControlClient to communicate with the container's control server
ctl := agentctl.NewControlClient(containerIP, AgentCtlPort, cm.logger)
// Wait for agentctl to be healthy
if err := cm.waitForHealth(ctx, ctl); err != nil {
@@ -115,7 +115,7 @@ func (cm *ContainerManager) LaunchContainer(ctx context.Context, config Containe
}
// waitForHealth waits for agentctl to be healthy with retries
func (cm *ContainerManager) waitForHealth(ctx context.Context, ctl *agentctl.StandaloneCtl) error {
func (cm *ContainerManager) waitForHealth(ctx context.Context, ctl *agentctl.ControlClient) error {
const maxRetries = 30
const retryDelay = 500 * time.Millisecond
apps/backend/internal/agent/lifecycle/events.go
+59 -67
View File
@@ -114,63 +114,64 @@ func (p *EventPublisher) PublishACPSessionCreated(execution *AgentExecution, ses
}
}
// PublishSessionUpdate publishes a session update to the event bus for WebSocket streaming.
func (p *EventPublisher) PublishSessionUpdate(execution *AgentExecution, update agentctl.SessionUpdate) {
// PublishAgentStreamEvent publishes an agent stream event to the event bus for WebSocket streaming.
// This is different from PublishAgentEvent which publishes lifecycle events (started, stopped, etc.).
func (p *EventPublisher) PublishAgentStreamEvent(execution *AgentExecution, event agentctl.AgentEvent) {
if p.eventBus == nil {
return
}
// Build the update data - our SessionUpdate type marshals cleanly
updateData := map[string]interface{}{
"type": update.Type,
// Build the event data - our AgentEvent type marshals cleanly
eventData := map[string]interface{}{
"type": event.Type,
}
if update.SessionID != "" {
updateData["session_id"] = update.SessionID
if event.SessionID != "" {
eventData["session_id"] = event.SessionID
}
if update.Text != "" {
updateData["text"] = update.Text
if event.Text != "" {
eventData["text"] = event.Text
}
if update.ToolCallID != "" {
updateData["tool_call_id"] = update.ToolCallID
if event.ToolCallID != "" {
eventData["tool_call_id"] = event.ToolCallID
}
if update.ToolName != "" {
updateData["tool_name"] = update.ToolName
if event.ToolName != "" {
eventData["tool_name"] = event.ToolName
}
if update.ToolTitle != "" {
updateData["tool_title"] = update.ToolTitle
if event.ToolTitle != "" {
eventData["tool_title"] = event.ToolTitle
}
if update.ToolStatus != "" {
updateData["tool_status"] = update.ToolStatus
if event.ToolStatus != "" {
eventData["tool_status"] = event.ToolStatus
}
if update.ToolArgs != nil {
updateData["tool_args"] = update.ToolArgs
if event.ToolArgs != nil {
eventData["tool_args"] = event.ToolArgs
}
if update.ToolResult != nil {
updateData["tool_result"] = update.ToolResult
if event.ToolResult != nil {
eventData["tool_result"] = event.ToolResult
}
if update.Error != "" {
updateData["error"] = update.Error
if event.Error != "" {
eventData["error"] = event.Error
}
if update.Data != nil {
updateData["data"] = update.Data
if event.Data != nil {
eventData["data"] = event.Data
}
// Build ACP message data
// Build agent event message data
data := map[string]interface{}{
"type": "session/update",
"type": "agent/event",
"timestamp": time.Now().UTC().Format(time.RFC3339Nano),
"agent_id": execution.ID,
"task_id": execution.TaskID,
"session_id": update.SessionID,
"data": updateData,
"session_id": event.SessionID,
"data": eventData,
}
event := bus.NewEvent(events.ACPMessage, "agent-manager", data)
busEvent := bus.NewEvent(events.ACPMessage, "agent-manager", data)
subject := events.BuildACPSubject(execution.TaskID)
if err := p.eventBus.Publish(context.Background(), subject, event); err != nil {
p.logger.Error("failed to publish session update",
if err := p.eventBus.Publish(context.Background(), subject, busEvent); err != nil {
p.logger.Error("failed to publish agent stream event",
zap.String("instance_id", execution.ID),
zap.String("task_id", execution.TaskID),
zap.Error(err))
@@ -320,8 +321,8 @@ func (p *EventPublisher) PublishToolCall(execution *AgentExecution, toolCallID,
}
}
// PublishToolCallComplete publishes a tool call completion event from a SessionUpdate.
func (p *EventPublisher) PublishToolCallComplete(execution *AgentExecution, update agentctl.SessionUpdate) {
// PublishToolCallComplete publishes a tool call completion event from an AgentEvent.
func (p *EventPublisher) PublishToolCallComplete(execution *AgentExecution, event agentctl.AgentEvent) {
if p.eventBus == nil {
return
}
@@ -334,14 +335,14 @@ func (p *EventPublisher) PublishToolCallComplete(execution *AgentExecution, upda
"agent_id": execution.ID,
"task_id": execution.TaskID,
"session_id": sessionID,
"tool_call_id": update.ToolCallID,
"status": update.ToolStatus,
"tool_call_id": event.ToolCallID,
"status": event.ToolStatus,
}
event := bus.NewEvent(events.ToolCallComplete, "agent-manager", data)
busEvent := bus.NewEvent(events.ToolCallComplete, "agent-manager", data)
subject := events.ToolCallComplete + "." + execution.TaskID
if err := p.eventBus.Publish(context.Background(), subject, event); err != nil {
if err := p.eventBus.Publish(context.Background(), subject, busEvent); err != nil {
p.logger.Error("failed to publish tool_call_complete event",
zap.String("instance_id", execution.ID),
zap.String("task_id", execution.TaskID),
@@ -349,16 +350,15 @@ func (p *EventPublisher) PublishToolCallComplete(execution *AgentExecution, upda
}
}
// PublishPermissionRequest publishes a permission request event.
// The event is structured as a protocol.Message so it can be parsed by the watcher.
func (p *EventPublisher) PublishPermissionRequest(execution *AgentExecution, notification *agentctl.PermissionNotification) {
// PublishPermissionRequest publishes a permission request event to the event bus.
func (p *EventPublisher) PublishPermissionRequest(execution *AgentExecution, event agentctl.AgentEvent) {
if p.eventBus == nil {
return
}
// Convert options to a serializable format
options := make([]map[string]interface{}, len(notification.Options))
for i, opt := range notification.Options {
options := make([]map[string]interface{}, len(event.PermissionOptions))
for i, opt := range event.PermissionOptions {
options[i] = map[string]interface{}{
"option_id": opt.OptionID,
"name": opt.Name,
@@ -366,32 +366,24 @@ func (p *EventPublisher) PublishPermissionRequest(execution *AgentExecution, not
}
}
// Build payload data (goes into msg.Data when parsed as protocol.Message)
payloadData := map[string]interface{}{
"pending_id": notification.PendingID,
"session_id": notification.SessionID,
"tool_call_id": notification.ToolCallID,
"title": notification.Title,
"options": options,
"action_type": notification.ActionType,
"action_details": notification.ActionDetails,
"created_at": notification.CreatedAt,
}
// Structure as protocol.Message so watcher can parse it correctly
data := map[string]interface{}{
"type": "permission_request",
"timestamp": time.Now().UTC().Format(time.RFC3339Nano),
"agent_id": execution.ID,
"task_id": execution.TaskID,
"session_id": notification.SessionID,
"data": payloadData,
"type": "permission_request",
"timestamp": time.Now().UTC().Format(time.RFC3339Nano),
"agent_id": execution.ID,
"task_id": execution.TaskID,
"session_id": execution.SessionID,
"pending_id": event.PendingID,
"tool_call_id": event.ToolCallID,
"title": event.PermissionTitle,
"options": options,
"action_type": event.ActionType,
"action_details": event.ActionDetails,
}
event := bus.NewEvent(events.ACPMessage, "agent-manager", data)
subject := events.BuildACPSubject(execution.TaskID)
busEvent := bus.NewEvent(events.PermissionRequestReceived, "agent-manager", data)
subject := events.PermissionRequestReceived + "." + execution.TaskID
if err := p.eventBus.Publish(context.Background(), subject, event); err != nil {
if err := p.eventBus.Publish(context.Background(), subject, busEvent); err != nil {
p.logger.Error("failed to publish permission_request event",
zap.String("instance_id", execution.ID),
zap.String("task_id", execution.TaskID),
@@ -399,7 +391,7 @@ func (p *EventPublisher) PublishPermissionRequest(execution *AgentExecution, not
} else {
p.logger.Debug("published permission_request event",
zap.String("task_id", execution.TaskID),
zap.String("pending_id", notification.PendingID),
zap.String("title", notification.Title))
zap.String("pending_id", event.PendingID),
zap.String("title", event.PermissionTitle))
}
}
apps/backend/internal/agent/lifecycle/manager.go
+35 -41
View File
@@ -99,10 +99,9 @@ func NewManager(
// Initialize stream manager with callbacks that delegate to manager methods
mgr.streamManager = NewStreamManager(log, StreamCallbacks{
OnSessionUpdate: mgr.handleSessionUpdate,
OnPermission: mgr.handlePermissionNotification,
OnGitStatus: mgr.handleGitStatusUpdate,
OnFileChange: mgr.handleFileChangeNotification,
OnAgentEvent: mgr.handleAgentEvent,
OnGitStatus: mgr.handleGitStatusUpdate,
OnFileChange: mgr.handleFileChangeNotification,
})
// Set session manager dependencies for full orchestration
@@ -708,34 +707,33 @@ func (m *Manager) initializeACPSession(ctx context.Context, execution *AgentExec
return m.sessionManager.InitializeAndPrompt(ctx, execution, agentConfig, taskDescription, m.MarkReady)
}
// handlePermissionNotification processes incoming permission requests from the agent
func (m *Manager) handlePermissionNotification(execution *AgentExecution, notification *agentctl.PermissionNotification) {
// Publish permission request event to the event bus (logged at Debug level there)
m.eventPublisher.PublishPermissionRequest(execution, notification)
// handlePermissionRequestEvent processes permission requests from the unified agent event stream
func (m *Manager) handlePermissionRequestEvent(execution *AgentExecution, event agentctl.AgentEvent) {
m.eventPublisher.PublishPermissionRequest(execution, event)
}
// handleSessionUpdate processes incoming session updates from the agent
func (m *Manager) handleSessionUpdate(execution *AgentExecution, update agentctl.SessionUpdate) {
// Handle different update types based on the Type field
switch update.Type {
// handleAgentEvent processes incoming agent events from the agent
func (m *Manager) handleAgentEvent(execution *AgentExecution, event agentctl.AgentEvent) {
// Handle different event types based on the Type field
switch event.Type {
case "message_chunk":
m.updateExecutionProgress(execution.ID, 50)
// Accumulate message content for saving as comment when a step completes
if update.Text != "" {
if event.Text != "" {
execution.messageMu.Lock()
execution.messageBuffer.WriteString(update.Text)
execution.messageBuffer.WriteString(event.Text)
execution.messageMu.Unlock()
}
case "reasoning":
// Accumulate reasoning/thinking content
execution.messageMu.Lock()
if update.ReasoningText != "" {
execution.reasoningBuffer.WriteString(update.ReasoningText)
if event.ReasoningText != "" {
execution.reasoningBuffer.WriteString(event.ReasoningText)
}
if update.ReasoningSummary != "" {
execution.summaryBuffer.WriteString(update.ReasoningSummary)
if event.ReasoningSummary != "" {
execution.summaryBuffer.WriteString(event.ReasoningSummary)
}
execution.messageMu.Unlock()
@@ -746,21 +744,21 @@ func (m *Manager) handleSessionUpdate(execution *AgentExecution, update agentctl
m.logger.Debug("tool call started",
zap.String("execution_id", execution.ID),
zap.String("tool_call_id", update.ToolCallID),
zap.String("tool_name", update.ToolName))
zap.String("tool_call_id", event.ToolCallID),
zap.String("tool_name", event.ToolName))
m.updateExecutionProgress(execution.ID, 60)
// Publish tool call as a comment so it appears in the chat
m.eventPublisher.PublishToolCall(execution, update.ToolCallID, update.ToolTitle, update.ToolStatus, update.ToolArgs)
m.eventPublisher.PublishToolCall(execution, event.ToolCallID, event.ToolTitle, event.ToolStatus, event.ToolArgs)
case "tool_update":
// Check if tool call completed
switch update.ToolStatus {
switch event.ToolStatus {
case "complete", "completed":
m.updateExecutionProgress(execution.ID, 80)
m.eventPublisher.PublishToolCallComplete(execution, update)
m.eventPublisher.PublishToolCallComplete(execution, event)
case "error", "failed":
m.eventPublisher.PublishToolCallComplete(execution, update)
m.eventPublisher.PublishToolCallComplete(execution, event)
}
case "plan":
@@ -770,12 +768,12 @@ func (m *Manager) handleSessionUpdate(execution *AgentExecution, update agentctl
case "error":
m.logger.Error("agent error",
zap.String("execution_id", execution.ID),
zap.String("error", update.Error))
zap.String("error", event.Error))
case "complete":
m.logger.Info("agent turn complete",
zap.String("execution_id", execution.ID),
zap.String("session_id", update.SessionID))
zap.String("session_id", event.SessionID))
// Flush accumulated message buffer as a comment
m.flushMessageBufferAsComment(execution)
@@ -786,10 +784,19 @@ func (m *Manager) handleSessionUpdate(execution *AgentExecution, update agentctl
zap.String("execution_id", execution.ID),
zap.Error(err))
}
case "permission_request":
m.logger.Debug("permission request received",
zap.String("execution_id", execution.ID),
zap.String("pending_id", event.PendingID),
zap.String("title", event.PermissionTitle))
// Handle permission request inline
m.handlePermissionRequestEvent(execution, event)
}
// Publish session update to event bus for WebSocket streaming
m.eventPublisher.PublishSessionUpdate(execution, update)
// Publish agent stream event to event bus for WebSocket streaming
m.eventPublisher.PublishAgentStreamEvent(execution, event)
}
// handleGitStatusUpdate processes git status updates from the workspace tracker
@@ -1195,17 +1202,4 @@ func (m *Manager) RespondToPermissionBySessionID(sessionID, pendingID, optionID
return m.RespondToPermission(execution.ID, pendingID, optionID, cancelled)
}
// GetPendingPermissionsForSession returns all pending permissions for a session
func (m *Manager) GetPendingPermissionsForSession(ctx context.Context, sessionID string) ([]agentctl.PermissionNotification, error) {
execution, exists := m.executionStore.GetBySessionID(sessionID)
if !exists {
// No active execution for this session, so no pending permissions
return nil, nil
}
if execution.agentctl == nil {
return nil, nil
}
return execution.agentctl.GetPendingPermissions(ctx)
}
apps/backend/internal/agent/lifecycle/runtime_docker.go
+2 -2
View File
@@ -154,8 +154,8 @@ func (r *DockerRuntime) RecoverInstances(ctx context.Context) ([]*RuntimeInstanc
containerIP = "127.0.0.1"
}
// Query the ControlServer to get the instance port
ctl := agentctl.NewStandaloneCtl(containerIP, AgentCtlPort, r.logger)
// Query the control server to get the instance port
ctl := agentctl.NewControlClient(containerIP, AgentCtlPort, r.logger)
instanceInfo, err := ctl.GetInstance(ctx, instanceID)
if err != nil {
r.logger.Warn("failed to get instance info from container",
apps/backend/internal/agent/lifecycle/runtime_standalone.go
+2 -2
View File
@@ -14,14 +14,14 @@ import (
// StandaloneRuntime implements Runtime for standalone agentctl execution.
// In this mode, a single agentctl control server manages multiple agent instances.
type StandaloneRuntime struct {
ctl *agentctl.StandaloneCtl
ctl *agentctl.ControlClient
host string
port int
logger *logger.Logger
}
// NewStandaloneRuntime creates a new standalone runtime.
func NewStandaloneRuntime(ctl *agentctl.StandaloneCtl, host string, port int, log *logger.Logger) *StandaloneRuntime {
func NewStandaloneRuntime(ctl *agentctl.ControlClient, host string, port int, log *logger.Logger) *StandaloneRuntime {
return &StandaloneRuntime{
ctl: ctl,
host: host,
apps/backend/internal/agent/lifecycle/streams.go
+6 -29
View File
@@ -12,10 +12,9 @@ import (
// StreamCallbacks defines callbacks for stream events
type StreamCallbacks struct {
OnSessionUpdate func(execution *AgentExecution, update agentctl.SessionUpdate)
OnPermission func(execution *AgentExecution, notification *agentctl.PermissionNotification)
OnGitStatus func(execution *AgentExecution, update *agentctl.GitStatusUpdate)
OnFileChange func(execution *AgentExecution, notification *agentctl.FileChangeNotification)
OnAgentEvent func(execution *AgentExecution, event agentctl.AgentEvent)
OnGitStatus func(execution *AgentExecution, update *agentctl.GitStatusUpdate)
OnFileChange func(execution *AgentExecution, notification *agentctl.FileChangeNotification)
}
// StreamManager manages WebSocket streams to agent executions
@@ -36,7 +35,6 @@ func NewStreamManager(log *logger.Logger, callbacks StreamCallbacks) *StreamMana
// If ready is non-nil, it will be closed when the updates stream is connected.
func (sm *StreamManager) ConnectAll(execution *AgentExecution, ready chan<- struct{}) {
go sm.connectUpdatesStream(execution, ready)
go sm.connectPermissionStream(execution)
go sm.connectGitStatusStream(execution)
go sm.connectFileChangesStream(execution)
}
@@ -74,9 +72,9 @@ func (sm *StreamManager) ReconnectAll(execution *AgentExecution) {
func (sm *StreamManager) connectUpdatesStream(execution *AgentExecution, ready chan<- struct{}) {
ctx := context.Background()
err := execution.agentctl.StreamUpdates(ctx, func(update agentctl.SessionUpdate) {
if sm.callbacks.OnSessionUpdate != nil {
sm.callbacks.OnSessionUpdate(execution, update)
err := execution.agentctl.StreamUpdates(ctx, func(event agentctl.AgentEvent) {
if sm.callbacks.OnAgentEvent != nil {
sm.callbacks.OnAgentEvent(execution, event)
}
})
@@ -94,28 +92,7 @@ func (sm *StreamManager) connectUpdatesStream(execution *AgentExecution, ready c
}
}
// connectPermissionStream handles the permission WebSocket stream
func (sm *StreamManager) connectPermissionStream(execution *AgentExecution) {
ctx := context.Background()
sm.logger.Debug("connecting to permission stream",
zap.String("instance_id", execution.ID),
zap.String("task_id", execution.TaskID))
err := execution.agentctl.StreamPermissions(ctx, func(notification *agentctl.PermissionNotification) {
sm.logger.Debug("received permission notification from stream",
zap.String("pending_id", notification.PendingID),
zap.String("title", notification.Title))
if sm.callbacks.OnPermission != nil {
sm.callbacks.OnPermission(execution, notification)
}
})
if err != nil {
sm.logger.Error("failed to connect to permission stream",
zap.String("instance_id", execution.ID),
zap.Error(err))
}
}
// connectGitStatusStream handles git status stream with retry logic
func (sm *StreamManager) connectGitStatusStream(execution *AgentExecution) {
apps/backend/internal/agentctl/client/acp.go
+21 -166
View File
@@ -10,9 +10,19 @@ import (
"github.com/coder/acp-go-sdk"
"github.com/gorilla/websocket"
"github.com/kandev/kandev/internal/agentctl/types/streams"
"go.uber.org/zap"
)
// Re-export stream types for convenience.
type (
AgentEvent = streams.AgentEvent
PermissionNotification = streams.PermissionNotification
PermissionOption = streams.PermissionOption
PermissionRespondRequest = streams.PermissionRespondRequest
PermissionRespondResponse = streams.PermissionRespondResponse
)
// AgentInfo contains information about the connected agent.
type AgentInfo struct {
Name string `json:"name"`
@@ -237,70 +247,12 @@ func (c *Client) Prompt(ctx context.Context, text string) (*PromptResponse, erro
return &result, nil
}
// 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"`
OperationID string `json:"operation_id,omitempty"` // Turn/operation ID for cancellation
// Note: AgentEvent, PermissionNotification, PermissionOption, and
// PermissionRespondRequest are re-exported from streams package at the top of this file.
// 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"`
// 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
type PermissionOption struct {
OptionID string `json:"option_id"`
Name string `json:"name"`
Kind string `json:"kind"` // allow_once, allow_always, reject_once, reject_always
}
// PermissionNotification is received when the agent requests permission
type PermissionNotification struct {
PendingID string `json:"pending_id"`
SessionID string `json:"session_id"`
ToolCallID string `json:"tool_call_id"`
Title string `json:"title"`
Options []PermissionOption `json:"options"`
ActionType string `json:"action_type,omitempty"` // command, file_write, file_read, network, mcp_tool, other
ActionDetails map[string]interface{} `json:"action_details,omitempty"` // Additional details about the action
CreatedAt time.Time `json:"created_at"`
}
// PermissionResponse is the user's response to a permission request
type PermissionRespondRequest struct {
PendingID string `json:"pending_id"`
OptionID string `json:"option_id,omitempty"`
Cancelled bool `json:"cancelled,omitempty"`
}
// PermissionRespondResponse from agentctl
type PermissionRespondResponse struct {
Success bool `json:"success"`
Error string `json:"error,omitempty"`
}
// StreamUpdates opens a WebSocket connection for streaming session updates
func (c *Client) StreamUpdates(ctx context.Context, handler func(SessionUpdate)) error {
// StreamUpdates opens a WebSocket connection for streaming agent events.
// Events include message chunks, reasoning, tool calls, plan updates, and completion/error.
func (c *Client) StreamUpdates(ctx context.Context, handler func(AgentEvent)) error {
wsURL := "ws" + c.baseURL[4:] + "/api/v1/acp/stream"
conn, _, err := websocket.DefaultDialer.DialContext(ctx, wsURL, nil)
@@ -336,99 +288,33 @@ func (c *Client) StreamUpdates(ctx context.Context, handler func(SessionUpdate))
return
}
var update SessionUpdate
if err := json.Unmarshal(message, &update); err != nil {
c.logger.Warn("failed to parse session update", zap.Error(err))
var event AgentEvent
if err := json.Unmarshal(message, &event); err != nil {
c.logger.Warn("failed to parse agent event", zap.Error(err))
continue
}
handler(update)
handler(event)
}
}()
return nil
}
// CloseUpdatesStream closes the updates stream connection
// CloseUpdatesStream closes the agent events stream connection.
func (c *Client) CloseUpdatesStream() {
c.CloseACPStream()
}
// CloseACPStream closes the ACP stream connection
func (c *Client) CloseACPStream() {
c.mu.Lock()
defer c.mu.Unlock()
if c.acpConn != nil {
if err := c.acpConn.Close(); err != nil {
c.logger.Debug("failed to close ACP stream", zap.Error(err))
c.logger.Debug("failed to close agent events stream", zap.Error(err))
}
c.acpConn = nil
}
}
// StreamPermissions opens a WebSocket connection for streaming permission requests
func (c *Client) StreamPermissions(ctx context.Context, handler func(*PermissionNotification)) error {
wsURL := "ws" + c.baseURL[4:] + "/api/v1/acp/permissions/stream"
conn, _, err := websocket.DefaultDialer.DialContext(ctx, wsURL, nil)
if err != nil {
return fmt.Errorf("failed to connect to permission stream: %w", err)
}
c.mu.Lock()
c.permissionConn = conn
c.mu.Unlock()
c.logger.Info("connected to permission stream", zap.String("url", wsURL))
// Read messages in a goroutine
go func() {
defer func() {
c.mu.Lock()
c.permissionConn = nil
c.mu.Unlock()
if err := conn.Close(); err != nil {
c.logger.Debug("failed to close permissions websocket", zap.Error(err))
}
}()
for {
_, message, err := conn.ReadMessage()
if err != nil {
if websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseGoingAway) {
c.logger.Info("permission stream closed normally")
} else {
c.logger.Debug("permission stream error", zap.Error(err))
}
return
}
var notification PermissionNotification
if err := json.Unmarshal(message, &notification); err != nil {
c.logger.Warn("failed to parse permission notification", zap.Error(err))
continue
}
handler(&notification)
}
}()
return nil
}
// ClosePermissionStream closes the permission stream connection
func (c *Client) ClosePermissionStream() {
c.mu.Lock()
defer c.mu.Unlock()
if c.permissionConn != nil {
if err := c.permissionConn.Close(); err != nil {
c.logger.Debug("failed to close permission stream", zap.Error(err))
}
c.permissionConn = nil
}
}
// RespondToPermission sends a response to a permission request
func (c *Client) RespondToPermission(ctx context.Context, pendingID, optionID string, cancelled bool) error {
@@ -478,35 +364,4 @@ func (c *Client) RespondToPermission(ctx context.Context, pendingID, optionID st
return nil
}
// GetPendingPermissions retrieves all pending permission requests from agentctl
func (c *Client) GetPendingPermissions(ctx context.Context) ([]PermissionNotification, error) {
req, err := http.NewRequestWithContext(ctx, "GET", c.baseURL+"/api/v1/acp/permissions", nil)
if err != nil {
return nil, err
}
resp, err := c.httpClient.Do(req)
if err != nil {
return nil, err
}
defer func() {
if err := resp.Body.Close(); err != nil {
c.logger.Debug("failed to close permissions response body", zap.Error(err))
}
}()
respBody, err := readResponseBody(resp)
if err != nil {
return nil, fmt.Errorf("failed to read response body: %w", err)
}
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
return nil, fmt.Errorf("get pending permissions failed with status %d: %s", resp.StatusCode, string(respBody))
}
var result []PermissionNotification
if err := json.Unmarshal(respBody, &result); err != nil {
return nil, fmt.Errorf("failed to parse permissions response: %w", err)
}
return result, nil
}
apps/backend/internal/agentctl/client/client.go
+9 -25
View File
@@ -24,7 +24,6 @@ type Client struct {
// WebSocket connections for streaming
acpConn *websocket.Conn
permissionConn *websocket.Conn
gitStatusConn *websocket.Conn
fileChangesConn *websocket.Conn
shellConn *websocket.Conn
@@ -251,7 +250,8 @@ func (c *Client) BaseURL() string {
return c.baseURL
}
// Re-export types from types package for convenience
// Re-export types from types package for convenience.
// These types are defined in the streams subpackage and re-exported through types.
type (
GitStatusUpdate = types.GitStatusUpdate
FileInfo = types.FileInfo
@@ -263,6 +263,9 @@ type (
FileContentRequest = types.FileContentRequest
FileContentResponse = types.FileContentResponse
FileChangeNotification = types.FileChangeNotification
ShellMessage = types.ShellMessage
ShellStatusResponse = types.ShellStatusResponse
ShellBufferResponse = types.ShellBufferResponse
)
// StreamGitStatus opens a WebSocket connection for streaming git status updates
@@ -453,23 +456,16 @@ func (c *Client) CloseFileChangesStream() {
}
}
// Close closes all connections (ACP, permissions, git status, file changes, shell)
// Close closes all connections
func (c *Client) Close() {
c.CloseACPStream()
c.ClosePermissionStream()
c.CloseUpdatesStream()
c.CloseGitStatusStream()
c.CloseFileChangesStream()
c.CloseShellStream()
}
// ShellStatusResponse from agentctl shell status endpoint
type ShellStatusResponse struct {
Running bool `json:"running"`
Pid int `json:"pid"`
Shell string `json:"shell"`
Cwd string `json:"cwd"`
StartedAt string `json:"started_at"`
}
// Note: ShellStatusResponse, ShellMessage, ShellBufferResponse are re-exported
// from streams package at the top of this file in the type block.
// ShellStatus gets the status of the embedded shell session
func (c *Client) ShellStatus(ctx context.Context) (*ShellStatusResponse, error) {
@@ -496,18 +492,6 @@ func (c *Client) ShellStatus(ctx context.Context) (*ShellStatusResponse, error)
return &result, nil
}
// ShellMessage represents a message to/from the shell
type ShellMessage struct {
Type string `json:"type"` // "input", "output", "ping", "pong", "exit"
Data string `json:"data,omitempty"`
Code int `json:"code,omitempty"` // For exit type
}
// ShellBufferResponse is the response from the shell buffer endpoint
type ShellBufferResponse struct {
Data string `json:"data"`
}
// ShellBuffer returns the buffered shell output
func (c *Client) ShellBuffer(ctx context.Context) (string, error) {
req, err := http.NewRequestWithContext(ctx, "GET", c.baseURL+"/api/v1/shell/buffer", nil)
apps/backend/internal/agentctl/client/standalone.go → apps/backend/internal/agentctl/client/control.go
+26 -25
View File
@@ -1,5 +1,5 @@
// Package agentctl provides a client for communicating with agentctl.
// This file contains the StandaloneCtl client for the multi-instance control API.
// This file contains the ControlClient for the agentctl control server API.
package client
import (
@@ -14,9 +14,10 @@ import (
"go.uber.org/zap"
)
// StandaloneCtl is a client for the agentctl control API in standalone mode.
// ControlClient is a client for the agentctl control server API.
// It manages creation and deletion of agent instances.
type StandaloneCtl struct {
// Used by both Docker and standalone runtimes.
type ControlClient struct {
baseURL string
httpClient *http.Client
logger *logger.Logger
@@ -50,25 +51,25 @@ type InstanceInfo struct {
CreatedAt time.Time `json:"created_at"`
}
// NewStandaloneCtl creates a new StandaloneCtl client.
func NewStandaloneCtl(host string, port int, log *logger.Logger) *StandaloneCtl {
return &StandaloneCtl{
// NewControlClient creates a new ControlClient for the agentctl control server.
func NewControlClient(host string, port int, log *logger.Logger) *ControlClient {
return &ControlClient{
baseURL: fmt.Sprintf("http://%s:%d", host, port),
httpClient: &http.Client{
Timeout: 30 * time.Second,
},
logger: log.WithFields(zap.String("component", "standalone-agentctl")),
logger: log.WithFields(zap.String("component", "agentctl-control")),
}
}
// Health checks if the standalone agentctl is healthy.
func (s *StandaloneCtl) Health(ctx context.Context) error {
req, err := http.NewRequestWithContext(ctx, "GET", s.baseURL+"/health", nil)
// Health checks if the agentctl control server is healthy.
func (c *ControlClient) Health(ctx context.Context) error {
req, err := http.NewRequestWithContext(ctx, "GET", c.baseURL+"/health", nil)
if err != nil {
return err
}
resp, err := s.httpClient.Do(req)
resp, err := c.httpClient.Do(req)
if err != nil {
return err
}
@@ -83,19 +84,19 @@ func (s *StandaloneCtl) Health(ctx context.Context) error {
}
// CreateInstance creates a new agent instance.
func (s *StandaloneCtl) CreateInstance(ctx context.Context, req *CreateInstanceRequest) (*CreateInstanceResponse, error) {
func (c *ControlClient) CreateInstance(ctx context.Context, req *CreateInstanceRequest) (*CreateInstanceResponse, error) {
body, err := json.Marshal(req)
if err != nil {
return nil, fmt.Errorf("failed to marshal request: %w", err)
}
httpReq, err := http.NewRequestWithContext(ctx, "POST", s.baseURL+"/api/v1/instances", bytes.NewReader(body))
httpReq, err := http.NewRequestWithContext(ctx, "POST", c.baseURL+"/api/v1/instances", bytes.NewReader(body))
if err != nil {
return nil, err
}
httpReq.Header.Set("Content-Type", "application/json")
resp, err := s.httpClient.Do(httpReq)
resp, err := c.httpClient.Do(httpReq)
if err != nil {
return nil, fmt.Errorf("failed to create instance: %w", err)
}
@@ -116,7 +117,7 @@ func (s *StandaloneCtl) CreateInstance(ctx context.Context, req *CreateInstanceR
return nil, fmt.Errorf("failed to decode response: %w", err)
}
s.logger.Info("created agent instance",
c.logger.Info("created agent instance",
zap.String("instance_id", result.ID),
zap.Int("port", result.Port))
@@ -124,13 +125,13 @@ func (s *StandaloneCtl) CreateInstance(ctx context.Context, req *CreateInstanceR
}
// DeleteInstance stops and removes an agent instance.
func (s *StandaloneCtl) DeleteInstance(ctx context.Context, instanceID string) error {
req, err := http.NewRequestWithContext(ctx, "DELETE", s.baseURL+"/api/v1/instances/"+instanceID, nil)
func (c *ControlClient) DeleteInstance(ctx context.Context, instanceID string) error {
req, err := http.NewRequestWithContext(ctx, "DELETE", c.baseURL+"/api/v1/instances/"+instanceID, nil)
if err != nil {
return err
}
resp, err := s.httpClient.Do(req)
resp, err := c.httpClient.Do(req)
if err != nil {
return fmt.Errorf("failed to delete instance: %w", err)
}
@@ -148,18 +149,18 @@ func (s *StandaloneCtl) DeleteInstance(ctx context.Context, instanceID string) e
return fmt.Errorf("failed to delete instance: %s (status %d)", errResp.Error, resp.StatusCode)
}
s.logger.Info("deleted agent instance", zap.String("instance_id", instanceID))
c.logger.Info("deleted agent instance", zap.String("instance_id", instanceID))
return nil
}
// GetInstance gets information about a specific instance.
func (s *StandaloneCtl) GetInstance(ctx context.Context, instanceID string) (*InstanceInfo, error) {
req, err := http.NewRequestWithContext(ctx, "GET", s.baseURL+"/api/v1/instances/"+instanceID, nil)
func (c *ControlClient) GetInstance(ctx context.Context, instanceID string) (*InstanceInfo, error) {
req, err := http.NewRequestWithContext(ctx, "GET", c.baseURL+"/api/v1/instances/"+instanceID, nil)
if err != nil {
return nil, err
}
resp, err := s.httpClient.Do(req)
resp, err := c.httpClient.Do(req)
if err != nil {
return nil, fmt.Errorf("failed to get instance: %w", err)
}
@@ -180,13 +181,13 @@ func (s *StandaloneCtl) GetInstance(ctx context.Context, instanceID string) (*In
}
// ListInstances lists all running agent instances.
func (s *StandaloneCtl) ListInstances(ctx context.Context) ([]*InstanceInfo, error) {
req, err := http.NewRequestWithContext(ctx, "GET", s.baseURL+"/api/v1/instances", nil)
func (c *ControlClient) ListInstances(ctx context.Context) ([]*InstanceInfo, error) {
req, err := http.NewRequestWithContext(ctx, "GET", c.baseURL+"/api/v1/instances", nil)
if err != nil {
return nil, err
}
resp, err := s.httpClient.Do(req)
resp, err := c.httpClient.Do(req)
if err != nil {
return nil, fmt.Errorf("failed to list instances: %w", err)
}
apps/backend/internal/agentctl/server/adapter/acp_adapter.go
+21 -21
View File
@@ -34,7 +34,7 @@ type ACPAdapter struct {
agentInfo *AgentInfo
// Update channel
updatesCh chan SessionUpdate
updatesCh chan AgentEvent
// Permission handler
permissionHandler PermissionHandler
@@ -52,7 +52,7 @@ func NewACPAdapter(stdin io.Writer, stdout io.Reader, cfg *Config, log *logger.L
logger: log.WithFields(zap.String("adapter", "acp")),
stdin: stdin,
stdout: stdout,
updatesCh: make(chan SessionUpdate, 100),
updatesCh: make(chan AgentEvent, 100),
}
}
@@ -62,7 +62,7 @@ func (a *ACPAdapter) Initialize(ctx context.Context) error {
a.logger.Info("initializing ACP adapter",
zap.String("workdir", a.cfg.WorkDir))
// Create ACP client with update handler that converts to SessionUpdate
// Create ACP client with update handler that converts to AgentEvent
a.acpClient = acpclient.NewClient(
acpclient.WithLogger(a.logger.Zap()),
acpclient.WithWorkspaceRoot(a.cfg.WorkDir),
@@ -190,8 +190,8 @@ func (a *ACPAdapter) Cancel(ctx context.Context) error {
})
}
// Updates returns the channel for session updates.
func (a *ACPAdapter) Updates() <-chan SessionUpdate {
// Updates returns the channel for agent events.
func (a *ACPAdapter) Updates() <-chan AgentEvent {
return a.updatesCh
}
@@ -244,20 +244,20 @@ func (a *ACPAdapter) GetACPConnection() *acp.ClientSideConnection {
return a.acpConn
}
// handleACPUpdate converts ACP SessionNotification to protocol-agnostic SessionUpdate.
// handleACPUpdate converts ACP SessionNotification to protocol-agnostic AgentEvent.
func (a *ACPAdapter) handleACPUpdate(n acp.SessionNotification) {
update := a.convertNotification(n)
if update != nil {
event := a.convertNotification(n)
if event != nil {
select {
case a.updatesCh <- *update:
case a.updatesCh <- *event:
default:
a.logger.Warn("updates channel full, dropping notification")
}
}
}
// convertNotification converts an ACP SessionNotification to a SessionUpdate.
func (a *ACPAdapter) convertNotification(n acp.SessionNotification) *SessionUpdate {
// convertNotification converts an ACP SessionNotification to an AgentEvent.
func (a *ACPAdapter) convertNotification(n acp.SessionNotification) *AgentEvent {
u := n.Update
// Log what update types we're receiving
@@ -271,8 +271,8 @@ func (a *ACPAdapter) convertNotification(n acp.SessionNotification) *SessionUpda
switch {
case u.AgentMessageChunk != nil:
if u.AgentMessageChunk.Content.Text != nil {
return &SessionUpdate{
Type: UpdateTypeMessageChunk,
return &AgentEvent{
Type: EventTypeMessageChunk,
SessionID: string(n.SessionId),
Text: u.AgentMessageChunk.Content.Text.Text,
}
@@ -281,8 +281,8 @@ func (a *ACPAdapter) convertNotification(n acp.SessionNotification) *SessionUpda
case u.AgentThoughtChunk != nil:
// Agent thinking/reasoning - map to the new reasoning type
if u.AgentThoughtChunk.Content.Text != nil {
return &SessionUpdate{
Type: UpdateTypeReasoning,
return &AgentEvent{
Type: EventTypeReasoning,
SessionID: string(n.SessionId),
ReasoningText: u.AgentThoughtChunk.Content.Text.Text,
}
@@ -324,8 +324,8 @@ func (a *ACPAdapter) convertNotification(n acp.SessionNotification) *SessionUpda
status = "running"
}
return &SessionUpdate{
Type: UpdateTypeToolCall,
return &AgentEvent{
Type: EventTypeToolCall,
SessionID: string(n.SessionId),
ToolCallID: string(u.ToolCall.ToolCallId),
ToolName: string(u.ToolCall.Kind), // Kind is effectively the tool name
@@ -343,8 +343,8 @@ func (a *ACPAdapter) convertNotification(n acp.SessionNotification) *SessionUpda
if status == "completed" {
status = "complete"
}
return &SessionUpdate{
Type: UpdateTypeToolUpdate,
return &AgentEvent{
Type: EventTypeToolUpdate,
SessionID: string(n.SessionId),
ToolCallID: string(u.ToolCallUpdate.ToolCallId),
ToolStatus: status,
@@ -359,8 +359,8 @@ func (a *ACPAdapter) convertNotification(n acp.SessionNotification) *SessionUpda
Priority: string(e.Priority),
}
}
return &SessionUpdate{
Type: UpdateTypePlan,
return &AgentEvent{
Type: EventTypePlan,
SessionID: string(n.SessionId),
PlanEntries: entries,
}
apps/backend/internal/agentctl/server/adapter/adapter.go
+20 -63
View File
@@ -6,6 +6,7 @@ import (
"context"
"github.com/kandev/kandev/internal/agentctl/types"
"github.com/kandev/kandev/internal/agentctl/types/streams"
)
// Re-export permission types from the shared types package for convenience.
@@ -14,71 +15,27 @@ import (
type (
PermissionRequest = types.PermissionRequest
PermissionResponse = types.PermissionResponse
PermissionOption = types.PermissionOption
PermissionOption = streams.PermissionOption
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
// Re-export stream types for convenience.
type (
AgentEvent = streams.AgentEvent
PlanEntry = streams.PlanEntry
)
// 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. 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"`
ToolTitle string `json:"tool_title,omitempty"`
ToolStatus string `json:"tool_status,omitempty"` // "started", "running", "completed", "error"
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"`
// Error fields (for "error" type)
Error string `json:"error,omitempty"`
// Raw data for protocol-specific extensions
Data map[string]interface{} `json:"data,omitempty"`
}
// PlanEntry represents an entry in the agent's execution plan
type PlanEntry struct {
Description string `json:"description,omitempty"` // Content/description of the task
Status string `json:"status,omitempty"` // "pending", "in_progress", "completed", "failed"
Priority string `json:"priority,omitempty"` // Relative importance
}
// Re-export agent event type constants from streams package.
const (
EventTypeMessageChunk = streams.EventTypeMessageChunk
EventTypeReasoning = streams.EventTypeReasoning
EventTypeToolCall = streams.EventTypeToolCall
EventTypeToolUpdate = streams.EventTypeToolUpdate
EventTypePlan = streams.EventTypePlan
EventTypeComplete = streams.EventTypeComplete
EventTypeError = streams.EventTypeError
EventTypePermissionRequest = streams.EventTypePermissionRequest
)
// AgentInfo contains information about the connected agent.
type AgentInfo struct {
@@ -88,7 +45,7 @@ type AgentInfo struct {
// AgentAdapter defines the interface for protocol adapters.
// Each adapter translates a specific protocol (ACP, REST, MCP, etc.) into the
// normalized SessionUpdate format that agentctl exposes via its HTTP API.
// normalized AgentEvent format that agentctl exposes via its HTTP API.
type AgentAdapter interface {
// Initialize establishes the connection with the agent and exchanges capabilities.
// For subprocess-based agents (ACP), this sends the initialize request.
@@ -112,9 +69,9 @@ type AgentAdapter interface {
// Cancel cancels the current operation.
Cancel(ctx context.Context) error
// Updates returns a channel that receives session updates.
// Updates returns a channel that receives agent events.
// The channel is closed when the adapter is closed.
Updates() <-chan SessionUpdate
Updates() <-chan AgentEvent
// GetSessionID returns the current session ID.
GetSessionID() string
apps/backend/internal/agentctl/server/adapter/adapter_test.go
+7 -8
View File
@@ -6,7 +6,7 @@ import (
)
func TestSessionUpdate_Fields(t *testing.T) {
update := SessionUpdate{
update := AgentEvent{
Type: "tool_call",
SessionID: "session-123",
Text: "test message",
@@ -56,7 +56,7 @@ func TestSessionUpdate_Fields(t *testing.T) {
}
func TestSessionUpdate_JSONSerialization(t *testing.T) {
update := SessionUpdate{
update := AgentEvent{
Type: "message_chunk",
SessionID: "session-abc",
Text: "Hello, world!",
@@ -68,7 +68,7 @@ func TestSessionUpdate_JSONSerialization(t *testing.T) {
t.Fatalf("failed to marshal SessionUpdate: %v", err)
}
var decoded SessionUpdate
var decoded AgentEvent
if err := json.Unmarshal(data, &decoded); err != nil {
t.Fatalf("failed to unmarshal SessionUpdate: %v", err)
}
@@ -86,7 +86,7 @@ func TestSessionUpdate_JSONSerialization(t *testing.T) {
func TestSessionUpdate_JSONOmitEmpty(t *testing.T) {
// Test that omitempty fields are not included when empty
update := SessionUpdate{
update := AgentEvent{
Type: "error",
}
@@ -257,7 +257,7 @@ func TestSessionUpdate_UpdateTypes(t *testing.T) {
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
update := SessionUpdate{Type: tc.updateType}
update := AgentEvent{Type: tc.updateType}
if update.Type != tc.updateType {
t.Errorf("expected Type %q, got %q", tc.updateType, update.Type)
}
@@ -288,7 +288,7 @@ func TestPlanEntry_StatusValues(t *testing.T) {
func TestSessionUpdate_WithComplexToolArgs(t *testing.T) {
// Test with complex nested ToolArgs
update := SessionUpdate{
update := AgentEvent{
Type: "tool_call",
ToolCallID: "call-123",
ToolName: "file_edit",
@@ -311,7 +311,7 @@ func TestSessionUpdate_WithComplexToolArgs(t *testing.T) {
t.Fatalf("failed to marshal: %v", err)
}
var decoded SessionUpdate
var decoded AgentEvent
if err := json.Unmarshal(data, &decoded); err != nil {
t.Fatalf("failed to unmarshal: %v", err)
}
@@ -328,4 +328,3 @@ func TestSessionUpdate_WithComplexToolArgs(t *testing.T) {
t.Errorf("expected 2 locations, got %d", len(locations))
}
}
apps/backend/internal/agentctl/server/adapter/codex_adapter.go
+30 -30
View File
@@ -39,7 +39,7 @@ type CodexAdapter struct {
agentInfo *AgentInfo
// Update channel
updatesCh chan SessionUpdate
updatesCh chan AgentEvent
// Permission handler
permissionHandler PermissionHandler
@@ -65,7 +65,7 @@ func NewCodexAdapter(stdin io.Writer, stdout io.Reader, cfg *Config, log *logger
stdout: stdout,
ctx: ctx,
cancel: cancel,
updatesCh: make(chan SessionUpdate, 100),
updatesCh: make(chan AgentEvent, 100),
}
}
@@ -292,8 +292,8 @@ func (a *CodexAdapter) Cancel(ctx context.Context) error {
return err
}
// Updates returns the channel for session updates.
func (a *CodexAdapter) Updates() <-chan SessionUpdate {
// Updates returns the channel for agent events.
func (a *CodexAdapter) Updates() <-chan AgentEvent {
return a.updatesCh
}
@@ -346,8 +346,8 @@ func (a *CodexAdapter) Close() error {
return nil
}
// sendUpdate safely sends an update to the updates channel.
func (a *CodexAdapter) sendUpdate(update SessionUpdate) {
// sendUpdate safely sends an event to the updates channel.
func (a *CodexAdapter) sendUpdate(update AgentEvent) {
select {
case a.updatesCh <- update:
default:
@@ -355,7 +355,7 @@ func (a *CodexAdapter) sendUpdate(update SessionUpdate) {
}
}
// handleNotification processes Codex notifications and emits SessionUpdates.
// handleNotification processes Codex notifications and emits AgentEvents.
func (a *CodexAdapter) handleNotification(method string, params json.RawMessage) {
a.mu.RLock()
threadID := a.threadID
@@ -373,8 +373,8 @@ func (a *CodexAdapter) handleNotification(method string, params json.RawMessage)
a.mu.Lock()
a.messageBuffer += p.Delta
a.mu.Unlock()
a.sendUpdate(SessionUpdate{
Type: UpdateTypeMessageChunk,
a.sendUpdate(AgentEvent{
Type: EventTypeMessageChunk,
SessionID: threadID,
OperationID: turnID,
Text: p.Delta,
@@ -389,8 +389,8 @@ func (a *CodexAdapter) handleNotification(method string, params json.RawMessage)
a.mu.Lock()
a.reasoningBuffer += p.Delta
a.mu.Unlock()
a.sendUpdate(SessionUpdate{
Type: UpdateTypeReasoning,
a.sendUpdate(AgentEvent{
Type: EventTypeReasoning,
SessionID: threadID,
OperationID: turnID,
ReasoningText: p.Delta,
@@ -405,8 +405,8 @@ func (a *CodexAdapter) handleNotification(method string, params json.RawMessage)
a.mu.Lock()
a.summaryBuffer += p.Delta
a.mu.Unlock()
a.sendUpdate(SessionUpdate{
Type: UpdateTypeReasoning,
a.sendUpdate(AgentEvent{
Type: EventTypeReasoning,
SessionID: threadID,
OperationID: turnID,
ReasoningSummary: p.Delta,
@@ -418,13 +418,13 @@ func (a *CodexAdapter) handleNotification(method string, params json.RawMessage)
a.logger.Warn("failed to parse turn completed", zap.Error(err))
return
}
update := SessionUpdate{
Type: UpdateTypeComplete,
update := AgentEvent{
Type: EventTypeComplete,
SessionID: threadID,
OperationID: p.TurnID,
}
if !p.Success && p.Error != "" {
update.Type = UpdateTypeError
update.Type = EventTypeError
update.Error = p.Error
}
a.sendUpdate(update)
@@ -435,8 +435,8 @@ func (a *CodexAdapter) handleNotification(method string, params json.RawMessage)
a.logger.Warn("failed to parse turn diff updated", zap.Error(err))
return
}
a.sendUpdate(SessionUpdate{
Type: UpdateTypeMessageChunk,
a.sendUpdate(AgentEvent{
Type: EventTypeMessageChunk,
SessionID: threadID,
OperationID: p.TurnID,
Diff: p.Diff,
@@ -455,8 +455,8 @@ func (a *CodexAdapter) handleNotification(method string, params json.RawMessage)
Status: e.Status,
}
}
a.sendUpdate(SessionUpdate{
Type: UpdateTypePlan,
a.sendUpdate(AgentEvent{
Type: EventTypePlan,
SessionID: threadID,
OperationID: p.TurnID,
PlanEntries: entries,
@@ -468,8 +468,8 @@ func (a *CodexAdapter) handleNotification(method string, params json.RawMessage)
a.logger.Warn("failed to parse error notification", zap.Error(err))
return
}
a.sendUpdate(SessionUpdate{
Type: UpdateTypeError,
a.sendUpdate(AgentEvent{
Type: EventTypeError,
SessionID: threadID,
Error: p.Message,
})
@@ -487,8 +487,8 @@ func (a *CodexAdapter) handleNotification(method string, params json.RawMessage)
// Item types: "userMessage", "agentMessage", "commandExecution", "fileChange", "reasoning"
switch p.Item.Type {
case "commandExecution":
a.sendUpdate(SessionUpdate{
Type: UpdateTypeToolCall,
a.sendUpdate(AgentEvent{
Type: EventTypeToolCall,
SessionID: threadID,
OperationID: turnID,
ToolCallID: p.Item.ID,
@@ -509,8 +509,8 @@ func (a *CodexAdapter) handleNotification(method string, params json.RawMessage)
title += fmt.Sprintf(" (+%d more)", len(p.Item.Changes)-1)
}
}
a.sendUpdate(SessionUpdate{
Type: UpdateTypeToolCall,
a.sendUpdate(AgentEvent{
Type: EventTypeToolCall,
SessionID: threadID,
OperationID: turnID,
ToolCallID: p.Item.ID,
@@ -535,8 +535,8 @@ func (a *CodexAdapter) handleNotification(method string, params json.RawMessage)
if p.Item.Status == "failed" {
status = "error"
}
update := SessionUpdate{
Type: UpdateTypeToolUpdate,
update := AgentEvent{
Type: EventTypeToolUpdate,
SessionID: threadID,
OperationID: turnID,
ToolCallID: p.Item.ID,
@@ -567,8 +567,8 @@ func (a *CodexAdapter) handleNotification(method string, params json.RawMessage)
a.logger.Warn("failed to parse command output delta", zap.Error(err))
return
}
a.sendUpdate(SessionUpdate{
Type: UpdateTypeToolUpdate,
a.sendUpdate(AgentEvent{
Type: EventTypeToolUpdate,
SessionID: threadID,
OperationID: turnID,
ToolCallID: p.ItemID,
apps/backend/internal/agentctl/server/api/acp.go
-79
View File
@@ -319,82 +319,3 @@ func (s *Server) handlePermissionRespond(c *gin.Context) {
})
}
// PendingPermissionResponse represents a pending permission for the API
type PendingPermissionResponse struct {
PendingID string `json:"pending_id"`
SessionID string `json:"session_id"`
ToolCallID string `json:"tool_call_id"`
Title string `json:"title"`
Options []PermissionOptionJSON `json:"options"`
CreatedAt string `json:"created_at"`
}
// PermissionOptionJSON is the JSON representation of a permission option
type PermissionOptionJSON struct {
OptionID string `json:"option_id"`
Name string `json:"name"`
Kind string `json:"kind"`
}
func (s *Server) handleGetPendingPermissions(c *gin.Context) {
pending := s.procMgr.GetPendingPermissions()
result := make([]PendingPermissionResponse, 0, len(pending))
for _, p := range pending {
options := make([]PermissionOptionJSON, len(p.Request.Options))
for i, opt := range p.Request.Options {
options[i] = PermissionOptionJSON{
OptionID: opt.OptionID,
Name: opt.Name,
Kind: opt.Kind,
}
}
result = append(result, PendingPermissionResponse{
PendingID: p.ID,
SessionID: p.Request.SessionID,
ToolCallID: p.Request.ToolCallID,
Title: p.Request.Title,
Options: options,
CreatedAt: p.CreatedAt.Format(time.RFC3339),
})
}
c.JSON(http.StatusOK, result)
}
// handlePermissionStreamWS streams permission request notifications via WebSocket
func (s *Server) handlePermissionStreamWS(c *gin.Context) {
conn, err := s.upgrader.Upgrade(c.Writer, c.Request, nil)
if err != nil {
s.logger.Error("WebSocket upgrade failed", zap.Error(err))
return
}
defer func() {
if err := conn.Close(); err != nil {
s.logger.Debug("failed to close permission websocket", zap.Error(err))
}
}()
s.logger.Info("Permission stream WebSocket connected")
// Get the permission requests channel
permissionCh := s.procMgr.GetPermissionRequests()
// Stream permission notifications to WebSocket
for notification := range permissionCh {
s.logger.Info("streaming permission notification to backend",
zap.String("pending_id", notification.PendingID),
zap.String("title", notification.Title))
data, err := json.Marshal(notification)
if err != nil {
s.logger.Error("failed to marshal permission notification", zap.Error(err))
continue
}
if err := conn.WriteMessage(websocket.TextMessage, data); err != nil {
s.logger.Debug("WebSocket write error", zap.Error(err))
return
}
}
}
apps/backend/internal/agentctl/server/api/server.go
-2
View File
@@ -72,8 +72,6 @@ func (s *Server) setupRoutes() {
api.GET("/acp/stream", s.handleACPStreamWS)
// Permission request handling
api.GET("/acp/permissions", s.handleGetPendingPermissions)
api.GET("/acp/permissions/stream", s.handlePermissionStreamWS)
api.POST("/acp/permissions/respond", s.handlePermissionRespond)
// Workspace monitoring (git status, files)
apps/backend/internal/agentctl/server/process/manager.go
+25 -37
View File
@@ -15,6 +15,7 @@ import (
"github.com/kandev/kandev/internal/agentctl/server/adapter"
"github.com/kandev/kandev/internal/agentctl/server/config"
"github.com/kandev/kandev/internal/agentctl/server/shell"
"github.com/kandev/kandev/internal/agentctl/types/streams"
"github.com/kandev/kandev/internal/common/logger"
"github.com/kandev/kandev/pkg/agent"
"go.uber.org/zap"
@@ -80,11 +81,8 @@ type Manager struct {
// Protocol adapter for agent communication
adapter adapter.AgentAdapter
// Session update notifications (protocol-agnostic)
updatesCh chan adapter.SessionUpdate
// Permission request notifications (sent to backend)
permissionCh chan *PermissionNotification
// Agent event notifications (protocol-agnostic)
updatesCh chan adapter.AgentEvent
// Pending permission requests waiting for user response
pendingPermissions map[string]*PendingPermission
@@ -104,8 +102,7 @@ func NewManager(cfg *config.InstanceConfig, log *logger.Logger) *Manager {
cfg: cfg,
logger: log.WithFields(zap.String("component", "process-manager")),
workspaceTracker: NewWorkspaceTracker(cfg.WorkDir, log),
updatesCh: make(chan adapter.SessionUpdate, 100),
permissionCh: make(chan *PermissionNotification, 10),
updatesCh: make(chan adapter.AgentEvent, 100),
pendingPermissions: make(map[string]*PendingPermission),
}
m.status.Store(StatusStopped)
@@ -344,8 +341,8 @@ func (m *Manager) forwardUpdates() {
}
}
// GetUpdates returns the channel for session update notifications
func (m *Manager) GetUpdates() <-chan adapter.SessionUpdate {
// GetUpdates returns the channel for agent event notifications
func (m *Manager) GetUpdates() <-chan adapter.AgentEvent {
return m.updatesCh
}
@@ -586,38 +583,39 @@ func (m *Manager) autoApprovePermission(req *adapter.PermissionRequest) (*adapte
}, nil
}
// sendPermissionNotification sends a permission request notification to the backend
// sendPermissionNotification sends a permission request notification through the updates channel
func (m *Manager) sendPermissionNotification(pending *PendingPermission) {
notification := &PermissionNotification{
PendingID: pending.ID,
SessionID: pending.Request.SessionID,
ToolCallID: pending.Request.ToolCallID,
Title: pending.Request.Title,
Options: pending.Request.Options,
ActionType: pending.Request.ActionType,
ActionDetails: pending.Request.ActionDetails,
CreatedAt: pending.CreatedAt,
// Convert options to streams.PermissionOption (types.PermissionOption is an alias)
options := make([]streams.PermissionOption, len(pending.Request.Options))
for i, opt := range pending.Request.Options {
options[i] = streams.PermissionOption(opt)
}
m.logger.Info("sending permission notification to backend",
event := adapter.AgentEvent{
Type: adapter.EventTypePermissionRequest,
SessionID: pending.Request.SessionID,
ToolCallID: pending.Request.ToolCallID,
PendingID: pending.ID,
PermissionTitle: pending.Request.Title,
PermissionOptions: options,
ActionType: pending.Request.ActionType,
ActionDetails: pending.Request.ActionDetails,
}
m.logger.Info("sending permission notification via updates channel",
zap.String("pending_id", pending.ID),
zap.String("title", pending.Request.Title),
zap.String("action_type", pending.Request.ActionType))
select {
case m.permissionCh <- notification:
case m.updatesCh <- event:
// Sent successfully
default:
m.logger.Warn("permission channel full, dropping notification",
m.logger.Warn("updates channel full, dropping permission notification",
zap.String("pending_id", pending.ID))
}
}
// GetPermissionRequests returns the channel for permission request notifications
func (m *Manager) GetPermissionRequests() <-chan *PermissionNotification {
return m.permissionCh
}
// RespondToPermission responds to a pending permission request
func (m *Manager) RespondToPermission(pendingID string, optionID string, cancelled bool) error {
m.permissionMu.RLock()
@@ -645,17 +643,7 @@ func (m *Manager) RespondToPermission(pendingID string, optionID string, cancell
}
}
// GetPendingPermissions returns all pending permission requests
func (m *Manager) GetPendingPermissions() []*PendingPermission {
m.permissionMu.RLock()
defer m.permissionMu.RUnlock()
result := make([]*PendingPermission, 0, len(m.pendingPermissions))
for _, p := range m.pendingPermissions {
result = append(result, p)
}
return result
}
// Shell returns the embedded shell session, or nil if not available
func (m *Manager) Shell() *shell.Session {
apps/backend/internal/agentctl/types/streams/agent.go
+129
View File
@@ -0,0 +1,129 @@
package streams
// AgentEvent type constants define the types of events streamed from the agent.
const (
// EventTypeMessageChunk indicates streaming text content from the agent.
EventTypeMessageChunk = "message_chunk"
// EventTypeReasoning indicates chain-of-thought or thinking content.
EventTypeReasoning = "reasoning"
// EventTypeToolCall indicates a tool invocation has started.
EventTypeToolCall = "tool_call"
// EventTypeToolUpdate indicates a tool status update (running, completed, etc.).
EventTypeToolUpdate = "tool_update"
// EventTypePlan indicates agent plan/task list updates.
EventTypePlan = "plan"
// EventTypeComplete indicates the turn or session has completed.
EventTypeComplete = "complete"
// EventTypeError indicates an error occurred.
EventTypeError = "error"
// EventTypePermissionRequest indicates the agent is requesting permission.
EventTypePermissionRequest = "permission_request"
)
// AgentEvent is the message type streamed from the agent process.
// This represents protocol-agnostic events from the agent, normalized from
// various underlying protocols (ACP, Codex, Claude Code, etc.).
//
// Stream endpoint: ws://.../api/v1/agent/events
type AgentEvent struct {
// Type identifies the event type. Use EventType* constants:
// "message_chunk", "reasoning", "tool_call", "tool_update", "plan", "complete", "error"
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 contains streaming text content from the agent.
Text string `json:"text,omitempty"`
// --- Reasoning fields (for "reasoning" type) ---
// ReasoningText contains full reasoning/chain-of-thought content.
ReasoningText string `json:"reasoning_text,omitempty"`
// ReasoningSummary contains a summarized version of reasoning (if available).
ReasoningSummary string `json:"reasoning_summary,omitempty"`
// --- Tool call fields (for "tool_call" and "tool_update" types) ---
// ToolCallID uniquely identifies the tool invocation.
ToolCallID string `json:"tool_call_id,omitempty"`
// ToolName is the name of the tool being invoked.
ToolName string `json:"tool_name,omitempty"`
// ToolTitle is a human-readable title for the tool call.
ToolTitle string `json:"tool_title,omitempty"`
// ToolStatus indicates the current status: "started", "running", "completed", "error".
ToolStatus string `json:"tool_status,omitempty"`
// ToolArgs contains the arguments passed to the tool.
ToolArgs map[string]interface{} `json:"tool_args,omitempty"`
// ToolResult contains the result from the tool execution.
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 contains the agent's execution plan/task list.
PlanEntries []PlanEntry `json:"plan_entries,omitempty"`
// --- Error fields (for "error" type) ---
// Error contains error message when Type is "error".
Error string `json:"error,omitempty"`
// --- Permission request fields (for "permission_request" type) ---
// PendingID uniquely identifies this pending permission request.
PendingID string `json:"pending_id,omitempty"`
// PermissionTitle is a human-readable description of the action requiring permission.
PermissionTitle string `json:"permission_title,omitempty"`
// PermissionOptions contains the available permission choices.
PermissionOptions []PermissionOption `json:"permission_options,omitempty"`
// ActionType categorizes the action requiring approval.
// Use ActionType* constants: "command", "file_write", "file_read", "network", "mcp_tool", "other".
ActionType string `json:"action_type,omitempty"`
// ActionDetails contains structured details about the action.
ActionDetails map[string]interface{} `json:"action_details,omitempty"`
// --- Extension fields ---
// Data contains raw protocol-specific extensions.
Data map[string]interface{} `json:"data,omitempty"`
}
// PlanEntry represents an entry in the agent's execution plan.
type PlanEntry struct {
// Description is the content/description of the task.
Description string `json:"description,omitempty"`
// Status indicates task status: "pending", "in_progress", "completed", "failed".
Status string `json:"status,omitempty"`
// Priority indicates relative importance.
Priority string `json:"priority,omitempty"`
}
apps/backend/internal/agentctl/types/streams/doc.go
+51
View File
@@ -0,0 +1,51 @@
// Package streams defines the protocol message types for agentctl WebSocket streams.
//
// The agentctl client produces messages to the following channels/streams:
//
// # Agent Events Stream (/api/v1/agent/events)
//
// Streams real-time events from the agent process including message chunks,
// reasoning/thinking content, tool invocations, plan updates, and completion
// or error notifications. This stream is protocol-agnostic and works with
// any agent backend (ACP, Codex, Claude Code, etc.).
//
// Message type: AgentEvent
//
// Event types (use EventType* constants):
// - message_chunk: Streaming text content from the agent
// - reasoning: Chain-of-thought or thinking content
// - tool_call: A tool invocation has started
// - tool_update: Tool status update (running, completed, error)
// - plan: Agent plan/task list updates
// - complete: The turn or operation has completed
// - error: An error occurred
//
// # Permission Stream (/api/v1/acp/permissions/stream)
//
// Streams permission requests from the agent when it needs approval for
// actions like file writes, shell commands, or network access.
//
// Message type: PermissionNotification
//
// # Git Status Stream (/api/v1/workspace/git-status/stream)
//
// Streams git status updates when the workspace state changes.
//
// Message type: GitStatusUpdate
//
// # File Changes Stream (/api/v1/workspace/file-changes/stream)
//
// Streams file system change notifications when files are created,
// modified, deleted, or renamed in the workspace.
//
// Message type: FileChangeNotification
//
// # Shell Stream (/api/v1/shell/stream)
//
// Bidirectional WebSocket for interactive shell I/O.
//
// Message type: ShellMessage (input/output/ping/pong/exit)
//
// All streams use JSON-encoded messages over WebSocket connections.
package streams
apps/backend/internal/agentctl/types/streams/file.go
+130
View File
@@ -0,0 +1,130 @@
package streams
import "time"
// File change operation constants.
const (
// FileOpCreate indicates a file was created.
FileOpCreate = "create"
// FileOpWrite indicates a file was written/modified.
FileOpWrite = "write"
// FileOpRemove indicates a file was removed.
FileOpRemove = "remove"
// FileOpRename indicates a file was renamed.
FileOpRename = "rename"
// FileOpChmod indicates file permissions changed.
FileOpChmod = "chmod"
// FileOpRefresh indicates a refresh/rescan of the file.
FileOpRefresh = "refresh"
)
// FileChangeNotification is the message type streamed via the file changes stream.
// Represents a filesystem change notification.
//
// Stream endpoint: ws://.../api/v1/workspace/file-changes/stream
type FileChangeNotification struct {
// Timestamp is when the change was detected.
Timestamp time.Time `json:"timestamp"`
// Path is the file path relative to workspace root.
Path string `json:"path"`
// Operation indicates the type of change. Use FileOp* constants:
// "create", "write", "remove", "rename", "chmod", "refresh".
Operation string `json:"operation"`
}
// FileListUpdate represents a file listing update.
// Used for initial sync or full refresh of workspace files.
type FileListUpdate struct {
// Timestamp is when the listing was captured.
Timestamp time.Time `json:"timestamp"`
// Files contains the list of files in the workspace.
Files []FileEntry `json:"files"`
}
// FileEntry represents a file in the workspace.
type FileEntry struct {
// Path is the file path relative to workspace root.
Path string `json:"path"`
// IsDir indicates if this is a directory.
IsDir bool `json:"is_dir"`
// Size is the file size in bytes (for files only).
Size int64 `json:"size,omitempty"`
}
// FileTreeNode represents a node in the file tree.
type FileTreeNode struct {
// Name is the file or directory name.
Name string `json:"name"`
// Path is the full path relative to workspace root.
Path string `json:"path"`
// IsDir indicates if this is a directory.
IsDir bool `json:"is_dir"`
// Size is the file size in bytes (for files only).
Size int64 `json:"size,omitempty"`
// Children contains child nodes (for directories only).
Children []*FileTreeNode `json:"children,omitempty"`
}
// FileTreeRequest represents a request for file tree.
//
// HTTP endpoint: GET /api/v1/workspace/tree
type FileTreeRequest struct {
// Path is the path to get tree for (relative to workspace root).
Path string `json:"path"`
// Depth is the depth to traverse (0 = unlimited, 1 = immediate children only).
Depth int `json:"depth"`
}
// FileTreeResponse represents a response with file tree.
type FileTreeResponse struct {
// RequestID identifies the request.
RequestID string `json:"request_id"`
// Root is the root node of the file tree.
Root *FileTreeNode `json:"root"`
// Error contains error message if the request failed.
Error string `json:"error,omitempty"`
}
// FileContentRequest represents a request for file content.
//
// HTTP endpoint: GET /api/v1/workspace/file/content
type FileContentRequest struct {
// Path is the file path (relative to workspace root).
Path string `json:"path"`
}
// FileContentResponse represents a response with file content.
type FileContentResponse struct {
// RequestID identifies the request.
RequestID string `json:"request_id"`
// Path is the file path.
Path string `json:"path"`
// Content is the file content.
Content string `json:"content"`
// Size is the file size in bytes.
Size int64 `json:"size"`
// Error contains error message if the request failed.
Error string `json:"error,omitempty"`
}
apps/backend/internal/agentctl/types/streams/git.go
+64
View File
@@ -0,0 +1,64 @@
package streams
import "time"
// GitStatusUpdate is the message type streamed via the git status stream.
// Represents the current git state of the workspace.
//
// Stream endpoint: ws://.../api/v1/workspace/git-status/stream
type GitStatusUpdate struct {
// Timestamp is when this status was captured.
Timestamp time.Time `json:"timestamp"`
// Modified contains paths of modified files.
Modified []string `json:"modified"`
// Added contains paths of added/staged files.
Added []string `json:"added"`
// Deleted contains paths of deleted files.
Deleted []string `json:"deleted"`
// Untracked contains paths of untracked files.
Untracked []string `json:"untracked"`
// Renamed contains paths of renamed files.
Renamed []string `json:"renamed"`
// Ahead is the number of commits ahead of the remote branch.
Ahead int `json:"ahead"`
// Behind is the number of commits behind the remote branch.
Behind int `json:"behind"`
// Branch is the current local branch name.
Branch string `json:"branch"`
// RemoteBranch is the tracked remote branch (e.g., "origin/main").
RemoteBranch string `json:"remote_branch,omitempty"`
// Files contains detailed information about each changed file.
Files map[string]FileInfo `json:"files,omitempty"`
}
// FileInfo represents detailed information about a file's git status.
type FileInfo struct {
// Path is the file path relative to workspace root.
Path string `json:"path"`
// Status indicates the file status: "modified", "added", "deleted", "untracked", "renamed".
Status string `json:"status"`
// Additions is the number of added lines.
Additions int `json:"additions,omitempty"`
// Deletions is the number of deleted lines.
Deletions int `json:"deletions,omitempty"`
// OldPath is the original path for renamed files.
OldPath string `json:"old_path,omitempty"`
// Diff contains the unified diff content for this file.
Diff string `json:"diff,omitempty"`
}
apps/backend/internal/agentctl/types/streams/permission.go
+103
View File
@@ -0,0 +1,103 @@
package streams
import "time"
// Permission action types categorize the kind of action requiring approval.
const (
// ActionTypeCommand indicates shell command execution.
ActionTypeCommand = "command"
// ActionTypeFileWrite indicates file modification or creation.
ActionTypeFileWrite = "file_write"
// ActionTypeFileRead indicates file read (for sensitive files).
ActionTypeFileRead = "file_read"
// ActionTypeNetwork indicates network access.
ActionTypeNetwork = "network"
// ActionTypeMCPTool indicates MCP tool invocation.
ActionTypeMCPTool = "mcp_tool"
// ActionTypeOther indicates other/unknown action type.
ActionTypeOther = "other"
)
// PermissionNotification is the message type streamed via the permissions stream.
// Received when the agent requests permission for an action.
//
// Stream endpoint: ws://.../api/v1/acp/permissions/stream
type PermissionNotification struct {
// PendingID uniquely identifies this pending permission request.
PendingID string `json:"pending_id"`
// SessionID is the session making the request.
SessionID string `json:"session_id"`
// ToolCallID is the tool call that triggered this permission request.
ToolCallID string `json:"tool_call_id"`
// Title is a human-readable description of the action.
Title string `json:"title"`
// Options contains the available permission choices.
Options []PermissionOption `json:"options"`
// ActionType categorizes the action requiring approval.
// Use ActionType* constants: "command", "file_write", "file_read", "network", "mcp_tool", "other".
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"`
// CreatedAt is when the permission request was created.
CreatedAt time.Time `json:"created_at"`
}
// PermissionOption represents a permission choice presented to the user.
type PermissionOption struct {
// OptionID uniquely identifies this option.
OptionID string `json:"option_id"`
// Name is a human-readable name for the option.
Name string `json:"name"`
// Kind indicates the type of permission: "allow_once", "allow_always",
// "reject_once", "reject_always".
Kind string `json:"kind"`
// Metadata contains protocol-specific option data.
// For Codex: {"for_session": true} for session-wide approvals.
Metadata map[string]interface{} `json:"metadata,omitempty"`
}
// PermissionRespondRequest is sent to respond to a permission request.
//
// HTTP endpoint: POST /api/v1/acp/permissions/respond
type PermissionRespondRequest struct {
// PendingID is the ID of the permission request to respond to.
PendingID string `json:"pending_id"`
// OptionID is the selected option ID.
OptionID string `json:"option_id,omitempty"`
// Cancelled indicates if the request was cancelled.
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"`
}
// PermissionRespondResponse is the response from the permission respond endpoint.
type PermissionRespondResponse struct {
// Success indicates if the response was accepted.
Success bool `json:"success"`
// Error contains error message if Success is false.
Error string `json:"error,omitempty"`
}
apps/backend/internal/agentctl/types/streams/shell.go
+64
View File
@@ -0,0 +1,64 @@
package streams
// Shell message type constants.
const (
// ShellMsgTypeInput indicates user input to the shell.
ShellMsgTypeInput = "input"
// ShellMsgTypeOutput indicates output from the shell.
ShellMsgTypeOutput = "output"
// ShellMsgTypePing is a ping message for keepalive.
ShellMsgTypePing = "ping"
// ShellMsgTypePong is a pong response to ping.
ShellMsgTypePong = "pong"
// ShellMsgTypeExit indicates the shell has exited.
ShellMsgTypeExit = "exit"
)
// ShellMessage is the message type for bidirectional shell stream.
// Used for interactive shell I/O over WebSocket.
//
// Stream endpoint: ws://.../api/v1/shell/stream
type ShellMessage struct {
// Type indicates the message type. Use ShellMsgType* constants:
// "input", "output", "ping", "pong", "exit".
Type string `json:"type"`
// Data contains the shell input or output data.
Data string `json:"data,omitempty"`
// Code is the exit code (for "exit" type only).
Code int `json:"code,omitempty"`
}
// ShellStatusResponse is the response from the shell status endpoint.
//
// HTTP endpoint: GET /api/v1/shell/status
type ShellStatusResponse struct {
// Running indicates if the shell is currently running.
Running bool `json:"running"`
// Pid is the process ID of the shell.
Pid int `json:"pid"`
// Shell is the shell type (e.g., "/bin/bash").
Shell string `json:"shell"`
// Cwd is the current working directory.
Cwd string `json:"cwd"`
// StartedAt is when the shell was started (ISO 8601 format).
StartedAt string `json:"started_at"`
}
// ShellBufferResponse is the response from the shell buffer endpoint.
//
// HTTP endpoint: GET /api/v1/shell/buffer
type ShellBufferResponse struct {
// Data contains the buffered shell output.
Data string `json:"data"`
}
apps/backend/internal/agentctl/types/types.go
+83 -110
View File
@@ -1,23 +1,85 @@
// Package types provides shared types for agentctl packages.
// This breaks import cycles between adapter and acp packages.
//
// For stream protocol message types, see the streams subpackage.
package types
import (
"context"
"time"
"github.com/kandev/kandev/internal/agentctl/types/streams"
)
// Permission action types - categorize the kind of action requiring approval
// Re-export stream types for convenience.
// New code may import from the streams package directly.
type (
// Agent event stream types
AgentEvent = streams.AgentEvent
PlanEntry = streams.PlanEntry
// Permission stream types
PermissionNotification = streams.PermissionNotification
PermissionOption = streams.PermissionOption
PermissionRespondRequest = streams.PermissionRespondRequest
PermissionRespondResponse = streams.PermissionRespondResponse
// Git stream types
GitStatusUpdate = streams.GitStatusUpdate
FileInfo = streams.FileInfo
// File stream types
FileChangeNotification = streams.FileChangeNotification
FileListUpdate = streams.FileListUpdate
FileEntry = streams.FileEntry
FileTreeNode = streams.FileTreeNode
FileTreeRequest = streams.FileTreeRequest
FileTreeResponse = streams.FileTreeResponse
FileContentRequest = streams.FileContentRequest
FileContentResponse = streams.FileContentResponse
// Shell stream types
ShellMessage = streams.ShellMessage
ShellStatusResponse = streams.ShellStatusResponse
ShellBufferResponse = streams.ShellBufferResponse
)
// Re-export stream constants for convenience.
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
// Agent event types (preferred)
EventTypeMessageChunk = streams.EventTypeMessageChunk
EventTypeReasoning = streams.EventTypeReasoning
EventTypeToolCall = streams.EventTypeToolCall
EventTypeToolUpdate = streams.EventTypeToolUpdate
EventTypePlan = streams.EventTypePlan
EventTypeComplete = streams.EventTypeComplete
EventTypeError = streams.EventTypeError
// Permission action types
ActionTypeCommand = streams.ActionTypeCommand
ActionTypeFileWrite = streams.ActionTypeFileWrite
ActionTypeFileRead = streams.ActionTypeFileRead
ActionTypeNetwork = streams.ActionTypeNetwork
ActionTypeMCPTool = streams.ActionTypeMCPTool
ActionTypeOther = streams.ActionTypeOther
// File operation types
FileOpCreate = streams.FileOpCreate
FileOpWrite = streams.FileOpWrite
FileOpRemove = streams.FileOpRemove
FileOpRename = streams.FileOpRename
FileOpChmod = streams.FileOpChmod
FileOpRefresh = streams.FileOpRefresh
// Shell message types
ShellMsgTypeInput = streams.ShellMsgTypeInput
ShellMsgTypeOutput = streams.ShellMsgTypeOutput
ShellMsgTypePing = streams.ShellMsgTypePing
ShellMsgTypePong = streams.ShellMsgTypePong
ShellMsgTypeExit = streams.ShellMsgTypeExit
)
// PermissionRequest represents a permission request from the agent
// PermissionRequest represents a permission request from the agent.
// This is used internally by adapters and is not sent over streams directly.
type PermissionRequest struct {
SessionID string `json:"session_id"`
ToolCallID string `json:"tool_call_id"`
@@ -35,18 +97,8 @@ type PermissionRequest struct {
ActionDetails map[string]interface{} `json:"action_details,omitempty"`
}
// PermissionOption represents a permission choice
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
// PermissionResponse is the user's response to a permission request.
// This is used internally by adapters.
type PermissionResponse struct {
OptionID string `json:"option_id,omitempty"`
Cancelled bool `json:"cancelled,omitempty"`
@@ -56,96 +108,17 @@ type PermissionResponse struct {
ResponseMetadata map[string]interface{} `json:"response_metadata,omitempty"`
}
// PermissionHandler is called when the agent requests permission for an action
// PermissionHandler is called when the agent requests permission for an action.
type PermissionHandler func(ctx context.Context, req *PermissionRequest) (*PermissionResponse, error)
// GitStatusUpdate represents a git status update
type GitStatusUpdate struct {
Timestamp time.Time `json:"timestamp"`
Modified []string `json:"modified"`
Added []string `json:"added"`
Deleted []string `json:"deleted"`
Untracked []string `json:"untracked"`
Renamed []string `json:"renamed"`
Ahead int `json:"ahead"`
Behind int `json:"behind"`
Branch string `json:"branch"`
RemoteBranch string `json:"remote_branch,omitempty"`
Files map[string]FileInfo `json:"files,omitempty"`
}
// Subscriber types for internal use.
type (
// GitStatusSubscriber is a channel that receives git status updates.
GitStatusSubscriber chan GitStatusUpdate
// FileInfo represents information about a file
type FileInfo struct {
Path string `json:"path"`
Status string `json:"status"` // modified, added, deleted, untracked, renamed
Additions int `json:"additions,omitempty"`
Deletions int `json:"deletions,omitempty"`
OldPath string `json:"old_path,omitempty"` // For renamed files
Diff string `json:"diff,omitempty"`
}
// FileListUpdate represents a file listing update
type FileListUpdate struct {
Timestamp time.Time `json:"timestamp"`
Files []FileEntry `json:"files"`
}
// FileEntry represents a file in the workspace
type FileEntry struct {
Path string `json:"path"`
IsDir bool `json:"is_dir"`
Size int64 `json:"size,omitempty"`
}
// FileTreeNode represents a node in the file tree
type FileTreeNode struct {
Name string `json:"name"`
Path string `json:"path"`
IsDir bool `json:"is_dir"`
Size int64 `json:"size,omitempty"`
Children []*FileTreeNode `json:"children,omitempty"`
}
// FileTreeRequest represents a request for file tree
type FileTreeRequest struct {
Path string `json:"path"` // Path to get tree for (relative to workspace root)
Depth int `json:"depth"` // Depth to traverse (0 = unlimited, 1 = immediate children only)
}
// FileTreeResponse represents a response with file tree
type FileTreeResponse struct {
RequestID string `json:"request_id"`
Root *FileTreeNode `json:"root"`
Error string `json:"error,omitempty"`
}
// FileContentRequest represents a request for file content
type FileContentRequest struct {
Path string `json:"path"` // Path to file (relative to workspace root)
}
// FileContentResponse represents a response with file content
type FileContentResponse struct {
RequestID string `json:"request_id"`
Path string `json:"path"`
Content string `json:"content"`
Size int64 `json:"size"`
Error string `json:"error,omitempty"`
}
// FileChangeNotification represents a filesystem change notification
type FileChangeNotification struct {
Timestamp time.Time `json:"timestamp"`
Path string `json:"path"`
Operation string `json:"operation"` // create, write, remove, rename, chmod, refresh
}
// GitStatusSubscriber is a channel that receives git status updates
type GitStatusSubscriber chan GitStatusUpdate
// FilesSubscriber is a channel that receives file listing updates
type FilesSubscriber chan FileListUpdate
// FileChangeSubscriber is a channel that receives file change notifications
type FileChangeSubscriber chan FileChangeNotification
// FilesSubscriber is a channel that receives file listing updates.
FilesSubscriber chan FileListUpdate
// FileChangeSubscriber is a channel that receives file change notifications.
FileChangeSubscriber chan FileChangeNotification
)
apps/backend/internal/events/types.go
+4 -3
View File
@@ -95,9 +95,10 @@ const (
// Event types for agent prompts
const (
PromptComplete = "prompt.complete" // Agent finished responding to a prompt
ToolCallStarted = "tool_call.started" // Agent started a tool call
ToolCallComplete = "tool_call.complete" // Agent finished a tool call
PromptComplete = "prompt.complete" // Agent finished responding to a prompt
ToolCallStarted = "tool_call.started" // Agent started a tool call
ToolCallComplete = "tool_call.complete" // Agent finished a tool call
PermissionRequestReceived = "permission_request.received" // Agent requested permission
)
// Event types for workspace/git status
apps/backend/internal/gateway/websocket/client.go
+4 -28
View File
@@ -169,11 +169,8 @@ func (c *Client) handleSubscribe(msg *ws.Message) {
})
c.sendMessage(resp)
// Send historical logs if available
// Send historical logs if available (now includes pending permission request messages)
c.sendHistoricalLogs(req.TaskID)
// Send pending permissions if available
c.sendPendingPermissions(req.TaskID)
}
type UserSubscribeRequest struct {
@@ -275,30 +272,9 @@ func (c *Client) sendHistoricalLogs(taskID string) {
}
}
// sendPendingPermissions sends any pending permission requests to the client
func (c *Client) sendPendingPermissions(taskID string) {
ctx := context.Background()
permissions, err := c.hub.GetPendingPermissions(ctx, taskID)
if err != nil {
c.logger.Error("Failed to get pending permissions",
zap.String("task_id", taskID),
zap.Error(err))
return
}
if len(permissions) == 0 {
return
}
c.logger.Debug("Sending pending permissions",
zap.String("task_id", taskID),
zap.Int("count", len(permissions)))
// Send each pending permission as a notification
for _, permission := range permissions {
c.sendMessage(permission)
}
}
// sendPendingPermissions is deprecated. Pending permissions are now stored as messages
// and sent via sendHistoricalLogs. This function is kept for reference but is unused.
// func (c *Client) sendPendingPermissions(taskID string) {}
// handleUnsubscribe handles task.unsubscribe action
func (c *Client) handleUnsubscribe(msg *ws.Message) {
apps/backend/internal/gateway/websocket/hub.go
+11 -15
View File
@@ -14,8 +14,9 @@ import (
// HistoricalLogsProvider is a function that retrieves historical logs for a task
type HistoricalLogsProvider func(ctx context.Context, taskID string) ([]*ws.Message, error)
// PendingPermissionsProvider is a function that retrieves pending permissions for a task
type PendingPermissionsProvider func(ctx context.Context, taskID string) ([]*ws.Message, error)
// PendingPermissionsProvider is deprecated. Pending permissions are now stored as messages
// and retrieved via the HistoricalLogsProvider.
// type PendingPermissionsProvider func(ctx context.Context, taskID string) ([]*ws.Message, error)
// Hub manages all WebSocket client connections
type Hub struct {
@@ -41,8 +42,8 @@ type Hub struct {
// Optional provider for historical logs on subscription
historicalLogsProvider HistoricalLogsProvider
// Optional provider for pending permissions on subscription
pendingPermissionsProvider PendingPermissionsProvider
// NOTE: pendingPermissionsProvider is deprecated. Pending permissions are now stored
// as messages and retrieved via the historicalLogsProvider.
mu sync.RWMutex
logger *logger.Logger
@@ -392,15 +393,10 @@ func (h *Hub) GetHistoricalLogs(ctx context.Context, taskID string) ([]*ws.Messa
return h.historicalLogsProvider(ctx, taskID)
}
// SetPendingPermissionsProvider sets the provider for pending permissions on subscription
func (h *Hub) SetPendingPermissionsProvider(provider PendingPermissionsProvider) {
h.pendingPermissionsProvider = provider
}
// SetPendingPermissionsProvider is deprecated. Pending permissions are now stored as messages
// and retrieved via the HistoricalLogsProvider. This method is a no-op for backwards compatibility.
// func (h *Hub) SetPendingPermissionsProvider(provider PendingPermissionsProvider) {}
// GetPendingPermissions retrieves pending permissions for a task if a provider is set
func (h *Hub) GetPendingPermissions(ctx context.Context, taskID string) ([]*ws.Message, error) {
if h.pendingPermissionsProvider == nil {
return nil, nil
}
return h.pendingPermissionsProvider(ctx, taskID)
}
// GetPendingPermissions is deprecated. Pending permissions are now stored as messages
// and retrieved via the HistoricalLogsProvider. This method always returns nil.
// func (h *Hub) GetPendingPermissions(ctx context.Context, taskID string) ([]*ws.Message, error) {}
apps/backend/internal/orchestrator/service.go
+77 -11
View File
@@ -57,6 +57,8 @@ type MessageCreator interface {
CreateToolCallMessage(ctx context.Context, taskID, toolCallID, title, status, agentSessionID string, args map[string]interface{}) error
UpdateToolCallMessage(ctx context.Context, taskID, toolCallID, status, result, agentSessionID string) error
CreateSessionMessage(ctx context.Context, taskID, content, agentSessionID, messageType string, metadata map[string]interface{}, requestsInput bool) error
CreatePermissionRequestMessage(ctx context.Context, taskID, sessionID, pendingID, toolCallID, title string, options []map[string]interface{}, actionType string, actionDetails map[string]interface{}) (string, error)
UpdatePermissionMessage(ctx context.Context, sessionID, pendingID, status string) error
}
// Service is the main orchestrator service
@@ -152,16 +154,17 @@ func NewService(
// Create the watcher with event handlers that wire everything together
handlers := watcher.EventHandlers{
OnTaskStateChanged: s.handleTaskStateChanged,
OnAgentReady: s.handleAgentReady,
OnAgentCompleted: s.handleAgentCompleted,
OnAgentFailed: s.handleAgentFailed,
OnACPMessage: s.handleACPMessage,
OnACPSessionCreated: s.handleACPSessionCreated,
OnPromptComplete: s.handlePromptComplete,
OnToolCallStarted: s.handleToolCallStarted,
OnToolCallComplete: s.handleToolCallComplete,
OnGitStatusUpdated: s.handleGitStatusUpdated,
OnTaskStateChanged: s.handleTaskStateChanged,
OnAgentReady: s.handleAgentReady,
OnAgentCompleted: s.handleAgentCompleted,
OnAgentFailed: s.handleAgentFailed,
OnACPMessage: s.handleACPMessage,
OnACPSessionCreated: s.handleACPSessionCreated,
OnPromptComplete: s.handlePromptComplete,
OnToolCallStarted: s.handleToolCallStarted,
OnToolCallComplete: s.handleToolCallComplete,
OnPermissionRequest: s.handlePermissionRequest,
OnGitStatusUpdated: s.handleGitStatusUpdated,
}
s.watcher = watcher.NewWatcher(eventBus, handlers, log)
@@ -554,7 +557,31 @@ func (s *Service) RespondToPermission(ctx context.Context, sessionID, pendingID,
zap.String("pending_id", pendingID),
zap.String("option_id", optionID),
zap.Bool("cancelled", cancelled))
return s.executor.RespondToPermission(ctx, sessionID, pendingID, optionID, cancelled)
// Respond to the permission via agentctl
if err := s.executor.RespondToPermission(ctx, sessionID, pendingID, optionID, cancelled); err != nil {
return err
}
// Determine status based on response
status := "approved"
if cancelled {
status = "rejected"
}
// Update the permission message with the new status
if s.messageCreator != nil {
if err := s.messageCreator.UpdatePermissionMessage(ctx, sessionID, pendingID, status); err != nil {
s.logger.Warn("failed to update permission message status",
zap.String("session_id", sessionID),
zap.String("pending_id", pendingID),
zap.String("status", status),
zap.Error(err))
// Don't fail the whole operation if message update fails
}
}
return nil
}
// CompleteTask explicitly completes a task and stops its agent
@@ -1180,3 +1207,42 @@ func (s *Service) handleToolCallComplete(ctx context.Context, data watcher.ToolC
s.updateTaskSessionState(ctx, data.TaskID, data.TaskSessionID, models.TaskSessionStateRunning, "", false)
}
}
// handlePermissionRequest handles permission request events and saves as message
func (s *Service) handlePermissionRequest(ctx context.Context, data watcher.PermissionRequestData) {
s.logger.Info("handling permission request",
zap.String("task_id", data.TaskID),
zap.String("pending_id", data.PendingID),
zap.String("title", data.Title))
if data.TaskSessionID == "" {
s.logger.Warn("missing session_id for permission_request",
zap.String("task_id", data.TaskID),
zap.String("pending_id", data.PendingID))
return
}
if s.messageCreator != nil {
_, err := s.messageCreator.CreatePermissionRequestMessage(
ctx,
data.TaskID,
data.TaskSessionID,
data.PendingID,
data.ToolCallID,
data.Title,
data.Options,
data.ActionType,
data.ActionDetails,
)
if err != nil {
s.logger.Error("failed to create permission request message",
zap.String("task_id", data.TaskID),
zap.String("pending_id", data.PendingID),
zap.Error(err))
} else {
s.logger.Info("created permission request message",
zap.String("task_id", data.TaskID),
zap.String("pending_id", data.PendingID))
}
}
}
apps/backend/internal/orchestrator/watcher/watcher.go
+62
View File
@@ -69,6 +69,19 @@ type ToolCallCompleteData struct {
Result string `json:"result,omitempty"`
}
// PermissionRequestData contains data from permission_request events
type PermissionRequestData struct {
TaskID string `json:"task_id"`
TaskSessionID string `json:"session_id"`
AgentID string `json:"agent_id"`
PendingID string `json:"pending_id"`
ToolCallID string `json:"tool_call_id"`
Title string `json:"title"`
Options []map[string]interface{} `json:"options"`
ActionType string `json:"action_type"`
ActionDetails map[string]interface{} `json:"action_details"`
}
// GitStatusData contains data from git status events
type GitStatusData struct {
TaskID string `json:"task_id"`
@@ -113,6 +126,9 @@ type EventHandlers struct {
OnToolCallStarted func(ctx context.Context, data ToolCallData)
OnToolCallComplete func(ctx context.Context, data ToolCallCompleteData)
// Permission request events
OnPermissionRequest func(ctx context.Context, data PermissionRequestData)
// Git status events
OnGitStatusUpdated func(ctx context.Context, data GitStatusData)
}
@@ -187,6 +203,12 @@ func (w *Watcher) Start(ctx context.Context) error {
return err
}
// Subscribe to permission request events
if err := w.subscribeToPermissionRequestEvents(); err != nil {
w.unsubscribeAll()
return err
}
// Subscribe to git status events
if err := w.subscribeToGitStatusEvents(); err != nil {
w.unsubscribeAll()
@@ -547,6 +569,46 @@ func (w *Watcher) createToolCallCompleteHandler() bus.EventHandler {
}
}
// subscribeToPermissionRequestEvents subscribes to permission request events
func (w *Watcher) subscribeToPermissionRequestEvents() error {
if w.handlers.OnPermissionRequest == nil {
return nil
}
subject := events.PermissionRequestReceived + ".*"
sub, err := w.eventBus.Subscribe(subject, w.createPermissionRequestHandler())
if err != nil {
w.logger.Error("Failed to subscribe to permission request events",
zap.String("subject", subject),
zap.Error(err))
return err
}
w.subscriptions = append(w.subscriptions, sub)
return nil
}
// createPermissionRequestHandler creates a bus.EventHandler for permission request events
func (w *Watcher) createPermissionRequestHandler() bus.EventHandler {
return func(ctx context.Context, event *bus.Event) error {
var data PermissionRequestData
if err := w.parseEventData(event.Data, &data); err != nil {
w.logger.Error("Failed to parse permission request event data",
zap.String("event_type", event.Type),
zap.String("event_id", event.ID),
zap.Error(err))
return nil
}
w.logger.Debug("Handling permission request event",
zap.String("task_id", data.TaskID),
zap.String("pending_id", data.PendingID),
zap.String("title", data.Title))
w.handlers.OnPermissionRequest(ctx, data)
return nil
}
}
// subscribeToGitStatusEvents subscribes to git status events
func (w *Watcher) subscribeToGitStatusEvents() error {
if w.handlers.OnGitStatusUpdated == nil {
apps/backend/internal/task/models/models.go
+2
View File
@@ -96,6 +96,8 @@ const (
MessageTypeError MessageType = "error"
// MessageTypeStatus is for status changes: started, completed, failed
MessageTypeStatus MessageType = "status"
// MessageTypePermissionRequest is for agent permission requests
MessageTypePermissionRequest MessageType = "permission_request"
)
// Message represents a message in a task session
apps/backend/internal/task/repository/interface.go
+1
View File
@@ -62,6 +62,7 @@ type Repository interface {
CreateMessage(ctx context.Context, message *models.Message) error
GetMessage(ctx context.Context, id string) (*models.Message, error)
GetMessageByToolCallID(ctx context.Context, sessionID, toolCallID string) (*models.Message, error)
GetMessageByPendingID(ctx context.Context, sessionID, pendingID string) (*models.Message, error)
UpdateMessage(ctx context.Context, message *models.Message) error
ListMessages(ctx context.Context, sessionID string) ([]*models.Message, error)
ListMessagesPaginated(ctx context.Context, sessionID string, opts ListMessagesOptions) ([]*models.Message, bool, error)
apps/backend/internal/task/repository/sqlite.go
+24 -2
View File
@@ -1546,7 +1546,7 @@ func (r *SQLiteRepository) DeleteMessage(ctx context.Context, id string) error {
return nil
}
// GetMessageByToolCallID retrieves a message by session ID and tool_call_id in metadata
// GetMessageByToolCallID retrieves a tool_call message by session ID and tool_call_id in metadata
func (r *SQLiteRepository) GetMessageByToolCallID(ctx context.Context, sessionID, toolCallID string) (*models.Message, error) {
message := &models.Message{}
var requestsInput int
@@ -1554,7 +1554,7 @@ func (r *SQLiteRepository) GetMessageByToolCallID(ctx context.Context, sessionID
var metadataJSON string
err := r.db.QueryRowContext(ctx, `
SELECT id, task_session_id, task_id, author_type, author_id, content, requests_input, type, metadata, created_at
FROM task_session_messages WHERE task_session_id = ? AND json_extract(metadata, '$.tool_call_id') = ?
FROM task_session_messages WHERE task_session_id = ? AND type = 'tool_call' AND json_extract(metadata, '$.tool_call_id') = ?
`, sessionID, toolCallID).Scan(&message.ID, &message.TaskSessionID, &message.TaskID, &message.AuthorType, &message.AuthorID,
&message.Content, &requestsInput, &messageType, &metadataJSON, &message.CreatedAt)
if err != nil {
@@ -1568,6 +1568,28 @@ func (r *SQLiteRepository) GetMessageByToolCallID(ctx context.Context, sessionID
return message, nil
}
// GetMessageByPendingID retrieves a message by session ID and pending_id in metadata
func (r *SQLiteRepository) GetMessageByPendingID(ctx context.Context, sessionID, pendingID string) (*models.Message, error) {
message := &models.Message{}
var requestsInput int
var messageType string
var metadataJSON string
err := r.db.QueryRowContext(ctx, `
SELECT id, task_session_id, task_id, author_type, author_id, content, requests_input, type, metadata, created_at
FROM task_session_messages WHERE task_session_id = ? AND json_extract(metadata, '$.pending_id') = ?
`, sessionID, pendingID).Scan(&message.ID, &message.TaskSessionID, &message.TaskID, &message.AuthorType, &message.AuthorID,
&message.Content, &requestsInput, &messageType, &metadataJSON, &message.CreatedAt)
if err != nil {
return nil, err
}
message.RequestsInput = requestsInput == 1
message.Type = models.MessageType(messageType)
if metadataJSON != "" {
_ = json.Unmarshal([]byte(metadataJSON), &message.Metadata)
}
return message, nil
}
// UpdateMessage updates an existing message
func (r *SQLiteRepository) UpdateMessage(ctx context.Context, message *models.Message) error {
metadataJSON, err := json.Marshal(message.Metadata)
apps/backend/internal/task/service/service.go
+63
View File
@@ -1611,6 +1611,69 @@ func (s *Service) UpdateToolCallMessage(ctx context.Context, sessionID, toolCall
return nil
}
// UpdatePermissionMessage updates a permission request message's status.
// It includes retry logic to handle race conditions.
func (s *Service) UpdatePermissionMessage(ctx context.Context, sessionID, pendingID, status string) error {
const maxRetries = 5
const retryDelay = 100 * time.Millisecond
var message *models.Message
var err error
// Retry loop to handle race condition
for attempt := 0; attempt < maxRetries; attempt++ {
message, err = s.repo.GetMessageByPendingID(ctx, sessionID, pendingID)
if err == nil {
break
}
if ctx.Err() != nil {
return ctx.Err()
}
if attempt < maxRetries-1 {
s.logger.Debug("permission message not found, retrying",
zap.String("session_id", sessionID),
zap.String("pending_id", pendingID),
zap.Int("attempt", attempt+1),
zap.Int("max_retries", maxRetries))
time.Sleep(retryDelay)
}
}
if err != nil {
s.logger.Warn("permission message not found for update after retries",
zap.String("session_id", sessionID),
zap.String("pending_id", pendingID),
zap.Int("retries", maxRetries),
zap.Error(err))
return err
}
if message.Metadata == nil {
message.Metadata = make(map[string]interface{})
}
message.Metadata["status"] = status
if err := s.repo.UpdateMessage(ctx, message); err != nil {
s.logger.Error("failed to update permission message",
zap.String("message_id", message.ID),
zap.String("pending_id", pendingID),
zap.Error(err))
return err
}
// Publish message.updated event
s.publishMessageEvent(ctx, events.MessageUpdated, message)
s.logger.Info("permission message updated",
zap.String("message_id", message.ID),
zap.String("pending_id", pendingID),
zap.String("status", status))
return nil
}
// publishMessageEvent publishes message events to the event bus
func (s *Service) publishMessageEvent(ctx context.Context, eventType string, message *models.Message) {
if s.eventBus == nil {
apps/web/app/task/[id]/page-client.tsx
-4
View File
@@ -5,7 +5,6 @@ import '@git-diff-view/react/styles/diff-view.css';
import { TooltipProvider } from '@kandev/ui/tooltip';
import { TaskTopBar } from '@/components/task/task-top-bar';
import { TaskLayout } from '@/components/task/task-layout';
import { CommandApprovalDialog } from '@/components/task/command-approval-dialog';
import { DebugOverlay } from '@/components/debug-overlay';
import type { Repository, Task } from '@/lib/types/http';
import { DEBUG_UI } from '@/lib/config';
@@ -174,9 +173,6 @@ export default function TaskPage({
workspaceId={task?.workspace_id ?? null}
boardId={task?.board_id ?? null}
/>
{/* Fallback dialog for permissions without matching tool call (e.g., workspace indexing) */}
{task?.id && <CommandApprovalDialog taskId={task.id} standaloneOnly />}
</div>
</TooltipProvider>
);
apps/web/components/task/chat/message-renderer.tsx
+15 -3
View File
@@ -3,6 +3,7 @@
import type { ReactElement } from 'react';
import type { Message } from '@/lib/types/http';
import { ChatMessage } from '@/components/task/chat/messages/chat-message';
import { PermissionRequestMessage } from '@/components/task/chat/messages/permission-request-message';
import { StatusMessage } from '@/components/task/chat/messages/status-message';
import { ToolCallMessage } from '@/components/task/chat/messages/tool-call-message';
import { ThinkingMessage } from '@/components/task/chat/messages/thinking-message';
@@ -11,6 +12,7 @@ import { TodoMessage } from '@/components/task/chat/messages/todo-message';
type AdapterContext = {
isTaskDescription: boolean;
taskId?: string;
permissionsByToolCallId?: Map<string, Message>;
};
type MessageAdapter = {
@@ -29,12 +31,21 @@ const adapters: MessageAdapter[] = [
},
{
matches: (comment) => comment.type === 'tool_call',
render: (comment, ctx) => <ToolCallMessage comment={comment} taskId={ctx.taskId} />,
render: (comment, ctx) => {
const toolCallId = (comment.metadata as { tool_call_id?: string } | undefined)?.tool_call_id;
const permissionMessage = toolCallId ? ctx.permissionsByToolCallId?.get(toolCallId) : undefined;
return <ToolCallMessage comment={comment} permissionMessage={permissionMessage} />;
},
},
{
matches: (comment) => comment.type === 'error' || comment.type === 'status' || comment.type === 'progress',
render: (comment) => <StatusMessage comment={comment} />,
},
{
// Standalone permission requests (no matching tool call)
matches: (comment) => comment.type === 'permission_request',
render: (comment) => <PermissionRequestMessage comment={comment} />,
},
{
matches: () => true,
render: (comment, ctx) => {
@@ -73,10 +84,11 @@ type MessageRendererProps = {
comment: Message;
isTaskDescription: boolean;
taskId?: string;
permissionsByToolCallId?: Map<string, Message>;
};
export function MessageRenderer({ comment, isTaskDescription, taskId }: MessageRendererProps) {
const ctx = { isTaskDescription, taskId };
export function MessageRenderer({ comment, isTaskDescription, taskId, permissionsByToolCallId }: MessageRendererProps) {
const ctx = { isTaskDescription, taskId, permissionsByToolCallId };
const adapter = adapters.find((entry) => entry.matches(comment, ctx)) ?? adapters[adapters.length - 1];
return adapter.render(comment, ctx);
}
apps/web/components/task/chat/messages/permission-request-message.tsx
+170
View File
@@ -0,0 +1,170 @@
'use client';
import { useCallback, useState } from 'react';
import {
IconAlertTriangle,
IconCheck,
IconX,
IconLoader2,
} from '@tabler/icons-react';
import { Button } from '@kandev/ui/button';
import { cn } from '@/lib/utils';
import { getWebSocketClient } from '@/lib/ws/connection';
import type { Message } from '@/lib/types/http';
type PermissionOption = {
option_id: string;
name: string;
kind: string;
};
type PermissionRequestMetadata = {
pending_id: string;
tool_call_id?: string;
options: PermissionOption[];
action_type: string;
action_details?: {
command?: string;
path?: string;
cwd?: string;
};
status?: 'pending' | 'approved' | 'rejected';
};
type PermissionRequestMessageProps = {
comment: Message;
};
export function PermissionRequestMessage({ comment }: PermissionRequestMessageProps) {
const [isResponding, setIsResponding] = useState(false);
const metadata = comment.metadata as PermissionRequestMetadata | undefined;
const isResolved = metadata?.status === 'approved' || metadata?.status === 'rejected';
const isPending = !isResolved;
const handleRespond = useCallback(
async (optionId: string, cancelled: boolean = false) => {
if (!metadata) return;
const client = getWebSocketClient();
if (!client) {
console.error('WebSocket client not available');
return;
}
setIsResponding(true);
try {
await client.request('permission.respond', {
session_id: comment.session_id,
pending_id: metadata.pending_id,
option_id: cancelled ? undefined : optionId,
cancelled,
});
} catch (error) {
console.error('Failed to respond to permission request:', error);
} finally {
setIsResponding(false);
}
},
[comment.session_id, metadata]
);
const handleApprove = useCallback(() => {
const allowOption = metadata?.options.find(
(opt) => opt.kind === 'allow_once' || opt.kind === 'allow_always'
);
if (allowOption) {
handleRespond(allowOption.option_id);
}
}, [metadata, handleRespond]);
const handleReject = useCallback(() => {
const rejectOption = metadata?.options.find(
(opt) => opt.kind === 'reject_once' || opt.kind === 'reject_always'
);
if (rejectOption) {
handleRespond(rejectOption.option_id);
} else {
handleRespond('', true);
}
}, [metadata, handleRespond]);
const getStatusBadge = () => {
if (metadata?.status === 'approved') {
return (
<span className="inline-flex items-center gap-1 text-xs text-green-600 dark:text-green-400">
<IconCheck className="h-3 w-3" /> Approved
</span>
);
}
if (metadata?.status === 'rejected') {
return (
<span className="inline-flex items-center gap-1 text-xs text-red-600 dark:text-red-400">
<IconX className="h-3 w-3" /> Rejected
</span>
);
}
return (
<span className="inline-flex items-center gap-1 text-xs text-amber-600 dark:text-amber-400">
<IconLoader2 className="h-3 w-3 animate-spin" /> Pending Approval
</span>
);
};
return (
<div
className={cn(
'w-full rounded-md border overflow-hidden',
isPending ? 'border-amber-500/50 bg-amber-500/5' : 'border-border/50 bg-muted/20'
)}
>
{/* Header */}
<div className="flex items-center gap-2 px-3 py-2 text-sm">
<IconAlertTriangle
className={cn(
'h-4 w-4 flex-shrink-0',
isPending ? 'text-amber-600 dark:text-amber-400' : 'text-muted-foreground'
)}
/>
<span className="flex-1 font-medium text-foreground truncate">
{comment.content || 'Permission Required'}
</span>
{getStatusBadge()}
</div>
{/* Approval buttons - only show when pending */}
{isPending && (
<div className="flex items-center gap-2 px-3 py-2 border-t border-amber-500/30 bg-amber-500/10">
<span className="text-xs text-amber-600 dark:text-amber-400 flex-1">
Approve this action?
</span>
<Button
size="sm"
variant="ghost"
onClick={handleReject}
disabled={isResponding}
className="h-7 px-2 text-red-600 hover:text-red-700 hover:bg-red-500/10"
>
<IconX className="h-4 w-4 mr-1" />
Deny
</Button>
<Button
size="sm"
variant="ghost"
onClick={handleApprove}
disabled={isResponding}
className="h-7 px-2 text-green-600 hover:text-green-700 hover:bg-green-500/10"
>
{isResponding ? (
<IconLoader2 className="h-4 w-4 mr-1 animate-spin" />
) : (
<IconCheck className="h-4 w-4 mr-1" />
)}
Approve
</Button>
</div>
)}
</div>
);
}
apps/web/components/task/chat/messages/tool-call-message.tsx
+84 -67
View File
@@ -16,10 +16,28 @@ import {
} from '@tabler/icons-react';
import { Button } from '@kandev/ui/button';
import { cn } 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 { useAppStore } from '@/components/state-provider';
import { getWebSocketClient } from '@/lib/ws/connection';
type PermissionOption = {
option_id: string;
name: string;
kind: string;
};
type PermissionRequestMetadata = {
pending_id: string;
tool_call_id: string;
options: PermissionOption[];
action_type: string;
action_details: {
command?: string;
path?: string;
cwd?: string;
};
status?: 'pending' | 'approved' | 'rejected';
};
function getToolIcon(toolName: string | undefined, className: string) {
const name = toolName?.toLowerCase() ?? '';
@@ -41,7 +59,15 @@ function getToolIcon(toolName: string | undefined, className: string) {
return <IconCode className={className} />;
}
function getStatusIcon(status?: string) {
function getStatusIcon(status?: string, permissionStatus?: string) {
// If there's a permission status, show that instead
if (permissionStatus === 'approved') {
return <IconCheck className="h-3.5 w-3.5 text-green-500" />;
}
if (permissionStatus === 'rejected') {
return <IconX className="h-3.5 w-3.5 text-red-500" />;
}
switch (status) {
case 'complete':
return <IconCheck className="h-3.5 w-3.5 text-green-500" />;
@@ -56,74 +82,64 @@ function getStatusIcon(status?: string) {
type ToolCallMessageProps = {
comment: Message;
taskId?: string;
permissionMessage?: Message;
};
export function ToolCallMessage({ comment, taskId }: ToolCallMessageProps) {
export function ToolCallMessage({ comment, permissionMessage }: ToolCallMessageProps) {
const [isExpanded, setIsExpanded] = useState(false);
const [isResponding, setIsResponding] = useState(false);
const metadata = comment.metadata as ToolCallMetadata | undefined;
const pendingPermissions = useAppStore((state) => state.permissions.pending);
const removePendingPermission = useAppStore((state) => state.removePendingPermission);
const activeSessionId = useAppStore((state) => state.tasks.activeSessionId);
const permissionMetadata = permissionMessage?.metadata as PermissionRequestMetadata | undefined;
const toolCallId = metadata?.tool_call_id;
const toolName = metadata?.tool_name ?? '';
const title = metadata?.title ?? comment.content ?? 'Tool call';
const status = metadata?.status;
const args = metadata?.args;
const result = metadata?.result;
// Find pending permission for this tool call (only if we have a valid tool_call_id)
const pendingPermission = pendingPermissions.find(
(p) =>
toolCallId &&
p.tool_call_id === toolCallId &&
(!activeSessionId || p.session_id === activeSessionId) &&
(!taskId || p.task_id === taskId)
);
const isPendingApproval = !!pendingPermission;
// Permission state
const hasPermission = !!permissionMessage;
const permissionStatus = permissionMetadata?.status;
const isPermissionPending = hasPermission && permissionStatus !== 'approved' && permissionStatus !== 'rejected';
const handleRespond = useCallback(
async (optionId: string, cancelled: boolean = false) => {
if (!pendingPermission) return;
setIsResponding(true);
if (!permissionMetadata || !permissionMessage) return;
const client = getWebSocketClient();
if (!client) {
console.error('WebSocket client not available');
setIsResponding(false);
return;
}
setIsResponding(true);
try {
await client.request('permission.respond', {
session_id: pendingPermission.session_id,
pending_id: pendingPermission.pending_id,
session_id: permissionMessage.session_id,
pending_id: permissionMetadata.pending_id,
option_id: cancelled ? undefined : optionId,
cancelled,
});
removePendingPermission(pendingPermission.pending_id);
} catch (error) {
console.error('Failed to respond to permission request:', error);
} finally {
setIsResponding(false);
}
},
[pendingPermission, removePendingPermission]
[permissionMessage, permissionMetadata]
);
const handleApprove = useCallback(() => {
const allowOption = pendingPermission?.options.find(
const allowOption = permissionMetadata?.options.find(
(opt) => opt.kind === 'allow_once' || opt.kind === 'allow_always'
);
if (allowOption) {
handleRespond(allowOption.option_id);
}
}, [pendingPermission, handleRespond]);
}, [permissionMetadata, handleRespond]);
const handleReject = useCallback(() => {
const rejectOption = pendingPermission?.options.find(
const rejectOption = permissionMetadata?.options.find(
(opt) => opt.kind === 'reject_once' || opt.kind === 'reject_always'
);
if (rejectOption) {
@@ -131,9 +147,12 @@ export function ToolCallMessage({ comment, taskId }: ToolCallMessageProps) {
} else {
handleRespond('', true);
}
}, [pendingPermission, handleRespond]);
}, [permissionMetadata, handleRespond]);
const toolIcon = getToolIcon(toolName, 'h-4 w-4 text-amber-600 dark:text-amber-400 flex-shrink-0');
const toolIcon = getToolIcon(toolName, cn(
'h-4 w-4 flex-shrink-0',
isPermissionPending ? 'text-amber-600 dark:text-amber-400' : 'text-amber-600 dark:text-amber-400'
));
const hasDetails = args && Object.keys(args).length > 0;
let filePath: string | undefined;
@@ -151,9 +170,7 @@ export function ToolCallMessage({ comment, taskId }: ToolCallMessageProps) {
return (
<div className={cn(
'w-full rounded-md border overflow-hidden',
isPendingApproval
? 'border-amber-500/50 bg-amber-500/5'
: 'border-border/50 bg-muted/20'
isPermissionPending ? 'border-amber-500/50 bg-amber-500/5' : 'border-border/50 bg-muted/20'
)}>
<button
type="button"
@@ -173,7 +190,7 @@ export function ToolCallMessage({ comment, taskId }: ToolCallMessageProps) {
{filePath}
</span>
)}
{!isPendingApproval && getStatusIcon(status)}
{getStatusIcon(status, permissionStatus)}
{hasDetails && (
isExpanded
? <IconChevronDown className="h-4 w-4 text-muted-foreground/50" />
@@ -181,39 +198,6 @@ export function ToolCallMessage({ comment, taskId }: ToolCallMessageProps) {
)}
</button>
{/* Inline approval buttons when permission is pending */}
{isPendingApproval && (
<div className="flex items-center gap-2 px-3 py-2 border-t border-amber-500/30 bg-amber-500/10">
<span className="text-xs text-amber-600 dark:text-amber-400 flex-1">
Approve this action?
</span>
<Button
size="sm"
variant="ghost"
onClick={handleReject}
disabled={isResponding}
className="h-7 px-2 text-red-600 hover:text-red-700 hover:bg-red-500/10"
>
<IconX className="h-4 w-4 mr-1" />
Deny
</Button>
<Button
size="sm"
variant="ghost"
onClick={handleApprove}
disabled={isResponding}
className="h-7 px-2 text-green-600 hover:text-green-700 hover:bg-green-500/10"
>
{isResponding ? (
<IconLoader2 className="h-4 w-4 mr-1 animate-spin" />
) : (
<IconCheck className="h-4 w-4 mr-1" />
)}
Approve
</Button>
</div>
)}
{isExpanded && hasDetails && (
<div className="border-t border-border/30 bg-background/50 p-3 space-y-2">
{args && Object.entries(args).map(([key, value]) => {
@@ -243,6 +227,39 @@ export function ToolCallMessage({ comment, taskId }: ToolCallMessageProps) {
)}
</div>
)}
{/* Inline permission approval UI */}
{isPermissionPending && (
<div className="flex items-center gap-2 px-3 py-2 border-t border-amber-500/30 bg-amber-500/10">
<span className="text-xs text-amber-600 dark:text-amber-400 flex-1">
Approve this action?
</span>
<Button
size="sm"
variant="ghost"
onClick={handleReject}
disabled={isResponding}
className="h-7 px-2 text-red-600 hover:text-red-700 hover:bg-red-500/10"
>
<IconX className="h-4 w-4 mr-1" />
Deny
</Button>
<Button
size="sm"
variant="ghost"
onClick={handleApprove}
disabled={isResponding}
className="h-7 px-2 text-green-600 hover:text-green-700 hover:bg-green-500/10"
>
{isResponding ? (
<IconLoader2 className="h-4 w-4 mr-1 animate-spin" />
) : (
<IconCheck className="h-4 w-4 mr-1" />
)}
Approve
</Button>
</div>
)}
</div>
);
}
apps/web/components/task/command-approval-dialog.tsx
-187
View File
@@ -1,187 +0,0 @@
'use client';
import { useCallback, useMemo, useState } from 'react';
import { IconTerminal2, IconAlertTriangle, IconLoader2 } from '@tabler/icons-react';
import {
AlertDialog,
AlertDialogAction,
AlertDialogCancel,
AlertDialogContent,
AlertDialogDescription,
AlertDialogFooter,
AlertDialogHeader,
AlertDialogTitle,
} from '@kandev/ui/alert-dialog';
import { useAppStore } from '@/components/state-provider';
import { getWebSocketClient } from '@/lib/ws/connection';
import type { Message } from '@/lib/types/http';
const EMPTY_MESSAGES: Message[] = [];
type CommandApprovalDialogProps = {
taskId: string;
/** Only show permissions that don't have a matching tool call (standalone permissions like workspace indexing) */
standaloneOnly?: boolean;
};
export function CommandApprovalDialog({ taskId, standaloneOnly = false }: CommandApprovalDialogProps) {
const [isResponding, setIsResponding] = useState(false);
const pendingPermissions = useAppStore((state) => state.permissions.pending);
const removePendingPermission = useAppStore((state) => state.removePendingPermission);
const activeSessionId = useAppStore((state) => state.tasks.activeSessionId);
const messages = useAppStore((state) =>
activeSessionId ? state.messages.bySession[activeSessionId] ?? EMPTY_MESSAGES : EMPTY_MESSAGES
);
// Get tool_call_ids from all tool call messages
const toolCallIds = useMemo(() => {
const ids = new Set<string>();
for (const msg of messages) {
const tcid = msg.metadata?.tool_call_id;
if (msg.type === 'tool_call' && typeof tcid === 'string' && tcid) {
ids.add(tcid);
}
}
return ids;
}, [messages]);
// Get the first pending permission for this task
// If standaloneOnly, filter out permissions that have a matching tool call
const permission = useMemo(() => {
return pendingPermissions.find((p) => {
if (activeSessionId) {
if (p.session_id !== activeSessionId) return false;
} else if (p.task_id !== taskId) {
return false;
}
if (standaloneOnly && p.tool_call_id && toolCallIds.has(p.tool_call_id)) {
return false; // This permission will be handled by inline UI
}
return true;
});
}, [pendingPermissions, activeSessionId, taskId, standaloneOnly, toolCallIds]);
const handleRespond = useCallback(
async (optionId: string, cancelled: boolean = false) => {
if (!permission) return;
setIsResponding(true);
const client = getWebSocketClient();
if (!client) {
console.error('WebSocket client not available');
setIsResponding(false);
return;
}
try {
await client.request('permission.respond', {
session_id: permission.session_id,
pending_id: permission.pending_id,
option_id: cancelled ? undefined : optionId,
cancelled,
});
removePendingPermission(permission.pending_id);
} catch (error) {
console.error('Failed to respond to permission request:', error);
} finally {
setIsResponding(false);
}
},
[permission, removePendingPermission]
);
const handleApprove = useCallback(() => {
// Find the "allow" option (allow_once or allow_always)
const allowOption = permission?.options.find(
(opt) => opt.kind === 'allow_once' || opt.kind === 'allow_always'
);
if (allowOption) {
handleRespond(allowOption.option_id);
}
}, [permission, handleRespond]);
const handleReject = useCallback(() => {
// Find the "reject" option (reject_once or reject_always)
const rejectOption = permission?.options.find(
(opt) => opt.kind === 'reject_once' || opt.kind === 'reject_always'
);
if (rejectOption) {
handleRespond(rejectOption.option_id);
} else {
// If no reject option, cancel the request
handleRespond('', true);
}
}, [permission, handleRespond]);
if (!permission) {
return null;
}
// Extract command details from action_details
const actionDetails = permission.action_details || {};
const command = actionDetails.command as string | undefined;
const cwd = actionDetails.cwd as string | undefined;
const reasoning = actionDetails.reasoning as string | undefined;
return (
<AlertDialog open={!!permission}>
<AlertDialogContent className="max-w-lg">
<AlertDialogHeader>
<div className="flex items-center gap-2">
<div className="flex h-10 w-10 items-center justify-center rounded-full bg-amber-500/10">
<IconAlertTriangle className="h-5 w-5 text-amber-500" />
</div>
<AlertDialogTitle className="text-base">
{permission.title || 'Command Execution Approval'}
</AlertDialogTitle>
</div>
<AlertDialogDescription className="text-left">
The agent is requesting permission to execute a command.
</AlertDialogDescription>
</AlertDialogHeader>
<div className="space-y-3">
{command && (
<div className="rounded-md border bg-muted/50 p-3">
<div className="mb-1 flex items-center gap-1.5 text-xs font-medium text-muted-foreground">
<IconTerminal2 className="h-3.5 w-3.5" />
Command
</div>
<code className="block whitespace-pre-wrap break-all font-mono text-sm">
{command}
</code>
</div>
)}
{cwd && (
<div className="text-xs text-muted-foreground">
<span className="font-medium">Working directory:</span> {cwd}
</div>
)}
{reasoning && (
<div className="text-sm text-muted-foreground">
<span className="font-medium">Reason:</span> {reasoning}
</div>
)}
</div>
<AlertDialogFooter>
<AlertDialogCancel onClick={handleReject} disabled={isResponding}>
Reject
</AlertDialogCancel>
<AlertDialogAction onClick={handleApprove} variant="default" disabled={isResponding}>
{isResponding ? (
<>
<IconLoader2 className="h-4 w-4 mr-1 animate-spin" />
Approving...
</>
) : (
'Approve'
)}
</AlertDialogAction>
</AlertDialogFooter>
</AlertDialogContent>
</AlertDialog>
);
}
apps/web/components/task/task-chat-panel.tsx
+57 -12
View File
@@ -1,6 +1,6 @@
'use client';
import { FormEvent, useCallback, useEffect, useRef, useState } from 'react';
import { FormEvent, useCallback, useEffect, useMemo, useRef, useState } from 'react';
import { flushSync } from 'react-dom';
import { useVirtualizer } from '@tanstack/react-virtual';
import { IconBrain, IconListCheck, IconLoader2, IconPaperclip } from '@tabler/icons-react';
@@ -68,18 +68,62 @@ export function TaskChatPanel({
const showLoadingState = (messagesLoading || isInitialLoading) && !isWorking;
const { loadMore, hasMore, isLoading: isLoadingMore } = useLazyLoadMessages(resolvedSessionId);
// Build a set of tool_call_ids that exist in the messages
const toolCallIds = useMemo(() => {
const set = new Set<string>();
for (const message of messages) {
if (message.type === 'tool_call') {
const toolCallId = (message.metadata as { tool_call_id?: string } | undefined)?.tool_call_id;
if (toolCallId) {
set.add(toolCallId);
}
}
}
return set;
}, [messages]);
// Build a lookup map of tool_call_id → permission_request message
const permissionsByToolCallId = useMemo(() => {
const map = new Map<string, Message>();
for (const message of messages) {
if (message.type === 'permission_request') {
const toolCallId = (message.metadata as { tool_call_id?: string } | undefined)?.tool_call_id;
if (toolCallId) {
map.set(toolCallId, message);
}
}
}
return map;
}, [messages]);
// Filter to only show chat-relevant types.
const visibleMessages = messages.filter((message) =>
!message.type ||
message.type === 'message' ||
message.type === 'content' ||
message.type === 'tool_call' ||
message.type === 'progress' ||
message.type === 'status' ||
message.type === 'error' ||
message.type === 'thinking' ||
message.type === 'todo'
);
// Permission requests are only hidden if they have a matching tool_call in the messages
// (so they can be displayed inline). Standalone permissions are shown as separate messages.
const visibleMessages = messages.filter((message) => {
// Standard visible types
if (
!message.type ||
message.type === 'message' ||
message.type === 'content' ||
message.type === 'tool_call' ||
message.type === 'progress' ||
message.type === 'status' ||
message.type === 'error' ||
message.type === 'thinking' ||
message.type === 'todo'
) {
return true;
}
// Permission requests: show if no matching tool_call exists (standalone permission)
if (message.type === 'permission_request') {
const toolCallId = (message.metadata as { tool_call_id?: string } | undefined)?.tool_call_id;
// Show as standalone if no tool_call_id or no matching tool_call message exists
return !toolCallId || !toolCallIds.has(toolCallId);
}
return false;
});
// Create a synthetic "user" message for the task description
const taskDescriptionMessage: Message | null =
@@ -248,6 +292,7 @@ export function TaskChatPanel({
comment={message}
isTaskDescription={message.id === 'task-description'}
taskId={taskId ?? undefined}
permissionsByToolCallId={permissionsByToolCallId}
/>
</div>
</div>
apps/web/lib/state/store.ts
-57
View File
@@ -175,30 +175,6 @@ export type DiffState = {
files: Array<{ path: string; status: 'A' | 'M' | 'D'; plus: number; minus: number }>;
};
export type PermissionOption = {
option_id: string;
name: string;
kind: string; // allow_once, allow_always, reject_once, reject_always
};
export type PendingPermission = {
pending_id: string;
task_id: string;
agent_instance_id: string;
session_id: string;
tool_call_id: string;
title: string;
options: PermissionOption[];
action_type?: string; // command, file_write, file_read, network, mcp_tool, other
action_details?: Record<string, unknown>;
created_at: string;
timestamp: string;
};
export type PermissionsState = {
pending: PendingPermission[];
};
export type ConnectionState = {
status: 'disconnected' | 'connecting' | 'connected' | 'error' | 'reconnecting';
error: string | null;
@@ -263,7 +239,6 @@ export type AppState = {
sessionAgentctl: SessionAgentctlState;
worktrees: WorktreesState;
sessionWorktreesBySessionId: SessionWorktreesState;
permissions: PermissionsState;
hydrate: (state: Partial<AppState>) => void;
setActiveWorkspace: (workspaceId: string | null) => void;
setWorkspaces: (workspaces: WorkspaceState['items']) => void;
@@ -317,9 +292,6 @@ export type AppState = {
setGitStatus: (sessionId: string, gitStatus: GitStatusEntry) => void;
clearGitStatus: (sessionId: string) => void;
bumpAgentProfilesVersion: () => void;
addPendingPermission: (permission: PendingPermission) => void;
removePendingPermission: (pendingId: string) => void;
clearPendingPermissions: (sessionId?: string) => void;
};
export type AppStore = ReturnType<typeof createAppStore>;
@@ -356,7 +328,6 @@ const defaultState: AppState = {
sessionAgentctl: { itemsBySessionId: {} },
worktrees: { items: {} },
sessionWorktreesBySessionId: { itemsBySessionId: {} },
permissions: { pending: [] },
hydrate: () => undefined,
setActiveWorkspace: () => undefined,
setWorkspaces: () => undefined,
@@ -396,9 +367,6 @@ const defaultState: AppState = {
setGitStatus: () => undefined,
clearGitStatus: () => undefined,
bumpAgentProfilesVersion: () => undefined,
addPendingPermission: () => undefined,
removePendingPermission: () => undefined,
clearPendingPermissions: () => undefined,
};
function mergeInitialState(
@@ -444,9 +412,6 @@ function mergeInitialState(
| 'setGitStatus'
| 'clearGitStatus'
| 'bumpAgentProfilesVersion'
| 'addPendingPermission'
| 'removePendingPermission'
| 'clearPendingPermissions'
> {
if (!initialState) return defaultState;
return {
@@ -478,7 +443,6 @@ function mergeInitialState(
...defaultState.sessionWorktreesBySessionId,
...initialState.sessionWorktreesBySessionId,
},
permissions: { ...defaultState.permissions, ...initialState.permissions },
};
}
@@ -521,7 +485,6 @@ export function createAppStore(initialState?: Partial<AppState>) {
if (state.sessionWorktreesBySessionId) {
Object.assign(draft.sessionWorktreesBySessionId, state.sessionWorktreesBySessionId);
}
if (state.permissions) Object.assign(draft.permissions, state.permissions);
}),
setActiveWorkspace: (workspaceId) =>
set((draft) => {
@@ -793,26 +756,6 @@ export function createAppStore(initialState?: Partial<AppState>) {
set((draft) => {
draft.agentProfiles.version += 1;
}),
addPendingPermission: (permission) =>
set((draft) => {
// Avoid duplicates
const exists = draft.permissions.pending.some((p) => p.pending_id === permission.pending_id);
if (!exists) {
draft.permissions.pending.push(permission);
}
}),
removePendingPermission: (pendingId) =>
set((draft) => {
draft.permissions.pending = draft.permissions.pending.filter((p) => p.pending_id !== pendingId);
}),
clearPendingPermissions: (sessionId) =>
set((draft) => {
if (sessionId) {
draft.permissions.pending = draft.permissions.pending.filter((p) => p.session_id !== sessionId);
} else {
draft.permissions.pending = [];
}
}),
}))
);
}
apps/web/lib/types/backend.ts
+1 -23
View File
@@ -36,8 +36,7 @@ export type BackendMessageType =
| 'agent.profile.updated'
| 'user.settings.updated'
| 'session.workspace.file.changes'
| 'session.shell.output'
| 'permission.requested';
| 'session.shell.output';
export type BackendMessage<T extends BackendMessageType, P> = {
id?: string;
@@ -254,26 +253,6 @@ export type ShellOutputPayload = {
code?: number;
};
export type PermissionOptionPayload = {
option_id: string;
name: string;
kind: string; // allow_once, allow_always, reject_once, reject_always
};
export type PermissionRequestedPayload = {
pending_id: string;
task_id: string;
agent_instance_id: string;
session_id: string;
tool_call_id: string;
title: string;
options: PermissionOptionPayload[];
action_type?: string; // command, file_write, file_read, network, mcp_tool, other
action_details?: Record<string, unknown>;
created_at: string;
timestamp: string;
};
export type BackendMessageMap = {
'kanban.update': BackendMessage<'kanban.update', KanbanUpdatePayload>;
'task.created': BackendMessage<'task.created', TaskEventPayload>;
@@ -313,7 +292,6 @@ export type BackendMessageMap = {
'user.settings.updated': BackendMessage<'user.settings.updated', UserSettingsUpdatedPayload>;
'session.workspace.file.changes': BackendMessage<'session.workspace.file.changes', FileChangeNotificationPayload>;
'session.shell.output': BackendMessage<'session.shell.output', ShellOutputPayload>;
'permission.requested': BackendMessage<'permission.requested', PermissionRequestedPayload>;
};
// Workspace file types
apps/web/lib/types/http.ts
+2 -1
View File
@@ -307,7 +307,8 @@ export type MessageType =
| 'error'
| 'status'
| 'thinking'
| 'todo';
| 'todo'
| 'permission_request';
export type Message = {
id: string;
apps/web/lib/ws/handlers/messages.ts
+3 -2
View File
@@ -1,6 +1,7 @@
import type { StoreApi } from 'zustand';
import type { AppState } from '@/lib/state/store';
import type { WsHandlers } from '@/lib/ws/handlers/types';
import type { MessageType } from '@/lib/types/http';
export function registerMessagesHandlers(store: StoreApi<AppState>): WsHandlers {
return {
@@ -16,7 +17,7 @@ export function registerMessagesHandlers(store: StoreApi<AppState>): WsHandlers
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',
type: (payload.type as MessageType) || 'message',
metadata: payload.metadata,
requests_input: payload.requests_input,
created_at: payload.created_at,
@@ -34,7 +35,7 @@ export function registerMessagesHandlers(store: StoreApi<AppState>): WsHandlers
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',
type: (payload.type as MessageType) || 'message',
metadata: payload.metadata,
requests_input: payload.requests_input,
created_at: payload.created_at,
apps/web/lib/ws/handlers/permissions.ts
-29
View File
@@ -1,29 +0,0 @@
import type { StoreApi } from 'zustand';
import type { AppState, PendingPermission } from '@/lib/state/store';
import type { WsHandlers } from '@/lib/ws/handlers/types';
import type { PermissionRequestedPayload } from '@/lib/types/backend';
export function registerPermissionsHandlers(store: StoreApi<AppState>): WsHandlers {
return {
'permission.requested': (message) => {
const payload = message.payload as PermissionRequestedPayload;
const state = store.getState();
// Convert to PendingPermission type (they're compatible)
const permission: PendingPermission = {
pending_id: payload.pending_id,
task_id: payload.task_id,
agent_instance_id: payload.agent_instance_id,
session_id: payload.session_id,
tool_call_id: payload.tool_call_id,
title: payload.title,
options: payload.options,
action_type: payload.action_type,
action_details: payload.action_details,
created_at: payload.created_at,
timestamp: payload.timestamp,
};
state.addPendingPermission(permission);
},
};
}
apps/web/lib/ws/router.ts
-2
View File
@@ -11,7 +11,6 @@ import { registerEnvironmentsHandlers } from '@/lib/ws/handlers/environments';
import { registerExecutorsHandlers } from '@/lib/ws/handlers/executors';
import { registerGitStatusHandlers } from '@/lib/ws/handlers/git-status';
import { registerKanbanHandlers } from '@/lib/ws/handlers/kanban';
import { registerPermissionsHandlers } from '@/lib/ws/handlers/permissions';
import { registerSystemEventsHandlers } from '@/lib/ws/handlers/system-events';
import { registerTasksHandlers } from '@/lib/ws/handlers/tasks';
import { registerTerminalsHandlers } from '@/lib/ws/handlers/terminals';
@@ -35,7 +34,6 @@ export function registerWsHandlers(store: StoreApi<AppState>) {
...registerMessagesHandlers(store),
...registerNotificationsHandlers(store),
...registerGitStatusHandlers(store),
...registerPermissionsHandlers(store),
...registerSystemEventsHandlers(store),
};
}