feat(orchestrator): add operator control endpoints

Co-authored-by: Jason Woltje <jason@diversecanvas.com>
Co-committed-by: Jason Woltje <jason@diversecanvas.com>

apps/orchestrator/Dockerfile

@@ -21,8 +21,7 @@ FROM base AS deps
 COPY packages/shared/package.json ./packages/shared/
 COPY packages/config/package.json ./packages/config/
 COPY apps/orchestrator/package.json ./apps/orchestrator/
-# Copy API prisma schema so prisma generate can run in the orchestrator build
-COPY apps/api/prisma ./apps/api/prisma
+# API schema is available via apps/orchestrator/prisma/schema.prisma symlink
 
 # Copy npm configuration for native binary architecture hints
 COPY .npmrc ./
@@ -48,9 +47,14 @@ COPY --from=deps /app/packages/shared/node_modules ./packages/shared/node_module
 COPY --from=deps /app/packages/config/node_modules ./packages/config/node_modules
 COPY --from=deps /app/apps/orchestrator/node_modules ./apps/orchestrator/node_modules
 
-# Copy API prisma schema and generate the Prisma client for the orchestrator
-COPY apps/api/prisma ./apps/api/prisma
-RUN pnpm --filter=@mosaic/orchestrator prisma:generate
+# The repo has apps/orchestrator/prisma/schema.prisma as a symlink for CI use.
+# Kaniko resolves destination symlinks on COPY, which fails because the symlink
+# target (../../api/prisma/schema.prisma) doesn't exist in the container.
+# Fix: remove the dangling symlink first, then copy the real schema file there.
+RUN rm -f apps/orchestrator/prisma/schema.prisma
+COPY apps/api/prisma/schema.prisma ./apps/orchestrator/prisma/schema.prisma
+# pnpm turbo build runs prisma:generate (--schema=./prisma/schema.prisma) from the
+# orchestrator package context — no cross-package project-root issues.
 
 # Build the orchestrator app using TurboRepo
 RUN pnpm turbo build --filter=@mosaic/orchestrator

apps/orchestrator/package.json

@@ -7,7 +7,7 @@
     "dev": "nest start --watch",
     "lint": "eslint src/",
     "lint:fix": "eslint src/ --fix",
-    "prisma:generate": "prisma generate --schema=../api/prisma/schema.prisma",
+    "prisma:generate": "prisma generate --schema=./prisma/schema.prisma",
     "start": "node dist/main.js",
     "start:debug": "nest start --debug --watch",
     "start:dev": "nest start --watch",
@@ -47,6 +47,7 @@
     "@types/express": "^5.0.1",
     "@types/node": "^22.13.4",
     "@vitest/coverage-v8": "^4.0.18",
+    "prisma": "^6.19.2",
     "ts-node": "^10.9.2",
     "tsconfig-paths": "^4.2.0",
     "typescript": "^5.8.2",

apps/orchestrator/prisma/schema.prisma

@@ -0,0 +1 @@
+../../api/prisma/schema.prisma

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

