diff --git a/apps/backend/internal/backendapp/boot_state.go b/apps/backend/internal/backendapp/boot_state.go index 3911df4563..940f89ae4a 100644 --- a/apps/backend/internal/backendapp/boot_state.go +++ b/apps/backend/internal/backendapp/boot_state.go @@ -80,6 +80,7 @@ func bootInitialState( state["agentRuntime"] = snapshot } } + builder.addAgentProfileRecentUseState(ctx, state) if route.Route == webapp.RouteSettings { // Resolve an active workspace rather than emitting null. The SPA derives @@ -243,6 +244,30 @@ func (b bootStateBuilder) addUserSettingsState(ctx context.Context, state map[st state["userSettings"] = mapUserSettingsState(response, workspaceID) } +func (b bootStateBuilder) addAgentProfileRecentUseState(ctx context.Context, state map[string]any) { + if b.p.userCtrl == nil { + return + } + records, err := b.p.userCtrl.GetAgentProfileRecentUse(ctx) + if err != nil { + b.logBootError("get agent profile recent use", err) + return + } + state["agentProfileRecentUse"] = mapAgentProfileRecentUseState(records) +} + +func mapAgentProfileRecentUseState(records []userdto.AgentProfileRecentUseDTO) map[string]any { + byContext := make(map[string]any, len(records)) + for _, record := range records { + byContext[string(record.Context)] = map[string]any{ + "profileIds": append([]string{}, record.ProfileIDs...), + "revision": record.Revision, + "updatedAt": record.UpdatedAt, + } + } + return map[string]any{"records": byContext, "loaded": true} +} + func (b bootStateBuilder) addSettingsRouteState(ctx context.Context, state map[string]any, path string) { switch path { case "/settings/prompts": diff --git a/apps/backend/internal/backendapp/boot_state_agent_profile_recent_use_test.go b/apps/backend/internal/backendapp/boot_state_agent_profile_recent_use_test.go new file mode 100644 index 0000000000..dd5a303acd --- /dev/null +++ b/apps/backend/internal/backendapp/boot_state_agent_profile_recent_use_test.go @@ -0,0 +1,46 @@ +package backendapp + +import ( + "net/http" + "net/http/httptest" + "testing" + + usermodels "github.com/kandev/kandev/internal/user/models" + "github.com/kandev/kandev/internal/webapp" +) + +func TestBootInitialStateIncludesAgentProfileRecentUseSeparately(t *testing.T) { + harness := newBootStateTestHarness(t) + if _, err := harness.userSvc.RecordAgentProfileRecentUse( + t.Context(), + usermodels.AgentProfileRecentUseQuickChat, + "profile-a", + ); err != nil { + t.Fatalf("record recent-use profile: %v", err) + } + + params := routeParams{userCtrl: harness.userCtrl} + request := httptest.NewRequest(http.MethodGet, "/", nil) + state := bootInitialState(t.Context(), request, params, webapp.RouteClassification{Route: webapp.RouteHome}) + recentUse, ok := state["agentProfileRecentUse"].(map[string]any) + if !ok { + t.Fatalf("agentProfileRecentUse = %#v, want separate state", state["agentProfileRecentUse"]) + } + if loaded, ok := recentUse["loaded"].(bool); !ok || !loaded { + t.Fatalf("agentProfileRecentUse.loaded = %#v, want true", recentUse["loaded"]) + } + records, ok := recentUse["records"].(map[string]any) + if !ok { + t.Fatalf("agentProfileRecentUse.records = %#v, want context map", recentUse["records"]) + } + record, ok := records[string(usermodels.AgentProfileRecentUseQuickChat)].(map[string]any) + if !ok { + t.Fatalf("quick-chat recent-use record = %#v", records[string(usermodels.AgentProfileRecentUseQuickChat)]) + } + if ids, ok := record["profileIds"].([]string); !ok || len(ids) != 1 || ids[0] != "profile-a" { + t.Fatalf("profileIds = %#v, want [profile-a]", record["profileIds"]) + } + if _, embedded := state["userSettings"].(map[string]any); embedded { + t.Fatal("recent-use state must not be embedded in userSettings") + } +} diff --git a/apps/backend/internal/backendapp/helpers.go b/apps/backend/internal/backendapp/helpers.go index 5079c5b0f1..2742136d12 100644 --- a/apps/backend/internal/backendapp/helpers.go +++ b/apps/backend/internal/backendapp/helpers.go @@ -1117,6 +1117,7 @@ func registerTaskRoutes(p routeParams, planService *taskservice.PlanService, han taskH := taskhandlers.RegisterTaskRoutes(p.router, p.gateway.Dispatcher, p.taskSvc, p.orchestratorSvc, p.taskRepo, planService, p.log) if p.services != nil && p.services.User != nil { taskH.SetTaskCreateLastUsedRecorder(p.services.User) + taskH.SetAgentProfileRecentUseRecorder(p.services.User) } if handoffSvc != nil { taskH.SetHandoffService(handoffSvc) diff --git a/apps/backend/internal/backendapp/orchestrator.go b/apps/backend/internal/backendapp/orchestrator.go index e76c4a3e17..8433029f2b 100644 --- a/apps/backend/internal/backendapp/orchestrator.go +++ b/apps/backend/internal/backendapp/orchestrator.go @@ -122,6 +122,7 @@ func provideOrchestrator( } orchestratorSvc := orchestrator.NewService(serviceCfg, eventBus, agentManagerClient, taskRepoAdapter, taskRepo, userSvc, secretStore, msgQueue, log) + orchestratorSvc.SetAgentProfileRecentUseRecorder(userSvc) if gitCredentialBroker != nil { orchestratorSvc.SetGitHubCredentialBroker(gitCredentialBroker, githubCredentialBrokerEndpoint(cfg)) } diff --git a/apps/backend/internal/events/types.go b/apps/backend/internal/events/types.go index deadb55dfe..180936a453 100644 --- a/apps/backend/internal/events/types.go +++ b/apps/backend/internal/events/types.go @@ -176,7 +176,8 @@ const ( // Event types for users const ( - UserSettingsUpdated = "user.settings.updated" + UserSettingsUpdated = "user.settings.updated" + UserAgentProfileRecentUseUpdated = "user.agent_profile_recent_use.updated" ) // PluginUserStateUpdated fires after a successful write/delete on a diff --git a/apps/backend/internal/gateway/websocket/user_notifications.go b/apps/backend/internal/gateway/websocket/user_notifications.go index 47a179d8e1..ca4f7131f4 100644 --- a/apps/backend/internal/gateway/websocket/user_notifications.go +++ b/apps/backend/internal/gateway/websocket/user_notifications.go @@ -27,6 +27,7 @@ func RegisterUserNotifications(ctx context.Context, eventBus bus.EventBus, hub * } b.subscribe(eventBus, events.UserSettingsUpdated, ws.ActionUserSettingsUpdated) + b.subscribe(eventBus, events.UserAgentProfileRecentUseUpdated, ws.ActionUserAgentProfileRecentUseUpdated) b.subscribe(eventBus, events.PluginUserStateUpdated, ws.ActionPluginUserStateUpdated) go func() { diff --git a/apps/backend/internal/gateway/websocket/user_notifications_test.go b/apps/backend/internal/gateway/websocket/user_notifications_test.go index cc95fdf1b1..af41103c1d 100644 --- a/apps/backend/internal/gateway/websocket/user_notifications_test.go +++ b/apps/backend/internal/gateway/websocket/user_notifications_test.go @@ -167,3 +167,54 @@ func TestRegisterUserNotifications_PluginUserStateRoutesUnderNATSTransport(t *te t.Fatal("expected the writing user's client to receive the notification even when event.Data arrives as a map (NATS transport)") } } + +func TestRegisterUserNotifications_AgentProfileRecentUseReachesOnlyOwner(t *testing.T) { + h := newTestHub(t) + owner := newTestClient("owner-conn") + other := newTestClient("other-conn") + registerTestClient(h, owner) + registerTestClient(h, other) + h.SubscribeToUser(owner, "user_1") + h.SubscribeToUser(other, "user_2") + + eventBus := bus.NewMemoryEventBus(testLoggerForUserNotifications(t)) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + RegisterUserNotifications(ctx, eventBus, h, testLoggerForUserNotifications(t)) + if err := eventBus.Publish(ctx, events.UserAgentProfileRecentUseUpdated, bus.NewEvent( + events.UserAgentProfileRecentUseUpdated, + "test", + map[string]any{ + "user_id": "user_1", + "context": "quick_chat", + "profile_ids": []string{"profile-a"}, + "revision": int64(1), + "updated_at": "2026-08-27T00:00:00Z", + }, + )); err != nil { + t.Fatalf("publish: %v", err) + } + + select { + case raw := <-owner.send: + var message ws.Message + if err := json.Unmarshal(raw, &message); err != nil { + t.Fatalf("unmarshal envelope: %v", err) + } + if message.Action != ws.ActionUserAgentProfileRecentUseUpdated { + t.Fatalf("action = %q, want %q", message.Action, ws.ActionUserAgentProfileRecentUseUpdated) + } + var payload map[string]any + if err := json.Unmarshal(message.Payload, &payload); err != nil { + t.Fatalf("unmarshal payload: %v", err) + } + if _, leaked := payload["user_id"]; leaked { + t.Fatalf("payload leaked user_id: %+v", payload) + } + default: + t.Fatal("expected owner notification") + } + if clientReceived(other) { + t.Fatal("expected other user's client to receive nothing") + } +} diff --git a/apps/backend/internal/orchestrator/agent_profile_recent_use.go b/apps/backend/internal/orchestrator/agent_profile_recent_use.go new file mode 100644 index 0000000000..c3a3f46ba5 --- /dev/null +++ b/apps/backend/internal/orchestrator/agent_profile_recent_use.go @@ -0,0 +1,60 @@ +package orchestrator + +import ( + "context" + "time" + + "go.uber.org/zap" + + "github.com/kandev/kandev/internal/auth/authn" + usermodels "github.com/kandev/kandev/internal/user/models" +) + +const deferredAgentProfileRecentUseTimeout = 5 * time.Second + +// AgentProfileRecentUseRecorder persists the bounded operational profile +// history. It is optional so orchestrator-only tests and installations without +// the user service retain their existing behavior. +type AgentProfileRecentUseRecorder interface { + RecordAgentProfileRecentUse( + ctx context.Context, + contextValue usermodels.AgentProfileRecentUseContext, + profileID string, + ) (*usermodels.AgentProfileRecentUse, error) +} + +// SetAgentProfileRecentUseRecorder wires the optional recorder used by +// deferred task-create launches. +func (s *Service) SetAgentProfileRecentUseRecorder(recorder AgentProfileRecentUseRecorder) { + s.agentProfileRecentUseRecorder = recorder +} + +// recordSuccessfulDeferredTaskProfileAsync updates task_create recency after +// a deferred launch has succeeded. The creator identity is restored from the +// persisted intent because workflow events do not carry the original request +// context. The write is best effort and bounded so launch lifecycle work never +// waits on preference persistence. +func (s *Service) recordSuccessfulDeferredTaskProfileAsync( + ctx context.Context, userID, profileID string, +) { + if s.agentProfileRecentUseRecorder == nil || userID == "" || profileID == "" { + return + } + go func() { + baseCtx := context.WithoutCancel(ctx) + recordCtx, cancel := context.WithTimeout( + authn.WithIdentity(baseCtx, authn.Identity{UserID: userID}), + deferredAgentProfileRecentUseTimeout, + ) + defer cancel() + if _, err := s.agentProfileRecentUseRecorder.RecordAgentProfileRecentUse( + recordCtx, usermodels.AgentProfileRecentUseTaskCreate, profileID, + ); err != nil { + s.logger.Warn("deferred task-create profile recency recording failed", + zap.String("user_id", userID), + zap.String("agent_profile_id", profileID), + zap.Error(err), + ) + } + }() +} diff --git a/apps/backend/internal/orchestrator/deferred_launch_consume_test.go b/apps/backend/internal/orchestrator/deferred_launch_consume_test.go index 438ab0e062..adcb275dcc 100644 --- a/apps/backend/internal/orchestrator/deferred_launch_consume_test.go +++ b/apps/backend/internal/orchestrator/deferred_launch_consume_test.go @@ -11,11 +11,13 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "github.com/kandev/kandev/internal/auth/authn" "github.com/kandev/kandev/internal/orchestrator/executor" "github.com/kandev/kandev/internal/orchestrator/watcher" "github.com/kandev/kandev/internal/task/models" sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite" taskservice "github.com/kandev/kandev/internal/task/service" + usermodels "github.com/kandev/kandev/internal/user/models" wfmodels "github.com/kandev/kandev/internal/workflow/models" v1 "github.com/kandev/kandev/pkg/api/v1" ) @@ -52,6 +54,25 @@ type launchCounter struct { fired chan struct{} } +type deferredRecentUseCall struct { + profileID string + userID string +} + +type deferredRecentUseRecorder struct { + calls chan deferredRecentUseCall +} + +func (r *deferredRecentUseRecorder) RecordAgentProfileRecentUse( + ctx context.Context, + _ usermodels.AgentProfileRecentUseContext, + profileID string, +) (*usermodels.AgentProfileRecentUse, error) { + identity, _ := authn.IdentityFromContext(ctx) + r.calls <- deferredRecentUseCall{profileID: profileID, userID: identity.UserID} + return nil, nil +} + func newLaunchCounter() *launchCounter { return &launchCounter{fired: make(chan struct{}, 8)} } @@ -89,6 +110,99 @@ func (l *launchCounter) awaitLaunch(alreadySeen int) bool { } } +func TestDeferredTaskCreateLaunchRecordsThePersistedCreatorAfterSuccess(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedChainStepTask(t, repo, deferredChainTaskID) + task, err := repo.GetTask(ctx, deferredChainTaskID) + require.NoError(t, err) + launch := task.Metadata[models.MetaKeyDeferredLaunch].(map[string]interface{}) + launch[models.DeferredLaunchUserIDKey] = "creator-1" + launch[models.DeferredLaunchRecordRecentUseKey] = true + require.NoError(t, repo.UpdateTask(ctx, task)) + + recorder := &deferredRecentUseRecorder{calls: make(chan deferredRecentUseCall, 1)} + svc := newDeferredLaunchTestService(t, repo, newLaunchCounter()) + svc.SetAgentProfileRecentUseRecorder(recorder) + + require.True(t, svc.launchDeferredTask(ctx, task, "test.deferred_launch", false)) + select { + case call := <-recorder.calls: + assert.Equal(t, "profile1", call.profileID) + assert.Equal(t, "creator-1", call.userID) + case <-time.After(2 * time.Second): + t.Fatal("successful deferred task-create launch did not record profile recency") + } + awaitLaunchedSession(t, repo, deferredChainTaskID) +} + +// A deferred launch without the selector attribution marker models an MCP +// create: the agent profile is operational launch input, not a UI selection. +// It must not reorder task_create history even if an old or forged record also +// contains a user ID. +func TestDeferredMCPLaunchDoesNotRecordRecencyWithoutSelectorAttribution(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedChainStepTask(t, repo, deferredChainTaskID) + task, err := repo.GetTask(ctx, deferredChainTaskID) + require.NoError(t, err) + launch := task.Metadata[models.MetaKeyDeferredLaunch].(map[string]interface{}) + launch[models.DeferredLaunchUserIDKey] = "mcp-user" + require.NoError(t, repo.UpdateTask(ctx, task)) + + recorder := &deferredRecentUseRecorder{calls: make(chan deferredRecentUseCall, 1)} + svc := newDeferredLaunchTestService(t, repo, newLaunchCounter()) + svc.SetAgentProfileRecentUseRecorder(recorder) + + require.True(t, svc.launchDeferredTask(ctx, task, "test.deferred_mcp_launch", false)) + awaitLaunchedSession(t, repo, deferredChainTaskID) + select { + case call := <-recorder.calls: + t.Fatalf("MCP deferred launch recorded profile %q for user %q", call.profileID, call.userID) + case <-time.After(100 * time.Millisecond): + } +} + +func TestFailedDeferredTaskCreateLaunchDoesNotRecordRecency(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedChainStepTask(t, repo, deferredChainTaskID) + task, err := repo.GetTask(ctx, deferredChainTaskID) + require.NoError(t, err) + launch := task.Metadata[models.MetaKeyDeferredLaunch].(map[string]interface{}) + launch[models.DeferredLaunchUserIDKey] = "creator-1" + launch[models.DeferredLaunchRecordRecentUseKey] = true + require.NoError(t, repo.UpdateTask(ctx, task)) + + taskRepo := newMockTaskRepo() + taskRepo.tasks[deferredChainTaskID] = &v1.Task{ + ID: deferredChainTaskID, WorkflowID: "wf1", Title: "Chain step", Description: "desc", State: v1.TaskStateCreated, + } + launchReturned := make(chan struct{}) + agentMgr := &mockAgentManager{ + repoForExecutionLookup: repo, + launchAgentFunc: func(context.Context, *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) { + close(launchReturned) + return nil, errors.New("executor refused the deferred launch") + }, + } + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, agentMgr) + recorder := &deferredRecentUseRecorder{calls: make(chan deferredRecentUseCall, 1)} + svc.SetAgentProfileRecentUseRecorder(recorder) + + require.True(t, svc.launchDeferredTask(ctx, task, "test.deferred_launch", false)) + select { + case <-launchReturned: + case <-time.After(2 * time.Second): + t.Fatal("deferred launch did not reach the failing agent boundary") + } + select { + case call := <-recorder.calls: + t.Fatalf("failed deferred launch recorded profile %q", call.profileID) + case <-time.After(100 * time.Millisecond): + } +} + // seedChainStepTask creates a task carrying a start-when-unblocked deferred // launch and no session — exactly what create_task_kandev with blocked_by and // start_agent leaves behind. @@ -384,6 +498,12 @@ func TestQueuePromotionOfBlockedWIPOverflowDoesNotLaunch(t *testing.T) { repo := setupTestRepo(t) seedSession(t, repo, "wip-occupant", "wip-occupant-session", "step-limited") seedChainStepTask(t, repo, deferredChainTaskID) + createdTask, err := repo.GetTask(ctx, deferredChainTaskID) + require.NoError(t, err) + createdLaunch := createdTask.Metadata[models.MetaKeyDeferredLaunch].(map[string]interface{}) + createdLaunch[models.DeferredLaunchUserIDKey] = "creator-1" + createdLaunch[models.DeferredLaunchRecordRecentUseKey] = true + require.NoError(t, repo.UpdateTask(ctx, createdTask)) counter := newLaunchCounter() steps := newMockStepGetter() @@ -399,6 +519,8 @@ func TestQueuePromotionOfBlockedWIPOverflowDoesNotLaunch(t *testing.T) { svc := newDeferredLaunchTestService(t, repo, counter) svc.SetWorkflowStepGetter(steps) + recorder := &deferredRecentUseRecorder{calls: make(chan deferredRecentUseCall, 1)} + svc.SetAgentProfileRecentUseRecorder(recorder) // Model an unresolved blocked_by edge. Promotion is deliberately allowed; // only the deferred agent launch must remain gated. dependencies := &launchDependencyReader{blocked: true} @@ -449,6 +571,13 @@ func TestQueuePromotionOfBlockedWIPOverflowDoesNotLaunch(t *testing.T) { require.NoError(t, err) _, promotionPending = started.Metadata[models.MetaKeyQueuePromotionPending] require.False(t, promotionPending, "resolved promotion left its lifecycle token behind") + select { + case call := <-recorder.calls: + assert.Equal(t, "profile1", call.profileID) + assert.Equal(t, "creator-1", call.userID) + case <-time.After(2 * time.Second): + t.Fatal("WIP-promoted task-create launch did not record profile recency") + } } func TestFailedPromotedDeferredLaunchRestoresTokenAndRetries(t *testing.T) { @@ -460,6 +589,9 @@ func TestFailedPromotedDeferredLaunchRestoresTokenAndRetries(t *testing.T) { )) task, err := baseRepo.GetTask(ctx, deferredChainTaskID) require.NoError(t, err) + launch := task.Metadata[models.MetaKeyDeferredLaunch].(map[string]interface{}) + launch[models.DeferredLaunchUserIDKey] = "creator-1" + launch[models.DeferredLaunchRecordRecentUseKey] = true task.WorkflowStepID = "step-draft" task.WIPAdmitted = true require.NoError(t, baseRepo.UpdateTask(ctx, task)) @@ -488,6 +620,8 @@ func TestFailedPromotedDeferredLaunchRestoresTokenAndRetries(t *testing.T) { } svc := createTestServiceWithScheduler(baseRepo, steps, taskRepo, agentMgr) svc.SetTaskDependencyReader(&resolvedDependencyReader{}) + recorder := &deferredRecentUseRecorder{calls: make(chan deferredRecentUseCall, 1)} + svc.SetAgentProfileRecentUseRecorder(recorder) svc.handleTaskQueuePromoted(ctx, watcher.TaskEventData{TaskID: deferredChainTaskID}) @@ -508,6 +642,11 @@ func TestFailedPromotedDeferredLaunchRestoresTokenAndRetries(t *testing.T) { time.Sleep(5 * time.Millisecond) } require.Equal(t, 0, sessionCount(t, baseRepo, deferredChainTaskID)) + select { + case call := <-recorder.calls: + t.Fatalf("failed promoted launch recorded profile %q", call.profileID) + case <-time.After(100 * time.Millisecond): + } // Repair the transient scheduler projection failure, then exercise the same // startup sweep that retries durable queue-promotion lifecycle tokens. @@ -534,6 +673,13 @@ func TestFailedPromotedDeferredLaunchRestoresTokenAndRetries(t *testing.T) { _, promotionPending := stored.Metadata[models.MetaKeyQueuePromotionPending] assert.False(t, promotionPending) assert.Equal(t, int32(1), agentLaunches.Load()) + select { + case call := <-recorder.calls: + assert.Equal(t, "profile1", call.profileID) + assert.Equal(t, "creator-1", call.userID) + case <-time.After(2 * time.Second): + t.Fatal("retried promoted task-create launch did not record profile recency") + } } // intentObserver records the dangerous state: does the task still carry a diff --git a/apps/backend/internal/orchestrator/event_handlers_workflow.go b/apps/backend/internal/orchestrator/event_handlers_workflow.go index dea385b089..b525391ff4 100644 --- a/apps/backend/internal/orchestrator/event_handlers_workflow.go +++ b/apps/backend/internal/orchestrator/event_handlers_workflow.go @@ -1428,6 +1428,8 @@ func (s *Service) launchDeferredTask(ctx context.Context, task *models.Task, eve AgentProfileID string `json:"agent_profile_id"` ExecutorID string `json:"executor_id"` ExecutorProfileID string `json:"executor_profile_id"` + UserID string `json:"user_id"` + RecordRecentUse bool `json:"record_recent_use"` Prompt string `json:"prompt"` PlanMode bool `json:"plan_mode"` Attachments []v1.MessageAttachment `json:"attachments"` @@ -1446,13 +1448,16 @@ func (s *Service) launchDeferredTask(ctx context.Context, task *models.Task, eve if !claimOK { return } - _, launchErr := s.LaunchSession(launchCtx, &LaunchSessionRequest{ + launchResp, launchErr := s.LaunchSession(launchCtx, &LaunchSessionRequest{ TaskID: task.ID, Intent: launchIntent, AgentProfileID: intent.AgentProfileID, ExecutorID: intent.ExecutorID, ExecutorProfileID: intent.ExecutorProfileID, WorkflowStepID: task.WorkflowStepID, Prompt: intent.Prompt, PlanMode: intent.PlanMode, Attachments: intent.Attachments, LaunchWorkspace: true, }) - if launchErr != nil { + if launchErr != nil || launchResp == nil || !launchResp.Success { + if launchErr == nil { + launchErr = errors.New("deferred launch returned an unsuccessful response") + } s.logger.Error(eventName+": failed to launch deferred task", zap.String("task_id", task.ID), zap.Error(launchErr)) s.restoreDeferredLaunch(launchCtx, task.ID, raw, eventName, metadataClaimed) if restoreQueuePromotion { @@ -1460,6 +1465,9 @@ func (s *Service) launchDeferredTask(ctx context.Context, task *models.Task, eve } return } + if intent.RecordRecentUse { + s.recordSuccessfulDeferredTaskProfileAsync(launchCtx, intent.UserID, launchResp.AgentProfileID) + } delete(task.Metadata, models.MetaKeyDeferredLaunch) if metadataClaimed { task.UpdatedAt = time.Now().UTC() diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index dd7f1457df..c46e01a0d0 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -528,6 +528,10 @@ type Service struct { // safe: both call sites guard on it. See SetSubagentContextRecorder. subagentContexts SubagentContextRecorder + // agentProfileRecentUseRecorder optionally persists the task_create profile + // selected by a deferred launch after its agent starts successfully. + agentProfileRecentUseRecorder AgentProfileRecentUseRecorder + // Turn service for managing session turns turnService TurnService diff --git a/apps/backend/internal/task/dto/converters_test.go b/apps/backend/internal/task/dto/converters_test.go index ca7d155633..926b7e9e3f 100644 --- a/apps/backend/internal/task/dto/converters_test.go +++ b/apps/backend/internal/task/dto/converters_test.go @@ -231,6 +231,46 @@ func TestFromTask_DerivesInterruptedFromMetadata(t *testing.T) { }) } +func TestFromTaskRedactsDeferredLaunchAttribution(t *testing.T) { + task := &models.Task{ + ID: "task-private-attribution", + Metadata: map[string]interface{}{ + models.MetaKeyDeferredLaunch: map[string]interface{}{ + "intent": "start", + "agent_profile_id": "profile-1", + models.DeferredLaunchUserIDKey: "user-private", + models.DeferredLaunchRecordRecentUseKey: true, + }, + }, + } + + got := FromTask(task) + launch, ok := got.Metadata[models.MetaKeyDeferredLaunch].(map[string]interface{}) + if !ok { + t.Fatalf("DTO deferred launch metadata = %#v, want a map", got.Metadata[models.MetaKeyDeferredLaunch]) + } + if _, exists := launch[models.DeferredLaunchUserIDKey]; exists { + t.Fatalf("DTO exposed deferred launch user_id = %v", launch[models.DeferredLaunchUserIDKey]) + } + if _, exists := launch[models.DeferredLaunchRecordRecentUseKey]; exists { + t.Fatalf("DTO exposed deferred launch recency marker = %v", launch[models.DeferredLaunchRecordRecentUseKey]) + } + if got := task.Metadata[models.MetaKeyDeferredLaunch].(map[string]interface{})[models.DeferredLaunchUserIDKey]; got != "user-private" { + t.Fatalf("redacting the DTO mutated the source task user_id = %v", got) + } + apiTask := task.ToAPI() + apiLaunch, ok := apiTask.Metadata[models.MetaKeyDeferredLaunch].(map[string]interface{}) + if !ok { + t.Fatalf("API deferred launch metadata = %#v, want a map", apiTask.Metadata[models.MetaKeyDeferredLaunch]) + } + if _, exists := apiLaunch[models.DeferredLaunchUserIDKey]; exists { + t.Fatalf("API exposed deferred launch user_id = %v", apiLaunch[models.DeferredLaunchUserIDKey]) + } + if _, exists := apiLaunch[models.DeferredLaunchRecordRecentUseKey]; exists { + t.Fatalf("API exposed deferred launch recency marker = %v", apiLaunch[models.DeferredLaunchRecordRecentUseKey]) + } +} + func TestTaskToAPI_DerivesInterruptedFromMetadata(t *testing.T) { now := time.Now().UTC() diff --git a/apps/backend/internal/task/dto/dto.go b/apps/backend/internal/task/dto/dto.go index c0daeaffd6..04248b8740 100644 --- a/apps/backend/internal/task/dto/dto.go +++ b/apps/backend/internal/task/dto/dto.go @@ -882,7 +882,7 @@ func FromTaskWithSessionInfo( ArchivedAt: task.ArchivedAt, CreatedAt: task.CreatedAt, UpdatedAt: task.UpdatedAt, - Metadata: task.Metadata, + Metadata: models.PublicTaskMetadata(task.Metadata), Interrupted: task.Metadata[models.MetaKeyInterruptedAt] != nil, AutoStartFailed: task.Metadata[models.MetaKeyAutoStartFailed] != nil, // Office extensions. AssigneeAgentProfileID is a read-time diff --git a/apps/backend/internal/task/handlers/task_handlers.go b/apps/backend/internal/task/handlers/task_handlers.go index 086c66f6ff..e1e8b4d0bb 100644 --- a/apps/backend/internal/task/handlers/task_handlers.go +++ b/apps/backend/internal/task/handlers/task_handlers.go @@ -31,22 +31,28 @@ type handlerRepo interface { } type TaskHandlers struct { - service *service.Service - orchestrator OrchestratorStarter - foregroundActivity dto.ForegroundActivityProvider - cancellationPending dto.CancellationPendingProvider - repo handlerRepo - planService *service.PlanService - handoffSvc *service.HandoffService - workspaceRestorer WorkspaceQuarantineRestorer - unarchiveRecoveryTimeout time.Duration - taskCreateLastUsedRecorder taskCreateLastUsedRecorder - onTaskCreatedWithPR func(ctx context.Context, taskID, sessionID, prURL, branch string) - logger *logger.Logger + service *service.Service + orchestrator OrchestratorStarter + foregroundActivity dto.ForegroundActivityProvider + cancellationPending dto.CancellationPendingProvider + repo handlerRepo + planService *service.PlanService + handoffSvc *service.HandoffService + workspaceRestorer WorkspaceQuarantineRestorer + unarchiveRecoveryTimeout time.Duration + taskCreateLastUsedRecorder taskCreateLastUsedRecorder + agentProfileRecentUseRecorder agentProfileRecentUseRecorder + onTaskCreatedWithPR func(ctx context.Context, taskID, sessionID, prURL, branch string) + logger *logger.Logger } const defaultUnarchiveRecoveryTimeout = 30 * time.Second +// agentProfileRecentUseTimeout bounds detached preference persistence. The +// launch response does not wait for this best-effort write, and a stuck store +// must not leave an unbounded background goroutine behind. +const agentProfileRecentUseTimeout = 5 * time.Second + func (h *TaskHandlers) detachedRecoveryTimeout() time.Duration { if h.unarchiveRecoveryTimeout > 0 { return h.unarchiveRecoveryTimeout @@ -62,6 +68,14 @@ type taskCreateLastUsedRecorder interface { RecordTaskCreateLastUsed(ctx context.Context, patch usermodels.TaskCreateLastUsed) error } +type agentProfileRecentUseRecorder interface { + RecordAgentProfileRecentUse( + ctx context.Context, + contextValue usermodels.AgentProfileRecentUseContext, + profileID string, + ) (*usermodels.AgentProfileRecentUse, error) +} + // SetHandoffService wires the office task-handoffs service used by the // Kanban subtask path to attach workspace-group membership and the // sequential blocker chain (handoffs phase 5). Optional — nil disables @@ -90,6 +104,21 @@ func (h *TaskHandlers) SetTaskCreateLastUsedRecorder(recorder taskCreateLastUsed h.taskCreateLastUsedRecorder = recorder } +func (h *TaskHandlers) SetAgentProfileRecentUseRecorder(recorder agentProfileRecentUseRecorder) { + h.agentProfileRecentUseRecorder = recorder +} + +func (h *TaskHandlers) recordSuccessfulTaskCreateProfileAsync(ctx context.Context, profileID string) { + if h.agentProfileRecentUseRecorder == nil || profileID == "" { + return + } + recordCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), agentProfileRecentUseTimeout) + go func() { + defer cancel() + h.recordSuccessfulTaskCreateProfile(recordCtx, profileID) + }() +} + // SetOnTaskCreatedWithPR sets a callback invoked when a task is created with a PR URL // in one of its repository inputs. The callback runs in a background goroutine. func (h *TaskHandlers) SetOnTaskCreatedWithPR(fn func(ctx context.Context, taskID, sessionID, prURL, branch string)) { diff --git a/apps/backend/internal/task/handlers/task_http_handlers.go b/apps/backend/internal/task/handlers/task_http_handlers.go index 49753d3fae..9672e15971 100644 --- a/apps/backend/internal/task/handlers/task_http_handlers.go +++ b/apps/backend/internal/task/handlers/task_http_handlers.go @@ -767,6 +767,7 @@ type createTaskResponse struct { dto.TaskDTO TaskSessionID string `json:"session_id,omitempty"` AgentExecutionID string `json:"agent_execution_id,omitempty"` + AgentProfileID string `json:"agent_profile_id,omitempty"` // Deduplicated and CreationComplete are required booleans (not // presence-only markers) on every create-idempotency outcome, per // docs/specs/tasks/requirements/external-id-idempotency.md. Deduplicated is true @@ -932,28 +933,29 @@ func (h *TaskHandlers) httpCreateTask(c *gin.Context) { } result, err := h.service.CreateTask(c.Request.Context(), &service.CreateTaskRequest{ - WorkspaceID: body.WorkspaceID, - WorkflowID: body.WorkflowID, - WorkflowStepID: body.WorkflowStepID, - Title: title, - Description: description, - AutoTitle: body.AutoTitle, - Autopilot: body.Autopilot, - Priority: body.Priority, - State: body.State, - Repositories: convertToServiceRepos(repos), - Position: body.Position, - Metadata: metadata, - DeferredLaunch: deferredLaunch, - PlanMode: body.PlanMode, - StartAgent: body.StartAgent, - ParentID: body.ParentID, - WorkspacePath: body.WorkspacePath, - BlockedBy: body.BlockedBy, - StartWhenUnblocked: body.StartWhenUnblocked, - ProjectID: body.ProjectID, - Labels: labels, - ExternalID: body.ExternalID, + WorkspaceID: body.WorkspaceID, + WorkflowID: body.WorkflowID, + WorkflowStepID: body.WorkflowStepID, + Title: title, + Description: description, + AutoTitle: body.AutoTitle, + Autopilot: body.Autopilot, + Priority: body.Priority, + State: body.State, + Repositories: convertToServiceRepos(repos), + Position: body.Position, + Metadata: metadata, + DeferredLaunch: deferredLaunch, + RecordAgentProfileRecentUse: true, + PlanMode: body.PlanMode, + StartAgent: body.StartAgent, + ParentID: body.ParentID, + WorkspacePath: body.WorkspacePath, + BlockedBy: body.BlockedBy, + StartWhenUnblocked: body.StartWhenUnblocked, + ProjectID: body.ProjectID, + Labels: labels, + ExternalID: body.ExternalID, }) if err != nil { handleNotFound(c, h.logger, err, "task not created") @@ -1058,7 +1060,7 @@ func (h *TaskHandlers) httpCreateTask(c *gin.Context) { return } - h.dispatchTaskSession(taskDTO.ID, taskDTO.Description, body, dispatch) + h.dispatchTaskSession(c.Request.Context(), taskDTO.ID, taskDTO.Description, body, dispatch) h.recordTaskCreateLastUsed(c.Request.Context(), body, repos) // Associate PR with task if any repository input contains a PR URL @@ -1450,6 +1452,7 @@ func (h *TaskHandlers) prepareTaskSession( h.logger.Error("failed to prepare session for task", zap.Error(err), zap.String("task_id", taskID)) } else { response.TaskSessionID = resp.SessionID + response.AgentProfileID = firstNonEmpty(resp.AgentProfileID, body.AgentProfileID) } return nil } @@ -1489,6 +1492,9 @@ func (h *TaskHandlers) prepareStartAgentSession( } sessionID := prepResp.SessionID response.TaskSessionID = sessionID + // The follow-up start resolves workflow and dynamic-profile overrides. Do + // not expose the requested profile as effective in this response; the async + // launch records the profile returned by the successful start. if updatedTask, updateErr := h.service.UpdateTaskState(ctx, taskID, v1.TaskStateScheduling); updateErr != nil { h.logger.Warn("failed to mark task scheduling after preparing start session", zap.Error(updateErr), @@ -1503,7 +1509,12 @@ func (h *TaskHandlers) prepareStartAgentSession( // dispatchTaskSession launches the agent asynchronously (create-sequence // step 8). Callers MUST only invoke this after settlement has succeeded — // dispatch is nil whenever there was nothing prepared to dispatch. -func (h *TaskHandlers) dispatchTaskSession(taskID, description string, body httpCreateTaskRequest, dispatch *startAgentDispatch) { +func (h *TaskHandlers) dispatchTaskSession( + ctx context.Context, + taskID, description string, + body httpCreateTaskRequest, + dispatch *startAgentDispatch, +) { if dispatch == nil { return } @@ -1511,7 +1522,7 @@ func (h *TaskHandlers) dispatchTaskSession(taskID, description string, body http // Launch agent asynchronously so the HTTP request can return immediately. // The frontend will receive WebSocket updates when the agent actually starts. go func() { - startCtx, cancel := context.WithTimeout(context.Background(), constants.AgentLaunchTimeout) + startCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), constants.AgentLaunchTimeout) defer cancel() launchResp, err := h.orchestrator.LaunchSession(startCtx, &orchestrator.LaunchSessionRequest{ TaskID: taskID, @@ -1527,13 +1538,33 @@ func (h *TaskHandlers) dispatchTaskSession(taskID, description string, body http h.logger.Error("failed to start agent for task (async)", zap.Error(err), zap.String("task_id", taskID), zap.String("session_id", sessionID)) return } + if launchResp == nil || !launchResp.Success { + h.logger.Warn("agent start returned no successful launch response", + zap.String("task_id", taskID), zap.String("session_id", sessionID)) + return + } h.logger.Info("agent started for task (async)", zap.String("task_id", taskID), zap.String("session_id", launchResp.SessionID), zap.String("execution_id", launchResp.AgentExecutionID)) + h.recordSuccessfulTaskCreateProfile(startCtx, launchResp.AgentProfileID) }() } +func (h *TaskHandlers) recordSuccessfulTaskCreateProfile(ctx context.Context, profileID string) { + if h.agentProfileRecentUseRecorder == nil || profileID == "" { + return + } + if _, err := h.agentProfileRecentUseRecorder.RecordAgentProfileRecentUse( + ctx, + usermodels.AgentProfileRecentUseTaskCreate, + profileID, + ); err != nil { + h.logger.Warn("failed to record agent profile recent use after successful task launch", + zap.String("profile_id", profileID), zap.Error(err)) + } +} + type httpUpdateTaskRequest struct { Title *string `json:"title,omitempty"` Description *string `json:"description,omitempty"` @@ -1898,8 +1929,9 @@ func (body *httpStartQuickChatRequest) validateRepositories() error { // httpStartQuickChatResponse is returned when a quick chat session is created. type httpStartQuickChatResponse struct { - TaskID string `json:"task_id"` - SessionID string `json:"session_id"` + TaskID string `json:"task_id"` + SessionID string `json:"session_id"` + AgentProfileID string `json:"agent_profile_id,omitempty"` } // quickChatParams holds resolved parameters for creating a quick chat session. @@ -2048,8 +2080,9 @@ func (h *TaskHandlers) httpStartQuickChat(c *gin.Context) { zap.String("workspace_id", workspaceID)) c.JSON(http.StatusOK, httpStartQuickChatResponse{ - TaskID: task.ID, - SessionID: resp.SessionID, + TaskID: task.ID, + SessionID: resp.SessionID, + AgentProfileID: firstNonEmpty(resp.AgentProfileID, params.agentProfileID), }) } @@ -2202,8 +2235,9 @@ func (h *TaskHandlers) httpStartConfigChat(c *gin.Context) { zap.String("workspace_id", workspaceID)) c.JSON(http.StatusOK, httpStartQuickChatResponse{ - TaskID: task.ID, - SessionID: sessionID, + TaskID: task.ID, + SessionID: sessionID, + AgentProfileID: firstNonEmpty(resp.AgentProfileID, agentProfileID), }) } diff --git a/apps/backend/internal/task/handlers/task_http_handlers_test.go b/apps/backend/internal/task/handlers/task_http_handlers_test.go index 25de378860..fdf79d9cfc 100644 --- a/apps/backend/internal/task/handlers/task_http_handlers_test.go +++ b/apps/backend/internal/task/handlers/task_http_handlers_test.go @@ -38,10 +38,12 @@ import ( // LaunchSession to short-circuit the two-phase create flow before its async // start goroutine spawns (keeping assertions race-free). type captureOrchestrator struct { - mu sync.Mutex - requests []*orchestrator.LaunchSessionRequest - prepErr error - startCreatedCalled chan struct{} + mu sync.Mutex + requests []*orchestrator.LaunchSessionRequest + prepErr error + startCreatedErr error + startCreatedResponse *orchestrator.LaunchSessionResponse + startCreatedCalled chan struct{} } func (m *captureOrchestrator) LaunchSession(_ context.Context, req *orchestrator.LaunchSessionRequest) (*orchestrator.LaunchSessionResponse, error) { @@ -54,12 +56,33 @@ func (m *captureOrchestrator) LaunchSession(_ context.Context, req *orchestrator default: } } + if req.Intent == orchestrator.IntentStartCreated { + if m.startCreatedErr != nil { + return nil, m.startCreatedErr + } + if m.startCreatedResponse != nil { + return m.startCreatedResponse, nil + } + } if m.prepErr != nil { return nil, m.prepErr } return &orchestrator.LaunchSessionResponse{SessionID: "sess-1"}, nil } +type captureAgentProfileRecentUseRecorder struct { + profileIDs chan string +} + +func (r *captureAgentProfileRecentUseRecorder) RecordAgentProfileRecentUse( + _ context.Context, + _ usermodels.AgentProfileRecentUseContext, + profileID string, +) (*usermodels.AgentProfileRecentUse, error) { + r.profileIDs <- profileID + return &usermodels.AgentProfileRecentUse{ProfileIDs: []string{profileID}}, nil +} + func (m *captureOrchestrator) EnsureSession(_ context.Context, _ string, _ ...orchestrator.EnsureSessionOptions) (*orchestrator.EnsureSessionResponse, error) { return nil, nil } @@ -273,7 +296,7 @@ func TestStartAgentForNewTask_SetsDeferredStart(t *testing.T) { body := httpCreateTaskRequest{StartAgent: true, AgentProfileID: "profile-1"} dispatch := h.prepareStartAgentSession(context.Background(), resp, "task-1", body, "step-1") require.Nil(t, dispatch, "a prepare failure must not produce a dispatch") - h.dispatchTaskSession("task-1", "do the thing", body, dispatch) + h.dispatchTaskSession(context.Background(), "task-1", "do the thing", body, dispatch) orch.mu.Lock() defer orch.mu.Unlock() @@ -284,6 +307,61 @@ func TestStartAgentForNewTask_SetsDeferredStart(t *testing.T) { "sync prepare must defer the start so the passthrough PTY is launched with the prompt by the follow-up IntentStartCreated") } +func TestDispatchTaskSessionRecordsOnlyTheSuccessfulEffectiveProfile(t *testing.T) { + t.Run("successful launch uses the resolved profile", func(t *testing.T) { + recorder := &captureAgentProfileRecentUseRecorder{profileIDs: make(chan string, 1)} + orch := &captureOrchestrator{startCreatedResponse: &orchestrator.LaunchSessionResponse{ + Success: true, + SessionID: "session-1", + AgentProfileID: "workflow-profile", + }} + h := &TaskHandlers{ + orchestrator: orch, + agentProfileRecentUseRecorder: recorder, + logger: newTestLogger(t), + } + + h.dispatchTaskSession( + context.Background(), + "task-1", + "do the thing", + httpCreateTaskRequest{AgentProfileID: "requested-profile"}, + &startAgentDispatch{sessionID: "session-1"}, + ) + + select { + case profileID := <-recorder.profileIDs: + assert.Equal(t, "workflow-profile", profileID) + case <-time.After(time.Second): + t.Fatal("successful launch did not record profile recency") + } + }) + + t.Run("failed launch does not record recency", func(t *testing.T) { + recorder := &captureAgentProfileRecentUseRecorder{profileIDs: make(chan string, 1)} + orch := &captureOrchestrator{startCreatedErr: errors.New("launch failed")} + h := &TaskHandlers{ + orchestrator: orch, + agentProfileRecentUseRecorder: recorder, + logger: newTestLogger(t), + } + + h.dispatchTaskSession( + context.Background(), + "task-1", + "do the thing", + httpCreateTaskRequest{AgentProfileID: "requested-profile"}, + &startAgentDispatch{sessionID: "session-1"}, + ) + + select { + case profileID := <-recorder.profileIDs: + t.Fatalf("failed launch recorded profile %q", profileID) + case <-time.After(50 * time.Millisecond): + } + }) +} + // configChatRepo returns a non-nil workspace so resolveConfigChatDefaults does // not nil-deref; everything else inherits mockRepository's no-op stubs. type configChatRepo struct { diff --git a/apps/backend/internal/task/handlers/task_ws_handlers.go b/apps/backend/internal/task/handlers/task_ws_handlers.go index 587fe95597..6d1caa35f3 100644 --- a/apps/backend/internal/task/handlers/task_ws_handlers.go +++ b/apps/backend/internal/task/handlers/task_ws_handlers.go @@ -176,21 +176,22 @@ func (h *TaskHandlers) wsCreateTask(ctx context.Context, msg *ws.Message) (*ws.M } result, err := h.service.CreateTask(ctx, &service.CreateTaskRequest{ - WorkspaceID: req.WorkspaceID, - WorkflowID: req.WorkflowID, - WorkflowStepID: req.WorkflowStepID, - Title: title, - Description: description, - Autopilot: req.Autopilot, - Priority: req.Priority, - State: req.State, - Repositories: convertToServiceRepos(repos), - Position: req.Position, - Metadata: req.Metadata, - DeferredLaunch: deferredLaunch, - PlanMode: req.PlanMode, - StartAgent: req.StartAgent, - ParentID: req.ParentID, + WorkspaceID: req.WorkspaceID, + WorkflowID: req.WorkflowID, + WorkflowStepID: req.WorkflowStepID, + Title: title, + Description: description, + Autopilot: req.Autopilot, + Priority: req.Priority, + State: req.State, + Repositories: convertToServiceRepos(repos), + Position: req.Position, + Metadata: req.Metadata, + DeferredLaunch: deferredLaunch, + RecordAgentProfileRecentUse: true, + PlanMode: req.PlanMode, + StartAgent: req.StartAgent, + ParentID: req.ParentID, }) if err != nil { h.logger.Error("failed to create task", zap.Error(err)) @@ -221,6 +222,10 @@ func (h *TaskHandlers) wsCreateTask(ctx context.Context, msg *ws.Message) (*ws.M } response.TaskSessionID = launchResp.SessionID response.AgentExecutionID = launchResp.AgentExecutionID + response.AgentProfileID = launchResp.AgentProfileID + if launchResp.Success { + h.recordSuccessfulTaskCreateProfileAsync(ctx, launchResp.AgentProfileID) + } } h.recordTaskCreateLastUsed(ctx, httpCreateTaskRequest{ WorkspaceID: req.WorkspaceID, diff --git a/apps/backend/internal/task/handlers/task_ws_handlers_test.go b/apps/backend/internal/task/handlers/task_ws_handlers_test.go index d5950132ae..e28d91a2d7 100644 --- a/apps/backend/internal/task/handlers/task_ws_handlers_test.go +++ b/apps/backend/internal/task/handlers/task_ws_handlers_test.go @@ -4,18 +4,50 @@ import ( "context" "fmt" "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/kandev/kandev/internal/auth/authn" + "github.com/kandev/kandev/internal/orchestrator" "github.com/kandev/kandev/internal/task/dto" "github.com/kandev/kandev/internal/task/models" "github.com/kandev/kandev/internal/task/service" + usermodels "github.com/kandev/kandev/internal/user/models" v1 "github.com/kandev/kandev/pkg/api/v1" ws "github.com/kandev/kandev/pkg/websocket" ) +type blockingAgentProfileRecentUseRecorder struct { + started chan struct{} + release chan struct{} +} + +func (r *blockingAgentProfileRecentUseRecorder) RecordAgentProfileRecentUse( + context.Context, + usermodels.AgentProfileRecentUseContext, + string, +) (*usermodels.AgentProfileRecentUse, error) { + close(r.started) + <-r.release + return nil, nil +} + +type successfulWSLaunchOrchestrator struct{} + +func (successfulWSLaunchOrchestrator) LaunchSession(context.Context, *orchestrator.LaunchSessionRequest) (*orchestrator.LaunchSessionResponse, error) { + return &orchestrator.LaunchSessionResponse{ + Success: true, + SessionID: "session-1", + AgentProfileID: "profile-1", + }, nil +} + +func (successfulWSLaunchOrchestrator) EnsureSession(context.Context, string, ...orchestrator.EnsureSessionOptions) (*orchestrator.EnsureSessionResponse, error) { + return nil, nil +} + // wsTaskRepo owns everything as "user-b" and records every write, so a request // made as "user-a" both fails and leaves no trace. type wsTaskRepo struct { @@ -456,3 +488,49 @@ func TestWSCreateTask_PlanModeStartAgentPreservesPlanMode(t *testing.T) { assert.Equal(t, true, pmFlag, "plan_mode=true must remain true in the deferred launch intent") } + +func TestWSCreateTaskDoesNotWaitForRecentUsePersistence(t *testing.T) { + repo := &wsTaskRepo{} + h := newWSTaskHandlers(t, repo) + recorder := &blockingAgentProfileRecentUseRecorder{ + started: make(chan struct{}), + release: make(chan struct{}), + } + h.orchestrator = successfulWSLaunchOrchestrator{} + h.agentProfileRecentUseRecorder = recorder + defer close(recorder.release) + + result := make(chan struct { + response *ws.Message + err error + }, 1) + go func() { + response, err := h.wsCreateTask(asUser("user-b"), wsWorkflowRequest(t, ws.ActionTaskCreate, map[string]any{ + "workspace_id": "ws-b", + "workflow_id": "wf-b", + "title": "Async recency", + "description": "Start without waiting for preference persistence", + "start_agent": true, + "agent_profile_id": "profile-1", + })) + result <- struct { + response *ws.Message + err error + }{response: response, err: err} + }() + + select { + case <-recorder.started: + case <-time.After(500 * time.Millisecond): + t.Fatal("recent-use recorder was not invoked") + } + + select { + case got := <-result: + require.NoError(t, got.err) + require.NotNil(t, got.response) + require.Equal(t, ws.MessageTypeResponse, got.response.Type) + case <-time.After(100 * time.Millisecond): + t.Fatal("successful task.create waited for best-effort recent-use persistence") + } +} diff --git a/apps/backend/internal/task/models/models.go b/apps/backend/internal/task/models/models.go index 97e8d619f1..4d3b3c44e2 100644 --- a/apps/backend/internal/task/models/models.go +++ b/apps/backend/internal/task/models/models.go @@ -129,6 +129,17 @@ const ( // its own intents and keeps a WIP-only intent from being read as a chain // step (and vice versa). DeferredLaunchStartWhenUnblockedKey = "start_when_unblocked" + // DeferredLaunchUserIDKey preserves the authenticated creator so a + // selector-backed deferred task-create launch can update that user's + // task_create history after the launch is promoted by the workflow engine. + // It is server-owned state inside the protected deferred launch record and + // must not be accepted from task metadata request surfaces. + DeferredLaunchUserIDKey = "user_id" + // DeferredLaunchRecordRecentUseKey marks a deferred launch that originated + // from a selector-backed task-create surface and is therefore eligible to + // update task_create profile history after promotion. It is server-owned + // state and is omitted from task DTOs. + DeferredLaunchRecordRecentUseKey = "record_recent_use" // MetaKeyWorkspacePath is the optional host folder for repo-less tasks // (set by CreateTask, read by the orchestrator when building a session). // Centralised here so the set/read sites can't drift apart. @@ -2420,7 +2431,7 @@ func (t *Task) ToAPI() *v1.Task { Repositories: repositories, CreatedAt: t.CreatedAt, UpdatedAt: t.UpdatedAt, - Metadata: t.Metadata, + Metadata: PublicTaskMetadata(t.Metadata), Interrupted: t.Metadata[MetaKeyInterruptedAt] != nil, AutoStartFailed: t.Metadata[MetaKeyAutoStartFailed] != nil, IsEphemeral: t.IsEphemeral, diff --git a/apps/backend/internal/task/models/public_metadata.go b/apps/backend/internal/task/models/public_metadata.go new file mode 100644 index 0000000000..ffc451207b --- /dev/null +++ b/apps/backend/internal/task/models/public_metadata.go @@ -0,0 +1,22 @@ +package models + +import "maps" + +// PublicTaskMetadata returns a detached task metadata projection for API +// responses. Deferred launch attribution is server-owned and must never expose +// the creator's raw user ID or the internal recency marker. +func PublicTaskMetadata(metadata map[string]interface{}) map[string]interface{} { + if metadata == nil { + return nil + } + public := maps.Clone(metadata) + deferred, ok := public[MetaKeyDeferredLaunch].(map[string]interface{}) + if !ok { + return public + } + publicDeferred := maps.Clone(deferred) + delete(publicDeferred, DeferredLaunchUserIDKey) + delete(publicDeferred, DeferredLaunchRecordRecentUseKey) + public[MetaKeyDeferredLaunch] = publicDeferred + return public +} diff --git a/apps/backend/internal/task/service/service_deferred_launch_test.go b/apps/backend/internal/task/service/service_deferred_launch_test.go index 7148f66b24..416ee1305d 100644 --- a/apps/backend/internal/task/service/service_deferred_launch_test.go +++ b/apps/backend/internal/task/service/service_deferred_launch_test.go @@ -5,14 +5,138 @@ import ( "errors" "testing" + "github.com/kandev/kandev/internal/auth/authn" "github.com/kandev/kandev/internal/task/models" "github.com/kandev/kandev/internal/task/repository" sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite" v1 "github.com/kandev/kandev/pkg/api/v1" + "github.com/stretchr/testify/require" ) const deferredLaunchTaskID = "task-deferred" +func TestCreateTaskPersistsDeferredLaunchOriginUserID(t *testing.T) { + svc, _, repo := createTestService(t) + ctx := authn.WithIdentity(context.Background(), authn.Identity{UserID: "user-created"}) + + require.NoError(t, repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws-origin", Name: "Origin"})) + require.NoError(t, repo.CreateWorkflow(ctx, &models.Workflow{ID: "wf-origin", WorkspaceID: "ws-origin", Name: "flow"})) + + result, err := svc.CreateTask(ctx, &CreateTaskRequest{ + WorkspaceID: "ws-origin", + WorkflowID: "wf-origin", + Title: "Deferred task", + StartAgent: true, + RecordAgentProfileRecentUse: true, + DeferredLaunch: map[string]interface{}{ + "intent": "start", + "agent_profile_id": "profile-1", + "prompt": "run this later", + }, + }) + require.NoError(t, err) + + launch, ok := result.Task.Metadata[models.MetaKeyDeferredLaunch].(map[string]interface{}) + require.True(t, ok, "created task must retain its deferred launch intent") + if got := launch[models.DeferredLaunchUserIDKey]; got != "user-created" { + t.Fatalf("deferred launch user_id = %v, want user-created", got) + } + if got := launch[models.DeferredLaunchRecordRecentUseKey]; got != true { + t.Fatalf("deferred launch record_recent_use = %v, want true", got) + } +} + +func TestCreateTaskDoesNotPersistDeferredLaunchOriginUserIDWithoutOptIn(t *testing.T) { + svc, _, repo := createTestService(t) + ctx := authn.WithIdentity(context.Background(), authn.Identity{UserID: "mcp-user"}) + + require.NoError(t, repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws-mcp", Name: "MCP"})) + require.NoError(t, repo.CreateWorkflow(ctx, &models.Workflow{ID: "wf-mcp", WorkspaceID: "ws-mcp", Name: "flow"})) + + result, err := svc.CreateTask(ctx, &CreateTaskRequest{ + WorkspaceID: "ws-mcp", + WorkflowID: "wf-mcp", + Title: "MCP deferred task", + StartAgent: true, + DeferredLaunch: map[string]interface{}{ + "intent": "start", + "agent_profile_id": "profile-mcp", + "prompt": "run this later", + }, + }) + require.NoError(t, err) + + launch, ok := result.Task.Metadata[models.MetaKeyDeferredLaunch].(map[string]interface{}) + require.True(t, ok, "created task must retain its deferred launch intent") + if _, exists := launch[models.DeferredLaunchUserIDKey]; exists { + t.Fatalf("deferred launch user_id = %v, want no recency attribution without opt-in", launch[models.DeferredLaunchUserIDKey]) + } +} + +func TestUpdateTaskCannotReplaceDeferredLaunchAttribution(t *testing.T) { + svc, _, repo := createTestService(t) + ctx := context.Background() + require.NoError(t, repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws-protected", Name: "Protected"})) + require.NoError(t, repo.CreateWorkflow(ctx, &models.Workflow{ID: "wf-protected", WorkspaceID: "ws-protected", Name: "flow"})) + require.NoError(t, repo.CreateTask(ctx, &models.Task{ + ID: "task-protected", WorkspaceID: "ws-protected", WorkflowID: "wf-protected", + Title: "Protected deferred task", State: v1.TaskStateCreated, + Metadata: map[string]interface{}{ + models.MetaKeyDeferredLaunch: map[string]interface{}{ + "intent": "start", + "agent_profile_id": "profile-protected", + models.DeferredLaunchUserIDKey: "user-a", + models.DeferredLaunchRecordRecentUseKey: true, + }, + }, + })) + + attackerCtx := authn.WithIdentity(ctx, authn.Identity{UserID: "user-b"}) + _, err := svc.UpdateTask(attackerCtx, "task-protected", &UpdateTaskRequest{ + Metadata: map[string]interface{}{ + "ordinary": "allowed", + models.MetaKeyDeferredLaunch: map[string]interface{}{ + "intent": "start", + "agent_profile_id": "profile-protected", + models.DeferredLaunchUserIDKey: "user-b", + models.DeferredLaunchRecordRecentUseKey: true, + }, + }, + }) + require.NoError(t, err) + + updated, err := repo.GetTask(ctx, "task-protected") + require.NoError(t, err) + launch, ok := updated.Metadata[models.MetaKeyDeferredLaunch].(map[string]interface{}) + require.True(t, ok, "deferred launch attribution must survive generic metadata updates") + if got := launch[models.DeferredLaunchUserIDKey]; got != "user-a" { + t.Fatalf("deferred launch user_id = %v, want immutable user-a", got) + } + if got := updated.Metadata["ordinary"]; got != "allowed" { + t.Fatalf("ordinary metadata = %v, want the caller's update preserved", got) + } + + _, err = svc.UpdateTaskMetadata(attackerCtx, "task-protected", map[string]interface{}{ + "ordinary_from_merge": "allowed", + models.MetaKeyDeferredLaunch: map[string]interface{}{ + models.DeferredLaunchUserIDKey: "user-b", + models.DeferredLaunchRecordRecentUseKey: true, + }, + }) + require.NoError(t, err) + + updated, err = repo.GetTask(ctx, "task-protected") + require.NoError(t, err) + launch, ok = updated.Metadata[models.MetaKeyDeferredLaunch].(map[string]interface{}) + require.True(t, ok, "deferred launch attribution must survive metadata merges") + if got := launch[models.DeferredLaunchUserIDKey]; got != "user-a" { + t.Fatalf("merged deferred launch user_id = %v, want immutable user-a", got) + } + if got := updated.Metadata["ordinary_from_merge"]; got != "allowed" { + t.Fatalf("ordinary metadata from merge = %v, want the caller's update preserved", got) + } +} + // seedDeferredLaunchTask creates a task carrying a pending start-when-unblocked // deferred launch, matching what create_task_kandev with blocked_by leaves. func seedDeferredLaunchTask(t *testing.T, repo *sqliterepo.Repository) { diff --git a/apps/backend/internal/task/service/service_events.go b/apps/backend/internal/task/service/service_events.go index 37af748915..f31d6b89d8 100644 --- a/apps/backend/internal/task/service/service_events.go +++ b/apps/backend/internal/task/service/service_events.go @@ -429,7 +429,7 @@ func (s *Service) publishTaskEventNow(ctx context.Context, eventType string, tas } s.addTaskWorkspaceFoldersToEvent(ctx, task, data) if task.Metadata != nil { - data["metadata"] = task.Metadata + data["metadata"] = models.PublicTaskMetadata(task.Metadata) } if oldState != nil { data["old_state"] = string(*oldState) diff --git a/apps/backend/internal/task/service/service_events_test.go b/apps/backend/internal/task/service/service_events_test.go index ad518ffc29..0cabc2b817 100644 --- a/apps/backend/internal/task/service/service_events_test.go +++ b/apps/backend/internal/task/service/service_events_test.go @@ -603,6 +603,40 @@ func TestPublishTaskUpdated_EmitsAutopilot(t *testing.T) { } } +func TestPublishTaskUpdatedRedactsDeferredLaunchAttribution(t *testing.T) { + svc, eventBus, _ := createTestService(t) + task := &models.Task{ + ID: "task-private-attribution", WorkspaceID: "ws-1", WorkflowID: "wf-1", WorkflowStepID: "step-1", + Metadata: map[string]interface{}{ + models.MetaKeyDeferredLaunch: map[string]interface{}{ + models.DeferredLaunchUserIDKey: "user-private", + models.DeferredLaunchRecordRecentUseKey: true, + }, + }, + } + + svc.PublishTaskUpdated(context.Background(), task) + + data := singlePublishedEventData(t, eventBus) + metadata, ok := data["metadata"].(map[string]interface{}) + if !ok { + t.Fatalf("event metadata = %#v, want a map", data["metadata"]) + } + launch, ok := metadata[models.MetaKeyDeferredLaunch].(map[string]interface{}) + if !ok { + t.Fatalf("event deferred launch metadata = %#v, want a map", metadata[models.MetaKeyDeferredLaunch]) + } + if _, exists := launch[models.DeferredLaunchUserIDKey]; exists { + t.Fatalf("event exposed deferred launch user_id = %v", launch[models.DeferredLaunchUserIDKey]) + } + if _, exists := launch[models.DeferredLaunchRecordRecentUseKey]; exists { + t.Fatalf("event exposed deferred launch recency marker = %v", launch[models.DeferredLaunchRecordRecentUseKey]) + } + if got := task.Metadata[models.MetaKeyDeferredLaunch].(map[string]interface{})[models.DeferredLaunchUserIDKey]; got != "user-private" { + t.Fatalf("redacting the event mutated the source task user_id = %v", got) + } +} + // TestPublishTaskUpdated_EmitsAutoStartFailed regression-tests the WS gap // found in Review round 2: setTaskAutoStartFailedMarker / // clearTaskAutoStartFailedMarker both call PublishTaskUpdated expecting open diff --git a/apps/backend/internal/task/service/service_requests.go b/apps/backend/internal/task/service/service_requests.go index af122ce53a..5e9e835353 100644 --- a/apps/backend/internal/task/service/service_requests.go +++ b/apps/backend/internal/task/service/service_requests.go @@ -80,7 +80,11 @@ type CreateTaskRequest struct { Position int `json:"position"` Metadata map[string]interface{} `json:"metadata,omitempty"` DeferredLaunch map[string]interface{} `json:"deferred_launch,omitempty"` - PlanMode bool `json:"plan_mode,omitempty"` + // RecordAgentProfileRecentUse opts this deferred launch into task_create + // profile-history attribution. Only the authenticated HTTP/WS selector + // surfaces set it; programmatic callers such as MCP must leave it false. + RecordAgentProfileRecentUse bool `json:"-"` + PlanMode bool `json:"plan_mode,omitempty"` // StartAgent reports that the caller intends to launch an agent for this // task right away. It only steers step resolution (see resolveWorkflowStep) diff --git a/apps/backend/internal/task/service/service_task_metadata.go b/apps/backend/internal/task/service/service_task_metadata.go new file mode 100644 index 0000000000..c162409da3 --- /dev/null +++ b/apps/backend/internal/task/service/service_task_metadata.go @@ -0,0 +1,35 @@ +package service + +import ( + "maps" + + "github.com/kandev/kandev/internal/task/models" +) + +// cloneTaskMetadata copies the top-level task metadata map so request-owned +// maps cannot be mutated while the service adds or protects server state. +func cloneTaskMetadata(metadata map[string]interface{}) map[string]interface{} { + if metadata == nil { + return nil + } + cloned := make(map[string]interface{}, len(metadata)) + maps.Copy(cloned, metadata) + return cloned +} + +// protectedTaskMetadataUpdate applies a generic metadata replacement while +// keeping the deferred launch intent owned by the server. The HTTP PATCH +// surface may replace ordinary metadata, but it cannot create, replace, or +// remove the launch record that carries deferred-start ownership. +func protectedTaskMetadataUpdate(existing, requested map[string]interface{}) map[string]interface{} { + updated := cloneTaskMetadata(requested) + if updated == nil { + updated = make(map[string]interface{}) + } + if deferred, ok := existing[models.MetaKeyDeferredLaunch]; ok { + updated[models.MetaKeyDeferredLaunch] = deferred + } else { + delete(updated, models.MetaKeyDeferredLaunch) + } + return updated +} diff --git a/apps/backend/internal/task/service/service_tasks.go b/apps/backend/internal/task/service/service_tasks.go index 93857b4116..a9e925f27e 100644 --- a/apps/backend/internal/task/service/service_tasks.go +++ b/apps/backend/internal/task/service/service_tasks.go @@ -5,7 +5,6 @@ import ( "context" "errors" "fmt" - "maps" "net/url" "os" "path/filepath" @@ -18,6 +17,7 @@ import ( v1 "github.com/kandev/kandev/pkg/api/v1" "go.uber.org/zap" + "github.com/kandev/kandev/internal/auth/authn" "github.com/kandev/kandev/internal/events" "github.com/kandev/kandev/internal/repoclone" "github.com/kandev/kandev/internal/steptelemetry" @@ -274,7 +274,7 @@ func (s *Service) prepareTaskForCreation(ctx context.Context, req *CreateTaskReq } workflowStepID := s.resolveWorkflowStep(ctx, req) - task := s.buildTask(req, workflowStepID) + task := s.buildTask(ctx, req, workflowStepID) task.ExternalID = externalID // Auto-assign identifier for office tasks @@ -723,7 +723,7 @@ func (s *Service) resolveWorkflowStep(ctx context.Context, req *CreateTaskReques } // buildTask constructs a Task model from the CreateTaskRequest. -func (s *Service) buildTask(req *CreateTaskRequest, workflowStepID string) *models.Task { +func (s *Service) buildTask(ctx context.Context, req *CreateTaskRequest, workflowStepID string) *models.Task { state := v1.TaskStateCreated if req.State != nil { state = *req.State @@ -743,19 +743,26 @@ func (s *Service) buildTask(req *CreateTaskRequest, workflowStepID string) *mode // so callers (e.g. onboarding) can omit it. priority = defaultPriority } - metadata := req.Metadata + metadata := cloneTaskMetadata(req.Metadata) + delete(metadata, models.MetaKeyDeferredLaunch) if req.DeferredLaunch != nil { if metadata == nil { metadata = make(map[string]interface{}) } - launch := req.DeferredLaunch + launch := cloneTaskMetadata(req.DeferredLaunch) + delete(launch, models.DeferredLaunchUserIDKey) + delete(launch, models.DeferredLaunchRecordRecentUseKey) + if req.RecordAgentProfileRecentUse { + if identity, ok := authn.IdentityFromContext(ctx); ok && identity.UserID != "" { + launch[models.DeferredLaunchUserIDKey] = identity.UserID + launch[models.DeferredLaunchRecordRecentUseKey] = true + } + } if ResolveStartWhenUnblocked(req) { // Mark the intent as a dependency-chain step. The record is the same // one WIP overflow persists — reused so "launch exactly once" and // restart survival are inherited — and the flag is what lets // dependency resolution recognise its own intents. - launch = make(map[string]interface{}, len(req.DeferredLaunch)+1) - maps.Copy(launch, req.DeferredLaunch) launch[models.DeferredLaunchStartWhenUnblockedKey] = true } metadata[models.MetaKeyDeferredLaunch] = launch @@ -1650,7 +1657,7 @@ func (s *Service) UpdateTask(ctx context.Context, id string, req *UpdateTaskRequ task.Position = *req.Position } if req.Metadata != nil { - task.Metadata = req.Metadata + task.Metadata = protectedTaskMetadataUpdate(task.Metadata, req.Metadata) } if req.Title != nil { task.Title = *req.Title diff --git a/apps/backend/internal/task/service/service_workflow.go b/apps/backend/internal/task/service/service_workflow.go index 8a5cd747e1..abbc5556c9 100644 --- a/apps/backend/internal/task/service/service_workflow.go +++ b/apps/backend/internal/task/service/service_workflow.go @@ -370,6 +370,11 @@ func (s *Service) UpdateTaskMetadata(ctx context.Context, id string, metadata ma task.Metadata = make(map[string]interface{}) } for k, v := range metadata { + // Deferred launch ownership is server-managed. Preserve it even if a + // future metadata endpoint forwards the whole request map here. + if k == models.MetaKeyDeferredLaunch { + continue + } task.Metadata[k] = v } task.UpdatedAt = time.Now().UTC() diff --git a/apps/backend/internal/user/controller/controller.go b/apps/backend/internal/user/controller/controller.go index 3784941fd7..c3aac3e95d 100644 --- a/apps/backend/internal/user/controller/controller.go +++ b/apps/backend/internal/user/controller/controller.go @@ -5,6 +5,7 @@ import ( "runtime" "github.com/kandev/kandev/internal/user/dto" + "github.com/kandev/kandev/internal/user/models" "github.com/kandev/kandev/internal/user/service" ) @@ -47,6 +48,38 @@ func (c *Controller) GetUserSettings(ctx context.Context) (dto.UserSettingsRespo }, nil } +// GetAgentProfileRecentUse returns the current user's independent operational +// profile histories. +func (c *Controller) GetAgentProfileRecentUse(ctx context.Context) ([]dto.AgentProfileRecentUseDTO, error) { + records, err := c.svc.GetAgentProfileRecentUse(ctx) + if err != nil { + return nil, err + } + result := make([]dto.AgentProfileRecentUseDTO, 0, len(records)) + for _, record := range records { + result = append(result, dto.FromAgentProfileRecentUse(record)) + } + return result, nil +} + +// RecordAgentProfileRecentUse records one successful profile launch in the +// requested operational selector context. +func (c *Controller) RecordAgentProfileRecentUse( + ctx context.Context, + contextValue string, + req dto.RecordAgentProfileRecentUseRequest, +) (dto.AgentProfileRecentUseDTO, error) { + record, err := c.svc.RecordAgentProfileRecentUse( + ctx, + models.AgentProfileRecentUseContext(contextValue), + req.AgentProfileID, + ) + if err != nil { + return dto.AgentProfileRecentUseDTO{}, err + } + return dto.FromAgentProfileRecentUse(record), nil +} + // UpdateUserSettings applies a partial settings patch and returns the // resulting settings with OS shell options. func (c *Controller) UpdateUserSettings(ctx context.Context, req dto.UpdateUserSettingsRequest) (dto.UserSettingsResponse, error) { diff --git a/apps/backend/internal/user/dto/dto.go b/apps/backend/internal/user/dto/dto.go index 5bf484c906..ea69dca3e5 100644 --- a/apps/backend/internal/user/dto/dto.go +++ b/apps/backend/internal/user/dto/dto.go @@ -88,6 +88,31 @@ type UserSettingsResponse struct { ShellOptions []ShellOption `json:"shell_options"` } +type AgentProfileRecentUseDTO struct { + Context models.AgentProfileRecentUseContext `json:"context"` + ProfileIDs []string `json:"profile_ids"` + Revision int64 `json:"revision"` + UpdatedAt string `json:"updated_at"` +} + +// FromAgentProfileRecentUse maps the persisted context history to its API +// representation without exposing the owning user id. +func FromAgentProfileRecentUse(record *models.AgentProfileRecentUse) AgentProfileRecentUseDTO { + if record == nil { + return AgentProfileRecentUseDTO{} + } + return AgentProfileRecentUseDTO{ + Context: record.Context, + ProfileIDs: append([]string{}, record.ProfileIDs...), + Revision: record.Revision, + UpdatedAt: record.UpdatedAt.Format(time.RFC3339), + } +} + +type RecordAgentProfileRecentUseRequest struct { + AgentProfileID string `json:"agent_profile_id"` +} + type ShellOption struct { Value string `json:"value"` Label string `json:"label"` diff --git a/apps/backend/internal/user/handlers/agent_profile_recent_use_test.go b/apps/backend/internal/user/handlers/agent_profile_recent_use_test.go new file mode 100644 index 0000000000..8b324a283a --- /dev/null +++ b/apps/backend/internal/user/handlers/agent_profile_recent_use_test.go @@ -0,0 +1,64 @@ +package handlers + +import ( + "bytes" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" +) + +func TestHTTPAgentProfileRecentUseRoundTrip(t *testing.T) { + router := newTestUserSettingsRouter(t) + request := httptest.NewRequest( + http.MethodPut, + "/api/v1/user/agent-profile-recent-use/task_create", + bytes.NewReader([]byte(`{"agent_profile_id":"profile-a"}`)), + ) + request.Header.Set("Content-Type", "application/json") + response := httptest.NewRecorder() + router.ServeHTTP(response, request) + if response.Code != http.StatusOK { + t.Fatalf("PUT recent-use status = %d, want %d: %s", response.Code, http.StatusOK, response.Body.String()) + } + + getResponse := httptest.NewRecorder() + router.ServeHTTP(getResponse, httptest.NewRequest( + http.MethodGet, + "/api/v1/user/agent-profile-recent-use", + nil, + )) + if getResponse.Code != http.StatusOK { + t.Fatalf("GET recent-use status = %d, want %d: %s", getResponse.Code, http.StatusOK, getResponse.Body.String()) + } + var records []struct { + Context string `json:"context"` + ProfileIDs []string `json:"profile_ids"` + Revision int64 `json:"revision"` + } + if err := json.Unmarshal(getResponse.Body.Bytes(), &records); err != nil { + t.Fatalf("decode GET recent-use: %v", err) + } + if len(records) != 1 || records[0].Context != "task_create" || records[0].Revision != 1 { + t.Fatalf("GET records = %+v, want one revision-one task-create record", records) + } + if len(records[0].ProfileIDs) != 1 || records[0].ProfileIDs[0] != "profile-a" { + t.Fatalf("profile ids = %v, want [profile-a]", records[0].ProfileIDs) + } +} + +func TestHTTPAgentProfileRecentUseRejectsInvalidInput(t *testing.T) { + router := newTestUserSettingsRouter(t) + for _, path := range []string{ + "/api/v1/user/agent-profile-recent-use/unknown", + "/api/v1/user/agent-profile-recent-use/task_create", + } { + request := httptest.NewRequest(http.MethodPut, path, bytes.NewReader([]byte(`{"agent_profile_id":""}`))) + request.Header.Set("Content-Type", "application/json") + response := httptest.NewRecorder() + router.ServeHTTP(response, request) + if response.Code != http.StatusBadRequest { + t.Fatalf("PUT %s status = %d, want %d", path, response.Code, http.StatusBadRequest) + } + } +} diff --git a/apps/backend/internal/user/handlers/handlers.go b/apps/backend/internal/user/handlers/handlers.go index 8f4d3bba89..1e15f22dca 100644 --- a/apps/backend/internal/user/handlers/handlers.go +++ b/apps/backend/internal/user/handlers/handlers.go @@ -38,6 +38,8 @@ func (h *Handlers) registerHTTP(router *gin.Engine) { api.GET("/user", h.httpGetUser) api.GET("/user/settings", h.httpGetUserSettings) api.PATCH("/user/settings", h.httpUpdateUserSettings) + api.GET("/user/agent-profile-recent-use", h.httpGetAgentProfileRecentUse) + api.PUT("/user/agent-profile-recent-use/:context", h.httpRecordAgentProfileRecentUse) } func (h *Handlers) registerWS(dispatcher *ws.Dispatcher) { @@ -65,6 +67,49 @@ func (h *Handlers) httpGetUserSettings(c *gin.Context) { c.JSON(http.StatusOK, resp) } +func (h *Handlers) httpGetAgentProfileRecentUse(c *gin.Context) { + records, err := h.controller.GetAgentProfileRecentUse(c.Request.Context()) + if err != nil { + h.logger.Error("failed to get agent profile recent use", zap.Error(err)) + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to get agent profile recent use"}) + return + } + c.JSON(http.StatusOK, records) +} + +const maxRecordAgentProfileRecentUseBodyBytes = 4 * 1024 + +func (h *Handlers) httpRecordAgentProfileRecentUse(c *gin.Context) { + c.Request.Body = http.MaxBytesReader(c.Writer, c.Request.Body, maxRecordAgentProfileRecentUseBodyBytes) + var req dto.RecordAgentProfileRecentUseRequest + if err := c.ShouldBindJSON(&req); err != nil { + var maxBytesError *http.MaxBytesError + if errors.As(err, &maxBytesError) { + c.JSON(http.StatusRequestEntityTooLarge, gin.H{"error": "request body too large"}) + return + } + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid payload"}) + return + } + record, err := h.controller.RecordAgentProfileRecentUse( + c.Request.Context(), + c.Param("context"), + req, + ) + if err != nil { + status := http.StatusInternalServerError + if errors.Is(err, service.ErrValidation) { + status = http.StatusBadRequest + } else if errors.Is(err, service.ErrAgentProfileRecentUseConflict) { + status = http.StatusConflict + } + h.logger.Error("failed to record agent profile recent use", zap.Error(err)) + c.JSON(status, gin.H{"error": "failed to record agent profile recent use"}) + return + } + c.JSON(http.StatusOK, record) +} + // maxUpdateUserSettingsBodyBytes bounds the update-settings request body via // http.MaxBytesReader, writing the 413 response below when exceeded. The // settings payload can legitimately carry several capped-but-large fields diff --git a/apps/backend/internal/user/models/models.go b/apps/backend/internal/user/models/models.go index 871e2130bc..04e7a5a485 100644 --- a/apps/backend/internal/user/models/models.go +++ b/apps/backend/internal/user/models/models.go @@ -234,3 +234,54 @@ type TaskCreateLastUsed struct { ExecutorProfileID string `json:"executor_profile_id"` WorkflowIDsByWorkspace map[string]string `json:"workflow_ids_by_workspace"` } + +// AgentProfileRecentUseContext identifies an operational selector whose +// profile order is remembered independently from user settings. +type AgentProfileRecentUseContext string + +const ( + AgentProfileRecentUseTaskCreate AgentProfileRecentUseContext = "task_create" + AgentProfileRecentUseTaskSession AgentProfileRecentUseContext = "task_session" + AgentProfileRecentUseQuickChat AgentProfileRecentUseContext = "quick_chat" + AgentProfileRecentUseConfigChat AgentProfileRecentUseContext = "config_chat" +) + +const ( + AgentProfileRecentUseMaxProfiles = 10 + AgentProfileRecentUseMaxContexts = 4 + AgentProfileRecentUseMaxProfileIDBytes = 255 +) + +// SupportedAgentProfileRecentUseContexts returns the closed set of contexts +// that may be persisted for a user. +func SupportedAgentProfileRecentUseContexts() []AgentProfileRecentUseContext { + return []AgentProfileRecentUseContext{ + AgentProfileRecentUseTaskCreate, + AgentProfileRecentUseTaskSession, + AgentProfileRecentUseQuickChat, + AgentProfileRecentUseConfigChat, + } +} + +// IsAgentProfileRecentUseContext reports whether context is supported. +func IsAgentProfileRecentUseContext(context AgentProfileRecentUseContext) bool { + switch context { + case AgentProfileRecentUseTaskCreate, + AgentProfileRecentUseTaskSession, + AgentProfileRecentUseQuickChat, + AgentProfileRecentUseConfigChat: + return true + default: + return false + } +} + +// AgentProfileRecentUse is the persisted move-to-front history for one user +// and operational selector context. +type AgentProfileRecentUse struct { + UserID string `json:"user_id"` + Context AgentProfileRecentUseContext `json:"context"` + ProfileIDs []string `json:"profile_ids"` + Revision int64 `json:"revision"` + UpdatedAt time.Time `json:"updated_at"` +} diff --git a/apps/backend/internal/user/service/agent_profile_recent_use.go b/apps/backend/internal/user/service/agent_profile_recent_use.go new file mode 100644 index 0000000000..60c0087e3f --- /dev/null +++ b/apps/backend/internal/user/service/agent_profile_recent_use.go @@ -0,0 +1,138 @@ +package service + +import ( + "context" + "database/sql" + "errors" + "fmt" + "time" + + "github.com/kandev/kandev/internal/events" + "github.com/kandev/kandev/internal/events/bus" + "github.com/kandev/kandev/internal/user/models" + "github.com/kandev/kandev/internal/user/store" + "go.uber.org/zap" +) + +const maxAgentProfileRecentUseCASAttempts = 3 + +var errAgentProfileRecentUseRepositoryUnavailable = errors.New("agent profile recent-use repository unavailable") + +// GetAgentProfileRecentUse returns the persisted operational profile histories +// for the current user. Missing contexts are omitted and read as empty. +func (s *Service) GetAgentProfileRecentUse(ctx context.Context) ([]*models.AgentProfileRecentUse, error) { + if s.recentUseRepo == nil { + return nil, errAgentProfileRecentUseRepositoryUnavailable + } + return s.recentUseRepo.ListAgentProfileRecentUse(ctx, s.settingsUserID(ctx)) +} + +// RecordAgentProfileRecentUse moves a successfully used profile to the front +// of its context history. It retries a bounded number of stale revisions so +// concurrent clients converge without making the launch path synchronous. +func (s *Service) RecordAgentProfileRecentUse( + ctx context.Context, + contextValue models.AgentProfileRecentUseContext, + profileID string, +) (*models.AgentProfileRecentUse, error) { + if err := validateAgentProfileRecentUseInput(contextValue, profileID); err != nil { + return nil, err + } + if s.recentUseRepo == nil { + return nil, errAgentProfileRecentUseRepositoryUnavailable + } + userID := s.settingsUserID(ctx) + for attempt := 0; attempt < maxAgentProfileRecentUseCASAttempts; attempt++ { + current, err := s.currentAgentProfileRecentUse(ctx, userID, contextValue) + if err != nil { + return nil, err + } + if len(current.ProfileIDs) > 0 && current.ProfileIDs[0] == profileID { + return current, nil + } + next := buildAgentProfileRecentUse(current, profileID) + updated, err := s.recentUseRepo.UpsertAgentProfileRecentUse(ctx, next, current.Revision) + if errors.Is(err, store.ErrAgentProfileRecentUseRevisionConflict) { + continue + } + if err != nil { + return nil, err + } + s.publishAgentProfileRecentUseEvent(ctx, updated) + return updated, nil + } + return nil, store.ErrAgentProfileRecentUseRevisionConflict +} + +func validateAgentProfileRecentUseInput( + contextValue models.AgentProfileRecentUseContext, + profileID string, +) error { + if !models.IsAgentProfileRecentUseContext(contextValue) { + return fmt.Errorf("%w: unsupported context", ErrValidation) + } + if profileID == "" || len(profileID) > models.AgentProfileRecentUseMaxProfileIDBytes { + return fmt.Errorf("%w: profile id must be between 1 and %d bytes", ErrValidation, models.AgentProfileRecentUseMaxProfileIDBytes) + } + return nil +} + +func (s *Service) currentAgentProfileRecentUse( + ctx context.Context, + userID string, + contextValue models.AgentProfileRecentUseContext, +) (*models.AgentProfileRecentUse, error) { + record, err := s.recentUseRepo.GetAgentProfileRecentUse(ctx, userID, contextValue) + if errors.Is(err, sql.ErrNoRows) { + return &models.AgentProfileRecentUse{ + UserID: userID, + Context: contextValue, + ProfileIDs: []string{}, + }, nil + } + return record, err +} + +func buildAgentProfileRecentUse( + current *models.AgentProfileRecentUse, + profileID string, +) *models.AgentProfileRecentUse { + profileIDs := make([]string, 0, models.AgentProfileRecentUseMaxProfiles) + profileIDs = append(profileIDs, profileID) + for _, existingID := range current.ProfileIDs { + if existingID == profileID || len(profileIDs) == models.AgentProfileRecentUseMaxProfiles { + continue + } + profileIDs = append(profileIDs, existingID) + } + return &models.AgentProfileRecentUse{ + UserID: current.UserID, + Context: current.Context, + ProfileIDs: profileIDs, + Revision: current.Revision + 1, + UpdatedAt: time.Now().UTC(), + } +} + +func (s *Service) publishAgentProfileRecentUseEvent( + ctx context.Context, + record *models.AgentProfileRecentUse, +) { + if s.eventBus == nil || record == nil { + return + } + data := map[string]interface{}{ + "user_id": record.UserID, + "context": record.Context, + "profile_ids": record.ProfileIDs, + "revision": record.Revision, + "updated_at": record.UpdatedAt.Format(time.RFC3339), + } + if err := s.eventBus.Publish( + ctx, + events.UserAgentProfileRecentUseUpdated, + bus.NewEvent(events.UserAgentProfileRecentUseUpdated, "user-service", data), + ); err != nil { + s.logger.Error("failed to publish agent profile recent-use event", zap.Error(err)) + } +} diff --git a/apps/backend/internal/user/service/agent_profile_recent_use_test.go b/apps/backend/internal/user/service/agent_profile_recent_use_test.go new file mode 100644 index 0000000000..d2b056142c --- /dev/null +++ b/apps/backend/internal/user/service/agent_profile_recent_use_test.go @@ -0,0 +1,162 @@ +package service + +import ( + "context" + "database/sql" + "errors" + "reflect" + "testing" + "time" + + "github.com/kandev/kandev/internal/common/logger" + eventtypes "github.com/kandev/kandev/internal/events" + "github.com/kandev/kandev/internal/user/models" + "github.com/kandev/kandev/internal/user/store" + "go.uber.org/zap" +) + +type recentUseRepository struct { + store.Repository + records map[models.AgentProfileRecentUseContext]*models.AgentProfileRecentUse + upsertCalls int + conflictsLeft int +} + +func (r *recentUseRepository) GetAgentProfileRecentUse( + _ context.Context, + _ string, + contextValue models.AgentProfileRecentUseContext, +) (*models.AgentProfileRecentUse, error) { + record := r.records[contextValue] + if record == nil { + return nil, sql.ErrNoRows + } + copy := *record + copy.ProfileIDs = append([]string(nil), record.ProfileIDs...) + return ©, nil +} + +func (r *recentUseRepository) ListAgentProfileRecentUse( + _ context.Context, + _ string, +) ([]*models.AgentProfileRecentUse, error) { + records := make([]*models.AgentProfileRecentUse, 0, len(r.records)) + for _, record := range r.records { + copy := *record + copy.ProfileIDs = append([]string(nil), record.ProfileIDs...) + records = append(records, ©) + } + return records, nil +} + +func (r *recentUseRepository) UpsertAgentProfileRecentUse( + _ context.Context, + record *models.AgentProfileRecentUse, + expectedRevision int64, +) (*models.AgentProfileRecentUse, error) { + r.upsertCalls++ + if r.conflictsLeft > 0 { + r.conflictsLeft-- + return nil, store.ErrAgentProfileRecentUseRevisionConflict + } + if current := r.records[record.Context]; current != nil && current.Revision != expectedRevision { + return nil, store.ErrAgentProfileRecentUseRevisionConflict + } + copy := *record + copy.ProfileIDs = append([]string(nil), record.ProfileIDs...) + copy.UpdatedAt = time.Unix(int64(r.upsertCalls), 0).UTC() + r.records[record.Context] = © + return ©, nil +} + +func newRecentUseService(t *testing.T, repo *recentUseRepository, eventBus *recordingEventBus) *Service { + t.Helper() + log, err := logger.NewFromZap(zap.NewNop()) + if err != nil { + t.Fatalf("create logger: %v", err) + } + return NewService(repo, eventBus, log) +} + +func TestRecordAgentProfileRecentUseMovesToFrontAndSuppressesNoop(t *testing.T) { + repo := &recentUseRepository{records: make(map[models.AgentProfileRecentUseContext]*models.AgentProfileRecentUse)} + events := &recordingEventBus{} + svc := newRecentUseService(t, repo, events) + ctx := context.Background() + contextValue := models.AgentProfileRecentUseTaskCreate + + for _, profileID := range []string{"profile-a", "profile-b", "profile-c"} { + if _, err := svc.RecordAgentProfileRecentUse(ctx, contextValue, profileID); err != nil { + t.Fatalf("record %q: %v", profileID, err) + } + } + if _, err := svc.RecordAgentProfileRecentUse(ctx, contextValue, "profile-b"); err != nil { + t.Fatalf("move profile-b: %v", err) + } + if _, err := svc.RecordAgentProfileRecentUse(ctx, contextValue, "profile-b"); err != nil { + t.Fatalf("repeat profile-b: %v", err) + } + + want := []string{"profile-b", "profile-c", "profile-a"} + if got := repo.records[contextValue].ProfileIDs; !reflect.DeepEqual(got, want) { + t.Fatalf("profile order = %v, want %v", got, want) + } + if repo.upsertCalls != 4 { + t.Fatalf("upsert calls = %d, want 4 after no-op suppression", repo.upsertCalls) + } + if len(events.publishedEvents) != 4 { + t.Fatalf("published events = %d, want 4", len(events.publishedEvents)) + } + if events.publishedSubjects[0] != eventtypes.UserAgentProfileRecentUseUpdated { + t.Fatalf("event subject = %q, want %q", events.publishedSubjects[0], eventtypes.UserAgentProfileRecentUseUpdated) + } +} + +func TestRecordAgentProfileRecentUseCapsHistoryAndRetriesConflicts(t *testing.T) { + repo := &recentUseRepository{ + records: make(map[models.AgentProfileRecentUseContext]*models.AgentProfileRecentUse), + conflictsLeft: 1, + } + svc := newRecentUseService(t, repo, &recordingEventBus{}) + for index := 0; index < models.AgentProfileRecentUseMaxProfiles+2; index++ { + profileID := "profile-" + string(rune('a'+index)) + if _, err := svc.RecordAgentProfileRecentUse(context.Background(), models.AgentProfileRecentUseQuickChat, profileID); err != nil { + t.Fatalf("record %q: %v", profileID, err) + } + } + + record := repo.records[models.AgentProfileRecentUseQuickChat] + if len(record.ProfileIDs) != models.AgentProfileRecentUseMaxProfiles { + t.Fatalf("profile count = %d, want %d", len(record.ProfileIDs), models.AgentProfileRecentUseMaxProfiles) + } + if record.ProfileIDs[0] != "profile-l" { + t.Fatalf("leading profile = %q, want profile-l", record.ProfileIDs[0]) + } + if repo.upsertCalls != models.AgentProfileRecentUseMaxProfiles+3 { + t.Fatalf("upsert calls = %d, want one conflict retry", repo.upsertCalls) + } +} + +func TestRecordAgentProfileRecentUseValidatesContextAndID(t *testing.T) { + repo := &recentUseRepository{records: make(map[models.AgentProfileRecentUseContext]*models.AgentProfileRecentUse)} + svc := newRecentUseService(t, repo, &recordingEventBus{}) + tests := []struct { + name string + context models.AgentProfileRecentUseContext + profileID string + }{ + {name: "unknown context", context: "other", profileID: "profile-a"}, + {name: "empty id", context: models.AgentProfileRecentUseTaskSession, profileID: ""}, + {name: "long id", context: models.AgentProfileRecentUseTaskSession, profileID: string(make([]byte, models.AgentProfileRecentUseMaxProfileIDBytes+1))}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if _, err := svc.RecordAgentProfileRecentUse(context.Background(), tt.context, tt.profileID); !errors.Is(err, ErrValidation) { + t.Fatalf("error = %v, want validation error", err) + } + }) + } + if repo.upsertCalls != 0 { + t.Fatalf("upsert calls = %d, want 0 for invalid input", repo.upsertCalls) + } +} diff --git a/apps/backend/internal/user/service/service.go b/apps/backend/internal/user/service/service.go index 62a0c74690..329a45c178 100644 --- a/apps/backend/internal/user/service/service.go +++ b/apps/backend/internal/user/service/service.go @@ -22,9 +22,10 @@ import ( // ErrUserSettingsConflict marks an exhausted conditional settings update. var ( - ErrUserNotFound = errors.New("user not found") - ErrValidation = errors.New("validation error") - ErrUserSettingsConflict = store.ErrUserSettingsRevisionConflict + ErrUserNotFound = errors.New("user not found") + ErrValidation = errors.New("validation error") + ErrUserSettingsConflict = store.ErrUserSettingsRevisionConflict + ErrAgentProfileRecentUseConflict = store.ErrAgentProfileRecentUseRevisionConflict ) const ( @@ -35,10 +36,11 @@ const ( ) type Service struct { - repo store.Repository - eventBus bus.EventBus - logger *logger.Logger - defaultUser string + repo store.Repository + recentUseRepo store.AgentProfileRecentUseRepository + eventBus bus.EventBus + logger *logger.Logger + defaultUser string } type UpdateUserSettingsRequest struct { @@ -110,11 +112,18 @@ type SystemMetricsDisplaySettingsPatch struct { // NewService builds the user settings service with its repository, event bus, // and logger. func NewService(repo store.Repository, eventBus bus.EventBus, log *logger.Logger) *Service { + recentUseRepo, _ := repo.(store.AgentProfileRecentUseRepository) + if recentUseLogger, ok := repo.(interface { + SetAgentProfileRecentUseLogger(*logger.Logger) + }); ok { + recentUseLogger.SetAgentProfileRecentUseLogger(log.WithFields(zap.String("component", "user-store"))) + } return &Service{ - repo: repo, - eventBus: eventBus, - logger: log.WithFields(zap.String("component", "user-service")), - defaultUser: store.DefaultUserID, + repo: repo, + recentUseRepo: recentUseRepo, + eventBus: eventBus, + logger: log.WithFields(zap.String("component", "user-service")), + defaultUser: store.DefaultUserID, } } diff --git a/apps/backend/internal/user/store/agent_profile_recent_use.go b/apps/backend/internal/user/store/agent_profile_recent_use.go new file mode 100644 index 0000000000..949b8b4508 --- /dev/null +++ b/apps/backend/internal/user/store/agent_profile_recent_use.go @@ -0,0 +1,171 @@ +package store + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "fmt" + "time" + + "go.uber.org/zap" + + "github.com/kandev/kandev/internal/user/models" +) + +var _ AgentProfileRecentUseRepository = (*sqliteRepository)(nil) + +const agentProfileRecentUseColumns = `user_id, context, profile_ids, revision, updated_at` + +// GetAgentProfileRecentUse reads one user's context history. +func (r *sqliteRepository) GetAgentProfileRecentUse( + ctx context.Context, + userID string, + contextValue models.AgentProfileRecentUseContext, +) (*models.AgentProfileRecentUse, error) { + row := r.ro.QueryRowContext(ctx, r.ro.Rebind(` + SELECT `+agentProfileRecentUseColumns+` + FROM user_agent_profile_recent_use + WHERE user_id = ? AND context = ? + `), userID, contextValue) + return scanAgentProfileRecentUse(row) +} + +// ListAgentProfileRecentUse reads all persisted histories for one user. +func (r *sqliteRepository) ListAgentProfileRecentUse( + ctx context.Context, + userID string, +) ([]*models.AgentProfileRecentUse, error) { + rows, err := r.ro.QueryxContext(ctx, r.ro.Rebind(` + SELECT `+agentProfileRecentUseColumns+` + FROM user_agent_profile_recent_use + WHERE user_id = ? + ORDER BY context ASC + `), userID) + if err != nil { + return nil, err + } + defer func() { _ = rows.Close() }() + + records := make([]*models.AgentProfileRecentUse, 0, models.AgentProfileRecentUseMaxContexts) + for rows.Next() { + record, scanErr := scanAgentProfileRecentUse(rows) + if scanErr != nil { + // A malformed context row must not make the other independent + // selector histories unavailable. Skip only this row and let the + // cursor continue to the remaining contexts. + if r.recentUseLogger != nil { + fields := []zap.Field{ + zap.String("user_id", userID), + zap.Error(scanErr), + } + if record != nil && record.Context != "" { + fields = append(fields, zap.String("context", string(record.Context))) + } + r.recentUseLogger.Warn("skipping malformed agent profile recent-use row", fields...) + } + continue + } + records = append(records, record) + } + return records, rows.Err() +} + +// UpsertAgentProfileRecentUse conditionally writes one context history. A +// missing row is created only for revision zero; existing rows update only +// when their current revision equals expectedRevision. +func (r *sqliteRepository) UpsertAgentProfileRecentUse( + ctx context.Context, + record *models.AgentProfileRecentUse, + expectedRevision int64, +) (*models.AgentProfileRecentUse, error) { + if err := validateAgentProfileRecentUseRecord(record); err != nil { + return nil, err + } + if expectedRevision < 0 || record.Revision != expectedRevision+1 { + return nil, fmt.Errorf("invalid agent profile recent-use revision") + } + profileIDs, err := json.Marshal(record.ProfileIDs) + if err != nil { + return nil, fmt.Errorf("marshal agent profile recent-use: %w", err) + } + updatedAt := record.UpdatedAt + if updatedAt.IsZero() { + updatedAt = time.Now().UTC() + } + query := ` + INSERT INTO user_agent_profile_recent_use + (user_id, context, profile_ids, revision, updated_at) + VALUES (?, ?, ?, ?, ?) + ON CONFLICT (user_id, context) DO UPDATE SET + profile_ids = excluded.profile_ids, + revision = excluded.revision, + updated_at = excluded.updated_at + WHERE user_agent_profile_recent_use.revision = ? + RETURNING ` + agentProfileRecentUseColumns + args := []any{ + record.UserID, + record.Context, + string(profileIDs), + record.Revision, + updatedAt, + expectedRevision, + } + updated, err := scanAgentProfileRecentUse(r.db.QueryRowContext(ctx, r.db.Rebind(query), args...)) + if err == nil { + return updated, nil + } + if errors.Is(err, sql.ErrNoRows) { + return nil, ErrAgentProfileRecentUseRevisionConflict + } + return nil, err +} + +func validateAgentProfileRecentUseRecord(record *models.AgentProfileRecentUse) error { + if record == nil || record.UserID == "" { + return errors.New("agent profile recent-use user is required") + } + if !models.IsAgentProfileRecentUseContext(record.Context) { + return fmt.Errorf("unsupported agent profile recent-use context: %q", record.Context) + } + if len(record.ProfileIDs) > models.AgentProfileRecentUseMaxProfiles { + return fmt.Errorf("agent profile recent-use exceeds %d profiles", models.AgentProfileRecentUseMaxProfiles) + } + seen := make(map[string]struct{}, len(record.ProfileIDs)) + for _, profileID := range record.ProfileIDs { + if profileID == "" || len(profileID) > models.AgentProfileRecentUseMaxProfileIDBytes { + return errors.New("agent profile recent-use profile id is invalid") + } + if _, exists := seen[profileID]; exists { + return errors.New("agent profile recent-use profile ids must be distinct") + } + seen[profileID] = struct{}{} + } + return nil +} + +func scanAgentProfileRecentUse(scanner interface{ Scan(dest ...any) error }) (*models.AgentProfileRecentUse, error) { + record := &models.AgentProfileRecentUse{} + var contextValue string + var profileIDsJSON string + if err := scanner.Scan( + &record.UserID, + &contextValue, + &profileIDsJSON, + &record.Revision, + &record.UpdatedAt, + ); err != nil { + return nil, err + } + record.Context = models.AgentProfileRecentUseContext(contextValue) + if err := json.Unmarshal([]byte(profileIDsJSON), &record.ProfileIDs); err != nil { + return record, fmt.Errorf("decode agent profile recent-use: %w", err) + } + if record.ProfileIDs == nil { + record.ProfileIDs = []string{} + } + if err := validateAgentProfileRecentUseRecord(record); err != nil { + return record, err + } + return record, nil +} diff --git a/apps/backend/internal/user/store/agent_profile_recent_use_test.go b/apps/backend/internal/user/store/agent_profile_recent_use_test.go new file mode 100644 index 0000000000..9a88cd51ea --- /dev/null +++ b/apps/backend/internal/user/store/agent_profile_recent_use_test.go @@ -0,0 +1,200 @@ +package store + +import ( + "context" + "testing" + "time" + + "github.com/jmoiron/sqlx" + _ "github.com/mattn/go-sqlite3" + "go.uber.org/zap" + "go.uber.org/zap/zaptest/observer" + + commonlogger "github.com/kandev/kandev/internal/common/logger" + "github.com/kandev/kandev/internal/user/models" +) + +func TestSQLiteRepositoryCreatesAgentProfileRecentUseTable(t *testing.T) { + conn, err := sqlx.Open("sqlite3", ":memory:") + if err != nil { + t.Fatalf("open sqlite: %v", err) + } + conn.SetMaxOpenConns(1) + t.Cleanup(func() { _ = conn.Close() }) + + repo, cleanup, err := Provide(conn, conn) + if err != nil { + t.Fatalf("create repository: %v", err) + } + _ = repo + t.Cleanup(func() { _ = cleanup() }) + + var tableName string + err = conn.GetContext(context.Background(), &tableName, ` + SELECT name FROM sqlite_master + WHERE type = 'table' AND name = 'user_agent_profile_recent_use' + `) + if err != nil { + t.Fatalf("find recent-use table: %v", err) + } + if tableName != "user_agent_profile_recent_use" { + t.Fatalf("table name = %q, want recent-use table", tableName) + } +} + +func TestSQLiteRepositoryPersistsAgentProfileRecentUseWithCAS(t *testing.T) { + conn := openRecentUseSQLite(t) + repo, cleanup, err := Provide(conn, conn) + if err != nil { + t.Fatalf("create repository: %v", err) + } + t.Cleanup(func() { _ = cleanup() }) + + ctx := context.Background() + record := &models.AgentProfileRecentUse{ + UserID: DefaultUserID, + Context: models.AgentProfileRecentUseTaskSession, + ProfileIDs: []string{"profile-a", "profile-b"}, + Revision: 1, + UpdatedAt: time.Now().UTC(), + } + if _, err := repo.UpsertAgentProfileRecentUse(ctx, record, 0); err != nil { + t.Fatalf("insert recent-use record: %v", err) + } + if _, err := repo.UpsertAgentProfileRecentUse(ctx, &models.AgentProfileRecentUse{ + UserID: DefaultUserID, + Context: models.AgentProfileRecentUseTaskSession, + ProfileIDs: []string{"profile-c"}, + Revision: 1, + UpdatedAt: time.Now().UTC(), + }, 0); err != ErrAgentProfileRecentUseRevisionConflict { + t.Fatalf("stale upsert error = %v, want revision conflict", err) + } + updated, err := repo.UpsertAgentProfileRecentUse(ctx, &models.AgentProfileRecentUse{ + UserID: DefaultUserID, + Context: models.AgentProfileRecentUseTaskSession, + ProfileIDs: []string{"profile-c"}, + Revision: 2, + UpdatedAt: time.Now().UTC(), + }, 1) + if err != nil { + t.Fatalf("conditional update: %v", err) + } + if updated.Revision != 2 || len(updated.ProfileIDs) != 1 || updated.ProfileIDs[0] != "profile-c" { + t.Fatalf("updated record = %+v", updated) + } +} + +func TestSQLiteRepositoryRecentUseMigrationReplayAndCascade(t *testing.T) { + conn := openRecentUseSQLite(t) + repo, cleanup, err := Provide(conn, conn) + if err != nil { + t.Fatalf("create repository: %v", err) + } + t.Cleanup(func() { _ = cleanup() }) + ctx := context.Background() + if _, err := repo.UpsertAgentProfileRecentUse(ctx, &models.AgentProfileRecentUse{ + UserID: DefaultUserID, + Context: models.AgentProfileRecentUseQuickChat, + ProfileIDs: []string{"profile-a"}, + Revision: 1, + UpdatedAt: time.Now().UTC(), + }, 0); err != nil { + t.Fatalf("insert record: %v", err) + } + if _, err := newSQLiteRepositoryWithDB(conn, conn); err != nil { + t.Fatalf("replay schema: %v", err) + } + if err := conn.GetContext(ctx, new(int), `SELECT COUNT(1) FROM user_agent_profile_recent_use`); err != nil { + t.Fatalf("query replayed table: %v", err) + } + if _, err := conn.Exec(`DELETE FROM users WHERE id = ?`, DefaultUserID); err != nil { + t.Fatalf("delete user: %v", err) + } + var count int + if err := conn.GetContext(ctx, &count, `SELECT COUNT(1) FROM user_agent_profile_recent_use`); err != nil { + t.Fatalf("count cascaded rows: %v", err) + } + if count != 0 { + t.Fatalf("recent-use rows after user deletion = %d, want 0", count) + } +} + +func TestSQLiteRepositoryRejectsInvalidRecentUseJSON(t *testing.T) { + conn := openRecentUseSQLite(t) + repo, cleanup, err := Provide(conn, conn) + if err != nil { + t.Fatalf("create repository: %v", err) + } + t.Cleanup(func() { _ = cleanup() }) + if _, err := conn.Exec(`INSERT INTO user_agent_profile_recent_use + (user_id, context, profile_ids, revision, updated_at) VALUES (?, ?, ?, ?, ?)`, + DefaultUserID, models.AgentProfileRecentUseConfigChat, "{}", 1, time.Now().UTC()); err != nil { + t.Fatalf("insert invalid JSON: %v", err) + } + if _, err := repo.GetAgentProfileRecentUse(context.Background(), DefaultUserID, models.AgentProfileRecentUseConfigChat); err == nil { + t.Fatal("invalid profile_ids JSON was accepted") + } +} + +func TestSQLiteRepositoryListSkipsMalformedContextRows(t *testing.T) { + conn := openRecentUseSQLite(t) + repo, cleanup, err := Provide(conn, conn) + if err != nil { + t.Fatalf("create repository: %v", err) + } + t.Cleanup(func() { _ = cleanup() }) + core, logs := observer.New(zap.WarnLevel) + log, err := commonlogger.NewFromZap(zap.New(core)) + if err != nil { + t.Fatalf("create observer logger: %v", err) + } + repo.SetAgentProfileRecentUseLogger(log) + ctx := context.Background() + if _, err := repo.UpsertAgentProfileRecentUse(ctx, &models.AgentProfileRecentUse{ + UserID: DefaultUserID, + Context: models.AgentProfileRecentUseQuickChat, + ProfileIDs: []string{"profile-valid"}, + Revision: 1, + UpdatedAt: time.Now().UTC(), + }, 0); err != nil { + t.Fatalf("insert valid record: %v", err) + } + if _, err := conn.Exec(`INSERT INTO user_agent_profile_recent_use + (user_id, context, profile_ids, revision, updated_at) VALUES (?, ?, ?, ?, ?)`, + DefaultUserID, models.AgentProfileRecentUseConfigChat, "{}", 1, time.Now().UTC()); err != nil { + t.Fatalf("insert malformed record: %v", err) + } + + records, err := repo.ListAgentProfileRecentUse(ctx, DefaultUserID) + if err != nil { + t.Fatalf("list records: %v", err) + } + if len(records) != 1 { + t.Fatalf("listed records = %d, want one valid context", len(records)) + } + if records[0].Context != models.AgentProfileRecentUseQuickChat || records[0].ProfileIDs[0] != "profile-valid" { + t.Fatalf("listed record = %+v, want valid quick-chat context", records[0]) + } + entries := logs.FilterMessage("skipping malformed agent profile recent-use row").All() + if len(entries) != 1 { + t.Fatalf("malformed-row diagnostics = %d, want one", len(entries)) + } + if got := entries[0].ContextMap()["context"]; got != string(models.AgentProfileRecentUseConfigChat) { + t.Fatalf("malformed-row context = %v, want %q", got, models.AgentProfileRecentUseConfigChat) + } +} + +func openRecentUseSQLite(t *testing.T) *sqlx.DB { + t.Helper() + conn, err := sqlx.Open("sqlite3", ":memory:") + if err != nil { + t.Fatalf("open sqlite: %v", err) + } + conn.SetMaxOpenConns(1) + if _, err := conn.Exec("PRAGMA foreign_keys = ON"); err != nil { + t.Fatalf("enable foreign keys: %v", err) + } + t.Cleanup(func() { _ = conn.Close() }) + return conn +} diff --git a/apps/backend/internal/user/store/sqlite.go b/apps/backend/internal/user/store/sqlite.go index 089aea256a..cd5f370689 100644 --- a/apps/backend/internal/user/store/sqlite.go +++ b/apps/backend/internal/user/store/sqlite.go @@ -12,6 +12,7 @@ import ( "github.com/jmoiron/sqlx" + commonlogger "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/internal/db" "github.com/kandev/kandev/internal/db/dialect" "github.com/kandev/kandev/internal/user/models" @@ -26,9 +27,10 @@ const ( ) type sqliteRepository struct { - db *sqlx.DB // writer - ro *sqlx.DB // reader - ownsDB bool + db *sqlx.DB // writer + ro *sqlx.DB // reader + ownsDB bool + recentUseLogger *commonlogger.Logger } var _ Repository = (*sqliteRepository)(nil) @@ -70,6 +72,15 @@ func (r *sqliteRepository) initSchema() error { created_at TIMESTAMP NOT NULL, updated_at TIMESTAMP NOT NULL ); + CREATE TABLE IF NOT EXISTS user_agent_profile_recent_use ( + user_id TEXT NOT NULL, + context TEXT NOT NULL, + profile_ids TEXT NOT NULL DEFAULT '[]', + revision BIGINT NOT NULL DEFAULT 0, + updated_at TIMESTAMP NOT NULL, + PRIMARY KEY (user_id, context), + FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE + ); ` if _, err := r.db.Exec(schema); err != nil { return err @@ -123,6 +134,13 @@ func (r *sqliteRepository) Close() error { return r.db.Close() } +// SetAgentProfileRecentUseLogger wires diagnostics for malformed persisted +// profile-history rows. The user service installs its scoped logger during +// construction; the repository remains usable without one in low-level tests. +func (r *sqliteRepository) SetAgentProfileRecentUseLogger(log *commonlogger.Logger) { + r.recentUseLogger = log +} + const userColumns = "id, email, display_name, role, status, created_at, updated_at" // GetUser returns the user row for the given id. diff --git a/apps/backend/internal/user/store/store.go b/apps/backend/internal/user/store/store.go index b71804a4bc..ea66746c8a 100644 --- a/apps/backend/internal/user/store/store.go +++ b/apps/backend/internal/user/store/store.go @@ -11,6 +11,10 @@ import ( // matched no rows because the expected revision was stale. var ErrUserSettingsRevisionConflict = errors.New("user settings revision conflict") +// ErrAgentProfileRecentUseRevisionConflict reports a conditional recency +// write that matched no row because another client changed the context. +var ErrAgentProfileRecentUseRevisionConflict = errors.New("agent profile recent-use revision conflict") + type Repository interface { GetUser(ctx context.Context, id string) (*models.User, error) GetDefaultUser(ctx context.Context) (*models.User, error) @@ -20,6 +24,15 @@ type Repository interface { Close() error } +// AgentProfileRecentUseRepository is the independent persistence surface for +// bounded operational profile histories. It is separate from Repository so +// existing user-settings consumers and test fakes do not inherit this concern. +type AgentProfileRecentUseRepository interface { + GetAgentProfileRecentUse(ctx context.Context, userID string, context models.AgentProfileRecentUseContext) (*models.AgentProfileRecentUse, error) + ListAgentProfileRecentUse(ctx context.Context, userID string) ([]*models.AgentProfileRecentUse, error) + UpsertAgentProfileRecentUse(ctx context.Context, record *models.AgentProfileRecentUse, expectedRevision int64) (*models.AgentProfileRecentUse, error) +} + // AccountRepository is the account-management surface consumed by // internal/auth. It is intentionally separate from Repository so existing // Repository fakes/tests are unaffected by the auth feature; *sqliteRepository diff --git a/apps/backend/pkg/websocket/actions.go b/apps/backend/pkg/websocket/actions.go index 21d414c802..01c6c7a816 100644 --- a/apps/backend/pkg/websocket/actions.go +++ b/apps/backend/pkg/websocket/actions.go @@ -362,9 +362,10 @@ const ( ActionWorktreeReset = "worktree.reset" // Reset HEAD to a commit (soft/hard) // User actions - ActionUserGet = "user.get" - ActionUserSettingsUpdate = "user.settings.update" - ActionUserSettingsUpdated = "user.settings.updated" + ActionUserGet = "user.get" + ActionUserSettingsUpdate = "user.settings.update" + ActionUserSettingsUpdated = "user.settings.updated" + ActionUserAgentProfileRecentUseUpdated = "user.agent_profile_recent_use.updated" // ActionPluginUserStateUpdated notifies the writing user's other WS // connections that one of their per-user plugin storage keys changed diff --git a/apps/web/components/config-chat/config-chat-setup.tsx b/apps/web/components/config-chat/config-chat-setup.tsx index 07f49927e0..156e818511 100644 --- a/apps/web/components/config-chat/config-chat-setup.tsx +++ b/apps/web/components/config-chat/config-chat-setup.tsx @@ -10,6 +10,7 @@ import { useFeature } from "@/hooks/domains/features/use-feature"; import { isSelectableAgentProfile } from "@/lib/state/slices/settings/types"; import type { QuickChatSessionKind } from "@/lib/state/slices/ui/types"; import { ConfigurationChatToggle } from "@/components/quick-chat/configuration-chat-toggle"; +import { orderAgentProfilesByRecentUse } from "@/lib/agent-profile-recent-use"; /** * Catalog KEYS, not copy. Resolving these at module scope would freeze them at @@ -47,9 +48,13 @@ function ProfileSelector({ onSelect }: { onSelect: (id: string) => void }) { const profiles = useAppStore((state) => state.agentProfiles.items ?? []); const dynamicRoutingEnabled = useFeature("dynamicAgentRouting"); const { t } = useTranslation(); + const recentProfileIds = useAppStore( + (state) => state.agentProfileRecentUse?.records.config_chat?.profileIds, + ); const selectableProfiles = profiles.filter((profile) => isSelectableAgentProfile(profile, dynamicRoutingEnabled), ); + const orderedProfiles = orderAgentProfilesByRecentUse(selectableProfiles, recentProfileIds); return (
@@ -59,7 +64,7 @@ function ProfileSelector({ onSelect }: { onSelect: (id: string) => void }) {

{t("configChat:agentProfileHelp")}

- {selectableProfiles.map((profile) => ( + {orderedProfiles.map((profile) => (