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

Merge pull request #4 from kdlbs/feat/acp-permission-requests

Browse Source 
feat: implement ACP permission request flow
This commit is contained in:
João Salavisa
2026-01-11 22:26:10 +00:00
committed by GitHub
parent 473b336b13 c1c9dfa2c1
commit 4706e3be40
29 changed files with 2504 additions and 751 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+3
View File
@@ -68,3 +68,6 @@ tmp/
apps/backend/bin/
# Go binaries (built locally)
apps/backend/kandev
apps/backend/agentctl
apps/backend/CODE_REVIEW_ISSUES.md
+52
View File
@@ -0,0 +1,52 @@
# Backend Code Review Issues
Generated: 2026-01-11
## High Priority
### 1. Security TODOs
- [ ] `internal/gateway/websocket/handler.go:20` - TODO: In production, validate origin
- [ ] `internal/gateway/websocket/handler.go:46` - TODO: Implement proper JWT validation
### 2. Deprecated API
- [ ] `internal/agent/docker/client.go:300-301` - `inspect.NetworkSettings.IPAddress` is deprecated (removed in Docker v29). Use `NetworkSettings.Networks` instead.
## Medium Priority
### 3. Unused Functions
- [ ] `internal/agent/lifecycle/manager.go:641` - `updateInstanceComplete` is unused
- [ ] `internal/agent/streaming/reader.go:232` - `marshalMessage` is unused
- [ ] `internal/task/repository/sqlite.go:617` - `scanTasks` is unused
### 4. Unused Field / Race Condition
- [ ] `internal/gateway/websocket/client.go:36` - `mu sync.RWMutex` field is unused. Either use it to protect `subscriptions` or remove it.
### 5. Outdated Comment
- [ ] `internal/agent/agentctl/acp.go:108` - Comment references `needs_input` which was removed. Update to: `// The StopReason indicates why the agent stopped (e.g., "end_turn")`
## Low Priority
### 6. Unused Test Helpers
- [ ] `internal/agent/lifecycle/manager_test.go:130` - `testManager` type unused
- [ ] `internal/agent/lifecycle/manager_test.go:135` - `newTestManager` function unused
### 7. Code Style (staticcheck S1000)
- [ ] `internal/agentctl/api/acp.go:215` - Use `for range ch` instead of `for { select {} }`
- [ ] `internal/agentctl/api/acp.go:336` - Use `for range ch` instead of `for { select {} }`
### 8. Hardcoded Sleep
- [ ] `internal/agent/lifecycle/manager.go:253` - `time.Sleep(500 * time.Millisecond)` - Consider proper ready-check mechanism
### 9. Ignored Error
- [ ] `internal/task/repository/sqlite.go:626` - `_ = json.Unmarshal(...)` ignores error
## Questions / Review Needed
### 10. InputRequestHandler Redundancy
Files: `internal/orchestrator/service.go` (lines 44-45, 71, 387-389, 507-512), `cmd/kandev/main.go` (lines 305-363)
The `InputRequestHandler` and `handleInputRequired` functionality still exists using `input_required` message type. Is this still needed given the new comment-based conversation flow?
### 11. Other TODOs
- [ ] `internal/gateway/websocket/hub.go:127` - TODO: Add topic-based routing for task-specific notifications
apps/backend/cmd/agentctl/main.go
+99
View File
@@ -0,0 +1,99 @@
// Package main is the entry point for the agentctl binary
// agentctl is a sidecar process that manages agent subprocess communication
// via HTTP API, bridging the agent's ACP protocol with the kandev backend.
package main
import (
"context"
"fmt"
"net/http"
"os"
"os/signal"
"syscall"
"time"
"github.com/kandev/kandev/internal/agentctl/api"
"github.com/kandev/kandev/internal/agentctl/config"
"github.com/kandev/kandev/internal/agentctl/process"
"github.com/kandev/kandev/internal/common/logger"
"go.uber.org/zap"
)
func main() {
// Load configuration from environment
cfg := config.Load()
// Initialize logger
log, err := logger.NewLogger(logger.LoggingConfig{
Level: cfg.LogLevel,
Format: cfg.LogFormat,
OutputPath: "stdout",
})
if err != nil {
fmt.Fprintf(os.Stderr, "failed to initialize logger: %v\n", err)
os.Exit(1)
}
defer log.Sync()
log.Info("starting agentctl",
zap.Int("port", cfg.Port),
zap.String("agent_command", cfg.AgentCommand),
zap.String("workdir", cfg.WorkDir),
zap.Bool("auto_start", cfg.AutoStart))
// Create process manager
procMgr := process.NewManager(cfg, log)
// Create HTTP API server
server := api.NewServer(cfg, procMgr, log)
// Start HTTP server
addr := fmt.Sprintf(":%d", cfg.Port)
httpServer := &http.Server{
Addr: addr,
Handler: server.Router(),
}
// Start server in goroutine
go func() {
log.Info("HTTP server starting", zap.String("address", addr))
if err := httpServer.ListenAndServe(); err != nil && err != http.ErrServerClosed {
log.Error("HTTP server error", zap.Error(err))
os.Exit(1)
}
}()
// Auto-start agent if configured
if cfg.AutoStart {
log.Info("auto-starting agent process")
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
if err := procMgr.Start(ctx); err != nil {
log.Error("failed to auto-start agent", zap.Error(err))
}
cancel()
}
// Wait for shutdown signal
quit := make(chan os.Signal, 1)
signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
<-quit
log.Info("shutting down agentctl...")
// Graceful shutdown
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
// Stop agent process
if err := procMgr.Stop(ctx); err != nil {
log.Error("error stopping agent process", zap.Error(err))
}
// Shutdown HTTP server
if err := httpServer.Shutdown(ctx); err != nil {
log.Error("error shutting down HTTP server", zap.Error(err))
}
log.Info("agentctl stopped")
}
apps/backend/cmd/kandev/main.go
+43 -2
View File
@@ -202,6 +202,9 @@ func main() {
serviceCfg := orchestrator.DefaultServiceConfig()
orchestratorSvc := orchestrator.NewService(serviceCfg, eventBus, nil, agentManagerClient, taskRepoAdapter, log)
// Set up comment creator for saving agent responses as comments
orchestratorSvc.SetCommentCreator(&commentCreatorAdapter{svc: taskSvc})
// ============================================
// WEBSOCKET GATEWAY (All communication via WebSocket)
// ============================================
@@ -396,6 +399,7 @@ func main() {
taskhandlers.RegisterBoardRoutes(router, gateway.Dispatcher, boardController, log)
taskhandlers.RegisterTaskRoutes(router, gateway.Dispatcher, taskController, log)
taskhandlers.RegisterRepositoryRoutes(router, gateway.Dispatcher, repositoryController, log)
taskhandlers.RegisterCommentRoutes(router, gateway.Dispatcher, taskSvc, &orchestratorAdapter{svc: orchestratorSvc}, log)
log.Info("Registered Task Service handlers (HTTP + WebSocket)")
// Health check (simple HTTP for load balancers/monitoring)
@@ -578,11 +582,16 @@ func (a *lifecycleAdapter) PromptAgent(ctx context.Context, agentInstanceID stri
return nil, err
}
return &executor.PromptResult{
StopReason: result.StopReason,
NeedsInput: result.NeedsInput,
StopReason: result.StopReason,
AgentMessage: result.AgentMessage,
}, nil
}
// RespondToPermissionByTaskID sends a response to a permission request for a task
func (a *lifecycleAdapter) RespondToPermissionByTaskID(ctx context.Context, taskID, pendingID, optionID string, cancelled bool) error {
return a.mgr.RespondToPermissionByTaskID(taskID, pendingID, optionID, cancelled)
}
// corsMiddleware returns a CORS middleware for WebSocket connections
func corsMiddleware() gin.HandlerFunc {
return func(c *gin.Context) {
@@ -599,3 +608,35 @@ func corsMiddleware() gin.HandlerFunc {
c.Next()
}
}
// orchestratorAdapter adapts the orchestrator.Service to the taskhandlers.OrchestratorService interface
type orchestratorAdapter struct {
svc *orchestrator.Service
}
// PromptTask forwards to the orchestrator service and converts the result type
func (a *orchestratorAdapter) PromptTask(ctx context.Context, taskID string, prompt string) (*taskhandlers.PromptResult, error) {
result, err := a.svc.PromptTask(ctx, taskID, prompt)
if err != nil {
return nil, err
}
return &taskhandlers.PromptResult{
StopReason: result.StopReason,
AgentMessage: result.AgentMessage,
}, nil
}
// commentCreatorAdapter adapts the task service to the orchestrator.CommentCreator interface
type commentCreatorAdapter struct {
svc *taskservice.Service
}
// CreateAgentComment creates a comment with author_type="agent"
func (a *commentCreatorAdapter) CreateAgentComment(ctx context.Context, taskID, content string) error {
_, err := a.svc.CreateComment(ctx, &taskservice.CreateCommentRequest{
TaskID: taskID,
Content: content,
AuthorType: "agent",
})
return err
}
apps/backend/dockerfiles/augment-agent/Dockerfile
-4
View File
@@ -39,10 +39,6 @@ RUN chmod +x /usr/local/bin/agentctl
# Create agent directory
WORKDIR /agent
# Copy agent script (legacy, for backward compatibility)
COPY dockerfiles/augment-agent/agent.sh /agent/agent.sh
RUN chmod +x /agent/agent.sh
# Copy entrypoint script
COPY dockerfiles/augment-agent/entrypoint.sh /agent/entrypoint.sh
RUN chmod +x /agent/entrypoint.sh
apps/backend/dockerfiles/augment-agent/agent.sh
-31
View File
@@ -1,31 +0,0 @@
#!/bin/bash
# Augment Agent - Runs in ACP (Agent Client Protocol) mode
# Uses JSON-RPC 2.0 over stdin/stdout for bidirectional communication
set -e
# Check if auggie is available
if ! command -v auggie &> /dev/null; then
echo '{"jsonrpc":"2.0","method":"error","params":{"message":"Auggie CLI not found"}}' >&2
exit 1
fi
# Check for authentication
if [ -z "$AUGMENT_SESSION_AUTH" ]; then
echo '{"jsonrpc":"2.0","method":"error","params":{"message":"AUGMENT_SESSION_AUTH required"}}' >&2
exit 1
fi
# Export for auggie to use
export AUGMENT_SESSION_AUTH
# Build auggie command
AUGGIE_ARGS="--acp"
# Add workspace root if we're in a workspace with files
if [ -d "/workspace" ] && [ "$(ls -A /workspace 2>/dev/null)" ]; then
AUGGIE_ARGS="$AUGGIE_ARGS --workspace-root /workspace"
fi
# Execute auggie in ACP mode - stdin/stdout will be used for JSON-RPC communication
exec auggie $AUGGIE_ARGS
apps/backend/internal/agent/acp/session.go
+173 -8
View File
@@ -28,6 +28,29 @@ type Session struct {
mu sync.RWMutex
}
// PendingPermission represents a permission request waiting for user response
type PendingPermission struct {
ID string // Unique ID for this pending request
RPCID interface{} // JSON-RPC request ID from agent
InstanceID string // Agent instance
TaskID string // Task being worked on
SessionID string // ACP session ID
ToolCallID string // Tool call requesting permission
Title string // Human-readable title
Options []jsonrpc.PermissionOption // Available choices
ResponseCh chan *PermissionResponseMsg // Channel to receive user response
CreatedAt time.Time
}
// PermissionResponseMsg is the message sent by the user in response to a permission request
type PermissionResponseMsg struct {
OptionID string
Cancelled bool
}
// PermissionRequestHandler is called when an agent requests permission from the user
type PermissionRequestHandler func(req *PendingPermission)
// UpdateHandler is called when the agent sends session updates
type UpdateHandler func(instanceID, taskID string, updateType string, data json.RawMessage)
@@ -37,17 +60,25 @@ type SessionManager struct {
mu sync.RWMutex
eventBus bus.EventBus
logger *logger.Logger
// Handler for session updates
updateHandler UpdateHandler
// Handler for permission requests
permissionHandler PermissionRequestHandler
// Pending permission requests waiting for user response
pendingPermissions map[string]*PendingPermission
permissionMu sync.RWMutex
}
// NewSessionManager creates a new session manager
func NewSessionManager(eventBus bus.EventBus, log *logger.Logger) *SessionManager {
return &SessionManager{
sessions: make(map[string]*Session),
eventBus: eventBus,
logger: log.WithFields(zap.String("component", "acp-session-manager")),
sessions: make(map[string]*Session),
pendingPermissions: make(map[string]*PendingPermission),
eventBus: eventBus,
logger: log.WithFields(zap.String("component", "acp-session-manager")),
}
}
@@ -56,6 +87,11 @@ func (m *SessionManager) SetUpdateHandler(handler UpdateHandler) {
m.updateHandler = handler
}
// SetPermissionHandler sets the handler for permission requests from agents
func (m *SessionManager) SetPermissionHandler(handler PermissionRequestHandler) {
m.permissionHandler = handler
}
// CreateSession creates a new ACP session for an agent instance
func (m *SessionManager) CreateSession(ctx context.Context, instanceID, taskID string, stdin io.WriteCloser, stdout io.Reader) error {
m.logger.Info("Creating ACP session",
@@ -395,8 +431,91 @@ func (m *SessionManager) handleRequestPermission(session *Session, id interface{
zap.String("title", reqParams.ToolCall.Title),
zap.Int("num_options", len(reqParams.Options)))
// Auto-approve: find the first "allow" option and select it
// In a real implementation, this would prompt the user or check policies
// If no permission handler is set, fall back to auto-approve
if m.permissionHandler == nil {
m.autoApprovePermission(session, id, &reqParams)
return
}
// Create a pending permission request
pendingID := fmt.Sprintf("%s-%d", session.InstanceID, time.Now().UnixNano())
pending := &PendingPermission{
ID: pendingID,
RPCID: id,
InstanceID: session.InstanceID,
TaskID: session.TaskID,
SessionID: reqParams.SessionID,
ToolCallID: reqParams.ToolCall.ToolCallID,
Title: reqParams.ToolCall.Title,
Options: reqParams.Options,
ResponseCh: make(chan *PermissionResponseMsg, 1),
CreatedAt: time.Now(),
}
// Store the pending request
m.permissionMu.Lock()
m.pendingPermissions[pendingID] = pending
m.permissionMu.Unlock()
// Notify the permission handler (which will send WebSocket notification to user)
m.permissionHandler(pending)
// Wait for user response in a goroutine to not block the read loop
go m.waitForPermissionResponse(session, pending)
}
// waitForPermissionResponse waits for user response and sends it to the agent
func (m *SessionManager) waitForPermissionResponse(session *Session, pending *PendingPermission) {
// Wait for response with a 5-minute timeout
timeout := time.After(5 * time.Minute)
select {
case resp := <-pending.ResponseCh:
m.logger.Info("Received permission response from user",
zap.String("pending_id", pending.ID),
zap.String("option_id", resp.OptionID),
zap.Bool("cancelled", resp.Cancelled))
var result jsonrpc.RequestPermissionResult
if resp.Cancelled {
result = jsonrpc.RequestPermissionResult{
Outcome: jsonrpc.PermissionOutcome{
Outcome: "cancelled",
},
}
} else {
result = jsonrpc.RequestPermissionResult{
Outcome: jsonrpc.PermissionOutcome{
Outcome: "selected",
OptionID: resp.OptionID,
},
}
}
if err := session.Client.SendResponse(pending.RPCID, result, nil); err != nil {
m.logger.Error("Failed to send permission response to agent", zap.Error(err))
}
case <-timeout:
m.logger.Warn("Permission request timed out, cancelling",
zap.String("pending_id", pending.ID))
result := jsonrpc.RequestPermissionResult{
Outcome: jsonrpc.PermissionOutcome{
Outcome: "cancelled",
},
}
session.Client.SendResponse(pending.RPCID, result, nil)
}
// Clean up
m.permissionMu.Lock()
delete(m.pendingPermissions, pending.ID)
m.permissionMu.Unlock()
}
// autoApprovePermission auto-approves a permission request (fallback behavior)
func (m *SessionManager) autoApprovePermission(session *Session, id interface{}, reqParams *jsonrpc.RequestPermissionParams) {
var selectedOptionID string
for _, opt := range reqParams.Options {
if opt.Kind == "allow_once" || opt.Kind == "allow_always" {
@@ -416,7 +535,6 @@ func (m *SessionManager) handleRequestPermission(session *Session, id interface{
zap.String("option_id", selectedOptionID))
}
// Send response
result := jsonrpc.RequestPermissionResult{
Outcome: jsonrpc.PermissionOutcome{
Outcome: "selected",
@@ -427,8 +545,55 @@ func (m *SessionManager) handleRequestPermission(session *Session, id interface{
if err := session.Client.SendResponse(id, result, nil); err != nil {
m.logger.Error("Failed to send permission response", zap.Error(err))
} else {
m.logger.Info("Sent permission response",
m.logger.Info("Sent auto-approved permission response",
zap.String("instance_id", session.InstanceID),
zap.String("selected_option", selectedOptionID))
}
}
// RespondToPermission sends a user's response to a pending permission request
func (m *SessionManager) RespondToPermission(pendingID string, optionID string, cancelled bool) error {
m.permissionMu.RLock()
pending, exists := m.pendingPermissions[pendingID]
m.permissionMu.RUnlock()
if !exists {
return fmt.Errorf("pending permission request not found: %s", pendingID)
}
// Send response through channel (non-blocking)
select {
case pending.ResponseCh <- &PermissionResponseMsg{
OptionID: optionID,
Cancelled: cancelled,
}:
m.logger.Info("Sent permission response to waiting goroutine",
zap.String("pending_id", pendingID),
zap.String("option_id", optionID))
return nil
default:
return fmt.Errorf("permission request already responded to or timed out")
}
}
// GetPendingPermission returns a pending permission request by ID
func (m *SessionManager) GetPendingPermission(pendingID string) (*PendingPermission, bool) {
m.permissionMu.RLock()
defer m.permissionMu.RUnlock()
pending, exists := m.pendingPermissions[pendingID]
return pending, exists
}
// GetPendingPermissionsForTask returns all pending permissions for a task
func (m *SessionManager) GetPendingPermissionsForTask(taskID string) []*PendingPermission {
m.permissionMu.RLock()
defer m.permissionMu.RUnlock()
var result []*PendingPermission
for _, pending := range m.pendingPermissions {
if pending.TaskID == taskID {
result = append(result, pending)
}
}
return result
}
apps/backend/internal/agent/agentctl/acp.go
+124
View File
@@ -146,6 +146,36 @@ func (c *Client) Prompt(ctx context.Context, text string) (*PromptResponse, erro
// SessionNotification represents a session update from the agent
type SessionNotification = acp.SessionNotification
// 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"`
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(acp.SessionNotification)) error {
wsURL := "ws" + c.baseURL[4:] + "/api/v1/acp/stream"
@@ -276,5 +306,99 @@ func (c *Client) CloseOutputStream() {
func (c *Client) Close() {
c.CloseACPStream()
c.CloseOutputStream()
c.ClosePermissionStream()
}
// 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()
conn.Close()
}()
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 {
c.permissionConn.Close()
c.permissionConn = nil
}
}
// RespondToPermission sends a response to a permission request
func (c *Client) RespondToPermission(ctx context.Context, pendingID, optionID string, cancelled bool) error {
reqBody := PermissionRespondRequest{
PendingID: pendingID,
OptionID: optionID,
Cancelled: cancelled,
}
body, err := json.Marshal(reqBody)
if err != nil {
return err
}
req, err := http.NewRequestWithContext(ctx, "POST", c.baseURL+"/api/v1/acp/permissions/respond", bytes.NewReader(body))
if err != nil {
return err
}
req.Header.Set("Content-Type", "application/json")
resp, err := c.httpClient.Do(req)
if err != nil {
return err
}
defer resp.Body.Close()
var result PermissionRespondResponse
if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
return err
}
if !result.Success {
return fmt.Errorf("permission response failed: %s", result.Error)
}
return nil
}
apps/backend/internal/agent/agentctl/client.go
+4 -3
View File
@@ -21,9 +21,10 @@ type Client struct {
logger *logger.Logger
// WebSocket connections for streaming
outputConn *websocket.Conn
acpConn *websocket.Conn
mu sync.RWMutex
outputConn *websocket.Conn
acpConn *websocket.Conn
permissionConn *websocket.Conn
mu sync.RWMutex
}
// StatusResponse from agentctl
apps/backend/internal/agent/lifecycle/manager.go
+208 -64
View File
@@ -43,6 +43,10 @@ type AgentInstance struct {
// agentctl client for this instance
agentctl *agentctl.Client
// Message buffer for accumulating agent response during a prompt
messageBuffer strings.Builder
messageMu sync.Mutex
}
// LaunchRequest contains parameters for launching an agent
@@ -311,13 +315,21 @@ func (m *Manager) initializeACPSession(ctx context.Context, instance *AgentInsta
// 3. Set up updates stream to receive session notifications
go m.handleUpdatesStream(instance)
// 4. Send the task prompt if provided
// 4. Set up permission stream to receive permission requests
go m.handlePermissionStream(instance)
// 5. Send the task prompt if provided
if taskDescription != "" {
m.logger.Info("sending ACP prompt",
zap.String("instance_id", instance.ID),
zap.String("session_id", sessionID),
zap.String("task_description", taskDescription))
// Clear message buffer before starting prompt
instance.messageMu.Lock()
instance.messageBuffer.Reset()
instance.messageMu.Unlock()
// Prompt is SYNCHRONOUS - it blocks until the agent completes the task
// Use a long timeout context for this
promptCtx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
@@ -331,7 +343,12 @@ func (m *Manager) initializeACPSession(ctx context.Context, instance *AgentInsta
return fmt.Errorf("prompt failed: %w", err)
}
// Check if agent needs input - if so, publish input_required event
// Extract accumulated message from buffer
instance.messageMu.Lock()
agentMessage := instance.messageBuffer.String()
instance.messageBuffer.Reset()
instance.messageMu.Unlock()
stopReason := ""
if resp != nil {
stopReason = string(resp.StopReason)
@@ -340,20 +357,16 @@ func (m *Manager) initializeACPSession(ctx context.Context, instance *AgentInsta
zap.String("instance_id", instance.ID),
zap.String("stop_reason", stopReason))
if resp != nil && resp.StopReason == "needs_input" {
m.logger.Info("agent requesting user input on initial prompt",
// Publish prompt_complete event with the agent's response (for saving as comment)
m.publishPromptComplete(instance, agentMessage)
// Prompt completed - mark agent as READY for follow-up prompts
m.logger.Info("agent ready for follow-up prompts",
zap.String("instance_id", instance.ID))
if err := m.MarkReady(instance.ID); err != nil {
m.logger.Error("failed to mark instance as ready",
zap.String("instance_id", instance.ID),
zap.String("task_id", instance.TaskID))
m.publishInputRequiredFromPrompt(instance, "The agent needs more information to proceed.")
} else {
// Prompt completed - mark agent as READY for follow-up prompts
m.logger.Info("agent ready for follow-up prompts",
zap.String("instance_id", instance.ID))
if err := m.MarkReady(instance.ID); err != nil {
m.logger.Error("failed to mark instance as ready",
zap.String("instance_id", instance.ID),
zap.Error(err))
}
zap.Error(err))
}
} else {
m.logger.Warn("no task description provided, marking as ready",
@@ -382,6 +395,74 @@ func (m *Manager) handleUpdatesStream(instance *AgentInstance) {
}
}
// handlePermissionStream handles streaming permission requests from the agent
func (m *Manager) handlePermissionStream(instance *AgentInstance) {
ctx := context.Background()
err := instance.agentctl.StreamPermissions(ctx, func(notification *agentctl.PermissionNotification) {
m.handlePermissionNotification(instance, notification)
})
if err != nil {
m.logger.Error("failed to connect to permission stream",
zap.String("instance_id", instance.ID),
zap.Error(err))
}
}
// handlePermissionNotification processes incoming permission requests from the agent
func (m *Manager) handlePermissionNotification(instance *AgentInstance, notification *agentctl.PermissionNotification) {
m.logger.Info("received permission request",
zap.String("instance_id", instance.ID),
zap.String("task_id", instance.TaskID),
zap.String("pending_id", notification.PendingID),
zap.String("title", notification.Title),
zap.Int("num_options", len(notification.Options)))
// Publish permission request event to the event bus
m.publishPermissionRequest(instance, notification)
}
// publishPermissionRequest publishes a permission request event
func (m *Manager) publishPermissionRequest(instance *AgentInstance, notification *agentctl.PermissionNotification) {
// Convert options to a serializable format
options := make([]map[string]interface{}, len(notification.Options))
for i, opt := range notification.Options {
options[i] = map[string]interface{}{
"option_id": opt.OptionID,
"name": opt.Name,
"kind": opt.Kind,
}
}
data := map[string]interface{}{
"type": "permission_request",
"task_id": instance.TaskID,
"agent_instance_id": instance.ID,
"pending_id": notification.PendingID,
"session_id": notification.SessionID,
"tool_call_id": notification.ToolCallID,
"title": notification.Title,
"options": options,
"created_at": notification.CreatedAt,
"timestamp": time.Now().UTC().Format(time.RFC3339),
}
event := bus.NewEvent(events.ACPMessage, "agent-manager", data)
subject := events.BuildACPSubject(instance.TaskID)
if err := m.eventBus.Publish(context.Background(), subject, event); err != nil {
m.logger.Error("failed to publish permission_request event",
zap.String("instance_id", instance.ID),
zap.String("task_id", instance.TaskID),
zap.Error(err))
} else {
m.logger.Info("published permission_request event",
zap.String("task_id", instance.TaskID),
zap.String("pending_id", notification.PendingID),
zap.String("title", notification.Title))
}
}
// handleSessionNotification processes incoming session notifications from the agent
func (m *Manager) handleSessionNotification(instance *AgentInstance, notification agentctl.SessionNotification) {
m.logger.Debug("received session notification",
@@ -395,6 +476,13 @@ func (m *Manager) handleSessionNotification(instance *AgentInstance, notificatio
m.logger.Debug("agent message chunk",
zap.String("instance_id", instance.ID))
m.updateInstanceProgress(instance.ID, 50)
// Accumulate message content for saving as comment when prompt completes
if update.AgentMessageChunk.Content.Text != nil {
instance.messageMu.Lock()
instance.messageBuffer.WriteString(update.AgentMessageChunk.Content.Text.Text)
instance.messageMu.Unlock()
}
}
if update.AgentThoughtChunk != nil {
@@ -465,39 +553,6 @@ func (m *Manager) publishSessionNotification(instance *AgentInstance, notificati
}
}
// publishInputRequiredFromPrompt publishes an input_required event when the agent needs user input
// This is called after a prompt completes with stopReason="needs_input"
func (m *Manager) publishInputRequiredFromPrompt(instance *AgentInstance, message string) {
if m.eventBus == nil {
return
}
// Structure must match protocol.Message with type "input_required"
data := map[string]interface{}{
"type": "input_required",
"timestamp": time.Now().UTC().Format(time.RFC3339Nano),
"agent_id": instance.ID,
"task_id": instance.TaskID,
"data": map[string]interface{}{
"message": message,
},
}
event := bus.NewEvent(events.ACPMessage, "agent-manager", data)
subject := events.BuildACPSubject(instance.TaskID)
if err := m.eventBus.Publish(context.Background(), subject, event); err != nil {
m.logger.Error("failed to publish input_required event",
zap.String("instance_id", instance.ID),
zap.String("task_id", instance.TaskID),
zap.Error(err))
} else {
m.logger.Info("published input_required event",
zap.String("task_id", instance.TaskID),
zap.String("message", message))
}
}
// publishSessionInfo publishes a session_info event with the session ID
// This allows the orchestrator to store the session ID in task metadata for resumption
func (m *Manager) publishSessionInfo(instance *AgentInstance, sessionID string) {
@@ -533,6 +588,42 @@ func (m *Manager) publishSessionInfo(instance *AgentInstance, sessionID string)
}
}
// publishPromptComplete publishes a prompt_complete event when an agent finishes responding
// This is used to save the agent's response as a comment on the task
func (m *Manager) publishPromptComplete(instance *AgentInstance, agentMessage string) {
if m.eventBus == nil {
return
}
// Only publish if there's actual content
if agentMessage == "" {
return
}
data := map[string]interface{}{
"type": "prompt_complete",
"timestamp": time.Now().UTC().Format(time.RFC3339Nano),
"agent_id": instance.ID,
"task_id": instance.TaskID,
"agent_message": agentMessage,
}
event := bus.NewEvent(events.PromptComplete, "agent-manager", data)
// Publish on task-specific subject so orchestrator can subscribe
subject := events.PromptComplete + "." + instance.TaskID
if err := m.eventBus.Publish(context.Background(), subject, event); err != nil {
m.logger.Error("failed to publish prompt_complete event",
zap.String("instance_id", instance.ID),
zap.String("task_id", instance.TaskID),
zap.Error(err))
} else {
m.logger.Info("published prompt_complete event",
zap.String("task_id", instance.TaskID),
zap.Int("message_length", len(agentMessage)))
}
}
// updateInstanceProgress updates an instance's progress
func (m *Manager) updateInstanceProgress(instanceID string, progress int) {
m.mu.Lock()
@@ -684,8 +775,8 @@ func (m *Manager) expandMountTemplate(source, workspacePath, taskID string) stri
// PromptAgent sends a follow-up prompt to a running agent
// PromptResult contains the result of a prompt operation
type PromptResult struct {
StopReason string // The reason the agent stopped (e.g., "end_turn", "needs_input")
NeedsInput bool // True if the agent is requesting user input
StopReason string // The reason the agent stopped (e.g., "end_turn")
AgentMessage string // The agent's accumulated response message
}
// PromptAgent sends a follow-up prompt to a running agent
@@ -710,6 +801,12 @@ func (m *Manager) PromptAgent(ctx context.Context, instanceID string, prompt str
// Set status to RUNNING while processing
instance.Status = v1.AgentStatusRunning
// Clear message buffer before starting new prompt
instance.messageMu.Lock()
instance.messageBuffer.Reset()
instance.messageMu.Unlock()
m.mu.Unlock()
m.logger.Info("sending prompt to agent",
@@ -722,24 +819,25 @@ func (m *Manager) PromptAgent(ctx context.Context, instanceID string, prompt str
return nil, err
}
// Extract accumulated message from buffer
instance.messageMu.Lock()
agentMessage := instance.messageBuffer.String()
instance.messageBuffer.Reset()
instance.messageMu.Unlock()
result := &PromptResult{
StopReason: string(resp.StopReason),
NeedsInput: resp.StopReason == "needs_input",
StopReason: string(resp.StopReason),
AgentMessage: agentMessage,
}
// Check if agent needs input - if so, publish input_required event
if result.NeedsInput {
m.logger.Info("agent requesting user input",
// Publish prompt_complete event with the agent's response (for saving as comment)
m.publishPromptComplete(instance, agentMessage)
// Prompt completed - mark as READY for next prompt
if err := m.MarkReady(instanceID); err != nil {
m.logger.Error("failed to mark instance as ready after prompt",
zap.String("instance_id", instanceID),
zap.String("task_id", instance.TaskID))
m.publishInputRequiredFromPrompt(instance, "The agent needs more information to proceed.")
} else {
// Prompt completed normally - mark as READY for next prompt
if err := m.MarkReady(instanceID); err != nil {
m.logger.Error("failed to mark instance as ready after prompt",
zap.String("instance_id", instanceID),
zap.Error(err))
}
zap.Error(err))
}
return result, nil
@@ -1076,4 +1174,50 @@ func (m *Manager) performCleanup(ctx context.Context) {
}
}
}
}
// RespondToPermission sends a response to a permission request for an agent instance
func (m *Manager) RespondToPermission(instanceID, pendingID, optionID string, cancelled bool) error {
m.mu.RLock()
instance, exists := m.instances[instanceID]
m.mu.RUnlock()
if !exists {
return fmt.Errorf("agent instance not found: %s", instanceID)
}
if instance.agentctl == nil {
return fmt.Errorf("agent instance has no agentctl client: %s", instanceID)
}
m.logger.Info("responding to permission request",
zap.String("instance_id", instanceID),
zap.String("pending_id", pendingID),
zap.String("option_id", optionID),
zap.Bool("cancelled", cancelled))
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
return instance.agentctl.RespondToPermission(ctx, pendingID, optionID, cancelled)
}
// RespondToPermissionByTaskID sends a response to a permission request for a task
func (m *Manager) RespondToPermissionByTaskID(taskID, pendingID, optionID string, cancelled bool) error {
// Find the instance by task ID
m.mu.RLock()
var instance *AgentInstance
for _, inst := range m.instances {
if inst.TaskID == taskID {
instance = inst
break
}
}
m.mu.RUnlock()
if instance == nil {
return fmt.Errorf("no agent instance found for task: %s", taskID)
}
return m.RespondToPermission(instance.ID, pendingID, optionID, cancelled)
}
apps/backend/internal/agent/registry/defaults.go
+5
View File
@@ -11,6 +11,11 @@ func DefaultAgents() []*AgentTypeConfig {
Tag: "latest",
WorkingDir: "/workspace",
RequiredEnv: []string{"AUGMENT_SESSION_AUTH"},
Env: map[string]string{
// Auto-approve permissions for now (until full permission flow is implemented)
// This allows the agent to proceed without waiting for user approval
"AGENTCTL_AUTO_APPROVE_PERMISSIONS": "true",
},
Mounts: []MountTemplate{
{Source: "{workspace}", Target: "/workspace", ReadOnly: false},
{Source: "{augment_sessions}", Target: "/root/.augment/sessions", ReadOnly: false},
apps/backend/internal/agentctl/acp/client.go
+119 -3
View File
@@ -16,13 +16,39 @@ import (
// UpdateHandler is called when session updates are received from the agent
type UpdateHandler func(notification acp.SessionNotification)
// PermissionRequestHandler is called when the agent requests permission
// Returns the selected option ID, or empty string with cancelled=true to cancel
type PermissionRequestHandler func(ctx context.Context, req *PermissionRequest) (*PermissionResponse, error)
// PermissionRequest represents a permission request from the agent
type PermissionRequest struct {
SessionID string `json:"session_id"`
ToolCallID string `json:"tool_call_id"`
Title string `json:"title"`
Options []PermissionOption `json:"options"`
}
// 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
}
// PermissionResponse is the user's response to a permission request
type PermissionResponse struct {
OptionID string `json:"option_id,omitempty"`
Cancelled bool `json:"cancelled,omitempty"`
}
// Client implements acp.Client interface and handles all agent requests
type Client struct {
logger *zap.Logger
workspaceRoot string
mu sync.RWMutex
updateHandler UpdateHandler
mu sync.RWMutex
updateHandler UpdateHandler
permissionHandler PermissionRequestHandler
}
// ClientOption configures a Client
@@ -49,6 +75,13 @@ func WithUpdateHandler(h UpdateHandler) ClientOption {
}
}
// WithPermissionHandler sets the handler for permission requests
func WithPermissionHandler(h PermissionRequestHandler) ClientOption {
return func(c *Client) {
c.permissionHandler = h
}
}
// NewClient creates a new ACP client implementation
func NewClient(opts ...ClientOption) *Client {
c := &Client{
@@ -68,8 +101,16 @@ func (c *Client) SetUpdateHandler(h UpdateHandler) {
c.updateHandler = h
}
// SetPermissionHandler sets the permission request handler (thread-safe)
func (c *Client) SetPermissionHandler(h PermissionRequestHandler) {
c.mu.Lock()
defer c.mu.Unlock()
c.permissionHandler = h
}
// RequestPermission handles permission requests from the agent
// Auto-approves by selecting the first "allow" option
// If a permission handler is set, it forwards the request to the handler.
// Otherwise, auto-approves by selecting the first "allow" option.
func (c *Client) RequestPermission(ctx context.Context, p acp.RequestPermissionRequest) (acp.RequestPermissionResponse, error) {
title := ""
if p.ToolCall.Title != nil {
@@ -91,6 +132,81 @@ func (c *Client) RequestPermission(ctx context.Context, p acp.RequestPermissionR
}, nil
}
// Check if we have a permission handler
c.mu.RLock()
handler := c.permissionHandler
c.mu.RUnlock()
if handler != nil {
// Forward to external handler (e.g., backend/user)
return c.forwardPermissionRequest(ctx, handler, p)
}
// Fall back to auto-approve
return c.autoApprovePermission(p)
}
// forwardPermissionRequest forwards the permission request to an external handler
func (c *Client) forwardPermissionRequest(ctx context.Context, handler PermissionRequestHandler, p acp.RequestPermissionRequest) (acp.RequestPermissionResponse, error) {
// Convert ACP types to our types
options := make([]PermissionOption, len(p.Options))
for i, opt := range p.Options {
options[i] = PermissionOption{
OptionID: string(opt.OptionId),
Name: opt.Name,
Kind: string(opt.Kind),
}
}
title := ""
if p.ToolCall.Title != nil {
title = *p.ToolCall.Title
}
req := &PermissionRequest{
SessionID: string(p.SessionId),
ToolCallID: string(p.ToolCall.ToolCallId),
Title: title,
Options: options,
}
c.logger.Info("forwarding permission request to handler",
zap.String("session_id", req.SessionID),
zap.String("tool_call_id", req.ToolCallID))
resp, err := handler(ctx, req)
if err != nil {
c.logger.Error("permission handler failed", zap.Error(err))
// On error, cancel the permission request
return acp.RequestPermissionResponse{
Outcome: acp.RequestPermissionOutcome{
Cancelled: &acp.RequestPermissionOutcomeCancelled{},
},
}, nil
}
if resp.Cancelled {
c.logger.Info("permission request cancelled by user")
return acp.RequestPermissionResponse{
Outcome: acp.RequestPermissionOutcome{
Cancelled: &acp.RequestPermissionOutcomeCancelled{},
},
}, nil
}
c.logger.Info("permission request approved by user",
zap.String("option_id", resp.OptionID))
return acp.RequestPermissionResponse{
Outcome: acp.RequestPermissionOutcome{
Selected: &acp.RequestPermissionOutcomeSelected{
OptionId: acp.PermissionOptionId(resp.OptionID),
},
},
}, nil
}
// autoApprovePermission auto-approves by selecting the first "allow" option
func (c *Client) autoApprovePermission(p acp.RequestPermissionRequest) (acp.RequestPermissionResponse, error) {
// Find the first "allow" option
var selectedOption *acp.PermissionOption
for i := range p.Options {
apps/backend/internal/agentctl/api/acp.go
+120
View File
@@ -233,3 +233,123 @@ func (s *Server) handleACPStreamWS(c *gin.Context) {
}
}
// PermissionRespondRequest is a request to respond to a permission request
type PermissionRespondRequest struct {
PendingID string `json:"pending_id" binding:"required"`
OptionID string `json:"option_id,omitempty"`
Cancelled bool `json:"cancelled,omitempty"`
}
// PermissionRespondResponse is the response to a permission respond call
type PermissionRespondResponse struct {
Success bool `json:"success"`
Error string `json:"error,omitempty"`
}
func (s *Server) handlePermissionRespond(c *gin.Context) {
var req PermissionRespondRequest
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, PermissionRespondResponse{
Success: false,
Error: "invalid request: " + err.Error(),
})
return
}
s.logger.Info("received permission response",
zap.String("pending_id", req.PendingID),
zap.String("option_id", req.OptionID),
zap.Bool("cancelled", req.Cancelled))
if err := s.procMgr.RespondToPermission(req.PendingID, req.OptionID, req.Cancelled); err != nil {
s.logger.Error("failed to respond to permission", zap.Error(err))
c.JSON(http.StatusNotFound, PermissionRespondResponse{
Success: false,
Error: err.Error(),
})
return
}
c.JSON(http.StatusOK, PermissionRespondResponse{
Success: true,
})
}
// 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 conn.Close()
s.logger.Info("Permission stream WebSocket connected")
// Get the permission requests channel
permissionCh := s.procMgr.GetPermissionRequests()
// Stream permission notifications to WebSocket
for {
select {
case notification, ok := <-permissionCh:
if !ok {
return
}
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/api/server.go
+5
View File
@@ -69,6 +69,11 @@ func (s *Server) setupRoutes() {
api.POST("/acp/prompt", s.handleACPPrompt)
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)
// Output streaming
api.GET("/output", s.handleGetOutput)
api.GET("/output/stream", s.handleOutputStreamWS)
apps/backend/internal/agentctl/config/config.go
+12 -8
View File
@@ -27,6 +27,9 @@ type Config struct {
// Auto-start the agent on agentctl startup
AutoStart bool
// Auto-approve permission requests (for testing/CI)
AutoApprovePermissions bool
// Logging configuration
LogLevel string
LogFormat string
@@ -44,14 +47,15 @@ func Load() *Config {
defaultCmd := "auggie --acp --workspace-root " + workDir
cfg := &Config{
Port: getEnvInt("AGENTCTL_PORT", 9999),
AgentCommand: getEnv("AGENTCTL_AGENT_COMMAND", defaultCmd),
WorkDir: workDir,
AutoStart: getEnvBool("AGENTCTL_AUTO_START", false),
LogLevel: getEnv("AGENTCTL_LOG_LEVEL", "info"),
LogFormat: getEnv("AGENTCTL_LOG_FORMAT", "json"),
OutputBufferSize: getEnvInt("AGENTCTL_OUTPUT_BUFFER_SIZE", 1000),
HealthCheckInterval: getEnvInt("AGENTCTL_HEALTH_CHECK_INTERVAL", 5),
Port: getEnvInt("AGENTCTL_PORT", 9999),
AgentCommand: getEnv("AGENTCTL_AGENT_COMMAND", defaultCmd),
WorkDir: workDir,
AutoStart: getEnvBool("AGENTCTL_AUTO_START", false),
AutoApprovePermissions: getEnvBool("AGENTCTL_AUTO_APPROVE_PERMISSIONS", false),
LogLevel: getEnv("AGENTCTL_LOG_LEVEL", "info"),
LogFormat: getEnv("AGENTCTL_LOG_FORMAT", "json"),
OutputBufferSize: getEnvInt("AGENTCTL_OUTPUT_BUFFER_SIZE", 1000),
HealthCheckInterval: getEnvInt("AGENTCTL_HEALTH_CHECK_INTERVAL", 5),
}
// Parse agent command into args
apps/backend/internal/agentctl/process/manager.go
+201 -4
View File
@@ -39,6 +39,24 @@ type errorWrapper struct {
err error
}
// PendingPermission represents a permission request waiting for user response
type PendingPermission struct {
ID string
Request *acpclient.PermissionRequest
ResponseCh chan *acpclient.PermissionResponse
CreatedAt time.Time
}
// PermissionNotification is sent 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 []acpclient.PermissionOption `json:"options"`
CreatedAt time.Time `json:"created_at"`
}
// Manager manages the agent subprocess
type Manager struct {
cfg *config.Config
@@ -64,6 +82,13 @@ type Manager struct {
// Session update notifications
updatesCh chan acp.SessionNotification
// Permission request notifications (sent to backend)
permissionCh chan *PermissionNotification
// Pending permission requests waiting for user response
pendingPermissions map[string]*PendingPermission
permissionMu sync.RWMutex
// Synchronization
mu sync.RWMutex
wg sync.WaitGroup
@@ -75,10 +100,12 @@ type Manager struct {
// NewManager creates a new process manager
func NewManager(cfg *config.Config, log *logger.Logger) *Manager {
m := &Manager{
cfg: cfg,
logger: log.WithFields(zap.String("component", "process-manager")),
outputBuffer: NewOutputBuffer(cfg.OutputBufferSize),
updatesCh: make(chan acp.SessionNotification, 100),
cfg: cfg,
logger: log.WithFields(zap.String("component", "process-manager")),
outputBuffer: NewOutputBuffer(cfg.OutputBufferSize),
updatesCh: make(chan acp.SessionNotification, 100),
permissionCh: make(chan *PermissionNotification, 10),
pendingPermissions: make(map[string]*PendingPermission),
}
m.status.Store(StatusStopped)
m.exitCode.Store(-1)
@@ -179,6 +206,7 @@ func (m *Manager) Start(ctx context.Context) error {
m.logger.Warn("updates channel full, dropping notification")
}
}),
acpclient.WithPermissionHandler(m.handlePermissionRequest),
)
// Create ACP SDK connection - it handles JSON-RPC communication
@@ -329,3 +357,172 @@ func (m *Manager) GetProcessInfo() map[string]interface{} {
return info
}
// handlePermissionRequest handles permission requests from the agent
// It stores the pending request and waits for a response from the backend
func (m *Manager) handlePermissionRequest(ctx context.Context, req *acpclient.PermissionRequest) (*acpclient.PermissionResponse, error) {
// Generate a unique ID for this permission request
pendingID := fmt.Sprintf("%s-%s-%d", req.SessionID, req.ToolCallID, time.Now().UnixNano())
m.logger.Info("handling permission request",
zap.String("pending_id", pendingID),
zap.String("session_id", req.SessionID),
zap.String("tool_call_id", req.ToolCallID),
zap.String("title", req.Title),
zap.Bool("auto_approve", m.cfg.AutoApprovePermissions))
// If auto-approve is enabled, immediately approve with the first "allow" option
if m.cfg.AutoApprovePermissions {
return m.autoApprovePermission(req)
}
// Create pending permission with response channel
pending := &PendingPermission{
ID: pendingID,
Request: req,
ResponseCh: make(chan *acpclient.PermissionResponse, 1),
CreatedAt: time.Now(),
}
// Store pending permission
m.permissionMu.Lock()
m.pendingPermissions[pendingID] = pending
m.permissionMu.Unlock()
// Clean up when done
defer func() {
m.permissionMu.Lock()
delete(m.pendingPermissions, pendingID)
m.permissionMu.Unlock()
}()
// Send notification to backend via updates channel
// We use a custom notification type that the backend will recognize
m.sendPermissionNotification(pending)
// Wait for response with timeout
select {
case resp := <-pending.ResponseCh:
m.logger.Info("received permission response",
zap.String("pending_id", pendingID),
zap.String("option_id", resp.OptionID),
zap.Bool("cancelled", resp.Cancelled))
return resp, nil
case <-ctx.Done():
m.logger.Warn("permission request context cancelled",
zap.String("pending_id", pendingID))
return &acpclient.PermissionResponse{Cancelled: true}, nil
case <-time.After(5 * time.Minute):
m.logger.Warn("permission request timed out",
zap.String("pending_id", pendingID))
return &acpclient.PermissionResponse{Cancelled: true}, nil
}
}
// autoApprovePermission automatically approves a permission request
// by selecting the first "allow" option, or the first option if no allow option exists
func (m *Manager) autoApprovePermission(req *acpclient.PermissionRequest) (*acpclient.PermissionResponse, error) {
if len(req.Options) == 0 {
m.logger.Warn("no options available for auto-approve, cancelling")
return &acpclient.PermissionResponse{Cancelled: true}, nil
}
// Find the first "allow" option
var selectedOption *acpclient.PermissionOption
for i := range req.Options {
opt := &req.Options[i]
if opt.Kind == "allow_once" || opt.Kind == "allow_always" {
selectedOption = opt
break
}
}
// If no allow option, use the first option
if selectedOption == nil {
selectedOption = &req.Options[0]
}
m.logger.Info("auto-approving permission request",
zap.String("option_id", selectedOption.OptionID),
zap.String("option_name", selectedOption.Name),
zap.String("kind", selectedOption.Kind))
return &acpclient.PermissionResponse{
OptionID: selectedOption.OptionID,
}, nil
}
// sendPermissionNotification sends a permission request notification to the backend
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,
CreatedAt: pending.CreatedAt,
}
m.logger.Info("sending permission notification to backend",
zap.String("pending_id", pending.ID),
zap.String("title", pending.Request.Title))
select {
case m.permissionCh <- notification:
// Sent successfully
default:
m.logger.Warn("permission channel full, dropping 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()
pending, ok := m.pendingPermissions[pendingID]
m.permissionMu.RUnlock()
if !ok {
return fmt.Errorf("pending permission not found: %s", pendingID)
}
m.logger.Info("responding to permission request",
zap.String("pending_id", pendingID),
zap.String("option_id", optionID),
zap.Bool("cancelled", cancelled))
// Send response (non-blocking since channel is buffered)
select {
case pending.ResponseCh <- &acpclient.PermissionResponse{
OptionID: optionID,
Cancelled: cancelled,
}:
return nil
default:
return fmt.Errorf("response channel full for pending permission: %s", pendingID)
}
}
// 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
}
// GetPendingPermission returns a specific pending permission request
func (m *Manager) GetPendingPermission(pendingID string) (*PendingPermission, bool) {
m.permissionMu.RLock()
defer m.permissionMu.RUnlock()
p, ok := m.pendingPermissions[pendingID]
return p, ok
}
apps/backend/internal/events/types.go
+5
View File
@@ -64,6 +64,11 @@ const (
ACPMessage = "acp.message" // Base subject for ACP messages
)
// Event types for agent prompts
const (
PromptComplete = "prompt.complete" // Agent finished responding to a prompt
)
// BuildACPSubject creates an ACP subject for a specific task
func BuildACPSubject(taskID string) string {
return ACPMessage + "." + taskID
apps/backend/internal/integration/orchestrator_test.go
+11 -1
View File
@@ -19,6 +19,7 @@ import (
"github.com/gorilla/websocket"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"go.uber.org/zap"
"github.com/kandev/kandev/internal/common/logger"
"github.com/kandev/kandev/internal/events"
@@ -343,10 +344,19 @@ func (s *SimulatedAgentManagerClient) PromptAgent(ctx context.Context, agentInst
return &executor.PromptResult{
StopReason: "end_turn",
NeedsInput: false,
}, nil
}
// RespondToPermissionByTaskID responds to a permission request for a task
func (s *SimulatedAgentManagerClient) RespondToPermissionByTaskID(ctx context.Context, taskID, pendingID, optionID string, cancelled bool) error {
s.logger.Info("simulated: responding to permission",
zap.String("task_id", taskID),
zap.String("pending_id", pendingID),
zap.String("option_id", optionID),
zap.Bool("cancelled", cancelled))
return nil
}
// CompleteAgent marks an agent as completed
func (s *SimulatedAgentManagerClient) CompleteAgent(instanceID string) {
s.mu.Lock()
apps/backend/internal/orchestrator/executor/executor.go
+27 -3
View File
@@ -22,8 +22,8 @@ var (
// PromptResult contains the result of a prompt operation
type PromptResult struct {
StopReason string // The reason the agent stopped (e.g., "end_turn", "needs_input")
NeedsInput bool // True if the agent is requesting user input
StopReason string // The reason the agent stopped (e.g., "end_turn")
AgentMessage string // The agent's accumulated response message
}
// AgentManagerClient is an interface for the Agent Manager service
@@ -44,6 +44,9 @@ type AgentManagerClient interface {
// PromptAgent sends a prompt to a running agent
// Returns PromptResult indicating if the agent needs input
PromptAgent(ctx context.Context, agentInstanceID string, prompt string) (*PromptResult, error)
// RespondToPermission sends a response to a permission request
RespondToPermissionByTaskID(ctx context.Context, taskID, pendingID, optionID string, cancelled bool) error
}
// LaunchAgentRequest contains parameters for launching an agent
@@ -228,6 +231,17 @@ func (e *Executor) Prompt(ctx context.Context, taskID string, prompt string) (*P
return e.agentManager.PromptAgent(ctx, agentInstanceID, prompt)
}
// RespondToPermission sends a response to a permission request for a task
func (e *Executor) RespondToPermission(ctx context.Context, taskID, pendingID, optionID string, cancelled bool) error {
e.logger.Info("responding to permission request",
zap.String("task_id", taskID),
zap.String("pending_id", pendingID),
zap.String("option_id", optionID),
zap.Bool("cancelled", cancelled))
return e.agentManager.RespondToPermissionByTaskID(ctx, taskID, pendingID, optionID, cancelled)
}
// GetExecution returns the current execution state for a task
func (e *Executor) GetExecution(taskID string) (*TaskExecution, bool) {
e.mu.RLock()
@@ -380,5 +394,15 @@ func (m *MockAgentManagerClient) PromptAgent(ctx context.Context, agentInstanceI
m.logger.Info("mock: prompting agent",
zap.String("agent_instance_id", agentInstanceID),
zap.Int("prompt_length", len(prompt)))
return &PromptResult{StopReason: "end_turn", NeedsInput: false}, nil
return &PromptResult{StopReason: "end_turn"}, nil
}
// RespondToPermissionByTaskID mocks responding to a permission request
func (m *MockAgentManagerClient) RespondToPermissionByTaskID(ctx context.Context, taskID, pendingID, optionID string, cancelled bool) error {
m.logger.Info("mock: responding to permission",
zap.String("task_id", taskID),
zap.String("pending_id", pendingID),
zap.String("option_id", optionID),
zap.Bool("cancelled", cancelled))
return nil
}
apps/backend/internal/orchestrator/scheduler/scheduler_test.go
+5 -1
View File
@@ -56,7 +56,11 @@ func (m *mockAgentManager) ListAgentTypes(ctx context.Context) ([]*v1.AgentType,
}
func (m *mockAgentManager) PromptAgent(ctx context.Context, agentInstanceID string, prompt string) (*executor.PromptResult, error) {
return &executor.PromptResult{StopReason: "end_turn", NeedsInput: false}, nil
return &executor.PromptResult{StopReason: "end_turn"}, nil
}
func (m *mockAgentManager) RespondToPermissionByTaskID(ctx context.Context, taskID, pendingID, optionID string, cancelled bool) error {
return nil
}
// testTaskRepository is an in-memory task repository for testing
apps/backend/internal/orchestrator/service.go
+48 -5
View File
@@ -44,6 +44,11 @@ func DefaultServiceConfig() ServiceConfig {
// InputRequestHandler is called when an agent requests user input
type InputRequestHandler func(ctx context.Context, taskID, agentID, message string) error
// CommentCreator is an interface for creating comments on tasks
type CommentCreator interface {
CreateAgentComment(ctx context.Context, taskID, content string) error
}
// Service is the main orchestrator service
type Service struct {
config ServiceConfig
@@ -65,6 +70,9 @@ type Service struct {
// Input request handler (for agent-user conversation)
inputRequestHandler InputRequestHandler
// Comment creator for saving agent responses
commentCreator CommentCreator
// Service state
mu sync.RWMutex
running bool
@@ -122,12 +130,18 @@ func NewService(
OnAgentCompleted: s.handleAgentCompleted,
OnAgentFailed: s.handleAgentFailed,
OnACPMessage: s.handleACPMessage,
OnPromptComplete: s.handlePromptComplete,
}
s.watcher = watcher.NewWatcher(eventBus, handlers, log)
return s
}
// SetCommentCreator sets the comment creator for saving agent responses
func (s *Service) SetCommentCreator(cc CommentCreator) {
s.commentCreator = cc
}
// Start starts all orchestrator components
func (s *Service) Start(ctx context.Context) error {
s.mu.Lock()
@@ -273,12 +287,11 @@ func (s *Service) StopTask(ctx context.Context, taskID string, reason string, fo
// PromptResult contains the result of a prompt operation
type PromptResult struct {
StopReason string // The reason the agent stopped (e.g., "end_turn", "needs_input")
NeedsInput bool // True if the agent is requesting user input
StopReason string // The reason the agent stopped (e.g., "end_turn")
AgentMessage string // The agent's accumulated response message
}
// PromptTask sends a follow-up prompt to a running agent for a task
// Returns PromptResult indicating if the agent needs input
func (s *Service) PromptTask(ctx context.Context, taskID string, prompt string) (*PromptResult, error) {
s.logger.Info("sending prompt to task agent",
zap.String("task_id", taskID),
@@ -288,11 +301,21 @@ func (s *Service) PromptTask(ctx context.Context, taskID string, prompt string)
return nil, err
}
return &PromptResult{
StopReason: result.StopReason,
NeedsInput: result.NeedsInput,
StopReason: result.StopReason,
AgentMessage: result.AgentMessage,
}, nil
}
// RespondToPermission sends a response to a permission request for a task
func (s *Service) RespondToPermission(ctx context.Context, taskID, pendingID, optionID string, cancelled bool) error {
s.logger.Info("responding to permission request",
zap.String("task_id", taskID),
zap.String("pending_id", pendingID),
zap.String("option_id", optionID),
zap.Bool("cancelled", cancelled))
return s.executor.RespondToPermission(ctx, taskID, pendingID, optionID, cancelled)
}
// CompleteTask explicitly completes a task and stops its agent
func (s *Service) CompleteTask(ctx context.Context, taskID string) error {
s.logger.Info("completing task",
@@ -490,3 +513,23 @@ func (s *Service) handleInputRequired(ctx context.Context, taskID string, msg *p
}
}
// handlePromptComplete handles prompt complete events and saves agent response as comment
func (s *Service) handlePromptComplete(ctx context.Context, data watcher.PromptCompleteData) {
s.logger.Info("handling prompt complete",
zap.String("task_id", data.TaskID),
zap.Int("message_length", len(data.AgentMessage)))
// Save agent response as a comment if we have a comment creator
if s.commentCreator != nil && data.AgentMessage != "" {
if err := s.commentCreator.CreateAgentComment(ctx, data.TaskID, data.AgentMessage); err != nil {
s.logger.Error("failed to create agent comment",
zap.String("task_id", data.TaskID),
zap.Error(err))
} else {
s.logger.Info("created agent comment",
zap.String("task_id", data.TaskID),
zap.Int("message_length", len(data.AgentMessage)))
}
}
}
apps/backend/internal/orchestrator/watcher/watcher.go
+58
View File
@@ -33,6 +33,13 @@ type AgentEventData struct {
ErrorMessage string `json:"error_message,omitempty"`
}
// PromptCompleteData contains data from prompt_complete events
type PromptCompleteData struct {
TaskID string `json:"task_id"`
AgentID string `json:"agent_id"`
AgentMessage string `json:"agent_message"`
}
// EventHandlers contains callbacks for different event types
type EventHandlers struct {
// Task events
@@ -50,6 +57,9 @@ type EventHandlers struct {
// ACP messages
OnACPMessage func(ctx context.Context, taskID string, msg *protocol.Message)
// Prompt events
OnPromptComplete func(ctx context.Context, data PromptCompleteData)
}
// Watcher subscribes to events and dispatches to handlers
@@ -104,6 +114,12 @@ func (w *Watcher) Start(ctx context.Context) error {
return err
}
// Subscribe to prompt complete events
if err := w.subscribeToPromptCompleteEvents(); err != nil {
w.unsubscribeAll()
return err
}
w.running = true
w.logger.Info("Event watcher started", zap.Int("subscriptions", len(w.subscriptions)))
return nil
@@ -302,6 +318,48 @@ func (w *Watcher) createACPMessageHandler() bus.EventHandler {
}
}
// subscribeToPromptCompleteEvents subscribes to prompt complete events
func (w *Watcher) subscribeToPromptCompleteEvents() error {
if w.handlers.OnPromptComplete == nil {
return nil
}
// Use wildcard to subscribe to all prompt complete events (prompt.complete.*)
subject := events.PromptComplete + ".*"
// Use regular subscription (each instance needs all messages for comment creation)
sub, err := w.eventBus.Subscribe(subject, w.createPromptCompleteHandler())
if err != nil {
w.logger.Error("Failed to subscribe to prompt complete events",
zap.String("subject", subject),
zap.Error(err))
return err
}
w.subscriptions = append(w.subscriptions, sub)
return nil
}
// createPromptCompleteHandler creates a bus.EventHandler for prompt complete events
func (w *Watcher) createPromptCompleteHandler() bus.EventHandler {
return func(ctx context.Context, event *bus.Event) error {
var data PromptCompleteData
if err := w.parseEventData(event.Data, &data); err != nil {
w.logger.Error("Failed to parse prompt complete event data",
zap.String("event_type", event.Type),
zap.String("event_id", event.ID),
zap.Error(err))
return nil // Don't return error to continue processing other events
}
w.logger.Debug("Handling prompt complete event",
zap.String("task_id", data.TaskID),
zap.Int("message_length", len(data.AgentMessage)))
w.handlers.OnPromptComplete(ctx, data)
return nil
}
}
// parseEventData converts event data map to a typed struct
func (w *Watcher) parseEventData(data map[string]interface{}, target interface{}) error {
// Marshal to JSON and unmarshal to target type
apps/backend/internal/orchestrator/wshandlers/handlers.go
+44 -1
View File
@@ -41,6 +41,7 @@ func (h *Handlers) RegisterHandlers(d *ws.Dispatcher) {
d.RegisterFunc(ws.ActionOrchestratorPrompt, h.PromptTask)
d.RegisterFunc(ws.ActionOrchestratorComplete, h.CompleteTask)
d.RegisterFunc(ws.ActionTaskLogs, h.GetTaskLogs)
d.RegisterFunc(ws.ActionPermissionRespond, h.RespondToPermission)
}
// GetStatus handles orchestrator.status action
@@ -203,7 +204,6 @@ func (h *Handlers) PromptTask(ctx context.Context, msg *ws.Message) (*ws.Message
return ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{
"success": true,
"needs_input": result.NeedsInput,
"stop_reason": result.StopReason,
})
}
@@ -294,3 +294,46 @@ func (h *Handlers) GetTaskLogs(ctx context.Context, msg *ws.Message) (*ws.Messag
Total: len(logs),
})
}
// PermissionRespondRequest is the payload for permission.respond
type PermissionRespondRequest struct {
TaskID string `json:"task_id"`
PendingID string `json:"pending_id"`
OptionID string `json:"option_id,omitempty"`
Cancelled bool `json:"cancelled,omitempty"`
}
// RespondToPermission handles permission.respond action
func (h *Handlers) RespondToPermission(ctx context.Context, msg *ws.Message) (*ws.Message, error) {
var req PermissionRespondRequest
if err := msg.ParsePayload(&req); err != nil {
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil)
}
if req.TaskID == "" {
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_id is required", nil)
}
if req.PendingID == "" {
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "pending_id is required", nil)
}
if !req.Cancelled && req.OptionID == "" {
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "option_id is required when not cancelled", nil)
}
h.logger.Info("responding to permission request",
zap.String("task_id", req.TaskID),
zap.String("pending_id", req.PendingID),
zap.String("option_id", req.OptionID),
zap.Bool("cancelled", req.Cancelled))
if err := h.service.RespondToPermission(ctx, req.TaskID, req.PendingID, req.OptionID, req.Cancelled); err != nil {
h.logger.Error("failed to respond to permission", zap.String("task_id", req.TaskID), zap.Error(err))
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to respond to permission: "+err.Error(), nil)
}
return ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{
"success": true,
"task_id": req.TaskID,
"pending_id": req.PendingID,
})
}
apps/backend/internal/task/handlers/comment_handlers.go
+118
View File
@@ -0,0 +1,118 @@
package handlers
import (
"context"
"github.com/gin-gonic/gin"
"go.uber.org/zap"
"github.com/kandev/kandev/internal/common/logger"
"github.com/kandev/kandev/internal/task/service"
ws "github.com/kandev/kandev/pkg/websocket"
)
// PromptResult represents the result of prompting an agent
type PromptResult struct {
StopReason string `json:"stop_reason"`
AgentMessage string `json:"agent_message"`
}
// OrchestratorService defines the interface for orchestrator operations
type OrchestratorService interface {
PromptTask(ctx context.Context, taskID string, prompt string) (*PromptResult, error)
}
// CommentHandlers handles WebSocket requests for comments
type CommentHandlers struct {
svc *service.Service
orchestrator OrchestratorService
logger *logger.Logger
}
// NewCommentHandlers creates a new CommentHandlers instance
func NewCommentHandlers(svc *service.Service, orchestrator OrchestratorService, log *logger.Logger) *CommentHandlers {
return &CommentHandlers{
svc: svc,
orchestrator: orchestrator,
logger: log.WithFields(zap.String("component", "task-comment-handlers")),
}
}
// RegisterCommentRoutes registers comment WebSocket handlers
func RegisterCommentRoutes(router *gin.Engine, dispatcher *ws.Dispatcher, svc *service.Service, orchestrator OrchestratorService, log *logger.Logger) {
handlers := NewCommentHandlers(svc, orchestrator, log)
handlers.registerWS(dispatcher)
}
func (h *CommentHandlers) registerWS(dispatcher *ws.Dispatcher) {
dispatcher.RegisterFunc(ws.ActionCommentAdd, h.wsAddComment)
dispatcher.RegisterFunc(ws.ActionCommentList, h.wsListComments)
}
// WS handlers
type wsAddCommentRequest struct {
TaskID string `json:"task_id"`
Content string `json:"content"`
AuthorID string `json:"author_id,omitempty"`
}
func (h *CommentHandlers) wsAddComment(ctx context.Context, msg *ws.Message) (*ws.Message, error) {
var req wsAddCommentRequest
if err := msg.ParsePayload(&req); err != nil {
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil)
}
if req.TaskID == "" {
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_id is required", nil)
}
if req.Content == "" {
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "content is required", nil)
}
comment, err := h.svc.CreateComment(ctx, &service.CreateCommentRequest{
TaskID: req.TaskID,
Content: req.Content,
AuthorType: "user",
AuthorID: req.AuthorID,
})
if err != nil {
h.logger.Error("failed to create comment", zap.Error(err))
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to create comment", nil)
}
// Auto-forward comment as prompt to running agent if orchestrator is available
// The agent's response will be saved as a comment via the prompt_complete event
if h.orchestrator != nil {
_, err := h.orchestrator.PromptTask(ctx, req.TaskID, req.Content)
if err != nil {
h.logger.Warn("failed to forward comment as prompt to agent",
zap.String("task_id", req.TaskID),
zap.Error(err))
}
}
return ws.NewResponse(msg.ID, msg.Action, comment)
}
type wsListCommentsRequest struct {
TaskID string `json:"task_id"`
}
func (h *CommentHandlers) wsListComments(ctx context.Context, msg *ws.Message) (*ws.Message, error) {
var req wsListCommentsRequest
if err := msg.ParsePayload(&req); err != nil {
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil)
}
if req.TaskID == "" {
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_id is required", nil)
}
comments, err := h.svc.ListComments(ctx, req.TaskID)
if err != nil {
h.logger.Error("failed to list comments", zap.Error(err))
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to list comments", nil)
}
return ws.NewResponse(msg.ID, msg.Action, comments)
}
apps/backend/kandev
BIN
View File
Binary file not shown.
apps/backend/pkg/api/v1/task.go
+27
View File
@@ -94,3 +94,30 @@ type CreateCommentRequest struct {
AuthorType string `json:"author_type,omitempty"` // Defaults to "user" if not specified
RequestsInput bool `json:"requests_input,omitempty"`
}
// PermissionOption represents a permission choice presented to the user
type PermissionOption struct {
OptionID string `json:"option_id"`
Name string `json:"name"`
Kind string `json:"kind"` // allow_once, allow_always, reject_once, reject_always
}
// PermissionRequest represents an agent's request for user permission
type PermissionRequest struct {
RequestID string `json:"request_id"` // Unique ID for this request (JSON-RPC ID)
TaskID string `json:"task_id"` // Task the agent is working on
InstanceID string `json:"instance_id"` // Agent instance ID
SessionID string `json:"session_id"` // ACP session ID
ToolCallID string `json:"tool_call_id"` // Tool call requesting permission
Title string `json:"title"` // Human-readable title
Description string `json:"description,omitempty"` // Additional context
Options []PermissionOption `json:"options"` // Available choices
CreatedAt time.Time `json:"created_at"`
}
// PermissionResponse represents the user's response to a permission request
type PermissionResponse struct {
RequestID string `json:"request_id" binding:"required"` // The request being responded to
OptionID string `json:"option_id,omitempty"` // Selected option (if not cancelled)
Cancelled bool `json:"cancelled,omitempty"` // True if user cancelled
}
apps/backend/pkg/websocket/actions.go
+4
View File
@@ -93,6 +93,10 @@ const (
ActionRepositoryScriptCreated = "repository.script.created"
ActionRepositoryScriptUpdated = "repository.script.updated"
ActionRepositoryScriptDeleted = "repository.script.deleted"
// Permission request actions (agent -> user -> agent)
ActionPermissionRequested = "permission.requested" // Agent requesting permission
ActionPermissionRespond = "permission.respond" // User responding to permission request
)
// Error codes
docs/WEBSOCKET_API.md
+718 -1
View File
@@ -127,9 +127,148 @@ All messages follow this JSON envelope format:
---
## Workspace Actions
Workspaces are the top-level organizational unit that contain boards and repositories.
### `workspace.list`
**Purpose:** List all workspaces.
**Request:**
```json
{
"id": "uuid",
"type": "request",
"action": "workspace.list",
"payload": {}
}
```
**Response:**
```json
{
"id": "uuid",
"type": "response",
"action": "workspace.list",
"payload": {
"workspaces": [
{
"id": "workspace-uuid",
"name": "My Project",
"description": "Main development workspace",
"owner_id": "user-uuid",
"created_at": "2026-01-10T12:00:00Z",
"updated_at": "2026-01-10T12:00:00Z"
}
]
}
}
```
### `workspace.create`
**Purpose:** Create a new workspace.
**Request:**
```json
{
"id": "uuid",
"type": "request",
"action": "workspace.create",
"payload": {
"name": "My Project",
"description": "Main development workspace",
"owner_id": "user-uuid"
}
}
```
| Payload Field | Type | Required | Description |
|--------------|------|----------|-------------|
| `name` | string | ✅ | Workspace name (max 255 chars) |
| `description` | string | ❌ | Workspace description |
| `owner_id` | string | ❌ | Owner user ID |
**Response:** Created workspace object
### `workspace.get`
**Purpose:** Get a workspace by ID.
**Request:**
```json
{
"id": "uuid",
"type": "request",
"action": "workspace.get",
"payload": {
"id": "workspace-uuid"
}
}
```
**Response:** Workspace object
### `workspace.update`
**Purpose:** Update a workspace.
**Request:**
```json
{
"id": "uuid",
"type": "request",
"action": "workspace.update",
"payload": {
"id": "workspace-uuid",
"name": "Updated Name",
"description": "Updated description"
}
}
```
| Payload Field | Type | Required | Description |
|--------------|------|----------|-------------|
| `id` | string | ✅ | Workspace ID |
| `name` | string | ❌ | New name |
| `description` | string | ❌ | New description |
**Response:** Updated workspace object
### `workspace.delete`
**Purpose:** Delete a workspace.
**Request:**
```json
{
"id": "uuid",
"type": "request",
"action": "workspace.delete",
"payload": {
"id": "workspace-uuid"
}
}
```
**Response:**
```json
{
"id": "uuid",
"type": "response",
"action": "workspace.delete",
"payload": {
"deleted": true
}
}
```
---
## Board Actions
Boards are the top-level containers that organize columns and tasks.
Boards are containers within workspaces that organize columns and tasks.
### `board.create`
@@ -427,6 +566,286 @@ Columns organize tasks within a board and represent workflow states.
---
## Repository Actions
Repositories represent Git repositories linked to a workspace.
### `repository.list`
**Purpose:** List all repositories in a workspace.
**Request:**
```json
{
"id": "uuid",
"type": "request",
"action": "repository.list",
"payload": {
"workspace_id": "workspace-uuid"
}
}
```
**Response:**
```json
{
"id": "uuid",
"type": "response",
"action": "repository.list",
"payload": {
"repositories": [
{
"id": "repo-uuid",
"workspace_id": "workspace-uuid",
"name": "my-project",
"source_type": "local",
"local_path": "/home/user/projects/my-project",
"provider": "",
"default_branch": "main",
"created_at": "2026-01-10T12:00:00Z",
"updated_at": "2026-01-10T12:00:00Z"
}
]
}
}
```
### `repository.create`
**Purpose:** Create a new repository link.
**Request:**
```json
{
"id": "uuid",
"type": "request",
"action": "repository.create",
"payload": {
"workspace_id": "workspace-uuid",
"name": "my-project",
"source_type": "local",
"local_path": "/home/user/projects/my-project",
"default_branch": "main",
"setup_script": "npm install",
"cleanup_script": "npm run clean"
}
}
```
| Payload Field | Type | Required | Description |
|--------------|------|----------|-------------|
| `workspace_id` | string | ✅ | Workspace ID |
| `name` | string | ✅ | Repository name |
| `source_type` | string | ❌ | `local` or `remote` |
| `local_path` | string | ❌ | Local filesystem path |
| `provider` | string | ❌ | Git provider (github, gitlab, etc.) |
| `provider_repo_id` | string | ❌ | Provider's repository ID |
| `provider_owner` | string | ❌ | Repository owner on provider |
| `provider_name` | string | ❌ | Repository name on provider |
| `default_branch` | string | ❌ | Default branch name |
| `setup_script` | string | ❌ | Script to run on setup |
| `cleanup_script` | string | ❌ | Script to run on cleanup |
**Response:** Created repository object
### `repository.get`
**Purpose:** Get a repository by ID.
**Request:**
```json
{
"id": "uuid",
"type": "request",
"action": "repository.get",
"payload": {
"id": "repo-uuid"
}
}
```
**Response:** Repository object
### `repository.update`
**Purpose:** Update a repository.
**Request:**
```json
{
"id": "uuid",
"type": "request",
"action": "repository.update",
"payload": {
"id": "repo-uuid",
"name": "updated-name",
"default_branch": "develop"
}
}
```
**Response:** Updated repository object
### `repository.delete`
**Purpose:** Delete a repository link.
**Request:**
```json
{
"id": "uuid",
"type": "request",
"action": "repository.delete",
"payload": {
"id": "repo-uuid"
}
}
```
**Response:**
```json
{
"id": "uuid",
"type": "response",
"action": "repository.delete",
"payload": {
"deleted": true
}
}
```
### `repository.script.list`
**Purpose:** List scripts for a repository.
**Request:**
```json
{
"id": "uuid",
"type": "request",
"action": "repository.script.list",
"payload": {
"repository_id": "repo-uuid"
}
}
```
**Response:**
```json
{
"id": "uuid",
"type": "response",
"action": "repository.script.list",
"payload": {
"scripts": [
{
"id": "script-uuid",
"repository_id": "repo-uuid",
"name": "build",
"command": "npm run build",
"position": 0
}
]
}
}
```
### `repository.script.create`
**Purpose:** Create a repository script.
**Request:**
```json
{
"id": "uuid",
"type": "request",
"action": "repository.script.create",
"payload": {
"repository_id": "repo-uuid",
"name": "test",
"command": "npm test",
"position": 1
}
}
```
| Payload Field | Type | Required | Description |
|--------------|------|----------|-------------|
| `repository_id` | string | ✅ | Repository ID |
| `name` | string | ✅ | Script name |
| `command` | string | ✅ | Command to execute |
| `position` | int | ❌ | Order position |
**Response:** Created script object
### `repository.script.get`
**Purpose:** Get a script by ID.
**Request:**
```json
{
"id": "uuid",
"type": "request",
"action": "repository.script.get",
"payload": {
"id": "script-uuid"
}
}
```
**Response:** Script object
### `repository.script.update`
**Purpose:** Update a script.
**Request:**
```json
{
"id": "uuid",
"type": "request",
"action": "repository.script.update",
"payload": {
"id": "script-uuid",
"name": "updated-name",
"command": "npm run updated-command"
}
}
```
**Response:** Updated script object
### `repository.script.delete`
**Purpose:** Delete a script.
**Request:**
```json
{
"id": "uuid",
"type": "request",
"action": "repository.script.delete",
"payload": {
"id": "script-uuid"
}
}
```
**Response:**
```json
{
"id": "uuid",
"type": "response",
"action": "repository.script.delete",
"payload": {
"deleted": true
}
}
```
---
## Task Actions
Tasks are work items that can be executed by AI agents.
@@ -757,6 +1176,106 @@ Subscriptions enable real-time streaming of agent output (ACP) to specific clien
---
## Comment Actions
Comments enable conversation between users and agents on tasks.
### `comment.add`
**Purpose:** Add a comment to a task. If an agent is running on the task, the comment is automatically forwarded as a prompt.
**Request:**
```json
{
"id": "uuid",
"type": "request",
"action": "comment.add",
"payload": {
"task_id": "task-uuid",
"content": "Please also add unit tests for the new function",
"author_id": "user-uuid"
}
}
```
| Payload Field | Type | Required | Description |
|--------------|------|----------|-------------|
| `task_id` | string | ✅ | Task ID |
| `content` | string | ✅ | Comment text |
| `author_id` | string | ❌ | Author user ID |
**Response:**
```json
{
"id": "uuid",
"type": "response",
"action": "comment.add",
"payload": {
"id": "comment-uuid",
"task_id": "task-uuid",
"author_type": "user",
"author_id": "user-uuid",
"content": "Please also add unit tests for the new function",
"requests_input": false,
"created_at": "2026-01-10T12:00:00Z"
}
}
```
### `comment.list`
**Purpose:** List all comments for a task.
**Request:**
```json
{
"id": "uuid",
"type": "request",
"action": "comment.list",
"payload": {
"task_id": "task-uuid"
}
}
```
**Response:**
```json
{
"id": "uuid",
"type": "response",
"action": "comment.list",
"payload": {
"comments": [
{
"id": "comment-uuid-1",
"task_id": "task-uuid",
"author_type": "user",
"author_id": "user-uuid",
"content": "Please implement the login feature",
"requests_input": false,
"created_at": "2026-01-10T12:00:00Z"
},
{
"id": "comment-uuid-2",
"task_id": "task-uuid",
"author_type": "agent",
"author_id": "agent-instance-uuid",
"content": "I've implemented the login feature. Should I also add password reset?",
"requests_input": true,
"created_at": "2026-01-10T12:05:00Z"
}
]
}
}
```
| Response Field | Type | Description |
|---------------|------|-------------|
| `author_type` | string | `user` or `agent` |
| `requests_input` | boolean | True if agent is asking for user input |
---
## Agent Actions
Agents are Docker containers running AI coding assistants.
@@ -1262,6 +1781,120 @@ The orchestrator coordinates task execution, managing the workflow between tasks
---
## Permission Actions
Agents may request user permission for certain operations. The permission flow uses notifications and responses.
### `permission.requested` (Notification)
**Purpose:** Server notifies client that an agent is requesting permission.
**When:** Agent calls a tool that requires user approval.
```json
{
"type": "notification",
"action": "permission.requested",
"payload": {
"task_id": "task-uuid",
"pending_id": "pending-request-uuid",
"instance_id": "agent-instance-uuid",
"session_id": "acp-session-uuid",
"tool_call_id": "tool-call-uuid",
"title": "Execute shell command",
"description": "Agent wants to run: npm install express",
"options": [
{
"option_id": "allow-once",
"name": "Allow Once",
"kind": "allow_once"
},
{
"option_id": "allow-always",
"name": "Always Allow",
"kind": "allow_always"
},
{
"option_id": "reject-once",
"name": "Deny",
"kind": "reject_once"
}
],
"created_at": "2026-01-10T12:00:00Z"
},
"timestamp": "2026-01-10T12:00:00Z"
}
```
| Payload Field | Type | Description |
|--------------|------|-------------|
| `task_id` | string | Task the agent is working on |
| `pending_id` | string | Unique ID for this pending request |
| `instance_id` | string | Agent instance ID |
| `session_id` | string | ACP session ID |
| `tool_call_id` | string | Tool call requesting permission |
| `title` | string | Human-readable title |
| `description` | string | Additional context |
| `options` | array | Available permission choices |
**Option Kinds:**
- `allow_once` - Allow this specific action
- `allow_always` - Allow this type of action permanently
- `reject_once` - Deny this specific action
- `reject_always` - Deny this type of action permanently
### `permission.respond`
**Purpose:** User responds to a permission request.
**Request:**
```json
{
"id": "uuid",
"type": "request",
"action": "permission.respond",
"payload": {
"task_id": "task-uuid",
"pending_id": "pending-request-uuid",
"option_id": "allow-once"
}
}
```
| Payload Field | Type | Required | Description |
|--------------|------|----------|-------------|
| `task_id` | string | ✅ | Task ID |
| `pending_id` | string | ✅ | Pending request ID from notification |
| `option_id` | string | ❌ | Selected option ID (required if not cancelled) |
| `cancelled` | boolean | ❌ | True to cancel the request |
**Response:**
```json
{
"id": "uuid",
"type": "response",
"action": "permission.respond",
"payload": {
"success": true
}
}
```
**Error Response (no pending request):**
```json
{
"id": "uuid",
"type": "error",
"action": "permission.respond",
"payload": {
"code": "NOT_FOUND",
"message": "No pending permission request for this task"
}
}
```
---
## Server Notifications (ACP Streaming)
These are push notifications from the server to subscribed clients. They have `type: "notification"` and no `id` field.
@@ -1403,6 +2036,34 @@ These are push notifications from the server to subscribed clients. They have `t
}
```
### `comment.added`
**Purpose:** New comment added to a task.
**When:** User or agent adds a comment.
```json
{
"type": "notification",
"action": "comment.added",
"payload": {
"task_id": "task-uuid",
"comment_id": "comment-uuid",
"author_type": "agent",
"author_id": "agent-instance-uuid",
"content": "I've completed the implementation. Would you like me to add tests?",
"requests_input": true,
"created_at": "2026-01-10T12:05:00Z"
},
"timestamp": "2026-01-10T12:05:00Z"
}
```
| Payload Field | Type | Description |
|--------------|------|-------------|
| `author_type` | string | `user` or `agent` |
| `requests_input` | boolean | True if agent is asking for user input |
### `task.updated`
**Purpose:** Task properties changed.
@@ -1522,17 +2183,41 @@ async function runTask() {
## Quick Reference
### Request Actions
| Action | Purpose |
|--------|---------|
| `health.check` | Server health status |
| **Workspace** | |
| `workspace.list` | List all workspaces |
| `workspace.create` | Create workspace |
| `workspace.get` | Get workspace by ID |
| `workspace.update` | Update workspace |
| `workspace.delete` | Delete workspace |
| **Board** | |
| `board.create` | Create board |
| `board.list` | List all boards |
| `board.get` | Get board by ID |
| `board.update` | Update board |
| `board.delete` | Delete board |
| **Column** | |
| `column.create` | Create column |
| `column.list` | List columns in board |
| `column.get` | Get column by ID |
| `column.update` | Update column |
| `column.delete` | Delete column |
| **Repository** | |
| `repository.list` | List repositories in workspace |
| `repository.create` | Create repository link |
| `repository.get` | Get repository by ID |
| `repository.update` | Update repository |
| `repository.delete` | Delete repository |
| `repository.script.list` | List repository scripts |
| `repository.script.create` | Create script |
| `repository.script.get` | Get script by ID |
| `repository.script.update` | Update script |
| `repository.script.delete` | Delete script |
| **Task** | |
| `task.create` | Create task |
| `task.list` | List tasks in board |
| `task.get` | Get task by ID |
@@ -1542,12 +2227,17 @@ async function runTask() {
| `task.state` | Change task state |
| `task.subscribe` | Subscribe to task notifications |
| `task.unsubscribe` | Unsubscribe from task |
| **Comment** | |
| `comment.add` | Add comment to task |
| `comment.list` | List comments for task |
| **Agent** | |
| `agent.list` | List agent instances |
| `agent.launch` | Launch agent (low-level) |
| `agent.status` | Get agent status |
| `agent.logs` | Get agent logs |
| `agent.stop` | Stop agent |
| `agent.types` | List available agent types |
| **Orchestrator** | |
| `orchestrator.status` | Get orchestrator status |
| `orchestrator.queue` | Get task queue |
| `orchestrator.trigger` | Queue task for execution |
@@ -1555,4 +2245,31 @@ async function runTask() {
| `orchestrator.stop` | Stop task execution |
| `orchestrator.prompt` | Send follow-up prompt |
| `orchestrator.complete` | Complete task |
| **Permission** | |
| `permission.respond` | Respond to permission request |
### Notification Actions (Server → Client)
| Action | Purpose |
|--------|---------|
| `acp.progress` | Agent progress update |
| `acp.log` | Agent log message |
| `acp.result` | Agent result artifact |
| `acp.error` | Agent error |
| `acp.status` | Agent status change |
| `acp.heartbeat` | Keep-alive heartbeat |
| `permission.requested` | Agent requesting permission |
| `comment.added` | New comment on task |
| `task.updated` | Task was updated |
| `task.created` | Task was created |
| `task.deleted` | Task was deleted |
| `workspace.created` | Workspace was created |
| `workspace.updated` | Workspace was updated |
| `workspace.deleted` | Workspace was deleted |
| `board.created` | Board was created |
| `board.updated` | Board was updated |
| `board.deleted` | Board was deleted |
| `column.created` | Column was created |
| `column.updated` | Column was updated |
| `column.deleted` | Column was deleted |
scripts/e2e-test.sh
+271 -612
View File
File diff suppressed because it is too large Load Diff