feat(notifications): emit at maintenance/turn/sync/reauth events

packages/backend/convex/agentJobs.ts

@@ -4,6 +4,7 @@ import type { Doc, Id } from './_generated/dataModel';
 import type { MutationCtx } from './_generated/server';
 import { internalMutation, mutation, query } from './_generated/server';
 import { getOwnedSpoon, getRequiredUserId, optionalText } from './model';
+import { emitNotification } from './notifications';
 import type { AgentRuntimeName } from './runtimeSupport';
 import { runtimesForProfile } from './runtimeSupport';
 import { assertWorkerToken } from './workerAuth';
@@ -1344,6 +1345,21 @@ export const updateStatus = mutation({
         }
         await ctx.db.patch(job.threadId, threadPatch);
       }
+      if (
+        (args.status === 'changes_ready' ||
+          args.status === 'draft_pr_opened') &&
+        job.status !== args.status
+      ) {
+        await emitNotification(ctx, {
+          userId: job.ownerId,
+          kind: 'agent_turn_finished',
+          title: 'Agent turn finished',
+          body: args.summary ?? job.summary ?? '',
+          link: `/threads/${job.threadId}`,
+          threadId: job.threadId,
+          spoonId: job.spoonId,
+        });
+      }
     }
     return { success: true };
   },
@@ -1687,6 +1703,17 @@ export const applyMaintenanceDecision = mutation({
       createdAt: now,
       updatedAt: now,
     });
+    if (status === 'waiting_for_user') {
+      await emitNotification(ctx, {
+        userId: job.ownerId,
+        kind: 'agent_needs_input',
+        title: 'Agent needs your input',
+        body: args.summary,
+        link: `/threads/${job.threadId}`,
+        threadId: job.threadId,
+        spoonId: job.spoonId,
+      });
+    }
     return { success: true };
   },
 });

packages/backend/convex/github.ts

@@ -8,6 +8,7 @@ import {
   query,
 } from './_generated/server';
 import { getRequiredUserId } from './model';
+import { emitNotification } from './notifications';
 
 export const getInstallUrl = query({
   args: {},
@@ -158,6 +159,15 @@ export const setConnectionStatusByInstallation = internalMutation({
     const now = Date.now();
     for (const connection of connections) {
       await ctx.db.patch(connection._id, { status, updatedAt: now });
+      if (status !== 'active') {
+        await emitNotification(ctx, {
+          userId: connection.userId,
+          kind: 'connection_needs_reauth',
+          title: 'GitHub connection needs re-authorization',
+          body: 'Reconnect your GitHub account to keep syncing.',
+          link: '/settings/integrations',
+        });
+      }
     }
 
     return { updated: connections.length };

packages/backend/convex/githubSync.ts

@@ -378,6 +378,16 @@ const refreshOwnedSpoon = async (
         error: message,
       }),
     ]);
+    if (!rateLimited) {
+      await ctx.runMutation(internal.notifications.emit, {
+        userId: ownerId,
+        kind: 'sync_failed',
+        title: `Sync failed for ${spoon.name}`,
+        body: message,
+        link: `/spoons/${spoonId}`,
+        spoonId,
+      });
+    }
     throw new ConvexError(message);
   }
 };

packages/backend/convex/threads.ts

@@ -15,6 +15,7 @@ import {
   optionalText,
   requireText,
 } from './model';
+import { emitNotification } from './notifications';
 
 const threadSource = v.union(
   v.literal('user_request'),
@@ -500,6 +501,15 @@ export const createMaintenanceThread = internalMutation({
         jobType: args.jobType,
       },
     );
+    await emitNotification(ctx, {
+      userId: args.ownerId,
+      kind: 'maintenance_thread',
+      title: args.title,
+      body: args.summary,
+      link: `/threads/${threadId}`,
+      spoonId: args.spoonId,
+      threadId,
+    });
     return threadId;
   },
 });

packages/backend/tests/unit/git-connection-status.test.ts

@@ -16,6 +16,13 @@ const seedConnection = async (
       email: 'a@b.c',
       name: 'A',
     });
