diff --git a/apps/backend/cmd/kandev/adapters.go b/apps/backend/cmd/kandev/adapters.go index 59f84f7ef0..cffa001566 100644 --- a/apps/backend/cmd/kandev/adapters.go +++ b/apps/backend/cmd/kandev/adapters.go @@ -214,8 +214,8 @@ type orchestratorAdapter struct { } // PromptTask forwards to the orchestrator service and converts the result type -func (a *orchestratorAdapter) PromptTask(ctx context.Context, taskID, taskSessionID, prompt, model string) (*taskhandlers.PromptResult, error) { - result, err := a.svc.PromptTask(ctx, taskID, taskSessionID, prompt, model) +func (a *orchestratorAdapter) PromptTask(ctx context.Context, taskID, taskSessionID, prompt, model string, planMode bool) (*taskhandlers.PromptResult, error) { + result, err := a.svc.PromptTask(ctx, taskID, taskSessionID, prompt, model, planMode) if err != nil { return nil, err } diff --git a/apps/backend/cmd/kandev/agents.go b/apps/backend/cmd/kandev/agents.go index 0d11cdcaf2..76614a4d72 100644 --- a/apps/backend/cmd/kandev/agents.go +++ b/apps/backend/cmd/kandev/agents.go @@ -47,7 +47,7 @@ func provideLifecycleManager( credsMgr.AddProvider(credentials.NewFileProvider(credsFile)) } - profileResolver := lifecycle.NewStoreProfileResolver(agentSettingsRepo) + profileResolver := lifecycle.NewStoreProfileResolver(agentSettingsRepo, agentRegistry) var agentRuntime lifecycle.Runtime var containerMgr *lifecycle.ContainerManager diff --git a/apps/backend/internal/agent/lifecycle/profile_resolver.go b/apps/backend/internal/agent/lifecycle/profile_resolver.go index b71d00cdc3..f997be6fe8 100644 --- a/apps/backend/internal/agent/lifecycle/profile_resolver.go +++ b/apps/backend/internal/agent/lifecycle/profile_resolver.go @@ -4,17 +4,20 @@ import ( "context" "fmt" + "github.com/kandev/kandev/internal/agent/registry" "github.com/kandev/kandev/internal/agent/settings/store" ) // StoreProfileResolver implements ProfileResolver using the agent settings store type StoreProfileResolver struct { - store store.Repository + store store.Repository + registry *registry.Registry } -// NewStoreProfileResolver creates a new profile resolver using the given store -func NewStoreProfileResolver(store store.Repository) *StoreProfileResolver { - return &StoreProfileResolver{store: store} +// NewStoreProfileResolver creates a new profile resolver using the given store and registry. +// The registry is used to look up the agent's default model when the profile doesn't specify one. +func NewStoreProfileResolver(store store.Repository, reg *registry.Registry) *StoreProfileResolver { + return &StoreProfileResolver{store: store, registry: reg} } // ResolveProfile looks up an agent profile by ID and returns the profile info @@ -31,12 +34,21 @@ func (r *StoreProfileResolver) ResolveProfile(ctx context.Context, profileID str return nil, fmt.Errorf("agent not found for profile: %w", err) } + // Determine the model to use: profile's model, or fall back to agent's default model + model := profile.Model + if model == "" && r.registry != nil { + // Look up the agent type in the registry to get the default model + if agentType, ok := r.registry.Get(agent.Name); ok { + model = agentType.ModelConfig.DefaultModel + } + } + return &AgentProfileInfo{ ProfileID: profile.ID, ProfileName: profile.Name, AgentID: agent.ID, AgentName: agent.Name, - Model: profile.Model, + Model: model, AutoApprove: profile.AutoApprove, DangerouslySkipPermissions: profile.DangerouslySkipPermissions, AllowIndexing: profile.AllowIndexing, diff --git a/apps/backend/internal/agent/lifecycle/profile_resolver_test.go b/apps/backend/internal/agent/lifecycle/profile_resolver_test.go index fe0e43092d..f0871e6bae 100644 --- a/apps/backend/internal/agent/lifecycle/profile_resolver_test.go +++ b/apps/backend/internal/agent/lifecycle/profile_resolver_test.go @@ -5,8 +5,10 @@ import ( "errors" "testing" + "github.com/kandev/kandev/internal/agent/registry" "github.com/kandev/kandev/internal/agent/settings/models" "github.com/kandev/kandev/internal/agent/settings/store" + "github.com/kandev/kandev/internal/common/logger" ) // MockRepository implements store.Repository for testing @@ -94,7 +96,7 @@ func (m *MockRepository) Close() error { func TestNewStoreProfileResolver(t *testing.T) { mockRepo := &MockRepository{} - resolver := NewStoreProfileResolver(mockRepo) + resolver := NewStoreProfileResolver(mockRepo, nil) if resolver == nil { t.Fatal("expected non-nil resolver") @@ -125,7 +127,7 @@ func TestStoreProfileResolver_ResolveProfile_Success(t *testing.T) { }, } - resolver := NewStoreProfileResolver(mockRepo) + resolver := NewStoreProfileResolver(mockRepo, nil) ctx := context.Background() info, err := resolver.ResolveProfile(ctx, "profile-123") @@ -169,7 +171,7 @@ func TestStoreProfileResolver_ResolveProfile_ProfileNotFound(t *testing.T) { }, } - resolver := NewStoreProfileResolver(mockRepo) + resolver := NewStoreProfileResolver(mockRepo, nil) ctx := context.Background() info, err := resolver.ResolveProfile(ctx, "non-existent-profile") @@ -205,7 +207,7 @@ func TestStoreProfileResolver_ResolveProfile_AgentNotFound(t *testing.T) { }, } - resolver := NewStoreProfileResolver(mockRepo) + resolver := NewStoreProfileResolver(mockRepo, nil) ctx := context.Background() info, err := resolver.ResolveProfile(ctx, "profile-123") @@ -222,3 +224,95 @@ func TestStoreProfileResolver_ResolveProfile_AgentNotFound(t *testing.T) { t.Errorf("expected error message to start with '%s', got '%s'", expectedMsg, err.Error()) } } + +func TestStoreProfileResolver_ResolveProfile_FallbackToRegistryDefaultModel(t *testing.T) { + mockRepo := &MockRepository{ + GetAgentProfileFn: func(ctx context.Context, id string) (*models.AgentProfile, error) { + return &models.AgentProfile{ + ID: "profile-123", + AgentID: "agent-456", + Name: "Default Profile", + Model: "", // Empty model - should fallback to registry + AutoApprove: false, + }, nil + }, + GetAgentFn: func(ctx context.Context, id string) (*models.Agent, error) { + return &models.Agent{ + ID: "agent-456", + Name: "claude-code", // Agent name matches registry key + }, nil + }, + } + + // Create a registry with a default model for claude-code + log, _ := logger.NewLogger(logger.LoggingConfig{Level: "error", Format: "console"}) + reg := registry.NewRegistry(log) + err := reg.Register(®istry.AgentTypeConfig{ + ID: "claude-code", + Name: "claude-code", + Cmd: []string{"claude"}, // Standalone agent uses Cmd + ResourceLimits: registry.ResourceLimits{ + MemoryMB: 1024, + CPUCores: 1, + TimeoutSeconds: 3600, + }, + ModelConfig: registry.ModelConfig{ + DefaultModel: "claude-sonnet-4-20250514", + }, + }) + if err != nil { + t.Fatalf("failed to register agent: %v", err) + } + + resolver := NewStoreProfileResolver(mockRepo, reg) + ctx := context.Background() + + info, err := resolver.ResolveProfile(ctx, "profile-123") + + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if info == nil { + t.Fatal("expected non-nil profile info") + } + // Should have fallback to registry's default model + if info.Model != "claude-sonnet-4-20250514" { + t.Errorf("expected Model 'claude-sonnet-4-20250514' (from registry), got '%s'", info.Model) + } +} + +func TestStoreProfileResolver_ResolveProfile_EmptyModelNoRegistry(t *testing.T) { + mockRepo := &MockRepository{ + GetAgentProfileFn: func(ctx context.Context, id string) (*models.AgentProfile, error) { + return &models.AgentProfile{ + ID: "profile-123", + AgentID: "agent-456", + Name: "Default Profile", + Model: "", // Empty model + }, nil + }, + GetAgentFn: func(ctx context.Context, id string) (*models.Agent, error) { + return &models.Agent{ + ID: "agent-456", + Name: "custom-agent", + }, nil + }, + } + + // No registry provided + resolver := NewStoreProfileResolver(mockRepo, nil) + ctx := context.Background() + + info, err := resolver.ResolveProfile(ctx, "profile-123") + + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if info == nil { + t.Fatal("expected non-nil profile info") + } + // Model should remain empty since no registry fallback available + if info.Model != "" { + t.Errorf("expected empty Model when no registry fallback, got '%s'", info.Model) + } +} diff --git a/apps/backend/internal/gateway/websocket/session_notifications.go b/apps/backend/internal/gateway/websocket/session_notifications.go index 6a167250ed..e3c2dd904f 100644 --- a/apps/backend/internal/gateway/websocket/session_notifications.go +++ b/apps/backend/internal/gateway/websocket/session_notifications.go @@ -2,6 +2,8 @@ package websocket import ( "context" + "sync" + "time" "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/internal/events" @@ -10,16 +12,29 @@ import ( "go.uber.org/zap" ) +// fileChangeDebounceWindow is the time to wait before flushing batched file changes +const fileChangeDebounceWindow = 100 * time.Millisecond + +// maxFileChangeBatchSize is the maximum number of file changes to batch before forcing a flush +const maxFileChangeBatchSize = 50 + type SessionStreamBroadcaster struct { hub *Hub subscriptions []bus.Subscription logger *logger.Logger + + // File change debouncing + fileChangeMu sync.Mutex + fileChangeBatch map[string][]any // sessionID -> list of file change payloads + fileChangeTimers map[string]*time.Timer // sessionID -> debounce timer } func RegisterSessionStreamNotifications(ctx context.Context, eventBus bus.EventBus, hub *Hub, log *logger.Logger) *SessionStreamBroadcaster { b := &SessionStreamBroadcaster{ - hub: hub, - logger: log.WithFields(zap.String("component", "ws-session-stream-broadcaster")), + hub: hub, + logger: log.WithFields(zap.String("component", "ws-session-stream-broadcaster")), + fileChangeBatch: make(map[string][]any), + fileChangeTimers: make(map[string]*time.Timer), } if eventBus == nil { return b @@ -28,7 +43,8 @@ func RegisterSessionStreamNotifications(ctx context.Context, eventBus bus.EventB b.subscribe(eventBus, events.BuildGitStatusWildcardSubject(), ws.ActionSessionGitStatus) b.subscribe(eventBus, events.BuildGitSnapshotWildcardSubject(), ws.ActionSessionGitSnapshot) b.subscribe(eventBus, events.BuildGitCommitRecordedWildcardSubject(), ws.ActionSessionGitCommit) - b.subscribe(eventBus, events.BuildFileChangeWildcardSubject(), ws.ActionWorkspaceFileChanges) + // Use debounced subscription for file changes to batch high-frequency events + b.subscribeFileChanges(eventBus, events.BuildFileChangeWildcardSubject(), ws.ActionWorkspaceFileChanges) b.subscribe(eventBus, events.BuildShellOutputWildcardSubject(), ws.ActionSessionShellOutput) b.subscribe(eventBus, events.BuildShellExitWildcardSubject(), ws.ActionSessionShellOutput) b.subscribe(eventBus, events.BuildProcessOutputWildcardSubject(), ws.ActionSessionProcessOutput) @@ -49,6 +65,15 @@ func (b *SessionStreamBroadcaster) Close() { } } b.subscriptions = nil + + // Stop all debounce timers + b.fileChangeMu.Lock() + for _, timer := range b.fileChangeTimers { + timer.Stop() + } + b.fileChangeTimers = nil + b.fileChangeBatch = nil + b.fileChangeMu.Unlock() } func (b *SessionStreamBroadcaster) subscribe(eventBus bus.EventBus, subject, action string) { @@ -72,14 +97,86 @@ func (b *SessionStreamBroadcaster) subscribe(eventBus bus.EventBus, subject, act b.subscriptions = append(b.subscriptions, sub) } -func extractSessionID(data interface{}) string { +// subscribeFileChanges subscribes to file change events with debouncing. +// Multiple file changes within the debounce window are batched into a single message. +func (b *SessionStreamBroadcaster) subscribeFileChanges(eventBus bus.EventBus, subject, action string) { + sub, err := eventBus.Subscribe(subject, func(ctx context.Context, event *bus.Event) error { + sessionID := extractSessionID(event.Data) + if sessionID == "" { + return nil + } + + b.fileChangeMu.Lock() + defer b.fileChangeMu.Unlock() + + // Add to batch + b.fileChangeBatch[sessionID] = append(b.fileChangeBatch[sessionID], event.Data) + + // Force flush if batch is too large + if len(b.fileChangeBatch[sessionID]) >= maxFileChangeBatchSize { + b.flushFileChangesLocked(sessionID, action) + return nil + } + + // Reset or create debounce timer + if timer, exists := b.fileChangeTimers[sessionID]; exists { + timer.Stop() + } + b.fileChangeTimers[sessionID] = time.AfterFunc(fileChangeDebounceWindow, func() { + b.fileChangeMu.Lock() + defer b.fileChangeMu.Unlock() + b.flushFileChangesLocked(sessionID, action) + }) + + return nil + }) + if err != nil { + b.logger.Error("failed to subscribe to file change events", zap.String("subject", subject), zap.Error(err)) + return + } + b.subscriptions = append(b.subscriptions, sub) +} + +// flushFileChangesLocked sends batched file changes. Must be called with fileChangeMu held. +func (b *SessionStreamBroadcaster) flushFileChangesLocked(sessionID, action string) { + batch := b.fileChangeBatch[sessionID] + if len(batch) == 0 { + return + } + + // Clear the batch + delete(b.fileChangeBatch, sessionID) + if timer, exists := b.fileChangeTimers[sessionID]; exists { + timer.Stop() + delete(b.fileChangeTimers, sessionID) + } + + // Create a single notification with all changes + payload := map[string]any{ + "session_id": sessionID, + "changes": batch, + } + + msg, err := ws.NewNotification(action, payload) + if err != nil { + b.logger.Error("failed to build batched file change notification", zap.String("action", action), zap.Error(err)) + return + } + + b.hub.BroadcastToSession(sessionID, msg) + b.logger.Debug("flushed batched file changes", + zap.String("session_id", sessionID), + zap.Int("count", len(batch))) +} + +func extractSessionID(data any) string { if data == nil { return "" } if typed, ok := data.(interface{ GetSessionID() string }); ok { return typed.GetSessionID() } - if m, ok := data.(map[string]interface{}); ok { + if m, ok := data.(map[string]any); ok { if sessionID, ok := m["session_id"].(string); ok { return sessionID } diff --git a/apps/backend/internal/gateway/websocket/session_notifications_test.go b/apps/backend/internal/gateway/websocket/session_notifications_test.go new file mode 100644 index 0000000000..fef5beb569 --- /dev/null +++ b/apps/backend/internal/gateway/websocket/session_notifications_test.go @@ -0,0 +1,153 @@ +package websocket + +import ( + "testing" + "time" +) + +func TestSessionStreamBroadcaster_FileChangeBatching(t *testing.T) { + // Test that multiple file changes are batched correctly + b := &SessionStreamBroadcaster{ + fileChangeBatch: make(map[string][]any), + fileChangeTimers: make(map[string]*time.Timer), + } + + sessionID := "test-session-123" + + // Add multiple file changes + for i := range 5 { + b.fileChangeMu.Lock() + b.fileChangeBatch[sessionID] = append(b.fileChangeBatch[sessionID], map[string]any{ + "path": "/test/file" + string(rune('0'+i)) + ".txt", + "operation": "write", + "session_id": sessionID, + }) + b.fileChangeMu.Unlock() + } + + // Check batch size + b.fileChangeMu.Lock() + batchLen := len(b.fileChangeBatch[sessionID]) + b.fileChangeMu.Unlock() + + if batchLen != 5 { + t.Errorf("expected batch size 5, got %d", batchLen) + } + + // Verify batch contents + b.fileChangeMu.Lock() + batch := b.fileChangeBatch[sessionID] + for i, item := range batch { + m, ok := item.(map[string]any) + if !ok { + t.Errorf("batch item %d is not a map", i) + continue + } + if m["session_id"] != sessionID { + t.Errorf("batch item %d has wrong session_id", i) + } + } + b.fileChangeMu.Unlock() +} + +func TestSessionStreamBroadcaster_MaxBatchSize(t *testing.T) { + // Test that batch reaches max size correctly + b := &SessionStreamBroadcaster{ + fileChangeBatch: make(map[string][]any), + fileChangeTimers: make(map[string]*time.Timer), + } + + sessionID := "test-session-456" + + // Add maxFileChangeBatchSize items + b.fileChangeMu.Lock() + for range maxFileChangeBatchSize { + b.fileChangeBatch[sessionID] = append(b.fileChangeBatch[sessionID], map[string]any{ + "path": "/test/file.txt", + }) + } + batchLen := len(b.fileChangeBatch[sessionID]) + b.fileChangeMu.Unlock() + + if batchLen != maxFileChangeBatchSize { + t.Errorf("expected batch size %d, got %d", maxFileChangeBatchSize, batchLen) + } + + // Verify the constant is set correctly + if maxFileChangeBatchSize != 50 { + t.Errorf("expected maxFileChangeBatchSize to be 50, got %d", maxFileChangeBatchSize) + } +} + +func TestSessionStreamBroadcaster_DebounceWindowConstant(t *testing.T) { + // Verify debounce window is set to expected value + expectedWindow := 100 * time.Millisecond + if fileChangeDebounceWindow != expectedWindow { + t.Errorf("expected fileChangeDebounceWindow to be %v, got %v", expectedWindow, fileChangeDebounceWindow) + } +} + +func TestSessionStreamBroadcaster_Close(t *testing.T) { + b := &SessionStreamBroadcaster{ + fileChangeBatch: make(map[string][]any), + fileChangeTimers: make(map[string]*time.Timer), + } + + // Add a timer + sessionID := "test-session-789" + b.fileChangeTimers[sessionID] = time.AfterFunc(time.Hour, func() {}) + b.fileChangeBatch[sessionID] = []any{map[string]any{"path": "/test.txt"}} + + // Close should clean up + b.Close() + + if b.fileChangeBatch != nil { + t.Error("expected fileChangeBatch to be nil after Close") + } + if b.fileChangeTimers != nil { + t.Error("expected fileChangeTimers to be nil after Close") + } +} + +func TestExtractSessionID(t *testing.T) { + tests := []struct { + name string + data any + expected string + }{ + { + name: "nil data", + data: nil, + expected: "", + }, + { + name: "map with session_id", + data: map[string]any{ + "session_id": "session-123", + "path": "/test.txt", + }, + expected: "session-123", + }, + { + name: "map without session_id", + data: map[string]any{ + "path": "/test.txt", + }, + expected: "", + }, + { + name: "non-map type", + data: "string value", + expected: "", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + result := extractSessionID(tt.data) + if result != tt.expected { + t.Errorf("extractSessionID(%v) = %q, want %q", tt.data, result, tt.expected) + } + }) + } +} diff --git a/apps/backend/internal/orchestrator/controller/controller.go b/apps/backend/internal/orchestrator/controller/controller.go index daf170a18d..57d240b6e9 100644 --- a/apps/backend/internal/orchestrator/controller/controller.go +++ b/apps/backend/internal/orchestrator/controller/controller.go @@ -139,7 +139,7 @@ func (c *Controller) StopTask(ctx context.Context, req dto.StopTaskRequest) (dto // PromptTask sends a prompt to a running task func (c *Controller) PromptTask(ctx context.Context, req dto.PromptTaskRequest) (dto.PromptTaskResponse, error) { - result, err := c.service.PromptTask(ctx, req.TaskID, req.TaskSessionID, req.Prompt, req.Model) + result, err := c.service.PromptTask(ctx, req.TaskID, req.TaskSessionID, req.Prompt, req.Model, req.PlanMode) if err != nil { return dto.PromptTaskResponse{}, err } diff --git a/apps/backend/internal/orchestrator/dto/dto.go b/apps/backend/internal/orchestrator/dto/dto.go index 9fcc6c93f0..94800a0c55 100644 --- a/apps/backend/internal/orchestrator/dto/dto.go +++ b/apps/backend/internal/orchestrator/dto/dto.go @@ -123,7 +123,8 @@ type PromptTaskRequest struct { TaskID string `json:"task_id"` TaskSessionID string `json:"session_id"` Prompt string `json:"prompt"` - Model string `json:"model,omitempty"` // Optional: switch to this model before processing prompt + Model string `json:"model,omitempty"` // Optional: switch to this model before processing prompt + PlanMode bool `json:"plan_mode,omitempty"` // Optional: enable plan mode (analyze before executing) } // PromptTaskResponse is the response for orchestrator.prompt diff --git a/apps/backend/internal/orchestrator/event_handlers.go b/apps/backend/internal/orchestrator/event_handlers.go index 67c0e6374f..2b22442904 100644 --- a/apps/backend/internal/orchestrator/event_handlers.go +++ b/apps/backend/internal/orchestrator/event_handlers.go @@ -408,11 +408,12 @@ func (s *Service) updateTaskSessionState(ctx context.Context, taskID, sessionID zap.String("new_state", string(nextState))) if s.eventBus != nil { _ = s.eventBus.Publish(ctx, events.TaskSessionStateChanged, bus.NewEvent(events.TaskSessionStateChanged, "task-session", map[string]interface{}{ - "task_id": taskID, - "session_id": sessionID, - "old_state": string(oldState), - "new_state": string(nextState), - "agent_profile_id": session.AgentProfileID, + "task_id": taskID, + "session_id": sessionID, + "old_state": string(oldState), + "new_state": string(nextState), + "agent_profile_id": session.AgentProfileID, + "agent_profile_snapshot": session.AgentProfileSnapshot, })) } } @@ -603,10 +604,11 @@ func (s *Service) handleContextWindowUpdated(ctx context.Context, data watcher.C events.TaskSessionStateChanged, "orchestrator", map[string]interface{}{ - "task_id": data.TaskID, - "session_id": session.ID, - "new_state": string(session.State), - "agent_profile_id": session.AgentProfileID, + "task_id": data.TaskID, + "session_id": session.ID, + "new_state": string(session.State), + "agent_profile_id": session.AgentProfileID, + "agent_profile_snapshot": session.AgentProfileSnapshot, "metadata": map[string]interface{}{ "context_window": contextWindowData, }, diff --git a/apps/backend/internal/orchestrator/task_operations.go b/apps/backend/internal/orchestrator/task_operations.go index 8755d7d56c..c83d734925 100644 --- a/apps/backend/internal/orchestrator/task_operations.go +++ b/apps/backend/internal/orchestrator/task_operations.go @@ -237,16 +237,29 @@ func (s *Service) StopSession(ctx context.Context, sessionID string, reason stri return s.executor.Stop(ctx, sessionID, reason, force) } +// PlanModePrefix is prepended to prompts when plan mode is enabled. +// This instructs agents to analyze and plan before making changes. +const PlanModePrefix = "[PLAN MODE: Analyze and plan before making changes. Do not execute yet.]\n\n" + // PromptTask sends a follow-up prompt to a running agent for a task session. -func (s *Service) PromptTask(ctx context.Context, taskID, sessionID string, prompt string, model string) (*PromptResult, error) { - s.logger.Debug("sending prompt to task agent", +// If planMode is true, a plan mode prefix is prepended to the prompt. +func (s *Service) PromptTask(ctx context.Context, taskID, sessionID string, prompt string, model string, planMode bool) (*PromptResult, error) { + s.logger.Debug("PromptTask called", zap.String("task_id", taskID), zap.String("session_id", sessionID), - zap.Int("prompt_length", len(prompt))) + zap.Int("prompt_length", len(prompt)), + zap.String("requested_model", model), + zap.Bool("plan_mode", planMode)) if sessionID == "" { return nil, fmt.Errorf("session_id is required") } + // Apply plan mode prefix if enabled + effectivePrompt := prompt + if planMode { + effectivePrompt = PlanModePrefix + prompt + } + // Check if model switching is requested if model != "" { // Get the current session to check current model @@ -255,30 +268,34 @@ func (s *Service) PromptTask(ctx context.Context, taskID, sessionID string, prom return nil, fmt.Errorf("failed to get session for model switch: %w", err) } - // Get current model - handle nil snapshot and missing key safely - currentModel := "" + // Get current model from agent profile snapshot + var currentModel string if session.AgentProfileSnapshot != nil { if m, ok := session.AgentProfileSnapshot["model"].(string); ok { currentModel = m } } - // Trigger switch if models differ OR if current model is unknown (empty string) + // Trigger switch if models differ if currentModel != model { - s.logger.Info("model switch requested", + s.logger.Info("switching model", zap.String("task_id", taskID), zap.String("session_id", sessionID), - zap.String("current_model", currentModel), - zap.String("new_model", model)) + zap.String("from", currentModel), + zap.String("to", model)) // Start a new turn for this prompt (model switch case) s.startTurnForSession(ctx, sessionID) // SwitchModel will stop agent, rebuild with new model, restart, and send prompt - switchResult, err := s.executor.SwitchModel(ctx, taskID, sessionID, model, prompt) + switchResult, err := s.executor.SwitchModel(ctx, taskID, sessionID, model, effectivePrompt) if err != nil { return nil, fmt.Errorf("model switch failed: %w", err) } + + // Update session state to RUNNING and publish event so frontend knows agent is active + s.setSessionRunning(ctx, taskID, sessionID) + return &PromptResult{ StopReason: switchResult.StopReason, AgentMessage: switchResult.AgentMessage, @@ -288,10 +305,13 @@ func (s *Service) PromptTask(ctx context.Context, taskID, sessionID string, prom } s.setSessionRunning(ctx, taskID, sessionID) - // Start a new turn for this prompt s.startTurnForSession(ctx, sessionID) - result, err := s.executor.Prompt(ctx, taskID, sessionID, prompt) + result, err := s.executor.Prompt(ctx, taskID, sessionID, effectivePrompt) if err != nil { + s.logger.Error("prompt failed", + zap.String("task_id", taskID), + zap.String("session_id", sessionID), + zap.Error(err)) return nil, err } return &PromptResult{ diff --git a/apps/backend/internal/task/handlers/message_handlers.go b/apps/backend/internal/task/handlers/message_handlers.go index cbd63f766b..b6ff6aa81b 100644 --- a/apps/backend/internal/task/handlers/message_handlers.go +++ b/apps/backend/internal/task/handlers/message_handlers.go @@ -27,7 +27,7 @@ type PromptResult struct { // OrchestratorService defines the interface for orchestrator operations type OrchestratorService interface { - PromptTask(ctx context.Context, taskID, sessionID, prompt, model string) (*PromptResult, error) + PromptTask(ctx context.Context, taskID, sessionID, prompt, model string, planMode bool) (*PromptResult, error) ResumeTaskSession(ctx context.Context, taskID, taskSessionID string) error } @@ -124,6 +124,7 @@ type wsAddMessageRequest struct { Content string `json:"content"` AuthorID string `json:"author_id,omitempty"` Model string `json:"model,omitempty"` + PlanMode bool `json:"plan_mode,omitempty"` } func (h *MessageHandlers) wsAddMessage(ctx context.Context, msg *ws.Message) (*ws.Message, error) { @@ -186,9 +187,10 @@ func (h *MessageHandlers) wsAddMessage(ctx context.Context, msg *ws.Message) (*w sessionID := req.TaskSessionID content := req.Content model := req.Model + planMode := req.PlanMode go func() { promptCtx := context.WithoutCancel(ctx) - _, err := h.orchestrator.PromptTask(promptCtx, taskID, sessionID, content, model) + _, err := h.orchestrator.PromptTask(promptCtx, taskID, sessionID, content, model, planMode) if err != nil { if errors.Is(err, executor.ErrExecutionNotFound) { if resumeErr := h.orchestrator.ResumeTaskSession(promptCtx, taskID, sessionID); resumeErr != nil { @@ -199,7 +201,7 @@ func (h *MessageHandlers) wsAddMessage(ctx context.Context, msg *ws.Message) (*w } else { for attempt := 0; attempt < 3; attempt++ { time.Sleep(500 * time.Millisecond) - _, err = h.orchestrator.PromptTask(promptCtx, taskID, sessionID, content, model) + _, err = h.orchestrator.PromptTask(promptCtx, taskID, sessionID, content, model, planMode) if err == nil { break } diff --git a/apps/web/app/task/[id]/page-client.tsx b/apps/web/app/task/[id]/page-client.tsx index e7e7583501..126b71354c 100644 --- a/apps/web/app/task/[id]/page-client.tsx +++ b/apps/web/app/task/[id]/page-client.tsx @@ -122,22 +122,44 @@ export default function TaskPage({ }, []); useEffect(() => { + console.log('[TaskPage] Active session setup effect', { + initialTaskId: initialTask?.id, + activeTaskId, + initialSessionId, + sessionIdProp: sessionId, + taskSessionId, + }); if (!initialTask?.id) return; - if (activeTaskId && activeTaskId !== initialTask.id) return; + if (activeTaskId && activeTaskId !== initialTask.id) { + console.log('[TaskPage] Skipping - different activeTaskId already set'); + return; + } if (initialSessionId) { + console.log('[TaskPage] Setting active session', initialTask.id, initialSessionId); setActiveSession(initialTask.id, initialSessionId); } else { + console.log('[TaskPage] Setting active task', initialTask.id); setActiveTask(initialTask.id); } - }, [activeTaskId, initialSessionId, initialTask?.id, setActiveSession, setActiveTask]); + }, [activeTaskId, initialSessionId, initialTask?.id, sessionId, taskSessionId, setActiveSession, setActiveTask]); useEffect(() => { + console.log('[TaskPage] Task fetch effect', { + activeTaskId, + initialTaskId: initialTask?.id, + taskDetailsId: taskDetails?.id, + shouldFetch: activeTaskId && taskDetails?.id !== activeTaskId, + }); if (!activeTaskId) return; if (taskDetails?.id === activeTaskId) return; + console.log('[TaskPage] Fetching task details for', activeTaskId); fetchTask(activeTaskId, { cache: 'no-store' }) - .then((response) => setTaskDetails(response)) - .catch((error) => console.error('Failed to load task details:', error)); - }, [activeTaskId, taskDetails?.id]); + .then((response) => { + console.log('[TaskPage] Fetched task details', response?.id); + setTaskDetails(response); + }) + .catch((error) => console.error('[TaskPage] Failed to load task details:', error)); + }, [activeTaskId, initialTask?.id, taskDetails?.id]); const { repositories } = useRepositories(task?.workspace_id ?? null, Boolean(task?.workspace_id)); const effectiveRepositories = repositories.length ? repositories : initialRepositories; const repository = useMemo( diff --git a/apps/web/components/task/model-selector.tsx b/apps/web/components/task/model-selector.tsx index 21018686f2..721c8d0b0b 100644 --- a/apps/web/components/task/model-selector.tsx +++ b/apps/web/components/task/model-selector.tsx @@ -5,7 +5,6 @@ import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from '@ import { useAppStore } from '@/components/state-provider'; import { useAvailableAgents } from '@/hooks/domains/settings/use-available-agents'; import { useSettingsData } from '@/hooks/domains/settings/use-settings-data'; -import { cn } from '@/lib/utils'; import type { Agent, AgentProfile, AvailableAgent } from '@/lib/types/http'; type ModelSelectorProps = { @@ -18,9 +17,8 @@ export function ModelSelector({ sessionId }: ModelSelectorProps) { const settingsAgents = useAppStore((state) => state.settingsAgents.items); const taskSessions = useAppStore((state) => state.taskSessions.items); - const pendingModels = useAppStore((state) => state.pendingModel.bySessionId); const activeModels = useAppStore((state) => state.activeModel.bySessionId); - const setPendingModel = useAppStore((state) => state.setPendingModel); + const setActiveModel = useAppStore((state) => state.setActiveModel); // Ensure available agents are loaded (contains model_config) // Note: We don't block on this loading - it only affects the dropdown options @@ -49,6 +47,7 @@ export function ModelSelector({ sessionId }: ModelSelectorProps) { } // Get available models from the agent discovery data + // Agent.name is the registry ID (e.g., "claude-code") which matches AvailableAgent.name let availableModels: { id: string; name: string; provider: string; context_window: number; is_default: boolean }[] = []; if (sessionProfile?.agent) { const agentName = sessionProfile.agent.name; @@ -58,13 +57,9 @@ export function ModelSelector({ sessionId }: ModelSelectorProps) { } } - // Priority: activeModel (after switch) > snapshot model - // Profile model fallback removed - all profiles now have models + // Priority: activeModel (user selection) > snapshot model (from backend) const activeModel = sessionId ? (activeModels[sessionId] || null) : null; const currentModel = activeModel || snapshotModelStr; - const pendingModel = sessionId ? (pendingModels[sessionId] || null) : null; - const displayedModel = pendingModel || currentModel || ''; - const hasPendingChange = pendingModel && pendingModel !== currentModel; // Ensure current model is included in the list (even if not in available models) const modelOptions = [...availableModels]; @@ -80,7 +75,8 @@ export function ModelSelector({ sessionId }: ModelSelectorProps) { const handleModelChange = (modelId: string) => { if (!sessionId) return; - setPendingModel(sessionId, modelId); + // Immediately update the active model - the actual switch happens on next message + setActiveModel(sessionId, modelId); }; // If no session or no model available, show disabled placeholder @@ -95,19 +91,11 @@ export function ModelSelector({ sessionId }: ModelSelectorProps) { } return ( - +
- {hasPendingChange && ( - pending - )}
diff --git a/apps/web/components/task/task-chat-panel.tsx b/apps/web/components/task/task-chat-panel.tsx index 7f5f4caca9..87eeabd9f2 100644 --- a/apps/web/components/task/task-chat-panel.tsx +++ b/apps/web/components/task/task-chat-panel.tsx @@ -56,7 +56,7 @@ export function TaskChatPanel({ ); // Model management - const { sessionModel, pendingModel, clearPendingModel, setActiveModel } = useSessionModel( + const { sessionModel, activeModel } = useSessionModel( resolvedSessionId, session?.agent_profile_id ); @@ -66,9 +66,8 @@ export function TaskChatPanel({ resolvedSessionId, taskId, sessionModel, - pendingModel, - setActiveModel, - clearPendingModel + activeModel, + planModeEnabled ); // Clear awaiting state when a new agent message arrives @@ -104,6 +103,10 @@ export function TaskChatPanel({ } else { await handleSendMessage(trimmed); } + // Reset plan mode after sending - it's a per-message instruction + if (planModeEnabled) { + setPlanModeEnabled(false); + } } finally { setIsSending(false); } diff --git a/apps/web/hooks/domains/session/use-session-model.ts b/apps/web/hooks/domains/session/use-session-model.ts index 1428a211af..d135b85215 100644 --- a/apps/web/hooks/domains/session/use-session-model.ts +++ b/apps/web/hooks/domains/session/use-session-model.ts @@ -19,16 +19,12 @@ export function useSessionModel( const sessionModel = sessionProfile?.model ?? null; - // Get pending model state for this session - const pendingModels = useAppStore((state) => state.pendingModel.bySessionId); - const clearPendingModel = useAppStore((state) => state.clearPendingModel); - const setActiveModel = useAppStore((state) => state.setActiveModel); - const pendingModel = resolvedSessionId ? pendingModels[resolvedSessionId] : null; + // Get active model state for this session (user's selected model) + const activeModels = useAppStore((state) => state.activeModel.bySessionId); + const activeModel = resolvedSessionId ? activeModels[resolvedSessionId] ?? null : null; return { sessionModel, - pendingModel, - clearPendingModel, - setActiveModel, + activeModel, }; } diff --git a/apps/web/hooks/use-message-handler.ts b/apps/web/hooks/use-message-handler.ts index 26537ba50e..5e383d55ff 100644 --- a/apps/web/hooks/use-message-handler.ts +++ b/apps/web/hooks/use-message-handler.ts @@ -5,9 +5,8 @@ export function useMessageHandler( resolvedSessionId: string | null, taskId: string | null, sessionModel: string | null, - pendingModel: string | null, - setActiveModel: (sessionId: string, modelId: string) => void, - clearPendingModel: (sessionId: string) => void + activeModel: string | null, + planMode: boolean = false ) { const handleSendMessage = useCallback( async (message: string) => { @@ -18,8 +17,8 @@ export function useMessageHandler( const client = getWebSocketClient(); if (!client) return; - // Include pending model in the request if one is selected - const modelToSend = pendingModel && pendingModel !== sessionModel ? pendingModel : undefined; + // Include active model in the request if it differs from the session model + const modelToSend = activeModel && activeModel !== sessionModel ? activeModel : undefined; await client.request( 'message.add', @@ -28,17 +27,12 @@ export function useMessageHandler( session_id: resolvedSessionId, content: message, ...(modelToSend && { model: modelToSend }), + ...(planMode && { plan_mode: true }), }, 10000 ); - - // Move pending model to active model after sending (it's now the model in use) - if (modelToSend && resolvedSessionId) { - setActiveModel(resolvedSessionId, modelToSend); - clearPendingModel(resolvedSessionId); - } }, - [resolvedSessionId, taskId, pendingModel, sessionModel, setActiveModel, clearPendingModel] + [resolvedSessionId, taskId, activeModel, sessionModel, planMode] ); return { handleSendMessage }; diff --git a/apps/web/lib/state/slices/session/session-slice.ts b/apps/web/lib/state/slices/session/session-slice.ts index 5a7952af73..b67c1df310 100644 --- a/apps/web/lib/state/slices/session/session-slice.ts +++ b/apps/web/lib/state/slices/session/session-slice.ts @@ -140,10 +140,21 @@ export const createSessionSlice: StateCreator< }), setTaskSession: (session) => set((draft) => { - draft.taskSessions.items[session.id] = session; + // Merge with existing session data to preserve fields like agent_profile_id + const existingSession = draft.taskSessions.items[session.id]; + draft.taskSessions.items[session.id] = existingSession + ? { ...existingSession, ...session } + : session; }), setTaskSessionsForTask: (taskId, sessions) => set((draft) => { + // Also update individual session items with merging to preserve existing data + sessions.forEach((session) => { + const existingSession = draft.taskSessions.items[session.id]; + draft.taskSessions.items[session.id] = existingSession + ? { ...existingSession, ...session } + : session; + }); draft.taskSessionsByTask.itemsByTaskId[taskId] = sessions; draft.taskSessionsByTask.loadingByTaskId[taskId] = false; draft.taskSessionsByTask.loadedByTaskId[taskId] = true; diff --git a/apps/web/lib/types/backend.ts b/apps/web/lib/types/backend.ts index 73d5db4720..9cb48e8050 100644 --- a/apps/web/lib/types/backend.ts +++ b/apps/web/lib/types/backend.ts @@ -162,6 +162,7 @@ export type TaskSessionStateChangedPayload = { old_state?: string; new_state: string; agent_profile_id?: string; + agent_profile_snapshot?: Record; metadata?: Record; }; @@ -415,10 +416,20 @@ export type FileContentResponse = { error?: string; }; -export type FileChangeNotificationPayload = { +// Single file change event (used within batched notifications) +export type FileChangeEvent = { timestamp: string; path: string; operation: 'create' | 'write' | 'remove' | 'rename' | 'chmod' | 'refresh'; + session_id: string; + task_id: string; + agent_id: string; +}; + +// Batched file change notification payload (multiple changes batched for efficiency) +export type FileChangeNotificationPayload = { + session_id: string; + changes: FileChangeEvent[]; }; // Open file tab for file viewer diff --git a/apps/web/lib/ws/handlers/agent-session.ts b/apps/web/lib/ws/handlers/agent-session.ts index 70e746ceec..77385df29b 100644 --- a/apps/web/lib/ws/handlers/agent-session.ts +++ b/apps/web/lib/ws/handlers/agent-session.ts @@ -17,16 +17,20 @@ export function registerTaskSessionHandlers(store: StoreApi): WsHandle // Also update or create the session object if we have the session ID if (sessionId) { const existingSession = store.getState().taskSessions.items[sessionId]; + const agentProfileId = payload.agent_profile_id; + const agentProfileSnapshot = payload.agent_profile_snapshot; + if (existingSession) { - // Update existing session - preserve all existing fields + // Update existing session - preserve all existing fields, but update snapshot if provided store.getState().setTaskSession({ ...existingSession, state: newState, + // Update snapshot if provided in the event + ...(agentProfileSnapshot ? { agent_profile_snapshot: agentProfileSnapshot } : {}), }); } else { // Create partial session object - // Include agent_profile_id from the payload if provided - const agentProfileId = payload.agent_profile_id; + // Include agent_profile_id and agent_profile_snapshot from the payload if provided store.getState().setTaskSession({ id: sessionId, task_id: taskId, @@ -34,6 +38,7 @@ export function registerTaskSessionHandlers(store: StoreApi): WsHandle started_at: '', updated_at: '', ...(agentProfileId ? { agent_profile_id: agentProfileId } : {}), + ...(agentProfileSnapshot ? { agent_profile_snapshot: agentProfileSnapshot } : {}), }); } @@ -41,7 +46,6 @@ export function registerTaskSessionHandlers(store: StoreApi): WsHandle if (sessionsByTask) { const hasSession = sessionsByTask.some((session) => session.id === sessionId); if (!hasSession) { - const agentProfileId = payload.agent_profile_id; store.getState().setTaskSessionsForTask(taskId, [...sessionsByTask, { id: sessionId, task_id: taskId, @@ -49,10 +53,15 @@ export function registerTaskSessionHandlers(store: StoreApi): WsHandle started_at: '', updated_at: '', ...(agentProfileId ? { agent_profile_id: agentProfileId } : {}), + ...(agentProfileSnapshot ? { agent_profile_snapshot: agentProfileSnapshot } : {}), }]); } else { const nextSessions = sessionsByTask.map((session) => - session.id === sessionId ? { ...session, state: newState } : session + session.id === sessionId ? { + ...session, + state: newState, + ...(agentProfileSnapshot ? { agent_profile_snapshot: agentProfileSnapshot } : {}), + } : session ); store.getState().setTaskSessionsForTask(taskId, nextSessions); }