feat: add leader AgentBus task summaries

control-plane/src/server.ts

@@ -18,6 +18,7 @@ import {
   AgentBusChannelService,
   mergeRuntimeChannelStatuses
 } from './agentbus-channels.js';
+import { LeaderNotificationService } from './leader-notification-service.js';
 import {
   TaskError,
   TaskService,
@@ -156,6 +157,16 @@ const channelUpdateSchema = z.object({
   enabled: z.boolean().optional()
 });
 const channelRotateKeySchema = z.object({ agentbus_key: z.string().min(1).max(4_000) });
+const leaderSummarySubscriptionSchema = z.object({
+  channel_id: z.string().uuid(),
+  recipient_address: z.string().trim().min(1).max(500).optional(),
+  conversation_id: z.string().trim().min(1).max(500).optional(),
+  include_manual: z.boolean(),
+  include_agentbus: z.boolean(),
+  enabled: z.boolean(),
+  target_verified: z.boolean().optional(),
+  expected_revision: z.number().int().min(0).optional()
+});
 const listTasksQuerySchema = z.object({
   status: z.string().max(80).optional(),
   search: z.string().max(200).optional(),
@@ -524,6 +535,11 @@ export async function buildServer({
     warn: (metadata, message) => app.log.warn(agentBusDiagnosticMetadata(metadata), message),
     error: (metadata, message) => app.log.error(agentBusDiagnosticMetadata(metadata), message)
   });
+  const leaderNotificationService = new LeaderNotificationService(config, {
+    info: (metadata, message) => app.log.info(agentBusDiagnosticMetadata(metadata), message),
+    warn: (metadata, message) => app.log.warn(agentBusDiagnosticMetadata(metadata), message),
+    error: (metadata, message) => app.log.error(agentBusDiagnosticMetadata(metadata), message)
+  });
   let agentBus: AgentBusManager | null = null;
 
   const getSession = async (request: FastifyRequest): Promise<ActiveSession> => {
@@ -805,6 +821,7 @@ export async function buildServer({
       tasks,
       organizationId: organization.id,
       scheduleParseQueue,
+      leaderNotifications: leaderNotificationService,
       logger: {
         info: (metadata, message) => app.log.info(agentBusDiagnosticMetadata(metadata), message),
         warn: (metadata, message) => app.log.warn(agentBusDiagnosticMetadata(metadata), message),
@@ -812,7 +829,7 @@ export async function buildServer({
       }
     });
     await agentBus.start();
-    app.addHook('onClose', async () => agentBus?.stop());
+    leaderNotificationService.startProjector(organization.id);
   }
 
   async function getAiProbe(): Promise<unknown> {
@@ -1166,6 +1183,35 @@ export async function buildServer({
     return { ok: true, ...result };
   });
 
+  app.get('/api/settings/leader-summary-subscriptions', async (request) => {
+    const session = await requireAdminSession(request);
+    return {
+      ok: true,
+      subscriptions: await leaderNotificationService.listSubscriptions(session.user.organizationId)
+    };
+  });
+
+  app.put('/api/settings/leader-summary-subscriptions/:leaderUserId', async (request) => {
+    const session = await requireAdminMutationSession(request);
+    const params = request.params as { leaderUserId: string };
+    const body = leaderSummarySubscriptionSchema.parse(request.body);
+    const subscription = await leaderNotificationService.upsertSubscription(
+      contextFor(session, request),
+      {
+        leaderUserId: params.leaderUserId,
+        channelId: body.channel_id,
+        recipientAddress: body.recipient_address,
+        conversationId: body.conversation_id,
+        includeManual: body.include_manual,
+        includeAgentBus: body.include_agentbus,
+        enabled: body.enabled,
+        targetVerified: body.target_verified,
+        expectedRevision: body.expected_revision
+      }
+    );
+    return { ok: true, subscription };
+  });
+
   app.post('/api/auth/logout', async (request, reply) => {
     setAuthNoStore(reply);
     const session = await requireAuthenticatedMutationSession(request);
@@ -1539,9 +1585,11 @@ export async function buildServer({
       diagnostic_event: 'service.closing',
       diagnostic_stage: 'shutdown'
     }, 'control plane closing');
+    await leaderNotificationService.stopProjector();
+    await agentBus?.stop();
     await closePool();
   });
-  return { app, auth, tasks, agentBus, channelService };
+  return { app, auth, tasks, agentBus, channelService, leaderNotificationService };
 }
 
 function installProcessDiagnostics(app: Awaited<ReturnType<typeof buildServer>>['app']): void {