diff --git a/packages/backend/convex/agentJobs.ts b/packages/backend/convex/agentJobs.ts index 364bed9..75aceeb 100644 --- a/packages/backend/convex/agentJobs.ts +++ b/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 }; }, }); diff --git a/packages/backend/convex/github.ts b/packages/backend/convex/github.ts index 918d724..d7ef6b9 100644 --- a/packages/backend/convex/github.ts +++ b/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 }; diff --git a/packages/backend/convex/githubSync.ts b/packages/backend/convex/githubSync.ts index a5cc6ce..bebe0bf 100644 --- a/packages/backend/convex/githubSync.ts +++ b/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); } }; diff --git a/packages/backend/convex/threads.ts b/packages/backend/convex/threads.ts index 8c28720..f57e420 100644 --- a/packages/backend/convex/threads.ts +++ b/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; }, }); diff --git a/packages/backend/tests/unit/git-connection-status.test.ts b/packages/backend/tests/unit/git-connection-status.test.ts index ba75ddb..a679148 100644 --- a/packages/backend/tests/unit/git-connection-status.test.ts +++ b/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', diff --git a/packages/backend/tests/unit/maintenance-thread.test.ts b/packages/backend/tests/unit/maintenance-thread.test.ts index 8b6068c..a070ebe 100644 --- a/packages/backend/tests/unit/maintenance-thread.test.ts +++ b/packages/backend/tests/unit/maintenance-thread.test.ts @@ -14,6 +14,13 @@ const seedSpoon = async (t: ReturnType) => 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', diff --git a/packages/backend/tests/unit/notifications.test.ts b/packages/backend/tests/unit/notifications.test.ts index a512f0d..9143dd8 100644 --- a/packages/backend/tests/unit/notifications.test.ts +++ b/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, + 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) => + 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, + 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, + ); + }); +});