+    // Disable email so a reauth emit does not schedule the best-effort node
+    // email action, whose async logging would race test teardown.
+    await ctx.db.insert('notificationPreferences', {
+      userId,
+      emailEnabled: false,
+      updatedAt: now,
+    });
     return await ctx.db.insert('gitConnections', {
       userId,
       provider: 'github',

packages/backend/tests/unit/maintenance-thread.test.ts

@@ -14,6 +14,13 @@ const seedSpoon = async (t: ReturnType<typeof convexTest>) =>
       email: 'maintenance-thread@example.com',
       name: 'Maintenance Thread Owner',
     });
+    // Disable email so the notification emit does not schedule the best-effort
+    // node email action, whose async logging would race test teardown.
+    await ctx.db.insert('notificationPreferences', {
+      userId: ownerId,
+      emailEnabled: false,
+      updatedAt: now,
+    });
     const spoonId = await ctx.db.insert('spoons', {
       ownerId,
       name: 'Maintenance Thread Spoon',

packages/backend/tests/unit/notifications.test.ts

@@ -1,11 +1,22 @@
 import { convexTest } from 'convex-test';
-import { describe, expect, test } from 'vitest';
+import { beforeAll, describe, expect, test } from 'vitest';
 
-import { internal } from '../../convex/_generated/api';
+import type { Id } from '../../convex/_generated/dataModel';
+import { api, internal } from '../../convex/_generated/api';
 import schema from '../../convex/schema';
 
 const modules = import.meta.glob('../../convex/**/*.*s');
 
+const countNotifications = async (
+  t: ReturnType<typeof convexTest>,
+  userId: Id<'users'>,
+  kind: string,
+) =>
+  await t.run(async (ctx) => {
+    const rows = await ctx.db.query('notifications').collect();
+    return rows.filter((n) => n.userId === userId && n.kind === kind).length;
+  });
+
 describe('notifications schema', () => {
   test('inserts and reads back a notifications row', async () => {
     const t = convexTest(schema, modules);
@@ -173,3 +184,189 @@ describe('emitNotification', () => {
     expect(typeof notification?.emailedAt).toBe('number');
   });
 });
+
+describe('emit points', () => {
+  beforeAll(() => {
+    process.env.SPOON_WORKER_TOKEN = 'test-worker-token';
+  });
+
+  const seedSpoon = async (t: ReturnType<typeof convexTest>) =>
+    await t.run(async (ctx) => {
+      const now = Date.now();
+      const ownerId = await ctx.db.insert('users', {
+        email: 'emit-points@example.com',
+        name: 'Emit Points Owner',
+      });
+      // Disable email so emitNotification does not schedule the best-effort
+      // node email action, which would log asynchronously and race teardown.
+      await ctx.db.insert('notificationPreferences', {
+        userId: ownerId,
+        emailEnabled: false,
+        updatedAt: now,
+      });
+      const spoonId = await ctx.db.insert('spoons', {
+        ownerId,
+        name: 'Emit Points Spoon',
+        provider: 'github',
+        upstreamOwner: 'upstream',
+        upstreamRepo: 'editor',
+        upstreamDefaultBranch: 'main',
+        upstreamUrl: 'https://github.com/upstream/editor',
+        forkOwner: 'team',
+        forkRepo: 'editor-spoon',
+        forkUrl: 'https://github.com/team/editor-spoon',
+        visibility: 'private',
+        maintenanceMode: 'watch',
+        syncCadence: 'daily',
+        productionRefStrategy: 'default_branch',
+        status: 'active',
+        createdAt: now,
+        updatedAt: now,
+      });
+      return { ownerId, spoonId };
+    });
+
+  const maintenanceArgs = (
+    ownerId: Id<'users'>,
+    spoonId: Id<'spoons'>,
+    dedupKey: string,
+  ) => ({
+    spoonId,
+    ownerId,
+    source: 'upstream_update' as const,
+    title: 'Upstream maintenance review',
+    summary: 'Upstream advanced; review recommended.',
+    upstreamTo: 'headsha',
+    dedupKey,
+    jobType: 'maintenance_review' as const,
+  });
+
+  test('createMaintenanceThread emits exactly one maintenance_thread notification, and the dedup call does not add another', async () => {
+    const t = convexTest(schema, modules);
+    const { ownerId, spoonId } = await seedSpoon(t);
+
+    await t.mutation(
+      internal.threads.createMaintenanceThread,
+      maintenanceArgs(ownerId, spoonId, 'base:head'),
+    );
+    expect(await countNotifications(t, ownerId, 'maintenance_thread')).toBe(1);
+
+    // Second call with the same dedupKey reuses the existing thread and must
+    // NOT emit a second notification.
+    await t.mutation(
+      internal.threads.createMaintenanceThread,
+      maintenanceArgs(ownerId, spoonId, 'base:head'),
+    );
+    expect(await countNotifications(t, ownerId, 'maintenance_thread')).toBe(1);
+  });
+
+  const seedThreadedJob = async (
+    t: ReturnType<typeof convexTest>,
+    ownerId: Id<'users'>,
+    spoonId: Id<'spoons'>,
+  ) =>
+    await t.run(async (ctx) => {
+      const now = Date.now();
+      const threadId = await ctx.db.insert('threads', {
+        ownerId,
+        spoonId,
+        title: 'Agent turn thread',
+        summary: 'Working.',
+        source: 'user_request',
+        status: 'running',
+        priority: 'normal',
+        createdAt: now,
+        updatedAt: now,
+      });
+      const requestId = await ctx.db.insert('agentRequests', {
+        spoonId,
+        ownerId,
+        prompt: 'x',
+        status: 'running',
+        createdAt: now,
+        updatedAt: now,
+      });
+      const jobId = await ctx.db.insert('agentJobs', {
+        spoonId,
+        ownerId,
+        agentRequestId: requestId,
+        threadId,
+        status: 'running',
+        prompt: 'x',
+        runtime: 'opencode',
+        baseBranch: 'main',
+        workBranch: 'spoon/x',
+        forkOwner: 'team',
+        forkRepo: 'editor-spoon',
+        forkUrl: 'https://github.com/team/editor-spoon',
+        upstreamOwner: 'upstream',
+        upstreamRepo: 'editor',
+        selectedSecretIds: [],
+        model: '',
+        reasoningEffort: 'medium',
+        claimedBy: 'worker-1',
+        createdAt: now,
+        updatedAt: now,
+      });
+      return jobId;
+    });
+
+  test('updateStatus to changes_ready emits one agent_turn_finished, and a repeat call at the same status does not add another', async () => {
+    const t = convexTest(schema, modules);
+    const { ownerId, spoonId } = await seedSpoon(t);
+    const jobId = await seedThreadedJob(t, ownerId, spoonId);
+
+    await t.mutation(api.agentJobs.updateStatus, {
+      workerToken: 'test-worker-token',
+      workerId: 'worker-1',
+      jobId,
+      status: 'changes_ready',
+      summary: 'Draft is ready.',
+    });
+    expect(await countNotifications(t, ownerId, 'agent_turn_finished')).toBe(1);
+
+    // Same status again: not a transition, so no second notification.
+    await t.mutation(api.agentJobs.updateStatus, {
+      workerToken: 'test-worker-token',
+      workerId: 'worker-1',
+      jobId,
+      status: 'changes_ready',
+      summary: 'Draft is ready.',
+    });
+    expect(await countNotifications(t, ownerId, 'agent_turn_finished')).toBe(1);
+  });
+
+  test('setConnectionStatusByInstallation to a non-active status emits one connection_needs_reauth for the owner', async () => {
+    const t = convexTest(schema, modules);
+    const userId = await t.run(async (ctx) => {
+      const now = Date.now();
+      const userId = await ctx.db.insert('users', {
+        email: 'reauth@example.com',
+        name: 'Reauth Owner',
+      });
+      await ctx.db.insert('notificationPreferences', {
+        userId,
+        emailEnabled: false,
+        updatedAt: now,
+      });
+      await ctx.db.insert('gitConnections', {
+        userId,
+        provider: 'github',
+        displayName: 'GitHub installation 77',
+        installationId: '77',
+        status: 'active',
+        connectedAt: now,
+        updatedAt: now,
+      });
+      return userId;
+    });
+
+    await t.mutation(internal.github.setConnectionStatusByInstallation, {
+      installationId: '77',
+      status: 'revoked',
+    });
+    expect(await countNotifications(t, userId, 'connection_needs_reauth')).toBe(
+      1,
+    );
+  });
+});