diff --git a/packages/api/services/notificationProjectionService.ts b/packages/api/services/notificationProjectionService.ts index 2e5dbb20f..7b68dd62c 100644 --- a/packages/api/services/notificationProjectionService.ts +++ b/packages/api/services/notificationProjectionService.ts @@ -29,6 +29,7 @@ interface ProjectionLogger { type NotificationEventWriter = Pick; export interface NotificationProjectionOptions { @@ -116,15 +117,23 @@ function compactDisplayText(value: unknown): string | undefined { : `${characters.slice(0, 319).join('')}…`; } +function cleanTaskDescription(value: unknown): string | undefined { + const compact = compactDisplayText(value)?.replace(/^New Issue:\s*/i, '').trim(); + if (!compact || /^(?:implementation (?:is )?complete(?:d)?|preparing (?:a )?(?:pr|pull request))\b/i.test(compact)) { + return undefined; + } + return compact; +} + function taskDescription(initial: Record): string | undefined { const issueRef = typeof initial.issueRef === 'object' && initial.issueRef !== null && !Array.isArray(initial.issueRef) ? initial.issueRef as Record : {}; - return compactDisplayText(initial.subtitle) - ?? compactDisplayText(initial.title) - ?? compactDisplayText(issueRef.title); + return cleanTaskDescription(initial.subtitle) + ?? cleanTaskDescription(initial.title) + ?? cleanTaskDescription(issueRef.title); } function stableKey(scope: string, ...parts: unknown[]): string { @@ -255,6 +264,9 @@ export class NotificationProjectionService { startStalledDetector(): void { if (this.stalledTimer) return; + void this.bestEffort('resolved activity cleanup', async () => { + await this.cleanupResolvedActivities(); + }); this.stalledTimer = setInterval(() => { void this.bestEffort('stalled activity', () => this.detectStalledActivities()); }, this.stalledCheckIntervalMs); @@ -367,6 +379,7 @@ export class NotificationProjectionService { } async detectStalledActivities(): Promise { + await this.cleanupResolvedActivities(); const cutoff = normalizeISO8601Timestamp(this.now().getTime() - this.stalledAfterMs); const rows = await this.database('notification_source_activity') .select( @@ -382,7 +395,10 @@ export class NotificationProjectionService { if (row.activity_type === 'task') { const issueNumber = positiveInteger(metadata.issueNumber); const prNumber = positiveInteger(metadata.prNumber); - await this.createPullRequestAwareEvent({ + await this.notifications.createSourceActivityNotificationEvent({ + type: 'task', key: row.activity_key, repository: row.repository, + lastActivityAt: row.last_activity_at, + }, { deduplicationKey: stableKey( 'task-stalled', row.activity_key, row.status, row.last_activity_at, ), @@ -397,9 +413,13 @@ export class NotificationProjectionService { body: `Active work for ${row.repository} has not reported progress.`, actions: taskActions({ active: true }), occurredAt: row.last_activity_at, - }, await this.loadInstanceMemberRecipients(), row.repository, prNumber); + }, await this.loadInstanceMemberRecipients()); } else { - await this.notifications.createNotificationEvent({ + await this.notifications.createSourceActivityNotificationEvent({ + type: 'indexing', key: row.activity_key, repository: row.repository, + ...(row.branch === null ? {} : { branch: row.branch }), + lastActivityAt: row.last_activity_at, + }, { deduplicationKey: stableKey( 'indexing-stalled', row.activity_key, row.status, row.last_activity_at, ), @@ -418,6 +438,16 @@ export class NotificationProjectionService { } } + /** + * Passively heals stale warning cards left by a missed lifecycle event or an + * older server version. Immutable notification events remain available for + * audit; only their active Inbox receipts are dismissed. + */ + async cleanupResolvedActivities(): Promise { + return this.database.transaction(transaction => + this.dismissResolvedActivityReceipts(transaction)); + } + async projectSystemSnapshot( snapshot: SystemHealthSnapshot, additionalAdministratorIds: readonly string[] = [], @@ -529,11 +559,12 @@ export class NotificationProjectionService { ...(context.issueNumber === undefined ? {} : { issueNumber: context.issueNumber }), ...(context.prNumber === undefined ? {} : { prNumber: context.prNumber }), }, - title: context.issueNumber === undefined + title: context.description ?? (context.issueNumber === undefined ? 'Implementation completed' - : `Implementation completed for issue #${context.issueNumber}`, - body: context.description - ?? `Implementation work for ${context.repository} is complete.`, + : `Issue #${context.issueNumber} implementation completed`), + body: context.issueNumber === undefined + ? 'Open task details to review the completed work.' + : `Issue #${context.issueNumber} is complete. Open task details to review the result.`, actions: taskActions({ followup: context.followupEligible, hasPullRequest: pullRequestUrl !== undefined, @@ -559,9 +590,8 @@ export class NotificationProjectionService { target: { type: 'pull_request', repository: context.repository, prNumber, }, - title: `PR #${prNumber} ready for review`, - body: context.description - ?? `Implementation is complete; review the changes in ${context.repository}.`, + title: context.description ?? `PR #${prNumber} ready for review`, + body: `PR #${prNumber} is ready for review.`, actions: [ ...(pullRequestUrl === undefined ? [] : ['open_pr' as const]), 'dismiss', @@ -680,10 +710,62 @@ export class NotificationProjectionService { .select('status', 'last_activity_at') .where({ activity_type: input.type, activity_key: input.key }) .first() as { status?: unknown; last_activity_at?: unknown } | undefined; - return stored?.status === input.status && stored.last_activity_at === input.occurredAt; + const accepted = stored?.status === input.status + && stored.last_activity_at === input.occurredAt; + if (accepted && completedAt !== null) { + await this.dismissResolvedActivityReceipts(transaction); + } + return accepted; }); } + private async dismissResolvedActivityReceipts( + transaction: Knex.Transaction, + ): Promise { + const timestamp = normalizeISO8601Timestamp(this.now()); + const resolvedEvents = transaction('notification_events as event') + .select('event.event_id') + .where({ 'event.severity': 'warning' }) + .andWhere((warning) => { + warning.where((task) => { + task.where({ 'event.kind': 'task' }).whereExists(function resolvedTask() { + this.select(transaction.raw('1')) + .from('notification_source_activity as activity') + .where({ 'activity.activity_type': 'task' }) + .whereNotNull('activity.completed_at') + .whereRaw( + "activity.activity_key = json_extract(event.target_json, '$.taskId')", + ); + }); + }).orWhere((indexing) => { + indexing.where({ 'event.kind': 'indexing' }) + .whereExists(function resolvedIndexing() { + this.select(transaction.raw('1')) + .from('notification_source_activity as activity') + .where({ 'activity.activity_type': 'indexing' }) + .whereNotNull('activity.completed_at') + .whereRaw( + "activity.repository = json_extract(event.target_json, '$.repository')", + ) + .whereRaw( + "activity.branch IS json_extract(event.target_json, '$.branch')", + ); + }); + }); + }); + const changed = await transaction('notification_user_states') + .where({ inbox_enabled: true }) + .whereNull('dismissed_at') + .whereIn('event_id', resolvedEvents) + .update({ + dismissed_at: transaction.raw( + 'CASE WHEN created_at > ? THEN created_at ELSE ? END', + [timestamp, timestamp], + ), + }); + return Number(changed); + } + private async loadInstanceMemberRecipients(): Promise { const rows = await this.database('instance_members').distinct('github_user_id') as Array<{ github_user_id?: unknown; diff --git a/packages/api/test/notificationProjectionService.test.ts b/packages/api/test/notificationProjectionService.test.ts index 2bef7d8f8..d8208b0ba 100644 --- a/packages/api/test/notificationProjectionService.test.ts +++ b/packages/api/test/notificationProjectionService.test.ts @@ -122,8 +122,8 @@ describe('notification lifecycle projection', { concurrency: false }, () => { ); assert.deepEqual(events.map(event => ({ title: event.title, body: event.body })), [ { - title: 'PR #42 ready for review', - body: 'Keep only the newest actionable Inbox update.', + title: 'Keep only the newest actionable Inbox update.', + body: 'PR #42 is ready for review.', }, { title: 'Review completed for PR #7', @@ -140,6 +140,29 @@ describe('notification lifecycle projection', { concurrency: false }, () => { assert.equal(await countNotificationEvents(database), 2); }); + test('uses the issue title instead of boilerplate completion text', async () => { + await database('tasks').insert({ + task_id: 'implementation-description', repository: 'integry/propr', issue_number: 2103, + pr_number: null, task_type: 'issue', + initial_job_data: JSON.stringify({ + title: 'New Issue: Inbox notification cleanup', + subtitle: 'Preparing a PR for issue #2103', + }), + }); + + await projection.projectTaskUpdate({ + eventType: TASK_UPDATE, taskId: 'implementation-description', state: 'completed', + repository: 'integry/propr', issueNumber: 2103, timestamp: iso(), + }); + + const event = await database('notification_events').first(); + assert.equal(event.title, 'Inbox notification cleanup'); + assert.equal( + event.body, + 'Issue #2103 is complete. Open task details to review the result.', + ); + }); + test('ignores stale task transitions and emits one stalled event per unchanged activity', async () => { const activeAt = iso(-30_000); await database('tasks').insert({ @@ -169,6 +192,77 @@ describe('notification lifecycle projection', { concurrency: false }, () => { assert.deepEqual(JSON.parse(events[0].advertised_actions_json), ['stop', 'dismiss']); }); + test('actively dismisses stalled cards when their task reaches a terminal state', async () => { + const processingAt = iso(-30_000); + await database('tasks').insert({ + task_id: 'task-resolved', repository: 'integry/propr', issue_number: 12, + pr_number: null, task_type: 'issue', initial_job_data: '{}', + }); + await projection.projectTaskUpdate({ + eventType: TASK_UPDATE, taskId: 'task-resolved', state: 'processing', + repository: 'integry/propr', issueNumber: 12, timestamp: processingAt, + }); + await projection.detectStalledActivities(); + assert.equal(await countUndismissedNotificationReceipts(database, 'task'), 2); + + clock += 1_000; + await projection.projectTaskUpdate({ + eventType: TASK_UPDATE, taskId: 'task-resolved', state: 'failed', + repository: 'integry/propr', issueNumber: 12, timestamp: iso(), + }); + + const active = await new NotificationService({ database }).listNotifications('admin-user'); + assert.deepEqual(active.notifications.map(notification => notification.title), [ + 'Task failed for issue #12', + ]); + const delayedStall = await new NotificationService({ + database, now: () => new Date(clock), + }).createSourceActivityNotificationEvent({ + type: 'task', key: 'task-resolved', repository: 'integry/propr', + lastActivityAt: processingAt, + }, { + eventId: 'delayed-stalled-card', deduplicationKey: 'delayed-stalled-card', + kind: 'task', severity: 'warning', + target: { + type: 'task', repository: 'integry/propr', taskId: 'task-resolved', issueNumber: 12, + }, + title: 'Task appears stalled', body: 'This delayed card must not be created.', + occurredAt: processingAt, + }, ['admin-user']); + assert.equal(delayedStall, null, 'a delayed detector cannot resurrect a stale card'); + assert.equal(await countNotificationEvents(database), 2, 'audit events are retained'); + }); + + test('passively dismisses a stale activity card created after resolution', async () => { + await database('tasks').insert({ + task_id: 'task-passive-cleanup', repository: 'integry/propr', issue_number: 13, + pr_number: null, task_type: 'issue', initial_job_data: '{}', + }); + await projection.projectTaskUpdate({ + eventType: TASK_UPDATE, taskId: 'task-passive-cleanup', state: 'failed', + repository: 'integry/propr', issueNumber: 13, timestamp: iso(), + }); + const notifications = new NotificationService({ database, now: () => new Date(clock) }); + await notifications.createNotificationEvent({ + eventId: 'legacy-stalled-card', deduplicationKey: 'legacy-stalled-card', + kind: 'task', severity: 'warning', + target: { + type: 'task', repository: 'integry/propr', taskId: 'task-passive-cleanup', + issueNumber: 13, + }, + title: 'Task appears stalled', body: 'This legacy card is no longer relevant.', + occurredAt: iso(), + }, ['admin-user']); + assert.equal(await countUndismissedNotificationReceipts(database, 'task'), 3); + + assert.equal(await projection.cleanupResolvedActivities(), 1); + assert.equal(await projection.cleanupResolvedActivities(), 0); + const active = await notifications.listNotifications('admin-user'); + assert.deepEqual(active.notifications.map(notification => notification.title), [ + 'Task failed for issue #13', + ]); + }); + test('projects a task failure once without copying error details', async () => { await database('tasks').insert({ task_id: 'task-failed', repository: 'integry/propr', issue_number: 99, @@ -280,11 +374,11 @@ describe('notification lifecycle projection', { concurrency: false }, () => { const listed = await new NotificationService({ database }).listNotifications('admin-user'); const lifecycleEvents = listed.notifications.filter(notification => [ 'Task failed for issue #101', - 'Implementation completed for issue #102', + 'Issue #102 implementation completed', 'Review completed for PR #7', ].includes(notification.title)); assert.deepEqual(lifecycleEvents.map(notification => notification.title).sort(), [ - 'Implementation completed for issue #102', + 'Issue #102 implementation completed', 'Review completed for PR #7', 'Task failed for issue #101', ]); @@ -398,6 +492,7 @@ describe('notification lifecycle projection', { concurrency: false }, () => { }, createPullRequestAttentionNotificationEvent: async () => null, createPullRequestNotificationEvent: async () => null, + createSourceActivityNotificationEvent: async () => null, reconcileSystemFailureTransition: async () => ({ accepted: true, event: null }), }, logger: { warn: message => warnings.push(message) }, diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 35e9bb9e6..c65ee1129 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -421,7 +421,7 @@ export { getNotificationPreferences, updateNotificationPreferences, updateNotificationPreference, upsertPushSubscription, listPushSubscriptions, revokePushSubscription, revokePushSubscriptionById, garbageCollectPushSubscriptions } from './services/notificationService.js'; -export type { NotificationRecipientInput, NotificationRecipient, CreateNotificationEventInput, NotificationListOptions, NotificationServiceOptions } from './services/notificationService.js'; +export type { NotificationRecipientInput, NotificationRecipient, CreateNotificationEventInput, NotificationListOptions, NotificationServiceOptions, NotificationSourceActivityIdentity } from './services/notificationService.js'; export { DEFAULT_NOTIFICATION_LIST_LIMIT, MAX_NOTIFICATION_LIST_LIMIT, NotificationQueryValidationError, parseNotificationListLimit, encodeNotificationCursor, decodeNotificationCursor } from './services/notificationPagination.js'; export type { NotificationCursor } from './services/notificationPagination.js'; diff --git a/packages/core/src/services/notificationService.ts b/packages/core/src/services/notificationService.ts index ad01155d8..edfb6f4f5 100644 --- a/packages/core/src/services/notificationService.ts +++ b/packages/core/src/services/notificationService.ts @@ -117,6 +117,14 @@ export interface SystemFailureTransitionResult { event: NotificationEvent<'system_failure'> | null; } +export interface NotificationSourceActivityIdentity { + type: 'task' | 'indexing'; + key: string; + repository: string; + branch?: string; + lastActivityAt: TimestampInput; +} + interface NotificationEventRow { event_id: string; deduplication_key: string; @@ -374,6 +382,73 @@ export class NotificationService { }); } + /** + * Create a stalled-activity event only while the activity snapshot that + * triggered it is still current. Terminal transitions can therefore race + * a detector safely: either this transaction creates the card first and + * the transition dismisses it, or this check observes the transition and + * skips the obsolete card. + */ + async createSourceActivityNotificationEvent( + source: NotificationSourceActivityIdentity, + input: CreateNotificationEventInput, + recipients: readonly NotificationRecipient[] = input.recipients ?? [] + ): Promise | null> { + assertIdentifier(source.key, 'notification source activity key'); + const lastActivityAt = normalizeISO8601Timestamp(source.lastActivityAt); + const event = this.prepareNotificationEvent(input); + const normalizedRecipients = normalizeRecipients(recipients); + + if (source.type !== input.kind || input.severity !== 'warning') { + throw new TypeError('source activity notifications must be matching warning events'); + } + if (input.target.repository !== source.repository) { + throw new TypeError('source activity notification repository must match its source'); + } + if (source.type === 'task') { + if (input.target.type !== 'task' || input.target.taskId !== source.key) { + throw new TypeError('task notification target must match its source activity'); + } + } else if ( + input.target.type !== 'indexing' + || input.target.branch !== source.branch + ) { + throw new TypeError('indexing notification target must match its source activity'); + } + + return this.database.transaction(async transaction => { + if ( + input.target.type === 'task' + && input.target.prNumber !== undefined + && !await this.pullRequestIsOpen( + transaction, + input.target.repository, + input.target.prNumber + ) + ) return null; + const current = await transaction('notification_source_activity') + .select('status', 'last_activity_at', 'completed_at') + .where({ + activity_type: source.type, + activity_key: source.key, + repository: source.repository, + branch: source.branch ?? null + }) + .first() as { + status?: unknown; + last_activity_at?: unknown; + completed_at?: unknown; + } | undefined; + if ( + current?.completed_at !== null + || (current.status !== 'queued' && current.status !== 'processing') + || current.last_activity_at !== lastActivityAt + ) return null; + + return this.persistNotificationEvent(transaction, event, normalizedRecipients); + }); + } + /** * Create or reuse a PR-attention event and supersede older cards in the * same transaction that checks the durable merge marker.