feat(orchestrator): add SSE events, queue controls, and mosaic rails sync

apps/orchestrator/src/api/agents/agent-events.service.ts

@@ -0,0 +1,70 @@
+import { Injectable, Logger, OnModuleInit } from "@nestjs/common";
+import { randomUUID } from "crypto";
+import { ValkeyService } from "../../valkey/valkey.service";
+import type { EventHandler, OrchestratorEvent } from "../../valkey/types";
+
+type UnsubscribeFn = () => void;
+
+@Injectable()
+export class AgentEventsService implements OnModuleInit {
+  private readonly logger = new Logger(AgentEventsService.name);
+  private readonly subscribers = new Map<string, EventHandler>();
+  private connected = false;
+
+  constructor(private readonly valkeyService: ValkeyService) {}
+
+  async onModuleInit(): Promise<void> {
+    if (this.connected) return;
+
+    await this.valkeyService.subscribeToEvents(
+      (event) => {
+        this.subscribers.forEach((handler) => {
+          void handler(event);
+        });
+      },
+      (error, _raw, channel) => {
+        this.logger.warn(`Event stream parse/validation warning on ${channel}: ${error.message}`);
+      }
+    );
+
+    this.connected = true;
+    this.logger.log("Agent event stream subscription active");
+  }
+
+  subscribe(handler: EventHandler): UnsubscribeFn {
+    const id = randomUUID();
+    this.subscribers.set(id, handler);
+    return () => {
+      this.subscribers.delete(id);
+    };
+  }
+
+  async getInitialSnapshot(): Promise<{
+    type: "stream.snapshot";
+    timestamp: string;
+    agents: number;
+    tasks: number;
+  }> {
+    const [agents, tasks] = await Promise.all([
+      this.valkeyService.listAgents(),
+      this.valkeyService.listTasks(),
+    ]);
+
+    return {
+      type: "stream.snapshot",
+      timestamp: new Date().toISOString(),
+      agents: agents.length,
+      tasks: tasks.length,
+    };
+  }
+
+  createHeartbeat(): OrchestratorEvent {
+    return {
+      type: "task.processing",
+      timestamp: new Date().toISOString(),
+      data: {
+        heartbeat: true,
+      },
+    };
+  }
+}

apps/orchestrator/src/api/agents/agents-killswitch.controller.spec.ts

@@ -4,6 +4,7 @@ import { QueueService } from "../../queue/queue.service";
 import { AgentSpawnerService } from "../../spawner/agent-spawner.service";
 import { AgentLifecycleService } from "../../spawner/agent-lifecycle.service";
 import { KillswitchService } from "../../killswitch/killswitch.service";