@@ -0,0 +1,68 @@
+import { Injectable } from "@nestjs/common";
+import type { Prisma } from "@prisma/client";
+import { PrismaService } from "../../prisma/prisma.service";
+
+@Injectable()
+export class AgentControlService {
+  constructor(private readonly prisma: PrismaService) {}
+
+  private toJsonValue(value: Record<string, unknown>): Prisma.InputJsonValue {
+    return value as Prisma.InputJsonValue;
+  }
+
+  private async createOperatorAuditLog(
+    agentId: string,
+    operatorId: string,
+    action: "inject" | "pause" | "resume",
+    payload: Record<string, unknown>
+  ): Promise<void> {
+    await this.prisma.operatorAuditLog.create({
+      data: {
+        sessionId: agentId,
+        userId: operatorId,
+        provider: "internal",
+        action,
+        metadata: this.toJsonValue({ payload }),
+      },
+    });
+  }
+
+  async injectMessage(agentId: string, operatorId: string, message: string): Promise<void> {
+    const treeEntry = await this.prisma.agentSessionTree.findUnique({
+      where: { sessionId: agentId },
+      select: { id: true },
+    });
+
+    if (treeEntry) {
+      await this.prisma.agentConversationMessage.create({
+        data: {
+          sessionId: agentId,
+          role: "operator",
+          content: message,
+          provider: "internal",
+          metadata: this.toJsonValue({}),
+        },
+      });
+    }
+
+    await this.createOperatorAuditLog(agentId, operatorId, "inject", { message });
+  }
+
+  async pauseAgent(agentId: string, operatorId: string): Promise<void> {
+    await this.prisma.agentSessionTree.updateMany({
+      where: { sessionId: agentId },
+      data: { status: "paused" },
+    });
+
+    await this.createOperatorAuditLog(agentId, operatorId, "pause", {});
+  }
+
+  async resumeAgent(agentId: string, operatorId: string): Promise<void> {
+    await this.prisma.agentSessionTree.updateMany({
+      where: { sessionId: agentId },
+      data: { status: "running" },
+    });
+
+    await this.createOperatorAuditLog(agentId, operatorId, "resume", {});
+  }
+}

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

@@ -6,6 +6,7 @@ import { AgentLifecycleService } from "../../spawner/agent-lifecycle.service";
 import { KillswitchService } from "../../killswitch/killswitch.service";
 import { AgentEventsService } from "./agent-events.service";
 import { AgentMessagesService } from "./agent-messages.service";
