[2103 by GPT-5.6 Sol] Inbox notification cleanup

packages/api/services/notificationProjectionService.ts

@@ -29,6 +29,7 @@ interface ProjectionLogger {
 type NotificationEventWriter = Pick<NotificationService,
   'createNotificationEvent' | 'createPullRequestNotificationEvent'
   | 'createPullRequestAttentionNotificationEvent'
+  | 'createSourceActivityNotificationEvent'
   | 'reconcileSystemFailureTransition'>;
 
 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, unknown>): string | undefined {
   const issueRef = typeof initial.issueRef === 'object'
     && initial.issueRef !== null
     && !Array.isArray(initial.issueRef)
     ? initial.issueRef as Record<string, unknown>
     : {};
-  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<void> {
+    await this.cleanupResolvedActivities();
     const cutoff = normalizeISO8601Timestamp(this.now().getTime() - this.stalledAfterMs);
     const rows = await this.database<SourceActivityRow>('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<number> {
+    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<number> {
+    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<NotificationRecipient[]> {
     const rows = await this.database('instance_members').distinct('github_user_id') as Array<{
       github_user_id?: unknown;

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) },

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';