Compare commits

..

50 Commits

Author SHA1 Message Date
495d78115e feat(orchestrator): add OpenClaw SSE bridge streaming
All checks were successful
ci/woodpecker/push/ci Pipeline was successful
2026-03-07 16:20:27 -06:00
563d59ad5d Merge pull request 'feat(orchestrator): MS23-P3-001 OpenClawProvider' (#735) from feat/ms23-p3-openclaw-provider into main
All checks were successful
ci/woodpecker/push/ci Pipeline was successful
2026-03-07 22:06:31 +00:00
da6e055113 feat(orchestrator): MS23-P3-001 OpenClawProvider
All checks were successful
ci/woodpecker/push/ci Pipeline was successful
2026-03-07 15:58:56 -06:00
0441d44f42 Merge pull request 'test(web): MS23-P2-009 Mission Control frontend tests' (#733) from test/ms23-p2 into main
All checks were successful
ci/woodpecker/push/ci Pipeline was successful
2026-03-07 21:38:36 +00:00
7147dc3503 test(web): MS23-P2-009 Mission Control frontend tests
All checks were successful
ci/woodpecker/push/ci Pipeline was successful
2026-03-07 15:27:36 -06:00
f0aa3b5a75 test(web): add Mission Control frontend tests for phase 2 gate
All checks were successful
ci/woodpecker/push/ci Pipeline was successful
2026-03-07 15:25:39 -06:00
11d64341b1 Merge pull request 'fix(web): MS23-P2 restore main branch CI' (#732) from fix/ms23-p2-main into main
All checks were successful
ci/woodpecker/push/ci Pipeline was successful
2026-03-07 21:08:54 +00:00
90d2fa7563 fix(web): MS23-P2 fix broken lint/typecheck on main
All checks were successful
ci/woodpecker/push/ci Pipeline was successful
2026-03-07 14:55:04 -06:00
31af6c26ec Merge pull request 'feat(web): MS23-P2-008 Panel grid responsive layout' (#731) from feat/ms23-p2-grid into main
All checks were successful
ci/woodpecker/push/ci Pipeline was successful
2026-03-07 20:48:36 +00:00
e4f942dde7 feat(web): MS23-P2-008 panel grid responsive layout
All checks were successful
ci/woodpecker/push/ci Pipeline was successful
2026-03-07 14:47:39 -06:00
4ea31c5749 Merge pull request 'feat: MS23-P2-007 AuditLogDrawer + audit log endpoint' (#730) from feat/ms23-p2-audit into main
All checks were successful
ci/woodpecker/push/ci Pipeline was successful
2026-03-07 20:40:21 +00:00
4792f7b70a feat: MS23-P2-007 AuditLogDrawer + audit log endpoint
All checks were successful
ci/woodpecker/push/ci Pipeline was successful
2026-03-07 14:40:06 -06:00
571094a099 Merge pull request 'feat(web): MS23-P2-004 Panel operator controls' (#729) from feat/ms23-p2-controls into main
Some checks failed
ci/woodpecker/push/ci Pipeline failed
2026-03-07 20:35:53 +00:00
adef5bdbb2 feat(web): MS23-P2-004 panel operator controls
All checks were successful
ci/woodpecker/push/ci Pipeline was successful
2026-03-07 14:34:31 -06:00
eb771d795a Merge pull request 'feat(web): MS23-P2-003 BargeInInput component' (#728) from feat/ms23-p2-barge into main
All checks were successful
ci/woodpecker/push/ci Pipeline was successful
2026-03-07 20:33:39 +00:00
c9aff531ea Merge pull request 'feat(web): MS23-P2-006 KillAllDialog confirmation modal' (#727) from feat/ms23-p2-killall into main
Some checks failed
ci/woodpecker/push/ci Pipeline failed
2026-03-07 20:33:03 +00:00
b2c751caca feat(web): MS23-P2-003 BargeInInput component
Some checks failed
ci/woodpecker/push/ci Pipeline failed
2026-03-07 14:32:12 -06:00
cd28428cf2 feat(web): MS23-P2-006 KillAllDialog confirmation modal
Some checks failed
ci/woodpecker/push/ci Pipeline failed
2026-03-07 14:31:38 -06:00
2c36569f85 Merge pull request 'feat(web): MS23-P2-005 GlobalAgentRoster sidebar tree' (#726) from feat/ms23-p2-roster into main
Some checks failed
ci/woodpecker/push/ci Pipeline failed
2026-03-07 20:20:17 +00:00
7c086db7e4 feat(web): MS23-P2-005 GlobalAgentRoster sidebar tree
Some checks failed
ci/woodpecker/push/ci Pipeline failed
2026-03-07 14:19:05 -06:00
577e6141e0 Merge pull request 'feat(web): MS23-P2-002 OrchestratorPanel SSE stream component' (#725) from feat/ms23-p2-panel into main
All checks were successful
ci/woodpecker/push/ci Pipeline was successful
2026-03-07 20:17:56 +00:00
631ba499e3 feat(web): MS23-P2-002 OrchestratorPanel SSE stream component
All checks were successful
ci/woodpecker/push/ci Pipeline was successful
2026-03-07 14:14:57 -06:00
a61106c24a Merge pull request 'feat(web): MS23-P2-001 Mission Control page route + layout shell' (#724) from feat/ms23-p2-page into main
All checks were successful
ci/woodpecker/push/ci Pipeline was successful
2026-03-07 20:00:55 +00:00
487aac6903 feat(web): add Mission Control page layout shell
All checks were successful
ci/woodpecker/push/ci Pipeline was successful
2026-03-07 13:59:23 -06:00
544e828e58 Merge pull request 'test(orchestrator): MS23-P1-006 Phase 1 provider system unit tests' (#723) from test/ms23-p1 into main
All checks were successful
ci/woodpecker/push/ci Pipeline was successful
2026-03-07 19:51:09 +00:00
9489bc63f8 test(orchestrator): add mission-control phase 1 gate coverage
All checks were successful
ci/woodpecker/push/ci Pipeline was successful
2026-03-07 13:47:49 -06:00
ad644799aa feat(orchestrator): MS23-P1-005 Mission Control proxy API (#722)
All checks were successful
ci/woodpecker/push/ci Pipeline was successful
Co-authored-by: Jason Woltje <jason@diversecanvas.com>
Co-committed-by: Jason Woltje <jason@diversecanvas.com>
2026-03-07 19:43:34 +00:00
81bf349270 Merge pull request 'feat(orchestrator): MS23-P1-004 AgentProviderConfig CRUD API' (#721) from feat/ms23-p1-provider-api into main
All checks were successful
ci/woodpecker/push/ci Pipeline was successful
2026-03-07 19:26:24 +00:00
bcada71e88 feat(orchestrator): add AgentProviderConfig CRUD API
All checks were successful
ci/woodpecker/push/ci Pipeline was successful
2026-03-07 13:22:56 -06:00
9cc82e7fcf feat(orchestrator): MS23-P1-003 AgentProviderRegistry (#720)
All checks were successful
ci/woodpecker/push/ci Pipeline was successful
Co-authored-by: Jason Woltje <jason@diversecanvas.com>
Co-committed-by: Jason Woltje <jason@diversecanvas.com>
2026-03-07 19:15:59 +00:00
4b135ae1f0 feat(orchestrator): MS23-P1-002 InternalAgentProvider (#719)
All checks were successful
ci/woodpecker/push/ci Pipeline was successful
Co-authored-by: Jason Woltje <jason@diversecanvas.com>
Co-committed-by: Jason Woltje <jason@diversecanvas.com>
2026-03-07 19:06:36 +00:00
364619b332 feat(shared): MS23-P1-001 IAgentProvider interface + agent types (#718)
All checks were successful
ci/woodpecker/push/ci Pipeline was successful
Co-authored-by: Jason Woltje <jason@diversecanvas.com>
Co-committed-by: Jason Woltje <jason@diversecanvas.com>
2026-03-07 18:29:18 +00:00
18ed3a5411 chore(tasks): MS23 Phase 0 complete — all 6 P0 tasks done (#717)
Co-authored-by: Jason Woltje <jason@diversecanvas.com>
Co-committed-by: Jason Woltje <jason@diversecanvas.com>
2026-03-07 18:13:12 +00:00
79ff3a921f test(orchestrator): MS23-P0-006 service unit tests for AgentMessages/Control/Tree (#716)
All checks were successful
ci/woodpecker/push/ci Pipeline was successful
Co-authored-by: Jason Woltje <jason@diversecanvas.com>
Co-committed-by: Jason Woltje <jason@diversecanvas.com>
2026-03-07 18:12:47 +00:00
76f06d0291 chore(tasks): MS23-P0-005 done, P0-006 in-progress (#715)
Co-authored-by: Jason Woltje <jason@diversecanvas.com>
Co-committed-by: Jason Woltje <jason@diversecanvas.com>
2026-03-07 17:58:20 +00:00
03dd25f028 feat(orchestrator): MS23-P0-005 subagent tree endpoint (#714)
All checks were successful
ci/woodpecker/push/ci Pipeline was successful
Co-authored-by: Jason Woltje <jason@diversecanvas.com>
Co-committed-by: Jason Woltje <jason@diversecanvas.com>
2026-03-07 17:57:55 +00:00
f3726de54e chore(tasks): MS23-P0-004 done, P0-005 in-progress (#713)
Co-authored-by: Jason Woltje <jason@diversecanvas.com>
Co-committed-by: Jason Woltje <jason@diversecanvas.com>
2026-03-07 17:43:35 +00:00
d0c6622de5 feat(orchestrator): MS23-P0-004 operator inject/pause/resume endpoints (#712)
All checks were successful
ci/woodpecker/push/ci Pipeline was successful
Co-authored-by: Jason Woltje <jason@diversecanvas.com>
Co-committed-by: Jason Woltje <jason@diversecanvas.com>
2026-03-07 17:43:11 +00:00
4749f52668 chore(tasks): MS23-P0-002 done, P0-003 done, P0-004 in-progress (#711)
Co-authored-by: Jason Woltje <jason@diversecanvas.com>
Co-committed-by: Jason Woltje <jason@diversecanvas.com>
2026-03-07 17:15:57 +00:00
e0b28c91c3 feat(orchestrator): MS23 per-agent message history and SSE stream (#702)
All checks were successful
ci/woodpecker/push/ci Pipeline was successful
Co-authored-by: Jason Woltje <jason@diversecanvas.com>
Co-committed-by: Jason Woltje <jason@diversecanvas.com>
2026-03-07 17:15:30 +00:00
fa7837af3e fix(orchestrator): rm symlink before schema COPY in Docker builder (#710)
All checks were successful
ci/woodpecker/push/ci Pipeline was successful
Co-authored-by: Jason Woltje <jason@diversecanvas.com>
Co-committed-by: Jason Woltje <jason@diversecanvas.com>
2026-03-07 17:00:38 +00:00
123cbce5cd fix(orchestrator): copy schema to overwrite dangling symlink in Docker (#709)
Some checks failed
ci/woodpecker/push/ci Pipeline failed
Co-authored-by: Jason Woltje <jason@diversecanvas.com>
Co-committed-by: Jason Woltje <jason@diversecanvas.com>
2026-03-07 16:45:54 +00:00
7d47e5ff99 fix(orchestrator): use symlink path ./prisma/schema.prisma in generate script (#708)
Some checks failed
ci/woodpecker/push/ci Pipeline failed
Co-authored-by: Jason Woltje <jason@diversecanvas.com>
Co-committed-by: Jason Woltje <jason@diversecanvas.com>
2026-03-07 16:31:49 +00:00
ef674206e7 fix(orchestrator): symlink prisma/schema.prisma to resolve Docker build root detection (#707)
Some checks failed
ci/woodpecker/push/ci Pipeline failed
Co-authored-by: Jason Woltje <jason@diversecanvas.com>
Co-committed-by: Jason Woltje <jason@diversecanvas.com>
2026-03-07 16:17:20 +00:00
977747599f fix(orchestrator): local prisma schema copy for Docker generate (#706)
Some checks failed
ci/woodpecker/push/ci Pipeline failed
Co-authored-by: Jason Woltje <jason@diversecanvas.com>
Co-committed-by: Jason Woltje <jason@diversecanvas.com>
2026-03-07 16:00:07 +00:00
fc4699ca51 fix(orchestrator): copy apps/api/package.json for prisma generate in Dockerfile (#705)
Some checks failed
ci/woodpecker/push/ci Pipeline failed
Co-authored-by: Jason Woltje <jason@diversecanvas.com>
Co-committed-by: Jason Woltje <jason@diversecanvas.com>
2026-03-07 15:38:46 +00:00
b61554800b fix(orchestrator): add prisma CLI devDependency for prisma:generate (#704)
Some checks failed
ci/woodpecker/push/ci Pipeline failed
Co-authored-by: Jason Woltje <jason@diversecanvas.com>
Co-committed-by: Jason Woltje <jason@diversecanvas.com>
2026-03-07 06:56:42 +00:00
98e892f23c fix(orchestrator): Dockerfile prisma generate + vitest reflect-metadata setup (#703)
Some checks failed
ci/woodpecker/push/ci Pipeline failed
Co-authored-by: Jason Woltje <jason@diversecanvas.com>
Co-committed-by: Jason Woltje <jason@diversecanvas.com>
2026-03-07 04:45:17 +00:00
de6faf659e feat(orchestrator): MS23 agent lifecycle ingestion service (#701)
Some checks failed
ci/woodpecker/push/ci Pipeline failed
Co-authored-by: Jason Woltje <jason@diversecanvas.com>
Co-committed-by: Jason Woltje <jason@diversecanvas.com>
2026-03-07 04:21:26 +00:00
49fa958444 chore(orchestrator): MS23 P0-001 done, P0-002 in-progress (#700)
Co-authored-by: Jason Woltje <jason@diversecanvas.com>
Co-committed-by: Jason Woltje <jason@diversecanvas.com>
2026-03-07 02:10:59 +00:00
86 changed files with 8742 additions and 39 deletions

View File

@@ -21,6 +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/
# API schema is available via apps/orchestrator/prisma/schema.prisma symlink
# Copy npm configuration for native binary architecture hints
COPY .npmrc ./
@@ -46,6 +47,15 @@ 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
# 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

View File

@@ -3,30 +3,33 @@
"version": "0.0.20",
"private": true,
"scripts": {
"dev": "nest start --watch",
"build": "nest build",
"dev": "nest start --watch",
"lint": "eslint src/",
"lint:fix": "eslint src/ --fix",
"prisma:generate": "prisma generate --schema=./prisma/schema.prisma",
"start": "node dist/main.js",
"start:dev": "nest start --watch",
"start:debug": "nest start --debug --watch",
"start:dev": "nest start --watch",
"start:prod": "node dist/main.js",
"test": "vitest",
"test:watch": "vitest watch",
"test:e2e": "vitest run --config tests/integration/vitest.config.ts",
"test:perf": "vitest run --config tests/performance/vitest.config.ts",
"typecheck": "tsc --noEmit",
"lint": "eslint src/",
"lint:fix": "eslint src/ --fix"
"test:watch": "vitest watch",
"typecheck": "tsc --noEmit"
},
"dependencies": {
"@anthropic-ai/sdk": "^0.72.1",
"@mosaic/config": "workspace:*",
"@mosaic/shared": "workspace:*",
"@nestjs/axios": "^4.0.1",
"@nestjs/bullmq": "^11.0.4",
"@nestjs/common": "^11.1.12",
"@nestjs/config": "^4.0.2",
"@nestjs/core": "^11.1.12",
"@nestjs/platform-express": "^11.1.12",
"@nestjs/throttler": "^6.5.0",
"@prisma/client": "^6.19.2",
"bullmq": "^5.67.2",
"class-transformer": "^0.5.1",
"class-validator": "^0.14.1",
@@ -45,6 +48,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",

View File

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

View File

@@ -0,0 +1,10 @@
import { Module } from "@nestjs/common";
import { PrismaModule } from "../prisma/prisma.module";
import { AgentIngestionService } from "./agent-ingestion.service";
@Module({
imports: [PrismaModule],
providers: [AgentIngestionService],
exports: [AgentIngestionService],
})
export class AgentIngestionModule {}

View File

@@ -0,0 +1,141 @@
import { Injectable, Logger } from "@nestjs/common";
import type { Prisma } from "@prisma/client";
import { PrismaService } from "../prisma/prisma.service";
export type AgentConversationRole = "agent" | "user" | "system" | "operator";
@Injectable()
export class AgentIngestionService {
private readonly logger = new Logger(AgentIngestionService.name);
constructor(private readonly prisma: PrismaService) {}
private toJsonValue(value: Record<string, unknown>): Prisma.InputJsonValue {
return value as Prisma.InputJsonValue;
}
async recordAgentSpawned(
agentId: string,
parentAgentId?: string,
missionId?: string,
taskId?: string,
agentType?: string
): Promise<void> {
await this.prisma.agentSessionTree.upsert({
where: { sessionId: agentId },
create: {
sessionId: agentId,
parentSessionId: parentAgentId ?? null,
missionId,
taskId,
agentType,
status: "spawning",
},
update: {
parentSessionId: parentAgentId ?? null,
missionId,
taskId,
agentType,
status: "spawning",
completedAt: null,
},
});
this.logger.debug(`Recorded spawned state for agent ${agentId}`);
}
async recordAgentStarted(agentId: string): Promise<void> {
await this.prisma.agentSessionTree.upsert({
where: { sessionId: agentId },
create: {
sessionId: agentId,
status: "running",
},
update: {
status: "running",
},
});
this.logger.debug(`Recorded running state for agent ${agentId}`);
}
async recordAgentCompleted(agentId: string): Promise<void> {
const completedAt = new Date();
await this.prisma.agentSessionTree.upsert({
where: { sessionId: agentId },
create: {
sessionId: agentId,
status: "completed",
completedAt,
},
update: {
status: "completed",
completedAt,
},
});
this.logger.debug(`Recorded completed state for agent ${agentId}`);
}
async recordAgentFailed(agentId: string, error?: string): Promise<void> {
const completedAt = new Date();
const metadata = error ? this.toJsonValue({ error }) : undefined;
await this.prisma.agentSessionTree.upsert({
where: { sessionId: agentId },
create: {
sessionId: agentId,
status: "failed",
completedAt,
...(metadata && { metadata }),
},
update: {
status: "failed",
completedAt,
...(metadata && { metadata }),
},
});
this.logger.debug(`Recorded failed state for agent ${agentId}`);
}
async recordAgentKilled(agentId: string): Promise<void> {
const completedAt = new Date();
await this.prisma.agentSessionTree.upsert({
where: { sessionId: agentId },
create: {
sessionId: agentId,
status: "killed",
completedAt,
},
update: {
status: "killed",
completedAt,
},
});
this.logger.debug(`Recorded killed state for agent ${agentId}`);
}
async recordMessage(
sessionId: string,
role: AgentConversationRole,
content: string,
provider = "internal",
metadata?: Record<string, unknown>
): Promise<void> {
await this.prisma.agentConversationMessage.create({
data: {
sessionId,
role,
content,
provider,
...(metadata && { metadata: this.toJsonValue(metadata) }),
},
});
this.logger.debug(`Recorded message for session ${sessionId}`);
}
}

View File

@@ -0,0 +1,54 @@
import {
Body,
Controller,
Delete,
Get,
Param,
Patch,
Post,
UseGuards,
UsePipes,
ValidationPipe,
} from "@nestjs/common";
import type { AgentProviderConfig } from "@prisma/client";
import { OrchestratorApiKeyGuard } from "../../common/guards/api-key.guard";
import { OrchestratorThrottlerGuard } from "../../common/guards/throttler.guard";
import { AgentProvidersService } from "./agent-providers.service";
import { CreateAgentProviderDto } from "./dto/create-agent-provider.dto";
import { UpdateAgentProviderDto } from "./dto/update-agent-provider.dto";
@Controller("agent-providers")
@UseGuards(OrchestratorApiKeyGuard, OrchestratorThrottlerGuard)
export class AgentProvidersController {
constructor(private readonly agentProvidersService: AgentProvidersService) {}
@Get()
async list(): Promise<AgentProviderConfig[]> {
return this.agentProvidersService.list();
}
@Get(":id")
async getById(@Param("id") id: string): Promise<AgentProviderConfig> {
return this.agentProvidersService.getById(id);
}
@Post()
@UsePipes(new ValidationPipe({ transform: true, whitelist: true }))
async create(@Body() dto: CreateAgentProviderDto): Promise<AgentProviderConfig> {
return this.agentProvidersService.create(dto);
}
@Patch(":id")
@UsePipes(new ValidationPipe({ transform: true, whitelist: true }))
async update(
@Param("id") id: string,
@Body() dto: UpdateAgentProviderDto
): Promise<AgentProviderConfig> {
return this.agentProvidersService.update(id, dto);
}
@Delete(":id")
async delete(@Param("id") id: string): Promise<AgentProviderConfig> {
return this.agentProvidersService.delete(id);
}
}

View File

@@ -0,0 +1,12 @@
import { Module } from "@nestjs/common";
import { PrismaModule } from "../../prisma/prisma.module";
import { OrchestratorApiKeyGuard } from "../../common/guards/api-key.guard";
import { AgentProvidersController } from "./agent-providers.controller";
import { AgentProvidersService } from "./agent-providers.service";
@Module({
imports: [PrismaModule],
controllers: [AgentProvidersController],
providers: [OrchestratorApiKeyGuard, AgentProvidersService],
})
export class AgentProvidersModule {}

View File

@@ -0,0 +1,211 @@
import { beforeEach, describe, expect, it, vi } from "vitest";
import { NotFoundException } from "@nestjs/common";
import { AgentProvidersService } from "./agent-providers.service";
import { PrismaService } from "../../prisma/prisma.service";
describe("AgentProvidersService", () => {
let service: AgentProvidersService;
let prisma: {
agentProviderConfig: {
findMany: ReturnType<typeof vi.fn>;
findUnique: ReturnType<typeof vi.fn>;
create: ReturnType<typeof vi.fn>;
update: ReturnType<typeof vi.fn>;
delete: ReturnType<typeof vi.fn>;
};
};
beforeEach(() => {
prisma = {
agentProviderConfig: {
findMany: vi.fn(),
findUnique: vi.fn(),
create: vi.fn(),
update: vi.fn(),
delete: vi.fn(),
},
};
service = new AgentProvidersService(prisma as unknown as PrismaService);
});
it("lists all provider configs", async () => {
const expected = [
{
id: "cfg-1",
workspaceId: "8bcd7eda-a122-4d6c-adfd-b152f6f75369",
name: "Primary",
provider: "openai",
gatewayUrl: "https://gateway.example.com",
credentials: {},
isActive: true,
createdAt: new Date("2026-03-07T18:00:00.000Z"),
updatedAt: new Date("2026-03-07T18:00:00.000Z"),
},
];
prisma.agentProviderConfig.findMany.mockResolvedValue(expected);
const result = await service.list();
expect(prisma.agentProviderConfig.findMany).toHaveBeenCalledWith({
orderBy: [{ createdAt: "desc" }, { id: "desc" }],
});
expect(result).toEqual(expected);
});
it("returns a single provider config", async () => {
const expected = {
id: "cfg-1",
workspaceId: "8bcd7eda-a122-4d6c-adfd-b152f6f75369",
name: "Primary",
provider: "openai",
gatewayUrl: "https://gateway.example.com",
credentials: { apiKeyRef: "vault:openai" },
isActive: true,
createdAt: new Date("2026-03-07T18:00:00.000Z"),
updatedAt: new Date("2026-03-07T18:00:00.000Z"),
};
prisma.agentProviderConfig.findUnique.mockResolvedValue(expected);
const result = await service.getById("cfg-1");
expect(prisma.agentProviderConfig.findUnique).toHaveBeenCalledWith({
where: { id: "cfg-1" },
});
expect(result).toEqual(expected);
});
it("throws NotFoundException when provider config is missing", async () => {
prisma.agentProviderConfig.findUnique.mockResolvedValue(null);
await expect(service.getById("missing")).rejects.toBeInstanceOf(NotFoundException);
});
it("creates a provider config with default credentials", async () => {
const created = {
id: "cfg-created",
workspaceId: "8bcd7eda-a122-4d6c-adfd-b152f6f75369",
name: "New Provider",
provider: "claude",
gatewayUrl: "https://gateway.example.com",
credentials: {},
isActive: true,
createdAt: new Date("2026-03-07T18:00:00.000Z"),
updatedAt: new Date("2026-03-07T18:00:00.000Z"),
};
prisma.agentProviderConfig.create.mockResolvedValue(created);
const result = await service.create({
workspaceId: "8bcd7eda-a122-4d6c-adfd-b152f6f75369",
name: "New Provider",
provider: "claude",
gatewayUrl: "https://gateway.example.com",
});
expect(prisma.agentProviderConfig.create).toHaveBeenCalledWith({
data: {
workspaceId: "8bcd7eda-a122-4d6c-adfd-b152f6f75369",
name: "New Provider",
provider: "claude",
gatewayUrl: "https://gateway.example.com",
credentials: {},
},
});
expect(result).toEqual(created);
});
it("updates a provider config", async () => {
prisma.agentProviderConfig.findUnique.mockResolvedValue({
id: "cfg-1",
workspaceId: "8bcd7eda-a122-4d6c-adfd-b152f6f75369",
name: "Primary",
provider: "openai",
gatewayUrl: "https://gateway.example.com",
credentials: {},
isActive: true,
createdAt: new Date("2026-03-07T18:00:00.000Z"),
updatedAt: new Date("2026-03-07T18:00:00.000Z"),
});
const updated = {
id: "cfg-1",
workspaceId: "8bcd7eda-a122-4d6c-adfd-b152f6f75369",
name: "Secondary",
provider: "openai",
gatewayUrl: "https://gateway2.example.com",
credentials: { apiKeyRef: "vault:new" },
isActive: false,
createdAt: new Date("2026-03-07T18:00:00.000Z"),
updatedAt: new Date("2026-03-07T19:00:00.000Z"),
};
prisma.agentProviderConfig.update.mockResolvedValue(updated);
const result = await service.update("cfg-1", {
name: "Secondary",
gatewayUrl: "https://gateway2.example.com",
credentials: { apiKeyRef: "vault:new" },
isActive: false,
});
expect(prisma.agentProviderConfig.update).toHaveBeenCalledWith({
where: { id: "cfg-1" },
data: {
name: "Secondary",
gatewayUrl: "https://gateway2.example.com",
credentials: { apiKeyRef: "vault:new" },
isActive: false,
},
});
expect(result).toEqual(updated);
});
it("throws NotFoundException when updating a missing provider config", async () => {
prisma.agentProviderConfig.findUnique.mockResolvedValue(null);
await expect(service.update("missing", { name: "Updated" })).rejects.toBeInstanceOf(
NotFoundException
);
expect(prisma.agentProviderConfig.update).not.toHaveBeenCalled();
});
it("deletes a provider config", async () => {
prisma.agentProviderConfig.findUnique.mockResolvedValue({
id: "cfg-1",
workspaceId: "8bcd7eda-a122-4d6c-adfd-b152f6f75369",
name: "Primary",
provider: "openai",
gatewayUrl: "https://gateway.example.com",
credentials: {},
isActive: true,
createdAt: new Date("2026-03-07T18:00:00.000Z"),
updatedAt: new Date("2026-03-07T18:00:00.000Z"),
});
const deleted = {
id: "cfg-1",
workspaceId: "8bcd7eda-a122-4d6c-adfd-b152f6f75369",
name: "Primary",
provider: "openai",
gatewayUrl: "https://gateway.example.com",
credentials: {},
isActive: true,
createdAt: new Date("2026-03-07T18:00:00.000Z"),
updatedAt: new Date("2026-03-07T18:00:00.000Z"),
};
prisma.agentProviderConfig.delete.mockResolvedValue(deleted);
const result = await service.delete("cfg-1");
expect(prisma.agentProviderConfig.delete).toHaveBeenCalledWith({
where: { id: "cfg-1" },
});
expect(result).toEqual(deleted);
});
it("throws NotFoundException when deleting a missing provider config", async () => {
prisma.agentProviderConfig.findUnique.mockResolvedValue(null);
await expect(service.delete("missing")).rejects.toBeInstanceOf(NotFoundException);
expect(prisma.agentProviderConfig.delete).not.toHaveBeenCalled();
});
});

View File

@@ -0,0 +1,71 @@
import { Injectable, NotFoundException } from "@nestjs/common";
import type { AgentProviderConfig, Prisma } from "@prisma/client";
import { PrismaService } from "../../prisma/prisma.service";
import { CreateAgentProviderDto } from "./dto/create-agent-provider.dto";
import { UpdateAgentProviderDto } from "./dto/update-agent-provider.dto";
@Injectable()
export class AgentProvidersService {
constructor(private readonly prisma: PrismaService) {}
async list(): Promise<AgentProviderConfig[]> {
return this.prisma.agentProviderConfig.findMany({
orderBy: [{ createdAt: "desc" }, { id: "desc" }],
});
}
async getById(id: string): Promise<AgentProviderConfig> {
const providerConfig = await this.prisma.agentProviderConfig.findUnique({
where: { id },
});
if (!providerConfig) {
throw new NotFoundException(`Agent provider config with id ${id} not found`);
}
return providerConfig;
}
async create(dto: CreateAgentProviderDto): Promise<AgentProviderConfig> {
return this.prisma.agentProviderConfig.create({
data: {
workspaceId: dto.workspaceId,
name: dto.name,
provider: dto.provider,
gatewayUrl: dto.gatewayUrl,
credentials: this.toJsonValue(dto.credentials ?? {}),
...(dto.isActive !== undefined ? { isActive: dto.isActive } : {}),
},
});
}
async update(id: string, dto: UpdateAgentProviderDto): Promise<AgentProviderConfig> {
await this.getById(id);
const data: Prisma.AgentProviderConfigUpdateInput = {
...(dto.workspaceId !== undefined ? { workspaceId: dto.workspaceId } : {}),
...(dto.name !== undefined ? { name: dto.name } : {}),
...(dto.provider !== undefined ? { provider: dto.provider } : {}),
...(dto.gatewayUrl !== undefined ? { gatewayUrl: dto.gatewayUrl } : {}),
...(dto.isActive !== undefined ? { isActive: dto.isActive } : {}),
...(dto.credentials !== undefined ? { credentials: this.toJsonValue(dto.credentials) } : {}),
};
return this.prisma.agentProviderConfig.update({
where: { id },
data,
});
}
async delete(id: string): Promise<AgentProviderConfig> {
await this.getById(id);
return this.prisma.agentProviderConfig.delete({
where: { id },
});
}
private toJsonValue(value: Record<string, unknown>): Prisma.InputJsonValue {
return value as Prisma.InputJsonValue;
}
}

View File

@@ -0,0 +1,26 @@
import { IsBoolean, IsNotEmpty, IsObject, IsOptional, IsString, IsUUID } from "class-validator";
export class CreateAgentProviderDto {
@IsUUID()
workspaceId!: string;
@IsString()
@IsNotEmpty()
name!: string;
@IsString()
@IsNotEmpty()
provider!: string;
@IsString()
@IsNotEmpty()
gatewayUrl!: string;
@IsOptional()
@IsObject()
credentials?: Record<string, unknown>;
@IsOptional()
@IsBoolean()
isActive?: boolean;
}

View File

@@ -0,0 +1,30 @@
import { IsBoolean, IsNotEmpty, IsObject, IsOptional, IsString, IsUUID } from "class-validator";
export class UpdateAgentProviderDto {
@IsOptional()
@IsUUID()
workspaceId?: string;
@IsOptional()
@IsString()
@IsNotEmpty()
name?: string;
@IsOptional()
@IsString()
@IsNotEmpty()
provider?: string;
@IsOptional()
@IsString()
@IsNotEmpty()
gatewayUrl?: string;
@IsOptional()
@IsObject()
credentials?: Record<string, unknown>;
@IsOptional()
@IsBoolean()
isActive?: boolean;
}

View File

@@ -0,0 +1,172 @@
import { describe, it, expect, beforeEach, afterEach, vi } from "vitest";
import { AgentControlService } from "./agent-control.service";
import { PrismaService } from "../../prisma/prisma.service";
import { KillswitchService } from "../../killswitch/killswitch.service";
describe("AgentControlService", () => {
let service: AgentControlService;
let prisma: {
agentSessionTree: {
findUnique: ReturnType<typeof vi.fn>;
updateMany: ReturnType<typeof vi.fn>;
};
agentConversationMessage: {
create: ReturnType<typeof vi.fn>;
};
operatorAuditLog: {
create: ReturnType<typeof vi.fn>;
};
};
let killswitchService: {
killAgent: ReturnType<typeof vi.fn>;
};
beforeEach(() => {
prisma = {
agentSessionTree: {
findUnique: vi.fn(),
updateMany: vi.fn().mockResolvedValue({ count: 1 }),
},
agentConversationMessage: {
create: vi.fn().mockResolvedValue(undefined),
},
operatorAuditLog: {
create: vi.fn().mockResolvedValue(undefined),
},
};
killswitchService = {
killAgent: vi.fn().mockResolvedValue(undefined),
};
service = new AgentControlService(
prisma as unknown as PrismaService,
killswitchService as unknown as KillswitchService
);
});
afterEach(() => {
vi.clearAllMocks();
});
describe("injectMessage", () => {
it("creates conversation message and audit log when tree entry exists", async () => {
prisma.agentSessionTree.findUnique.mockResolvedValue({ id: "tree-1" });
await service.injectMessage("agent-123", "operator-abc", "Please continue");
expect(prisma.agentSessionTree.findUnique).toHaveBeenCalledWith({
where: { sessionId: "agent-123" },
select: { id: true },
});
expect(prisma.agentConversationMessage.create).toHaveBeenCalledWith({
data: {
sessionId: "agent-123",
role: "operator",
content: "Please continue",
provider: "internal",
metadata: {},
},
});
expect(prisma.operatorAuditLog.create).toHaveBeenCalledWith({
data: {
sessionId: "agent-123",
userId: "operator-abc",
provider: "internal",
action: "inject",
metadata: {
payload: {
message: "Please continue",
},
},
},
});
});
it("creates only audit log when no tree entry exists", async () => {
prisma.agentSessionTree.findUnique.mockResolvedValue(null);
await service.injectMessage("agent-456", "operator-def", "Nudge message");
expect(prisma.agentConversationMessage.create).not.toHaveBeenCalled();
expect(prisma.operatorAuditLog.create).toHaveBeenCalledWith({
data: {
sessionId: "agent-456",
userId: "operator-def",
provider: "internal",
action: "inject",
metadata: {
payload: {
message: "Nudge message",
},
},
},
});
});
});
describe("pauseAgent", () => {
it("updates tree status to paused and creates audit log", async () => {
await service.pauseAgent("agent-789", "operator-pause");
expect(prisma.agentSessionTree.updateMany).toHaveBeenCalledWith({
where: { sessionId: "agent-789" },
data: { status: "paused" },
});
expect(prisma.operatorAuditLog.create).toHaveBeenCalledWith({
data: {
sessionId: "agent-789",
userId: "operator-pause",
provider: "internal",
action: "pause",
metadata: {
payload: {},
},
},
});
});
});
describe("resumeAgent", () => {
it("updates tree status to running and creates audit log", async () => {
await service.resumeAgent("agent-321", "operator-resume");
expect(prisma.agentSessionTree.updateMany).toHaveBeenCalledWith({
where: { sessionId: "agent-321" },
data: { status: "running" },
});
expect(prisma.operatorAuditLog.create).toHaveBeenCalledWith({
data: {
sessionId: "agent-321",
userId: "operator-resume",
provider: "internal",
action: "resume",
metadata: {
payload: {},
},
},
});
});
});
describe("killAgent", () => {
it("delegates kill to killswitch and logs audit", async () => {
await service.killAgent("agent-654", "operator-kill", false);
expect(killswitchService.killAgent).toHaveBeenCalledWith("agent-654");
expect(prisma.operatorAuditLog.create).toHaveBeenCalledWith({
data: {
sessionId: "agent-654",
userId: "operator-kill",
provider: "internal",
action: "kill",
metadata: {
payload: {
force: false,
},
},
},
});
});
});
});

View File

@@ -0,0 +1,77 @@
import { Injectable } from "@nestjs/common";
import type { Prisma } from "@prisma/client";
import { KillswitchService } from "../../killswitch/killswitch.service";
import { PrismaService } from "../../prisma/prisma.service";
@Injectable()
export class AgentControlService {
constructor(
private readonly prisma: PrismaService,
private readonly killswitchService: KillswitchService
) {}
private toJsonValue(value: Record<string, unknown>): Prisma.InputJsonValue {
return value as Prisma.InputJsonValue;
}
private async createOperatorAuditLog(
agentId: string,
operatorId: string,
action: "inject" | "pause" | "resume" | "kill",
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", {});
}
async killAgent(agentId: string, operatorId: string, force = true): Promise<void> {
await this.killswitchService.killAgent(agentId);
await this.createOperatorAuditLog(agentId, operatorId, "kill", { force });
}
}

View File

@@ -0,0 +1,103 @@
import { describe, it, expect, beforeEach, afterEach, vi } from "vitest";
import { AgentMessagesService } from "./agent-messages.service";
import { PrismaService } from "../../prisma/prisma.service";
describe("AgentMessagesService", () => {
let service: AgentMessagesService;
let prisma: {
agentConversationMessage: {
findMany: ReturnType<typeof vi.fn>;
count: ReturnType<typeof vi.fn>;
};
};
beforeEach(() => {
prisma = {
agentConversationMessage: {
findMany: vi.fn(),
count: vi.fn(),
},
};
service = new AgentMessagesService(prisma as unknown as PrismaService);
});
afterEach(() => {
vi.clearAllMocks();
});
describe("getMessages", () => {
it("returns paginated messages from Prisma", async () => {
const sessionId = "agent-123";
const messages = [
{
id: "msg-1",
sessionId,
provider: "internal",
role: "assistant",
content: "First message",
timestamp: new Date("2026-03-07T16:00:00.000Z"),
metadata: {},
},
{
id: "msg-2",
sessionId,
provider: "internal",
role: "user",
content: "Second message",
timestamp: new Date("2026-03-07T15:59:00.000Z"),
metadata: {},
},
];
prisma.agentConversationMessage.findMany.mockResolvedValue(messages);
prisma.agentConversationMessage.count.mockResolvedValue(2);
const result = await service.getMessages(sessionId, 50, 0);
expect(prisma.agentConversationMessage.findMany).toHaveBeenCalledWith({
where: { sessionId },
orderBy: { timestamp: "desc" },
take: 50,
skip: 0,
});
expect(prisma.agentConversationMessage.count).toHaveBeenCalledWith({ where: { sessionId } });
expect(result).toEqual({
messages,
total: 2,
});
});
it("applies limit and cursor (skip) correctly", async () => {
const sessionId = "agent-456";
const limit = 10;
const cursor = 20;
prisma.agentConversationMessage.findMany.mockResolvedValue([]);
prisma.agentConversationMessage.count.mockResolvedValue(42);
await service.getMessages(sessionId, limit, cursor);
expect(prisma.agentConversationMessage.findMany).toHaveBeenCalledWith({
where: { sessionId },
orderBy: { timestamp: "desc" },
take: limit,
skip: cursor,
});
});
it("returns empty messages array when no messages exist", async () => {
const sessionId = "agent-empty";
prisma.agentConversationMessage.findMany.mockResolvedValue([]);
prisma.agentConversationMessage.count.mockResolvedValue(0);
const result = await service.getMessages(sessionId, 25, 0);
expect(result).toEqual({
messages: [],
total: 0,
});
});
});
});

View File

@@ -0,0 +1,84 @@
import { Injectable } from "@nestjs/common";
import { type AgentConversationMessage, type Prisma } from "@prisma/client";
import { PrismaService } from "../../prisma/prisma.service";
@Injectable()
export class AgentMessagesService {
constructor(private readonly prisma: PrismaService) {}
async getMessages(
sessionId: string,
limit: number,
skip: number
): Promise<{
messages: AgentConversationMessage[];
total: number;
}> {
const where = { sessionId };
const [messages, total] = await Promise.all([
this.prisma.agentConversationMessage.findMany({
where,
orderBy: {
timestamp: "desc",
},
take: limit,
skip,
}),
this.prisma.agentConversationMessage.count({ where }),
]);
return {
messages,
total,
};
}
async getReplayMessages(sessionId: string, limit = 50): Promise<AgentConversationMessage[]> {
const messages = await this.prisma.agentConversationMessage.findMany({
where: { sessionId },
orderBy: {
timestamp: "desc",
},
take: limit,
});
return messages.reverse();
}
async getMessagesAfter(
sessionId: string,
lastSeenTimestamp: Date,
lastSeenMessageId: string | null
): Promise<AgentConversationMessage[]> {
const where: Prisma.AgentConversationMessageWhereInput = {
sessionId,
...(lastSeenMessageId
? {
OR: [
{
timestamp: {
gt: lastSeenTimestamp,
},
},
{
timestamp: lastSeenTimestamp,
id: {
gt: lastSeenMessageId,
},
},
],
}
: {
timestamp: {
gt: lastSeenTimestamp,
},
}),
};
return this.prisma.agentConversationMessage.findMany({
where,
orderBy: [{ timestamp: "asc" }, { id: "asc" }],
});
}
}

View File

@@ -0,0 +1,202 @@
import { Logger } from "@nestjs/common";
import type {
AgentMessage,
AgentSession,
AgentSessionList,
IAgentProvider,
InjectResult,
} from "@mosaic/shared";
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { AgentProviderRegistry } from "./agent-provider.registry";
import { InternalAgentProvider } from "./internal-agent.provider";
type MockProvider = IAgentProvider & {
listSessions: ReturnType<typeof vi.fn>;
getSession: ReturnType<typeof vi.fn>;
};
const emptyMessageStream = async function* (): AsyncIterable<AgentMessage> {
return;
};
const createProvider = (providerId: string, sessions: AgentSession[] = []): MockProvider => {
return {
providerId,
providerType: providerId,
displayName: providerId,
listSessions: vi.fn().mockResolvedValue({
sessions,
total: sessions.length,
} as AgentSessionList),
getSession: vi.fn().mockResolvedValue(null),
getMessages: vi.fn().mockResolvedValue([]),
injectMessage: vi.fn().mockResolvedValue({ accepted: true } as InjectResult),
pauseSession: vi.fn().mockResolvedValue(undefined),
resumeSession: vi.fn().mockResolvedValue(undefined),
killSession: vi.fn().mockResolvedValue(undefined),
streamMessages: vi.fn().mockReturnValue(emptyMessageStream()),
isAvailable: vi.fn().mockResolvedValue(true),
};
};
describe("AgentProviderRegistry", () => {
let registry: AgentProviderRegistry;
let internalProvider: MockProvider;
beforeEach(() => {
internalProvider = createProvider("internal");
registry = new AgentProviderRegistry(internalProvider as unknown as InternalAgentProvider);
});
afterEach(() => {
vi.restoreAllMocks();
});
it("registers InternalAgentProvider on module init", () => {
registry.onModuleInit();
expect(registry.getProvider("internal")).toBe(internalProvider);
});
it("registers providers and returns null for unknown provider ids", () => {
const externalProvider = createProvider("openclaw");
registry.registerProvider(externalProvider);
expect(registry.getProvider("openclaw")).toBe(externalProvider);
expect(registry.getProvider("missing")).toBeNull();
});
it("aggregates and sorts sessions from all providers", async () => {
const internalSessions: AgentSession[] = [
{
id: "session-older",
providerId: "internal",
providerType: "internal",
status: "active",
createdAt: new Date("2026-03-07T10:00:00.000Z"),
updatedAt: new Date("2026-03-07T10:10:00.000Z"),
},
];
const externalSessions: AgentSession[] = [
{
id: "session-newer",
providerId: "openclaw",
providerType: "external",
status: "paused",
createdAt: new Date("2026-03-07T09:00:00.000Z"),
updatedAt: new Date("2026-03-07T10:20:00.000Z"),
},
];
internalProvider.listSessions.mockResolvedValue({
sessions: internalSessions,
total: internalSessions.length,
} as AgentSessionList);
const externalProvider = createProvider("openclaw", externalSessions);
registry.onModuleInit();
registry.registerProvider(externalProvider);
const result = await registry.listAllSessions();
expect(result.map((session) => session.id)).toEqual(["session-newer", "session-older"]);
expect(internalProvider.listSessions).toHaveBeenCalledTimes(1);
expect(externalProvider.listSessions).toHaveBeenCalledTimes(1);
});
it("skips provider failures and logs warning", async () => {
const warnSpy = vi.spyOn(Logger.prototype, "warn").mockImplementation(() => undefined);
const healthyProvider = createProvider("healthy", [
{
id: "session-1",
providerId: "healthy",
providerType: "external",
status: "active",
createdAt: new Date("2026-03-07T11:00:00.000Z"),
updatedAt: new Date("2026-03-07T11:00:00.000Z"),
},
]);
const failingProvider = createProvider("failing");
failingProvider.listSessions.mockRejectedValue(new Error("provider offline"));
registry.onModuleInit();
registry.registerProvider(healthyProvider);
registry.registerProvider(failingProvider);
const result = await registry.listAllSessions();
expect(result).toHaveLength(1);
expect(result[0]?.id).toBe("session-1");
expect(warnSpy).toHaveBeenCalledWith(
expect.stringContaining("Failed to list sessions for provider failing")
);
});
it("finds a provider for an existing session", async () => {
const targetSession: AgentSession = {
id: "session-found",
providerId: "openclaw",
providerType: "external",
status: "active",
createdAt: new Date("2026-03-07T12:00:00.000Z"),
updatedAt: new Date("2026-03-07T12:10:00.000Z"),
};
const openclawProvider = createProvider("openclaw");
openclawProvider.getSession.mockResolvedValue(targetSession);
registry.onModuleInit();
registry.registerProvider(openclawProvider);
const result = await registry.getProviderForSession(targetSession.id);
expect(result).toEqual({
provider: openclawProvider,
session: targetSession,
});
expect(internalProvider.getSession).toHaveBeenCalledWith(targetSession.id);
expect(openclawProvider.getSession).toHaveBeenCalledWith(targetSession.id);
});
it("returns null when no provider has the requested session", async () => {
const openclawProvider = createProvider("openclaw");
registry.onModuleInit();
registry.registerProvider(openclawProvider);
await expect(registry.getProviderForSession("missing-session")).resolves.toBeNull();
});
it("continues searching providers when getSession throws", async () => {
const warnSpy = vi.spyOn(Logger.prototype, "warn").mockImplementation(() => undefined);
const failingProvider = createProvider("failing");
failingProvider.getSession.mockRejectedValue(new Error("provider timeout"));
const healthySession: AgentSession = {
id: "session-healthy",
providerId: "healthy",
providerType: "external",
status: "active",
createdAt: new Date("2026-03-07T12:15:00.000Z"),
updatedAt: new Date("2026-03-07T12:16:00.000Z"),
};
const healthyProvider = createProvider("healthy");
healthyProvider.getSession.mockResolvedValue(healthySession);
registry.onModuleInit();
registry.registerProvider(failingProvider);
registry.registerProvider(healthyProvider);
const result = await registry.getProviderForSession(healthySession.id);
expect(result).toEqual({ provider: healthyProvider, session: healthySession });
expect(warnSpy).toHaveBeenCalledWith(
expect.stringContaining("Failed to get session session-healthy for provider failing")
);
});
});

View File

@@ -0,0 +1,79 @@
import { Injectable, Logger, OnModuleInit } from "@nestjs/common";
import type { AgentSession, IAgentProvider } from "@mosaic/shared";
import { InternalAgentProvider } from "./internal-agent.provider";
@Injectable()
export class AgentProviderRegistry implements OnModuleInit {
private readonly logger = new Logger(AgentProviderRegistry.name);
private readonly providers = new Map<string, IAgentProvider>();
constructor(private readonly internalProvider: InternalAgentProvider) {}
onModuleInit(): void {
this.registerProvider(this.internalProvider);
}
registerProvider(provider: IAgentProvider): void {
const existingProvider = this.providers.get(provider.providerId);
if (existingProvider !== undefined) {
this.logger.warn(`Replacing existing provider registration for ${provider.providerId}`);
}
this.providers.set(provider.providerId, provider);
}
getProvider(providerId: string): IAgentProvider | null {
return this.providers.get(providerId) ?? null;
}
async getProviderForSession(
sessionId: string
): Promise<{ provider: IAgentProvider; session: AgentSession } | null> {
for (const provider of this.providers.values()) {
try {
const session = await provider.getSession(sessionId);
if (session !== null) {
return {
provider,
session,
};
}
} catch (error) {
this.logger.warn(
`Failed to get session ${sessionId} for provider ${provider.providerId}: ${this.toErrorMessage(error)}`
);
}
}
return null;
}
async listAllSessions(): Promise<AgentSession[]> {
const providers = [...this.providers.values()];
const sessionsByProvider = await Promise.all(
providers.map(async (provider) => {
try {
const { sessions } = await provider.listSessions();
return sessions;
} catch (error) {
this.logger.warn(
`Failed to list sessions for provider ${provider.providerId}: ${this.toErrorMessage(error)}`
);
return [];
}
})
);
return sessionsByProvider
.flat()
.sort((left, right) => right.updatedAt.getTime() - left.updatedAt.getTime());
}
private toErrorMessage(error: unknown): string {
if (error instanceof Error) {
return error.message;
}
return String(error);
}
}

View File

@@ -0,0 +1,245 @@
import { describe, it, expect, beforeEach, afterEach, vi } from "vitest";
import { AgentTreeService } from "./agent-tree.service";
import { PrismaService } from "../../prisma/prisma.service";
describe("AgentTreeService", () => {
let service: AgentTreeService;
let prisma: {
agentSessionTree: {
findMany: ReturnType<typeof vi.fn>;
count: ReturnType<typeof vi.fn>;
findUnique: ReturnType<typeof vi.fn>;
};
};
beforeEach(() => {
prisma = {
agentSessionTree: {
findMany: vi.fn(),
count: vi.fn(),
findUnique: vi.fn(),
},
};
service = new AgentTreeService(prisma as unknown as PrismaService);
});
afterEach(() => {
vi.clearAllMocks();
});
describe("listSessions", () => {
it("returns paginated sessions and cursor", async () => {
const sessions = [
{
id: "tree-2",
sessionId: "agent-2",
parentSessionId: null,
provider: "internal",
missionId: null,
taskId: "task-2",
taskSource: "queue",
agentType: "worker",
status: "running",
spawnedAt: new Date("2026-03-07T11:00:00.000Z"),
completedAt: null,
metadata: {},
},
{
id: "tree-1",
sessionId: "agent-1",
parentSessionId: null,
provider: "internal",
missionId: null,
taskId: "task-1",
taskSource: "queue",
agentType: "worker",
status: "running",
spawnedAt: new Date("2026-03-07T10:00:00.000Z"),
completedAt: null,
metadata: {},
},
];
prisma.agentSessionTree.findMany.mockResolvedValue(sessions);
prisma.agentSessionTree.count.mockResolvedValue(7);
const result = await service.listSessions(undefined, 2);
expect(prisma.agentSessionTree.findMany).toHaveBeenCalledWith({
where: undefined,
orderBy: [{ spawnedAt: "desc" }, { sessionId: "desc" }],
take: 2,
});
expect(prisma.agentSessionTree.count).toHaveBeenCalledWith();
expect(result.sessions).toEqual(sessions);
expect(result.total).toBe(7);
expect(result.cursor).toBeTypeOf("string");
});
it("applies cursor filter when provided", async () => {
prisma.agentSessionTree.findMany.mockResolvedValue([]);
prisma.agentSessionTree.count.mockResolvedValue(0);
const cursorDate = "2026-03-07T10:00:00.000Z";
const cursorSessionId = "agent-5";
const cursor = Buffer.from(
JSON.stringify({
spawnedAt: cursorDate,
sessionId: cursorSessionId,
}),
"utf8"
).toString("base64url");
await service.listSessions(cursor, 25);
expect(prisma.agentSessionTree.findMany).toHaveBeenCalledWith({
where: {
OR: [
{
spawnedAt: {
lt: new Date(cursorDate),
},
},
{
spawnedAt: new Date(cursorDate),
sessionId: {
lt: cursorSessionId,
},
},
],
},
orderBy: [{ spawnedAt: "desc" }, { sessionId: "desc" }],
take: 25,
});
});
it("ignores invalid cursor values", async () => {
prisma.agentSessionTree.findMany.mockResolvedValue([]);
prisma.agentSessionTree.count.mockResolvedValue(0);
await service.listSessions("invalid-cursor", 10);
expect(prisma.agentSessionTree.findMany).toHaveBeenCalledWith({
where: undefined,
orderBy: [{ spawnedAt: "desc" }, { sessionId: "desc" }],
take: 10,
});
});
});
describe("getSession", () => {
it("returns matching session entry", async () => {
const session = {
id: "tree-1",
sessionId: "agent-123",
parentSessionId: null,
provider: "internal",
missionId: null,
taskId: "task-1",
taskSource: "queue",
agentType: "worker",
status: "running",
spawnedAt: new Date("2026-03-07T11:00:00.000Z"),
completedAt: null,
metadata: {},
};
prisma.agentSessionTree.findUnique.mockResolvedValue(session);
const result = await service.getSession("agent-123");
expect(prisma.agentSessionTree.findUnique).toHaveBeenCalledWith({
where: { sessionId: "agent-123" },
});
expect(result).toEqual(session);
});
it("returns null when session does not exist", async () => {
prisma.agentSessionTree.findUnique.mockResolvedValue(null);
const result = await service.getSession("agent-missing");
expect(result).toBeNull();
});
});
describe("getTree", () => {
it("returns mapped entries from Prisma", async () => {
prisma.agentSessionTree.findMany.mockResolvedValue([
{
id: "tree-1",
sessionId: "agent-1",
parentSessionId: "agent-root",
provider: "internal",
missionId: "mission-1",
taskId: "task-1",
taskSource: "queue",
agentType: "worker",
status: "running",
spawnedAt: new Date("2026-03-07T10:00:00.000Z"),
completedAt: new Date("2026-03-07T11:00:00.000Z"),
metadata: {},
},
]);
const result = await service.getTree();
expect(prisma.agentSessionTree.findMany).toHaveBeenCalledWith({
orderBy: { spawnedAt: "desc" },
take: 200,
});
expect(result).toEqual([
{
sessionId: "agent-1",
parentSessionId: "agent-root",
status: "running",
agentType: "worker",
taskSource: "queue",
spawnedAt: "2026-03-07T10:00:00.000Z",
completedAt: "2026-03-07T11:00:00.000Z",
},
]);
});
it("returns empty array when no entries exist", async () => {
prisma.agentSessionTree.findMany.mockResolvedValue([]);
const result = await service.getTree();
expect(result).toEqual([]);
});
it("maps null parentSessionId and completedAt correctly", async () => {
prisma.agentSessionTree.findMany.mockResolvedValue([
{
id: "tree-2",
sessionId: "agent-root",
parentSessionId: null,
provider: "internal",
missionId: null,
taskId: null,
taskSource: null,
agentType: null,
status: "spawning",
spawnedAt: new Date("2026-03-07T09:00:00.000Z"),
completedAt: null,
metadata: {},
},
]);
const result = await service.getTree();
expect(result).toEqual([
{
sessionId: "agent-root",
parentSessionId: null,
status: "spawning",
agentType: null,
taskSource: null,
spawnedAt: "2026-03-07T09:00:00.000Z",
completedAt: null,
},
]);
});
});
});

View File

@@ -0,0 +1,146 @@
import { Injectable } from "@nestjs/common";
import type { AgentSessionTree, Prisma } from "@prisma/client";
import { AgentTreeResponseDto } from "./dto/agent-tree-response.dto";
import { PrismaService } from "../../prisma/prisma.service";
const DEFAULT_PAGE_LIMIT = 50;
const MAX_PAGE_LIMIT = 200;
interface SessionCursor {
spawnedAt: Date;
sessionId: string;
}
export interface AgentSessionTreeListResult {
sessions: AgentSessionTree[];
total: number;
cursor?: string;
}
@Injectable()
export class AgentTreeService {
constructor(private readonly prisma: PrismaService) {}
async listSessions(
cursor?: string,
limit = DEFAULT_PAGE_LIMIT
): Promise<AgentSessionTreeListResult> {
const safeLimit = this.normalizeLimit(limit);
const parsedCursor = this.parseCursor(cursor);
const where: Prisma.AgentSessionTreeWhereInput | undefined = parsedCursor
? {
OR: [
{
spawnedAt: {
lt: parsedCursor.spawnedAt,
},
},
{
spawnedAt: parsedCursor.spawnedAt,
sessionId: {
lt: parsedCursor.sessionId,
},
},
],
}
: undefined;
const [sessions, total] = await Promise.all([
this.prisma.agentSessionTree.findMany({
where,
orderBy: [{ spawnedAt: "desc" }, { sessionId: "desc" }],
take: safeLimit,
}),
this.prisma.agentSessionTree.count(),
]);
const nextCursor =
sessions.length === safeLimit
? this.serializeCursor(sessions[sessions.length - 1])
: undefined;
return {
sessions,
total,
...(nextCursor !== undefined ? { cursor: nextCursor } : {}),
};
}
async getSession(sessionId: string): Promise<AgentSessionTree | null> {
return this.prisma.agentSessionTree.findUnique({
where: { sessionId },
});
}
async getTree(): Promise<AgentTreeResponseDto[]> {
const entries = await this.prisma.agentSessionTree.findMany({
orderBy: { spawnedAt: "desc" },
take: 200,
});
const response: AgentTreeResponseDto[] = [];
for (const entry of entries) {
response.push({
sessionId: entry.sessionId,
parentSessionId: entry.parentSessionId ?? null,
status: entry.status,
agentType: entry.agentType ?? null,
taskSource: entry.taskSource ?? null,
spawnedAt: entry.spawnedAt.toISOString(),
completedAt: entry.completedAt?.toISOString() ?? null,
});
}
return response;
}
private normalizeLimit(limit: number): number {
const normalized = Number.isFinite(limit) ? Math.trunc(limit) : DEFAULT_PAGE_LIMIT;
if (normalized < 1) {
return 1;
}
return Math.min(normalized, MAX_PAGE_LIMIT);
}
private serializeCursor(entry: Pick<AgentSessionTree, "spawnedAt" | "sessionId">): string {
return Buffer.from(
JSON.stringify({
spawnedAt: entry.spawnedAt.toISOString(),
sessionId: entry.sessionId,
}),
"utf8"
).toString("base64url");
}
private parseCursor(cursor?: string): SessionCursor | null {
if (!cursor) {
return null;
}
try {
const decoded = Buffer.from(cursor, "base64url").toString("utf8");
const parsed = JSON.parse(decoded) as {
spawnedAt?: string;
sessionId?: string;
};
if (typeof parsed.spawnedAt !== "string" || typeof parsed.sessionId !== "string") {
return null;
}
const spawnedAt = new Date(parsed.spawnedAt);
if (Number.isNaN(spawnedAt.getTime())) {
return null;
}
return {
spawnedAt,
sessionId: parsed.sessionId,
};
} catch {
return null;
}
}
}

View File

@@ -5,6 +5,9 @@ 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 { AgentMessagesService } from "./agent-messages.service";
import { AgentControlService } from "./agent-control.service";
import { AgentTreeService } from "./agent-tree.service";
import type { KillAllResult } from "../../killswitch/killswitch.service";
describe("AgentsController - Killswitch Endpoints", () => {
@@ -27,6 +30,20 @@ describe("AgentsController - Killswitch Endpoints", () => {
subscribe: ReturnType<typeof vi.fn>;
getInitialSnapshot: ReturnType<typeof vi.fn>;
createHeartbeat: ReturnType<typeof vi.fn>;
getRecentEvents: ReturnType<typeof vi.fn>;
};
let mockMessagesService: {
getMessages: ReturnType<typeof vi.fn>;
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>;
};
let mockTreeService: {
getTree: ReturnType<typeof vi.fn>;
};
beforeEach(() => {
@@ -61,6 +78,23 @@ describe("AgentsController - Killswitch Endpoints", () => {
timestamp: new Date().toISOString(),
data: { heartbeat: true },
}),
getRecentEvents: vi.fn().mockReturnValue([]),
};
mockMessagesService = {
getMessages: vi.fn(),
getReplayMessages: vi.fn().mockResolvedValue([]),
getMessagesAfter: vi.fn().mockResolvedValue([]),
};
mockControlService = {
injectMessage: vi.fn().mockResolvedValue(undefined),
pauseAgent: vi.fn().mockResolvedValue(undefined),
resumeAgent: vi.fn().mockResolvedValue(undefined),
};
mockTreeService = {
getTree: vi.fn().mockResolvedValue([]),
};
controller = new AgentsController(
@@ -68,7 +102,10 @@ describe("AgentsController - Killswitch Endpoints", () => {
mockSpawnerService as unknown as AgentSpawnerService,
mockLifecycleService as unknown as AgentLifecycleService,
mockKillswitchService as unknown as KillswitchService,
mockEventsService as unknown as AgentEventsService
mockEventsService as unknown as AgentEventsService,
mockMessagesService as unknown as AgentMessagesService,
mockControlService as unknown as AgentControlService,
mockTreeService as unknown as AgentTreeService
);
});

View File

@@ -4,6 +4,9 @@ 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 { AgentMessagesService } from "./agent-messages.service";
import { AgentControlService } from "./agent-control.service";
import { AgentTreeService } from "./agent-tree.service";
import { describe, it, expect, beforeEach, afterEach, vi } from "vitest";
describe("AgentsController", () => {
@@ -30,6 +33,19 @@ describe("AgentsController", () => {
createHeartbeat: ReturnType<typeof vi.fn>;
getRecentEvents: ReturnType<typeof vi.fn>;
};
let messagesService: {
getMessages: ReturnType<typeof vi.fn>;
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>;
};
let treeService: {
getTree: ReturnType<typeof vi.fn>;
};
beforeEach(() => {
// Create mock services
@@ -69,13 +85,32 @@ describe("AgentsController", () => {
getRecentEvents: vi.fn().mockReturnValue([]),
};
messagesService = {
getMessages: vi.fn(),
getReplayMessages: vi.fn().mockResolvedValue([]),
getMessagesAfter: vi.fn().mockResolvedValue([]),
};
controlService = {
injectMessage: vi.fn().mockResolvedValue(undefined),
pauseAgent: vi.fn().mockResolvedValue(undefined),
resumeAgent: vi.fn().mockResolvedValue(undefined),
};
treeService = {
getTree: vi.fn().mockResolvedValue([]),
};
// 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,
eventsService as unknown as AgentEventsService
eventsService as unknown as AgentEventsService,
messagesService as unknown as AgentMessagesService,
controlService as unknown as AgentControlService,
treeService as unknown as AgentTreeService
);
});
@@ -87,6 +122,27 @@ describe("AgentsController", () => {
expect(controller).toBeDefined();
});
describe("getAgentTree", () => {
it("should return tree entries", async () => {
const entries = [
{
sessionId: "agent-1",
parentSessionId: null,
status: "running",
agentType: "worker",
taskSource: "internal",
spawnedAt: "2026-03-07T00:00:00.000Z",
completedAt: null,
},
];
treeService.getTree.mockResolvedValue(entries);
await expect(controller.getAgentTree()).resolves.toEqual(entries);
expect(treeService.getTree).toHaveBeenCalledTimes(1);
});
});
describe("listAgents", () => {
it("should return empty array when no agents exist", () => {
// Arrange
@@ -365,6 +421,93 @@ 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";
const query = {
limit: 25,
skip: 10,
};
const response = {
messages: [
{
id: "msg-1",
sessionId: agentId,
role: "agent",
content: "hello",
provider: "internal",
timestamp: new Date("2026-03-07T03:00:00.000Z"),
metadata: {},
},
],
total: 101,
};
messagesService.getMessages.mockResolvedValue(response);
const result = await controller.getAgentMessages(agentId, query);
expect(messagesService.getMessages).toHaveBeenCalledWith(agentId, 25, 10);
expect(result).toEqual(response);
});
it("should use default pagination values", async () => {
const agentId = "0b64079f-4487-42b9-92eb-cf8ea0042a64";
const query = {
limit: 50,
skip: 0,
};
messagesService.getMessages.mockResolvedValue({ messages: [], total: 0 });
await controller.getAgentMessages(agentId, query);
expect(messagesService.getMessages).toHaveBeenCalledWith(agentId, 50, 0);
});
});
describe("getRecentEvents", () => {
it("should return recent events with default limit", () => {
eventsService.getRecentEvents.mockReturnValue([

View File

@@ -14,7 +14,9 @@ import {
Sse,
MessageEvent,
Query,
Request,
} from "@nestjs/common";
import type { AgentConversationMessage } from "@prisma/client";
import { Throttle } from "@nestjs/throttler";
import { Observable } from "rxjs";
import { QueueService } from "../../queue/queue.service";
@@ -25,6 +27,13 @@ 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";
import { GetMessagesQueryDto } from "./dto/get-messages-query.dto";
import { AgentMessagesService } from "./agent-messages.service";
import { AgentControlService } from "./agent-control.service";
import { AgentTreeService } from "./agent-tree.service";
import { AgentTreeResponseDto } from "./dto/agent-tree-response.dto";
import { InjectAgentDto } from "./dto/inject-agent.dto";
import { PauseAgentDto, ResumeAgentDto } from "./dto/control-agent.dto";
/**
* Controller for agent management endpoints
@@ -47,7 +56,10 @@ export class AgentsController {
private readonly spawnerService: AgentSpawnerService,
private readonly lifecycleService: AgentLifecycleService,
private readonly killswitchService: KillswitchService,
private readonly eventsService: AgentEventsService
private readonly eventsService: AgentEventsService,
private readonly messagesService: AgentMessagesService,
private readonly agentControlService: AgentControlService,
private readonly agentTreeService: AgentTreeService
) {}
/**
@@ -69,6 +81,7 @@ export class AgentsController {
// Spawn agent using spawner service
const spawnResponse = this.spawnerService.spawnAgent({
taskId: dto.taskId,
...(dto.parentAgentId !== undefined ? { parentAgentId: dto.parentAgentId } : {}),
agentType: dto.agentType,
context: dto.context,
});
@@ -143,6 +156,13 @@ export class AgentsController {
};
}
@Get("tree")
@UseGuards(OrchestratorApiKeyGuard)
@Throttle({ default: { limit: 200, ttl: 60000 } })
async getAgentTree(): Promise<AgentTreeResponseDto[]> {
return this.agentTreeService.getTree();
}
/**
* List all agents
* @returns Array of all agent sessions with their status
@@ -185,6 +205,107 @@ export class AgentsController {
}
}
/**
* Get paginated message history for an agent.
*/
@Get(":agentId/messages")
@Throttle({ status: { limit: 200, ttl: 60000 } })
@UsePipes(new ValidationPipe({ transform: true, whitelist: true }))
async getAgentMessages(
@Param("agentId", ParseUUIDPipe) agentId: string,
@Query() query: GetMessagesQueryDto
): Promise<{
messages: AgentConversationMessage[];
total: number;
}> {
return this.messagesService.getMessages(agentId, query.limit, query.skip);
}
/**
* Stream per-agent conversation messages as server-sent events (SSE).
*/
@Sse(":agentId/messages/stream")
@Throttle({ status: { limit: 200, ttl: 60000 } })
streamAgentMessages(@Param("agentId", ParseUUIDPipe) agentId: string): Observable<MessageEvent> {
return new Observable<MessageEvent>((subscriber) => {
let isClosed = false;
let lastSeenTimestamp = new Date();
let lastSeenMessageId: string | null = null;
const emitMessage = (message: AgentConversationMessage): void => {
if (isClosed) {
return;
}
subscriber.next({
data: this.toMessageStreamPayload(message),
});
lastSeenTimestamp = message.timestamp;
lastSeenMessageId = message.id;
};
void this.messagesService
.getReplayMessages(agentId, 50)
.then((messages) => {
if (isClosed) {
return;
}
messages.forEach((message) => {
emitMessage(message);
});
if (messages.length === 0) {
lastSeenTimestamp = new Date();
lastSeenMessageId = null;
}
})
.catch((error: unknown) => {
this.logger.error(
`Failed to load replay messages for ${agentId}: ${error instanceof Error ? error.message : String(error)}`
);
lastSeenTimestamp = new Date();
lastSeenMessageId = null;
});
const pollInterval = setInterval(() => {
if (isClosed) {
return;
}
void this.messagesService
.getMessagesAfter(agentId, lastSeenTimestamp, lastSeenMessageId)
.then((messages) => {
if (isClosed || messages.length === 0) {
return;
}
messages.forEach((message) => {
emitMessage(message);
});
})
.catch((error: unknown) => {
this.logger.error(
`Failed to poll messages for ${agentId}: ${error instanceof Error ? error.message : String(error)}`
);
});
}, 1000);
const heartbeat = setInterval(() => {
if (!isClosed) {
subscriber.next({ data: { type: "heartbeat" } });
}
}, 15000);
return () => {
isClosed = true;
clearInterval(pollInterval);
clearInterval(heartbeat);
};
});
}
/**
* Get agent status
* @param agentId Agent ID to query
@@ -269,6 +390,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
@@ -301,4 +473,24 @@ export class AgentsController {
throw error;
}
}
private toMessageStreamPayload(message: AgentConversationMessage): {
messageId: string;
sessionId: string;
role: string;
content: string;
provider: string;
timestamp: string;
metadata: unknown;
} {
return {
messageId: message.id,
sessionId: message.sessionId,
role: message.role,
content: message.content,
provider: message.provider,
timestamp: message.timestamp.toISOString(),
metadata: message.metadata,
};
}
}

View File

@@ -6,10 +6,25 @@ 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";
import { PrismaModule } from "../../prisma/prisma.module";
import { AgentMessagesService } from "./agent-messages.service";
import { AgentControlService } from "./agent-control.service";
import { AgentTreeService } from "./agent-tree.service";
import { InternalAgentProvider } from "./internal-agent.provider";
import { AgentProviderRegistry } from "./agent-provider.registry";
@Module({
imports: [QueueModule, SpawnerModule, KillswitchModule, ValkeyModule],
imports: [QueueModule, SpawnerModule, KillswitchModule, ValkeyModule, PrismaModule],
controllers: [AgentsController],
providers: [OrchestratorApiKeyGuard, AgentEventsService],
providers: [
OrchestratorApiKeyGuard,
AgentEventsService,
AgentMessagesService,
AgentControlService,
AgentTreeService,
InternalAgentProvider,
AgentProviderRegistry,
],
exports: [InternalAgentProvider, AgentProviderRegistry],
})
export class AgentsModule {}

View File

@@ -0,0 +1,9 @@
export class AgentTreeResponseDto {
sessionId!: string;
parentSessionId!: string | null;
status!: string;
agentType!: string | null;
taskSource!: string | null;
spawnedAt!: string;
completedAt!: string | null;
}

View File

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

View File

@@ -0,0 +1,37 @@
import { plainToInstance } from "class-transformer";
import { validate } from "class-validator";
import { describe, expect, it } from "vitest";
import { GetMessagesQueryDto } from "./get-messages-query.dto";
describe("GetMessagesQueryDto", () => {
it("should use defaults when empty", async () => {
const dto = plainToInstance(GetMessagesQueryDto, {});
const errors = await validate(dto);
expect(errors).toHaveLength(0);
expect(dto.limit).toBe(50);
expect(dto.skip).toBe(0);
});
it("should reject limit greater than 200", async () => {
const dto = plainToInstance(GetMessagesQueryDto, {
limit: 201,
skip: 0,
});
const errors = await validate(dto);
expect(errors.length).toBeGreaterThan(0);
expect(errors.some((error) => error.property === "limit")).toBe(true);
});
it("should reject negative skip", async () => {
const dto = plainToInstance(GetMessagesQueryDto, {
limit: 50,
skip: -1,
});
const errors = await validate(dto);
expect(errors.length).toBeGreaterThan(0);
expect(errors.some((error) => error.property === "skip")).toBe(true);
});
});

View File

@@ -0,0 +1,17 @@
import { Type } from "class-transformer";
import { IsInt, IsOptional, Max, Min } from "class-validator";
export class GetMessagesQueryDto {
@IsOptional()
@Type(() => Number)
@IsInt()
@Min(1)
@Max(200)
limit = 50;
@IsOptional()
@Type(() => Number)
@IsInt()
@Min(0)
skip = 0;
}

View File

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

View File

@@ -116,6 +116,10 @@ export class SpawnAgentDto {
@IsOptional()
@IsIn(["strict", "standard", "minimal", "custom"])
gateProfile?: GateProfileType;
@IsOptional()
@IsString()
parentAgentId?: string;
}
/**

View File

@@ -0,0 +1,216 @@
import { beforeEach, describe, expect, it, vi } from "vitest";
import type { AgentConversationMessage, AgentSessionTree } from "@prisma/client";
import { AgentControlService } from "./agent-control.service";
import { AgentMessagesService } from "./agent-messages.service";
import { AgentTreeService } from "./agent-tree.service";
import { InternalAgentProvider } from "./internal-agent.provider";
describe("InternalAgentProvider", () => {
let provider: InternalAgentProvider;
let messagesService: {
getMessages: ReturnType<typeof vi.fn>;
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>;
killAgent: ReturnType<typeof vi.fn>;
};
let treeService: {
listSessions: ReturnType<typeof vi.fn>;
getSession: ReturnType<typeof vi.fn>;
};
beforeEach(() => {
messagesService = {
getMessages: vi.fn(),
getReplayMessages: vi.fn(),
getMessagesAfter: vi.fn(),
};
controlService = {
injectMessage: vi.fn().mockResolvedValue(undefined),
pauseAgent: vi.fn().mockResolvedValue(undefined),
resumeAgent: vi.fn().mockResolvedValue(undefined),
killAgent: vi.fn().mockResolvedValue(undefined),
};
treeService = {
listSessions: vi.fn(),
getSession: vi.fn(),
};
provider = new InternalAgentProvider(
messagesService as unknown as AgentMessagesService,
controlService as unknown as AgentControlService,
treeService as unknown as AgentTreeService
);
});
it("maps paginated sessions", async () => {
const sessionEntry: AgentSessionTree = {
id: "tree-1",
sessionId: "session-1",
parentSessionId: "parent-1",
provider: "internal",
missionId: null,
taskId: "task-123",
taskSource: "queue",
agentType: "worker",
status: "running",
spawnedAt: new Date("2026-03-07T10:00:00.000Z"),
completedAt: null,
metadata: { branch: "feat/test" },
};
treeService.listSessions.mockResolvedValue({
sessions: [sessionEntry],
total: 1,
cursor: "next-cursor",
});
const result = await provider.listSessions("cursor-1", 25);
expect(treeService.listSessions).toHaveBeenCalledWith("cursor-1", 25);
expect(result).toEqual({
sessions: [
{
id: "session-1",
providerId: "internal",
providerType: "internal",
label: "task-123",
status: "active",
parentSessionId: "parent-1",
createdAt: new Date("2026-03-07T10:00:00.000Z"),
updatedAt: new Date("2026-03-07T10:00:00.000Z"),
metadata: { branch: "feat/test" },
},
],
total: 1,
cursor: "next-cursor",
});
});
it("returns null for missing session", async () => {
treeService.getSession.mockResolvedValue(null);
const result = await provider.getSession("missing-session");
expect(treeService.getSession).toHaveBeenCalledWith("missing-session");
expect(result).toBeNull();
});
it("maps message history and parses skip cursor", async () => {
const message: AgentConversationMessage = {
id: "msg-1",
sessionId: "session-1",
provider: "internal",
role: "agent",
content: "hello",
timestamp: new Date("2026-03-07T10:05:00.000Z"),
metadata: { tokens: 42 },
};
messagesService.getMessages.mockResolvedValue({
messages: [message],
total: 10,
});
const result = await provider.getMessages("session-1", 30, "2");
expect(messagesService.getMessages).toHaveBeenCalledWith("session-1", 30, 2);
expect(result).toEqual([
{
id: "msg-1",
sessionId: "session-1",
role: "assistant",
content: "hello",
timestamp: new Date("2026-03-07T10:05:00.000Z"),
metadata: { tokens: 42 },
},
]);
});
it("routes control operations through AgentControlService", async () => {
const injectResult = await provider.injectMessage("session-1", "new instruction");
await provider.pauseSession("session-1");
await provider.resumeSession("session-1");
await provider.killSession("session-1", false);
expect(controlService.injectMessage).toHaveBeenCalledWith(
"session-1",
"internal-provider",
"new instruction"
);
expect(injectResult).toEqual({ accepted: true });
expect(controlService.pauseAgent).toHaveBeenCalledWith("session-1", "internal-provider");
expect(controlService.resumeAgent).toHaveBeenCalledWith("session-1", "internal-provider");
expect(controlService.killAgent).toHaveBeenCalledWith("session-1", "internal-provider", false);
});
it("streams replay and incremental messages", async () => {
const replayMessage: AgentConversationMessage = {
id: "msg-replay",
sessionId: "session-1",
provider: "internal",
role: "agent",
content: "replay",
timestamp: new Date("2026-03-07T10:00:00.000Z"),
metadata: {},
};
const incrementalMessage: AgentConversationMessage = {
id: "msg-live",
sessionId: "session-1",
provider: "internal",
role: "operator",
content: "live",
timestamp: new Date("2026-03-07T10:00:01.000Z"),
metadata: {},
};
messagesService.getReplayMessages.mockResolvedValue([replayMessage]);
messagesService.getMessagesAfter
.mockResolvedValueOnce([incrementalMessage])
.mockResolvedValueOnce([]);
const iterator = provider.streamMessages("session-1")[Symbol.asyncIterator]();
const first = await iterator.next();
const second = await iterator.next();
expect(first.done).toBe(false);
expect(first.value).toEqual({
id: "msg-replay",
sessionId: "session-1",
role: "assistant",
content: "replay",
timestamp: new Date("2026-03-07T10:00:00.000Z"),
metadata: {},
});
expect(second.done).toBe(false);
expect(second.value).toEqual({
id: "msg-live",
sessionId: "session-1",
role: "user",
content: "live",
timestamp: new Date("2026-03-07T10:00:01.000Z"),
metadata: {},
});
await iterator.return?.();
expect(messagesService.getReplayMessages).toHaveBeenCalledWith("session-1", 50);
expect(messagesService.getMessagesAfter).toHaveBeenCalledWith(
"session-1",
new Date("2026-03-07T10:00:00.000Z"),
"msg-replay"
);
});
it("reports provider availability", async () => {
await expect(provider.isAvailable()).resolves.toBe(true);
});
});

View File

@@ -0,0 +1,218 @@
import { Injectable } from "@nestjs/common";
import type {
AgentMessage,
AgentMessageRole,
AgentSession,
AgentSessionList,
AgentSessionStatus,
IAgentProvider,
InjectResult,
} from "@mosaic/shared";
import type { AgentConversationMessage, AgentSessionTree } from "@prisma/client";
import { AgentControlService } from "./agent-control.service";
import { AgentMessagesService } from "./agent-messages.service";
import { AgentTreeService } from "./agent-tree.service";
const DEFAULT_SESSION_LIMIT = 50;
const DEFAULT_MESSAGE_LIMIT = 50;
const MAX_MESSAGE_LIMIT = 200;
const STREAM_POLL_INTERVAL_MS = 1000;
const INTERNAL_OPERATOR_ID = "internal-provider";
@Injectable()
export class InternalAgentProvider implements IAgentProvider {
readonly providerId = "internal";
readonly providerType = "internal";
readonly displayName = "Internal Orchestrator";
constructor(
private readonly messagesService: AgentMessagesService,
private readonly controlService: AgentControlService,
private readonly treeService: AgentTreeService
) {}
async listSessions(cursor?: string, limit = DEFAULT_SESSION_LIMIT): Promise<AgentSessionList> {
const {
sessions,
total,
cursor: nextCursor,
} = await this.treeService.listSessions(cursor, limit);
return {
sessions: sessions.map((session) => this.toAgentSession(session)),
total,
...(nextCursor !== undefined ? { cursor: nextCursor } : {}),
};
}
async getSession(sessionId: string): Promise<AgentSession | null> {
const session = await this.treeService.getSession(sessionId);
return session ? this.toAgentSession(session) : null;
}
async getMessages(
sessionId: string,
limit = DEFAULT_MESSAGE_LIMIT,
before?: string
): Promise<AgentMessage[]> {
const safeLimit = this.normalizeMessageLimit(limit);
const skip = this.parseSkip(before);
const result = await this.messagesService.getMessages(sessionId, safeLimit, skip);
return result.messages.map((message) => this.toAgentMessage(message));
}
async injectMessage(sessionId: string, content: string): Promise<InjectResult> {
await this.controlService.injectMessage(sessionId, INTERNAL_OPERATOR_ID, content);
return {
accepted: true,
};
}
async pauseSession(sessionId: string): Promise<void> {
await this.controlService.pauseAgent(sessionId, INTERNAL_OPERATOR_ID);
}
async resumeSession(sessionId: string): Promise<void> {
await this.controlService.resumeAgent(sessionId, INTERNAL_OPERATOR_ID);
}
async killSession(sessionId: string, force = true): Promise<void> {
await this.controlService.killAgent(sessionId, INTERNAL_OPERATOR_ID, force);
}
async *streamMessages(sessionId: string): AsyncIterable<AgentMessage> {
const replayMessages = await this.messagesService.getReplayMessages(
sessionId,
DEFAULT_MESSAGE_LIMIT
);
let lastSeenTimestamp = new Date();
let lastSeenMessageId: string | null = null;
for (const message of replayMessages) {
yield this.toAgentMessage(message);
lastSeenTimestamp = message.timestamp;
lastSeenMessageId = message.id;
}
for (;;) {
const newMessages = await this.messagesService.getMessagesAfter(
sessionId,
lastSeenTimestamp,
lastSeenMessageId
);
for (const message of newMessages) {
yield this.toAgentMessage(message);
lastSeenTimestamp = message.timestamp;
lastSeenMessageId = message.id;
}
await this.delay(STREAM_POLL_INTERVAL_MS);
}
}
isAvailable(): Promise<boolean> {
return Promise.resolve(true);
}
private toAgentSession(session: AgentSessionTree): AgentSession {
const metadata = this.toMetadata(session.metadata);
return {
id: session.sessionId,
providerId: this.providerId,
providerType: this.providerType,
...(session.taskId !== null ? { label: session.taskId } : {}),
status: this.toSessionStatus(session.status),
...(session.parentSessionId !== null ? { parentSessionId: session.parentSessionId } : {}),
createdAt: session.spawnedAt,
updatedAt: session.completedAt ?? session.spawnedAt,
...(metadata !== undefined ? { metadata } : {}),
};
}
private toAgentMessage(message: AgentConversationMessage): AgentMessage {
const metadata = this.toMetadata(message.metadata);
return {
id: message.id,
sessionId: message.sessionId,
role: this.toMessageRole(message.role),
content: message.content,
timestamp: message.timestamp,
...(metadata !== undefined ? { metadata } : {}),
};
}
private toSessionStatus(status: string): AgentSessionStatus {
switch (status) {
case "running":
return "active";
case "paused":
return "paused";
case "completed":
return "completed";
case "failed":
case "killed":
return "failed";
case "spawning":
default:
return "idle";
}
}
private toMessageRole(role: string): AgentMessageRole {
switch (role) {
case "agent":
case "assistant":
return "assistant";
case "system":
return "system";
case "tool":
return "tool";
case "operator":
case "user":
default:
return "user";
}
}
private normalizeMessageLimit(limit: number): number {
const normalized = Number.isFinite(limit) ? Math.trunc(limit) : DEFAULT_MESSAGE_LIMIT;
if (normalized < 1) {
return 1;
}
return Math.min(normalized, MAX_MESSAGE_LIMIT);
}
private parseSkip(before?: string): number {
if (!before) {
return 0;
}
const parsed = Number.parseInt(before, 10);
if (Number.isNaN(parsed) || parsed < 0) {
return 0;
}
return parsed;
}
private toMetadata(value: unknown): Record<string, unknown> | undefined {
if (value !== null && typeof value === "object" && !Array.isArray(value)) {
return value as Record<string, unknown>;
}
return undefined;
}
private async delay(ms: number): Promise<void> {
await new Promise((resolve) => {
setTimeout(resolve, ms);
});
}
}

View File

@@ -0,0 +1,21 @@
import { Type } from "class-transformer";
import { IsInt, IsOptional, IsString, Max, Min } from "class-validator";
export class GetMissionControlAuditLogQueryDto {
@IsOptional()
@IsString()
sessionId?: string;
@IsOptional()
@Type(() => Number)
@IsInt()
@Min(1)
page = 1;
@IsOptional()
@Type(() => Number)
@IsInt()
@Min(1)
@Max(200)
limit = 50;
}

View File

@@ -0,0 +1,15 @@
import { Type } from "class-transformer";
import { IsInt, IsOptional, IsString, Max, Min } from "class-validator";
export class GetMissionControlMessagesQueryDto {
@IsOptional()
@Type(() => Number)
@IsInt()
@Min(1)
@Max(200)
limit?: number;
@IsOptional()
@IsString()
before?: string;
}

View File

@@ -0,0 +1,7 @@
import { IsBoolean, IsOptional } from "class-validator";
export class KillSessionDto {
@IsOptional()
@IsBoolean()
force?: boolean;
}

View File

@@ -0,0 +1,67 @@
import { beforeEach, describe, expect, it, vi } from "vitest";
import type { AgentSession } from "@mosaic/shared";
import type { PrismaService } from "../../prisma/prisma.service";
import { AgentProviderRegistry } from "../agents/agent-provider.registry";
import { MissionControlController } from "./mission-control.controller";
import { MissionControlService } from "./mission-control.service";
describe("MissionControlController", () => {
let controller: MissionControlController;
let registry: {
listAllSessions: ReturnType<typeof vi.fn>;
getProviderForSession: ReturnType<typeof vi.fn>;
};
beforeEach(() => {
registry = {
listAllSessions: vi.fn(),
getProviderForSession: vi.fn(),
};
const prisma = {
operatorAuditLog: {
create: vi.fn().mockResolvedValue(undefined),
},
};
const service = new MissionControlService(
registry as unknown as AgentProviderRegistry,
prisma as unknown as PrismaService
);
controller = new MissionControlController(service);
});
it("Phase 1 gate: unified sessions endpoint returns internal provider sessions", async () => {
const internalSession: AgentSession = {
id: "session-internal-1",
providerId: "internal",
providerType: "internal",
status: "active",
createdAt: new Date("2026-03-07T20:00:00.000Z"),
updatedAt: new Date("2026-03-07T20:01:00.000Z"),
};
const externalSession: AgentSession = {
id: "session-openclaw-1",
providerId: "openclaw",
providerType: "external",
status: "active",
createdAt: new Date("2026-03-07T20:02:00.000Z"),
updatedAt: new Date("2026-03-07T20:03:00.000Z"),
};
registry.listAllSessions.mockResolvedValue([internalSession, externalSession]);
const response = await controller.listSessions();
expect(registry.listAllSessions).toHaveBeenCalledTimes(1);
expect(response.sessions).toEqual([internalSession, externalSession]);
expect(response.sessions).toContainEqual(
expect.objectContaining({
id: "session-internal-1",
providerId: "internal",
})
);
});
});

View File

@@ -0,0 +1,192 @@
import {
Body,
Controller,
Get,
Header,
HttpCode,
MessageEvent,
Param,
Post,
Query,
Request,
Sse,
UseGuards,
UsePipes,
ValidationPipe,
} from "@nestjs/common";
import type { AgentMessage, AgentSession, InjectResult } from "@mosaic/shared";
import { Observable } from "rxjs";
import { AuthGuard } from "../../auth/guards/auth.guard";
import { InjectAgentDto } from "../agents/dto/inject-agent.dto";
import { GetMissionControlAuditLogQueryDto } from "./dto/get-mission-control-audit-log-query.dto";
import { GetMissionControlMessagesQueryDto } from "./dto/get-mission-control-messages-query.dto";
import { KillSessionDto } from "./dto/kill-session.dto";
import { MissionControlService, type MissionControlAuditLogPage } from "./mission-control.service";
const DEFAULT_OPERATOR_ID = "mission-control";
interface MissionControlRequest {
user?: {
id?: string;
};
}
@Controller("api/mission-control")
@UseGuards(AuthGuard)
export class MissionControlController {
constructor(private readonly missionControlService: MissionControlService) {}
@Get("sessions")
async listSessions(): Promise<{ sessions: AgentSession[] }> {
const sessions = await this.missionControlService.listSessions();
return { sessions };
}
@Get("sessions/:sessionId")
getSession(@Param("sessionId") sessionId: string): Promise<AgentSession> {
return this.missionControlService.getSession(sessionId);
}
@Get("sessions/:sessionId/messages")
@UsePipes(new ValidationPipe({ transform: true, whitelist: true }))
async getMessages(
@Param("sessionId") sessionId: string,
@Query() query: GetMissionControlMessagesQueryDto
): Promise<{ messages: AgentMessage[] }> {
const messages = await this.missionControlService.getMessages(
sessionId,
query.limit,
query.before
);
return { messages };
}
@Get("audit-log")
@UsePipes(new ValidationPipe({ transform: true, whitelist: true }))
getAuditLog(
@Query() query: GetMissionControlAuditLogQueryDto
): Promise<MissionControlAuditLogPage> {
return this.missionControlService.getAuditLog(query.sessionId, query.page, query.limit);
}
@Post("sessions/:sessionId/inject")
@HttpCode(200)
@UsePipes(new ValidationPipe({ transform: true, whitelist: true }))
injectMessage(
@Param("sessionId") sessionId: string,
@Body() dto: InjectAgentDto,
@Request() req: MissionControlRequest
): Promise<InjectResult> {
return this.missionControlService.injectMessage(
sessionId,
dto.message,
this.resolveOperatorId(req)
);
}
@Post("sessions/:sessionId/pause")
@HttpCode(200)
async pauseSession(
@Param("sessionId") sessionId: string,
@Request() req: MissionControlRequest
): Promise<{ message: string }> {
await this.missionControlService.pauseSession(sessionId, this.resolveOperatorId(req));
return { message: `Session ${sessionId} paused` };
}
@Post("sessions/:sessionId/resume")
@HttpCode(200)
async resumeSession(
@Param("sessionId") sessionId: string,
@Request() req: MissionControlRequest
): Promise<{ message: string }> {
await this.missionControlService.resumeSession(sessionId, this.resolveOperatorId(req));
return { message: `Session ${sessionId} resumed` };
}
@Post("sessions/:sessionId/kill")
@HttpCode(200)
@UsePipes(new ValidationPipe({ transform: true, whitelist: true }))
async killSession(
@Param("sessionId") sessionId: string,
@Body() dto: KillSessionDto,
@Request() req: MissionControlRequest
): Promise<{ message: string }> {
await this.missionControlService.killSession(
sessionId,
dto.force ?? true,
this.resolveOperatorId(req)
);
return { message: `Session ${sessionId} killed` };
}
@Sse("sessions/:sessionId/stream")
@Header("Content-Type", "text/event-stream")
@Header("Cache-Control", "no-cache")
streamSessionMessages(@Param("sessionId") sessionId: string): Observable<MessageEvent> {
return new Observable<MessageEvent>((subscriber) => {
let isClosed = false;
let iterator: AsyncIterator<AgentMessage> | null = null;
void this.missionControlService
.streamMessages(sessionId)
.then(async (stream) => {
iterator = stream[Symbol.asyncIterator]();
for (;;) {
if (isClosed) {
break;
}
const next = (await iterator.next()) as { done: boolean; value: AgentMessage };
if (next.done) {
break;
}
subscriber.next({
data: this.toStreamPayload(next.value),
});
}
subscriber.complete();
})
.catch((error: unknown) => {
subscriber.error(error);
});
return () => {
isClosed = true;
void iterator?.return?.();
};
});
}
private resolveOperatorId(req: MissionControlRequest): string {
const operatorId = req.user?.id;
return typeof operatorId === "string" && operatorId.length > 0
? operatorId
: DEFAULT_OPERATOR_ID;
}
private toStreamPayload(message: AgentMessage): {
id: string;
sessionId: string;
role: string;
content: string;
timestamp: string;
metadata?: Record<string, unknown>;
} {
return {
id: message.id,
sessionId: message.sessionId,
role: message.role,
content: message.content,
timestamp: message.timestamp.toISOString(),
...(message.metadata !== undefined ? { metadata: message.metadata } : {}),
};
}
}

View File

@@ -0,0 +1,13 @@
import { Module } from "@nestjs/common";
import { AgentsModule } from "../agents/agents.module";
import { AuthModule } from "../../auth/auth.module";
import { PrismaModule } from "../../prisma/prisma.module";
import { MissionControlController } from "./mission-control.controller";
import { MissionControlService } from "./mission-control.service";
@Module({
imports: [AgentsModule, AuthModule, PrismaModule],
controllers: [MissionControlController],
providers: [MissionControlService],
})
export class MissionControlModule {}

View File

@@ -0,0 +1,213 @@
import { NotFoundException } from "@nestjs/common";
import { beforeEach, describe, expect, it, vi } from "vitest";
import type { AgentMessage, AgentSession, IAgentProvider, InjectResult } from "@mosaic/shared";
import type { PrismaService } from "../../prisma/prisma.service";
import { AgentProviderRegistry } from "../agents/agent-provider.registry";
import { MissionControlService } from "./mission-control.service";
type MockProvider = IAgentProvider & {
listSessions: ReturnType<typeof vi.fn>;
getSession: ReturnType<typeof vi.fn>;
getMessages: ReturnType<typeof vi.fn>;
injectMessage: ReturnType<typeof vi.fn>;
pauseSession: ReturnType<typeof vi.fn>;
resumeSession: ReturnType<typeof vi.fn>;
killSession: ReturnType<typeof vi.fn>;
streamMessages: ReturnType<typeof vi.fn>;
};
const emptyMessageStream = async function* (): AsyncIterable<AgentMessage> {
return;
};
const createProvider = (providerId = "internal"): MockProvider => ({
providerId,
providerType: providerId,
displayName: providerId,
listSessions: vi.fn().mockResolvedValue({ sessions: [], total: 0 }),
getSession: vi.fn().mockResolvedValue(null),
getMessages: vi.fn().mockResolvedValue([]),
injectMessage: vi.fn().mockResolvedValue({ accepted: true } as InjectResult),
pauseSession: vi.fn().mockResolvedValue(undefined),
resumeSession: vi.fn().mockResolvedValue(undefined),
killSession: vi.fn().mockResolvedValue(undefined),
streamMessages: vi.fn().mockReturnValue(emptyMessageStream()),
isAvailable: vi.fn().mockResolvedValue(true),
});
describe("MissionControlService", () => {
let service: MissionControlService;
let registry: {
listAllSessions: ReturnType<typeof vi.fn>;
getProviderForSession: ReturnType<typeof vi.fn>;
};
let prisma: {
operatorAuditLog: {
create: ReturnType<typeof vi.fn>;
};
};
const session: AgentSession = {
id: "session-1",
providerId: "internal",
providerType: "internal",
status: "active",
createdAt: new Date("2026-03-07T14:00:00.000Z"),
updatedAt: new Date("2026-03-07T14:01:00.000Z"),
};
beforeEach(() => {
registry = {
listAllSessions: vi.fn().mockResolvedValue([session]),
getProviderForSession: vi.fn().mockResolvedValue(null),
};
prisma = {
operatorAuditLog: {
create: vi.fn().mockResolvedValue(undefined),
},
};
service = new MissionControlService(
registry as unknown as AgentProviderRegistry,
prisma as unknown as PrismaService
);
});
it("lists sessions from the registry", async () => {
await expect(service.listSessions()).resolves.toEqual([session]);
expect(registry.listAllSessions).toHaveBeenCalledTimes(1);
});
it("returns a session when it is found", async () => {
const provider = createProvider("internal");
registry.getProviderForSession.mockResolvedValue({ provider, session });
await expect(service.getSession(session.id)).resolves.toEqual(session);
});
it("throws NotFoundException when session lookup fails", async () => {
await expect(service.getSession("missing-session")).rejects.toBeInstanceOf(NotFoundException);
});
it("gets messages from the resolved provider", async () => {
const provider = createProvider("openclaw");
const messages: AgentMessage[] = [
{
id: "message-1",
sessionId: session.id,
role: "assistant",
content: "hello",
timestamp: new Date("2026-03-07T14:01:00.000Z"),
},
];
provider.getMessages.mockResolvedValue(messages);
registry.getProviderForSession.mockResolvedValue({ provider, session });
await expect(service.getMessages(session.id, 25, "10")).resolves.toEqual(messages);
expect(provider.getMessages).toHaveBeenCalledWith(session.id, 25, "10");
});
it("injects a message and writes an audit log", async () => {
const provider = createProvider("internal");
const injectResult: InjectResult = { accepted: true, messageId: "msg-1" };
provider.injectMessage.mockResolvedValue(injectResult);
registry.getProviderForSession.mockResolvedValue({ provider, session });
await expect(service.injectMessage(session.id, "ship it", "operator-1")).resolves.toEqual(
injectResult
);
expect(provider.injectMessage).toHaveBeenCalledWith(session.id, "ship it");
expect(prisma.operatorAuditLog.create).toHaveBeenCalledWith({
data: {
sessionId: session.id,
userId: "operator-1",
provider: "internal",
action: "inject",
content: "ship it",
metadata: {
payload: { message: "ship it" },
},
},
});
});
it("pauses and resumes using default operator id", async () => {
const provider = createProvider("openclaw");
registry.getProviderForSession.mockResolvedValue({ provider, session });
await service.pauseSession(session.id);
await service.resumeSession(session.id);
expect(provider.pauseSession).toHaveBeenCalledWith(session.id);
expect(provider.resumeSession).toHaveBeenCalledWith(session.id);
expect(prisma.operatorAuditLog.create).toHaveBeenNthCalledWith(1, {
data: {
sessionId: session.id,
userId: "mission-control",
provider: "openclaw",
action: "pause",
metadata: {
payload: {},
},
},
});
expect(prisma.operatorAuditLog.create).toHaveBeenNthCalledWith(2, {
data: {
sessionId: session.id,
userId: "mission-control",
provider: "openclaw",
action: "resume",
metadata: {
payload: {},
},
},
});
});
it("kills with provided force value and writes audit log", async () => {
const provider = createProvider("openclaw");
registry.getProviderForSession.mockResolvedValue({ provider, session });
await service.killSession(session.id, false, "operator-2");
expect(provider.killSession).toHaveBeenCalledWith(session.id, false);
expect(prisma.operatorAuditLog.create).toHaveBeenCalledWith({
data: {
sessionId: session.id,
userId: "operator-2",
provider: "openclaw",
action: "kill",
metadata: {
payload: { force: false },
},
},
});
});
it("resolves provider message stream", async () => {
const provider = createProvider("internal");
const messageStream = (async function* (): AsyncIterable<AgentMessage> {
yield {
id: "message-1",
sessionId: session.id,
role: "assistant",
content: "stream",
timestamp: new Date("2026-03-07T14:03:00.000Z"),
};
})();
provider.streamMessages.mockReturnValue(messageStream);
registry.getProviderForSession.mockResolvedValue({ provider, session });
await expect(service.streamMessages(session.id)).resolves.toBe(messageStream);
expect(provider.streamMessages).toHaveBeenCalledWith(session.id);
});
it("does not write audit log when session cannot be resolved", async () => {
await expect(service.pauseSession("missing-session")).rejects.toBeInstanceOf(NotFoundException);
expect(prisma.operatorAuditLog.create).not.toHaveBeenCalled();
});
});

View File

@@ -0,0 +1,186 @@
import { Injectable, NotFoundException } from "@nestjs/common";
import type { AgentMessage, AgentSession, IAgentProvider, InjectResult } from "@mosaic/shared";
import type { Prisma } from "@prisma/client";
import { PrismaService } from "../../prisma/prisma.service";
import { AgentProviderRegistry } from "../agents/agent-provider.registry";
type MissionControlAction = "inject" | "pause" | "resume" | "kill";
const DEFAULT_OPERATOR_ID = "mission-control";
export interface AuditLogEntry {
id: string;
userId: string;
sessionId: string;
provider: string;
action: string;
content: string | null;
metadata: Prisma.JsonValue;
createdAt: Date;
}
export interface MissionControlAuditLogPage {
items: AuditLogEntry[];
total: number;
page: number;
pages: number;
}
@Injectable()
export class MissionControlService {
constructor(
private readonly registry: AgentProviderRegistry,
private readonly prisma: PrismaService
) {}
listSessions(): Promise<AgentSession[]> {
return this.registry.listAllSessions();
}
async getSession(sessionId: string): Promise<AgentSession> {
const resolved = await this.registry.getProviderForSession(sessionId);
if (!resolved) {
throw new NotFoundException(`Session ${sessionId} not found`);
}
return resolved.session;
}
async getMessages(sessionId: string, limit?: number, before?: string): Promise<AgentMessage[]> {
const { provider } = await this.getProviderForSessionOrThrow(sessionId);
return provider.getMessages(sessionId, limit, before);
}
async getAuditLog(
sessionId: string | undefined,
page: number,
limit: number
): Promise<MissionControlAuditLogPage> {
const normalizedSessionId = sessionId?.trim();
const where: Prisma.OperatorAuditLogWhereInput =
normalizedSessionId && normalizedSessionId.length > 0
? { sessionId: normalizedSessionId }
: {};
const [total, items] = await this.prisma.$transaction([
this.prisma.operatorAuditLog.count({ where }),
this.prisma.operatorAuditLog.findMany({
where,
orderBy: { createdAt: "desc" },
skip: (page - 1) * limit,
take: limit,
}),
]);
return {
items,
total,
page,
pages: total === 0 ? 0 : Math.ceil(total / limit),
};
}
async injectMessage(
sessionId: string,
message: string,
operatorId = DEFAULT_OPERATOR_ID
): Promise<InjectResult> {
const { provider } = await this.getProviderForSessionOrThrow(sessionId);
const result = await provider.injectMessage(sessionId, message);
await this.writeOperatorAuditLog({
sessionId,
providerId: provider.providerId,
operatorId,
action: "inject",
content: message,
payload: { message },
});
return result;
}
async pauseSession(sessionId: string, operatorId = DEFAULT_OPERATOR_ID): Promise<void> {
const { provider } = await this.getProviderForSessionOrThrow(sessionId);
await provider.pauseSession(sessionId);
await this.writeOperatorAuditLog({
sessionId,
providerId: provider.providerId,
operatorId,
action: "pause",
payload: {},
});
}
async resumeSession(sessionId: string, operatorId = DEFAULT_OPERATOR_ID): Promise<void> {
const { provider } = await this.getProviderForSessionOrThrow(sessionId);
await provider.resumeSession(sessionId);
await this.writeOperatorAuditLog({
sessionId,
providerId: provider.providerId,
operatorId,
action: "resume",
payload: {},
});
}
async killSession(
sessionId: string,
force = true,
operatorId = DEFAULT_OPERATOR_ID
): Promise<void> {
const { provider } = await this.getProviderForSessionOrThrow(sessionId);
await provider.killSession(sessionId, force);
await this.writeOperatorAuditLog({
sessionId,
providerId: provider.providerId,
operatorId,
action: "kill",
payload: { force },
});
}
async streamMessages(sessionId: string): Promise<AsyncIterable<AgentMessage>> {
const { provider } = await this.getProviderForSessionOrThrow(sessionId);
return provider.streamMessages(sessionId);
}
private async getProviderForSessionOrThrow(
sessionId: string
): Promise<{ provider: IAgentProvider; session: AgentSession }> {
const resolved = await this.registry.getProviderForSession(sessionId);
if (!resolved) {
throw new NotFoundException(`Session ${sessionId} not found`);
}
return resolved;
}
private toJsonValue(value: Record<string, unknown>): Prisma.InputJsonValue {
return value as Prisma.InputJsonValue;
}
private async writeOperatorAuditLog(params: {
sessionId: string;
providerId: string;
operatorId: string;
action: MissionControlAction;
content?: string;
payload: Record<string, unknown>;
}): Promise<void> {
await this.prisma.operatorAuditLog.create({
data: {
sessionId: params.sessionId,
userId: params.operatorId,
provider: params.providerId,
action: params.action,
...(params.content !== undefined ? { content: params.content } : {}),
metadata: this.toJsonValue({ payload: params.payload }),
},
});
}
}

View File

@@ -0,0 +1,145 @@
import type { HttpService } from "@nestjs/axios";
import type { AgentMessage } from "@mosaic/shared";
import { Readable } from "node:stream";
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { OpenClawSseBridge } from "./openclaw-sse.bridge";
describe("OpenClawSseBridge", () => {
let bridge: OpenClawSseBridge;
let httpService: {
axiosRef: {
get: ReturnType<typeof vi.fn>;
};
};
beforeEach(() => {
httpService = {
axiosRef: {
get: vi.fn(),
},
};
bridge = new OpenClawSseBridge(httpService as unknown as HttpService);
});
afterEach(() => {
vi.useRealTimers();
});
it("maps message and status events, and skips heartbeats", async () => {
httpService.axiosRef.get.mockResolvedValue({
data: Readable.from([
'event: message\ndata: {"id":"msg-1","role":"assistant","content":"hello","timestamp":"2026-03-07T16:00:00.000Z"}\n\n',
"event: heartbeat\ndata: {}\n\n",
'event: status\ndata: {"status":"paused","timestamp":"2026-03-07T16:00:01.000Z"}\n\n',
"data: [DONE]\n\n",
]),
});
const messages = await collectMessages(
bridge.streamSession("https://gateway.example.com/", "session-1", {
Authorization: "Bearer test-token",
})
);
expect(httpService.axiosRef.get).toHaveBeenCalledWith(
"https://gateway.example.com/api/sessions/session-1/stream",
{
headers: {
Authorization: "Bearer test-token",
Accept: "text/event-stream",
},
responseType: "stream",
}
);
expect(messages).toHaveLength(2);
expect(messages[0]).toEqual({
id: "msg-1",
sessionId: "session-1",
role: "assistant",
content: "hello",
timestamp: new Date("2026-03-07T16:00:00.000Z"),
});
expect(messages[1]).toEqual({
id: expect.any(String),
sessionId: "session-1",
role: "system",
content: "Session status changed to paused",
timestamp: new Date("2026-03-07T16:00:01.000Z"),
metadata: {
status: "paused",
timestamp: "2026-03-07T16:00:01.000Z",
},
});
});
it("retries after disconnect and resumes streaming", async () => {
vi.useFakeTimers();
httpService.axiosRef.get
.mockResolvedValueOnce({
data: Readable.from([
'event: message\ndata: {"id":"msg-1","content":"first","timestamp":"2026-03-07T16:10:00.000Z"}\n\n',
]),
})
.mockResolvedValueOnce({
data: Readable.from(["data: [DONE]\n\n"]),
});
const consumePromise = collectMessages(
bridge.streamSession("https://gateway.example.com", "session-1", {
Authorization: "Bearer test-token",
})
);
await vi.advanceTimersByTimeAsync(2000);
const messages = await consumePromise;
expect(httpService.axiosRef.get).toHaveBeenCalledTimes(2);
expect(messages).toEqual([
{
id: "msg-1",
sessionId: "session-1",
role: "user",
content: "first",
timestamp: new Date("2026-03-07T16:10:00.000Z"),
},
]);
});
it("throws after exhausting reconnect retries", async () => {
vi.useFakeTimers();
httpService.axiosRef.get.mockRejectedValue(new Error("socket closed"));
const consumePromise = collectMessages(
bridge.streamSession("https://gateway.example.com", "session-1", {
Authorization: "Bearer test-token",
})
);
const rejection = expect(consumePromise).rejects.toThrow(
"Failed to reconnect OpenClaw stream for session session-1 after 5 retries: socket closed"
);
for (let attempt = 0; attempt < 5; attempt += 1) {
await vi.advanceTimersByTimeAsync(2000);
}
await rejection;
expect(httpService.axiosRef.get).toHaveBeenCalledTimes(6);
});
});
async function collectMessages(stream: AsyncIterable<AgentMessage>): Promise<AgentMessage[]> {
const messages: AgentMessage[] = [];
for await (const message of stream) {
messages.push(message);
}
return messages;
}

View File

@@ -0,0 +1,420 @@
import { HttpService } from "@nestjs/axios";
import { Injectable } from "@nestjs/common";
import type { AgentMessage, AgentMessageRole } from "@mosaic/shared";
import { randomUUID } from "node:crypto";
const STREAM_RETRY_DELAY_MS = 2000;
const STREAM_MAX_RETRIES = 5;
type JsonRecord = Record<string, unknown>;
type AsyncChunkStream = AsyncIterable<string | Uint8Array | Buffer>;
type ParsedStreamEvent =
| {
type: "message";
message: AgentMessage;
}
| {
type: "done";
};
@Injectable()
export class OpenClawSseBridge {
constructor(private readonly httpService: HttpService) {}
async *streamSession(
baseUrl: string,
sessionId: string,
headers: Record<string, string>
): AsyncIterable<AgentMessage> {
let retryCount = 0;
let lastError: unknown = new Error("OpenClaw stream disconnected");
while (retryCount <= STREAM_MAX_RETRIES) {
try {
const response = await this.httpService.axiosRef.get(
this.buildStreamUrl(baseUrl, sessionId),
{
headers: {
...headers,
Accept: "text/event-stream",
},
responseType: "stream",
}
);
const stream = this.asAsyncChunkStream(response.data);
if (stream === null) {
throw new Error("OpenClaw stream response is not readable");
}
retryCount = 0;
let streamCompleted = false;
for await (const event of this.parseStream(stream, sessionId)) {
if (event.type === "done") {
streamCompleted = true;
break;
}
yield event.message;
}
if (streamCompleted) {
return;
}
lastError = new Error("OpenClaw stream disconnected");
} catch (error) {
lastError = error;
}
if (retryCount >= STREAM_MAX_RETRIES) {
throw new Error(
`Failed to reconnect OpenClaw stream for session ${sessionId} after ${String(STREAM_MAX_RETRIES)} retries: ${this.toErrorMessage(lastError)}`
);
}
retryCount += 1;
await this.delay(STREAM_RETRY_DELAY_MS);
}
}
private async *parseStream(
stream: AsyncChunkStream,
sessionId: string
): AsyncGenerator<ParsedStreamEvent> {
const decoder = new TextDecoder();
let buffer = "";
for await (const chunk of stream) {
const textChunk = typeof chunk === "string" ? chunk : decoder.decode(chunk, { stream: true });
buffer += textChunk.replace(/\r\n/gu, "\n");
const rawEvents = buffer.split("\n\n");
buffer = rawEvents.pop() ?? "";
for (const rawEvent of rawEvents) {
const parsedEvent = this.parseRawEvent(rawEvent);
if (parsedEvent === null) {
continue;
}
if (parsedEvent.data === "[DONE]") {
yield {
type: "done",
};
return;
}
const payload = this.tryParseJson(parsedEvent.data) ?? parsedEvent.data;
const message = this.mapEventToMessage(parsedEvent.type, payload, sessionId);
if (message !== null) {
yield {
type: "message",
message,
};
}
}
}
buffer += decoder.decode();
const trailingEvent = this.parseRawEvent(buffer.trim());
if (trailingEvent === null) {
return;
}
if (trailingEvent.data === "[DONE]") {
yield {
type: "done",
};
return;
}
const payload = this.tryParseJson(trailingEvent.data) ?? trailingEvent.data;
const message = this.mapEventToMessage(trailingEvent.type, payload, sessionId);
if (message !== null) {
yield {
type: "message",
message,
};
}
}
private parseRawEvent(rawEvent: string): { type: string; data: string } | null {
if (rawEvent.trim().length === 0) {
return null;
}
let type = "message";
const dataLines: string[] = [];
for (const line of rawEvent.split("\n")) {
const trimmedLine = line.trimEnd();
if (trimmedLine.length === 0 || trimmedLine.startsWith(":")) {
continue;
}
if (trimmedLine.startsWith("event:")) {
type = trimmedLine.slice(6).trim().toLowerCase();
continue;
}
if (trimmedLine.startsWith("data:")) {
dataLines.push(trimmedLine.slice(5).trimStart());
}
}
if (dataLines.length > 0) {
return {
type,
data: dataLines.join("\n").trim(),
};
}
const trimmedEvent = rawEvent.trim();
if (trimmedEvent.startsWith("{") || trimmedEvent.startsWith("[")) {
return {
type,
data: trimmedEvent,
};
}
return null;
}
private mapEventToMessage(
eventType: string,
payload: unknown,
fallbackSessionId: string
): AgentMessage | null {
switch (eventType) {
case "heartbeat":
return null;
case "status":
return this.toStatusMessage(payload, fallbackSessionId);
case "message":
default:
return this.toAgentMessage(payload, fallbackSessionId);
}
}
private toStatusMessage(value: unknown, sessionId: string): AgentMessage | null {
if (typeof value === "string") {
const status = value.trim();
if (status.length === 0) {
return null;
}
return {
id: randomUUID(),
sessionId,
role: "system",
content: `Session status changed to ${status}`,
timestamp: new Date(),
metadata: {
status,
},
};
}
if (!this.isRecord(value)) {
return null;
}
const status = this.readString(value.status);
if (!status) {
return null;
}
return {
id: randomUUID(),
sessionId,
role: "system",
content: `Session status changed to ${status}`,
timestamp: this.parseDate(value.timestamp ?? value.updatedAt),
metadata: value,
};
}
private toAgentMessage(value: unknown, fallbackSessionId: string): AgentMessage | null {
if (typeof value === "string") {
const content = value.trim();
if (content.length === 0) {
return null;
}
return {
id: randomUUID(),
sessionId: fallbackSessionId,
role: "assistant",
content,
timestamp: new Date(),
};
}
let candidate: JsonRecord | null = null;
if (this.isRecord(value) && this.isRecord(value.message)) {
candidate = value.message;
} else if (this.isRecord(value)) {
candidate = value;
}
if (candidate === null) {
return null;
}
const sessionId = this.readString(candidate.sessionId) ?? fallbackSessionId;
if (!sessionId) {
return null;
}
const content = this.extractMessageContent(
candidate.content ?? candidate.text ?? candidate.message
);
if (content.length === 0) {
return null;
}
const metadata = this.toMetadata(candidate.metadata);
return {
id: this.readString(candidate.id) ?? this.readString(candidate.messageId) ?? randomUUID(),
sessionId,
role: this.toMessageRole(this.readString(candidate.role) ?? this.readString(candidate.type)),
content,
timestamp: this.parseDate(candidate.timestamp ?? candidate.createdAt),
...(metadata !== undefined ? { metadata } : {}),
};
}
private extractMessageContent(content: unknown): string {
if (typeof content === "string") {
return content.trim();
}
if (Array.isArray(content)) {
const parts: string[] = [];
for (const part of content) {
if (typeof part === "string") {
const trimmed = part.trim();
if (trimmed.length > 0) {
parts.push(trimmed);
}
continue;
}
if (!this.isRecord(part)) {
continue;
}
const text = this.readString(part.text) ?? this.readString(part.content);
if (text !== undefined && text.trim().length > 0) {
parts.push(text.trim());
}
}
return parts.join("\n\n").trim();
}
if (this.isRecord(content)) {
const text = this.readString(content.text) ?? this.readString(content.content);
return text?.trim() ?? "";
}
return "";
}
private toMessageRole(role?: string): AgentMessageRole {
switch (role?.toLowerCase()) {
case "assistant":
case "agent":
return "assistant";
case "system":
return "system";
case "tool":
return "tool";
case "operator":
case "user":
default:
return "user";
}
}
private parseDate(value: unknown, fallback = new Date()): Date {
if (value instanceof Date) {
return value;
}
if (typeof value === "string" || typeof value === "number") {
const parsed = new Date(value);
if (!Number.isNaN(parsed.getTime())) {
return parsed;
}
}
return fallback;
}
private toMetadata(value: unknown): Record<string, unknown> | undefined {
if (this.isRecord(value)) {
return value;
}
return undefined;
}
private buildStreamUrl(baseUrl: string, sessionId: string): string {
const normalizedBaseUrl = baseUrl.replace(/\/$/u, "");
return new URL(
`/api/sessions/${encodeURIComponent(sessionId)}/stream`,
`${normalizedBaseUrl}/`
).toString();
}
private tryParseJson(value: string): unknown {
try {
return JSON.parse(value) as unknown;
} catch {
return null;
}
}
private asAsyncChunkStream(value: unknown): AsyncChunkStream | null {
if (value !== null && typeof value === "object" && Symbol.asyncIterator in value) {
return value as AsyncChunkStream;
}
return null;
}
private isRecord(value: unknown): value is JsonRecord {
return typeof value === "object" && value !== null && !Array.isArray(value);
}
private readString(value: unknown): string | undefined {
if (typeof value !== "string") {
return undefined;
}
const trimmed = value.trim();
return trimmed.length > 0 ? trimmed : undefined;
}
private async delay(ms: number): Promise<void> {
await new Promise((resolve) => {
setTimeout(resolve, ms);
});
}
private toErrorMessage(error: unknown): string {
if (error instanceof Error) {
return error.message;
}
return String(error);
}
}

View File

@@ -0,0 +1,24 @@
import { HttpService } from "@nestjs/axios";
import { Injectable } from "@nestjs/common";
import type { AgentProviderConfig } from "@prisma/client";
import { EncryptionService } from "../../../security/encryption.service";
import { OpenClawSseBridge } from "./openclaw-sse.bridge";
import { OpenClawProvider } from "./openclaw.provider";
@Injectable()
export class OpenClawProviderFactory {
constructor(
private readonly encryptionService: EncryptionService,
private readonly httpService: HttpService,
private readonly openClawSseBridge: OpenClawSseBridge
) {}
createProvider(config: AgentProviderConfig): OpenClawProvider {
return new OpenClawProvider(
config,
this.encryptionService,
this.httpService,
this.openClawSseBridge
);
}
}

View File

@@ -0,0 +1,271 @@
import type { HttpService } from "@nestjs/axios";
import { ServiceUnavailableException } from "@nestjs/common";
import type { AgentProviderConfig } from "@prisma/client";
import { beforeEach, describe, expect, it, vi } from "vitest";
import { EncryptionService } from "../../../security/encryption.service";
import { OpenClawSseBridge } from "./openclaw-sse.bridge";
import { OpenClawProvider } from "./openclaw.provider";
describe("OpenClawProvider", () => {
let provider: OpenClawProvider;
let httpService: {
axiosRef: {
get: ReturnType<typeof vi.fn>;
post: ReturnType<typeof vi.fn>;
};
};
let encryptionService: {
decryptIfNeeded: ReturnType<typeof vi.fn>;
};
let sseBridge: {
streamSession: ReturnType<typeof vi.fn>;
};
const config: AgentProviderConfig = {
id: "cfg-openclaw-1",
workspaceId: "workspace-1",
name: "openclaw-home",
provider: "openclaw",
gatewayUrl: "https://gateway.example.com/",
credentials: {
apiToken: "enc:token-value",
displayName: "Home OpenClaw",
},
isActive: true,
createdAt: new Date("2026-03-07T15:00:00.000Z"),
updatedAt: new Date("2026-03-07T15:00:00.000Z"),
};
beforeEach(() => {
httpService = {
axiosRef: {
get: vi.fn(),
post: vi.fn(),
},
};
encryptionService = {
decryptIfNeeded: vi.fn().mockReturnValue("plain-token"),
};
sseBridge = {
streamSession: vi.fn(),
};
provider = new OpenClawProvider(
config,
encryptionService as unknown as EncryptionService,
httpService as unknown as HttpService,
sseBridge as unknown as OpenClawSseBridge
);
});
it("maps listSessions from OpenClaw API", async () => {
httpService.axiosRef.get.mockResolvedValue({
data: {
sessions: [
{
id: "session-1",
status: "running",
createdAt: "2026-03-07T15:01:00.000Z",
updatedAt: "2026-03-07T15:02:00.000Z",
},
],
total: 1,
cursor: "next-cursor",
},
});
const result = await provider.listSessions("cursor-1", 25);
expect(httpService.axiosRef.get).toHaveBeenCalledWith(
"https://gateway.example.com/api/sessions",
{
headers: {
Authorization: "Bearer plain-token",
},
params: {
cursor: "cursor-1",
limit: 25,
},
}
);
expect(result).toEqual({
sessions: [
{
id: "session-1",
providerId: "openclaw-home",
providerType: "openclaw",
status: "active",
createdAt: new Date("2026-03-07T15:01:00.000Z"),
updatedAt: new Date("2026-03-07T15:02:00.000Z"),
},
],
total: 1,
cursor: "next-cursor",
});
expect(encryptionService.decryptIfNeeded).toHaveBeenCalledWith("enc:token-value");
});
it("returns null from getSession when OpenClaw returns 404", async () => {
httpService.axiosRef.get.mockRejectedValue({
response: {
status: 404,
},
});
await expect(provider.getSession("missing-session")).resolves.toBeNull();
});
it("maps getMessages response", async () => {
httpService.axiosRef.get.mockResolvedValue({
data: {
messages: [
{
id: "message-1",
sessionId: "session-1",
role: "agent",
content: "hello",
timestamp: "2026-03-07T15:03:00.000Z",
metadata: {
tokens: 128,
},
},
],
},
});
const result = await provider.getMessages("session-1", 20, "before-cursor");
expect(httpService.axiosRef.get).toHaveBeenCalledWith(
"https://gateway.example.com/api/messages",
{
headers: {
Authorization: "Bearer plain-token",
},
params: {
sessionId: "session-1",
limit: 20,
before: "before-cursor",
},
}
);
expect(result).toEqual([
{
id: "message-1",
sessionId: "session-1",
role: "assistant",
content: "hello",
timestamp: new Date("2026-03-07T15:03:00.000Z"),
metadata: {
tokens: 128,
},
},
]);
});
it("maps inject and control endpoints", async () => {
httpService.axiosRef.post
.mockResolvedValueOnce({
data: {
accepted: true,
messageId: "message-2",
},
})
.mockResolvedValueOnce({ data: {} })
.mockResolvedValueOnce({ data: {} })
.mockResolvedValueOnce({ data: {} });
await expect(provider.injectMessage("session-1", "barge in")).resolves.toEqual({
accepted: true,
messageId: "message-2",
});
await provider.pauseSession("session-1");
await provider.resumeSession("session-1");
await provider.killSession("session-1", false);
expect(httpService.axiosRef.post).toHaveBeenNthCalledWith(
1,
"https://gateway.example.com/api/sessions/session-1/inject",
{ content: "barge in" },
{
headers: {
Authorization: "Bearer plain-token",
},
}
);
expect(httpService.axiosRef.post).toHaveBeenNthCalledWith(
2,
"https://gateway.example.com/api/sessions/session-1/pause",
{},
{
headers: {
Authorization: "Bearer plain-token",
},
}
);
expect(httpService.axiosRef.post).toHaveBeenNthCalledWith(
3,
"https://gateway.example.com/api/sessions/session-1/resume",
{},
{
headers: {
Authorization: "Bearer plain-token",
},
}
);
expect(httpService.axiosRef.post).toHaveBeenNthCalledWith(
4,
"https://gateway.example.com/api/sessions/session-1/kill",
{ force: false },
{
headers: {
Authorization: "Bearer plain-token",
},
}
);
});
it("delegates streaming to OpenClawSseBridge", async () => {
const streamedMessage = {
id: "message-stream",
sessionId: "session-stream",
role: "assistant",
content: "stream hello",
timestamp: new Date("2026-03-07T16:00:00.000Z"),
};
sseBridge.streamSession.mockReturnValue(
(async function* () {
yield streamedMessage;
})()
);
const messages: Array<unknown> = [];
for await (const message of provider.streamMessages("session-stream")) {
messages.push(message);
}
expect(sseBridge.streamSession).toHaveBeenCalledWith(
"https://gateway.example.com",
"session-stream",
{
Authorization: "Bearer plain-token",
}
);
expect(messages).toEqual([streamedMessage]);
});
it("throws ServiceUnavailableException for request failures", async () => {
httpService.axiosRef.get.mockRejectedValue(new Error("gateway unreachable"));
await expect(provider.listSessions()).rejects.toBeInstanceOf(ServiceUnavailableException);
});
it("returns false from isAvailable when gateway check fails", async () => {
httpService.axiosRef.get.mockRejectedValue(new Error("gateway unreachable"));
await expect(provider.isAvailable()).resolves.toBe(false);
});
});

View File

@@ -0,0 +1,613 @@
import { HttpService } from "@nestjs/axios";
import { Injectable, ServiceUnavailableException } from "@nestjs/common";
import type {
AgentMessage,
AgentMessageRole,
AgentSession,
AgentSessionList,
AgentSessionStatus,
IAgentProvider,
InjectResult,
} from "@mosaic/shared";
import type { AgentProviderConfig } from "@prisma/client";
import { randomUUID } from "node:crypto";
import { EncryptionService } from "../../../security/encryption.service";
import { OpenClawSseBridge } from "./openclaw-sse.bridge";
const DEFAULT_SESSION_LIMIT = 50;
const DEFAULT_MESSAGE_LIMIT = 50;
const MAX_MESSAGE_LIMIT = 200;
const OPENCLAW_PROVIDER_TYPE = "openclaw";
const API_TOKEN_KEYS = ["apiToken", "token", "bearerToken"] as const;
const DISPLAY_NAME_KEYS = ["displayName", "label"] as const;
type JsonRecord = Record<string, unknown>;
interface HttpErrorWithResponse {
response?: {
status?: number;
};
}
@Injectable()
export class OpenClawProvider implements IAgentProvider {
readonly providerId: string;
readonly providerType = OPENCLAW_PROVIDER_TYPE;
readonly displayName: string;
constructor(
private readonly config: AgentProviderConfig,
private readonly encryptionService: EncryptionService,
private readonly httpService: HttpService,
private readonly sseBridge: OpenClawSseBridge
) {
this.providerId = this.config.name;
this.displayName = this.resolveDisplayName();
}
validateBaseUrl(): void {
void this.resolveBaseUrl();
}
validateToken(): void {
void this.resolveApiToken();
}
async listSessions(cursor?: string, limit = DEFAULT_SESSION_LIMIT): Promise<AgentSessionList> {
const safeLimit = this.normalizeLimit(limit, DEFAULT_SESSION_LIMIT);
const params: Record<string, number | string> = { limit: safeLimit };
if (typeof cursor === "string" && cursor.length > 0) {
params.cursor = cursor;
}
try {
const response = await this.httpService.axiosRef.get(this.buildUrl("/api/sessions"), {
headers: this.authHeaders(),
params,
});
const page = this.extractSessionPage(response.data);
const sessions = page.records
.map((record) => this.toAgentSession(record))
.filter((session): session is AgentSession => session !== null);
return {
sessions,
total: page.total ?? sessions.length,
...(page.cursor !== undefined ? { cursor: page.cursor } : {}),
};
} catch (error) {
throw this.toServiceUnavailable("list sessions", error);
}
}
async getSession(sessionId: string): Promise<AgentSession | null> {
try {
const response = await this.httpService.axiosRef.get(
this.buildUrl(`/api/sessions/${encodeURIComponent(sessionId)}`),
{
headers: this.authHeaders(),
}
);
const payload = this.unwrapContainer(response.data, ["session", "data"]);
return this.toAgentSession(payload);
} catch (error) {
if (this.getHttpStatus(error) === 404) {
return null;
}
throw this.toServiceUnavailable(`get session ${sessionId}`, error);
}
}
async getMessages(
sessionId: string,
limit = DEFAULT_MESSAGE_LIMIT,
before?: string
): Promise<AgentMessage[]> {
const safeLimit = this.normalizeLimit(limit, DEFAULT_MESSAGE_LIMIT);
const params: Record<string, number | string> = {
sessionId,
limit: safeLimit,
};
if (typeof before === "string" && before.length > 0) {
params.before = before;
}
try {
const response = await this.httpService.axiosRef.get(this.buildUrl("/api/messages"), {
headers: this.authHeaders(),
params,
});
return this.extractMessageRecords(response.data)
.map((record) => this.toAgentMessage(record, sessionId))
.filter((message): message is AgentMessage => message !== null);
} catch (error) {
throw this.toServiceUnavailable(`get messages for session ${sessionId}`, error);
}
}
async injectMessage(sessionId: string, content: string): Promise<InjectResult> {
try {
const response = await this.httpService.axiosRef.post(
this.buildUrl(`/api/sessions/${encodeURIComponent(sessionId)}/inject`),
{ content },
{
headers: this.authHeaders(),
}
);
const payload = this.isRecord(response.data) ? response.data : {};
return {
accepted: typeof payload.accepted === "boolean" ? payload.accepted : true,
...(this.readString(payload.messageId) !== undefined
? { messageId: this.readString(payload.messageId) }
: {}),
};
} catch (error) {
throw this.toServiceUnavailable(`inject message into session ${sessionId}`, error);
}
}
async pauseSession(sessionId: string): Promise<void> {
try {
await this.httpService.axiosRef.post(
this.buildUrl(`/api/sessions/${encodeURIComponent(sessionId)}/pause`),
{},
{
headers: this.authHeaders(),
}
);
} catch (error) {
throw this.toServiceUnavailable(`pause session ${sessionId}`, error);
}
}
async resumeSession(sessionId: string): Promise<void> {
try {
await this.httpService.axiosRef.post(
this.buildUrl(`/api/sessions/${encodeURIComponent(sessionId)}/resume`),
{},
{
headers: this.authHeaders(),
}
);
} catch (error) {
throw this.toServiceUnavailable(`resume session ${sessionId}`, error);
}
}
async killSession(sessionId: string, force = true): Promise<void> {
try {
await this.httpService.axiosRef.post(
this.buildUrl(`/api/sessions/${encodeURIComponent(sessionId)}/kill`),
{ force },
{
headers: this.authHeaders(),
}
);
} catch (error) {
throw this.toServiceUnavailable(`kill session ${sessionId}`, error);
}
}
async *streamMessages(sessionId: string): AsyncIterable<AgentMessage> {
try {
yield* this.sseBridge.streamSession(this.resolveBaseUrl(), sessionId, this.authHeaders());
} catch (error) {
throw this.toServiceUnavailable(`stream messages for session ${sessionId}`, error);
}
}
async isAvailable(): Promise<boolean> {
try {
this.validateBaseUrl();
this.validateToken();
await this.httpService.axiosRef.get(this.buildUrl("/api/sessions"), {
headers: this.authHeaders(),
params: { limit: 1 },
});
return true;
} catch {
return false;
}
}
private extractSessionPage(payload: unknown): {
records: unknown[];
total?: number;
cursor?: string;
} {
if (Array.isArray(payload)) {
return {
records: payload,
total: payload.length,
};
}
if (!this.isRecord(payload)) {
return {
records: [],
};
}
let records: unknown[] = [];
if (Array.isArray(payload.sessions)) {
records = payload.sessions;
} else if (Array.isArray(payload.items)) {
records = payload.items;
} else if (Array.isArray(payload.data)) {
records = payload.data;
}
const total = typeof payload.total === "number" ? payload.total : undefined;
const cursor = this.readString(payload.cursor) ?? this.readString(payload.nextCursor);
return {
records,
total,
...(cursor !== undefined ? { cursor } : {}),
};
}
private extractMessageRecords(payload: unknown): unknown[] {
if (Array.isArray(payload)) {
return payload;
}
if (!this.isRecord(payload)) {
return [];
}
if (Array.isArray(payload.messages)) {
return payload.messages;
}
if (Array.isArray(payload.items)) {
return payload.items;
}
if (Array.isArray(payload.data)) {
return payload.data;
}
return [];
}
private unwrapContainer(payload: unknown, keys: string[]): unknown {
if (!this.isRecord(payload)) {
return payload;
}
for (const key of keys) {
if (key in payload) {
return payload[key];
}
}
return payload;
}
private toAgentSession(record: unknown): AgentSession | null {
if (!this.isRecord(record)) {
return null;
}
const id =
this.readString(record.id) ??
this.readString(record.sessionId) ??
this.readString(record.key);
if (!id) {
return null;
}
const createdAt = this.parseDate(record.createdAt ?? record.spawnedAt ?? record.startedAt);
const updatedAt = this.parseDate(
record.updatedAt ?? record.completedAt ?? record.lastActivityAt ?? record.endedAt,
createdAt
);
const label =
this.readString(record.label) ??
this.readString(record.title) ??
this.readString(record.name) ??
undefined;
const parentSessionId = this.readString(record.parentSessionId) ?? undefined;
const metadata = this.toMetadata(record.metadata);
return {
id,
providerId: this.providerId,
providerType: this.providerType,
...(label !== undefined ? { label } : {}),
status: this.toSessionStatus(this.readString(record.status)),
...(parentSessionId !== undefined ? { parentSessionId } : {}),
createdAt,
updatedAt,
...(metadata !== undefined ? { metadata } : {}),
};
}
private toAgentMessage(value: unknown, fallbackSessionId?: string): AgentMessage | null {
if (typeof value === "string") {
const content = value.trim();
if (content.length === 0 || fallbackSessionId === undefined) {
return null;
}
return {
id: randomUUID(),
sessionId: fallbackSessionId,
role: "assistant",
content,
timestamp: new Date(),
};
}
let candidate: JsonRecord | null = null;
if (this.isRecord(value) && this.isRecord(value.message)) {
candidate = value.message;
} else if (this.isRecord(value)) {
candidate = value;
}
if (candidate === null) {
return null;
}
const sessionId = this.readString(candidate.sessionId) ?? fallbackSessionId;
if (!sessionId) {
return null;
}
const content = this.extractMessageContent(
candidate.content ?? candidate.text ?? candidate.message
);
if (content.length === 0) {
return null;
}
const metadata = this.toMetadata(candidate.metadata);
return {
id: this.readString(candidate.id) ?? this.readString(candidate.messageId) ?? randomUUID(),
sessionId,
role: this.toMessageRole(this.readString(candidate.role) ?? this.readString(candidate.type)),
content,
timestamp: this.parseDate(candidate.timestamp ?? candidate.createdAt),
...(metadata !== undefined ? { metadata } : {}),
};
}
private extractMessageContent(content: unknown): string {
if (typeof content === "string") {
return content.trim();
}
if (Array.isArray(content)) {
const parts: string[] = [];
for (const part of content) {
if (typeof part === "string") {
const trimmed = part.trim();
if (trimmed.length > 0) {
parts.push(trimmed);
}
continue;
}
if (!this.isRecord(part)) {
continue;
}
const text = this.readString(part.text) ?? this.readString(part.content);
if (text !== undefined && text.trim().length > 0) {
parts.push(text.trim());
}
}
return parts.join("\n\n").trim();
}
if (this.isRecord(content)) {
const text = this.readString(content.text) ?? this.readString(content.content);
return text?.trim() ?? "";
}
return "";
}
private toSessionStatus(status?: string): AgentSessionStatus {
switch (status?.toLowerCase()) {
case "active":
case "running":
return "active";
case "paused":
return "paused";
case "completed":
case "done":
case "succeeded":
return "completed";
case "failed":
case "error":
case "killed":
case "terminated":
case "cancelled":
return "failed";
case "idle":
case "pending":
case "queued":
default:
return "idle";
}
}
private toMessageRole(role?: string): AgentMessageRole {
switch (role?.toLowerCase()) {
case "assistant":
case "agent":
return "assistant";
case "system":
return "system";
case "tool":
return "tool";
case "operator":
case "user":
default:
return "user";
}
}
private normalizeLimit(value: number, fallback: number): number {
const normalized = Number.isFinite(value) ? Math.trunc(value) : fallback;
if (normalized < 1) {
return 1;
}
return Math.min(normalized, MAX_MESSAGE_LIMIT);
}
private parseDate(value: unknown, fallback = new Date()): Date {
if (value instanceof Date) {
return value;
}
if (typeof value === "string" || typeof value === "number") {
const parsed = new Date(value);
if (!Number.isNaN(parsed.getTime())) {
return parsed;
}
}
return fallback;
}
private toMetadata(value: unknown): Record<string, unknown> | undefined {
if (this.isRecord(value)) {
return value;
}
return undefined;
}
private resolveDisplayName(): string {
const credentials = this.readCredentials();
for (const key of DISPLAY_NAME_KEYS) {
const value = this.readString(credentials[key]);
if (value !== undefined) {
return value;
}
}
return this.config.name;
}
private resolveBaseUrl(): string {
const configRecord = this.config as unknown as JsonRecord;
const rawBaseUrl =
this.readString(this.config.gatewayUrl) ?? this.readString(configRecord.baseUrl);
if (rawBaseUrl === undefined) {
throw new Error(`OpenClaw provider ${this.providerId} is missing gateway URL`);
}
try {
const parsed = new URL(rawBaseUrl);
return parsed.toString().replace(/\/$/u, "");
} catch {
throw new Error(`OpenClaw provider ${this.providerId} has invalid gateway URL`);
}
}
private resolveApiToken(): string {
const configRecord = this.config as unknown as JsonRecord;
const credentials = this.readCredentials();
const rawToken =
this.readString(configRecord.apiToken) ??
this.readString(configRecord.token) ??
this.readString(configRecord.bearerToken) ??
this.findFirstString(credentials, API_TOKEN_KEYS);
if (rawToken === undefined) {
throw new Error(`OpenClaw provider ${this.providerId} is missing apiToken credentials`);
}
try {
return this.encryptionService.decryptIfNeeded(rawToken);
} catch (error) {
throw new Error(`Failed to decrypt API token: ${this.toErrorMessage(error)}`);
}
}
private readCredentials(): JsonRecord {
return this.isRecord(this.config.credentials) ? this.config.credentials : {};
}
private findFirstString(record: JsonRecord, keys: readonly string[]): string | undefined {
for (const key of keys) {
const value = this.readString(record[key]);
if (value !== undefined) {
return value;
}
}
return undefined;
}
private authHeaders(extraHeaders: Record<string, string> = {}): Record<string, string> {
return {
Authorization: `Bearer ${this.resolveApiToken()}`,
...extraHeaders,
};
}
private buildUrl(path: string): string {
return new URL(path, `${this.resolveBaseUrl()}/`).toString();
}
private isRecord(value: unknown): value is JsonRecord {
return typeof value === "object" && value !== null && !Array.isArray(value);
}
private readString(value: unknown): string | undefined {
if (typeof value !== "string") {
return undefined;
}
const trimmed = value.trim();
return trimmed.length > 0 ? trimmed : undefined;
}
private getHttpStatus(error: unknown): number | undefined {
if (typeof error !== "object" || error === null || !("response" in error)) {
return undefined;
}
const response = (error as HttpErrorWithResponse).response;
return typeof response?.status === "number" ? response.status : undefined;
}
private toServiceUnavailable(operation: string, error: unknown): ServiceUnavailableException {
return new ServiceUnavailableException(
`OpenClaw provider ${this.providerId} failed to ${operation}: ${this.toErrorMessage(error)}`
);
}
private toErrorMessage(error: unknown): string {
if (error instanceof Error) {
return error.message;
}
return String(error);
}
}

View File

@@ -0,0 +1,131 @@
import { Logger } from "@nestjs/common";
import type { AgentProviderConfig } from "@prisma/client";
import { beforeEach, describe, expect, it, vi } from "vitest";
import { PrismaService } from "../../prisma/prisma.service";
import { AgentProviderRegistry } from "../agents/agent-provider.registry";
import { OpenClawProviderFactory } from "./openclaw/openclaw.provider-factory";
import { ProvidersModule } from "./providers.module";
type MockOpenClawProvider = {
providerId: string;
validateBaseUrl: ReturnType<typeof vi.fn>;
validateToken: ReturnType<typeof vi.fn>;
isAvailable: ReturnType<typeof vi.fn>;
};
describe("ProvidersModule", () => {
let moduleRef: ProvidersModule;
let prisma: {
agentProviderConfig: {
findMany: ReturnType<typeof vi.fn>;
};
};
let registry: {
registerProvider: ReturnType<typeof vi.fn>;
};
let factory: {
createProvider: ReturnType<typeof vi.fn>;
};
const config: AgentProviderConfig = {
id: "cfg-openclaw-1",
workspaceId: "workspace-1",
name: "openclaw-home",
provider: "openclaw",
gatewayUrl: "https://gateway.example.com",
credentials: { apiToken: "enc:token-value" },
isActive: true,
createdAt: new Date("2026-03-07T15:00:00.000Z"),
updatedAt: new Date("2026-03-07T15:00:00.000Z"),
};
beforeEach(() => {
prisma = {
agentProviderConfig: {
findMany: vi.fn(),
},
};
registry = {
registerProvider: vi.fn(),
};
factory = {
createProvider: vi.fn(),
};
moduleRef = new ProvidersModule(
prisma as unknown as PrismaService,
registry as unknown as AgentProviderRegistry,
factory as unknown as OpenClawProviderFactory
);
});
it("registers reachable OpenClaw providers", async () => {
const provider: MockOpenClawProvider = {
providerId: "openclaw-home",
validateBaseUrl: vi.fn(),
validateToken: vi.fn(),
isAvailable: vi.fn().mockResolvedValue(true),
};
prisma.agentProviderConfig.findMany.mockResolvedValue([config]);
factory.createProvider.mockReturnValue(provider);
await moduleRef.onModuleInit();
expect(prisma.agentProviderConfig.findMany).toHaveBeenCalledWith({
where: {
provider: "openclaw",
isActive: true,
},
orderBy: [{ createdAt: "asc" }, { id: "asc" }],
});
expect(factory.createProvider).toHaveBeenCalledWith(config);
expect(provider.validateBaseUrl).toHaveBeenCalledTimes(1);
expect(provider.validateToken).toHaveBeenCalledTimes(1);
expect(provider.isAvailable).toHaveBeenCalledTimes(1);
expect(registry.registerProvider).toHaveBeenCalledWith(provider);
});
it("skips provider registration when gateway is unreachable", async () => {
const warnSpy = vi.spyOn(Logger.prototype, "warn").mockImplementation(() => undefined);
const provider: MockOpenClawProvider = {
providerId: "openclaw-home",
validateBaseUrl: vi.fn(),
validateToken: vi.fn(),
isAvailable: vi.fn().mockResolvedValue(false),
};
prisma.agentProviderConfig.findMany.mockResolvedValue([config]);
factory.createProvider.mockReturnValue(provider);
await moduleRef.onModuleInit();
expect(registry.registerProvider).not.toHaveBeenCalled();
expect(warnSpy).toHaveBeenCalledWith(
expect.stringContaining("Skipping OpenClaw provider openclaw-home")
);
});
it("skips provider registration when token decryption fails", async () => {
const errorSpy = vi.spyOn(Logger.prototype, "error").mockImplementation(() => undefined);
const provider: MockOpenClawProvider = {
providerId: "openclaw-home",
validateBaseUrl: vi.fn(),
validateToken: vi.fn().mockImplementation(() => {
throw new Error("Failed to decrypt API token");
}),
isAvailable: vi.fn().mockResolvedValue(true),
};
prisma.agentProviderConfig.findMany.mockResolvedValue([config]);
factory.createProvider.mockReturnValue(provider);
await moduleRef.onModuleInit();
expect(registry.registerProvider).not.toHaveBeenCalled();
expect(errorSpy).toHaveBeenCalledWith(expect.stringContaining("token decryption failed"));
expect(provider.isAvailable).not.toHaveBeenCalled();
});
});

View File

@@ -0,0 +1,95 @@
import { HttpModule } from "@nestjs/axios";
import { Logger, Module, OnModuleInit } from "@nestjs/common";
import type { AgentProviderConfig } from "@prisma/client";
import { PrismaModule } from "../../prisma/prisma.module";
import { PrismaService } from "../../prisma/prisma.service";
import { EncryptionService } from "../../security/encryption.service";
import { AgentProviderRegistry } from "../agents/agent-provider.registry";
import { AgentsModule } from "../agents/agents.module";
import { OpenClawProviderFactory } from "./openclaw/openclaw.provider-factory";
import { OpenClawSseBridge } from "./openclaw/openclaw-sse.bridge";
const OPENCLAW_PROVIDER_TYPE = "openclaw";
@Module({
imports: [
AgentsModule,
PrismaModule,
HttpModule.register({
timeout: 10000,
maxRedirects: 5,
}),
],
providers: [EncryptionService, OpenClawSseBridge, OpenClawProviderFactory],
})
export class ProvidersModule implements OnModuleInit {
private readonly logger = new Logger(ProvidersModule.name);
constructor(
private readonly prisma: PrismaService,
private readonly registry: AgentProviderRegistry,
private readonly openClawProviderFactory: OpenClawProviderFactory
) {}
async onModuleInit(): Promise<void> {
const configs = await this.prisma.agentProviderConfig.findMany({
where: {
provider: OPENCLAW_PROVIDER_TYPE,
isActive: true,
},
orderBy: [{ createdAt: "asc" }, { id: "asc" }],
});
for (const config of configs) {
await this.registerProvider(config);
}
}
private async registerProvider(config: AgentProviderConfig): Promise<void> {
const provider = this.openClawProviderFactory.createProvider(config);
try {
provider.validateBaseUrl();
} catch (error) {
this.logger.warn(
`Skipping OpenClaw provider ${config.name}: invalid configuration (${this.toErrorMessage(error)})`
);
return;
}
try {
provider.validateToken();
} catch (error) {
this.logger.error(
`Skipping OpenClaw provider ${config.name}: token decryption failed (${this.toErrorMessage(error)})`
);
return;
}
try {
const available = await provider.isAvailable();
if (!available) {
this.logger.warn(
`Skipping OpenClaw provider ${config.name}: gateway ${config.gatewayUrl} is unreachable`
);
return;
}
} catch (error) {
this.logger.warn(
`Skipping OpenClaw provider ${config.name}: gateway ${config.gatewayUrl} is unreachable (${this.toErrorMessage(error)})`
);
return;
}
this.registry.registerProvider(provider);
this.logger.log(`Registered OpenClaw provider ${provider.providerId}`);
}
private toErrorMessage(error: unknown): string {
if (error instanceof Error) {
return error.message;
}
return String(error);
}
}

View File

@@ -4,7 +4,10 @@ import { BullModule } from "@nestjs/bullmq";
import { ThrottlerModule } from "@nestjs/throttler";
import { HealthModule } from "./api/health/health.module";
import { AgentsModule } from "./api/agents/agents.module";
import { MissionControlModule } from "./api/mission-control/mission-control.module";
import { QueueApiModule } from "./api/queue/queue-api.module";
import { AgentProvidersModule } from "./api/agent-providers/agent-providers.module";
import { ProvidersModule } from "./api/providers/providers.module";
import { CoordinatorModule } from "./coordinator/coordinator.module";
import { BudgetModule } from "./budget/budget.module";
import { CIModule } from "./ci";
@@ -51,6 +54,9 @@ import { orchestratorConfig } from "./config/orchestrator.config";
]),
HealthModule,
AgentsModule,
AgentProvidersModule,
ProvidersModule,
MissionControlModule,
QueueApiModule,
CoordinatorModule,
BudgetModule,

View File

@@ -0,0 +1,9 @@
import { Module } from "@nestjs/common";
import { OrchestratorApiKeyGuard } from "../common/guards/api-key.guard";
import { AuthGuard } from "./guards/auth.guard";
@Module({
providers: [OrchestratorApiKeyGuard, AuthGuard],
exports: [AuthGuard],
})
export class AuthModule {}

View File

@@ -0,0 +1,11 @@
import { CanActivate, ExecutionContext, Injectable } from "@nestjs/common";
import { OrchestratorApiKeyGuard } from "../../common/guards/api-key.guard";
@Injectable()
export class AuthGuard implements CanActivate {
constructor(private readonly apiKeyGuard: OrchestratorApiKeyGuard) {}
canActivate(context: ExecutionContext): boolean | Promise<boolean> {
return this.apiKeyGuard.canActivate(context);
}
}

View File

@@ -0,0 +1,9 @@
import { Global, Module } from "@nestjs/common";
import { PrismaService } from "./prisma.service";
@Global()
@Module({
providers: [PrismaService],
exports: [PrismaService],
})
export class PrismaModule {}

View File

@@ -0,0 +1,26 @@
import { Injectable, Logger, OnModuleDestroy, OnModuleInit } from "@nestjs/common";
import { PrismaClient } from "@prisma/client";
/**
* Lightweight Prisma service for orchestrator ingestion persistence.
*/
@Injectable()
export class PrismaService extends PrismaClient implements OnModuleInit, OnModuleDestroy {
private readonly logger = new Logger(PrismaService.name);
constructor() {
super({
log: process.env.NODE_ENV === "development" ? ["warn", "error"] : ["error"],
});
}
async onModuleInit(): Promise<void> {
await this.$connect();
this.logger.log("Database connection established");
}
async onModuleDestroy(): Promise<void> {
await this.$disconnect();
this.logger.log("Database connection closed");
}
}

View File

@@ -0,0 +1,85 @@
import { Injectable } from "@nestjs/common";
import { ConfigService } from "@nestjs/config";
import { createDecipheriv, hkdfSync } from "node:crypto";
const ALGORITHM = "aes-256-gcm";
const ENCRYPTED_PREFIX = "enc:";
const IV_LENGTH = 12;
const AUTH_TAG_LENGTH = 16;
const DERIVED_KEY_LENGTH = 32;
const HKDF_SALT = "mosaic.crypto.v1";
const HKDF_INFO = "mosaic-db-secret-encryption";
@Injectable()
export class EncryptionService {
private key: Buffer | null = null;
constructor(private readonly configService: ConfigService) {}
decryptIfNeeded(value: string): string {
if (!this.isEncrypted(value)) {
return value;
}
return this.decrypt(value);
}
decrypt(encrypted: string): string {
if (!this.isEncrypted(encrypted)) {
throw new Error("Value is not encrypted");
}
const payloadBase64 = encrypted.slice(ENCRYPTED_PREFIX.length);
try {
const payload = Buffer.from(payloadBase64, "base64");
if (payload.length < IV_LENGTH + AUTH_TAG_LENGTH) {
throw new Error("Encrypted payload is too short");
}
const iv = payload.subarray(0, IV_LENGTH);
const authTag = payload.subarray(payload.length - AUTH_TAG_LENGTH);
const ciphertext = payload.subarray(IV_LENGTH, payload.length - AUTH_TAG_LENGTH);
const decipher = createDecipheriv(ALGORITHM, this.getOrCreateKey(), iv);
decipher.setAuthTag(authTag);
return Buffer.concat([decipher.update(ciphertext), decipher.final()]).toString("utf8");
} catch {
throw new Error("Failed to decrypt value");
}
}
isEncrypted(value: string): boolean {
return value.startsWith(ENCRYPTED_PREFIX);
}
private getOrCreateKey(): Buffer {
if (this.key !== null) {
return this.key;
}
const secret = this.configService.get<string>("MOSAIC_SECRET_KEY");
if (!secret) {
throw new Error(
"orchestrator: MOSAIC_SECRET_KEY is required. Set it in your config or via MOSAIC_SECRET_KEY."
);
}
if (secret.length < 32) {
throw new Error("MOSAIC_SECRET_KEY must be at least 32 characters");
}
this.key = Buffer.from(
hkdfSync(
"sha256",
Buffer.from(secret, "utf8"),
Buffer.from(HKDF_SALT, "utf8"),
Buffer.from(HKDF_INFO, "utf8"),
DERIVED_KEY_LENGTH
)
);
return this.key;
}
}

View File

@@ -1,6 +1,7 @@
import { Injectable, Logger, Inject, forwardRef } from "@nestjs/common";
import { Injectable, Logger, Inject, Optional, forwardRef } from "@nestjs/common";
import { ValkeyService } from "../valkey/valkey.service";
import { AgentSpawnerService } from "./agent-spawner.service";
import { AgentIngestionService } from "../agent-ingestion/agent-ingestion.service";
import type { AgentState, AgentStatus, AgentEvent } from "../valkey/types";
import { isValidAgentTransition } from "../valkey/types/state.types";
@@ -32,7 +33,8 @@ export class AgentLifecycleService {
constructor(
private readonly valkeyService: ValkeyService,
@Inject(forwardRef(() => AgentSpawnerService))
private readonly spawnerService: AgentSpawnerService
private readonly spawnerService: AgentSpawnerService,
@Optional() private readonly agentIngestionService?: AgentIngestionService
) {
this.logger.log("AgentLifecycleService initialized");
}
@@ -55,6 +57,25 @@ export class AgentLifecycleService {
return createdState;
}
private async recordLifecycleIngestion(
agentId: string,
event: "started" | "completed" | "failed" | "killed",
record: (ingestionService: AgentIngestionService) => Promise<void>
): Promise<void> {
if (!this.agentIngestionService) {
return;
}
try {
await record(this.agentIngestionService);
} catch (error: unknown) {
const errorMessage = error instanceof Error ? error.message : String(error);
this.logger.error(
`Failed to record agent ${event} ingestion for ${agentId}: ${errorMessage}`
);
}
}
/**
* Acquire a per-agent mutex to serialize state transitions.
* Uses promise chaining: each caller chains onto the previous lock,
@@ -118,6 +139,10 @@ export class AgentLifecycleService {
// Emit event
await this.publishStateChangeEvent("agent.running", updatedState);
await this.recordLifecycleIngestion(agentId, "started", (ingestionService) =>
ingestionService.recordAgentStarted(agentId)
);
this.logger.log(`Agent ${agentId} transitioned to running`);
return updatedState;
});
@@ -155,6 +180,10 @@ export class AgentLifecycleService {
// Emit event
await this.publishStateChangeEvent("agent.completed", updatedState);
await this.recordLifecycleIngestion(agentId, "completed", (ingestionService) =>
ingestionService.recordAgentCompleted(agentId)
);
// Schedule session cleanup
this.spawnerService.scheduleSessionCleanup(agentId);
@@ -192,6 +221,10 @@ export class AgentLifecycleService {
// Emit event
await this.publishStateChangeEvent("agent.failed", updatedState, error);
await this.recordLifecycleIngestion(agentId, "failed", (ingestionService) =>
ingestionService.recordAgentFailed(agentId, error)
);
// Schedule session cleanup
this.spawnerService.scheduleSessionCleanup(agentId);
@@ -228,6 +261,10 @@ export class AgentLifecycleService {
// Emit event
await this.publishStateChangeEvent("agent.killed", updatedState);
await this.recordLifecycleIngestion(agentId, "killed", (ingestionService) =>
ingestionService.recordAgentKilled(agentId)
);
// Schedule session cleanup
this.spawnerService.scheduleSessionCleanup(agentId);

View File

@@ -1,4 +1,11 @@
import { Injectable, Logger, HttpException, HttpStatus, OnModuleDestroy } from "@nestjs/common";
import {
Injectable,
Logger,
HttpException,
HttpStatus,
OnModuleDestroy,
Optional,
} from "@nestjs/common";
import { ConfigService } from "@nestjs/config";
import Anthropic from "@anthropic-ai/sdk";
import { randomUUID } from "crypto";
@@ -8,6 +15,7 @@ import {
AgentSession,
AgentType,
} from "./types/agent-spawner.types";
import { AgentIngestionService } from "../agent-ingestion/agent-ingestion.service";
/**
* Default delay in milliseconds before cleaning up sessions after terminal states
@@ -30,7 +38,10 @@ export class AgentSpawnerService implements OnModuleDestroy {
private readonly sessionCleanupDelayMs: number;
private readonly cleanupTimers = new Map<string, NodeJS.Timeout>();
constructor(private readonly configService: ConfigService) {
constructor(
private readonly configService: ConfigService,
@Optional() private readonly agentIngestionService?: AgentIngestionService
) {
const configuredProvider = this.configService.get<string>("orchestrator.aiProvider");
this.aiProvider = this.normalizeAiProvider(configuredProvider);
@@ -98,6 +109,25 @@ export class AgentSpawnerService implements OnModuleDestroy {
this.cleanupTimers.clear();
}
private recordSpawnedAgentIngestion(agentId: string, request: SpawnAgentRequest): void {
if (!this.agentIngestionService) {
return;
}
void this.agentIngestionService
.recordAgentSpawned(
agentId,
request.parentAgentId,
undefined,
request.taskId,
request.agentType
)
.catch((error: unknown) => {
const errorMessage = error instanceof Error ? error.message : String(error);
this.logger.error(`Failed to record spawned ingestion for ${agentId}: ${errorMessage}`);
});
}
/**
* Spawn a new agent with the given configuration
* @param request Agent spawn request
@@ -130,6 +160,8 @@ export class AgentSpawnerService implements OnModuleDestroy {
// Store session
this.sessions.set(agentId, session);
this.recordSpawnedAgentIngestion(agentId, request);
this.logger.log(`Agent spawned successfully: ${agentId} (type: ${request.agentType})`);
// NOTE: Actual Claude SDK integration will be implemented in next iteration (see issue #TBD)

View File

@@ -3,9 +3,10 @@ import { AgentSpawnerService } from "./agent-spawner.service";
import { AgentLifecycleService } from "./agent-lifecycle.service";
import { DockerSandboxService } from "./docker-sandbox.service";
import { ValkeyModule } from "../valkey/valkey.module";
import { AgentIngestionModule } from "../agent-ingestion/agent-ingestion.module";
@Module({
imports: [ValkeyModule],
imports: [ValkeyModule, AgentIngestionModule],
providers: [AgentSpawnerService, AgentLifecycleService, DockerSandboxService],
exports: [AgentSpawnerService, AgentLifecycleService, DockerSandboxService],
})

View File

@@ -40,6 +40,8 @@ export interface SpawnAgentOptions {
export interface SpawnAgentRequest {
/** Unique task identifier */
taskId: string;
/** Optional parent session identifier for subagent lineage */
parentAgentId?: string;
/** Type of agent to spawn */
agentType: AgentType;
/** Context for task execution */

View File

@@ -4,6 +4,7 @@ export default defineConfig({
test: {
globals: true,
environment: "node",
setupFiles: ["reflect-metadata"],
exclude: ["**/node_modules/**", "**/dist/**", "**/tests/integration/**"],
include: ["src/**/*.spec.ts", "src/**/*.test.ts"],
coverage: {

View File

@@ -132,7 +132,7 @@ describe("KanbanPage add task flow", (): void => {
});
// Click the "+ Add task" button in the To Do column
const addTaskButtons = screen.getAllByRole("button", { name: /\+ Add task/i });
const addTaskButtons = await screen.findAllByRole("button", { name: /\+ Add task/i });
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
await user.click(addTaskButtons[0]!); // First column is "To Do"
@@ -165,7 +165,7 @@ describe("KanbanPage add task flow", (): void => {
});
// Click the "+ Add task" button
const addTaskButtons = screen.getAllByRole("button", { name: /\+ Add task/i });
const addTaskButtons = await screen.findAllByRole("button", { name: /\+ Add task/i });
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
await user.click(addTaskButtons[0]!);

View File

@@ -0,0 +1,5 @@
import { MissionControlLayout } from "@/components/mission-control/MissionControlLayout";
export default function MissionControlPage(): React.JSX.Element {
return <MissionControlLayout />;
}

View File

@@ -156,6 +156,26 @@ function IconTerminal(): React.JSX.Element {
);
}
function IconMissionControl(): React.JSX.Element {
return (
<svg
width="16"
height="16"
viewBox="0 0 16 16"
fill="none"
stroke="currentColor"
strokeWidth="1.5"
aria-hidden="true"
>
<circle cx="8" cy="8" r="1.5" />
<path d="M11 5a4.25 4.25 0 0 1 0 6" />
<path d="M5 5a4.25 4.25 0 0 0 0 6" />
<path d="M13.5 2.5a7.75 7.75 0 0 1 0 11" />
<path d="M2.5 2.5a7.75 7.75 0 0 0 0 11" />
</svg>
);
}
function IconSettings(): React.JSX.Element {
return (
<svg
@@ -260,6 +280,11 @@ const NAV_GROUPS: NavGroup[] = [
label: "Terminal",
icon: <IconTerminal />,
},
{
href: "/mission-control",
label: "Mission Control",
icon: <IconMissionControl />,
},
],
},
{

View File

@@ -0,0 +1,205 @@
import { beforeEach, describe, expect, it, vi } from "vitest";
import { render, screen, waitFor } from "@testing-library/react";
import userEvent from "@testing-library/user-event";
import { QueryClient, QueryClientProvider } from "@tanstack/react-query";
import type { ButtonHTMLAttributes, HTMLAttributes, ReactNode } from "react";
interface MockButtonProps extends ButtonHTMLAttributes<HTMLButtonElement> {
children: ReactNode;
}
interface MockBadgeProps extends HTMLAttributes<HTMLElement> {
children: ReactNode;
}
interface AuditLogEntry {
id: string;
userId: string;
sessionId: string;
provider: string;
action: string;
content: string | null;
metadata: unknown;
createdAt: string;
}
interface AuditLogResponse {
items: AuditLogEntry[];
total: number;
page: number;
pages: number;
}
const mockApiGet = vi.fn<(endpoint: string) => Promise<AuditLogResponse>>();
vi.mock("@/lib/api/client", () => ({
apiGet: (endpoint: string): Promise<AuditLogResponse> => mockApiGet(endpoint),
}));
vi.mock("@/components/ui/button", () => ({
Button: ({ children, ...props }: MockButtonProps): React.JSX.Element => (
<button {...props}>{children}</button>
),
}));
vi.mock("@/components/ui/badge", () => ({
Badge: ({ children, ...props }: MockBadgeProps): React.JSX.Element => (
<span {...props}>{children}</span>
),
}));
import { AuditLogDrawer } from "./AuditLogDrawer";
function renderWithQueryClient(ui: React.JSX.Element): ReturnType<typeof render> {
const queryClient = new QueryClient({
defaultOptions: {
queries: { retry: false },
mutations: { retry: false },
},
});
return render(<QueryClientProvider client={queryClient}>{ui}</QueryClientProvider>);
}
function responseWith(items: AuditLogEntry[], page: number, pages: number): AuditLogResponse {
return {
items,
total: items.length,
page,
pages,
};
}
describe("AuditLogDrawer", (): void => {
beforeEach((): void => {
vi.clearAllMocks();
mockApiGet.mockResolvedValue(responseWith([], 1, 0));
});
it("opens from trigger text and renders empty state", async (): Promise<void> => {
const user = userEvent.setup();
renderWithQueryClient(<AuditLogDrawer trigger="Audit" />);
await user.click(screen.getByRole("button", { name: "Audit" }));
await waitFor((): void => {
expect(screen.getByText("Audit Log")).toBeInTheDocument();
expect(screen.getByText("No audit entries found.")).toBeInTheDocument();
});
});
it("renders audit entries with action, session id, and payload", async (): Promise<void> => {
const user = userEvent.setup();
mockApiGet.mockResolvedValue(
responseWith(
[
{
id: "entry-1",
userId: "operator-1",
sessionId: "1234567890abcdef",
provider: "internal",
action: "inject",
content: "Run diagnostics",
metadata: { payload: { ignored: true } },
createdAt: "2026-03-07T19:00:00.000Z",
},
],
1,
1
)
);
renderWithQueryClient(<AuditLogDrawer trigger="Audit" />);
await user.click(screen.getByRole("button", { name: "Audit" }));
await waitFor((): void => {
expect(screen.getByText("inject")).toBeInTheDocument();
expect(screen.getByText("12345678")).toBeInTheDocument();
expect(screen.getByText("Run diagnostics")).toBeInTheDocument();
});
});
it("supports pagination and metadata payload summary", async (): Promise<void> => {
const user = userEvent.setup();
mockApiGet.mockImplementation((endpoint: string): Promise<AuditLogResponse> => {
const query = endpoint.split("?")[1] ?? "";
const params = new URLSearchParams(query);
const page = Number(params.get("page") ?? "1");
if (page === 1) {
return Promise.resolve({
items: [
{
id: "entry-page-1",
userId: "operator-2",
sessionId: "abcdefgh12345678",
provider: "internal",
action: "pause",
content: "",
metadata: { payload: { reason: "hold" } },
createdAt: "2026-03-07T19:01:00.000Z",
},
],
total: 2,
page: 1,
pages: 2,
});
}
return Promise.resolve({
items: [
{
id: "entry-page-2",
userId: "operator-3",
sessionId: "zzzz111122223333",
provider: "internal",
action: "kill",
content: null,
metadata: { payload: { force: true } },
createdAt: "2026-03-07T19:02:00.000Z",
},
],
total: 2,
page: 2,
pages: 2,
});
});
renderWithQueryClient(<AuditLogDrawer trigger="Audit" />);
await user.click(screen.getByRole("button", { name: "Audit" }));
await waitFor((): void => {
expect(screen.getByText("Page 1 of 2")).toBeInTheDocument();
expect(screen.getByText("reason=hold")).toBeInTheDocument();
});
await user.click(screen.getByRole("button", { name: "Next" }));
await waitFor((): void => {
expect(screen.getByText("Page 2 of 2")).toBeInTheDocument();
expect(screen.getByText("force=true")).toBeInTheDocument();
});
});
it("includes sessionId filter in query string", async (): Promise<void> => {
const user = userEvent.setup();
renderWithQueryClient(<AuditLogDrawer trigger="Audit" sessionId="session 7" />);
await user.click(screen.getByRole("button", { name: "Audit" }));
await waitFor((): void => {
expect(mockApiGet).toHaveBeenCalled();
});
const firstCall = mockApiGet.mock.calls[0];
const endpoint = firstCall?.[0] ?? "";
expect(endpoint).toContain("sessionId=session+7");
});
});

View File

@@ -0,0 +1,322 @@
"use client";
import { isValidElement, useEffect, useMemo, useState } from "react";
import type { ReactNode } from "react";
import { format } from "date-fns";
import { useQuery } from "@tanstack/react-query";
import { Loader2 } from "lucide-react";
import { Badge } from "@/components/ui/badge";
import type { BadgeVariant } from "@/components/ui/badge";
import { Button } from "@/components/ui/button";
import { ScrollArea } from "@/components/ui/scroll-area";
import {
Sheet,
SheetContent,
SheetDescription,
SheetHeader,
SheetTitle,
SheetTrigger,
} from "@/components/ui/sheet";
import { apiGet } from "@/lib/api/client";
const AUDIT_LOG_REFRESH_INTERVAL_MS = 10_000;
const AUDIT_LOG_PAGE_SIZE = 50;
const SUMMARY_MAX_LENGTH = 120;
interface AuditLogDrawerProps {
sessionId?: string;
trigger: ReactNode;
}
interface AuditLogEntry {
id: string;
userId: string;
sessionId: string;
provider: string;
action: string;
content: string | null;
metadata: unknown;
createdAt: string;
}
interface AuditLogResponse {
items: AuditLogEntry[];
total: number;
page: number;
pages: number;
}
function isRecord(value: unknown): value is Record<string, unknown> {
return typeof value === "object" && value !== null;
}
function truncateText(value: string, maxLength: number): string {
if (value.length <= maxLength) {
return value;
}
return `${value.slice(0, maxLength - 1)}`;
}
function truncateSessionId(sessionId: string): string {
return sessionId.slice(0, 8);
}
function formatTimestamp(value: string): string {
const parsed = new Date(value);
if (Number.isNaN(parsed.getTime())) {
return "Unknown";
}
return format(parsed, "yyyy-MM-dd HH:mm:ss");
}
function stringifyPayloadValue(value: unknown): string {
if (typeof value === "string") {
return value;
}
if (typeof value === "number" || typeof value === "boolean") {
return String(value);
}
try {
return JSON.stringify(value);
} catch {
return "[unserializable]";
}
}
function getPayloadSummary(entry: AuditLogEntry): string {
const metadata = isRecord(entry.metadata) ? entry.metadata : undefined;
const payload = metadata && isRecord(metadata.payload) ? metadata.payload : undefined;
if (typeof entry.content === "string" && entry.content.trim().length > 0) {
return truncateText(entry.content.trim(), SUMMARY_MAX_LENGTH);
}
if (payload) {
const summary = Object.entries(payload)
.map(([key, value]) => `${key}=${stringifyPayloadValue(value)}`)
.join(", ");
if (summary.length > 0) {
return truncateText(summary, SUMMARY_MAX_LENGTH);
}
}
return "—";
}
function getActionVariant(action: string): BadgeVariant {
switch (action) {
case "inject":
return "badge-blue";
case "pause":
return "status-warning";
case "resume":
return "status-success";
case "kill":
return "status-error";
default:
return "status-neutral";
}
}
async function fetchAuditLog(
sessionId: string | undefined,
page: number
): Promise<AuditLogResponse> {
const params = new URLSearchParams({
page: String(page),
limit: String(AUDIT_LOG_PAGE_SIZE),
});
const normalizedSessionId = sessionId?.trim();
if (normalizedSessionId && normalizedSessionId.length > 0) {
params.set("sessionId", normalizedSessionId);
}
return apiGet<AuditLogResponse>(`/api/mission-control/audit-log?${params.toString()}`);
}
export function AuditLogDrawer({ sessionId, trigger }: AuditLogDrawerProps): React.JSX.Element {
const [open, setOpen] = useState(false);
const [page, setPage] = useState(1);
const triggerElement = useMemo(
() =>
isValidElement(trigger) ? (
trigger
) : (
<Button variant="outline" size="sm">
{trigger}
</Button>
),
[trigger]
);
const auditLogQuery = useQuery<AuditLogResponse>({
queryKey: ["mission-control", "audit-log", sessionId ?? "all", page],
queryFn: async (): Promise<AuditLogResponse> => fetchAuditLog(sessionId, page),
enabled: open,
refetchInterval: open ? AUDIT_LOG_REFRESH_INTERVAL_MS : false,
});
useEffect(() => {
if (open) {
setPage(1);
}
}, [open, sessionId]);
useEffect(() => {
const pages = auditLogQuery.data?.pages;
if (pages !== undefined && pages > 0 && page > pages) {
setPage(pages);
}
}, [auditLogQuery.data?.pages, page]);
const totalItems = auditLogQuery.data?.total ?? 0;
const totalPages = auditLogQuery.data?.pages ?? 0;
const items = auditLogQuery.data?.items ?? [];
const canGoPrevious = page > 1;
const canGoNext = totalPages > 0 && page < totalPages;
const errorMessage =
auditLogQuery.error instanceof Error ? auditLogQuery.error.message : "Failed to load audit log";
return (
<Sheet open={open} onOpenChange={setOpen}>
<SheetTrigger asChild>{triggerElement}</SheetTrigger>
<SheetContent className="sm:max-w-[920px]">
<SheetHeader>
<div className="flex items-center justify-between gap-3">
<SheetTitle>Audit Log</SheetTitle>
{auditLogQuery.isFetching ? (
<Loader2 className="h-4 w-4 animate-spin text-muted-foreground" aria-hidden="true" />
) : null}
</div>
<SheetDescription>
{sessionId
? `Showing actions for session ${sessionId}.`
: "Showing operator actions across all mission control sessions."}
</SheetDescription>
</SheetHeader>
<div className="mt-4 flex min-h-0 flex-1 flex-col gap-3">
<div className="rounded-md border border-border/70">
<ScrollArea className="h-[64vh]">
<table className="w-full min-w-[760px] border-collapse text-sm">
<thead className="sticky top-0 z-10 bg-muted/90">
<tr>
<th className="px-3 py-2 text-left font-medium text-muted-foreground">
Timestamp
</th>
<th className="px-3 py-2 text-left font-medium text-muted-foreground">
Action
</th>
<th className="px-3 py-2 text-left font-medium text-muted-foreground">
Session
</th>
<th className="px-3 py-2 text-left font-medium text-muted-foreground">
Operator
</th>
<th className="px-3 py-2 text-left font-medium text-muted-foreground">
Payload
</th>
</tr>
</thead>
<tbody>
{auditLogQuery.isLoading ? (
<tr>
<td
colSpan={5}
className="px-3 py-6 text-center text-sm text-muted-foreground"
>
Loading audit log...
</td>
</tr>
) : auditLogQuery.error ? (
<tr>
<td colSpan={5} className="px-3 py-6 text-center text-sm text-red-500">
{errorMessage}
</td>
</tr>
) : items.length === 0 ? (
<tr>
<td
colSpan={5}
className="px-3 py-6 text-center text-sm text-muted-foreground"
>
No audit entries found.
</td>
</tr>
) : (
items.map((entry) => {
const payloadSummary = getPayloadSummary(entry);
return (
<tr key={entry.id} className="border-t border-border/60 align-top">
<td className="px-3 py-2 font-mono text-xs text-muted-foreground">
{formatTimestamp(entry.createdAt)}
</td>
<td className="px-3 py-2">
<Badge variant={getActionVariant(entry.action)} className="capitalize">
{entry.action}
</Badge>
</td>
<td className="px-3 py-2 font-mono text-xs" title={entry.sessionId}>
{truncateSessionId(entry.sessionId)}
</td>
<td
className="px-3 py-2 text-xs text-muted-foreground"
title={entry.userId}
>
{entry.userId}
</td>
<td className="px-3 py-2 text-xs" title={payloadSummary}>
{payloadSummary}
</td>
</tr>
);
})
)}
</tbody>
</table>
</ScrollArea>
</div>
<div className="flex items-center justify-between">
<p className="text-xs text-muted-foreground">{totalItems} total entries</p>
<div className="flex items-center gap-2">
<Button
variant="outline"
size="sm"
disabled={!canGoPrevious || auditLogQuery.isFetching}
onClick={() => {
setPage((currentPage) => Math.max(1, currentPage - 1));
}}
>
Previous
</Button>
<span className="text-xs text-muted-foreground">
Page {page} of {Math.max(totalPages, 1)}
</span>
<Button
variant="outline"
size="sm"
disabled={!canGoNext || auditLogQuery.isFetching}
onClick={() => {
setPage((currentPage) => currentPage + 1);
}}
>
Next
</Button>
</div>
</div>
</div>
</SheetContent>
</Sheet>
);
}

View File

@@ -0,0 +1,155 @@
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { fireEvent, render, screen, waitFor } from "@testing-library/react";
import userEvent from "@testing-library/user-event";
import * as MosaicUi from "@mosaic/ui";
import type { ButtonHTMLAttributes, ReactNode } from "react";
interface MockButtonProps extends ButtonHTMLAttributes<HTMLButtonElement> {
children: ReactNode;
}
const mockApiPost = vi.fn<(endpoint: string, body?: unknown) => Promise<{ message?: string }>>();
const mockShowToast = vi.fn<(message: string, variant?: string) => void>();
const useToastSpy = vi.spyOn(MosaicUi, "useToast");
vi.mock("@/lib/api/client", () => ({
apiPost: (endpoint: string, body?: unknown): Promise<{ message?: string }> =>
mockApiPost(endpoint, body),
}));
vi.mock("@/components/ui/button", () => ({
Button: ({ children, ...props }: MockButtonProps): React.JSX.Element => (
<button {...props}>{children}</button>
),
}));
import { BargeInInput } from "./BargeInInput";
describe("BargeInInput", (): void => {
beforeEach((): void => {
vi.clearAllMocks();
vi.stubGlobal("fetch", vi.fn());
mockApiPost.mockResolvedValue({ message: "ok" });
useToastSpy.mockReturnValue({
showToast: mockShowToast,
removeToast: vi.fn(),
} as ReturnType<typeof MosaicUi.useToast>);
});
afterEach((): void => {
vi.unstubAllGlobals();
});
it("renders input controls and keeps send disabled for empty content", (): void => {
render(<BargeInInput sessionId="session-1" />);
expect(screen.getByLabelText("Inject message")).toBeInTheDocument();
expect(screen.getByRole("checkbox", { name: "Pause before send" })).not.toBeChecked();
expect(screen.getByRole("button", { name: "Send" })).toBeDisabled();
});
it("sends a trimmed message and clears the textarea", async (): Promise<void> => {
const onSent = vi.fn<() => void>();
const user = userEvent.setup();
render(<BargeInInput sessionId="session-1" onSent={onSent} />);
const textarea = screen.getByLabelText("Inject message");
await user.type(textarea, " execute plan ");
await user.click(screen.getByRole("button", { name: "Send" }));
await waitFor((): void => {
expect(mockApiPost).toHaveBeenCalledWith("/api/mission-control/sessions/session-1/inject", {
content: "execute plan",
});
});
expect(onSent).toHaveBeenCalledTimes(1);
expect(textarea).toHaveValue("");
});
it("pauses and resumes the session around injection when checkbox is enabled", async (): Promise<void> => {
const user = userEvent.setup();
render(<BargeInInput sessionId="session-2" />);
await user.click(screen.getByRole("checkbox", { name: "Pause before send" }));
await user.type(screen.getByLabelText("Inject message"), "hello world");
await user.click(screen.getByRole("button", { name: "Send" }));
await waitFor((): void => {
expect(mockApiPost).toHaveBeenCalledTimes(3);
});
const calls = mockApiPost.mock.calls as [string, unknown?][];
expect(calls[0]).toEqual(["/api/mission-control/sessions/session-2/pause", undefined]);
expect(calls[1]).toEqual([
"/api/mission-control/sessions/session-2/inject",
{ content: "hello world" },
]);
expect(calls[2]).toEqual(["/api/mission-control/sessions/session-2/resume", undefined]);
});
it("submits with Enter and does not submit on Shift+Enter", async (): Promise<void> => {
const user = userEvent.setup();
render(<BargeInInput sessionId="session-3" />);
const textarea = screen.getByLabelText("Inject message");
await user.type(textarea, "first");
fireEvent.keyDown(textarea, { key: "Enter", code: "Enter", shiftKey: true });
expect(mockApiPost).not.toHaveBeenCalled();
fireEvent.keyDown(textarea, { key: "Enter", code: "Enter", shiftKey: false });
await waitFor((): void => {
expect(mockApiPost).toHaveBeenCalledWith("/api/mission-control/sessions/session-3/inject", {
content: "first",
});
});
});
it("shows an inline error and toast when injection fails", async (): Promise<void> => {
const user = userEvent.setup();
mockApiPost.mockRejectedValueOnce(new Error("Injection failed"));
render(<BargeInInput sessionId="session-4" />);
await user.type(screen.getByLabelText("Inject message"), "help");
await user.click(screen.getByRole("button", { name: "Send" }));
await waitFor((): void => {
expect(screen.getByRole("alert")).toHaveTextContent("Injection failed");
});
expect(mockShowToast).toHaveBeenCalledWith("Injection failed", "error");
});
it("reports resume failures after a successful send", async (): Promise<void> => {
const user = userEvent.setup();
mockApiPost
.mockResolvedValueOnce({ message: "paused" })
.mockResolvedValueOnce({ message: "sent" })
.mockRejectedValueOnce(new Error("resume failed"));
render(<BargeInInput sessionId="session-5" />);
await user.click(screen.getByRole("checkbox", { name: "Pause before send" }));
await user.type(screen.getByLabelText("Inject message"), "deploy now");
await user.click(screen.getByRole("button", { name: "Send" }));
await waitFor((): void => {
expect(screen.getByRole("alert")).toHaveTextContent(
"Message sent, but failed to resume session: resume failed"
);
});
expect(mockShowToast).toHaveBeenCalledWith(
"Message sent, but failed to resume session: resume failed",
"error"
);
});
});

View File

@@ -0,0 +1,147 @@
"use client";
import { useCallback, useState, type KeyboardEvent } from "react";
import { Loader2 } from "lucide-react";
import { useToast } from "@mosaic/ui";
import { Button } from "@/components/ui/button";
import { apiPost } from "@/lib/api/client";
const MAX_ROWS = 4;
const TEXTAREA_MAX_HEIGHT_REM = 6.5;
interface BargeInMutationResponse {
message?: string;
}
export interface BargeInInputProps {
sessionId: string;
onSent?: () => void;
}
function getErrorMessage(error: unknown): string {
if (error instanceof Error && error.message.trim().length > 0) {
return error.message;
}
return "Failed to send message to the session.";
}
export function BargeInInput({ sessionId, onSent }: BargeInInputProps): React.JSX.Element {
const { showToast } = useToast();
const [content, setContent] = useState("");
const [pauseBeforeSend, setPauseBeforeSend] = useState(false);
const [isSending, setIsSending] = useState(false);
const [errorMessage, setErrorMessage] = useState<string | null>(null);
const handleSend = useCallback(async (): Promise<void> => {
const trimmedContent = content.trim();
if (!trimmedContent || isSending) {
return;
}
const encodedSessionId = encodeURIComponent(sessionId);
const baseEndpoint = `/api/mission-control/sessions/${encodedSessionId}`;
let didPause = false;
let didInject = false;
setIsSending(true);
setErrorMessage(null);
try {
if (pauseBeforeSend) {
await apiPost<BargeInMutationResponse>(`${baseEndpoint}/pause`);
didPause = true;
}
await apiPost<BargeInMutationResponse>(`${baseEndpoint}/inject`, { content: trimmedContent });
didInject = true;
setContent("");
onSent?.();
} catch (error) {
const message = getErrorMessage(error);
setErrorMessage(message);
showToast(message, "error");
} finally {
if (didPause) {
try {
await apiPost<BargeInMutationResponse>(`${baseEndpoint}/resume`);
} catch (resumeError) {
const resumeMessage = getErrorMessage(resumeError);
const message = didInject
? `Message sent, but failed to resume session: ${resumeMessage}`
: `Failed to resume session: ${resumeMessage}`;
setErrorMessage(message);
showToast(message, "error");
}
}
setIsSending(false);
}
}, [content, isSending, onSent, pauseBeforeSend, sessionId, showToast]);
const handleKeyDown = useCallback(
(event: KeyboardEvent<HTMLTextAreaElement>): void => {
if (event.key === "Enter" && !event.shiftKey) {
event.preventDefault();
void handleSend();
}
},
[handleSend]
);
const isSendDisabled = isSending || content.trim().length === 0;
return (
<div className="space-y-2">
<textarea
value={content}
onChange={(event) => {
setContent(event.target.value);
}}
onKeyDown={handleKeyDown}
disabled={isSending}
rows={MAX_ROWS}
placeholder="Inject a message into this session..."
className="block w-full resize-y rounded-md border border-border bg-background px-3 py-2 text-sm leading-5 text-foreground outline-none placeholder:text-muted-foreground disabled:cursor-not-allowed disabled:opacity-60"
style={{ maxHeight: `${String(TEXTAREA_MAX_HEIGHT_REM)}rem` }}
aria-label="Inject message"
/>
<div className="flex items-center justify-between gap-3">
<label className="flex items-center gap-2 text-sm text-muted-foreground">
<input
type="checkbox"
checked={pauseBeforeSend}
onChange={(event) => {
setPauseBeforeSend(event.target.checked);
}}
disabled={isSending}
className="h-4 w-4 rounded border-border"
/>
<span>Pause before send</span>
</label>
<Button
type="button"
variant="primary"
size="sm"
disabled={isSendDisabled}
onClick={() => {
void handleSend();
}}
>
{isSending ? (
<span className="flex items-center gap-2">
<Loader2 className="h-4 w-4 animate-spin" aria-hidden="true" />
Sending...
</span>
) : (
"Send"
)}
</Button>
</div>
{errorMessage ? (
<p role="alert" className="text-sm text-red-500">
{errorMessage}
</p>
) : null}
</div>
);
}

View File

@@ -0,0 +1,203 @@
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { fireEvent, render, screen, waitFor } from "@testing-library/react";
import { QueryClient, QueryClientProvider } from "@tanstack/react-query";
import type { ButtonHTMLAttributes, HTMLAttributes, ReactNode } from "react";
interface MockButtonProps extends ButtonHTMLAttributes<HTMLButtonElement> {
children: ReactNode;
}
interface MockContainerProps extends HTMLAttributes<HTMLElement> {
children: ReactNode;
}
interface MockSession {
id: string;
providerId: string;
providerType: string;
status: "active" | "paused" | "killed";
createdAt: string;
updatedAt: string;
metadata?: Record<string, unknown>;
}
const mockApiGet = vi.fn<(endpoint: string) => Promise<MockSession[]>>();
const mockApiPost = vi.fn<(endpoint: string, body?: unknown) => Promise<{ message: string }>>();
const mockKillAllDialog = vi.fn<() => React.JSX.Element>();
vi.mock("@/lib/api/client", () => ({
apiGet: (endpoint: string): Promise<MockSession[]> => mockApiGet(endpoint),
apiPost: (endpoint: string, body?: unknown): Promise<{ message: string }> =>
mockApiPost(endpoint, body),
}));
vi.mock("@/components/mission-control/KillAllDialog", () => ({
KillAllDialog: (): React.JSX.Element => mockKillAllDialog(),
}));
vi.mock("@/components/ui/button", () => ({
Button: ({ children, ...props }: MockButtonProps): React.JSX.Element => (
<button {...props}>{children}</button>
),
}));
vi.mock("@/components/ui/badge", () => ({
Badge: ({ children, ...props }: MockContainerProps): React.JSX.Element => (
<span {...props}>{children}</span>
),
}));
vi.mock("@/components/ui/card", () => ({
Card: ({ children, ...props }: MockContainerProps): React.JSX.Element => (
<section {...props}>{children}</section>
),
CardHeader: ({ children, ...props }: MockContainerProps): React.JSX.Element => (
<header {...props}>{children}</header>
),
CardContent: ({ children, ...props }: MockContainerProps): React.JSX.Element => (
<div {...props}>{children}</div>
),
CardTitle: ({ children, ...props }: MockContainerProps): React.JSX.Element => (
<h2 {...props}>{children}</h2>
),
}));
import { GlobalAgentRoster } from "./GlobalAgentRoster";
function renderWithQueryClient(ui: React.JSX.Element): ReturnType<typeof render> {
const queryClient = new QueryClient({
defaultOptions: {
queries: { retry: false },
mutations: { retry: false },
},
});
return render(<QueryClientProvider client={queryClient}>{ui}</QueryClientProvider>);
}
function makeSession(overrides: Partial<MockSession>): MockSession {
return {
id: "session-12345678",
providerId: "internal",
providerType: "internal",
status: "active",
createdAt: "2026-03-07T10:00:00.000Z",
updatedAt: "2026-03-07T10:01:00.000Z",
...overrides,
};
}
function getRowForSessionLabel(label: string): HTMLElement {
const sessionLabel = screen.getByText(label);
const row = sessionLabel.closest('[role="button"]');
if (!(row instanceof HTMLElement)) {
throw new Error(`Expected a row element for session label ${label}`);
}
return row;
}
describe("GlobalAgentRoster", (): void => {
beforeEach((): void => {
vi.clearAllMocks();
vi.stubGlobal("fetch", vi.fn());
mockApiGet.mockResolvedValue([]);
mockApiPost.mockResolvedValue({ message: "ok" });
mockKillAllDialog.mockImplementation(
(): React.JSX.Element => <div data-testid="kill-all-dialog">kill-all-dialog</div>
);
});
afterEach((): void => {
vi.unstubAllGlobals();
});
it("renders the empty state when no active sessions are returned", async (): Promise<void> => {
renderWithQueryClient(<GlobalAgentRoster />);
await waitFor((): void => {
expect(screen.getByText("No active agents")).toBeInTheDocument();
});
expect(screen.queryByTestId("kill-all-dialog")).not.toBeInTheDocument();
});
it("groups sessions by provider and shows kill-all control when sessions exist", async (): Promise<void> => {
mockApiGet.mockResolvedValue([
makeSession({ id: "alpha123456", providerId: "internal", providerType: "internal" }),
makeSession({ id: "bravo123456", providerId: "codex", providerType: "openai" }),
]);
renderWithQueryClient(<GlobalAgentRoster />);
await waitFor((): void => {
expect(screen.getByText("internal")).toBeInTheDocument();
expect(screen.getByText("codex (openai)")).toBeInTheDocument();
});
expect(screen.getByText("alpha123")).toBeInTheDocument();
expect(screen.getByText("bravo123")).toBeInTheDocument();
expect(screen.getByTestId("kill-all-dialog")).toBeInTheDocument();
});
it("calls onSelectSession on row click and keyboard activation", async (): Promise<void> => {
const onSelectSession = vi.fn<(sessionId: string) => void>();
mockApiGet.mockResolvedValue([makeSession({ id: "target123456" })]);
renderWithQueryClient(<GlobalAgentRoster onSelectSession={onSelectSession} />);
await waitFor((): void => {
expect(screen.getByText("target12")).toBeInTheDocument();
});
const row = getRowForSessionLabel("target12");
fireEvent.click(row);
fireEvent.keyDown(row, { key: "Enter" });
expect(onSelectSession).toHaveBeenCalledTimes(2);
expect(onSelectSession).toHaveBeenNthCalledWith(1, "target123456");
expect(onSelectSession).toHaveBeenNthCalledWith(2, "target123456");
});
it("kills a session from the roster", async (): Promise<void> => {
mockApiGet.mockResolvedValue([makeSession({ id: "killme123456" })]);
renderWithQueryClient(<GlobalAgentRoster />);
await waitFor((): void => {
expect(screen.getByRole("button", { name: "Kill session killme12" })).toBeInTheDocument();
});
fireEvent.click(screen.getByRole("button", { name: "Kill session killme12" }));
await waitFor((): void => {
expect(mockApiPost).toHaveBeenCalledWith("/api/mission-control/sessions/killme123456/kill", {
force: false,
});
});
});
it("collapses and reopens provider groups", async (): Promise<void> => {
mockApiGet.mockResolvedValue([makeSession({ id: "grouped12345" })]);
renderWithQueryClient(<GlobalAgentRoster />);
await waitFor((): void => {
expect(screen.getByText("grouped1")).toBeInTheDocument();
});
fireEvent.click(screen.getByRole("button", { name: /internal/i }));
expect(screen.queryByText("grouped1")).not.toBeInTheDocument();
fireEvent.click(screen.getByRole("button", { name: /internal/i }));
expect(screen.getByText("grouped1")).toBeInTheDocument();
});
});

View File

@@ -0,0 +1,292 @@
"use client";
import { useMemo, useState } from "react";
import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query";
import type { AgentSession } from "@mosaic/shared";
import { ChevronRight, Loader2, X } from "lucide-react";
import { KillAllDialog } from "@/components/mission-control/KillAllDialog";
import { Badge } from "@/components/ui/badge";
import type { BadgeVariant } from "@/components/ui/badge";
import { Button } from "@/components/ui/button";
import { Card, CardContent, CardHeader, CardTitle } from "@/components/ui/card";
import { Collapsible } from "@/components/ui/collapsible";
import { ScrollArea } from "@/components/ui/scroll-area";
import { Skeleton } from "@/components/ui/skeleton";
import { apiGet, apiPost } from "@/lib/api/client";
const SESSIONS_QUERY_KEY = ["mission-control", "sessions"] as const;
const SESSIONS_POLL_INTERVAL_MS = 5_000;
type MissionControlSessionStatus = AgentSession["status"] | "killed";
interface MissionControlSession extends Omit<AgentSession, "status" | "createdAt" | "updatedAt"> {
status: MissionControlSessionStatus;
createdAt: string | Date;
updatedAt: string | Date;
}
interface SessionsPayload {
sessions: MissionControlSession[];
}
interface ProviderSessionGroup {
providerId: string;
providerType: string;
sessions: MissionControlSession[];
}
export interface GlobalAgentRosterProps {
onSelectSession?: (sessionId: string) => void;
selectedSessionId?: string | undefined;
}
function getStatusVariant(status: MissionControlSessionStatus): BadgeVariant {
switch (status) {
case "active":
return "status-success";
case "paused":
return "status-warning";
case "killed":
return "status-error";
default:
return "status-neutral";
}
}
function truncateSessionId(sessionId: string): string {
return sessionId.slice(0, 8);
}
function resolveProviderName(providerId: string, providerType: string): string {
return providerId === providerType ? providerId : `${providerId} (${providerType})`;
}
function groupByProvider(sessions: MissionControlSession[]): ProviderSessionGroup[] {
const grouped = new Map<string, ProviderSessionGroup>();
for (const session of sessions) {
const existing = grouped.get(session.providerId);
if (existing) {
existing.sessions.push(session);
continue;
}
grouped.set(session.providerId, {
providerId: session.providerId,
providerType: session.providerType,
sessions: [session],
});
}
return Array.from(grouped.values()).sort((a, b) => a.providerId.localeCompare(b.providerId));
}
async function fetchSessions(): Promise<MissionControlSession[]> {
const payload = await apiGet<MissionControlSession[] | SessionsPayload>(
"/api/mission-control/sessions"
);
return Array.isArray(payload) ? payload : payload.sessions;
}
function toKillAllSessions(sessions: MissionControlSession[]): AgentSession[] {
return sessions
.filter(
(session): session is MissionControlSession & { status: AgentSession["status"] } =>
session.status !== "killed"
)
.map((session) => ({
...session,
createdAt:
session.createdAt instanceof Date ? session.createdAt : new Date(session.createdAt),
updatedAt:
session.updatedAt instanceof Date ? session.updatedAt : new Date(session.updatedAt),
}));
}
export function GlobalAgentRoster({
onSelectSession,
selectedSessionId,
}: GlobalAgentRosterProps): React.JSX.Element {
const queryClient = useQueryClient();
const [openProviders, setOpenProviders] = useState<Record<string, boolean>>({});
const sessionsQuery = useQuery<MissionControlSession[]>({
queryKey: SESSIONS_QUERY_KEY,
queryFn: fetchSessions,
refetchInterval: SESSIONS_POLL_INTERVAL_MS,
});
const killMutation = useMutation({
mutationFn: async (sessionId: string): Promise<string> => {
await apiPost<{ message: string }>(`/api/mission-control/sessions/${sessionId}/kill`, {
force: false,
});
return sessionId;
},
onSuccess: (): void => {
void queryClient.invalidateQueries({ queryKey: SESSIONS_QUERY_KEY });
},
});
const groupedSessions = useMemo(
() => groupByProvider(sessionsQuery.data ?? []),
[sessionsQuery.data]
);
const killAllSessions = useMemo(
() => toKillAllSessions(sessionsQuery.data ?? []),
[sessionsQuery.data]
);
const totalSessionCount = sessionsQuery.data?.length ?? 0;
const pendingKillSessionId = killMutation.isPending ? killMutation.variables : undefined;
const toggleProvider = (providerId: string): void => {
setOpenProviders((prev) => ({
...prev,
[providerId]: !(prev[providerId] ?? true),
}));
};
const isProviderOpen = (providerId: string): boolean => openProviders[providerId] ?? true;
const handleKillAllComplete = (): void => {
void queryClient.invalidateQueries({ queryKey: SESSIONS_QUERY_KEY });
};
return (
<Card className="flex h-full min-h-0 flex-col">
<CardHeader className="pb-2">
<CardTitle className="flex items-center justify-between gap-2 text-base">
<span>Agent Roster</span>
<div className="flex items-center gap-2">
{totalSessionCount > 0 ? (
<KillAllDialog sessions={killAllSessions} onComplete={handleKillAllComplete} />
) : null}
{sessionsQuery.isFetching && !sessionsQuery.isLoading ? (
<Loader2 className="h-4 w-4 animate-spin text-muted-foreground" aria-hidden="true" />
) : null}
</div>
</CardTitle>
</CardHeader>
<CardContent className="min-h-0 flex-1 px-3 pb-3">
{sessionsQuery.isLoading ? (
<ScrollArea className="h-full">
<div className="space-y-2 pr-1">
{Array.from({ length: 6 }).map((_, index) => (
<Skeleton key={`roster-skeleton-${String(index)}`} className="h-10 w-full" />
))}
</div>
</ScrollArea>
) : sessionsQuery.error ? (
<div className="flex h-full items-center justify-center text-center text-sm text-red-500">
Failed to load agents: {sessionsQuery.error.message}
</div>
) : groupedSessions.length === 0 ? (
<div className="flex h-full items-center justify-center text-sm text-muted-foreground">
No active agents
</div>
) : (
<ScrollArea className="h-full">
<div className="space-y-3 pr-1">
{groupedSessions.map((group) => {
const providerOpen = isProviderOpen(group.providerId);
return (
<Collapsible key={group.providerId} open={providerOpen} className="space-y-1">
<button
type="button"
onClick={() => {
toggleProvider(group.providerId);
}}
className="flex w-full items-center gap-2 rounded-md px-1 py-1 text-left text-sm hover:bg-muted/40"
aria-expanded={providerOpen}
>
<ChevronRight
className={`h-4 w-4 text-muted-foreground transition-transform ${providerOpen ? "rotate-90" : ""}`}
aria-hidden="true"
/>
<span className="font-medium">
{resolveProviderName(group.providerId, group.providerType)}
</span>
<span className="ml-auto text-xs text-muted-foreground">
{group.sessions.length}
</span>
</button>
{providerOpen ? (
<div className="space-y-1 pl-2">
{group.sessions.map((session) => {
const isSelected = selectedSessionId === session.id;
const isKilling = pendingKillSessionId === session.id;
return (
<div
key={session.id}
role="button"
tabIndex={0}
onClick={() => {
onSelectSession?.(session.id);
}}
onKeyDown={(event) => {
if (event.key === "Enter" || event.key === " ") {
event.preventDefault();
onSelectSession?.(session.id);
}
}}
className="flex items-center justify-between gap-2 rounded-md border border-transparent px-2 py-1.5 transition-colors hover:bg-muted/40"
style={
isSelected
? {
borderColor: "rgba(47, 128, 255, 0.35)",
backgroundColor: "rgba(47, 128, 255, 0.08)",
}
: undefined
}
>
<div className="flex min-w-0 items-center gap-2">
<span className="font-mono text-xs" title={session.id}>
{truncateSessionId(session.id)}
</span>
<Badge
variant={getStatusVariant(session.status)}
className="capitalize"
>
{session.status}
</Badge>
</div>
<Button
variant="ghost"
size="sm"
className="h-7 min-h-7 w-7 min-w-7 p-0"
disabled={isKilling}
onClick={(event) => {
event.stopPropagation();
killMutation.mutate(session.id);
}}
aria-label={`Kill session ${truncateSessionId(session.id)}`}
>
{isKilling ? (
<Loader2
className="h-3.5 w-3.5 animate-spin"
aria-hidden="true"
/>
) : (
<X className="h-3.5 w-3.5" aria-hidden="true" />
)}
</Button>
</div>
);
})}
</div>
) : null}
</Collapsible>
);
})}
</div>
</ScrollArea>
)}
</CardContent>
</Card>
);
}

View File

@@ -0,0 +1,170 @@
import { beforeEach, describe, expect, it, vi } from "vitest";
import { render, screen, waitFor } from "@testing-library/react";
import userEvent from "@testing-library/user-event";
import type {
ButtonHTMLAttributes,
InputHTMLAttributes,
LabelHTMLAttributes,
ReactNode,
} from "react";
interface MockButtonProps extends ButtonHTMLAttributes<HTMLButtonElement> {
children: ReactNode;
}
interface MockInputProps extends InputHTMLAttributes<HTMLInputElement> {
children?: ReactNode;
}
interface MockLabelProps extends LabelHTMLAttributes<HTMLLabelElement> {
children: ReactNode;
}
interface MockSession {
id: string;
providerId: string;
providerType: string;
status: "active" | "paused";
createdAt: Date;
updatedAt: Date;
}
const mockApiPost = vi.fn<(endpoint: string, body?: unknown) => Promise<{ message: string }>>();
vi.mock("@/lib/api/client", () => ({
apiPost: (endpoint: string, body?: unknown): Promise<{ message: string }> =>
mockApiPost(endpoint, body),
}));
vi.mock("@/components/ui/button", () => ({
Button: ({ children, ...props }: MockButtonProps): React.JSX.Element => (
<button {...props}>{children}</button>
),
}));
vi.mock("@/components/ui/input", () => ({
Input: ({ ...props }: MockInputProps): React.JSX.Element => <input {...props} />,
}));
vi.mock("@/components/ui/label", () => ({
Label: ({ children, ...props }: MockLabelProps): React.JSX.Element => (
<label {...props}>{children}</label>
),
}));
import { KillAllDialog } from "./KillAllDialog";
function makeSession(overrides: Partial<MockSession>): MockSession {
return {
id: "session-1",
providerId: "internal",
providerType: "internal",
status: "active",
createdAt: new Date("2026-03-07T10:00:00.000Z"),
updatedAt: new Date("2026-03-07T10:01:00.000Z"),
...overrides,
};
}
describe("KillAllDialog", (): void => {
beforeEach((): void => {
vi.clearAllMocks();
mockApiPost.mockResolvedValue({ message: "killed" });
});
it("renders trigger button and requires exact confirmation text", async (): Promise<void> => {
const user = userEvent.setup();
render(<KillAllDialog sessions={[makeSession({})]} />);
await user.click(screen.getByRole("button", { name: "Kill All" }));
const confirmInput = screen.getByLabelText("Type KILL ALL to confirm");
const confirmButton = screen.getByRole("button", { name: "Kill All Agents" });
expect(confirmButton).toBeDisabled();
await user.type(confirmInput, "kill all");
expect(confirmButton).toBeDisabled();
await user.clear(confirmInput);
await user.type(confirmInput, "KILL ALL");
expect(confirmButton).toBeEnabled();
});
it("kills only internal sessions by default and invokes completion callback", async (): Promise<void> => {
const onComplete = vi.fn<() => void>();
const user = userEvent.setup();
render(
<KillAllDialog
sessions={[
makeSession({ id: "internal-1", providerType: "internal" }),
makeSession({ id: "external-1", providerType: "external" }),
]}
onComplete={onComplete}
/>
);
await user.click(screen.getByRole("button", { name: "Kill All" }));
await user.type(screen.getByLabelText("Type KILL ALL to confirm"), "KILL ALL");
await user.click(screen.getByRole("button", { name: "Kill All Agents" }));
await waitFor((): void => {
expect(mockApiPost).toHaveBeenCalledWith("/api/mission-control/sessions/internal-1/kill", {
force: true,
});
});
expect(mockApiPost).not.toHaveBeenCalledWith("/api/mission-control/sessions/external-1/kill", {
force: true,
});
expect(onComplete).toHaveBeenCalledTimes(1);
});
it("kills all providers when all scope is selected", async (): Promise<void> => {
const user = userEvent.setup();
render(
<KillAllDialog
sessions={[
makeSession({ id: "internal-2", providerType: "internal" }),
makeSession({ id: "external-2", providerType: "external" }),
]}
/>
);
await user.click(screen.getByRole("button", { name: "Kill All" }));
await user.click(screen.getByRole("radio", { name: /All providers \(2\)/ }));
await user.type(screen.getByLabelText("Type KILL ALL to confirm"), "KILL ALL");
await user.click(screen.getByRole("button", { name: "Kill All Agents" }));
await waitFor((): void => {
expect(mockApiPost).toHaveBeenCalledWith("/api/mission-control/sessions/internal-2/kill", {
force: true,
});
expect(mockApiPost).toHaveBeenCalledWith("/api/mission-control/sessions/external-2/kill", {
force: true,
});
});
});
it("shows empty-scope warning when internal sessions are unavailable", async (): Promise<void> => {
const user = userEvent.setup();
render(
<KillAllDialog
sessions={[
makeSession({ id: "external-only", providerId: "ext", providerType: "external" }),
]}
/>
);
await user.click(screen.getByRole("button", { name: "Kill All" }));
await user.type(screen.getByLabelText("Type KILL ALL to confirm"), "KILL ALL");
expect(screen.getByText("No sessions in the selected scope.")).toBeInTheDocument();
expect(screen.getByRole("button", { name: "Kill All Agents" })).toBeDisabled();
});
});

View File

@@ -0,0 +1,224 @@
"use client";
import { useEffect, useMemo, useRef, useState } from "react";
import type { AgentSession } from "@mosaic/shared";
import { Loader2 } from "lucide-react";
import { Button } from "@/components/ui/button";
import {
Dialog,
DialogContent,
DialogDescription,
DialogFooter,
DialogHeader,
DialogTitle,
DialogTrigger,
} from "@/components/ui/dialog";
import { Input } from "@/components/ui/input";
import { Label } from "@/components/ui/label";
import { apiPost } from "@/lib/api/client";
const CONFIRM_TEXT = "KILL ALL";
const AUTO_CLOSE_DELAY_MS = 2_000;
type KillScope = "internal" | "all";
export interface KillAllDialogProps {
sessions: AgentSession[];
onComplete?: () => void;
}
export function KillAllDialog({ sessions, onComplete }: KillAllDialogProps): React.JSX.Element {
const [open, setOpen] = useState(false);
const [scope, setScope] = useState<KillScope>("internal");
const [confirmationInput, setConfirmationInput] = useState("");
const [isKilling, setIsKilling] = useState(false);
const [completedCount, setCompletedCount] = useState(0);
const [targetCount, setTargetCount] = useState(0);
const [successCount, setSuccessCount] = useState<number | null>(null);
const closeTimeoutRef = useRef<ReturnType<typeof setTimeout> | null>(null);
const internalSessions = useMemo(
() => sessions.filter((session) => session.providerType.toLowerCase() === "internal"),
[sessions]
);
const scopedSessions = useMemo(
() => (scope === "all" ? sessions : internalSessions),
[scope, sessions, internalSessions]
);
const hasConfirmation = confirmationInput === CONFIRM_TEXT;
const isConfirmDisabled =
isKilling || successCount !== null || !hasConfirmation || scopedSessions.length === 0;
useEffect((): (() => void) => {
return (): void => {
if (closeTimeoutRef.current !== null) {
clearTimeout(closeTimeoutRef.current);
}
};
}, []);
const resetState = (): void => {
setScope("internal");
setConfirmationInput("");
setIsKilling(false);
setCompletedCount(0);
setTargetCount(0);
setSuccessCount(null);
};
const handleOpenChange = (nextOpen: boolean): void => {
if (!nextOpen && isKilling) {
return;
}
if (!nextOpen) {
if (closeTimeoutRef.current !== null) {
clearTimeout(closeTimeoutRef.current);
}
resetState();
}
setOpen(nextOpen);
};
const handleKillAll = async (): Promise<void> => {
if (isConfirmDisabled) {
return;
}
const targetSessions = [...scopedSessions];
setIsKilling(true);
setCompletedCount(0);
setTargetCount(targetSessions.length);
setSuccessCount(null);
const killRequests = targetSessions.map(async (session) => {
try {
await apiPost<{ message: string }>(`/api/mission-control/sessions/${session.id}/kill`, {
force: true,
});
return true;
} catch {
return false;
} finally {
setCompletedCount((currentCount) => currentCount + 1);
}
});
const results = await Promise.all(killRequests);
const successfulKills = results.filter(Boolean).length;
setIsKilling(false);
setSuccessCount(successfulKills);
onComplete?.();
closeTimeoutRef.current = setTimeout(() => {
setOpen(false);
resetState();
}, AUTO_CLOSE_DELAY_MS);
};
return (
<Dialog open={open} onOpenChange={handleOpenChange}>
<DialogTrigger asChild>
<Button variant="danger" size="sm">
Kill All
</Button>
</DialogTrigger>
<DialogContent className="sm:max-w-[520px]">
<DialogHeader>
<DialogTitle>Kill All Agents</DialogTitle>
<DialogDescription>
This force-kills every selected agent session. This action cannot be undone.
</DialogDescription>
</DialogHeader>
<div className="space-y-4 py-2">
<fieldset className="space-y-2">
<legend className="text-sm font-medium">Scope</legend>
<label className="flex cursor-pointer items-center gap-2 text-sm text-foreground">
<input
type="radio"
name="kill-all-scope"
checked={scope === "internal"}
disabled={isKilling}
onChange={() => {
setScope("internal");
}}
/>
<span>Internal provider only ({internalSessions.length})</span>
</label>
<label className="flex cursor-pointer items-center gap-2 text-sm text-foreground">
<input
type="radio"
name="kill-all-scope"
checked={scope === "all"}
disabled={isKilling}
onChange={() => {
setScope("all");
}}
/>
<span>All providers ({sessions.length})</span>
</label>
</fieldset>
<div className="space-y-2">
<Label htmlFor="kill-all-confirmation-input">Type KILL ALL to confirm</Label>
<Input
id="kill-all-confirmation-input"
value={confirmationInput}
onChange={(event: React.ChangeEvent<HTMLInputElement>) => {
setConfirmationInput(event.target.value);
}}
placeholder={CONFIRM_TEXT}
autoComplete="off"
disabled={isKilling}
/>
</div>
{scopedSessions.length === 0 ? (
<p className="text-sm text-red-500">No sessions in the selected scope.</p>
) : null}
{isKilling ? (
<div className="flex items-center gap-2 text-sm text-muted-foreground">
<Loader2 className="h-4 w-4 animate-spin" aria-hidden="true" />
<span>
Killing {completedCount} of {targetCount} agents...
</span>
</div>
) : successCount !== null ? (
<p className="text-sm text-muted-foreground">
Killed {successCount} of {targetCount} agents. Closing...
</p>
) : null}
</div>
<DialogFooter>
<Button
type="button"
variant="outline"
disabled={isKilling}
onClick={() => {
setOpen(false);
}}
>
Cancel
</Button>
<Button
type="button"
variant="danger"
disabled={isConfirmDisabled}
onClick={() => {
void handleKillAll();
}}
>
Kill All Agents
</Button>
</DialogFooter>
</DialogContent>
</Dialog>
);
}

View File

@@ -0,0 +1,70 @@
import { render, screen } from "@testing-library/react";
import type { ButtonHTMLAttributes, ReactNode } from "react";
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
interface MockButtonProps extends ButtonHTMLAttributes<HTMLButtonElement> {
children: ReactNode;
}
const mockGlobalAgentRoster = vi.fn();
const mockMissionControlPanel = vi.fn();
vi.mock("@/components/mission-control/AuditLogDrawer", () => ({
AuditLogDrawer: ({ trigger }: { trigger: ReactNode }): React.JSX.Element => (
<div data-testid="audit-log-drawer">{trigger}</div>
),
}));
vi.mock("@/components/mission-control/GlobalAgentRoster", () => ({
GlobalAgentRoster: (props: unknown): React.JSX.Element => {
mockGlobalAgentRoster(props);
return <div data-testid="global-agent-roster" />;
},
}));
vi.mock("@/components/mission-control/MissionControlPanel", () => ({
MissionControlPanel: (props: unknown): React.JSX.Element => {
mockMissionControlPanel(props);
return <div data-testid="mission-control-panel" />;
},
MAX_PANEL_COUNT: 6,
MIN_PANEL_COUNT: 1,
}));
vi.mock("@/components/ui/button", () => ({
Button: ({ children, ...props }: MockButtonProps): React.JSX.Element => (
<button {...props}>{children}</button>
),
}));
import { MissionControlLayout } from "./MissionControlLayout";
describe("MissionControlLayout", (): void => {
beforeEach((): void => {
vi.clearAllMocks();
vi.stubGlobal("fetch", vi.fn());
});
afterEach((): void => {
vi.unstubAllGlobals();
});
it("renders without crashing", (): void => {
render(<MissionControlLayout />);
expect(screen.getByRole("region", { name: "Mission Control" })).toBeInTheDocument();
expect(screen.getByRole("button", { name: "Audit Log" })).toBeInTheDocument();
});
it("renders sidebar and panel grid container", (): void => {
render(<MissionControlLayout />);
const region = screen.getByRole("region", { name: "Mission Control" });
expect(region.querySelector(".grid")).toBeInTheDocument();
expect(region.querySelector("aside")).toBeInTheDocument();
expect(region.querySelector("main")).toBeInTheDocument();
expect(screen.getByTestId("global-agent-roster")).toBeInTheDocument();
expect(screen.getByTestId("mission-control-panel")).toBeInTheDocument();
});
});

View File

@@ -0,0 +1,115 @@
"use client";
import { useCallback, useState } from "react";
import { AuditLogDrawer } from "@/components/mission-control/AuditLogDrawer";
import { GlobalAgentRoster } from "@/components/mission-control/GlobalAgentRoster";
import {
MAX_PANEL_COUNT,
MIN_PANEL_COUNT,
MissionControlPanel,
type PanelConfig,
} from "@/components/mission-control/MissionControlPanel";
import { Button } from "@/components/ui/button";
const INITIAL_PANELS: PanelConfig[] = [{}];
export function MissionControlLayout(): React.JSX.Element {
const [panels, setPanels] = useState<PanelConfig[]>(INITIAL_PANELS);
const [selectedSessionId, setSelectedSessionId] = useState<string>();
const handleSelectSession = useCallback((sessionId: string): void => {
setSelectedSessionId(sessionId);
setPanels((currentPanels) => {
if (currentPanels.some((panel) => panel.sessionId === sessionId)) {
return currentPanels;
}
const firstEmptyPanelIndex = currentPanels.findIndex(
(panel) => panel.sessionId === undefined
);
if (firstEmptyPanelIndex >= 0) {
return currentPanels.map((panel, index) =>
index === firstEmptyPanelIndex ? { ...panel, sessionId } : panel
);
}
if (currentPanels.length >= MAX_PANEL_COUNT) {
return currentPanels;
}
return [...currentPanels, { sessionId }];
});
}, []);
const handleAddPanel = useCallback((): void => {
setPanels((currentPanels) => {
if (currentPanels.length >= MAX_PANEL_COUNT) {
return currentPanels;
}
return [...currentPanels, {}];
});
}, []);
const handleRemovePanel = useCallback((panelIndex: number): void => {
setPanels((currentPanels) => {
if (panelIndex < 0 || panelIndex >= currentPanels.length) {
return currentPanels;
}
if (currentPanels.length <= MIN_PANEL_COUNT) {
return currentPanels;
}
const nextPanels = currentPanels.filter((_, index) => index !== panelIndex);
return nextPanels.length === 0 ? INITIAL_PANELS : nextPanels;
});
}, []);
const handleExpandPanel = useCallback((panelIndex: number): void => {
setPanels((currentPanels) => {
if (panelIndex < 0 || panelIndex >= currentPanels.length) {
return currentPanels;
}
const shouldExpand = !currentPanels[panelIndex]?.expanded;
return currentPanels.map((panel, index) => ({
...panel,
expanded: shouldExpand && index === panelIndex,
}));
});
}, []);
return (
<section className="flex h-full min-h-0 flex-col overflow-hidden" aria-label="Mission Control">
<header className="mb-3 flex items-center justify-end">
<AuditLogDrawer
trigger={
<Button variant="outline" size="sm">
Audit Log
</Button>
}
/>
</header>
<div className="grid min-h-0 flex-1 gap-4 xl:grid-cols-[280px_minmax(0,1fr)]">
<aside className="h-full min-h-0">
<GlobalAgentRoster
onSelectSession={handleSelectSession}
{...(selectedSessionId !== undefined ? { selectedSessionId } : {})}
/>
</aside>
<main className="h-full min-h-0 overflow-hidden">
<MissionControlPanel
panels={panels}
onAddPanel={handleAddPanel}
onRemovePanel={handleRemovePanel}
onExpandPanel={handleExpandPanel}
/>
</main>
</div>
</section>
);
}

View File

@@ -0,0 +1,153 @@
import { beforeEach, describe, expect, it, vi } from "vitest";
import { fireEvent, render, screen, waitFor } from "@testing-library/react";
import type { ButtonHTMLAttributes, ReactNode } from "react";
import { MAX_PANEL_COUNT, MissionControlPanel, type PanelConfig } from "./MissionControlPanel";
interface MockButtonProps extends ButtonHTMLAttributes<HTMLButtonElement> {
children: ReactNode;
}
interface MockOrchestratorPanelProps {
sessionId?: string;
onClose?: () => void;
closeDisabled?: boolean;
onExpand?: () => void;
expanded?: boolean;
}
const mockOrchestratorPanel = vi.fn<(props: MockOrchestratorPanelProps) => React.JSX.Element>();
vi.mock("@/components/mission-control/OrchestratorPanel", () => ({
OrchestratorPanel: (props: MockOrchestratorPanelProps): React.JSX.Element =>
mockOrchestratorPanel(props),
}));
vi.mock("@/components/ui/button", () => ({
Button: ({ children, ...props }: MockButtonProps): React.JSX.Element => (
<button {...props}>{children}</button>
),
}));
function buildPanels(count: number): PanelConfig[] {
return Array.from({ length: count }, (_, index) => ({
sessionId: `session-${String(index + 1)}`,
}));
}
describe("MissionControlPanel", (): void => {
beforeEach((): void => {
vi.clearAllMocks();
mockOrchestratorPanel.mockImplementation(
({ sessionId, closeDisabled, expanded }: MockOrchestratorPanelProps): React.JSX.Element => (
<div
data-testid="orchestrator-panel"
data-session-id={sessionId ?? ""}
data-close-disabled={String(closeDisabled ?? false)}
data-expanded={String(expanded ?? false)}
/>
)
);
});
it("renders the panel grid and default heading", (): void => {
render(
<MissionControlPanel
panels={[{}]}
onAddPanel={vi.fn<() => void>()}
onRemovePanel={vi.fn<(index: number) => void>()}
onExpandPanel={vi.fn<(index: number) => void>()}
/>
);
expect(screen.getByRole("heading", { name: "Panels" })).toBeInTheDocument();
expect(screen.getAllByTestId("orchestrator-panel")).toHaveLength(1);
});
it("calls onAddPanel when the add button is clicked", (): void => {
const onAddPanel = vi.fn<() => void>();
render(
<MissionControlPanel
panels={[{}]}
onAddPanel={onAddPanel}
onRemovePanel={vi.fn<(index: number) => void>()}
onExpandPanel={vi.fn<(index: number) => void>()}
/>
);
fireEvent.click(screen.getByRole("button", { name: "Add panel" }));
expect(onAddPanel).toHaveBeenCalledTimes(1);
});
it("disables add panel at the configured maximum", (): void => {
render(
<MissionControlPanel
panels={buildPanels(MAX_PANEL_COUNT)}
onAddPanel={vi.fn<() => void>()}
onRemovePanel={vi.fn<(index: number) => void>()}
onExpandPanel={vi.fn<(index: number) => void>()}
/>
);
const addButton = screen.getByRole("button", { name: "Add panel" });
expect(addButton).toBeDisabled();
expect(addButton).toHaveAttribute("title", "Maximum of 6 panels");
});
it("passes closeDisabled=false when more than one panel exists", (): void => {
render(
<MissionControlPanel
panels={buildPanels(2)}
onAddPanel={vi.fn<() => void>()}
onRemovePanel={vi.fn<(index: number) => void>()}
onExpandPanel={vi.fn<(index: number) => void>()}
/>
);
const renderedPanels = screen.getAllByTestId("orchestrator-panel");
expect(renderedPanels).toHaveLength(2);
for (const panel of renderedPanels) {
expect(panel).toHaveAttribute("data-close-disabled", "false");
}
});
it("renders only the expanded panel in focused mode", (): void => {
render(
<MissionControlPanel
panels={[{ sessionId: "session-1" }, { sessionId: "session-2", expanded: true }]}
onAddPanel={vi.fn<() => void>()}
onRemovePanel={vi.fn<(index: number) => void>()}
onExpandPanel={vi.fn<(index: number) => void>()}
/>
);
const renderedPanels = screen.getAllByTestId("orchestrator-panel");
expect(renderedPanels).toHaveLength(1);
expect(renderedPanels[0]).toHaveAttribute("data-session-id", "session-2");
expect(renderedPanels[0]).toHaveAttribute("data-expanded", "true");
});
it("handles Escape key by toggling expanded panel", async (): Promise<void> => {
const onExpandPanel = vi.fn<(index: number) => void>();
render(
<MissionControlPanel
panels={[{ sessionId: "session-1", expanded: true }, { sessionId: "session-2" }]}
onAddPanel={vi.fn<() => void>()}
onRemovePanel={vi.fn<(index: number) => void>()}
onExpandPanel={onExpandPanel}
/>
);
fireEvent.keyDown(window, { key: "Escape" });
await waitFor((): void => {
expect(onExpandPanel).toHaveBeenCalledWith(0);
});
});
});

View File

@@ -0,0 +1,107 @@
"use client";
import { useEffect } from "react";
import { OrchestratorPanel } from "@/components/mission-control/OrchestratorPanel";
import { Button } from "@/components/ui/button";
export interface PanelConfig {
sessionId?: string;
expanded?: boolean;
}
interface MissionControlPanelProps {
panels: PanelConfig[];
onAddPanel: () => void;
onRemovePanel: (index: number) => void;
onExpandPanel: (index: number) => void;
}
export const MIN_PANEL_COUNT = 1;
export const MAX_PANEL_COUNT = 6;
export function MissionControlPanel({
panels,
onAddPanel,
onRemovePanel,
onExpandPanel,
}: MissionControlPanelProps): React.JSX.Element {
const expandedPanelIndex = panels.findIndex((panel) => panel.expanded);
const expandedPanel = expandedPanelIndex >= 0 ? panels[expandedPanelIndex] : undefined;
const canAddPanel = panels.length < MAX_PANEL_COUNT;
const canRemovePanel = panels.length > MIN_PANEL_COUNT;
useEffect(() => {
if (expandedPanelIndex < 0) {
return;
}
const handleKeyDown = (event: KeyboardEvent): void => {
if (event.key === "Escape") {
onExpandPanel(expandedPanelIndex);
}
};
window.addEventListener("keydown", handleKeyDown);
return (): void => {
window.removeEventListener("keydown", handleKeyDown);
};
}, [expandedPanelIndex, onExpandPanel]);
return (
<div className="flex h-full min-h-0 flex-col gap-3">
<div className="flex items-center justify-between">
<h2 className="text-sm font-medium text-muted-foreground">Panels</h2>
<Button
type="button"
variant="outline"
size="icon"
onClick={onAddPanel}
disabled={!canAddPanel}
aria-label="Add panel"
title={canAddPanel ? "Add panel" : "Maximum of 6 panels"}
>
<span aria-hidden="true" className="text-lg leading-none">
+
</span>
</Button>
</div>
<div className="min-h-0 flex-1">
{expandedPanelIndex >= 0 && expandedPanel ? (
<div className="h-full min-h-0">
<OrchestratorPanel
{...(expandedPanel.sessionId !== undefined
? { sessionId: expandedPanel.sessionId }
: {})}
onClose={() => {
onRemovePanel(expandedPanelIndex);
}}
closeDisabled={!canRemovePanel}
onExpand={() => {
onExpandPanel(expandedPanelIndex);
}}
expanded
/>
</div>
) : (
<div className="grid h-full min-h-0 auto-rows-fr grid-cols-1 gap-4 overflow-y-auto pr-1 md:grid-cols-2 xl:grid-cols-3">
{panels.map((panel, index) => (
<OrchestratorPanel
key={`panel-${String(index)}`}
{...(panel.sessionId !== undefined ? { sessionId: panel.sessionId } : {})}
onClose={() => {
onRemovePanel(index);
}}
closeDisabled={!canRemovePanel}
onExpand={() => {
onExpandPanel(index);
}}
expanded={panel.expanded ?? false}
/>
))}
</div>
)}
</div>
</div>
);
}

View File

@@ -0,0 +1,218 @@
import { afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
import { render, screen } from "@testing-library/react";
import type { ButtonHTMLAttributes, HTMLAttributes, ReactNode } from "react";
interface MockButtonProps extends ButtonHTMLAttributes<HTMLButtonElement> {
children: ReactNode;
}
interface MockContainerProps extends HTMLAttributes<HTMLElement> {
children: ReactNode;
}
type MockConnectionStatus = "connected" | "connecting" | "error";
type MockRole = "user" | "assistant" | "tool" | "system";
interface MockMessage {
id: string;
role: MockRole;
content: string;
timestamp: string;
}
interface MockSession {
id: string;
status: string;
}
interface MockSessionStreamResult {
messages: MockMessage[];
status: MockConnectionStatus;
error: string | null;
}
interface MockSessionsResult {
sessions: MockSession[];
loading: boolean;
error: Error | null;
}
interface MockPanelControlsProps {
sessionId: string;
status: string;
onStatusChange?: (nextStatus: string) => void;
}
interface MockBargeInInputProps {
sessionId: string;
}
const mockUseSessionStream = vi.fn<(sessionId: string) => MockSessionStreamResult>();
const mockUseSessions = vi.fn<() => MockSessionsResult>();
const mockPanelControls = vi.fn<(props: MockPanelControlsProps) => React.JSX.Element>();
const mockBargeInInput = vi.fn<(props: MockBargeInInputProps) => React.JSX.Element>();
vi.mock("date-fns", () => ({
formatDistanceToNow: (): string => "moments ago",
}));
vi.mock("@/hooks/useMissionControl", () => ({
useSessionStream: (sessionId: string): MockSessionStreamResult => mockUseSessionStream(sessionId),
useSessions: (): MockSessionsResult => mockUseSessions(),
}));
vi.mock("@/components/mission-control/PanelControls", () => ({
PanelControls: (props: MockPanelControlsProps): React.JSX.Element => mockPanelControls(props),
}));
vi.mock("@/components/mission-control/BargeInInput", () => ({
BargeInInput: (props: MockBargeInInputProps): React.JSX.Element => mockBargeInInput(props),
}));
vi.mock("@/components/ui/button", () => ({
Button: ({ children, ...props }: MockButtonProps): React.JSX.Element => (
<button {...props}>{children}</button>
),
}));
vi.mock("@/components/ui/badge", () => ({
Badge: ({ children, ...props }: MockContainerProps): React.JSX.Element => (
<span {...props}>{children}</span>
),
}));
vi.mock("@/components/ui/card", () => ({
Card: ({ children, ...props }: MockContainerProps): React.JSX.Element => (
<section {...props}>{children}</section>
),
CardHeader: ({ children, ...props }: MockContainerProps): React.JSX.Element => (
<header {...props}>{children}</header>
),
CardContent: ({ children, ...props }: MockContainerProps): React.JSX.Element => (
<div {...props}>{children}</div>
),
CardTitle: ({ children, ...props }: MockContainerProps): React.JSX.Element => (
<h2 {...props}>{children}</h2>
),
}));
vi.mock("@/components/ui/scroll-area", () => ({
ScrollArea: ({ children, ...props }: MockContainerProps): React.JSX.Element => (
<div {...props}>{children}</div>
),
}));
import { OrchestratorPanel } from "./OrchestratorPanel";
beforeAll((): void => {
Object.defineProperty(window.HTMLElement.prototype, "scrollIntoView", {
configurable: true,
value: vi.fn(),
});
});
describe("OrchestratorPanel", (): void => {
beforeEach((): void => {
vi.clearAllMocks();
vi.stubGlobal("fetch", vi.fn());
mockUseSessionStream.mockReturnValue({
messages: [],
status: "connecting",
error: null,
});
mockUseSessions.mockReturnValue({
sessions: [],
loading: false,
error: null,
});
mockPanelControls.mockImplementation(
({ status }: MockPanelControlsProps): React.JSX.Element => (
<div data-testid="panel-controls">status:{status}</div>
)
);
mockBargeInInput.mockImplementation(
({ sessionId }: MockBargeInInputProps): React.JSX.Element => (
<textarea aria-label="barge-input" data-session-id={sessionId} />
)
);
});
afterEach((): void => {
vi.unstubAllGlobals();
});
it("renders a selectable empty state when no session is provided", (): void => {
render(<OrchestratorPanel />);
expect(screen.getByText("Select an agent to view its stream")).toBeInTheDocument();
expect(screen.queryByText("Session: session-1")).not.toBeInTheDocument();
});
it("renders connection indicator and panel controls for an active session", (): void => {
mockUseSessionStream.mockReturnValue({
messages: [],
status: "connected",
error: null,
});
mockUseSessions.mockReturnValue({
sessions: [{ id: "session-1", status: "paused" }],
loading: false,
error: null,
});
render(<OrchestratorPanel sessionId="session-1" />);
expect(screen.getByText("Connected")).toBeInTheDocument();
expect(screen.getByText("Session: session-1")).toBeInTheDocument();
expect(screen.getByText("Waiting for messages...")).toBeInTheDocument();
expect(screen.getByTestId("panel-controls")).toHaveTextContent("status:paused");
});
it("renders stream messages with role and content", (): void => {
mockUseSessionStream.mockReturnValue({
status: "connected",
error: null,
messages: [
{
id: "msg-1",
role: "assistant",
content: "Mission accepted.",
timestamp: "2026-03-07T18:42:00.000Z",
},
],
});
render(<OrchestratorPanel sessionId="session-2" />);
expect(screen.getByText("assistant")).toBeInTheDocument();
expect(screen.getByText("Mission accepted.")).toBeInTheDocument();
expect(screen.getByText("moments ago")).toBeInTheDocument();
expect(screen.getByLabelText("barge-input")).toHaveAttribute("data-session-id", "session-2");
});
it("renders stream error text when the session has no messages", (): void => {
mockUseSessionStream.mockReturnValue({
messages: [],
status: "error",
error: "Mission Control stream disconnected.",
});
render(<OrchestratorPanel sessionId="session-3" />);
expect(screen.getByText("Error")).toBeInTheDocument();
expect(screen.getByText("Mission Control stream disconnected.")).toBeInTheDocument();
});
it("respects close button disabled state in panel actions", (): void => {
const onClose = vi.fn<() => void>();
render(<OrchestratorPanel onClose={onClose} closeDisabled />);
expect(screen.getByRole("button", { name: "Remove panel" })).toBeDisabled();
});
});

View File

@@ -0,0 +1,217 @@
"use client";
import { useEffect, useRef, useState } from "react";
import { formatDistanceToNow } from "date-fns";
import { BargeInInput } from "@/components/mission-control/BargeInInput";
import { Badge } from "@/components/ui/badge";
import type { BadgeVariant } from "@/components/ui/badge";
import { Button } from "@/components/ui/button";
import { Card, CardContent, CardHeader, CardTitle } from "@/components/ui/card";
import { PanelControls } from "@/components/mission-control/PanelControls";
import { ScrollArea } from "@/components/ui/scroll-area";
import {
useSessionStream,
useSessions,
type MissionControlConnectionStatus,
type MissionControlMessageRole,
} from "@/hooks/useMissionControl";
const ROLE_BADGE_VARIANT: Record<MissionControlMessageRole, BadgeVariant> = {
user: "badge-blue",
assistant: "status-success",
tool: "badge-amber",
system: "badge-muted",
};
const CONNECTION_DOT_CLASS: Record<MissionControlConnectionStatus, string> = {
connected: "bg-emerald-500",
connecting: "bg-amber-500",
error: "bg-red-500",
};
const CONNECTION_TEXT: Record<MissionControlConnectionStatus, string> = {
connected: "Connected",
connecting: "Connecting",
error: "Error",
};
export interface OrchestratorPanelProps {
sessionId?: string;
onClose?: () => void;
closeDisabled?: boolean;
onExpand?: () => void;
expanded?: boolean;
}
interface PanelHeaderActionsProps {
onClose?: () => void;
closeDisabled?: boolean;
onExpand?: () => void;
expanded?: boolean;
}
function PanelHeaderActions({
onClose,
closeDisabled = false,
onExpand,
expanded = false,
}: PanelHeaderActionsProps): React.JSX.Element | null {
if (!onClose && !onExpand) {
return null;
}
return (
<div className="flex items-center gap-1">
{onExpand ? (
<Button
type="button"
variant="ghost"
size="icon"
className="h-7 w-7"
onClick={onExpand}
aria-label={expanded ? "Collapse panel" : "Expand panel"}
title={expanded ? "Collapse panel" : "Expand panel"}
>
<span aria-hidden="true" className="text-base leading-none">
{expanded ? "↙" : "↗"}
</span>
</Button>
) : null}
{onClose ? (
<Button
type="button"
variant="ghost"
size="icon"
className="h-7 w-7"
onClick={onClose}
disabled={closeDisabled}
aria-label="Remove panel"
title="Remove panel"
>
<span aria-hidden="true" className="text-base leading-none">
×
</span>
</Button>
) : null}
</div>
);
}
function formatRelativeTimestamp(timestamp: string): string {
const parsedDate = new Date(timestamp);
if (Number.isNaN(parsedDate.getTime())) {
return "just now";
}
return formatDistanceToNow(parsedDate, { addSuffix: true });
}
export function OrchestratorPanel({
sessionId,
onClose,
closeDisabled,
onExpand,
expanded,
}: OrchestratorPanelProps): React.JSX.Element {
const { messages, status, error } = useSessionStream(sessionId ?? "");
const { sessions } = useSessions();
const bottomAnchorRef = useRef<HTMLDivElement | null>(null);
const [optimisticStatus, setOptimisticStatus] = useState<string | null>(null);
const selectedSessionStatus = sessions.find((session) => session.id === sessionId)?.status;
const controlsStatus = optimisticStatus ?? selectedSessionStatus ?? "unknown";
const panelHeaderActionProps = {
...(onClose !== undefined ? { onClose } : {}),
...(closeDisabled !== undefined ? { closeDisabled } : {}),
...(onExpand !== undefined ? { onExpand } : {}),
...(expanded !== undefined ? { expanded } : {}),
};
useEffect(() => {
bottomAnchorRef.current?.scrollIntoView({ block: "end" });
}, [messages.length]);
useEffect(() => {
setOptimisticStatus(null);
}, [sessionId, selectedSessionStatus]);
if (!sessionId) {
return (
<Card className="flex h-full min-h-[220px] flex-col">
<CardHeader>
<div className="flex items-start justify-between gap-2">
<CardTitle className="text-base">Orchestrator Panel</CardTitle>
<PanelHeaderActions {...panelHeaderActionProps} />
</div>
</CardHeader>
<CardContent className="flex flex-1 items-center justify-center text-sm text-muted-foreground">
Select an agent to view its stream
</CardContent>
</Card>
);
}
return (
<Card className="flex h-full min-h-[220px] flex-col">
<CardHeader className="space-y-2">
<div className="flex items-start justify-between gap-2">
<CardTitle className="text-base">Orchestrator Panel</CardTitle>
<PanelHeaderActions {...panelHeaderActionProps} />
</div>
<div className="flex flex-col gap-2 sm:flex-row sm:items-start sm:justify-between">
<div className="flex items-center gap-2 text-xs text-muted-foreground">
<span
className={`h-2.5 w-2.5 rounded-full ${CONNECTION_DOT_CLASS[status]} ${
status === "connecting" ? "animate-pulse" : ""
}`}
aria-hidden="true"
/>
<span>{CONNECTION_TEXT[status]}</span>
</div>
<PanelControls
sessionId={sessionId}
status={controlsStatus}
onStatusChange={setOptimisticStatus}
/>
</div>
<p className="truncate text-xs text-muted-foreground">Session: {sessionId}</p>
</CardHeader>
<CardContent className="flex min-h-0 flex-1 flex-col p-0">
<div className="min-h-0 flex-1">
<ScrollArea className="h-full w-full">
<div className="flex min-h-full flex-col gap-3 p-4">
{messages.length === 0 ? (
<p className="mt-6 text-center text-sm text-muted-foreground">
{error ?? "Waiting for messages..."}
</p>
) : (
messages.map((message) => (
<article
key={message.id}
className="rounded-lg border border-border/70 bg-card px-3 py-2"
>
<div className="mb-2 flex items-center justify-between gap-2">
<Badge variant={ROLE_BADGE_VARIANT[message.role]} className="uppercase">
{message.role}
</Badge>
<time className="text-xs text-muted-foreground">
{formatRelativeTimestamp(message.timestamp)}
</time>
</div>
<p className="whitespace-pre-wrap break-words text-sm text-foreground">
{message.content}
</p>
</article>
))
)}
<div ref={bottomAnchorRef} />
</div>
</ScrollArea>
</div>
<div className="border-t border-border/70 p-3">
<BargeInInput sessionId={sessionId} />
</div>
</CardContent>
</Card>
);
}

View File

@@ -0,0 +1,161 @@
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { render, screen, waitFor } from "@testing-library/react";
import userEvent from "@testing-library/user-event";
import { QueryClient, QueryClientProvider } from "@tanstack/react-query";
import type { ButtonHTMLAttributes, HTMLAttributes, ReactNode } from "react";
interface MockButtonProps extends ButtonHTMLAttributes<HTMLButtonElement> {
children: ReactNode;
}
interface MockBadgeProps extends HTMLAttributes<HTMLElement> {
children: ReactNode;
}
const mockApiPost = vi.fn<(endpoint: string, body?: unknown) => Promise<{ message: string }>>();
vi.mock("@/lib/api/client", () => ({
apiPost: (endpoint: string, body?: unknown): Promise<{ message: string }> =>
mockApiPost(endpoint, body),
}));
vi.mock("@/components/ui/button", () => ({
Button: ({ children, ...props }: MockButtonProps): React.JSX.Element => (
<button {...props}>{children}</button>
),
}));
vi.mock("@/components/ui/badge", () => ({
Badge: ({ children, ...props }: MockBadgeProps): React.JSX.Element => (
<span {...props}>{children}</span>
),
}));
import { PanelControls } from "./PanelControls";
function renderWithQueryClient(ui: React.JSX.Element): ReturnType<typeof render> {
const queryClient = new QueryClient({
defaultOptions: {
queries: { retry: false },
mutations: { retry: false },
},
});
return render(<QueryClientProvider client={queryClient}>{ui}</QueryClientProvider>);
}
describe("PanelControls", (): void => {
beforeEach((): void => {
vi.clearAllMocks();
vi.stubGlobal("fetch", vi.fn());
mockApiPost.mockResolvedValue({ message: "ok" });
});
afterEach((): void => {
vi.unstubAllGlobals();
});
it("renders action buttons with correct disabled state for active sessions", (): void => {
renderWithQueryClient(<PanelControls sessionId="session-1" status="active" />);
expect(screen.getByRole("button", { name: "Pause session" })).toBeEnabled();
expect(screen.getByRole("button", { name: "Resume session" })).toBeDisabled();
expect(screen.getByRole("button", { name: "Gracefully kill session" })).toBeEnabled();
expect(screen.getByRole("button", { name: "Force kill session" })).toBeEnabled();
});
it("disables all action buttons when session is already killed", (): void => {
renderWithQueryClient(<PanelControls sessionId="session-2" status="killed" />);
expect(screen.getByRole("button", { name: "Pause session" })).toBeDisabled();
expect(screen.getByRole("button", { name: "Resume session" })).toBeDisabled();
expect(screen.getByRole("button", { name: "Gracefully kill session" })).toBeDisabled();
expect(screen.getByRole("button", { name: "Force kill session" })).toBeDisabled();
});
it("pauses a running session and reports the next status", async (): Promise<void> => {
const onStatusChange = vi.fn<(status: string) => void>();
const user = userEvent.setup();
renderWithQueryClient(
<PanelControls
sessionId="session with space"
status="active"
onStatusChange={onStatusChange}
/>
);
await user.click(screen.getByRole("button", { name: "Pause session" }));
await waitFor((): void => {
expect(mockApiPost).toHaveBeenCalledWith(
"/api/mission-control/sessions/session%20with%20space/pause",
undefined
);
});
expect(onStatusChange).toHaveBeenCalledWith("paused");
});
it("asks for graceful kill confirmation before submitting", async (): Promise<void> => {
const onStatusChange = vi.fn<(status: string) => void>();
const user = userEvent.setup();
renderWithQueryClient(
<PanelControls sessionId="session-4" status="active" onStatusChange={onStatusChange} />
);
await user.click(screen.getByRole("button", { name: "Gracefully kill session" }));
expect(
screen.getByText("Gracefully stop this agent after it finishes the current step?")
).toBeInTheDocument();
await user.click(screen.getByRole("button", { name: "Confirm" }));
await waitFor((): void => {
expect(mockApiPost).toHaveBeenCalledWith("/api/mission-control/sessions/session-4/kill", {
force: false,
});
});
expect(onStatusChange).toHaveBeenCalledWith("killed");
});
it("sends force kill after confirmation", async (): Promise<void> => {
const onStatusChange = vi.fn<(status: string) => void>();
const user = userEvent.setup();
renderWithQueryClient(
<PanelControls sessionId="session-5" status="paused" onStatusChange={onStatusChange} />
);
await user.click(screen.getByRole("button", { name: "Force kill session" }));
expect(screen.getByText("This will hard-kill the agent immediately.")).toBeInTheDocument();
await user.click(screen.getByRole("button", { name: "Confirm" }));
await waitFor((): void => {
expect(mockApiPost).toHaveBeenCalledWith("/api/mission-control/sessions/session-5/kill", {
force: true,
});
});
expect(onStatusChange).toHaveBeenCalledWith("killed");
});
it("shows an error badge when an action fails", async (): Promise<void> => {
const user = userEvent.setup();
mockApiPost.mockRejectedValueOnce(new Error("unable to pause"));
renderWithQueryClient(<PanelControls sessionId="session-6" status="active" />);
await user.click(screen.getByRole("button", { name: "Pause session" }));
await waitFor((): void => {
expect(screen.getByText("unable to pause")).toBeInTheDocument();
});
});
});

View File

@@ -0,0 +1,259 @@
"use client";
import { useEffect, useState } from "react";
import { useMutation, useQueryClient } from "@tanstack/react-query";
import { Loader2 } from "lucide-react";
import { Badge } from "@/components/ui/badge";
import { Button } from "@/components/ui/button";
import { apiPost } from "@/lib/api/client";
const SESSIONS_QUERY_KEY = ["mission-control", "sessions"] as const;
type PanelAction = "pause" | "resume" | "graceful-kill" | "force-kill";
type KillConfirmationState = "graceful" | "force" | null;
interface PanelActionResult {
nextStatus: string;
}
export interface PanelControlsProps {
sessionId: string;
// eslint-disable-next-line @typescript-eslint/no-redundant-type-constituents
status: "active" | "paused" | "killed" | string;
onStatusChange?: (newStatus: string) => void;
}
function getErrorMessage(error: unknown): string {
if (error instanceof Error && error.message.trim().length > 0) {
return error.message;
}
return "Failed to update agent session.";
}
export function PanelControls({
sessionId,
status,
onStatusChange,
}: PanelControlsProps): React.JSX.Element {
const queryClient = useQueryClient();
const [errorMessage, setErrorMessage] = useState<string | null>(null);
const [confirmingKill, setConfirmingKill] = useState<KillConfirmationState>(null);
useEffect(() => {
setErrorMessage(null);
setConfirmingKill(null);
}, [sessionId]);
const controlMutation = useMutation({
mutationFn: async (action: PanelAction): Promise<PanelActionResult> => {
switch (action) {
case "pause":
await apiPost<{ message: string }>(
`/api/mission-control/sessions/${encodeURIComponent(sessionId)}/pause`
);
return { nextStatus: "paused" };
case "resume":
await apiPost<{ message: string }>(
`/api/mission-control/sessions/${encodeURIComponent(sessionId)}/resume`
);
return { nextStatus: "active" };
case "graceful-kill":
await apiPost<{ message: string }>(
`/api/mission-control/sessions/${encodeURIComponent(sessionId)}/kill`,
{ force: false }
);
return { nextStatus: "killed" };
case "force-kill":
await apiPost<{ message: string }>(
`/api/mission-control/sessions/${encodeURIComponent(sessionId)}/kill`,
{ force: true }
);
return { nextStatus: "killed" };
}
},
onSuccess: ({ nextStatus }): void => {
setErrorMessage(null);
setConfirmingKill(null);
onStatusChange?.(nextStatus);
void queryClient.invalidateQueries({ queryKey: SESSIONS_QUERY_KEY });
},
onError: (error: unknown): void => {
setConfirmingKill(null);
setErrorMessage(getErrorMessage(error));
},
});
const normalizedStatus = status.toLowerCase();
const isKilled = normalizedStatus === "killed";
const isBusy = controlMutation.isPending;
const pendingAction = isBusy ? controlMutation.variables : undefined;
const submitAction = (action: PanelAction): void => {
setErrorMessage(null);
controlMutation.mutate(action);
};
const pauseDisabled = isBusy || normalizedStatus === "paused" || isKilled;
const resumeDisabled = isBusy || normalizedStatus === "active" || isKilled;
const gracefulKillDisabled = isBusy || isKilled;
const forceKillDisabled = isBusy || isKilled;
return (
<div className="flex flex-col items-end gap-2">
<div className="flex flex-wrap items-center justify-end gap-1.5">
<Button
type="button"
size="sm"
variant="secondary"
onClick={() => {
submitAction("pause");
}}
disabled={pauseDisabled}
aria-label="Pause session"
>
{pendingAction === "pause" ? (
<Loader2 className="h-3.5 w-3.5 animate-spin" aria-hidden="true" />
) : (
<span aria-hidden="true"></span>
)}
<span>Pause</span>
</Button>
<Button
type="button"
size="sm"
variant="secondary"
onClick={() => {
submitAction("resume");
}}
disabled={resumeDisabled}
aria-label="Resume session"
>
{pendingAction === "resume" ? (
<Loader2 className="h-3.5 w-3.5 animate-spin" aria-hidden="true" />
) : (
<span aria-hidden="true"></span>
)}
<span>Resume</span>
</Button>
<div className="relative">
<Button
type="button"
size="sm"
variant="secondary"
onClick={() => {
setErrorMessage(null);
setConfirmingKill((current) => (current === "graceful" ? null : "graceful"));
}}
disabled={gracefulKillDisabled}
aria-label="Gracefully kill session"
>
{pendingAction === "graceful-kill" ? (
<Loader2 className="h-3.5 w-3.5 animate-spin" aria-hidden="true" />
) : (
<span aria-hidden="true"></span>
)}
<span>Graceful Kill</span>
</Button>
{confirmingKill === "graceful" ? (
<div className="absolute right-0 top-[calc(100%+0.375rem)] z-20 w-72 rounded-md border border-border bg-card p-2 shadow-lg">
<p className="text-xs text-muted-foreground">
Gracefully stop this agent after it finishes the current step?
</p>
<div className="mt-2 flex justify-end gap-1.5">
<Button
type="button"
size="sm"
variant="ghost"
onClick={() => {
setConfirmingKill(null);
}}
disabled={isBusy}
>
Cancel
</Button>
<Button
type="button"
size="sm"
variant="secondary"
onClick={() => {
submitAction("graceful-kill");
}}
disabled={isBusy}
>
{pendingAction === "graceful-kill" ? (
<Loader2 className="h-3.5 w-3.5 animate-spin" aria-hidden="true" />
) : null}
<span>Confirm</span>
</Button>
</div>
</div>
) : null}
</div>
<div className="relative">
<Button
type="button"
size="sm"
variant="danger"
onClick={() => {
setErrorMessage(null);
setConfirmingKill((current) => (current === "force" ? null : "force"));
}}
disabled={forceKillDisabled}
aria-label="Force kill session"
>
{pendingAction === "force-kill" ? (
<Loader2 className="h-3.5 w-3.5 animate-spin" aria-hidden="true" />
) : (
<span aria-hidden="true">💀</span>
)}
<span>Force Kill</span>
</Button>
{confirmingKill === "force" ? (
<div className="absolute right-0 top-[calc(100%+0.375rem)] z-20 w-72 rounded-md border border-border bg-card p-2 shadow-lg">
<p className="text-xs text-muted-foreground">
This will hard-kill the agent immediately.
</p>
<div className="mt-2 flex justify-end gap-1.5">
<Button
type="button"
size="sm"
variant="ghost"
onClick={() => {
setConfirmingKill(null);
}}
disabled={isBusy}
>
Cancel
</Button>
<Button
type="button"
size="sm"
variant="danger"
onClick={() => {
submitAction("force-kill");
}}
disabled={isBusy}
>
{pendingAction === "force-kill" ? (
<Loader2 className="h-3.5 w-3.5 animate-spin" aria-hidden="true" />
) : null}
<span>Confirm</span>
</Button>
</div>
</div>
) : null}
</div>
</div>
{errorMessage ? (
<Badge variant="status-error" className="max-w-[32rem] whitespace-normal text-xs">
{errorMessage}
</Badge>
) : null}
</div>
);
}

View File

@@ -102,5 +102,5 @@ describe("OnboardingWizard", () => {
await waitFor(() => {
expect(mockPush).toHaveBeenCalledWith("/");
});
});
}, 10_000);
});

View File

@@ -0,0 +1,13 @@
import * as React from "react";
export interface CollapsibleProps extends React.HTMLAttributes<HTMLDivElement> {
open?: boolean;
}
export function Collapsible({
open = true,
className = "",
...props
}: CollapsibleProps): React.JSX.Element {
return <div data-state={open ? "open" : "closed"} className={className} {...props} />;
}

View File

@@ -0,0 +1,15 @@
import * as React from "react";
export type ScrollAreaProps = React.HTMLAttributes<HTMLDivElement>;
export const ScrollArea = React.forwardRef<HTMLDivElement, ScrollAreaProps>(
({ className = "", children, ...props }, ref) => {
return (
<div ref={ref} className={`relative overflow-hidden ${className}`} {...props}>
<div className="h-full w-full overflow-auto">{children}</div>
</div>
);
}
);
ScrollArea.displayName = "ScrollArea";

View File

@@ -0,0 +1,137 @@
import * as React from "react";
import { X } from "lucide-react";
export interface SheetProps {
open?: boolean;
onOpenChange?: (open: boolean) => void;
children?: React.ReactNode;
}
export interface SheetTriggerProps {
children?: React.ReactNode;
asChild?: boolean;
}
export interface SheetContentProps {
children?: React.ReactNode;
className?: string;
}
export interface SheetHeaderProps {
children?: React.ReactNode;
className?: string;
}
export interface SheetTitleProps {
children?: React.ReactNode;
className?: string;
}
export interface SheetDescriptionProps {
children?: React.ReactNode;
className?: string;
}
const SheetContext = React.createContext<{
open?: boolean;
onOpenChange?: (open: boolean) => void;
}>({});
export function Sheet({ open, onOpenChange, children }: SheetProps): React.JSX.Element {
const contextValue: { open?: boolean; onOpenChange?: (open: boolean) => void } = {};
if (open !== undefined) {
contextValue.open = open;
}
if (onOpenChange !== undefined) {
contextValue.onOpenChange = onOpenChange;
}
return <SheetContext.Provider value={contextValue}>{children}</SheetContext.Provider>;
}
export function SheetTrigger({ children, asChild }: SheetTriggerProps): React.JSX.Element {
const { onOpenChange } = React.useContext(SheetContext);
if (asChild && React.isValidElement(children)) {
return React.cloneElement(children, {
onClick: () => onOpenChange?.(true),
} as React.HTMLAttributes<HTMLElement>);
}
return (
<button type="button" onClick={() => onOpenChange?.(true)}>
{children}
</button>
);
}
export function SheetContent({
children,
className = "",
}: SheetContentProps): React.JSX.Element | null {
const { open, onOpenChange } = React.useContext(SheetContext);
React.useEffect(() => {
if (!open) {
return;
}
const onKeyDown = (event: KeyboardEvent): void => {
if (event.key === "Escape") {
onOpenChange?.(false);
}
};
window.addEventListener("keydown", onKeyDown);
return (): void => {
window.removeEventListener("keydown", onKeyDown);
};
}, [onOpenChange, open]);
if (!open) {
return null;
}
return (
<div className="fixed inset-0 z-50">
<button
type="button"
aria-label="Close sheet"
className="absolute inset-0 h-full w-full bg-black/50"
onClick={() => onOpenChange?.(false)}
/>
<div
className={`absolute inset-y-0 right-0 z-10 flex h-full w-full max-w-3xl flex-col border-l border-border bg-background p-6 shadow-xl ${className}`}
>
<button
type="button"
onClick={() => onOpenChange?.(false)}
className="absolute right-4 top-4 rounded-sm opacity-70 transition-opacity hover:opacity-100"
>
<X className="h-4 w-4" />
<span className="sr-only">Close</span>
</button>
{children}
</div>
</div>
);
}
export function SheetHeader({ children, className = "" }: SheetHeaderProps): React.JSX.Element {
return <div className={`space-y-1 pr-8 ${className}`}>{children}</div>;
}
export function SheetTitle({ children, className = "" }: SheetTitleProps): React.JSX.Element {
return <h2 className={`text-lg font-semibold ${className}`}>{children}</h2>;
}
export function SheetDescription({
children,
className = "",
}: SheetDescriptionProps): React.JSX.Element {
return <p className={`text-sm text-muted-foreground ${className}`}>{children}</p>;
}

View File

@@ -0,0 +1,15 @@
import * as React from "react";
export type SkeletonProps = React.HTMLAttributes<HTMLDivElement>;
export const Skeleton = React.forwardRef<HTMLDivElement, SkeletonProps>(
({ className = "", ...props }, ref) => (
<div
ref={ref}
className={`animate-pulse rounded-md bg-[rgb(var(--surface-2))] ${className}`}
{...props}
/>
)
);
Skeleton.displayName = "Skeleton";

View File

@@ -0,0 +1,189 @@
"use client";
import { useEffect, useRef, useState } from "react";
import { useQuery } from "@tanstack/react-query";
import type { AgentMessageRole, AgentSessionStatus } from "@mosaic/shared";
import { apiGet } from "@/lib/api/client";
const MISSION_CONTROL_SESSIONS_QUERY_KEY = ["mission-control", "sessions"] as const;
const SESSIONS_REFRESH_INTERVAL_MS = 15_000;
export type MissionControlMessageRole = AgentMessageRole;
export interface MissionControlSession {
id: string;
providerId: string;
providerType: string;
label?: string;
status: AgentSessionStatus;
parentSessionId?: string;
createdAt: string;
updatedAt: string;
metadata?: Record<string, unknown>;
}
interface MissionControlSessionsResponse {
sessions: MissionControlSession[];
}
export interface MissionControlStreamMessage {
id: string;
sessionId: string;
role: MissionControlMessageRole;
content: string;
timestamp: string;
metadata?: Record<string, unknown>;
}
export type MissionControlConnectionStatus = "connecting" | "connected" | "error";
export interface UseSessionsResult {
sessions: MissionControlSession[];
loading: boolean;
error: Error | null;
}
export interface UseSessionStreamResult {
messages: MissionControlStreamMessage[];
status: MissionControlConnectionStatus;
error: string | null;
}
function isRecord(value: unknown): value is Record<string, unknown> {
return typeof value === "object" && value !== null;
}
function isMessageRole(value: unknown): value is MissionControlMessageRole {
return value === "assistant" || value === "system" || value === "tool" || value === "user";
}
function isMissionControlStreamMessage(value: unknown): value is MissionControlStreamMessage {
if (!isRecord(value)) {
return false;
}
const { id, sessionId, role, content, timestamp, metadata } = value;
if (
typeof id !== "string" ||
typeof sessionId !== "string" ||
!isMessageRole(role) ||
typeof content !== "string" ||
typeof timestamp !== "string"
) {
return false;
}
if (metadata !== undefined && !isRecord(metadata)) {
return false;
}
return true;
}
/**
* Fetches Mission Control sessions.
*/
export function useSessions(): UseSessionsResult {
const query = useQuery<MissionControlSessionsResponse>({
queryKey: MISSION_CONTROL_SESSIONS_QUERY_KEY,
queryFn: async (): Promise<MissionControlSessionsResponse> => {
return apiGet<MissionControlSessionsResponse>("/api/mission-control/sessions");
},
refetchInterval: SESSIONS_REFRESH_INTERVAL_MS,
});
return {
sessions: query.data?.sessions ?? [],
loading: query.isLoading,
error: query.error ?? null,
};
}
/**
* Backward-compatible alias for early Mission Control integration.
*/
export function useMissionControl(): UseSessionsResult {
return useSessions();
}
/**
* Streams Mission Control session messages over SSE.
*/
export function useSessionStream(sessionId: string): UseSessionStreamResult {
const [messages, setMessages] = useState<MissionControlStreamMessage[]>([]);
const [status, setStatus] = useState<MissionControlConnectionStatus>("connecting");
const [error, setError] = useState<string | null>(null);
const eventSourceRef = useRef<EventSource | null>(null);
useEffect(() => {
if (eventSourceRef.current !== null) {
eventSourceRef.current.close();
eventSourceRef.current = null;
}
setMessages([]);
setError(null);
if (!sessionId) {
setStatus("connecting");
return;
}
if (typeof EventSource === "undefined") {
setStatus("error");
setError("Mission Control stream is not supported by this browser.");
return;
}
setStatus("connecting");
const source = new EventSource(
`/api/mission-control/sessions/${encodeURIComponent(sessionId)}/stream`
);
eventSourceRef.current = source;
source.onopen = (): void => {
setStatus("connected");
setError(null);
};
source.onmessage = (event: MessageEvent<string>): void => {
try {
const parsed = JSON.parse(event.data) as unknown;
if (!isMissionControlStreamMessage(parsed)) {
return;
}
setMessages((previousMessages) => [...previousMessages, parsed]);
} catch {
// Ignore malformed events from the stream.
}
};
source.onerror = (): void => {
if (source.readyState === EventSource.CONNECTING) {
setStatus("connecting");
setError(null);
return;
}
setStatus("error");
setError("Mission Control stream disconnected.");
};
return (): void => {
source.close();
if (eventSourceRef.current === source) {
eventSourceRef.current = null;
}
};
}, [sessionId]);
return {
messages,
status,
error,
};
}

View File

@@ -122,11 +122,11 @@ 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-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 |
| 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 | done | 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 | 2026-03-07 | 15K | — | |
| MS23-P0-005 | done | 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 | done | 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 | codex | 2026-03-07 | 2026-03-07 | 20K | — | Phase 0 gate: SSE stream verified via curl |
### Phase 1 — Provider Interface (Plugin Architecture)

View File

@@ -0,0 +1,78 @@
// Agent message roles
export type AgentMessageRole = "user" | "assistant" | "system" | "tool";
// A single message in an agent conversation
export interface AgentMessage {
id: string;
sessionId: string;
role: AgentMessageRole;
content: string;
timestamp: Date;
metadata?: Record<string, unknown>;
}
// Session lifecycle status
export type AgentSessionStatus = "active" | "paused" | "completed" | "failed" | "idle";
// An agent session (conversation thread)
export interface AgentSession {
id: string;
providerId: string; // which provider owns this session
providerType: string; // "internal" | "openclaw" | etc.
label?: string;
status: AgentSessionStatus;
parentSessionId?: string; // for subagent trees
createdAt: Date;
updatedAt: Date;
metadata?: Record<string, unknown>;
}
// Result of listing sessions
export interface AgentSessionList {
sessions: AgentSession[];
total: number;
cursor?: string;
}
// Result of injecting a message
export interface InjectResult {
accepted: boolean;
messageId?: string;
}
// The IAgentProvider interface — every provider (internal, OpenClaw, future) implements this
export interface IAgentProvider {
readonly providerId: string;
readonly providerType: string;
readonly displayName: string;
// Session management
listSessions(cursor?: string, limit?: number): Promise<AgentSessionList>;
getSession(sessionId: string): Promise<AgentSession | null>;
getMessages(sessionId: string, limit?: number, before?: string): Promise<AgentMessage[]>;
// Control operations
injectMessage(sessionId: string, content: string): Promise<InjectResult>;
pauseSession(sessionId: string): Promise<void>;
resumeSession(sessionId: string): Promise<void>;
killSession(sessionId: string, force?: boolean): Promise<void>;
// SSE streaming — returns an AsyncIterable of AgentMessage events
streamMessages(sessionId: string): AsyncIterable<AgentMessage>;
// Health
isAvailable(): Promise<boolean>;
}
// Provider configuration stored in DB (AgentProviderConfig model from P0 schema)
export interface AgentProviderConfig {
id: string;
providerId: string;
providerType: string;
displayName: string;
baseUrl?: string;
apiToken?: string;
enabled: boolean;
createdAt: Date;
updatedAt: Date;
}

View File

@@ -134,3 +134,6 @@ export * from "./widget.types";
// Export WebSocket types
export * from "./websocket.types";
// Export agent provider types
export * from "./agent-provider.types";

27
pnpm-lock.yaml generated
View File

@@ -319,6 +319,9 @@ importers:
'@mosaic/shared':
specifier: workspace:*
version: link:../../packages/shared
'@nestjs/axios':
specifier: ^4.0.1
version: 4.0.1(@nestjs/common@11.1.12(class-transformer@0.5.1)(class-validator@0.14.3)(reflect-metadata@0.2.2)(rxjs@7.8.2))(axios@1.13.5)(rxjs@7.8.2)
'@nestjs/bullmq':
specifier: ^11.0.4
version: 11.0.4(@nestjs/common@11.1.12(class-transformer@0.5.1)(class-validator@0.14.3)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/core@11.1.12)(bullmq@5.67.2)
@@ -337,6 +340,9 @@ importers:
'@nestjs/throttler':
specifier: ^6.5.0
version: 6.5.0(@nestjs/common@11.1.12(class-transformer@0.5.1)(class-validator@0.14.3)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/core@11.1.12)(reflect-metadata@0.2.2)
'@prisma/client':
specifier: ^6.19.2
version: 6.19.2(prisma@6.19.2(magicast@0.3.5)(typescript@5.9.3))(typescript@5.9.3)
bullmq:
specifier: ^5.67.2
version: 5.67.2
@@ -386,6 +392,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)
@@ -8005,7 +8014,7 @@ snapshots:
chalk: 5.6.2
commander: 12.1.0
dotenv: 17.2.4
drizzle-orm: 0.41.0(@opentelemetry/api@1.9.0)(@prisma/client@5.22.0(prisma@6.19.2(magicast@0.3.5)(typescript@5.9.3)))(@types/pg@8.16.0)(better-sqlite3@12.6.2)(kysely@0.28.10)(pg@8.17.2)(postgres@3.4.8)(prisma@6.19.2(magicast@0.3.5)(typescript@5.9.3))
drizzle-orm: 0.41.0(@opentelemetry/api@1.9.0)(@prisma/client@6.19.2(prisma@6.19.2(magicast@0.3.5)(typescript@5.9.3))(typescript@5.9.3))(@types/pg@8.16.0)(better-sqlite3@12.6.2)(kysely@0.28.10)(pg@8.17.2)(postgres@3.4.8)(prisma@6.19.2(magicast@0.3.5)(typescript@5.9.3))
open: 10.2.0
pg: 8.17.2
prettier: 3.8.1
@@ -11345,7 +11354,7 @@ snapshots:
optionalDependencies:
'@prisma/client': 5.22.0(prisma@6.19.2(magicast@0.3.5)(typescript@5.9.3))
better-sqlite3: 12.6.2
drizzle-orm: 0.41.0(@opentelemetry/api@1.9.0)(@prisma/client@5.22.0(prisma@6.19.2(magicast@0.3.5)(typescript@5.9.3)))(@types/pg@8.16.0)(better-sqlite3@12.6.2)(kysely@0.28.10)(pg@8.17.2)(postgres@3.4.8)(prisma@6.19.2(magicast@0.3.5)(typescript@5.9.3))
drizzle-orm: 0.41.0(@opentelemetry/api@1.9.0)(@prisma/client@6.19.2(prisma@6.19.2(magicast@0.3.5)(typescript@5.9.3))(typescript@5.9.3))(@types/pg@8.16.0)(better-sqlite3@12.6.2)(kysely@0.28.10)(pg@8.17.2)(postgres@3.4.8)(prisma@6.19.2(magicast@0.3.5)(typescript@5.9.3))
next: 16.1.6(@babel/core@7.28.6)(@opentelemetry/api@1.9.0)(react-dom@19.2.4(react@19.2.4))(react@19.2.4)
pg: 8.17.2
prisma: 6.19.2(magicast@0.3.5)(typescript@5.9.3)
@@ -11370,7 +11379,7 @@ snapshots:
optionalDependencies:
'@prisma/client': 6.19.2(prisma@6.19.2(magicast@0.3.5)(typescript@5.9.3))(typescript@5.9.3)
better-sqlite3: 12.6.2
drizzle-orm: 0.41.0(@opentelemetry/api@1.9.0)(@prisma/client@5.22.0(prisma@6.19.2(magicast@0.3.5)(typescript@5.9.3)))(@types/pg@8.16.0)(better-sqlite3@12.6.2)(kysely@0.28.10)(pg@8.17.2)(postgres@3.4.8)(prisma@6.19.2(magicast@0.3.5)(typescript@5.9.3))
drizzle-orm: 0.41.0(@opentelemetry/api@1.9.0)(@prisma/client@6.19.2(prisma@6.19.2(magicast@0.3.5)(typescript@5.9.3))(typescript@5.9.3))(@types/pg@8.16.0)(better-sqlite3@12.6.2)(kysely@0.28.10)(pg@8.17.2)(postgres@3.4.8)(prisma@6.19.2(magicast@0.3.5)(typescript@5.9.3))
next: 16.1.6(@babel/core@7.28.6)(@opentelemetry/api@1.9.0)(react-dom@19.2.4(react@19.2.4))(react@19.2.4)
pg: 8.17.2
prisma: 6.19.2(magicast@0.3.5)(typescript@5.9.3)
@@ -12194,17 +12203,6 @@ snapshots:
dotenv@17.2.4: {}
drizzle-orm@0.41.0(@opentelemetry/api@1.9.0)(@prisma/client@5.22.0(prisma@6.19.2(magicast@0.3.5)(typescript@5.9.3)))(@types/pg@8.16.0)(better-sqlite3@12.6.2)(kysely@0.28.10)(pg@8.17.2)(postgres@3.4.8)(prisma@6.19.2(magicast@0.3.5)(typescript@5.9.3)):
optionalDependencies:
'@opentelemetry/api': 1.9.0
'@prisma/client': 5.22.0(prisma@6.19.2(magicast@0.3.5)(typescript@5.9.3))
'@types/pg': 8.16.0
better-sqlite3: 12.6.2
kysely: 0.28.10
pg: 8.17.2
postgres: 3.4.8
prisma: 6.19.2(magicast@0.3.5)(typescript@5.9.3)
drizzle-orm@0.41.0(@opentelemetry/api@1.9.0)(@prisma/client@6.19.2(prisma@6.19.2(magicast@0.3.5)(typescript@5.9.3))(typescript@5.9.3))(@types/pg@8.16.0)(better-sqlite3@12.6.2)(kysely@0.28.10)(pg@8.17.2)(postgres@3.4.8)(prisma@6.19.2(magicast@0.3.5)(typescript@5.9.3)):
optionalDependencies:
'@opentelemetry/api': 1.9.0
@@ -12215,7 +12213,6 @@ snapshots:
pg: 8.17.2
postgres: 3.4.8
prisma: 6.19.2(magicast@0.3.5)(typescript@5.9.3)
optional: true
dunder-proto@1.0.1:
dependencies: