Collapse filing sync notifications into one batch surface

lib/server/api/app.ts

@@ -73,6 +73,7 @@ import {
 import { answerSearchQuery, searchKnowledgeBase } from '@/lib/server/search';
 import {
   enqueueTask,
+  findOrEnqueueTask,
   findInFlightTask,
   getTaskById,
   getTaskTimeline,
@@ -340,7 +341,7 @@ async function queueAutoFilingSync(
   metadata?: { category?: unknown; tags?: unknown }
 ) {
   try {
-    await enqueueTask({
+    await findOrEnqueueTask({
       userId,
       taskType: 'sync_filings',
       payload: buildSyncFilingsPayload({
@@ -1459,7 +1460,7 @@ export const app = new Elysia({ prefix: '/api' })
     if (shouldQueueSync) {
       try {
         const watchlistItem = await getWatchlistItemByTicker(session.user.id, ticker);
-        await enqueueTask({
+        await findOrEnqueueTask({
           userId: session.user.id,
           taskType: 'sync_filings',
           payload: buildSyncFilingsPayload({
@@ -1661,7 +1662,7 @@ export const app = new Elysia({ prefix: '/api' })
 
     try {
       const limit = typeof payload.limit === 'number' ? payload.limit : Number(payload.limit);
-      const task = await enqueueTask({
+      const task = await findOrEnqueueTask({
         userId: session.user.id,
         taskType: 'sync_filings',
         payload: buildSyncFilingsPayload({

lib/server/api/task-workflow-hybrid.e2e.test.ts

@@ -467,6 +467,60 @@ if (process.env.RUN_TASK_WORKFLOW_E2E === '1') {
     expect(task.payload.tags).toEqual(['semis', 'ai']);
   });
 
+  it('reuses the same in-flight filing sync task for repeated same-ticker requests', async () => {
+    const first = await jsonRequest('POST', '/api/filings/sync', {
+      ticker: 'NVDA',
+      limit: 20
+    });
+    const second = await jsonRequest('POST', '/api/filings/sync', {
+      ticker: 'nvda',
+      limit: 20
+    });
+
+    expect(first.response.status).toBe(200);
+    expect(second.response.status).toBe(200);
+
+    const firstTask = (first.json as { task: { id: string } }).task;
+    const secondTask = (second.json as { task: { id: string } }).task;
+
+    expect(secondTask.id).toBe(firstTask.id);
+
+    const tasksResponse = await jsonRequest('GET', '/api/tasks?limit=10&status=queued&status=running');
+    expect(tasksResponse.response.status).toBe(200);
+
+    const tasks = (tasksResponse.json as {
+      tasks: Array<{ id: string; task_type: string; payload: { ticker?: string } }>;
+    }).tasks.filter((task) => task.task_type === 'sync_filings' && task.payload.ticker === 'NVDA');
+
+    expect(tasks).toHaveLength(1);
+  });
+
+  it('lets different tickers queue independent filing sync tasks', async () => {
+    const nvda = await jsonRequest('POST', '/api/filings/sync', { ticker: 'NVDA', limit: 20 });
+    const msft = await jsonRequest('POST', '/api/filings/sync', { ticker: 'MSFT', limit: 20 });
+    const aapl = await jsonRequest('POST', '/api/filings/sync', { ticker: 'AAPL', limit: 20 });
+
+    const ids = [
+      (nvda.json as { task: { id: string } }).task.id,
+      (msft.json as { task: { id: string } }).task.id,
+      (aapl.json as { task: { id: string } }).task.id
+    ];
+
+    expect(new Set(ids).size).toBe(3);
+
+    const tasksResponse = await jsonRequest('GET', '/api/tasks?limit=10&status=queued&status=running');
+    expect(tasksResponse.response.status).toBe(200);
+
+    const syncTickers = (tasksResponse.json as {
+      tasks: Array<{ task_type: string; payload: { ticker?: string } }>;
+    }).tasks
+      .filter((task) => task.task_type === 'sync_filings')
+      .map((task) => task.payload.ticker)
+      .filter((ticker): ticker is string => typeof ticker === 'string');
+
+    expect(syncTickers.sort()).toEqual(['AAPL', 'MSFT', 'NVDA']);
+  });
+
   it('scopes the filings endpoint by ticker while leaving the global endpoint mixed', async () => {
     if (!sqliteClient) {
       throw new Error('sqlite client not initialized');

lib/server/tasks.ts

@@ -129,6 +129,27 @@ export async function enqueueTask(input: EnqueueTaskInput) {
   }
 }
 
+export async function findOrEnqueueTask(input: EnqueueTaskInput) {
+  if (!input.resourceKey) {
+    return await enqueueTask(input);
+  }
+
+  const existingTask = await findInFlightTaskByResourceKey(
+    input.userId,
+    input.taskType,
+    input.resourceKey
+  );
+
+  if (existingTask) {
+    const reconciledTask = await reconcileTaskWithWorkflow(existingTask);
+    if (reconciledTask.status === 'queued' || reconciledTask.status === 'running') {
+      return reconciledTask;
+    }
+  }
+
+  return await enqueueTask(input);
+}
+
 export async function findInFlightTask(userId: string, taskType: TaskType, resourceKey: string) {
   const task = await findInFlightTaskByResourceKey(userId, taskType, resourceKey);