+import { AgentControlService } from "./agent-control.service";
 import type { KillAllResult } from "../../killswitch/killswitch.service";
 
 describe("AgentsController - Killswitch Endpoints", () => {
@@ -35,6 +36,11 @@ describe("AgentsController - Killswitch Endpoints", () => {
     getReplayMessages: ReturnType<typeof vi.fn>;
     getMessagesAfter: ReturnType<typeof vi.fn>;
   };
+  let mockControlService: {
+    injectMessage: ReturnType<typeof vi.fn>;
+    pauseAgent: ReturnType<typeof vi.fn>;
+    resumeAgent: ReturnType<typeof vi.fn>;
+  };
 
   beforeEach(() => {
     mockKillswitchService = {
@@ -77,13 +83,20 @@ describe("AgentsController - Killswitch Endpoints", () => {
       getMessagesAfter: vi.fn().mockResolvedValue([]),
     };
 
+    mockControlService = {
+      injectMessage: vi.fn().mockResolvedValue(undefined),
+      pauseAgent: vi.fn().mockResolvedValue(undefined),
+      resumeAgent: vi.fn().mockResolvedValue(undefined),
+    };
+
     controller = new AgentsController(
       mockQueueService as unknown as QueueService,
       mockSpawnerService as unknown as AgentSpawnerService,
       mockLifecycleService as unknown as AgentLifecycleService,
       mockKillswitchService as unknown as KillswitchService,
       mockEventsService as unknown as AgentEventsService,
-      mockMessagesService as unknown as AgentMessagesService
+      mockMessagesService as unknown as AgentMessagesService,
+      mockControlService as unknown as AgentControlService
     );
   });
 

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

@@ -5,6 +5,7 @@ import { AgentLifecycleService } from "../../spawner/agent-lifecycle.service";
 import { KillswitchService } from "../../killswitch/killswitch.service";
 import { AgentEventsService } from "./agent-events.service";
 import { AgentMessagesService } from "./agent-messages.service";
+import { AgentControlService } from "./agent-control.service";
 import { describe, it, expect, beforeEach, afterEach, vi } from "vitest";
 
 describe("AgentsController", () => {
@@ -36,6 +37,11 @@ describe("AgentsController", () => {
     getReplayMessages: ReturnType<typeof vi.fn>;
     getMessagesAfter: ReturnType<typeof vi.fn>;
   };
+  let controlService: {
+    injectMessage: ReturnType<typeof vi.fn>;
+    pauseAgent: ReturnType<typeof vi.fn>;
+    resumeAgent: ReturnType<typeof vi.fn>;
+  };
 
   beforeEach(() => {
     // Create mock services
@@ -81,6 +87,12 @@ describe("AgentsController", () => {
       getMessagesAfter: vi.fn().mockResolvedValue([]),
     };
 
+    controlService = {
+      injectMessage: vi.fn().mockResolvedValue(undefined),
+      pauseAgent: vi.fn().mockResolvedValue(undefined),
+      resumeAgent: vi.fn().mockResolvedValue(undefined),
+    };
+
     // Create controller with mocked services
     controller = new AgentsController(
       queueService as unknown as QueueService,
@@ -88,7 +100,8 @@ describe("AgentsController", () => {
       lifecycleService as unknown as AgentLifecycleService,
       killswitchService as unknown as KillswitchService,
       eventsService as unknown as AgentEventsService,
-      messagesService as unknown as AgentMessagesService
+      messagesService as unknown as AgentMessagesService,
+      controlService as unknown as AgentControlService
     );
   });
 
@@ -378,6 +391,47 @@ describe("AgentsController", () => {
     });
   });
 
+  describe("agent control endpoints", () => {
+    const agentId = "0b64079f-4487-42b9-92eb-cf8ea0042a64";
+
+    it("should inject an operator message", async () => {
+      const req = { apiKey: "control-key" };
+
+      const result = await controller.injectAgentMessage(
+        agentId,
+        { message: "pause and summarize" },
+        req
+      );
+
+      expect(controlService.injectMessage).toHaveBeenCalledWith(
+        agentId,
+        "control-key",
+        "pause and summarize"
+      );
+      expect(result).toEqual({ message: `Message injected into agent ${agentId}` });
+    });
+
+    it("should default operator id when request api key is missing", async () => {
+      await controller.injectAgentMessage(agentId, { message: "continue" }, {});
+
+      expect(controlService.injectMessage).toHaveBeenCalledWith(agentId, "operator", "continue");
+    });
+
+    it("should pause an agent", async () => {
+      const result = await controller.pauseAgent(agentId, {}, { apiKey: "ops-user" });
+
+      expect(controlService.pauseAgent).toHaveBeenCalledWith(agentId, "ops-user");
+      expect(result).toEqual({ message: `Agent ${agentId} paused` });
+    });
+
+    it("should resume an agent", async () => {
+      const result = await controller.resumeAgent(agentId, {}, { apiKey: "ops-user" });
+
+      expect(controlService.resumeAgent).toHaveBeenCalledWith(agentId, "ops-user");
+      expect(result).toEqual({ message: `Agent ${agentId} resumed` });
+    });
+  });
+
   describe("getAgentMessages", () => {
     it("should return paginated message history", async () => {
       const agentId = "0b64079f-4487-42b9-92eb-cf8ea0042a64";

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

@@ -14,6 +14,7 @@ import {
   Sse,
   MessageEvent,
   Query,
+  Request,
 } from "@nestjs/common";
 import type { AgentConversationMessage } from "@prisma/client";
 import { Throttle } from "@nestjs/throttler";
@@ -28,6 +29,9 @@ import { OrchestratorThrottlerGuard } from "../../common/guards/throttler.guard"
 import { AgentEventsService } from "./agent-events.service";
 import { GetMessagesQueryDto } from "./dto/get-messages-query.dto";
 import { AgentMessagesService } from "./agent-messages.service";
+import { AgentControlService } from "./agent-control.service";
+import { InjectAgentDto } from "./dto/inject-agent.dto";
+import { PauseAgentDto, ResumeAgentDto } from "./dto/control-agent.dto";
 
 /**
  * Controller for agent management endpoints
@@ -51,7 +55,8 @@ export class AgentsController {
     private readonly lifecycleService: AgentLifecycleService,
     private readonly killswitchService: KillswitchService,
     private readonly eventsService: AgentEventsService,
-    private readonly messagesService: AgentMessagesService
+    private readonly messagesService: AgentMessagesService,
+    private readonly agentControlService: AgentControlService
   ) {}
 
   /**
@@ -374,6 +379,57 @@ export class AgentsController {
     }
   }
 
+  @Post(":agentId/inject")
+  @Throttle({ default: { limit: 10, ttl: 60000 } })
+  @HttpCode(200)
+  @UsePipes(new ValidationPipe({ transform: true, whitelist: true }))
+  async injectAgentMessage(
+    @Param("agentId", ParseUUIDPipe) agentId: string,
+    @Body() dto: InjectAgentDto,
+    @Request() req: { apiKey?: string }
+  ): Promise<{ message: string }> {
+    const operatorId = req.apiKey ?? "operator";
+    await this.agentControlService.injectMessage(agentId, operatorId, dto.message);
+
+    return {
+      message: `Message injected into agent ${agentId}`,
+    };
+  }
+
+  @Post(":agentId/pause")
+  @Throttle({ default: { limit: 10, ttl: 60000 } })
+  @HttpCode(200)
+  @UsePipes(new ValidationPipe({ transform: true, whitelist: true }))
+  async pauseAgent(
+    @Param("agentId", ParseUUIDPipe) agentId: string,
+    @Body() _dto: PauseAgentDto,
+    @Request() req: { apiKey?: string }
+  ): Promise<{ message: string }> {
+    const operatorId = req.apiKey ?? "operator";
+    await this.agentControlService.pauseAgent(agentId, operatorId);
+
+    return {
+      message: `Agent ${agentId} paused`,
+    };
+  }
+
+  @Post(":agentId/resume")
+  @Throttle({ default: { limit: 10, ttl: 60000 } })
+  @HttpCode(200)
+  @UsePipes(new ValidationPipe({ transform: true, whitelist: true }))
+  async resumeAgent(
+    @Param("agentId", ParseUUIDPipe) agentId: string,
+    @Body() _dto: ResumeAgentDto,
+    @Request() req: { apiKey?: string }
+  ): Promise<{ message: string }> {
+    const operatorId = req.apiKey ?? "operator";
+    await this.agentControlService.resumeAgent(agentId, operatorId);
+
+    return {
+      message: `Agent ${agentId} resumed`,
+    };
+  }
+
   /**
    * Kill all active agents
    * @returns Summary of kill operation

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

@@ -8,10 +8,16 @@ import { OrchestratorApiKeyGuard } from "../../common/guards/api-key.guard";
 import { AgentEventsService } from "./agent-events.service";
 import { PrismaModule } from "../../prisma/prisma.module";
 import { AgentMessagesService } from "./agent-messages.service";
+import { AgentControlService } from "./agent-control.service";
 
 @Module({
   imports: [QueueModule, SpawnerModule, KillswitchModule, ValkeyModule, PrismaModule],
   controllers: [AgentsController],
-  providers: [OrchestratorApiKeyGuard, AgentEventsService, AgentMessagesService],
+  providers: [
+    OrchestratorApiKeyGuard,
+    AgentEventsService,
+    AgentMessagesService,
+    AgentControlService,
+  ],
 })
 export class AgentsModule {}

apps/orchestrator/src/api/agents/dto/control-agent.dto.ts

@@ -0,0 +1,3 @@
+export class PauseAgentDto {}
+
+export class ResumeAgentDto {}

apps/orchestrator/src/api/agents/dto/inject-agent.dto.ts

@@ -0,0 +1,7 @@
+import { IsNotEmpty, IsString } from "class-validator";
+
+export class InjectAgentDto {
+  @IsString()
+  @IsNotEmpty()
+  message!: string;
+}

docs/TASKS.md

@@ -122,9 +122,9 @@ Target version: `v0.0.23`
 | id          | status      | milestone     | description                                                                                      | issue | repo         | branch                 | depends_on                                      | blocks                                                      | agent | started_at | completed_at | estimate | used | notes                                         |
 | ----------- | ----------- | ------------- | ------------------------------------------------------------------------------------------------ | ----- | ------------ | ---------------------- | ----------------------------------------------- | ----------------------------------------------------------- | ----- | ---------- | ------------ | -------- | ---- | --------------------------------------------- |
 | MS23-P0-001 | done        | p0-foundation | Prisma schema: AgentConversationMessage, AgentSessionTree, AgentProviderConfig, OperatorAuditLog | #693  | api          | feat/ms23-p0-schema    | —                                               | MS23-P0-002,MS23-P0-003,MS23-P0-004,MS23-P0-005,MS23-P1-001 | codex | 2026-03-06 | 2026-03-06   | 15K      | —    | taskSource field per mosaic-queue note in PRD |
-| MS23-P0-002 | in-progress | p0-foundation | Agent message ingestion: wire spawner/lifecycle to write messages to DB                          | #693  | orchestrator | feat/ms23-p0-ingestion | MS23-P0-001                                     | MS23-P0-006                                                 | codex | 2026-03-06 | —            | 20K      | —    |                                               |
-| MS23-P0-003 | not-started | p0-foundation | Orchestrator API: GET /agents/:id/messages + SSE stream endpoint                                 | #693  | orchestrator | feat/ms23-p0-stream    | MS23-P0-001                                     | MS23-P0-006                                                 | —     | —          | —            | 20K      | —    |                                               |
-| MS23-P0-004 | not-started | p0-foundation | Orchestrator API: POST /agents/:id/inject + pause/resume endpoints                               | #693  | orchestrator | feat/ms23-p0-controls  | MS23-P0-001                                     | MS23-P0-006                                                 | —     | —          | —            | 15K      | —    |                                               |
+| MS23-P0-002 | done        | p0-foundation | Agent message ingestion: wire spawner/lifecycle to write messages to DB                          | #693  | orchestrator | feat/ms23-p0-ingestion | MS23-P0-001                                     | MS23-P0-006                                                 | codex | 2026-03-06 | 2026-03-07      | 20K      | —    |                                               |
+| MS23-P0-003 | done        | p0-foundation | Orchestrator API: GET /agents/:id/messages + SSE stream endpoint                                 | #693  | orchestrator | feat/ms23-p0-stream    | MS23-P0-001                                     | MS23-P0-006                                                 | codex | 2026-03-06 | 2026-03-07      | 20K      | —    |                                               |
+| MS23-P0-004 | in-progress | p0-foundation | Orchestrator API: POST /agents/:id/inject + pause/resume endpoints                               | #693  | orchestrator | feat/ms23-p0-controls  | MS23-P0-001                                     | MS23-P0-006                                                 | codex | 2026-03-07     | —            | 15K      | —    |                                               |
 | MS23-P0-005 | not-started | p0-foundation | Subagent tree: parentAgentId on spawn registration + GET /agents/tree                            | #693  | orchestrator | feat/ms23-p0-tree      | MS23-P0-001                                     | MS23-P0-006                                                 | —     | —          | —            | 15K      | —    |                                               |
 | MS23-P0-006 | not-started | p0-foundation | Unit + integration tests for all P0 orchestrator endpoints                                       | #693  | orchestrator | test/ms23-p0           | MS23-P0-002,MS23-P0-003,MS23-P0-004,MS23-P0-005 | MS23-P1-001                                                 | —     | —          | —            | 20K      | —    | Phase 0 gate: SSE stream verified via curl    |
 

pnpm-lock.yaml

@@ -389,6 +389,9 @@ importers:
       '@vitest/coverage-v8':
         specifier: ^4.0.18
         version: 4.0.18(vitest@4.0.18(@opentelemetry/api@1.9.0)(@types/node@22.19.7)(jiti@2.6.1)(jsdom@26.1.0)(terser@5.46.0)(tsx@4.21.0)(yaml@2.8.2))
+      prisma:
+        specifier: ^6.19.2
+        version: 6.19.2(magicast@0.3.5)(typescript@5.9.3)
       ts-node:
         specifier: ^10.9.2
         version: 10.9.2(@swc/core@1.15.11)(@types/node@22.19.7)(typescript@5.9.3)