+import { AgentEventsService } from "./agent-events.service";
 import type { KillAllResult } from "../../killswitch/killswitch.service";
 
 describe("AgentsController - Killswitch Endpoints", () => {
@@ -20,6 +21,12 @@ describe("AgentsController - Killswitch Endpoints", () => {
   };
   let mockLifecycleService: {
     getAgentLifecycleState: ReturnType<typeof vi.fn>;
+    registerSpawnedAgent: ReturnType<typeof vi.fn>;
+  };
+  let mockEventsService: {
+    subscribe: ReturnType<typeof vi.fn>;
+    getInitialSnapshot: ReturnType<typeof vi.fn>;
+    createHeartbeat: ReturnType<typeof vi.fn>;
   };
 
   beforeEach(() => {
@@ -38,13 +45,30 @@ describe("AgentsController - Killswitch Endpoints", () => {
 
     mockLifecycleService = {
       getAgentLifecycleState: vi.fn(),
+      registerSpawnedAgent: vi.fn(),
+    };
+
+    mockEventsService = {
+      subscribe: vi.fn().mockReturnValue(() => {}),
+      getInitialSnapshot: vi.fn().mockResolvedValue({
+        type: "stream.snapshot",
+        timestamp: new Date().toISOString(),
+        agents: 0,
+        tasks: 0,
+      }),
+      createHeartbeat: vi.fn().mockReturnValue({
+        type: "task.processing",
+        timestamp: new Date().toISOString(),
+        data: { heartbeat: true },
+      }),
     };
 
     controller = new AgentsController(
       mockQueueService as unknown as QueueService,
       mockSpawnerService as unknown as AgentSpawnerService,
       mockLifecycleService as unknown as AgentLifecycleService,
-      mockKillswitchService as unknown as KillswitchService
+      mockKillswitchService as unknown as KillswitchService,
+      mockEventsService as unknown as AgentEventsService
     );
   });
 

apps/orchestrator/src/api/agents/agents.controller.spec.ts

@@ -3,6 +3,7 @@ import { QueueService } from "../../queue/queue.service";
 import { AgentSpawnerService } from "../../spawner/agent-spawner.service";
 import { AgentLifecycleService } from "../../spawner/agent-lifecycle.service";
 import { KillswitchService } from "../../killswitch/killswitch.service";
+import { AgentEventsService } from "./agent-events.service";
 import { describe, it, expect, beforeEach, afterEach, vi } from "vitest";
 
 describe("AgentsController", () => {
@@ -17,11 +18,17 @@ describe("AgentsController", () => {
   };
   let lifecycleService: {
     getAgentLifecycleState: ReturnType<typeof vi.fn>;
+    registerSpawnedAgent: ReturnType<typeof vi.fn>;
   };
   let killswitchService: {
     killAgent: ReturnType<typeof vi.fn>;
     killAllAgents: ReturnType<typeof vi.fn>;
   };
+  let eventsService: {
+    subscribe: ReturnType<typeof vi.fn>;
+    getInitialSnapshot: ReturnType<typeof vi.fn>;
+    createHeartbeat: ReturnType<typeof vi.fn>;
+  };
 
   beforeEach(() => {
     // Create mock services
@@ -37,6 +44,7 @@ describe("AgentsController", () => {
 
     lifecycleService = {
       getAgentLifecycleState: vi.fn(),
+      registerSpawnedAgent: vi.fn().mockResolvedValue(undefined),
     };
 
     killswitchService = {
@@ -44,12 +52,28 @@ describe("AgentsController", () => {
       killAllAgents: vi.fn(),
     };
 
+    eventsService = {
+      subscribe: vi.fn().mockReturnValue(() => {}),
+      getInitialSnapshot: vi.fn().mockResolvedValue({
+        type: "stream.snapshot",
+        timestamp: new Date().toISOString(),
+        agents: 0,
+        tasks: 0,
+      }),
+      createHeartbeat: vi.fn().mockReturnValue({
+        type: "task.processing",
+        timestamp: new Date().toISOString(),
+        data: { heartbeat: true },
+      }),
+    };
+
     // Create controller with mocked services
     controller = new AgentsController(
       queueService as unknown as QueueService,
       spawnerService as unknown as AgentSpawnerService,
       lifecycleService as unknown as AgentLifecycleService,
-      killswitchService as unknown as KillswitchService
+      killswitchService as unknown as KillswitchService,
+      eventsService as unknown as AgentEventsService
     );
   });
 
@@ -195,6 +219,10 @@ describe("AgentsController", () => {
       expect(queueService.addTask).toHaveBeenCalledWith(validRequest.taskId, validRequest.context, {
         priority: 5,
       });
+      expect(lifecycleService.registerSpawnedAgent).toHaveBeenCalledWith(
+        agentId,
+        validRequest.taskId
+      );
       expect(result).toEqual({
         agentId,
         status: "spawning",

apps/orchestrator/src/api/agents/agents.controller.ts

@@ -11,8 +11,11 @@ import {
   HttpCode,
   UseGuards,
   ParseUUIDPipe,
+  Sse,
+  MessageEvent,
 } from "@nestjs/common";
 import { Throttle } from "@nestjs/throttler";
+import { Observable } from "rxjs";
 import { QueueService } from "../../queue/queue.service";
 import { AgentSpawnerService } from "../../spawner/agent-spawner.service";
 import { AgentLifecycleService } from "../../spawner/agent-lifecycle.service";
@@ -20,6 +23,7 @@ import { KillswitchService } from "../../killswitch/killswitch.service";
 import { SpawnAgentDto, SpawnAgentResponseDto } from "./dto/spawn-agent.dto";
 import { OrchestratorApiKeyGuard } from "../../common/guards/api-key.guard";
 import { OrchestratorThrottlerGuard } from "../../common/guards/throttler.guard";
+import { AgentEventsService } from "./agent-events.service";
 
 /**
  * Controller for agent management endpoints
@@ -41,7 +45,8 @@ export class AgentsController {
     private readonly queueService: QueueService,
     private readonly spawnerService: AgentSpawnerService,
     private readonly lifecycleService: AgentLifecycleService,
-    private readonly killswitchService: KillswitchService
+    private readonly killswitchService: KillswitchService,
+    private readonly eventsService: AgentEventsService
   ) {}
 
   /**
@@ -67,6 +72,9 @@ export class AgentsController {
         context: dto.context,
       });
 
+      // Persist initial lifecycle state in Valkey.
+      await this.lifecycleService.registerSpawnedAgent(spawnResponse.agentId, dto.taskId);
+
       // Queue task in Valkey
       await this.queueService.addTask(dto.taskId, dto.context, {
         priority: 5, // Default priority
@@ -85,6 +93,41 @@ export class AgentsController {
     }
   }
 
+  /**
+   * Stream orchestrator events as server-sent events (SSE)
+   */
+  @Sse("events")
+  @Throttle({ status: { limit: 200, ttl: 60000 } })
+  streamEvents(): Observable<MessageEvent> {
+    return new Observable<MessageEvent>((subscriber) => {
+      let isClosed = false;
+
+      const unsubscribe = this.eventsService.subscribe((event) => {
+        if (!isClosed) {
+          subscriber.next({ data: event });
+        }
+      });
+
+      void this.eventsService.getInitialSnapshot().then((snapshot) => {
+        if (!isClosed) {
+          subscriber.next({ data: snapshot });
+        }
+      });
+
+      const heartbeat = setInterval(() => {
+        if (!isClosed) {
+          subscriber.next({ data: this.eventsService.createHeartbeat() });
+        }
+      }, 15000);
+
+      return () => {
+        isClosed = true;
+        clearInterval(heartbeat);
+        unsubscribe();
+      };
+    });
+  }
+
   /**
    * List all agents
    * @returns Array of all agent sessions with their status

apps/orchestrator/src/api/agents/agents.module.ts

@@ -5,10 +5,11 @@ import { SpawnerModule } from "../../spawner/spawner.module";
 import { KillswitchModule } from "../../killswitch/killswitch.module";
 import { ValkeyModule } from "../../valkey/valkey.module";
 import { OrchestratorApiKeyGuard } from "../../common/guards/api-key.guard";
+import { AgentEventsService } from "./agent-events.service";
 
 @Module({
   imports: [QueueModule, SpawnerModule, KillswitchModule, ValkeyModule],
   controllers: [AgentsController],
-  providers: [OrchestratorApiKeyGuard],
+  providers: [OrchestratorApiKeyGuard, AgentEventsService],
 })
 export class AgentsModule {}

apps/orchestrator/src/api/queue/queue-api.module.ts

@@ -0,0 +1,11 @@
+import { Module } from "@nestjs/common";
+import { QueueController } from "./queue.controller";
+import { QueueModule } from "../../queue/queue.module";
+import { OrchestratorApiKeyGuard } from "../../common/guards/api-key.guard";
+
+@Module({
+  imports: [QueueModule],
+  controllers: [QueueController],
+  providers: [OrchestratorApiKeyGuard],
+})
+export class QueueApiModule {}

apps/orchestrator/src/api/queue/queue.controller.spec.ts

@@ -0,0 +1,65 @@
+import { describe, it, expect, beforeEach, afterEach, vi } from "vitest";
+import { QueueController } from "./queue.controller";
+import { QueueService } from "../../queue/queue.service";
+
+describe("QueueController", () => {
+  let controller: QueueController;
+  let queueService: {
+    getStats: ReturnType<typeof vi.fn>;
+    pause: ReturnType<typeof vi.fn>;
+    resume: ReturnType<typeof vi.fn>;
+  };
+
+  beforeEach(() => {
+    queueService = {
+      getStats: vi.fn(),
+      pause: vi.fn(),
+      resume: vi.fn(),
+    };
+
+    controller = new QueueController(queueService as unknown as QueueService);
+  });
+
+  afterEach(() => {
+    vi.clearAllMocks();
+  });
+
+  it("should return queue stats", async () => {
+    queueService.getStats.mockResolvedValue({
+      pending: 5,
+      active: 1,
+      completed: 10,
+      failed: 2,
+      delayed: 0,
+    });
+
+    const result = await controller.getStats();
+
+    expect(queueService.getStats).toHaveBeenCalledOnce();
+    expect(result).toEqual({
+      pending: 5,
+      active: 1,
+      completed: 10,
+      failed: 2,
+      delayed: 0,
+    });
+  });
+
+  it("should pause queue processing", async () => {
+    queueService.pause.mockResolvedValue(undefined);
+
+    const result = await controller.pause();
+
+    expect(queueService.pause).toHaveBeenCalledOnce();
+    expect(result).toEqual({ message: "Queue processing paused" });
+  });
+
+  it("should resume queue processing", async () => {
+    queueService.resume.mockResolvedValue(undefined);
+
+    const result = await controller.resume();
+
+    expect(queueService.resume).toHaveBeenCalledOnce();
+    expect(result).toEqual({ message: "Queue processing resumed" });
+  });
+});

apps/orchestrator/src/api/queue/queue.controller.ts

@@ -0,0 +1,39 @@
+import { Controller, Get, HttpCode, Post, UseGuards } from "@nestjs/common";
+import { Throttle } from "@nestjs/throttler";
+import { QueueService } from "../../queue/queue.service";
+import { OrchestratorApiKeyGuard } from "../../common/guards/api-key.guard";
+import { OrchestratorThrottlerGuard } from "../../common/guards/throttler.guard";
+
+@Controller("queue")
+@UseGuards(OrchestratorApiKeyGuard, OrchestratorThrottlerGuard)
+export class QueueController {
+  constructor(private readonly queueService: QueueService) {}
+
+  @Get("stats")
+  @Throttle({ status: { limit: 200, ttl: 60000 } })
+  async getStats(): Promise<{
+    pending: number;
+    active: number;
+    completed: number;
+    failed: number;
+    delayed: number;
+  }> {
+    return this.queueService.getStats();
+  }
+
+  @Post("pause")
+  @Throttle({ strict: { limit: 10, ttl: 60000 } })
+  @HttpCode(200)
+  async pause(): Promise<{ message: string }> {
+    await this.queueService.pause();
+    return { message: "Queue processing paused" };
+  }
+
+  @Post("resume")
+  @Throttle({ strict: { limit: 10, ttl: 60000 } })
+  @HttpCode(200)
+  async resume(): Promise<{ message: string }> {
+    await this.queueService.resume();
+    return { message: "Queue processing resumed" };
+  }
+}