diff --git a/apps/backend/internal/backendapp/gateway.go b/apps/backend/internal/backendapp/gateway.go index 7d97aed349..fdd8bd48e7 100644 --- a/apps/backend/internal/backendapp/gateway.go +++ b/apps/backend/internal/backendapp/gateway.go @@ -3,6 +3,7 @@ package backendapp import ( "context" "fmt" + "time" "go.uber.org/zap" @@ -299,19 +300,38 @@ func provideGateway( notificationSvc := notificationservice.NewService(notificationRepo, taskRepo, gateway.Hub, log) notificationCtrl := notificationcontroller.NewController(notificationSvc) if eventBus != nil { - _, err = eventBus.Subscribe(events.TaskSessionStateChanged, func(ctx context.Context, event *bus.Event) error { + _, err = eventBus.Subscribe(events.TurnCompleted, func(ctx context.Context, event *bus.Event) error { data, ok := event.Data.(map[string]interface{}) if !ok { return nil } taskID, _ := data["task_id"].(string) sessionID, _ := data["session_id"].(string) - newState, _ := data["new_state"].(string) - notificationSvc.HandleTaskSessionStateChanged(ctx, taskID, sessionID, newState) + turnID, _ := data["id"].(string) + if isAbandonedTurnCompletion(data) { + return nil + } + notificationSvc.HandleTaskTurnFinished(ctx, taskID, sessionID, turnID) return nil }) if err != nil { - log.Error("Failed to subscribe to task session notifications", zap.Error(err)) + log.Error("Failed to subscribe to turn notifications", zap.Error(err)) + } + + _, err = eventBus.Subscribe(events.MessageAdded, func(ctx context.Context, event *bus.Event) error { + data, ok := event.Data.(map[string]interface{}) + if !ok { + return nil + } + taskID, sessionID, pendingID, eligible := clarificationNotificationOccurrence(data) + if !eligible { + return nil + } + notificationSvc.HandleClarificationRequested(ctx, taskID, sessionID, pendingID) + return nil + }) + if err != nil { + log.Error("Failed to subscribe to clarification notifications", zap.Error(err)) } _, err = eventBus.Subscribe(events.OfficeInboxItem, func(ctx context.Context, event *bus.Event) error { @@ -340,6 +360,29 @@ func provideGateway( return gateway, notificationSvc, notificationCtrl, terminalSvc, nil } +func isAbandonedTurnCompletion(data map[string]interface{}) bool { + startedAt, started := data["started_at"].(time.Time) + completedAt, completed := data["completed_at"].(*time.Time) + return started && completed && completedAt != nil && startedAt.Equal(*completedAt) +} + +func clarificationNotificationOccurrence(data map[string]interface{}) (taskID, sessionID, pendingID string, ok bool) { + if data["type"] != "clarification_request" || data["requests_input"] != true || data["author_type"] != "agent" { + return "", "", "", false + } + metadata, metadataOK := data["metadata"].(map[string]interface{}) + pendingID, pendingOK := metadata["pending_id"].(string) + if !metadataOK || !pendingOK || pendingID == "" { + return "", "", "", false + } + taskID, taskOK := data["task_id"].(string) + sessionID, sessionOK := data["session_id"].(string) + if !taskOK || !sessionOK || taskID == "" || sessionID == "" { + return "", "", "", false + } + return taskID, sessionID, pendingID, true +} + type createdChangeAssociationRouter struct { resolveRepositoryID func(context.Context, string, string) string resolveWorkspaceID func(context.Context, string) (string, error) diff --git a/apps/backend/internal/backendapp/gateway_git_association_test.go b/apps/backend/internal/backendapp/gateway_git_association_test.go index 72a511c70a..7ec00cc064 100644 --- a/apps/backend/internal/backendapp/gateway_git_association_test.go +++ b/apps/backend/internal/backendapp/gateway_git_association_test.go @@ -4,8 +4,57 @@ import ( "context" "errors" "testing" + "time" ) +func TestIsAbandonedTurnCompletion(t *testing.T) { + startedAt := time.Date(2026, time.July, 24, 10, 0, 0, 0, time.UTC) + if !isAbandonedTurnCompletion(map[string]interface{}{"started_at": startedAt, "completed_at": &startedAt}) { + t.Fatal("an orphan turn completed at its start time must not notify") + } + completedAt := startedAt.Add(time.Second) + if isAbandonedTurnCompletion(map[string]interface{}{"started_at": startedAt, "completed_at": &completedAt}) { + t.Fatal("a live turn completion must remain eligible for notifications") + } +} + +func TestClarificationNotificationOccurrence(t *testing.T) { + tests := []struct { + name string + data map[string]interface{} + wantTaskID string + wantSessionID string + wantPendingID string + wantEligible bool + }{ + { + name: "structured clarification requesting input", + data: map[string]interface{}{ + "type": "clarification_request", "requests_input": true, "author_type": "agent", + "task_id": "task-1", "session_id": "session-1", "metadata": map[string]interface{}{"pending_id": "pending-1"}, + }, + wantTaskID: "task-1", wantSessionID: "session-1", wantPendingID: "pending-1", wantEligible: true, + }, + {name: "ordinary message", data: map[string]interface{}{"type": "message", "requests_input": true}, wantEligible: false}, + {name: "clarification without input request", data: map[string]interface{}{"type": "clarification_request", "requests_input": false}, wantEligible: false}, + {name: "missing metadata", data: map[string]interface{}{"type": "clarification_request", "requests_input": true}, wantEligible: false}, + {name: "wrong metadata type", data: map[string]interface{}{"type": "clarification_request", "requests_input": true, "metadata": "pending-1"}, wantEligible: false}, + {name: "empty pending ID", data: map[string]interface{}{"type": "clarification_request", "requests_input": true, "metadata": map[string]interface{}{"pending_id": ""}}, wantEligible: false}, + {name: "user authored clarification", data: map[string]interface{}{"type": "clarification_request", "requests_input": true, "author_type": "user", "task_id": "task-1", "session_id": "session-1", "metadata": map[string]interface{}{"pending_id": "pending-1"}}, wantEligible: false}, + {name: "system authored clarification", data: map[string]interface{}{"type": "clarification_request", "requests_input": true, "author_type": "system", "task_id": "task-1", "session_id": "session-1", "metadata": map[string]interface{}{"pending_id": "pending-1"}}, wantEligible: false}, + {name: "missing task ID", data: map[string]interface{}{"type": "clarification_request", "requests_input": true, "author_type": "agent", "session_id": "session-1", "metadata": map[string]interface{}{"pending_id": "pending-1"}}, wantEligible: false}, + {name: "missing session ID", data: map[string]interface{}{"type": "clarification_request", "requests_input": true, "author_type": "agent", "task_id": "task-1", "metadata": map[string]interface{}{"pending_id": "pending-1"}}, wantEligible: false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + taskID, sessionID, pendingID, eligible := clarificationNotificationOccurrence(tt.data) + if taskID != tt.wantTaskID || sessionID != tt.wantSessionID || pendingID != tt.wantPendingID || eligible != tt.wantEligible { + t.Fatalf("occurrence = (%q, %q, %q, %v), want (%q, %q, %q, %v)", taskID, sessionID, pendingID, eligible, tt.wantTaskID, tt.wantSessionID, tt.wantPendingID, tt.wantEligible) + } + }) + } +} + func TestCreatedChangeAssociationRouterRoutesGitLabSingleAndMultiRepo(t *testing.T) { var got []string router := createdChangeAssociationRouter{ diff --git a/apps/backend/internal/notifications/controller/controller.go b/apps/backend/internal/notifications/controller/controller.go index 18b4c59956..8f503cc0d6 100644 --- a/apps/backend/internal/notifications/controller/controller.go +++ b/apps/backend/internal/notifications/controller/controller.go @@ -48,11 +48,15 @@ func (c *Controller) CreateProvider(ctx context.Context, req dto.UpsertProviderR } providerType := models.ProviderType(req.Type) config := req.Config - provider, err := c.service.CreateProvider(ctx, userID, name, providerType, config, enabled, req.Events) + events := []string{service.EventTaskSessionClarificationAsked} + if req.Events != nil { + events = *req.Events + } + provider, err := c.service.CreateProvider(ctx, userID, name, providerType, config, enabled, events) if err != nil { return dto.NotificationProviderDTO{}, err } - return dto.FromProvider(provider, req.Events), nil + return dto.FromProvider(provider, events), nil } func (c *Controller) UpdateProvider(ctx context.Context, providerID string, req dto.UpdateProviderRequest) (dto.NotificationProviderDTO, error) { diff --git a/apps/backend/internal/notifications/controller/controller_test.go b/apps/backend/internal/notifications/controller/controller_test.go new file mode 100644 index 0000000000..7ee7ed72b1 --- /dev/null +++ b/apps/backend/internal/notifications/controller/controller_test.go @@ -0,0 +1,103 @@ +package controller + +import ( + "context" + "encoding/json" + "testing" + + "github.com/kandev/kandev/internal/common/logger" + "github.com/kandev/kandev/internal/notifications/dto" + "github.com/kandev/kandev/internal/notifications/models" + "github.com/kandev/kandev/internal/notifications/service" + "go.uber.org/zap" +) + +func TestCreateProviderDefaultsOmittedEventsToClarificationOnly(t *testing.T) { + testCreateProviderEvents(t, decodeCreateProviderRequest(t, `{"type":"local"}`), []string{service.EventTaskSessionClarificationAsked}) +} + +func TestCreateProviderPreservesExplicitEmptyEvents(t *testing.T) { + testCreateProviderEvents(t, decodeCreateProviderRequest(t, `{"type":"local","events":[]}`), []string{}) +} + +func TestCreateProviderPreservesExplicitEvents(t *testing.T) { + testCreateProviderEvents(t, decodeCreateProviderRequest(t, `{"type":"local","events":["session.turn_finished"]}`), []string{service.EventTaskSessionTurnFinished}) +} + +func decodeCreateProviderRequest(t *testing.T, body string) dto.UpsertProviderRequest { + t.Helper() + var request dto.UpsertProviderRequest + if err := json.Unmarshal([]byte(body), &request); err != nil { + t.Fatalf("decode request: %v", err) + } + return request +} + +func testCreateProviderEvents(t *testing.T, request dto.UpsertProviderRequest, want []string) { + t.Helper() + log, err := logger.NewFromZap(zap.NewNop()) + if err != nil { + t.Fatalf("create logger: %v", err) + } + repo := &controllerRepository{} + controller := NewController(service.NewService(repo, nil, nil, log)) + provider, err := controller.CreateProvider(context.Background(), request) + if err != nil { + t.Fatalf("create provider: %v", err) + } + if got := repo.events[provider.ID]; !sameEvents(got, want) { + t.Fatalf("stored events = %#v, want %#v", got, want) + } +} + +func sameEvents(got, want []string) bool { + if len(got) != len(want) { + return false + } + for index := range got { + if got[index] != want[index] { + return false + } + } + return true +} + +type controllerRepository struct { + providers map[string]*models.Provider + events map[string][]string +} + +func (r *controllerRepository) CreateProvider(_ context.Context, provider *models.Provider) error { + if r.providers == nil { + r.providers = make(map[string]*models.Provider) + } + provider.ID = "provider-1" + r.providers[provider.ID] = provider + return nil +} + +func (r *controllerRepository) UpdateProvider(context.Context, *models.Provider) error { return nil } +func (r *controllerRepository) GetProvider(_ context.Context, id string) (*models.Provider, error) { + return r.providers[id], nil +} +func (r *controllerRepository) ListProvidersByUser(context.Context, string) ([]*models.Provider, error) { + return nil, nil +} +func (r *controllerRepository) DeleteProvider(context.Context, string) error { return nil } +func (r *controllerRepository) ListSubscriptionsByProvider(context.Context, string) ([]*models.Subscription, error) { + return nil, nil +} +func (r *controllerRepository) ReplaceSubscriptions(_ context.Context, providerID, _ string, events []string) error { + if r.events == nil { + r.events = make(map[string][]string) + } + r.events[providerID] = append([]string(nil), events...) + return nil +} +func (r *controllerRepository) InsertDelivery(context.Context, *models.Delivery) (bool, error) { + return false, nil +} +func (r *controllerRepository) DeleteDelivery(context.Context, string, string, string) error { + return nil +} +func (r *controllerRepository) Close() error { return nil } diff --git a/apps/backend/internal/notifications/dto/dto.go b/apps/backend/internal/notifications/dto/dto.go index fa07b6f5b6..47b2e238ce 100644 --- a/apps/backend/internal/notifications/dto/dto.go +++ b/apps/backend/internal/notifications/dto/dto.go @@ -28,7 +28,7 @@ type UpsertProviderRequest struct { Type string `json:"type"` Config map[string]interface{} `json:"config,omitempty"` Enabled *bool `json:"enabled,omitempty"` - Events []string `json:"events,omitempty"` + Events *[]string `json:"events,omitempty"` } type UpdateProviderRequest struct { diff --git a/apps/backend/internal/notifications/models/models.go b/apps/backend/internal/notifications/models/models.go index aa494a5c2d..c39fabcdc8 100644 --- a/apps/backend/internal/notifications/models/models.go +++ b/apps/backend/internal/notifications/models/models.go @@ -37,5 +37,6 @@ type Delivery struct { ProviderID string `json:"provider_id"` EventType string `json:"event_type"` TaskSessionID string `json:"session_id"` + OccurrenceID string `json:"occurrence_id"` CreatedAt time.Time `json:"created_at"` } diff --git a/apps/backend/internal/notifications/providers/local.go b/apps/backend/internal/notifications/providers/local.go index f2b1686540..3ce148d7e2 100644 --- a/apps/backend/internal/notifications/providers/local.go +++ b/apps/backend/internal/notifications/providers/local.go @@ -28,15 +28,17 @@ func (p *LocalProvider) Send(_ context.Context, message Message) error { if p.hub == nil { return fmt.Errorf("websocket hub not available") } - msg, err := ws.NewNotification(ws.ActionSessionWaitingForInput, map[string]interface{}{ - "task_id": message.TaskID, - "session_id": message.TaskSessionID, - "title": message.Title, - "body": message.Body, + msg, err := ws.NewNotification(message.EventType, map[string]interface{}{ + "task_id": message.TaskID, + "session_id": message.TaskSessionID, + "occurrence_id": message.OccurrenceID, + "title": message.Title, + "body": message.Body, }) if err != nil { return err } + msg.ID = message.OccurrenceID p.hub.BroadcastToUser(message.UserID, msg) return nil } diff --git a/apps/backend/internal/notifications/providers/provider.go b/apps/backend/internal/notifications/providers/provider.go index 16591dfe24..5e8deb4c03 100644 --- a/apps/backend/internal/notifications/providers/provider.go +++ b/apps/backend/internal/notifications/providers/provider.go @@ -10,6 +10,7 @@ type Message struct { Body string TaskID string TaskSessionID string + OccurrenceID string UserID string Config map[string]interface{} } diff --git a/apps/backend/internal/notifications/service/service.go b/apps/backend/internal/notifications/service/service.go index b374f7e349..0694fc8891 100644 --- a/apps/backend/internal/notifications/service/service.go +++ b/apps/backend/internal/notifications/service/service.go @@ -20,9 +20,10 @@ import ( ) const ( - EventTaskSessionWaitingForInput = "session.waiting_for_input" - EventOfficeInboxItem = "office.inbox_item" - desktopNativeNotificationsEnv = "KANDEV_DESKTOP_NATIVE_NOTIFICATIONS" + EventTaskSessionTurnFinished = "session.turn_finished" + EventTaskSessionClarificationAsked = "session.clarification_requested" + EventOfficeInboxItem = "office.inbox_item" + desktopNativeNotificationsEnv = "KANDEV_DESKTOP_NATIVE_NOTIFICATIONS" ) var ErrProviderNotFound = errors.New("notification provider not found") @@ -66,7 +67,7 @@ func (s *Service) AppriseAvailable() bool { } func (s *Service) AvailableEvents() []string { - return []string{EventTaskSessionWaitingForInput, EventOfficeInboxItem} + return []string{EventTaskSessionTurnFinished, EventTaskSessionClarificationAsked, EventOfficeInboxItem} } func (s *Service) ListProviders(ctx context.Context, userID string) ([]*models.Provider, map[string][]string, error) { @@ -166,8 +167,16 @@ type ProviderUpdate struct { Events *[]string } -func (s *Service) HandleTaskSessionStateChanged(ctx context.Context, taskID, sessionID, newState string) { - if newState != "WAITING_FOR_INPUT" { +func (s *Service) HandleTaskTurnFinished(ctx context.Context, taskID, sessionID, turnID string) { + s.handleSemanticOccurrence(ctx, taskID, sessionID, turnID, EventTaskSessionTurnFinished) +} + +func (s *Service) HandleClarificationRequested(ctx context.Context, taskID, sessionID, pendingID string) { + s.handleSemanticOccurrence(ctx, taskID, sessionID, pendingID, EventTaskSessionClarificationAsked) +} + +func (s *Service) handleSemanticOccurrence(ctx context.Context, taskID, sessionID, occurrenceID, eventType string) { + if occurrenceID == "" { return } userID := userstore.DefaultUserID @@ -176,20 +185,21 @@ func (s *Service) HandleTaskSessionStateChanged(ctx context.Context, taskID, ses s.logger.Error("failed to load notification providers", zap.Error(err)) return } - title, body := s.buildWaitingForInputMessage(ctx, taskID) + title, body := s.buildSemanticMessage(ctx, taskID, eventType) for _, provider := range providers { if !provider.Enabled { continue } events := subscriptions[provider.ID] - if !containsEvent(events, EventTaskSessionWaitingForInput) { + if !containsEvent(events, eventType) { continue } delivery := &models.Delivery{ UserID: userID, ProviderID: provider.ID, - EventType: EventTaskSessionWaitingForInput, + EventType: eventType, TaskSessionID: sessionID, + OccurrenceID: occurrenceID, } inserted, err := s.repo.InsertDelivery(ctx, delivery) if err != nil { @@ -199,14 +209,16 @@ func (s *Service) HandleTaskSessionStateChanged(ctx context.Context, taskID, ses if !inserted { continue } - if err := s.dispatchProvider(ctx, provider, waitingForInputPayload{ + if err := s.dispatchProvider(ctx, provider, notificationPayload{ TaskID: taskID, TaskSessionID: sessionID, + OccurrenceID: occurrenceID, + EventType: eventType, Title: title, Body: body, }); err != nil { s.logger.Warn("notification delivery failed", zap.String("provider_id", provider.ID), zap.Error(err)) - _ = s.repo.DeleteDelivery(ctx, provider.ID, EventTaskSessionWaitingForInput, sessionID) + _ = s.repo.DeleteDelivery(ctx, provider.ID, eventType, occurrenceID) } } } @@ -253,32 +265,34 @@ func (s *Service) dispatchGenericNotification(ctx context.Context, provider *mod }) } -type waitingForInputPayload struct { +type notificationPayload struct { TaskID string TaskSessionID string + OccurrenceID string + EventType string Title string Body string } -func (s *Service) dispatchProvider(ctx context.Context, provider *models.Provider, payload waitingForInputPayload) error { +func (s *Service) dispatchProvider(ctx context.Context, provider *models.Provider, payload notificationPayload) error { adapter := s.providers[provider.Type] if adapter == nil { return fmt.Errorf("unknown provider type: %s", provider.Type) } return adapter.Send(ctx, providers.Message{ - EventType: EventTaskSessionWaitingForInput, + EventType: payload.EventType, Title: payload.Title, Body: payload.Body, TaskID: payload.TaskID, TaskSessionID: payload.TaskSessionID, + OccurrenceID: payload.OccurrenceID, UserID: userstore.DefaultUserID, Config: provider.Config, }) } -func (s *Service) buildWaitingForInputMessage(ctx context.Context, taskID string) (string, string) { - title := "Task needs your input" - body := "An agent is waiting for your input." +func (s *Service) buildSemanticMessage(ctx context.Context, taskID, eventType string) (string, string) { + title, body := semanticMessageCopy(eventType, "") if taskID == "" || s.taskRepo == nil { return title, body } @@ -287,11 +301,24 @@ func (s *Service) buildWaitingForInputMessage(ctx context.Context, taskID string return title, body } if task.Title != "" { - body = fmt.Sprintf("An agent is waiting for your input on \"%s\".", task.Title) + _, body = semanticMessageCopy(eventType, task.Title) } return title, body } +func semanticMessageCopy(eventType, taskTitle string) (string, string) { + if eventType == EventTaskSessionClarificationAsked { + if taskTitle == "" { + return "Agent needs your answer", "The agent asked a question." + } + return "Agent needs your answer", fmt.Sprintf("The agent asked a question on \"%s\".", taskTitle) + } + if taskTitle == "" { + return "Agent turn finished", "The agent finished a turn." + } + return "Agent turn finished", fmt.Sprintf("The agent finished a turn on \"%s\".", taskTitle) +} + func (s *Service) ensureDefaultProviders(ctx context.Context, userID string) error { providers, err := s.repo.ListProvidersByUser(ctx, userID) if err != nil { @@ -320,7 +347,7 @@ func (s *Service) ensureDefaultProviders(ctx context.Context, userID string) err return err } if err := s.repo.ReplaceSubscriptions(ctx, provider.ID, userID, []string{ - EventTaskSessionWaitingForInput, + EventTaskSessionClarificationAsked, EventOfficeInboxItem, }); err != nil { return err @@ -354,7 +381,7 @@ func (s *Service) ensureSystemProvider(ctx context.Context, userID string) error return err } return s.repo.ReplaceSubscriptions(ctx, provider.ID, userID, []string{ - EventTaskSessionWaitingForInput, + EventTaskSessionClarificationAsked, EventOfficeInboxItem, }) } @@ -369,7 +396,7 @@ func (s *Service) TestProvider(ctx context.Context, providerID string) error { return fmt.Errorf("unknown provider type: %s", provider.Type) } return adapter.Send(ctx, providers.Message{ - EventType: EventTaskSessionWaitingForInput, + EventType: EventTaskSessionClarificationAsked, Title: "Test notification", Body: "If you can read this, notifications are working.", Config: provider.Config, @@ -386,8 +413,9 @@ func (s *Service) validateProvider(providerType models.ProviderType, config map[ func (s *Service) validateEvents(events []string) error { allowed := map[string]struct{}{ - EventTaskSessionWaitingForInput: {}, - EventOfficeInboxItem: {}, + EventTaskSessionTurnFinished: {}, + EventTaskSessionClarificationAsked: {}, + EventOfficeInboxItem: {}, } for _, event := range events { if _, ok := allowed[event]; !ok { diff --git a/apps/backend/internal/notifications/service/service_test.go b/apps/backend/internal/notifications/service/service_test.go index ef667688c1..ea757850b1 100644 --- a/apps/backend/internal/notifications/service/service_test.go +++ b/apps/backend/internal/notifications/service/service_test.go @@ -1,13 +1,98 @@ package service import ( + "context" "testing" "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/internal/notifications/models" + "github.com/kandev/kandev/internal/notifications/providers" + taskmodels "github.com/kandev/kandev/internal/task/models" "go.uber.org/zap" ) +type notificationTestRepository struct { + providers []*models.Provider + subscriptions map[string][]*models.Subscription + deliveries []*models.Delivery +} + +func (r *notificationTestRepository) CreateProvider(_ context.Context, provider *models.Provider) error { + r.providers = append(r.providers, provider) + return nil +} +func (r *notificationTestRepository) UpdateProvider(context.Context, *models.Provider) error { + return nil +} +func (r *notificationTestRepository) GetProvider(_ context.Context, id string) (*models.Provider, error) { + for _, provider := range r.providers { + if provider.ID == id { + return provider, nil + } + } + return nil, nil +} +func (r *notificationTestRepository) ListProvidersByUser(context.Context, string) ([]*models.Provider, error) { + return r.providers, nil +} +func (r *notificationTestRepository) DeleteProvider(context.Context, string) error { return nil } +func (r *notificationTestRepository) ListSubscriptionsByProvider(_ context.Context, providerID string) ([]*models.Subscription, error) { + return r.subscriptions[providerID], nil +} +func (r *notificationTestRepository) ReplaceSubscriptions(context.Context, string, string, []string) error { + return nil +} +func (r *notificationTestRepository) InsertDelivery(_ context.Context, delivery *models.Delivery) (bool, error) { + for _, existing := range r.deliveries { + if existing.ProviderID == delivery.ProviderID && existing.EventType == delivery.EventType && existing.OccurrenceID == delivery.OccurrenceID { + return false, nil + } + } + r.deliveries = append(r.deliveries, delivery) + return true, nil +} +func (r *notificationTestRepository) DeleteDelivery(_ context.Context, providerID, eventType, occurrenceID string) error { + for index, delivery := range r.deliveries { + if delivery.ProviderID == providerID && delivery.EventType == eventType && delivery.OccurrenceID == occurrenceID { + r.deliveries = append(r.deliveries[:index], r.deliveries[index+1:]...) + return nil + } + } + return nil +} +func (r *notificationTestRepository) Close() error { return nil } + +type captureProvider struct{ messages []providers.Message } + +func (*captureProvider) Available() bool { return true } +func (*captureProvider) Validate(map[string]interface{}) error { return nil } +func (p *captureProvider) Send(_ context.Context, message providers.Message) error { + p.messages = append(p.messages, message) + return nil +} + +type failOnceProvider struct { + attempts int + messages []providers.Message +} + +func (*failOnceProvider) Available() bool { return true } +func (*failOnceProvider) Validate(map[string]interface{}) error { return nil } +func (p *failOnceProvider) Send(_ context.Context, message providers.Message) error { + p.attempts++ + if p.attempts == 1 { + return context.DeadlineExceeded + } + p.messages = append(p.messages, message) + return nil +} + +type notificationTestTaskGetter struct{ task *taskmodels.Task } + +func (g notificationTestTaskGetter) GetTask(context.Context, string) (*taskmodels.Task, error) { + return g.task, nil +} + func TestNewServiceSuppressesSystemProviderForDesktopOwnedLaunch(t *testing.T) { t.Setenv("KANDEV_DESKTOP_NATIVE_NOTIFICATIONS", "true") log, err := logger.NewFromZap(zap.NewNop()) @@ -38,3 +123,85 @@ func TestNewServiceRetainsSystemProviderForNonDesktopLaunch(t *testing.T) { t.Fatal("system notification provider must remain enabled outside the desktop-owned launch") } } + +func TestSemanticOccurrencesUseEventSpecificCopyAndOccurrenceIdempotency(t *testing.T) { + log, err := logger.NewFromZap(zap.NewNop()) + if err != nil { + t.Fatalf("create logger: %v", err) + } + repo := ¬ificationTestRepository{ + providers: []*models.Provider{{ID: "provider-1", Type: models.ProviderTypeLocal, Enabled: true}}, + subscriptions: map[string][]*models.Subscription{ + "provider-1": { + {ProviderID: "provider-1", EventType: EventTaskSessionTurnFinished, Enabled: true}, + {ProviderID: "provider-1", EventType: EventTaskSessionClarificationAsked, Enabled: true}, + }, + }, + } + service := NewService(repo, notificationTestTaskGetter{task: &taskmodels.Task{Title: "Fix delivery"}}, nil, log) + capture := &captureProvider{} + service.providers[models.ProviderTypeLocal] = capture + + service.HandleTaskTurnFinished(context.Background(), "task-1", "session-1", "turn-1") + service.HandleTaskTurnFinished(context.Background(), "task-1", "session-1", "turn-1") + service.HandleTaskTurnFinished(context.Background(), "task-1", "session-1", "turn-2") + service.HandleClarificationRequested(context.Background(), "task-1", "session-1", "pending-1") + + if len(capture.messages) != 3 { + t.Fatalf("sent %d notifications, want 3 semantic occurrences", len(capture.messages)) + } + if got := capture.messages[0]; got.EventType != EventTaskSessionTurnFinished || got.Title != "Agent turn finished" || got.Body != "The agent finished a turn on \"Fix delivery\"." { + t.Fatalf("turn notification = %#v", got) + } + if got := capture.messages[2]; got.EventType != EventTaskSessionClarificationAsked || got.Title != "Agent needs your answer" || got.Body != "The agent asked a question on \"Fix delivery\"." { + t.Fatalf("clarification notification = %#v", got) + } + if len(repo.deliveries) != 3 || repo.deliveries[0].OccurrenceID != "turn-1" || repo.deliveries[2].OccurrenceID != "pending-1" { + t.Fatalf("delivery occurrences = %#v", repo.deliveries) + } +} + +func TestFailedSemanticDeliveryReleasesOnlyItsOccurrenceClaim(t *testing.T) { + log, err := logger.NewFromZap(zap.NewNop()) + if err != nil { + t.Fatalf("create logger: %v", err) + } + repo := ¬ificationTestRepository{ + providers: []*models.Provider{{ID: "provider-1", Type: models.ProviderTypeLocal, Enabled: true}}, + subscriptions: map[string][]*models.Subscription{ + "provider-1": {{ProviderID: "provider-1", EventType: EventTaskSessionTurnFinished, Enabled: true}}, + }, + } + service := NewService(repo, nil, nil, log) + provider := &failOnceProvider{} + service.providers[models.ProviderTypeLocal] = provider + + service.HandleTaskTurnFinished(context.Background(), "task-1", "session-1", "turn-1") + service.HandleTaskTurnFinished(context.Background(), "task-1", "session-1", "turn-2") + service.HandleTaskTurnFinished(context.Background(), "task-1", "session-1", "turn-1") + + if provider.attempts != 3 { + t.Fatalf("send attempts = %d, want failed occurrence replayed independently", provider.attempts) + } + if len(provider.messages) != 2 { + t.Fatalf("successful messages = %#v", provider.messages) + } +} + +func TestProviderSendsClarificationAction(t *testing.T) { + log, err := logger.NewFromZap(zap.NewNop()) + if err != nil { + t.Fatalf("create logger: %v", err) + } + provider := &models.Provider{ID: "provider-1", Type: models.ProviderTypeLocal} + service := NewService(¬ificationTestRepository{providers: []*models.Provider{provider}}, nil, nil, log) + capture := &captureProvider{} + service.providers[models.ProviderTypeLocal] = capture + + if err := service.TestProvider(context.Background(), provider.ID); err != nil { + t.Fatalf("test provider: %v", err) + } + if len(capture.messages) != 1 || capture.messages[0].EventType != EventTaskSessionClarificationAsked { + t.Fatalf("test message = %#v, want clarification action", capture.messages) + } +} diff --git a/apps/backend/internal/notifications/store/interface.go b/apps/backend/internal/notifications/store/interface.go index 0b2f895373..550aaf665e 100644 --- a/apps/backend/internal/notifications/store/interface.go +++ b/apps/backend/internal/notifications/store/interface.go @@ -17,7 +17,7 @@ type Repository interface { ReplaceSubscriptions(ctx context.Context, providerID, userID string, events []string) error InsertDelivery(ctx context.Context, delivery *models.Delivery) (bool, error) - DeleteDelivery(ctx context.Context, providerID, eventType, taskSessionID string) error + DeleteDelivery(ctx context.Context, providerID, eventType, occurrenceID string) error Close() error } diff --git a/apps/backend/internal/notifications/store/sqlite.go b/apps/backend/internal/notifications/store/sqlite.go index ecaf2bec2b..d63f664181 100644 --- a/apps/backend/internal/notifications/store/sqlite.go +++ b/apps/backend/internal/notifications/store/sqlite.go @@ -2,8 +2,11 @@ package store import ( "context" + "database/sql" "encoding/json" + "errors" "fmt" + "strings" "time" "github.com/google/uuid" @@ -76,8 +79,9 @@ func (r *sqliteRepository) initSchema() error { provider_id TEXT NOT NULL, event_type TEXT NOT NULL, task_session_id TEXT NOT NULL, + occurrence_id TEXT NOT NULL, created_at TIMESTAMP NOT NULL, - UNIQUE(provider_id, event_type, task_session_id) + UNIQUE(provider_id, event_type, occurrence_id) ); CREATE INDEX IF NOT EXISTS idx_notification_providers_user_id ON notification_providers(user_id); @@ -86,10 +90,144 @@ func (r *sqliteRepository) initSchema() error { CREATE INDEX IF NOT EXISTS idx_notification_deliveries_session_id ON notification_deliveries(task_session_id); ` - _, err := r.db.Exec(schema) + if _, err := r.db.Exec(schema); err != nil { + return err + } + if err := r.migrateDeliveries(); err != nil { + return err + } + return r.migrateLegacySubscriptions() +} + +func (r *sqliteRepository) migrateDeliveries() error { + if dialect.IsPostgres(r.db.DriverName()) { + return r.migratePostgresDeliveries() + } + var schema string + err := r.db.QueryRow(`SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'notification_deliveries'`).Scan(&schema) + if errors.Is(err, sql.ErrNoRows) || strings.Contains(schema, "occurrence_id") { + return nil + } + if err != nil { + return fmt.Errorf("inspect notification deliveries schema: %w", err) + } + tx, err := r.db.Beginx() + if err != nil { + return err + } + defer func() { _ = tx.Rollback() }() + for _, statement := range []string{ + `CREATE TABLE notification_deliveries_new ( + id TEXT PRIMARY KEY, user_id TEXT NOT NULL, provider_id TEXT NOT NULL, + event_type TEXT NOT NULL, task_session_id TEXT NOT NULL, occurrence_id TEXT NOT NULL, + created_at TIMESTAMP NOT NULL, UNIQUE(provider_id, event_type, occurrence_id) + )`, + `INSERT INTO notification_deliveries_new (id, user_id, provider_id, event_type, task_session_id, occurrence_id, created_at) + SELECT id, user_id, provider_id, event_type, task_session_id, CAST(task_session_id AS TEXT), created_at + FROM notification_deliveries`, + `DROP TABLE notification_deliveries`, + `ALTER TABLE notification_deliveries_new RENAME TO notification_deliveries`, + `CREATE INDEX IF NOT EXISTS idx_notification_deliveries_session_id ON notification_deliveries(task_session_id)`, + } { + if _, err := tx.Exec(statement); err != nil { + return fmt.Errorf("migrate notification deliveries: %w", err) + } + } + return tx.Commit() +} + +func (r *sqliteRepository) migratePostgresDeliveries() error { + _, err := r.db.Exec(` + ALTER TABLE notification_deliveries + ADD COLUMN IF NOT EXISTS occurrence_id TEXT NOT NULL DEFAULT ''; + UPDATE notification_deliveries SET occurrence_id = task_session_id WHERE occurrence_id = ''; + DO $$ + DECLARE old_constraint text; + BEGIN + SELECT con.conname INTO old_constraint + FROM pg_constraint con + JOIN pg_class rel ON rel.oid = con.conrelid + JOIN pg_namespace nsp ON nsp.oid = rel.relnamespace + WHERE rel.relname = 'notification_deliveries' AND nsp.nspname = current_schema() + AND con.contype = 'u' + AND (SELECT array_agg(attr.attname::text ORDER BY cols.ordinality) + FROM unnest(con.conkey) WITH ORDINALITY AS cols(attnum, ordinality) + JOIN pg_attribute attr ON attr.attrelid = con.conrelid AND attr.attnum = cols.attnum) + = ARRAY['provider_id', 'event_type', 'task_session_id']; + IF old_constraint IS NOT NULL THEN + EXECUTE format('ALTER TABLE notification_deliveries DROP CONSTRAINT %I', old_constraint); + END IF; + IF NOT EXISTS ( + SELECT 1 FROM pg_constraint con JOIN pg_class rel ON rel.oid = con.conrelid + JOIN pg_namespace nsp ON nsp.oid = rel.relnamespace + WHERE rel.relname = 'notification_deliveries' AND nsp.nspname = current_schema() + AND con.contype = 'u' + AND (SELECT array_agg(attr.attname::text ORDER BY cols.ordinality) + FROM unnest(con.conkey) WITH ORDINALITY AS cols(attnum, ordinality) + JOIN pg_attribute attr ON attr.attrelid = con.conrelid AND attr.attnum = cols.attnum) + = ARRAY['provider_id', 'event_type', 'occurrence_id'] + ) THEN + ALTER TABLE notification_deliveries ADD CONSTRAINT notification_deliveries_provider_event_occurrence_key + UNIQUE (provider_id, event_type, occurrence_id); + END IF; + END $$; + `) return err } +func (r *sqliteRepository) migrateLegacySubscriptions() error { + tx, err := r.db.Beginx() + if err != nil { + return err + } + defer func() { _ = tx.Rollback() }() + rows, err := tx.Queryx(r.db.Rebind(` + SELECT id, provider_id, enabled FROM notification_subscriptions + WHERE event_type = ? + `), "session.waiting_for_input") + if err != nil { + return err + } + type legacySubscription struct { + id, providerID string + enabled int + } + legacy := make([]legacySubscription, 0) + for rows.Next() { + var subscription legacySubscription + if err := rows.Scan(&subscription.id, &subscription.providerID, &subscription.enabled); err != nil { + return err + } + legacy = append(legacy, subscription) + } + if err := rows.Close(); err != nil { + return err + } + if err := rows.Err(); err != nil { + return err + } + for _, subscription := range legacy { + var semanticEnabled int + err := tx.Get(&semanticEnabled, r.db.Rebind(`SELECT enabled FROM notification_subscriptions WHERE provider_id = ? AND event_type = ?`), subscription.providerID, "session.clarification_requested") + if errors.Is(err, sql.ErrNoRows) { + if _, err := tx.Exec(r.db.Rebind(`UPDATE notification_subscriptions SET event_type = ? WHERE id = ?`), "session.clarification_requested", subscription.id); err != nil { + return err + } + continue + } + if err != nil { + return err + } + if _, err := tx.Exec(r.db.Rebind(`UPDATE notification_subscriptions SET enabled = ? WHERE provider_id = ? AND event_type = ?`), dialect.BoolToInt(subscription.enabled == 1 || semanticEnabled == 1), subscription.providerID, "session.clarification_requested"); err != nil { + return err + } + if _, err := tx.Exec(r.db.Rebind(`DELETE FROM notification_subscriptions WHERE id = ?`), subscription.id); err != nil { + return err + } + } + return tx.Commit() +} + func (r *sqliteRepository) CreateProvider(ctx context.Context, provider *models.Provider) error { provider.ID = uuid.New().String() now := time.Now().UTC() @@ -224,13 +362,16 @@ func (r *sqliteRepository) ReplaceSubscriptions(ctx context.Context, providerID, } func (r *sqliteRepository) InsertDelivery(ctx context.Context, delivery *models.Delivery) (bool, error) { + if delivery.OccurrenceID == "" { + return false, errors.New("notification delivery occurrence ID is required") + } delivery.ID = uuid.New().String() delivery.CreatedAt = time.Now().UTC() result, err := r.db.ExecContext(ctx, r.db.Rebind(` - INSERT INTO notification_deliveries (id, user_id, provider_id, event_type, task_session_id, created_at) - VALUES (?, ?, ?, ?, ?, ?) + INSERT INTO notification_deliveries (id, user_id, provider_id, event_type, task_session_id, occurrence_id, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?) ON CONFLICT DO NOTHING - `), delivery.ID, delivery.UserID, delivery.ProviderID, delivery.EventType, delivery.TaskSessionID, delivery.CreatedAt) + `), delivery.ID, delivery.UserID, delivery.ProviderID, delivery.EventType, delivery.TaskSessionID, delivery.OccurrenceID, delivery.CreatedAt) if err != nil { return false, err } @@ -241,11 +382,11 @@ func (r *sqliteRepository) InsertDelivery(ctx context.Context, delivery *models. return rows > 0, nil } -func (r *sqliteRepository) DeleteDelivery(ctx context.Context, providerID, eventType, taskSessionID string) error { +func (r *sqliteRepository) DeleteDelivery(ctx context.Context, providerID, eventType, occurrenceID string) error { _, err := r.db.ExecContext(ctx, r.db.Rebind(` DELETE FROM notification_deliveries - WHERE provider_id = ? AND event_type = ? AND task_session_id = ? - `), providerID, eventType, taskSessionID) + WHERE provider_id = ? AND event_type = ? AND occurrence_id = ? + `), providerID, eventType, occurrenceID) return err } diff --git a/apps/backend/internal/notifications/store/sqlite_test.go b/apps/backend/internal/notifications/store/sqlite_test.go new file mode 100644 index 0000000000..c407189127 --- /dev/null +++ b/apps/backend/internal/notifications/store/sqlite_test.go @@ -0,0 +1,333 @@ +package store + +import ( + "context" + "fmt" + "path/filepath" + "testing" + + "github.com/jmoiron/sqlx" + "github.com/kandev/kandev/internal/db" + "github.com/kandev/kandev/internal/notifications/models" + "github.com/kandev/kandev/internal/testutil" +) + +func openNotificationTestDB(t *testing.T) *sqlx.DB { + t.Helper() + conn, err := db.OpenSQLite(filepath.Join(t.TempDir(), "notifications.db")) + if err != nil { + t.Fatalf("open SQLite database: %v", err) + } + database := sqlx.NewDb(conn, "sqlite3") + t.Cleanup(func() { _ = database.Close() }) + return database +} + +func TestSQLiteRepositoryDeliverySchemaUsesOccurrenceID(t *testing.T) { + database := openNotificationTestDB(t) + if _, err := newSQLiteRepositoryWithDB(database, database); err != nil { + t.Fatalf("create repository: %v", err) + } + + rows, err := database.Queryx(`PRAGMA table_info(notification_deliveries)`) + if err != nil { + t.Fatalf("inspect delivery schema: %v", err) + } + defer func() { _ = rows.Close() }() + + for rows.Next() { + var cid int + var name, columnType string + var notNull, primaryKey int + var defaultValue any + if err := rows.Scan(&cid, &name, &columnType, ¬Null, &defaultValue, &primaryKey); err != nil { + t.Fatalf("scan delivery schema: %v", err) + } + if name == "occurrence_id" { + return + } + } + t.Fatal("notification_deliveries must retain a semantic occurrence_id") +} + +func TestSQLiteRepositoryMigratesLegacyWaitingSubscriptionToClarificationOnly(t *testing.T) { + database := openNotificationTestDB(t) + ctx := context.Background() + legacySchema := ` + CREATE TABLE notification_providers ( + id TEXT PRIMARY KEY, user_id TEXT NOT NULL, name TEXT NOT NULL, type TEXT NOT NULL, + config TEXT DEFAULT '{}', enabled INTEGER NOT NULL DEFAULT 1, + created_at TIMESTAMP NOT NULL, updated_at TIMESTAMP NOT NULL + ); + CREATE TABLE notification_subscriptions ( + id TEXT PRIMARY KEY, user_id TEXT NOT NULL, provider_id TEXT NOT NULL, + event_type TEXT NOT NULL, enabled INTEGER NOT NULL DEFAULT 1, + created_at TIMESTAMP NOT NULL, updated_at TIMESTAMP NOT NULL, + UNIQUE(provider_id, event_type) + ); + ` + if _, err := database.Exec(legacySchema); err != nil { + t.Fatalf("create legacy schema: %v", err) + } + if _, err := database.ExecContext(ctx, ` + INSERT INTO notification_providers (id, user_id, name, type, created_at, updated_at) + VALUES ('provider-1', 'user-1', 'Legacy', 'local', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP); + INSERT INTO notification_subscriptions (id, user_id, provider_id, event_type, enabled, created_at, updated_at) + VALUES ('subscription-1', 'user-1', 'provider-1', 'session.waiting_for_input', 1, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP); + `); err != nil { + t.Fatalf("seed legacy subscription: %v", err) + } + + repo, err := newSQLiteRepositoryWithDB(database, database) + if err != nil { + t.Fatalf("migrate legacy schema: %v", err) + } + subscriptions, err := repo.ListSubscriptionsByProvider(ctx, "provider-1") + if err != nil { + t.Fatalf("list migrated subscriptions: %v", err) + } + if len(subscriptions) != 1 || subscriptions[0].EventType != "session.clarification_requested" { + t.Fatalf("legacy subscription = %#v, want clarification subscription only", subscriptions) + } +} + +func TestSQLiteRepositoryMergesLegacySubscriptionIdempotently(t *testing.T) { + database := openNotificationTestDB(t) + ctx := context.Background() + if _, err := database.Exec(` + CREATE TABLE notification_providers ( + id TEXT PRIMARY KEY, user_id TEXT NOT NULL, name TEXT NOT NULL, type TEXT NOT NULL, + config TEXT DEFAULT '{}', enabled INTEGER NOT NULL DEFAULT 1, + created_at TIMESTAMP NOT NULL, updated_at TIMESTAMP NOT NULL + ); + CREATE TABLE notification_subscriptions ( + id TEXT PRIMARY KEY, user_id TEXT NOT NULL, provider_id TEXT NOT NULL, + event_type TEXT NOT NULL, enabled INTEGER NOT NULL DEFAULT 1, + created_at TIMESTAMP NOT NULL, updated_at TIMESTAMP NOT NULL, + UNIQUE(provider_id, event_type) + ); + INSERT INTO notification_providers (id, user_id, name, type, created_at, updated_at) + VALUES ('provider-1', 'user-1', 'Legacy', 'local', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP); + INSERT INTO notification_subscriptions (id, user_id, provider_id, event_type, enabled, created_at, updated_at) VALUES + ('legacy', 'user-1', 'provider-1', 'session.waiting_for_input', 1, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP), + ('semantic', 'user-1', 'provider-1', 'session.clarification_requested', 0, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP); + `); err != nil { + t.Fatalf("seed legacy and semantic subscriptions: %v", err) + } + if _, err := newSQLiteRepositoryWithDB(database, database); err != nil { + t.Fatalf("first migration: %v", err) + } + repo, err := newSQLiteRepositoryWithDB(database, database) + if err != nil { + t.Fatalf("replay migration: %v", err) + } + subscriptions, err := repo.ListSubscriptionsByProvider(ctx, "provider-1") + if err != nil { + t.Fatalf("list migrated subscriptions: %v", err) + } + if len(subscriptions) != 1 || subscriptions[0].EventType != "session.clarification_requested" || !subscriptions[0].Enabled { + t.Fatalf("merged subscriptions = %#v, want one enabled clarification subscription", subscriptions) + } +} + +func TestSQLiteRepositoryDeliveryIdempotencyIsScopedToOccurrence(t *testing.T) { + database := openNotificationTestDB(t) + repo, err := newSQLiteRepositoryWithDB(database, database) + if err != nil { + t.Fatalf("create repository: %v", err) + } + ctx := context.Background() + first, err := repo.InsertDelivery(ctx, &models.Delivery{UserID: "user-1", ProviderID: "provider-1", EventType: "session.turn_finished", TaskSessionID: "session-1", OccurrenceID: "turn-1"}) + if err != nil || !first { + t.Fatalf("insert first occurrence: inserted=%v err=%v", first, err) + } + replayed, err := repo.InsertDelivery(ctx, &models.Delivery{UserID: "user-1", ProviderID: "provider-1", EventType: "session.turn_finished", TaskSessionID: "session-1", OccurrenceID: "turn-1"}) + if err != nil || replayed { + t.Fatalf("replay occurrence: inserted=%v err=%v", replayed, err) + } + later, err := repo.InsertDelivery(ctx, &models.Delivery{UserID: "user-1", ProviderID: "provider-1", EventType: "session.turn_finished", TaskSessionID: "session-1", OccurrenceID: "turn-2"}) + if err != nil || !later { + t.Fatalf("insert later occurrence: inserted=%v err=%v", later, err) + } + if _, err := repo.InsertDelivery(ctx, &models.Delivery{UserID: "user-1", ProviderID: "provider-1", EventType: "session.turn_finished", TaskSessionID: "session-1"}); err == nil { + t.Fatal("empty occurrence ID must be rejected") + } +} + +func TestSQLiteRepositoryMigratesLegacyDeliveriesPreservingHistoryAndReplayability(t *testing.T) { + database := openNotificationTestDB(t) + seedLegacyDeliverySchema(t, database) + + repo, err := newSQLiteRepositoryWithDB(database, database) + if err != nil { + t.Fatalf("first migration: %v", err) + } + if _, err := newSQLiteRepositoryWithDB(database, database); err != nil { + t.Fatalf("replay migration: %v", err) + } + + assertMigratedLegacyDeliveries(t, database) + assertDeliverySchema(t, database) + assertOccurrenceUniqueness(t, repo) +} + +func TestSQLiteRepositoryDeliveryMigrationRollsBackOnMalformedLegacyTable(t *testing.T) { + database := openNotificationTestDB(t) + if _, err := database.Exec(` + CREATE TABLE notification_deliveries ( + id TEXT PRIMARY KEY, user_id TEXT NOT NULL, provider_id TEXT NOT NULL, + event_type TEXT NOT NULL, task_session_id TEXT NOT NULL, + UNIQUE(provider_id, event_type, task_session_id) + ); + INSERT INTO notification_deliveries (id, user_id, provider_id, event_type, task_session_id) + VALUES ('delivery-1', 'user-1', 'provider-1', 'session.waiting_for_input', 'session-1'); + `); err != nil { + t.Fatalf("seed malformed legacy deliveries: %v", err) + } + + if _, err := newSQLiteRepositoryWithDB(database, database); err == nil { + t.Fatal("migration must fail after beginning when legacy history is malformed") + } + var count int + if err := database.Get(&count, `SELECT COUNT(*) FROM notification_deliveries WHERE id = 'delivery-1'`); err != nil || count != 1 { + t.Fatalf("legacy delivery after rollback = count %d, err %v", count, err) + } + var occurrenceColumns int + if err := database.Get(&occurrenceColumns, `SELECT COUNT(*) FROM pragma_table_info('notification_deliveries') WHERE name = 'occurrence_id'`); err != nil || occurrenceColumns != 0 { + t.Fatalf("legacy schema after rollback = occurrence columns %d, err %v", occurrenceColumns, err) + } +} + +func TestPostgresRepositoryMigratesLegacyDeliveriesReplayably(t *testing.T) { + database := testutil.OpenIsolatedPostgres(t, testutil.PostgresDSNFromEnv(t)) + seedLegacyDeliverySchema(t, database) + + repo, err := newSQLiteRepositoryWithDB(database, database) + if err != nil { + t.Fatalf("first migration: %v", err) + } + if _, err := newSQLiteRepositoryWithDB(database, database); err != nil { + t.Fatalf("replay migration: %v", err) + } + + assertMigratedLegacyDeliveries(t, database) + assertDeliverySchema(t, database) + assertOccurrenceUniqueness(t, repo) +} + +func TestPostgresRepositoryFreshSchemaReinitializes(t *testing.T) { + database := testutil.OpenIsolatedPostgres(t, testutil.PostgresDSNFromEnv(t)) + if _, err := newSQLiteRepositoryWithDB(database, database); err != nil { + t.Fatalf("first schema initialization: %v", err) + } + if _, err := newSQLiteRepositoryWithDB(database, database); err != nil { + t.Fatalf("replayed schema initialization: %v", err) + } +} + +func seedLegacyDeliverySchema(t *testing.T, database *sqlx.DB) { + t.Helper() + if _, err := database.Exec(` + CREATE TABLE notification_deliveries ( + id TEXT PRIMARY KEY, user_id TEXT NOT NULL, provider_id TEXT NOT NULL, + event_type TEXT NOT NULL, task_session_id TEXT NOT NULL, created_at TIMESTAMP NOT NULL, + UNIQUE(provider_id, event_type, task_session_id) + ); + INSERT INTO notification_deliveries (id, user_id, provider_id, event_type, task_session_id, created_at) VALUES + ('delivery-1', 'user-1', 'provider-1', 'session.waiting_for_input', 'session-1', CURRENT_TIMESTAMP), + ('delivery-2', 'user-1', 'provider-1', 'session.waiting_for_input', 'session-2', CURRENT_TIMESTAMP); + `); err != nil { + t.Fatalf("seed legacy deliveries: %v", err) + } +} + +func assertMigratedLegacyDeliveries(t *testing.T, database *sqlx.DB) { + t.Helper() + var rows []struct { + ID string `db:"id"` + OccurrenceID string `db:"occurrence_id"` + } + if err := database.Select(&rows, `SELECT id, occurrence_id FROM notification_deliveries ORDER BY id`); err != nil { + t.Fatalf("list migrated deliveries: %v", err) + } + if got := fmt.Sprint(rows); got != "[{delivery-1 session-1} {delivery-2 session-2}]" { + t.Fatalf("migrated delivery history = %s", got) + } + var occurrenceColumns int + if err := database.Get(&occurrenceColumns, database.Rebind(`SELECT COUNT(*) FROM information_schema.columns WHERE table_schema = current_schema() AND table_name = ? AND column_name = ?`), "notification_deliveries", "occurrence_id"); err != nil && database.DriverName() != "sqlite3" { + t.Fatalf("inspect postgres occurrence column: %v", err) + } + if database.DriverName() == "sqlite3" { + if err := database.Get(&occurrenceColumns, `SELECT COUNT(*) FROM pragma_table_info('notification_deliveries') WHERE name = 'occurrence_id'`); err != nil { + t.Fatalf("inspect sqlite occurrence column: %v", err) + } + } + if occurrenceColumns != 1 { + t.Fatalf("occurrence_id column count = %d, want 1", occurrenceColumns) + } +} + +func assertOccurrenceUniqueness(t *testing.T, repo *sqliteRepository) { + t.Helper() + ctx := context.Background() + first := &models.Delivery{UserID: "user-1", ProviderID: "provider-1", EventType: "session.turn_finished", TaskSessionID: "session-3", OccurrenceID: "turn-1"} + if inserted, err := repo.InsertDelivery(ctx, first); err != nil || !inserted { + t.Fatalf("insert first occurrence: inserted=%v err=%v", inserted, err) + } + later := &models.Delivery{UserID: "user-1", ProviderID: "provider-1", EventType: "session.turn_finished", TaskSessionID: "session-3", OccurrenceID: "turn-2"} + if inserted, err := repo.InsertDelivery(ctx, later); err != nil || !inserted { + t.Fatalf("insert later same-session occurrence: inserted=%v err=%v", inserted, err) + } + if inserted, err := repo.InsertDelivery(ctx, first); err != nil || inserted { + t.Fatalf("replay same occurrence: inserted=%v err=%v", inserted, err) + } +} + +func assertDeliverySchema(t *testing.T, database *sqlx.DB) { + t.Helper() + if database.DriverName() == "sqlite3" { + assertSQLiteDeliverySchema(t, database) + return + } + assertPostgresDeliverySchema(t, database) +} + +func assertSQLiteDeliverySchema(t *testing.T, database *sqlx.DB) { + t.Helper() + var indexNames []string + if err := database.Select(&indexNames, `SELECT name FROM pragma_index_list('notification_deliveries') WHERE "unique" = 1`); err != nil { + t.Fatalf("list sqlite delivery indexes: %v", err) + } + for _, indexName := range indexNames { + var columns []string + if err := database.Select(&columns, `SELECT name FROM pragma_index_info(?) ORDER BY seqno`, indexName); err != nil { + t.Fatalf("inspect sqlite index %s: %v", indexName, err) + } + if fmt.Sprint(columns) == "[provider_id event_type occurrence_id]" { + return + } + } + t.Fatalf("sqlite unique indexes = %#v, want provider/event/occurrence uniqueness", indexNames) +} + +func assertPostgresDeliverySchema(t *testing.T, database *sqlx.DB) { + t.Helper() + var definitions []string + if err := database.Select(&definitions, ` + SELECT pg_get_constraintdef(con.oid) + FROM pg_constraint con + JOIN pg_class rel ON rel.oid = con.conrelid + JOIN pg_namespace nsp ON nsp.oid = rel.relnamespace + WHERE rel.relname = 'notification_deliveries' + AND nsp.nspname = current_schema() + AND con.contype = 'u' + `); err != nil { + t.Fatalf("list postgres delivery constraints: %v", err) + } + for _, definition := range definitions { + if definition == "UNIQUE (provider_id, event_type, occurrence_id)" { + return + } + } + t.Fatalf("postgres unique constraints = %#v, want provider/event/occurrence uniqueness", definitions) +} diff --git a/apps/backend/pkg/websocket/actions.go b/apps/backend/pkg/websocket/actions.go index e366f141a5..5b782d2678 100644 --- a/apps/backend/pkg/websocket/actions.go +++ b/apps/backend/pkg/websocket/actions.go @@ -190,7 +190,6 @@ const ( ActionSessionMessageUpdated = "session.message.updated" ActionSessionMessageDeleted = "session.message.deleted" ActionSessionStateChanged = "session.state_changed" - ActionSessionWaitingForInput = "session.waiting_for_input" ActionSessionAgentctlStarting = "session.agentctl_starting" ActionSessionAgentctlReady = "session.agentctl_ready" ActionSessionAgentctlError = "session.agentctl_error" @@ -393,6 +392,12 @@ const ( ActionMCPGetTaskConversation = "mcp.get_task_conversation" ) +const ( + ActionSessionTurnFinished = "session.turn_finished" + ActionSessionClarificationRequested = "session.clarification_requested" + ActionOfficeInboxItem = "office.inbox_item" +) + // GitHub integration actions const ( ActionGitHubStatus = "github.status" diff --git a/apps/desktop/src-tauri/src/native_notifications.rs b/apps/desktop/src-tauri/src/native_notifications.rs index 7ae3fe7073..413a4a11a9 100644 --- a/apps/desktop/src-tauri/src/native_notifications.rs +++ b/apps/desktop/src-tauri/src/native_notifications.rs @@ -84,7 +84,10 @@ impl NativeNotificationState { } fn validate_request(request: &NativeNotificationRequest) -> Result<(), String> { - if !request.event_id.starts_with("session.waiting_for_input:") + if !request.event_id.starts_with("session.turn_finished:") + && !request + .event_id + .starts_with("session.clarification_requested:") && !request.event_id.starts_with("session.failed:") { return Err("unsupported native notification event".to_string()); @@ -154,8 +157,8 @@ mod tests { fn request(event_id: &str, task_id: &str) -> NativeNotificationRequest { NativeNotificationRequest { event_id: event_id.to_string(), - title: "Task needs your input".to_string(), - body: "Waiting".to_string(), + title: "Agent needs your answer".to_string(), + body: "The agent asked a question.".to_string(), task_id: task_id.to_string(), session_id: Some("session-1".to_string()), } @@ -164,7 +167,7 @@ mod tests { #[test] fn event_identity_is_delivered_at_most_once() { let state = NativeNotificationState::default(); - let request = request("session.waiting_for_input:session-1", "task-1"); + let request = request("session.clarification_requested:session-1", "task-1"); assert!(state.claim(&request.event_id)); assert!(!state.claim(&request.event_id)); @@ -182,13 +185,13 @@ mod tests { } #[test] - fn bridge_rejects_events_outside_the_two_notification_types() { - let request = request("office.inbox_item:item-1", "task-1"); - - assert_eq!( - validate_request(&request), - Err("unsupported native notification event".to_string()) - ); + fn bridge_accepts_semantic_session_events() { + for event_id in [ + "session.turn_finished:turn-1", + "session.clarification_requested:request-1", + ] { + assert_eq!(validate_request(&request(event_id, "task-1")), Ok(())); + } } #[test] diff --git a/apps/packages/types/src/backend.ts b/apps/packages/types/src/backend.ts index 8b2a10cadd..1da1e846d4 100644 --- a/apps/packages/types/src/backend.ts +++ b/apps/packages/types/src/backend.ts @@ -17,7 +17,9 @@ export type BackendMessageType = | "column.created" | "column.updated" | "column.deleted" - | "task_session.waiting_for_input"; + | "session.turn_finished" + | "session.clarification_requested" + | "office.inbox_item"; export type BackendMessage = { id?: string; @@ -78,9 +80,17 @@ export type SystemErrorPayload = { code?: string; }; -export type TaskSessionWaitingForInputPayload = { +export type TaskSessionNotificationPayload = { task_id: string; - task_session_id: string; + session_id: string; + occurrence_id: string; + title: string; + body: string; +}; + +export type OfficeInboxItemNotificationPayload = { + task_id?: string; + session_id?: string; title: string; body: string; }; @@ -133,8 +143,10 @@ export type BackendMessageMap = { "column.created": BackendMessage<"column.created", ColumnPayload>; "column.updated": BackendMessage<"column.updated", ColumnPayload>; "column.deleted": BackendMessage<"column.deleted", ColumnPayload>; - "task_session.waiting_for_input": BackendMessage< - "task_session.waiting_for_input", - TaskSessionWaitingForInputPayload + "session.turn_finished": BackendMessage<"session.turn_finished", TaskSessionNotificationPayload>; + "session.clarification_requested": BackendMessage< + "session.clarification_requested", + TaskSessionNotificationPayload >; + "office.inbox_item": BackendMessage<"office.inbox_item", OfficeInboxItemNotificationPayload>; }; diff --git a/apps/web/components/settings/notification-events-table.tsx b/apps/web/components/settings/notification-events-table.tsx index 82eb07dc5b..a78d3d4ddb 100644 --- a/apps/web/components/settings/notification-events-table.tsx +++ b/apps/web/components/settings/notification-events-table.tsx @@ -15,19 +15,137 @@ type Props = { onTestProvider: (providerId: string) => Promise; }; -export function NotificationEventsTable({ +function eventMeta(eventType: string) { + return ( + EVENT_LABELS[eventType] ?? { + title: eventType, + description: "Notify when this event occurs.", + } + ); +} + +function EventCheckbox({ + provider, + baselineProviders, + eventType, + onToggleEvent, + mobile = false, +}: { + provider: NotificationProvider; + baselineProviders: NotificationProvider[]; + eventType: string; + onToggleEvent: Props["onToggleEvent"]; + mobile?: boolean; +}) { + const meta = eventMeta(eventType); + const checked = (provider.events ?? []).includes(eventType); + const baselineChecked = ( + baselineProviders.find((candidate) => candidate.id === provider.id)?.events ?? [] + ).includes(eventType); + const checkbox = ( + onToggleEvent(provider, eventType)} + /> + ); + if (!mobile) return checkbox; + return ( + + ); +} + +function TestProviderButton({ + provider, + onTestProvider, + mobile = false, +}: Pick & { + provider: NotificationProvider; + mobile?: boolean; +}) { + if (provider.type === "local") return null; + return ( + + + + + + Send test notification + + + ); +} + +function MobileEventList({ tableProviders, baselineProviders, tableEvents, onToggleEvent, onTestProvider, }: Props) { - if (tableProviders.length === 0) { - return

No providers configured yet.

; - } - return ( -
+
+ {tableEvents.map((eventType) => { + const meta = eventMeta(eventType); + return ( +
+
+

{meta.title}

+

{meta.description}

+
+
+ {tableProviders.map((provider) => ( +
+ + {provider.name} + + + +
+ ))} +
+
+ ); + })} +
+ ); +} + +function DesktopEventTable({ + tableProviders, + baselineProviders, + tableEvents, + onToggleEvent, + onTestProvider, +}: Props) { + return ( +
@@ -36,24 +154,7 @@ export function NotificationEventsTable({ ))} @@ -61,34 +162,25 @@ export function NotificationEventsTable({ {tableEvents.map((eventType) => { - const meta = EVENT_LABELS[eventType] ?? { - title: eventType, - description: "Notify when this event occurs.", - }; + const meta = eventMeta(eventType); return ( - {tableProviders.map((provider) => { - const checked = (provider.events ?? []).includes(eventType); - const baselineChecked = ( - baselineProviders.find((candidate) => candidate.id === provider.id)?.events ?? - [] - ).includes(eventType); - return ( - - ); - })} + {tableProviders.map((provider) => ( + + ))} ); })} @@ -97,3 +189,16 @@ export function NotificationEventsTable({ ); } + +export function NotificationEventsTable(props: Props) { + if (props.tableProviders.length === 0) { + return

No providers configured yet.

; + } + + return ( + <> + + + + ); +} diff --git a/apps/web/components/settings/notification-sound-section.tsx b/apps/web/components/settings/notification-sound-section.tsx index 93b1a59e62..eebcd25b45 100644 --- a/apps/web/components/settings/notification-sound-section.tsx +++ b/apps/web/components/settings/notification-sound-section.tsx @@ -50,7 +50,8 @@ export function NotificationSoundSection({
Notification Sound

- Play a sound on this device when an agent needs your input. + Play a sound on this device when a selected agent turn, question, or Office event + occurs.

({ setNotificationProviders: vi.fn(), })); const PAGER_URL = "json://pager"; +const CLARIFICATION_EVENT = "session.clarification_requested"; +const SEMANTIC_NOTIFICATION_EVENTS = [CLARIFICATION_EVENT, "session.turn_finished"]; vi.mock("@/lib/api", () => ({ createNotificationProvider: mocks.createNotificationProvider, @@ -35,11 +37,16 @@ const savedProvider: NotificationProvider = { updated_at: "", }; +let notificationProviders = [savedProvider]; +let notificationEvents = ["task.completed"]; +let notificationProvidersLoaded = true; + vi.mock("@/hooks/domains/settings/use-notification-providers", () => ({ useNotificationProviders: () => ({ - providers: [savedProvider], - events: ["task.completed"], + providers: notificationProviders, + events: notificationEvents, appriseAvailable: true, + loaded: notificationProvidersLoaded, }), })); @@ -64,10 +71,86 @@ const createdProvider: NotificationProvider = { beforeEach(() => { vi.clearAllMocks(); + notificationProviders = [savedProvider]; + notificationEvents = ["task.completed"]; + notificationProvidersLoaded = true; mocks.createNotificationProvider.mockResolvedValue(createdProvider); }); -describe("notification provider drafts", () => { +describe("notification provider draft hydration", () => { + it("hydrates a cold provider load into the draft", () => { + notificationProviders = []; + notificationEvents = []; + notificationProvidersLoaded = false; + const { result, rerender } = renderHook(useHarness); + + notificationProviders = [savedProvider]; + notificationEvents = SEMANTIC_NOTIFICATION_EVENTS; + notificationProvidersLoaded = true; + rerender(); + + expect(result.current.state.providers).toEqual([savedProvider]); + expect(result.current.state.baselineProviders).toEqual([savedProvider]); + expect(result.current.state.notificationEvents).toEqual(SEMANTIC_NOTIFICATION_EVENTS); + }); + + it("hydrates the first provider load while retaining a typed create draft", async () => { + notificationProviders = []; + notificationEvents = []; + notificationProvidersLoaded = false; + const { result, rerender } = renderHook(useHarness); + + act(() => { + result.current.actions.openAppriseForm("create"); + result.current.state.setAppriseName("Pager"); + result.current.state.setAppriseUrls(PAGER_URL); + }); + + notificationProviders = [savedProvider]; + notificationEvents = SEMANTIC_NOTIFICATION_EVENTS; + notificationProvidersLoaded = true; + rerender(); + + expect(result.current.state.providers).toEqual([savedProvider]); + expect(result.current.state.baselineProviders).toEqual([savedProvider]); + expect(result.current.state.appriseName).toBe("Pager"); + expect(result.current.state.appriseUrls).toBe(PAGER_URL); + + await act(() => result.current.saveRequest.run()); + + expect(result.current.state.providers).toEqual([savedProvider, createdProvider]); + expect(mocks.setNotificationProviders).toHaveBeenCalledWith( + expect.objectContaining({ items: [savedProvider, createdProvider] }), + ); + }); + + it("does not replace an edited draft when providers refresh", () => { + const { result, rerender } = renderHook(useHarness); + act(() => result.current.actions.handleAppriseNameEdit(savedProvider.id, "Draft name")); + + const refreshedProvider = { ...savedProvider, name: "Server refresh" }; + notificationProviders = [refreshedProvider]; + rerender(); + + expect(result.current.state.providers[0]?.name).toBe("Draft name"); + expect(result.current.state.baselineProviders[0]?.name).toBe(savedProvider.name); + }); + + it("hydrates a deferred provider refresh after discarding the draft", () => { + const { result, rerender } = renderHook(useHarness); + act(() => result.current.actions.handleAppriseNameEdit(savedProvider.id, "Draft name")); + const refreshedProvider = { ...savedProvider, name: "Server refresh" }; + notificationProviders = [refreshedProvider]; + rerender(); + + act(() => result.current.actions.discard()); + + expect(result.current.state.providers).toEqual([refreshedProvider]); + expect(result.current.state.baselineProviders).toEqual([refreshedProvider]); + }); +}); + +describe("notification provider draft saving", () => { it("stages a new Apprise provider until the shared save runs", async () => { const { result } = renderHook(useHarness); @@ -85,6 +168,7 @@ describe("notification provider drafts", () => { expect.objectContaining({ name: "Pager", config: { urls: [PAGER_URL] }, + events: [CLARIFICATION_EVENT], }), ); expect(result.current.state.providers).toContainEqual(createdProvider); diff --git a/apps/web/components/settings/notifications-settings-actions.ts b/apps/web/components/settings/notifications-settings-actions.ts index 2f3fdb827b..acd6bf6e69 100644 --- a/apps/web/components/settings/notifications-settings-actions.ts +++ b/apps/web/components/settings/notifications-settings-actions.ts @@ -1,6 +1,6 @@ "use client"; -import { useState } from "react"; +import { useEffect, useRef, useState } from "react"; import { createNotificationProvider, deleteNotificationProvider, @@ -22,6 +22,22 @@ type ProviderUpdatePayload = { type AppriseFormMode = "create" | "edit"; +type NotificationDraft = { + providers: NotificationProvider[]; + baselineProviders: NotificationProvider[]; + appriseEdits: Record; + appriseNameEdits: Record; + pendingDeletes: Set; + showAppriseForm: boolean; + appriseFormMode: AppriseFormMode; +}; + +type NotificationHydrationSource = { + providers: NotificationProvider[] | undefined; + events: string[] | undefined; + appriseAvailable: boolean | undefined; +}; + export function formatAppriseUrls(value: unknown): string { if (Array.isArray(value)) { return value.filter((item): item is string => typeof item === "string").join("\n"); @@ -89,19 +105,56 @@ function buildAppriseEdits(providers: NotificationProvider[]) { return { urls, names }; } +function hasDraftChanges({ + providers, + baselineProviders, + appriseEdits, + appriseNameEdits, + pendingDeletes, + showAppriseForm, + appriseFormMode, +}: NotificationDraft) { + return ( + (showAppriseForm && appriseFormMode === "create") || + pendingDeletes.size > 0 || + providers.some((provider) => { + const baseline = baselineProviders.find((item) => item.id === provider.id); + if (!baseline) return false; + if (buildProviderUpdate(provider, baseline)) return true; + if (provider.type !== "apprise") return false; + const currentValue = appriseEdits[provider.id] ?? formatAppriseUrls(provider.config?.urls); + const baselineValue = formatAppriseUrls(baseline.config?.urls); + const nameValue = appriseNameEdits[provider.id] ?? provider.name; + return currentValue !== baselineValue || nameValue !== baseline.name; + }) + ); +} + +function isSameHydrationSource( + current: NotificationHydrationSource | undefined, + next: NotificationHydrationSource, +) { + return ( + current?.providers === next.providers && + current?.events === next.events && + current?.appriseAvailable === next.appriseAvailable + ); +} + export function useNotificationsState() { const { providers: storeProviders, events: storeEvents, appriseAvailable: storeAppriseAvailable, + loaded, } = useNotificationProviders(); const setNotificationProviders = useAppStore((state) => state.setNotificationProviders); const [providers, setProviders] = useState(() => storeProviders ?? []); const [baselineProviders, setBaselineProviders] = useState( () => storeProviders ?? [], ); - const [notificationEvents] = useState(() => storeEvents ?? []); - const [appriseAvailable] = useState(() => storeAppriseAvailable ?? true); + const [notificationEvents, setNotificationEvents] = useState(() => storeEvents ?? []); + const [appriseAvailable, setAppriseAvailable] = useState(() => storeAppriseAvailable ?? true); const [appriseName, setAppriseName] = useState(""); const [appriseUrls, setAppriseUrls] = useState(""); const [appriseEdits, setAppriseEdits] = useState>( @@ -114,13 +167,61 @@ export function useNotificationsState() { const [appriseFormMode, setAppriseFormMode] = useState("create"); const [activeAppriseId, setActiveAppriseId] = useState(null); const [pendingDeletes, setPendingDeletes] = useState>(new Set()); + const hydratedSource = useRef(undefined); + + useEffect(() => { + if (!loaded) return; + const source = { + providers: storeProviders, + events: storeEvents, + appriseAvailable: storeAppriseAvailable, + }; + if (isSameHydrationSource(hydratedSource.current, source)) { + return; + } + const draft = { + providers, + baselineProviders, + appriseEdits, + appriseNameEdits, + pendingDeletes, + showAppriseForm, + appriseFormMode, + }; + if (hydratedSource.current && hasDraftChanges(draft)) { + return; + } + hydratedSource.current = source; + const nextProviders = storeProviders ?? []; + setProviders(nextProviders); + setBaselineProviders(nextProviders); + setNotificationEvents(storeEvents ?? []); + setAppriseAvailable(storeAppriseAvailable ?? true); + const edits = buildAppriseEdits(nextProviders); + setAppriseEdits(edits.urls); + setAppriseNameEdits(edits.names); + }, [ + appriseEdits, + appriseFormMode, + appriseNameEdits, + baselineProviders, + loaded, + pendingDeletes, + providers, + showAppriseForm, + storeAppriseAvailable, + storeEvents, + storeProviders, + ]); return { providers, setProviders, baselineProviders, setBaselineProviders, notificationEvents, + setNotificationEvents, appriseAvailable, + setAppriseAvailable, appriseName, setAppriseName, appriseUrls, @@ -190,10 +291,7 @@ export function useSaveRequest(state: NotificationsState) { type: "apprise", config: { urls: createDraft.urls }, enabled: true, - events: - state.notificationEvents.length > 0 - ? state.notificationEvents - : DEFAULT_NOTIFICATION_EVENTS, + events: DEFAULT_NOTIFICATION_EVENTS, }) : null; const updatedById = new Map(updated.map((provider) => [provider.id, provider])); @@ -244,22 +342,15 @@ export function useIsDirty(state: NotificationsState) { showAppriseForm, appriseFormMode, } = state; - return ( - (showAppriseForm && appriseFormMode === "create") || - pendingDeletes.size > 0 || - providers.some((provider) => { - const baseline = baselineProviders.find((item) => item.id === provider.id); - if (!baseline) return false; - if (buildProviderUpdate(provider, baseline)) return true; - if (provider.type === "apprise") { - const currentValue = appriseEdits[provider.id] ?? formatAppriseUrls(provider.config?.urls); - const baselineValue = formatAppriseUrls(baseline.config?.urls); - const nameValue = appriseNameEdits[provider.id] ?? provider.name; - return currentValue !== baselineValue || nameValue !== baseline.name; - } - return false; - }) - ); + return hasDraftChanges({ + providers, + baselineProviders, + appriseEdits, + appriseNameEdits, + pendingDeletes, + showAppriseForm, + appriseFormMode, + }); } function useAppriseProviderActions(state: NotificationsState) { diff --git a/apps/web/components/settings/notifications-settings.tsx b/apps/web/components/settings/notifications-settings.tsx index 52731780f9..f049a887fb 100644 --- a/apps/web/components/settings/notifications-settings.tsx +++ b/apps/web/components/settings/notifications-settings.tsx @@ -41,7 +41,7 @@ function DesktopNotificationsSection({
Desktop Notifications

- Notify this device when an agent needs your input. + Notify this device when a selected agent turn, question, or Office event occurs.

diff --git a/apps/web/e2e/helpers/notifications-capture.ts b/apps/web/e2e/helpers/notifications-capture.ts index 219661f7a6..290d30ba6a 100644 --- a/apps/web/e2e/helpers/notifications-capture.ts +++ b/apps/web/e2e/helpers/notifications-capture.ts @@ -13,7 +13,7 @@ export type CapturedNotification = { * The e2e setupPage fixture overrides `window.Notification` with a stub that * never displays a real OS-level toast (those are noisy on developer * machines - Chromium pops the macOS notification center on every agent - * "waiting for input" event during a run). The stub instead records each + * configured session event during a run). The stub instead records each * attempted notification on `window.__kandevTestNotifications` so tests can * assert on what *would* have notified. * diff --git a/apps/web/e2e/tests/settings/mobile-notification-events.spec.ts b/apps/web/e2e/tests/settings/mobile-notification-events.spec.ts new file mode 100644 index 0000000000..d6bca50963 --- /dev/null +++ b/apps/web/e2e/tests/settings/mobile-notification-events.spec.ts @@ -0,0 +1,121 @@ +import { test, expect } from "../../fixtures/test-base"; +import type { Locator } from "@playwright/test"; +import type { ApiClient } from "../../helpers/api-client"; + +const TURN_FINISHED = "session.turn_finished"; +const CLARIFICATION_REQUESTED = "session.clarification_requested"; +const PROVIDER_NAMES = [ + "E2E mobile semantic notifications alpha", + "E2E mobile semantic notifications beta", + "E2E mobile semantic notifications gamma", +]; + +type SeededProvider = { + id: string; +}; + +async function seedNotificationProvider( + apiClient: ApiClient, + name: string, +): Promise { + const response = await apiClient.rawRequest("POST", "/api/v1/notification-providers", { + name, + type: "local", + events: [TURN_FINISHED, CLARIFICATION_REQUESTED], + }); + expect(response.ok).toBe(true); + return (await response.json()) as SeededProvider; +} + +async function expectViewportContained(locator: Locator) { + const box = await locator.boundingBox(); + expect(box).not.toBeNull(); + const viewportWidth = await locator.page().evaluate(() => window.innerWidth); + expect(box!.x).toBeGreaterThanOrEqual(0); + expect(box!.x + box!.width).toBeLessThanOrEqual(viewportWidth); +} + +test.describe("Mobile notification event settings", () => { + test("shows both semantic event rows as touch-operable without horizontal overflow", async ({ + testPage, + apiClient, + }) => { + const providers = await Promise.all( + PROVIDER_NAMES.map((name) => seedNotificationProvider(apiClient, name)), + ); + + try { + await testPage.goto("/settings/general/notifications"); + + const eventContainer = testPage.getByTestId("notification-events-mobile-list"); + const turnFinished = testPage.getByRole("checkbox", { + name: `Agent turn finished for ${PROVIDER_NAMES[0]}`, + }); + const needsAnswer = testPage.getByRole("checkbox", { + name: `Agent needs an answer for ${PROVIDER_NAMES[0]}`, + }); + const turnFinishedTarget = testPage.getByTestId( + `notification-event-toggle-${TURN_FINISHED}-${providers[0].id}`, + ); + const needsAnswerTarget = testPage.getByTestId( + `notification-event-toggle-${CLARIFICATION_REQUESTED}-${providers[0].id}`, + ); + await expect(eventContainer).toBeVisible(); + await expect( + eventContainer.getByText(PROVIDER_NAMES[2], { exact: true }).first(), + ).toBeVisible(); + await expect(eventContainer.getByText("Agent turn finished", { exact: true })).toBeVisible(); + await expect( + eventContainer.getByText("Notify after each completed agent turn."), + ).toBeVisible(); + await expect( + eventContainer.getByText("Agent needs an answer", { exact: true }), + ).toBeVisible(); + await expect( + eventContainer.getByText("Notify when the agent explicitly asks you a question."), + ).toBeVisible(); + await expect(turnFinished).toBeVisible(); + await expect(needsAnswer).toBeVisible(); + + await expectViewportContained( + eventContainer.getByText("Agent turn finished", { exact: true }), + ); + await expectViewportContained( + eventContainer.getByText("Notify after each completed agent turn."), + ); + await expectViewportContained( + eventContainer.getByText("Agent needs an answer", { exact: true }), + ); + await expectViewportContained( + eventContainer.getByText("Notify when the agent explicitly asks you a question."), + ); + await expectViewportContained(turnFinished); + await expectViewportContained(needsAnswer); + for (const target of [turnFinishedTarget, needsAnswerTarget]) { + const box = await target.boundingBox(); + expect(box).not.toBeNull(); + expect(box!.width).toBeGreaterThanOrEqual(44); + expect(box!.height).toBeGreaterThanOrEqual(44); + } + expect( + await eventContainer.evaluate((element) => element.scrollWidth <= element.clientWidth), + ).toBe(true); + + for (const target of [turnFinishedTarget, needsAnswerTarget]) { + await target.scrollIntoViewIfNeeded(); + await target.tap(); + } + await expect(turnFinished).not.toBeChecked(); + await expect(needsAnswer).not.toBeChecked(); + expect( + await testPage.evaluate(() => document.documentElement.scrollWidth > window.innerWidth), + ).toBe(false); + } finally { + await Promise.all( + providers.map((provider) => + apiClient.rawRequest("DELETE", `/api/v1/notification-providers/${provider.id}`), + ), + ); + } + }); +}); diff --git a/apps/web/e2e/tests/settings/settings-manual-save.spec.ts b/apps/web/e2e/tests/settings/settings-manual-save.spec.ts index 7fac2a848b..a689c6953b 100644 --- a/apps/web/e2e/tests/settings/settings-manual-save.spec.ts +++ b/apps/web/e2e/tests/settings/settings-manual-save.spec.ts @@ -1,8 +1,28 @@ import { test, expect } from "../../fixtures/test-base"; +import type { ApiClient } from "../../helpers/api-client"; const APPEARANCE_PATH = "/settings/general/appearance"; const TERMINAL_PATH = "/settings/general/terminal"; const NOTIFICATIONS_PATH = "/settings/general/notifications"; +const CLARIFICATION_REQUESTED = "session.clarification_requested"; +const PROVIDER_NAME = "E2E semantic notifications"; + +type SeededProvider = { + id: string; +}; + +async function seedNotificationProvider( + apiClient: ApiClient, + events: string[], +): Promise { + const response = await apiClient.rawRequest("POST", "/api/v1/notification-providers", { + name: PROVIDER_NAME, + type: "local", + events, + }); + expect(response.ok).toBe(true); + return (await response.json()) as SeededProvider; +} test.describe("Settings manual save", () => { test("keeps Appearance changes local and guards dirty navigation", async ({ @@ -89,6 +109,43 @@ test.describe("Settings manual save", () => { } }); + test("loads seeded semantic events and persists one independently selected event", async ({ + testPage, + apiClient, + }) => { + const provider = await seedNotificationProvider(apiClient, [CLARIFICATION_REQUESTED]); + + try { + await testPage.goto(NOTIFICATIONS_PATH); + + const turnFinished = testPage.getByRole("checkbox", { + name: `Agent turn finished for ${PROVIDER_NAME}`, + }); + const needsAnswer = testPage.getByRole("checkbox", { + name: `Agent needs an answer for ${PROVIDER_NAME}`, + }); + await expect(turnFinished).toBeVisible(); + await expect(needsAnswer).toBeVisible(); + await expect(turnFinished).not.toBeChecked(); + await expect(needsAnswer).toBeChecked(); + + await turnFinished.click(); + await testPage + .getByTestId("settings-floating-save") + .getByRole("button", { name: "Save changes" }) + .click(); + await expect(testPage.getByTestId("settings-floating-save")).not.toBeVisible({ + timeout: 15_000, + }); + + await testPage.reload(); + await expect(turnFinished).toBeChecked(); + await expect(needsAnswer).toBeChecked(); + } finally { + await apiClient.rawRequest("DELETE", `/api/v1/notification-providers/${provider.id}`); + } + }); + test("keeps notification sound changes local until Save", async ({ testPage }) => { await testPage.addInitScript(() => { window.localStorage.setItem( diff --git a/apps/web/hooks/use-session-failure-toast.test.ts b/apps/web/hooks/use-session-failure-toast.test.ts index e6e7de254f..6a1ce67701 100644 --- a/apps/web/hooks/use-session-failure-toast.test.ts +++ b/apps/web/hooks/use-session-failure-toast.test.ts @@ -9,7 +9,6 @@ const mockSetNotificationProvidersLoading = vi.fn(); const mockToast = vi.fn(); let mockProviders: Array> = []; let mockProvidersLoaded = true; -const WAITING_EVENT = "session.waiting_for_input"; vi.mock("@/lib/api", () => ({ listNotificationProviders: vi.fn(), @@ -57,7 +56,7 @@ describe("useSessionFailureToast native delivery", () => { expect(mockClearNotification).toHaveBeenCalledWith(null); }); - it("also emits one native failure notification when the local preference allows it", () => { + it("also emits one native failure notification when the local provider is enabled", () => { const invoke = vi.fn().mockResolvedValue("shown"); (window as Window & { __TAURI_INTERNALS__?: unknown }).__TAURI_INTERNALS__ = { invoke, @@ -68,7 +67,7 @@ describe("useSessionFailureToast native delivery", () => { id: "local", type: "local", enabled: true, - events: [WAITING_EVENT], + events: [], }, ]; mockNotification = { sessionId: "s-native", taskId: "t-1", message: "boom" }; @@ -100,7 +99,7 @@ describe("useSessionFailureToast native delivery", () => { id: "local", type: "local", enabled: false, - events: [WAITING_EVENT], + events: [], }, ]; mockNotification = { sessionId: "s-disabled", taskId: "t-1", message: "boom" }; @@ -126,12 +125,12 @@ describe("useSessionFailureToast native delivery", () => { type: "local", config: {}, enabled: true, - events: [WAITING_EVENT], + events: [], created_at: "2026-07-15T00:00:00Z", updated_at: "2026-07-15T00:00:00Z", }, ], - events: [WAITING_EVENT], + events: [], apprise_available: false, }); mockNotification = { sessionId: "s-lazy", taskId: "t-1", message: "boom" }; @@ -141,7 +140,7 @@ describe("useSessionFailureToast native delivery", () => { await waitFor(() => expect(invoke).toHaveBeenCalledTimes(1)); expect(mockSetNotificationProviders).toHaveBeenCalledWith({ items: expect.any(Array), - events: [WAITING_EVENT], + events: [], appriseAvailable: false, loaded: true, loading: false, diff --git a/apps/web/hooks/use-session-failure-toast.ts b/apps/web/hooks/use-session-failure-toast.ts index 65863d6ad9..4c36342f18 100644 --- a/apps/web/hooks/use-session-failure-toast.ts +++ b/apps/web/hooks/use-session-failure-toast.ts @@ -4,19 +4,13 @@ import { useEffect, useRef } from "react"; import { useAppStore } from "@/components/state-provider"; import { useToast } from "@/components/toast-provider"; import { nativeNotifications } from "@/lib/desktop/native-notification-client"; -import { NOTIFICATION_EVENT_TASK_SESSION_WAITING_FOR_INPUT } from "@/lib/notifications/events"; import { listNotificationProviders } from "@/lib/api"; import type { NotificationProvider } from "@/lib/types/http"; function localSessionNotificationsEnabled( providers: Array<{ type: string; enabled: boolean; events: string[] }>, ): boolean { - return providers.some( - (provider) => - provider.type === "local" && - provider.enabled && - provider.events.includes(NOTIFICATION_EVENT_TASK_SESSION_WAITING_FOR_INPUT), - ); + return providers.some((provider) => provider.type === "local" && provider.enabled); } /** Watches for session failure notifications and shows an error toast. Mount once inside ToastProvider. */ diff --git a/apps/web/lib/notifications/events.ts b/apps/web/lib/notifications/events.ts index 03a2880667..c3c2348a4a 100644 --- a/apps/web/lib/notifications/events.ts +++ b/apps/web/lib/notifications/events.ts @@ -1,10 +1,20 @@ -export const NOTIFICATION_EVENT_TASK_SESSION_WAITING_FOR_INPUT = "session.waiting_for_input"; +export const NOTIFICATION_EVENT_SESSION_TURN_FINISHED = "session.turn_finished"; +export const NOTIFICATION_EVENT_SESSION_CLARIFICATION_REQUESTED = "session.clarification_requested"; +export const NOTIFICATION_EVENT_OFFICE_INBOX_ITEM = "office.inbox_item"; -export const DEFAULT_NOTIFICATION_EVENTS = [NOTIFICATION_EVENT_TASK_SESSION_WAITING_FOR_INPUT]; +export const DEFAULT_NOTIFICATION_EVENTS = [NOTIFICATION_EVENT_SESSION_CLARIFICATION_REQUESTED]; export const EVENT_LABELS: Record = { - [NOTIFICATION_EVENT_TASK_SESSION_WAITING_FOR_INPUT]: { - title: "A session has finished", - description: "Notify when an agent is waiting for your reply.", + [NOTIFICATION_EVENT_SESSION_TURN_FINISHED]: { + title: "Agent turn finished", + description: "Notify after each completed agent turn.", + }, + [NOTIFICATION_EVENT_SESSION_CLARIFICATION_REQUESTED]: { + title: "Agent needs an answer", + description: "Notify when the agent explicitly asks you a question.", + }, + [NOTIFICATION_EVENT_OFFICE_INBOX_ITEM]: { + title: "Office inbox item", + description: "Notify when a new inbox item needs your attention.", }, }; diff --git a/apps/web/lib/types/backend.ts b/apps/web/lib/types/backend.ts index d77536372d..09901f8905 100644 --- a/apps/web/lib/types/backend.ts +++ b/apps/web/lib/types/backend.ts @@ -244,9 +244,17 @@ export type TaskSessionStateChangedPayload = { task_environment_id?: string; }; -export type TaskSessionWaitingForInputPayload = { +export type TaskSessionNotificationPayload = { task_id: string; session_id: string; + occurrence_id: string; + title: string; + body: string; +}; + +export type OfficeInboxItemNotificationPayload = { + task_id?: string; + session_id?: string; title: string; body: string; }; @@ -492,10 +500,15 @@ export type BackendMessageMap = OfficeBackendMessageMap & "session.state_changed", TaskSessionStateChangedPayload >; - "session.waiting_for_input": BackendMessage< - "session.waiting_for_input", - TaskSessionWaitingForInputPayload + "session.turn_finished": BackendMessage< + "session.turn_finished", + TaskSessionNotificationPayload >; + "session.clarification_requested": BackendMessage< + "session.clarification_requested", + TaskSessionNotificationPayload + >; + "office.inbox_item": BackendMessage<"office.inbox_item", OfficeInboxItemNotificationPayload>; "session.agentctl_starting": BackendMessage< "session.agentctl_starting", TaskSessionAgentctlPayload diff --git a/apps/web/lib/ws/handlers/notifications.test.ts b/apps/web/lib/ws/handlers/notifications.test.ts index c4771cb4ce..3302466dc8 100644 --- a/apps/web/lib/ws/handlers/notifications.test.ts +++ b/apps/web/lib/ws/handlers/notifications.test.ts @@ -1,7 +1,10 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import type { StoreApi } from "zustand"; import type { AppState } from "@/lib/state/store"; -import { NOTIFICATION_EVENT_TASK_SESSION_WAITING_FOR_INPUT } from "@/lib/notifications/events"; +import { + NOTIFICATION_EVENT_SESSION_CLARIFICATION_REQUESTED, + NOTIFICATION_EVENT_SESSION_TURN_FINISHED, +} from "@/lib/notifications/events"; import type { BackendMessageMap } from "@/lib/types/backend"; import { registerNotificationsHandlers } from "./notifications"; @@ -14,9 +17,13 @@ import { playWaitingForInputSound } from "@/lib/notifications/sound"; const TASK_ID = "task-1"; const SESSION_ID = "session-1"; const OTHER_TASK_ID = "task-2"; -const MESSAGE_TITLE = "Task needs your input"; -const MESSAGE_BODY = "An agent is waiting for your input."; +const MESSAGE_TITLE = "Agent needs your answer"; +const MESSAGE_BODY = 'The agent asked a question on "Task one".'; const notificationMock = vi.fn(); +type SemanticNotificationMessage = + | BackendMessageMap["session.clarification_requested"] + | BackendMessageMap["session.turn_finished"] + | BackendMessageMap["office.inbox_item"]; function makeStore(overrides: Partial = {}) { const state = { @@ -28,25 +35,48 @@ function makeStore(overrides: Partial = {}) { return { getState: () => state } as unknown as StoreApi; } -function makeMessage(id = "message-1"): BackendMessageMap["session.waiting_for_input"] { +function makeMessage( + id = "message-1", + action: + | "session.turn_finished" + | "session.clarification_requested" + | "office.inbox_item" = "session.clarification_requested", + title = MESSAGE_TITLE, + body = MESSAGE_BODY, +): + | BackendMessageMap["session.clarification_requested"] + | BackendMessageMap["session.turn_finished"] + | BackendMessageMap["office.inbox_item"] { return { id, type: "notification", - action: "session.waiting_for_input", + action, payload: { task_id: TASK_ID, session_id: SESSION_ID, - title: MESSAGE_TITLE, - body: MESSAGE_BODY, + occurrence_id: "pending-1", + title, + body, }, - }; + } as SemanticNotificationMessage; } -function getHandler(store: StoreApi) { - return registerNotificationsHandlers(store)["session.waiting_for_input"]!; +function getHandler( + store: StoreApi, + action: + | "session.turn_finished" + | "session.clarification_requested" + | "office.inbox_item" = "session.clarification_requested", +) { + return ( + registerNotificationsHandlers(store) as unknown as Record< + string, + (message: SemanticNotificationMessage) => void + > + )[action]!; } -describe("session.waiting_for_input handler", () => { +describe("semantic session notification handlers", () => { beforeEach(() => { vi.clearAllMocks(); vi.stubGlobal( @@ -60,6 +90,16 @@ describe("session.waiting_for_input handler", () => { delete (window as Window & { __TAURI_INTERNALS__?: unknown }).__TAURI_INTERNALS__; }); + it("registers handlers for both semantic notification events", () => { + const handlers = registerNotificationsHandlers(makeStore()) as Record; + + expect(handlers[NOTIFICATION_EVENT_SESSION_TURN_FINISHED]).toEqual(expect.any(Function)); + expect(handlers[NOTIFICATION_EVENT_SESSION_CLARIFICATION_REQUESTED]).toEqual( + expect.any(Function), + ); + expect(handlers["office.inbox_item"]).toEqual(expect.any(Function)); + }); + it("delegates event deduplication to the native notification command", () => { const invoke = vi.fn().mockResolvedValue("shown"); (window as Window & { __TAURI_INTERNALS__?: unknown }).__TAURI_INTERNALS__ = { @@ -75,7 +115,7 @@ describe("session.waiting_for_input handler", () => { expect(invoke).toHaveBeenCalledTimes(3); expect(invoke).toHaveBeenNthCalledWith(1, "show_native_notification", { request: { - eventId: "session.waiting_for_input:message-1", + eventId: "session.clarification_requested:message-1", title: MESSAGE_TITLE, body: MESSAGE_BODY, taskId: TASK_ID, @@ -84,7 +124,7 @@ describe("session.waiting_for_input handler", () => { }); expect(invoke).toHaveBeenNthCalledWith(2, "show_native_notification", { request: { - eventId: "session.waiting_for_input:message-1", + eventId: "session.clarification_requested:message-1", title: MESSAGE_TITLE, body: MESSAGE_BODY, taskId: TASK_ID, @@ -93,7 +133,7 @@ describe("session.waiting_for_input handler", () => { }); expect(invoke).toHaveBeenNthCalledWith(3, "show_native_notification", { request: { - eventId: "session.waiting_for_input:message-2", + eventId: "session.clarification_requested:message-2", title: MESSAGE_TITLE, body: MESSAGE_BODY, taskId: TASK_ID, @@ -134,15 +174,6 @@ describe("session.waiting_for_input handler", () => { expect(playWaitingForInputSound).toHaveBeenCalledTimes(1); }); - it("suppresses both channels while the session has no completed turns", () => { - const store = makeStore({ turns: { bySession: {} } } as unknown as Partial); - - getHandler(store)(makeMessage()); - - expect(playWaitingForInputSound).not.toHaveBeenCalled(); - expect(notificationMock).not.toHaveBeenCalled(); - }); - it("suppresses both channels while the user is viewing the task", () => { const store = makeStore({ tasks: { activeTaskId: TASK_ID } } as unknown as Partial); @@ -153,7 +184,7 @@ describe("session.waiting_for_input handler", () => { }); }); -describe("session.waiting_for_input malformed payloads", () => { +describe("semantic session notification malformed payloads", () => { beforeEach(() => { vi.clearAllMocks(); vi.stubGlobal( @@ -176,7 +207,7 @@ describe("session.waiting_for_input malformed payloads", () => { const message = { ...makeMessage(), payload: { ...makeMessage().payload, task_id: undefined }, - } as unknown as BackendMessageMap["session.waiting_for_input"]; + } as unknown as BackendMessageMap["session.clarification_requested"]; getHandler(makeStore())(message); @@ -184,7 +215,7 @@ describe("session.waiting_for_input malformed payloads", () => { expect(notificationMock).toHaveBeenCalledWith(MESSAGE_TITLE, { body: MESSAGE_BODY }); }); - it("uses the task and session as the native identity when the envelope ID is absent", () => { + it("uses the browser notification fallback when a semantic event lacks every occurrence identity", () => { const invoke = vi.fn().mockResolvedValue("shown"); (window as Window & { __TAURI_INTERNALS__?: unknown }).__TAURI_INTERNALS__ = { invoke, @@ -193,14 +224,96 @@ describe("session.waiting_for_input malformed payloads", () => { const message = { ...makeMessage(), id: undefined, - } as unknown as BackendMessageMap["session.waiting_for_input"]; + payload: { ...makeMessage().payload, occurrence_id: undefined }, + } as unknown as BackendMessageMap["session.clarification_requested"]; + + getHandler(makeStore())(message); + + expect(invoke).not.toHaveBeenCalled(); + expect(notificationMock).toHaveBeenCalledWith(MESSAGE_TITLE, { body: MESSAGE_BODY }); + }); + + it("uses the semantic occurrence as the native identity when the envelope ID is absent", () => { + const invoke = vi.fn().mockResolvedValue("shown"); + (window as Window & { __TAURI_INTERNALS__?: unknown }).__TAURI_INTERNALS__ = { + invoke, + transformCallback: vi.fn(), + }; + const message = { + ...makeMessage(), + id: undefined, + payload: { ...makeMessage().payload, occurrence_id: "pending-1" }, + } as unknown as BackendMessageMap["session.clarification_requested"]; getHandler(makeStore())(message); expect(invoke).toHaveBeenCalledWith("show_native_notification", { request: expect.objectContaining({ - eventId: `${NOTIFICATION_EVENT_TASK_SESSION_WAITING_FOR_INPUT}:${TASK_ID}:${SESSION_ID}`, + eventId: `${NOTIFICATION_EVENT_SESSION_CLARIFICATION_REQUESTED}:pending-1`, + }), + }); + }); + + it("keeps turn-finished copy and identity through native delivery", () => { + const invoke = vi.fn().mockResolvedValue("shown"); + (window as Window & { __TAURI_INTERNALS__?: unknown }).__TAURI_INTERNALS__ = { + invoke, + transformCallback: vi.fn(), + }; + + getHandler( + makeStore(), + "session.turn_finished", + )( + makeMessage( + "turn-1", + "session.turn_finished", + "Agent turn finished", + 'The agent finished a turn on "Task one".', + ), + ); + + expect(invoke).toHaveBeenCalledWith("show_native_notification", { + request: expect.objectContaining({ + eventId: "session.turn_finished:turn-1", + title: "Agent turn finished", + body: 'The agent finished a turn on "Task one".', }), }); }); }); + +describe("office inbox notification delivery", () => { + beforeEach(() => { + vi.clearAllMocks(); + vi.stubGlobal( + "Notification", + Object.assign(notificationMock, { permission: "granted" as NotificationPermission }), + ); + }); + + afterEach(() => { + vi.unstubAllGlobals(); + delete (window as Window & { __TAURI_INTERNALS__?: unknown }).__TAURI_INTERNALS__; + }); + + it("delivers office inbox items through the browser fallback", () => { + const invoke = vi.fn().mockResolvedValue("shown"); + (window as Window & { __TAURI_INTERNALS__?: unknown }).__TAURI_INTERNALS__ = { + invoke, + transformCallback: vi.fn(), + }; + + const message = { + ...makeMessage("inbox-1", "office.inbox_item", "", ""), + payload: { session_id: undefined, title: "", body: "" }, + } as BackendMessageMap["office.inbox_item"]; + + getHandler(makeStore(), "office.inbox_item")(message); + + expect(invoke).not.toHaveBeenCalled(); + expect(notificationMock).toHaveBeenCalledWith("New inbox item", { + body: "A new item needs your attention.", + }); + }); +}); diff --git a/apps/web/lib/ws/handlers/notifications.ts b/apps/web/lib/ws/handlers/notifications.ts index 9712eda6c5..a0286d9169 100644 --- a/apps/web/lib/ws/handlers/notifications.ts +++ b/apps/web/lib/ws/handlers/notifications.ts @@ -1,22 +1,20 @@ import type { StoreApi } from "zustand"; -import { NOTIFICATION_EVENT_TASK_SESSION_WAITING_FOR_INPUT } from "@/lib/notifications/events"; +import { + NOTIFICATION_EVENT_OFFICE_INBOX_ITEM, + NOTIFICATION_EVENT_SESSION_CLARIFICATION_REQUESTED, + NOTIFICATION_EVENT_SESSION_TURN_FINISHED, +} from "@/lib/notifications/events"; import { playWaitingForInputSound } from "@/lib/notifications/sound"; import { nativeNotifications } from "@/lib/desktop/native-notification-client"; import type { AppState } from "@/lib/state/store"; -import type { TaskSessionWaitingForInputPayload } from "@/lib/types/backend"; +import type { + OfficeInboxItemNotificationPayload, + TaskSessionNotificationPayload, +} from "@/lib/types/backend"; import type { WsHandlers } from "@/lib/ws/handlers/types"; /** Check whether the notification should be suppressed. */ -function shouldSuppressNotification( - state: AppState, - taskId: string | undefined, - sessionId: string | undefined, -): string | null { - // Suppress during initial preparation — session has no completed turns yet. - if (sessionId) { - const turns = state.turns.bySession[sessionId]; - if (!turns || turns.length === 0) return "session has no completed turns"; - } +function shouldSuppressNotification(state: AppState, taskId: string | undefined): string | null { // Suppress when user is actively viewing this task. if (document.visibilityState === "visible" && taskId && state.tasks.activeTaskId === taskId) { return "user is viewing this task"; @@ -25,39 +23,86 @@ function shouldSuppressNotification( } /** Show the desktop notification when the browser permission allows it. */ -function showDesktopNotification(payload: TaskSessionWaitingForInputPayload): void { +type NotificationPayload = TaskSessionNotificationPayload | OfficeInboxItemNotificationPayload; + +function isSemanticSessionNotification(eventType: string): boolean { + return ( + eventType === NOTIFICATION_EVENT_SESSION_TURN_FINISHED || + eventType === NOTIFICATION_EVENT_SESSION_CLARIFICATION_REQUESTED + ); +} + +function showDesktopNotification(payload: NotificationPayload): void { if (typeof Notification === "undefined" || Notification.permission !== "granted") return; - const title = payload.title || "Task needs your input"; - const body = payload.body || "An agent is waiting for your input."; - new Notification(title, { body }); + new Notification(payload.title, { body: payload.body }); +} + +function registerNotificationHandler( + store: StoreApi, + eventType: string, + fallback: Pick, +) { + return (message: { id?: string; payload: NotificationPayload }) => { + const sessionId = message.payload?.session_id; + const taskId = message.payload?.task_id; + const state = store.getState(); + + const reason = shouldSuppressNotification(state, taskId); + if (reason) return; + + playWaitingForInputSound(); + const payload = { + ...message.payload, + title: message.payload.title || fallback.title, + body: message.payload.body || fallback.body, + }; + const occurrenceID = + message.id ?? + ("occurrence_id" in message.payload ? message.payload.occurrence_id : undefined); + const nativeEventID = isSemanticSessionNotification(eventType) + ? occurrenceID + : (message.id ?? `${taskId}:${sessionId ?? "unknown"}`); + if (nativeNotifications.isAvailable() && taskId && nativeEventID) { + void nativeNotifications + .show({ + eventId: `${eventType}:${nativeEventID}`, + title: payload.title, + body: payload.body, + taskId, + sessionId, + }) + .catch(() => undefined); + } else { + showDesktopNotification(payload); + } + }; } export function registerNotificationsHandlers(store: StoreApi): WsHandlers { return { - [NOTIFICATION_EVENT_TASK_SESSION_WAITING_FOR_INPUT]: (message) => { - const sessionId = message.payload?.session_id as string | undefined; - const taskId = message.payload?.task_id as string | undefined; - const state = store.getState(); - - const reason = shouldSuppressNotification(state, taskId, sessionId); - if (reason) return; - - // Sound is its own opt-in channel — it plays regardless of Notification permission. - playWaitingForInputSound(); - if (nativeNotifications.isAvailable() && taskId) { - const eventId = message.id ?? `${taskId}:${sessionId ?? "unknown"}`; - void nativeNotifications - .show({ - eventId: `${NOTIFICATION_EVENT_TASK_SESSION_WAITING_FOR_INPUT}:${eventId}`, - title: message.payload.title || "Task needs your input", - body: message.payload.body || "An agent is waiting for your input.", - taskId, - sessionId, - }) - .catch(() => undefined); - } else { - showDesktopNotification(message.payload); - } - }, + [NOTIFICATION_EVENT_SESSION_TURN_FINISHED]: registerNotificationHandler( + store, + NOTIFICATION_EVENT_SESSION_TURN_FINISHED, + { + title: "Agent turn finished", + body: "The agent finished a turn.", + }, + ), + [NOTIFICATION_EVENT_SESSION_CLARIFICATION_REQUESTED]: registerNotificationHandler( + store, + NOTIFICATION_EVENT_SESSION_CLARIFICATION_REQUESTED, + { + title: "Agent needs your answer", + body: "The agent asked a question.", + }, + ), + [NOTIFICATION_EVENT_OFFICE_INBOX_ITEM]: registerNotificationHandler( + store, + NOTIFICATION_EVENT_OFFICE_INBOX_ITEM, + { + title: "New inbox item", + body: "A new item needs your attention.", + }, + ), }; } diff --git a/docs/decisions/0039-native-desktop-integration-boundary.md b/docs/decisions/0039-native-desktop-integration-boundary.md index e5692f3ee1..5a9f8a5a1d 100644 --- a/docs/decisions/0039-native-desktop-integration-boundary.md +++ b/docs/decisions/0039-native-desktop-integration-boundary.md @@ -1,6 +1,6 @@ # 0039: Native Desktop Integration Boundary -**Status:** accepted +**Status:** accepted (amended 2026-07-24) **Date:** 2026-07-15 **Area:** desktop, frontend, backend, infra @@ -46,10 +46,14 @@ The app checks on startup and periodically, but download/install/restart always confirmation in the existing System > Updates page. The backend is cleanly stopped before updater relaunch. -Desktop-owned launches reuse existing notification preferences and events but route waiting-for- -input and session-failure delivery through the native notifier. They suppress the legacy backend -shell-command and Web Notification delivery for those same events so one event yields at most one -native notification. Browser, CLI, and service launches retain their existing channels. Dock +Desktop-owned launches reuse existing notification preferences and events but +route selected turn-finished, clarification-requested, and session-failure +delivery through the native notifier. The semantic session event boundary is +defined by +[the semantic notification-event decision](2026-07-24-semantic-notification-events.md). +Desktop launches suppress the legacy backend shell-command and Web Notification +delivery for those same events so one event yields at most one native +notification. Browser, CLI, and service launches retain their existing channels. Dock reopen and second-instance activation focus the window but never infer a notification route. The upstream desktop notification API cannot correlate desktop body clicks with a custom payload, so the app does not navigate from generic focus or activation events. diff --git a/docs/decisions/2026-07-24-semantic-notification-events.md b/docs/decisions/2026-07-24-semantic-notification-events.md new file mode 100644 index 0000000000..6ecc18d10f --- /dev/null +++ b/docs/decisions/2026-07-24-semantic-notification-events.md @@ -0,0 +1,70 @@ +# Semantic notification events come from domain occurrences + +**Status:** accepted +**Date:** 2026-07-24 +**Area:** backend, frontend, desktop + +## Context + +The session lifecycle uses `WAITING_FOR_INPUT` for several operationally +different situations: startup readiness, the idle state after a normal turn, +and an actual clarification request. The notification service subscribed to +that state transition and translated every occurrence into “needs your input.” +Its once-per-session delivery key compounded the problem: a startup or ordinary +turn notification could consume the key before the agent asked a real question. + +The product now exposes ordinary turn completion and explicit clarification as +independently selectable notification events. Their occurrence boundaries must +remain stable even if the orchestrator's lifecycle states change. + +## Decision + +User-facing notifications will be sourced from semantic domain events, not +inferred from generic session lifecycle state. + +- `session.turn_finished` is derived from the durable completion of a specific + agent turn and uses the turn ID as its occurrence identity. +- `session.clarification_requested` is derived from Kandev's structured + agent-authored clarification-request message and uses the shared + pending/request ID as its occurrence identity. Task and session identity are + required. +- Notification idempotency is keyed by provider, semantic event type, and + occurrence identity. The task-session ID remains recorded for routing and + audit but is not the deduplication boundary. +- `session.waiting_for_input` is retired as an emitted notification event. + Existing subscriptions migrate to clarification requests only. +- Local WebSocket and desktop notifications preserve the semantic event type + and occurrence identity rather than collapsing both cases back into a + waiting-state action. + +The user-visible contract is defined by +[Semantic Notifications](../specs/platform/notifications.md). + +## Consequences + +- Startup and ordinary idle transitions cannot claim that the agent needs an + answer. +- Users can opt into every completed turn without coupling that choice to + clarification alerts. +- More than one turn or clarification in a session can notify, while replaying + one domain occurrence remains idempotent. +- Notification persistence requires an occurrence identifier and a replayable + migration from the legacy session-scoped unique key. +- New notification-producing features must identify a semantic domain + occurrence and stable occurrence ID instead of subscribing to a convenient + lifecycle state. + +## Alternatives Considered + +1. **Keep `WAITING_FOR_INPUT` and inspect the previous state.** Rejected because + the same transition still represents both normal completion and + clarification, and lifecycle changes would silently alter user-facing + semantics. +2. **Rename the existing notification to “turn ended.”** Rejected because it + would still fail to distinguish an explicit question and would preserve the + once-per-session suppression bug. +3. **Send both messages from every waiting transition.** Rejected because it + would create notifications for events that did not happen and prevent users + from selecting the two intents independently. +4. **Deduplicate only in memory.** Rejected because replay and process restart + require durable idempotency. diff --git a/docs/decisions/INDEX.md b/docs/decisions/INDEX.md index 26da771c58..b06c844193 100644 --- a/docs/decisions/INDEX.md +++ b/docs/decisions/INDEX.md @@ -44,7 +44,7 @@ Read individual ADRs for full context. Create new ones via `/record decision` or | 0036 | [Normalize ACP shell output at the adapter boundary](0036-normalize-acp-shell-output-at-adapter-boundary.md) | accepted | backend, frontend, protocol | 2026-07-14 | | 0037 | [Resource-aware frontend unit tests](0037-resource-aware-frontend-unit-tests.md) | accepted | frontend, infra | 2026-07-14 | | 0038 | [Quick Chat Repository Isolation](0038-quick-chat-repository-isolation.md) | accepted | backend, frontend | 2026-07-14 | -| 0039 | [Native desktop integration boundary](0039-native-desktop-integration-boundary.md) | accepted | desktop, frontend, backend, infra | 2026-07-15 | +| 0039 | [Native desktop integration boundary](0039-native-desktop-integration-boundary.md) | accepted (amended 2026-07-24) | desktop, frontend, backend, infra | 2026-07-15 | | 0040 | [Separate updater integrity from OS publisher identity](0040-separate-updater-integrity-from-os-publisher-identity.md) | accepted | desktop, infra, workflow | 2026-07-15 | | 0041 | [Backend-owned portable user settings](0041-backend-owned-portable-user-settings.md) | accepted | backend, frontend | 2026-07-15 | | 0042 | [Project shell output and fetch it on demand](0042-project-shell-output-and-fetch-on-demand.md) | accepted | backend, frontend, protocol | 2026-07-16 | @@ -69,3 +69,4 @@ Read individual ADRs for full context. Create new ones via `/record decision` or | 2026-07-20-repository-provider-origin-identity | [Persist Provider Origin In Repository Identity](2026-07-20-repository-provider-origin-identity.md) | accepted | backend, frontend | 2026-07-20 | | 2026-07-21-portable-status-bar-order | [Portable Status Bar Order](2026-07-21-portable-status-bar-order.md) | accepted | backend, frontend, protocol | 2026-07-21 | | 2026-07-21-work-item-reference-search | [Backend-normalized work-item references](2026-07-21-work-item-reference-search.md) | accepted | backend, frontend, protocol | 2026-07-21 | +| 2026-07-24-semantic-notification-events | [Semantic notification events come from domain occurrences](2026-07-24-semantic-notification-events.md) | accepted | backend, frontend, desktop | 2026-07-24 | diff --git a/docs/plans/semantic-notifications/plan.md b/docs/plans/semantic-notifications/plan.md new file mode 100644 index 0000000000..cc4e6cbd44 --- /dev/null +++ b/docs/plans/semantic-notifications/plan.md @@ -0,0 +1,216 @@ +--- +spec: docs/specs/platform/notifications.md +created: 2026-07-24 +status: implemented +--- + +# Implementation Plan: Semantic Notifications + +## Overview + +Replace the overloaded `session.waiting_for_input` trigger with two +independently selectable semantic notifications: +`session.turn_finished` and `session.clarification_requested`. Source them from +durable turn/message events, give each occurrence durable idempotency, migrate +legacy subscriptions to clarification only, and carry the distinction through +Apprise, local browser, and native desktop delivery. + +In the same settings surface, fix cold-load hydration so an asynchronously +loaded provider cannot be mistaken for missing configuration. Preserve unsaved +manual-save drafts once hydration has completed. + +The domain-event boundary is recorded in +[the semantic notification ADR](../../decisions/2026-07-24-semantic-notification-events.md). + +## Backend + +### Semantic event sources and messages + +Likely files: + +- `apps/backend/internal/backendapp/gateway.go` +- `apps/backend/internal/notifications/service/service.go` +- `apps/backend/internal/notifications/service/service_test.go` +- `apps/backend/internal/notifications/models/models.go` +- `apps/backend/pkg/websocket/actions.go` + +Changes: + +- Stop subscribing the notification service to generic + `task.session_state_changed` transitions. +- Subscribe to canonical completed-turn and message-added events. +- Translate one durable completed agent turn into + `session.turn_finished`, keyed by turn ID. +- Translate only a structured `clarification_request` message with + `author_type=agent`, `requests_input=true`, and non-empty task/session/request + identity into `session.clarification_requested`, keyed by the request's shared + `pending_id`. This produces one notification for a multi-question + clarification bundle. +- Produce the exact event-specific title and body from the spec, while keeping + existing task/session/provider routing. +- Do not emit either notification on startup, readiness, user answers, or + generic waiting-state transitions. + +### Occurrence-scoped persistence and migration + +Likely files: + +- `apps/backend/internal/notifications/store/sqlite.go` +- `apps/backend/internal/notifications/store/sqlite_test.go` +- Notification models and repository interfaces used by the store + +Changes: + +- Add a non-empty occurrence ID to delivery records. +- Replace session-scoped uniqueness with + `(provider_id, event_type, occurrence_id)` uniqueness on both SQLite and + Postgres paths. Preserve the task-session ID as data. +- Make the schema upgrade replayable and safe for legacy databases whose + delivery table contains the old inline unique constraint. +- Migrate every legacy `session.waiting_for_input` subscription to + `session.clarification_requested`, merging enabled state idempotently when a + semantic row already exists. Never create a turn-finished subscription from + the legacy key. +- Update the available-event catalog and default subscriptions so + clarification is on by default and turn completion is opt-in. +- Test fresh schema creation, legacy migration, repeated migration, replay of + one occurrence, and multiple occurrences in one session. + +## Frontend and Desktop + +### Settings hydration and event choices + +Likely files: + +- `apps/web/components/settings/notifications-settings-actions.ts` +- `apps/web/components/settings/notifications-settings-actions.test.tsx` +- `apps/web/hooks/domains/settings/use-notification-providers.ts` +- `apps/web/lib/notifications/events.ts` + +Changes: + +- Add the two event rows with the labels and descriptions from the spec. +- Initialize the route-local notification draft when the asynchronous provider + load first completes, rather than snapshotting the initial empty store. +- Hydrate only while the draft is clean/uninitialized so later store refreshes + cannot erase unsaved edits. +- Make new-provider defaults clarification-only. Keep the existing shared + floating Save behavior and provider-specific event checkboxes. +- Add a regression test that starts with an empty store, resolves a provider, + and proves the provider and its selections appear without remounting. Add a + second test proving a refresh does not replace a dirty draft. + +### Local WebSocket and native notification parity + +Likely files: + +- `apps/backend/internal/notifications/providers/local.go` +- `apps/backend/internal/notifications/providers/local_test.go` +- `apps/web/lib/types/backend.ts` +- `apps/web/lib/ws/handlers/notifications.ts` +- `apps/web/lib/ws/handlers/notifications.test.ts` +- `apps/desktop/src-tauri/src/native_notifications.rs` + +Changes: + +- Preserve the semantic event as the local WebSocket action instead of always + broadcasting the legacy waiting action. +- Register both semantic actions in the web client and use event-specific + fallback copy, while retaining existing active-task suppression, sound, and + browser/native delivery behavior. +- Permit both semantic event prefixes at the desktop native boundary and keep + legacy-prefix acceptance only where compatibility requires it. +- Carry the backend occurrence ID through Local WebSocket delivery and use it + as the native de-duplication identity. +- Add frontend and Rust tests for both event types and for rejection of + unrelated native notification events. + +## E2E, Mobile, and Documentation + +Likely files: + +- `apps/web/e2e/tests/settings/settings-manual-save.spec.ts` +- A focused `apps/web/e2e/tests/settings/mobile-*.spec.ts` +- `docs/public/websocket-api.md` +- `docs/public/desktop-app.md` +- `docs/specs/desktop-tauri-app/spec.md` +- `docs/specs/office/inbox.md` +- `docs/decisions/0039-native-desktop-integration-boundary.md` + +Changes: + +- Verify a cold settings load displays a seeded provider and both independent + event controls. +- Verify one event selection can be changed and saved on desktop. +- Render the event/provider matrix as stacked event cards on narrow viewports, + then verify both event rows remain readable and touch-operable at the + repository's canonical mobile viewport without horizontal overflow. +- Update public WebSocket and desktop notification terminology and reconcile + older specs/ADRs that name the retired waiting-state action. + +## Waves + +Wave 1: + +- [x] [task-01-backend-events-and-persistence](task-01-backend-events-and-persistence.md) + +Wave 2: + +- [x] [task-02-frontend-and-desktop](task-02-frontend-and-desktop.md) + +Wave 3: + +- [x] [task-03-e2e-docs-and-verification](task-03-e2e-docs-and-verification.md) + +## Verification + +Targeted backend: + +```bash +cd apps/backend +go test ./internal/notifications/... ./internal/backendapp/... +``` + +Targeted frontend and desktop: + +```bash +cd apps +pnpm --filter @kandev/web test -- --run \ + components/settings/notifications-settings-actions.test.tsx \ + lib/ws/handlers/notifications.test.ts +cd web && pnpm run typecheck +cd ../desktop/src-tauri && cargo test native_notifications +``` + +Focused E2E: + +```bash +cd apps/web +pnpm e2e:run --project chromium \ + tests/settings/settings-manual-save.spec.ts +pnpm e2e:run --project mobile-chrome \ + tests/settings/mobile-notification-events.spec.ts +``` + +Final repository verification: + +```bash +make fmt +make typecheck +make test +make lint +``` + +## Risks + +- The current delivery table embeds its legacy unique constraint in schema + creation, so SQLite migration may require a transactional table rebuild while + Postgres requires a dialect-specific constraint/index migration. +- Clarification bundles contain several messages; filtering on + `requests_input=true` and using their shared request identity is essential to + avoid one alert per question. +- A completed-turn domain event may also participate in recovery paths. Tests + must prove occurrence IDs make replay idempotent without suppressing later + turns. +- Frontend hydration must distinguish initial data arrival from a later server + refresh while the user has unsaved changes. diff --git a/docs/plans/semantic-notifications/task-01-backend-events-and-persistence.md b/docs/plans/semantic-notifications/task-01-backend-events-and-persistence.md new file mode 100644 index 0000000000..80c30dbeaa --- /dev/null +++ b/docs/plans/semantic-notifications/task-01-backend-events-and-persistence.md @@ -0,0 +1,58 @@ +--- +id: "01-backend-events-and-persistence" +title: "Emit and persist semantic notification occurrences" +status: done +wave: 1 +depends_on: [] +plan: "plan.md" +spec: "../../specs/platform/notifications.md" +--- + +# Task 01: Emit and Persist Semantic Notification Occurrences + +## Acceptance + +- Startup and generic session waiting transitions send no notification. +- Each completed agent turn emits `session.turn_finished` with the specified + copy and turn occurrence ID. +- Each structured clarification bundle emits one + `session.clarification_requested` with the specified copy and request + occurrence ID. +- Replaying an occurrence is idempotent, while later occurrences in the same + session can notify. +- Legacy waiting subscriptions migrate idempotently to clarification only + without losing providers, credentials, or delivery history. + +## Verification + +```bash +cd apps/backend +go test ./internal/notifications/... ./internal/backendapp/... +``` + +## Files likely touched + +- `apps/backend/internal/backendapp/gateway.go` +- `apps/backend/internal/notifications/models/models.go` +- `apps/backend/internal/notifications/service/service.go` +- `apps/backend/internal/notifications/store/sqlite.go` +- `apps/backend/internal/notifications/providers/local.go` +- `apps/backend/pkg/websocket/actions.go` +- Related backend tests + +## Dependencies + +None. + +## Inputs + +- Spec sections `Data Model`, `State and Event Semantics`, and + `Persistence Guarantees`. +- ADR `2026-07-24-semantic-notification-events`. +- Existing task `turn.completed` and `message.added` event payloads. + +## Output contract + +Report the exact event filters, occurrence-ID sources, dialect migration +strategy, legacy-subscription merge behavior, files changed, tests run, and +residual risks. diff --git a/docs/plans/semantic-notifications/task-02-frontend-and-desktop.md b/docs/plans/semantic-notifications/task-02-frontend-and-desktop.md new file mode 100644 index 0000000000..8c0bda85a5 --- /dev/null +++ b/docs/plans/semantic-notifications/task-02-frontend-and-desktop.md @@ -0,0 +1,58 @@ +--- +id: "02-frontend-and-desktop" +title: "Expose semantic notification settings and delivery" +status: done +wave: 2 +depends_on: ["01-backend-events-and-persistence"] +plan: "plan.md" +spec: "../../specs/platform/notifications.md" +--- + +# Task 02: Expose Semantic Notification Settings and Delivery + +## Acceptance + +- Settings show independent Agent turn finished and Agent needs an answer + controls with clarification selected by default for a new provider. +- A provider fetched after cold mount hydrates the draft without a remount. +- A later provider refresh does not overwrite unsaved notification edits. +- Local browser and native desktop delivery preserve each semantic action and + use its event-specific copy. +- Existing active-task suppression and sound behavior remain intact. + +## Verification + +```bash +cd apps +pnpm --filter @kandev/web test -- --run \ + components/settings/notifications-settings-actions.test.tsx \ + lib/ws/handlers/notifications.test.ts +cd web && pnpm run typecheck +cd ../desktop/src-tauri && cargo test native_notifications +``` + +## Files likely touched + +- `apps/web/components/settings/notifications-settings-actions.ts` +- `apps/web/components/settings/notifications-settings-actions.test.tsx` +- `apps/web/lib/notifications/events.ts` +- `apps/web/lib/types/backend.ts` +- `apps/web/lib/ws/handlers/notifications.ts` +- `apps/web/lib/ws/handlers/notifications.test.ts` +- `apps/desktop/src-tauri/src/native_notifications.rs` + +## Dependencies + +- Task 01 defines the canonical event/action strings and provider API catalog. + +## Inputs + +- Spec sections `What`, `Failure Modes`, and `Scenarios`. +- Existing settings-route save coordinator contract. +- Existing local notification and native desktop bridges. + +## Output contract + +Report the hydration guard, dirty-draft behavior, UI defaults, action/type +changes, desktop allowlist behavior, files changed, tests run, and residual +risks. diff --git a/docs/plans/semantic-notifications/task-03-e2e-docs-and-verification.md b/docs/plans/semantic-notifications/task-03-e2e-docs-and-verification.md new file mode 100644 index 0000000000..fe431e44d5 --- /dev/null +++ b/docs/plans/semantic-notifications/task-03-e2e-docs-and-verification.md @@ -0,0 +1,63 @@ +--- +id: "03-e2e-docs-and-verification" +title: "Verify notification settings across viewports and update contracts" +status: done +wave: 3 +depends_on: + - "01-backend-events-and-persistence" + - "02-frontend-and-desktop" +plan: "plan.md" +spec: "../../specs/platform/notifications.md" +--- + +# Task 03: Verify Notification Settings and Update Contracts + +## Acceptance + +- A browser E2E proves a seeded provider appears on a cold settings load and + both semantic event controls can be selected independently and saved. +- A mobile E2E at 390px proves both event rows are readable, touch-operable, + and do not cause horizontal overflow. +- Public WebSocket/desktop docs and existing durable specs/ADRs use the semantic + notification names and accurately describe compatibility behavior. +- Targeted checks and the repository verification suite pass. + +## Verification + +```bash +cd apps/web +pnpm e2e:run --project chromium \ + tests/settings/settings-manual-save.spec.ts +pnpm e2e:run --project mobile-chrome \ + tests/settings/mobile-notification-events.spec.ts +cd ../../ +make fmt +make typecheck +make test +make lint +``` + +## Files likely touched + +- `apps/web/e2e/tests/settings/settings-manual-save.spec.ts` +- `apps/web/e2e/tests/settings/mobile-notification-events.spec.ts` +- `docs/public/websocket-api.md` +- `docs/public/desktop-app.md` +- `docs/specs/desktop-tauri-app/spec.md` +- `docs/specs/office/inbox.md` +- `docs/decisions/0039-native-desktop-integration-boundary.md` + +## Dependencies + +- Tasks 01 and 02 complete the behavior exercised by E2E. + +## Inputs + +- Entire semantic notifications spec. +- Mobile-parity canonical viewport and interaction requirements. +- Docs-maintainer terminology and contract checks. + +## Output contract + +Report E2E fixture setup and cleanup, desktop/mobile evidence, documentation +files reconciled, full verification results, and any remaining gaps. diff --git a/docs/public/desktop-app.md b/docs/public/desktop-app.md index 376ffcd5b7..3bb1ade511 100644 --- a/docs/public/desktop-app.md +++ b/docs/public/desktop-app.md @@ -145,7 +145,14 @@ Kandev has no registered public URL scheme, file association, or command-line de External `http`, `https`, and `mailto` destinations open through the system browser or mail client. The desktop bridge rejects URLs with embedded credentials, unsupported schemes, `localhost`/subdomains of `localhost`, loopback IPs, and unspecified IPs; Kandev routes, previews, downloads, and blob URLs remain in the WebView. RFC 1918/private-LAN hosts are not categorically blocked by this validator. -Native notifications are limited to a session waiting for input or failing, and respect the application's notification preference and OS permission. Requests carry task/session identifiers for validation and event de-duplication, but the desktop registers no notification activation/deep-link handler. Clicking a notification is therefore not a supported navigation path. +Native notifications are limited to selected agent-turn-finished and +agent-needs-an-answer events, plus session failures. Semantic session events +respect each provider's event selections. Session failures remain an independent +safety alert whenever the Local provider is enabled. Delivery also respects the +application's notification preference and OS permission. Requests carry +task/session and semantic event identifiers for validation and event +de-duplication, but the desktop registers no notification activation/deep-link +handler. Clicking a notification is therefore not a supported navigation path. ## Troubleshooting diff --git a/docs/public/websocket-api.md b/docs/public/websocket-api.md index 0a2eb5061e..9eb3dbbf7f 100644 --- a/docs/public/websocket-api.md +++ b/docs/public/websocket-api.md @@ -758,7 +758,9 @@ File changes are batched for up to 100 ms and flushed immediately at 50 entries. | Routing key | Action | Notes | |-------------|--------|-------| | subscribed user | `user.settings.updated` | Requires `user.subscribe`. | -| subscribed user | `session.waiting_for_input` | Local user notification with task/session/title/body fields; it is not delivered by `session.subscribe`. | +| subscribed user | `session.turn_finished` | Local user notification for a completed agent turn, with task/session/occurrence/title/body fields; the turn ID is sent as both the envelope `id` and payload `occurrence_id`. It is not delivered by `session.subscribe`. | +| subscribed user | `session.clarification_requested` | Local user notification for a structured agent question that needs an answer, with task/session/occurrence/title/body fields; the clarification pending ID is sent as both the envelope `id` and payload `occurrence_id`. It is not delivered by `session.subscribe`. | +| subscribed user | `office.inbox_item` | Local notification for a selected Office inbox event, with title/body fields. | | subscribed run | `run.event.appended` | Future events only; there is no replay cursor. | | metrics subscribers | `system.metrics.updated` | Live resource snapshot; collection interest follows subscribers. | diff --git a/docs/specs/INDEX.md b/docs/specs/INDEX.md index c276f99c86..4b6febdc16 100644 --- a/docs/specs/INDEX.md +++ b/docs/specs/INDEX.md @@ -38,6 +38,7 @@ Product-wide capabilities that are not tied to a single feature area. |---|---| | [plugins](plugins/spec.md) | draft | | [plugins — marketplace](plugins/marketplace.md) | building | +| [semantic-notifications](platform/notifications.md) | shipped | | [workspace-git-status](platform/workspace-git-status.md) | shipped | ## tasks/ — task & workflow model diff --git a/docs/specs/desktop-tauri-app/spec.md b/docs/specs/desktop-tauri-app/spec.md index 6ab7628597..8777975f7e 100644 --- a/docs/specs/desktop-tauri-app/spec.md +++ b/docs/specs/desktop-tauri-app/spec.md @@ -26,7 +26,8 @@ implemented launch and backend-lifecycle behavior, it provides: - New Task, Check for Updates, Help, external-link, and standard application commands; - persisted window size, position, and maximized state, restored onto a visible display; - signed, prompt-before-install desktop updates through the existing System > Updates page; -- native notifications for waiting-for-input and session-failure events; +- native notifications for selected turn-finished, clarification-requested, and + session-failure events; - focus/reopen behavior for a second launch and Dock activation where supported. The macOS red window close control quits Kandev and stops its owned backend. It does not hide the @@ -134,7 +135,9 @@ OS publisher-identity signing is configured. For desktop-owned launches, Kandev uses native OS notifications for: -- a session waiting for user input; and +- a completed agent turn when that provider event is selected; +- a structured clarification request that needs a user answer when that + provider event is selected; - a session that failed. Notification delivery respects the existing notification settings and suppresses duplicate @@ -146,6 +149,10 @@ so a raw notification click cannot be correlated to an arbitrary payload on ever Generic focus and activation events therefore never navigate. Repeated delivery of the same event identity is deduplicated. +Turn-finished and clarification-requested delivery follows the selected event +checkboxes. Session-failure delivery is an independent safety alert whenever +the Local provider is enabled; it does not reuse either semantic subscription. + ## Desktop bridge and permissions The Go-served SPA may receive a small versioned set of desktop events and commands for menu @@ -244,9 +251,10 @@ display before being shown. Kandev prompts in the existing Updates page before any download, install, or restart. - **GIVEN** a Linux AppImage installation, **WHEN** the user confirms a valid update, **THEN** the signed AppImage updater artifact installs and Kandev relaunches after clean backend shutdown. -- **GIVEN** a session waits for input while Kandev is not focused, **WHEN** its native notification - is delivered, **THEN** Kandev sends no duplicate browser or shell notification and generic app - focus or activation does not navigate to an inferred task. +- **GIVEN** a selected turn-finished or clarification-requested event occurs + while Kandev is not focused, **WHEN** its native notification is delivered, + **THEN** Kandev sends no duplicate browser or shell notification and generic + app focus or activation does not navigate to an inferred task. - **GIVEN** a saved window position belongs to a disconnected display, **WHEN** Kandev reopens, **THEN** the main window appears within a currently visible display. - **GIVEN** the same settings page is opened in a browser or narrow/mobile viewport, **WHEN** it diff --git a/docs/specs/office/inbox.md b/docs/specs/office/inbox.md index f16e60eb11..929179afd4 100644 --- a/docs/specs/office/inbox.md +++ b/docs/specs/office/inbox.md @@ -121,7 +121,11 @@ A sibling `execution_state` JSON field tracks current stage, which participants ### Notifications -A new event type `office.inbox_item` is added alongside the existing `session.waiting_for_input` event. Users subscribe per provider at `/settings/general`. Defaults: Local (browser) and System (OS) auto-subscribed when Office is enabled. Notification content: item type, summary, deep link to `/office/inbox`. +A new event type `office.inbox_item` is available alongside +`session.turn_finished` and `session.clarification_requested`. Users subscribe +per provider at `/settings/general`. Defaults: Local (browser) and System (OS) +auto-subscribed when Office is enabled. Notification content: item type, +summary, deep link to `/office/inbox`. ### Approval flow diff --git a/docs/specs/platform/notifications.md b/docs/specs/platform/notifications.md new file mode 100644 index 0000000000..f5827a2ca4 --- /dev/null +++ b/docs/specs/platform/notifications.md @@ -0,0 +1,171 @@ +--- +status: shipped +created: 2026-07-24 +owner: kandev +--- + +# Semantic Notifications + +## Why + +Kandev currently treats every transition to a session's generic +`WAITING_FOR_INPUT` state as though the agent asked the user a question. That +state is also used before the first turn and after an ordinary completed turn, +so notifications can claim that input is required when it is not. The same +event is deduplicated once per session, which can let an early false positive +suppress a later real clarification request. + +The notification settings page also snapshots an empty provider store during a +cold load. Providers fetched immediately afterward remain absent from the +rendered draft, making an intact configuration look lost. + +## What + +- Notification settings expose two independent session events: + - **Agent turn finished** — notify after each completed agent turn. + - **Agent needs an answer** — notify when the agent explicitly requests a + user response through Kandev's clarification flow. +- A user may enable either event, both events, or neither event for each + notification provider. +- When a provider-create request omits its event selection, the clarification + event is enabled by default and turn-finished remains opt-in. An explicitly + empty selection enables neither event. +- Existing `session.waiting_for_input` subscriptions migrate to **Agent needs an + answer** only. Migration never opts a provider into **Agent turn finished**. +- Session startup, readiness, idle state, and generic + `WAITING_FOR_INPUT` transitions do not themselves send either notification. +- A turn-finished notification uses: + - title: `Agent turn finished` + - body: `The agent finished a turn on "".` +- A clarification notification uses: + - title: `Agent needs your answer` + - body: `The agent asked a question on "".` +- The local browser and native desktop notification path uses the same semantic + event and copy as remote providers such as Apprise. +- Native session-failure alerts remain an independent safety channel whenever + the Local provider is enabled; they are not controlled by either semantic + event checkbox. +- On a cold settings-page load, the provider list and saved event selections + appear after the provider request completes. Initial hydration must not + overwrite edits the user has already made. + +## Data Model + +The durable provider subscription keys are: + +- `session.turn_finished` +- `session.clarification_requested` + +Each delivery records both the task-session ID and a semantic occurrence ID: + +- turn-finished occurrence ID: the completed turn ID; +- clarification occurrence ID: the clarification request's pending/request ID. + +Delivery idempotency is scoped to provider, event type, and occurrence ID. +Consequently, replaying one occurrence does not duplicate a delivery, while +separate turns or clarification requests in the same session may each notify. + +`session.waiting_for_input` is a legacy subscription key accepted only by the +data migration. It is not offered by the settings API after migration and is +not emitted as a new notification event. + +## API Surface + +- The notification-provider API returns the two semantic events in its + available-event catalog. +- Provider create and update requests persist the selected semantic event keys. +- An omitted create-time event field applies the clarification-only default; + an explicitly empty event array persists no subscriptions. +- The WebSocket gateway may emit `session.turn_finished` and + `session.clarification_requested` for the local notification provider. +- Local semantic notifications carry the occurrence ID through the WebSocket + envelope/payload so native de-duplication uses the same identity as backend + delivery de-duplication. +- Existing provider URLs, enabled state, and unrelated event subscriptions are + unchanged. + +## State and Event Semantics + +| Domain occurrence | Turn finished | Needs an answer | +|---|---:|---:| +| Session starts or becomes ready | No | No | +| Session enters generic `WAITING_FOR_INPUT` | No | No | +| An agent turn is durably completed | Yes, when selected | No | +| An agent-authored structured clarification request is created | No | Yes, when selected | +| The user answers a clarification | No | No | +| The same domain event is replayed | No duplicate | No duplicate | +| A later occurrence happens in the same session | Yes | Yes | + +A clarification bundle containing several questions is one occurrence and +produces at most one notification per provider. User-authored, system-authored, +malformed, or unscoped clarification-shaped messages do not notify. + +## Permissions + +Notification configuration continues to use the existing settings +authorization rules. This change grants no additional access to task content. +Notification bodies contain the task title but not the question text. + +## Failure Modes + +- A provider send failure releases that occurrence's de-duplication claim so a + replay may retry it. It does not mark a later occurrence as delivered. +- Replayed event-bus messages do not create duplicate deliveries for the same + provider and occurrence. +- A provider refresh received while persisted-provider drafts are clean may + hydrate their baseline. It must not replace unsaved edits to an existing + provider. A pre-load new-provider form remains intact while the authoritative + provider list hydrates around it. + +## Persistence Guarantees + +- Existing provider records and credentials survive the schema migration. +- An enabled legacy `session.waiting_for_input` subscription becomes an enabled + `session.clarification_requested` subscription. A disabled legacy + subscription remains disabled unless an already-existing semantic + subscription is enabled. +- Migration is idempotent and does not create duplicate subscriptions. +- Existing delivery history remains available for audit, but legacy + session-level deduplication does not suppress new semantic occurrences. +- The de-duplication claim is inserted atomically with its semantic occurrence + identity. A failed provider send removes that exact + provider/event/occurrence claim. + +## Scenarios + +- **GIVEN** a provider subscribed to clarification requests, **WHEN** a session + starts and waits for its first user turn, **THEN** no notification is sent. +- **GIVEN** a provider subscribed only to turn completion, **WHEN** an ordinary + agent turn completes, **THEN** it receives `Agent turn finished` and no copy + claims that user input is required. +- **GIVEN** a provider subscribed only to clarification requests, **WHEN** an + agent issues a structured question bundle, **THEN** it receives one + `Agent needs your answer` notification. +- **GIVEN** two completed turns in one session, **WHEN** turn completion is + selected, **THEN** each turn can produce one notification. +- **GIVEN** a clarification event is delivered and then replayed, **WHEN** both + copies reach the notification service, **THEN** only one delivery is sent. +- **GIVEN** an existing provider subscribed to + `session.waiting_for_input`, **WHEN** the database is upgraded, **THEN** the + provider is subscribed to clarification requests and not to turn completion. +- **GIVEN** saved providers exist, **WHEN** the notifications settings route is + opened from a cold frontend store, **THEN** the providers and their selections + appear after loading without requiring a remount. +- **GIVEN** a 390px mobile viewport, **WHEN** both event rows are displayed, + **THEN** a stacked mobile composition keeps every label, description, provider, + and control readable and touch operable without horizontal scrolling. + +## Out of Scope + +- Inferring that arbitrary prose ending in a question mark requires an answer. +- Including clarification text or other conversation content in notification + payloads. +- Per-task notification policy, schedules, batching, or quiet hours. +- Retrying provider deliveries beyond the existing notification-provider + behavior. +- Adding a new notification-settings fetch-error surface. +- Changing session lifecycle states or the chat waiting-state presentation. + +## Implementation Plan + +See [the implementation plan](../../plans/semantic-notifications/plan.md).
{provider.name} - {provider.type !== "local" && ( - - - - - - Send test notification - - - )} +
{meta.title}
{meta.description}
 -
- onToggleEvent(provider, eventType)} - /> -
- 		+
+ +
+