fix: model selector (#90)

This commit is contained in:
Carlos Florêncio
2026-01-25 00:02:31 +00:00
committed by GitHub
parent d6e2380bdf
commit a3d144cbb8
19 changed files with 513 additions and 98 deletions
+2 -2
View File
@@ -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
}
+1 -1
View File
@@ -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
@@ -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,
@@ -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(&registry.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)
}
}
@@ -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
}
@@ -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)
}
})
}
}
@@ -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
}
@@ -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
@@ -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,
},
@@ -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{
@@ -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
}
+27 -5
View File
@@ -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(
+7 -19
View File
@@ -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 (
<Select value={displayedModel} onValueChange={handleModelChange}>
<SelectTrigger
className={cn(
'w-[180px] cursor-pointer',
hasPendingChange && 'border-primary/60 bg-primary/5'
)}
>
<Select value={currentModel} onValueChange={handleModelChange}>
<SelectTrigger className="w-[180px] cursor-pointer">
<div className="flex items-center gap-1.5">
<IconCpu className="h-3.5 w-3.5 text-muted-foreground" />
<SelectValue placeholder="Select model" />
{hasPendingChange && (
<span className="ml-auto text-[10px] font-medium text-primary">pending</span>
)}
</div>
</SelectTrigger>
<SelectContent>
+7 -4
View File
@@ -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);
}
@@ -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,
};
}
+6 -12
View File
@@ -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 };
@@ -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;
+12 -1
View File
@@ -162,6 +162,7 @@ export type TaskSessionStateChangedPayload = {
old_state?: string;
new_state: string;
agent_profile_id?: string;
agent_profile_snapshot?: Record<string, unknown>;
metadata?: Record<string, unknown>;
};
@@ -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
+14 -5
View File
@@ -17,16 +17,20 @@ export function registerTaskSessionHandlers(store: StoreApi<AppState>): 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<AppState>): WsHandle
started_at: '',
updated_at: '',
...(agentProfileId ? { agent_profile_id: agentProfileId } : {}),
...(agentProfileSnapshot ? { agent_profile_snapshot: agentProfileSnapshot } : {}),
});
}
@@ -41,7 +46,6 @@ export function registerTaskSessionHandlers(store: StoreApi<AppState>): 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<AppState>): 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);
}