Compare commits

...

5 Commits

Author SHA1 Message Date
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
19 changed files with 1421 additions and 5 deletions

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

@@ -1,6 +1,7 @@
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;
@@ -16,6 +17,9 @@ describe("AgentControlService", () => {
create: ReturnType<typeof vi.fn>;
};
};
let killswitchService: {
killAgent: ReturnType<typeof vi.fn>;
};
beforeEach(() => {
prisma = {
@@ -31,7 +35,14 @@ describe("AgentControlService", () => {
},
};
service = new AgentControlService(prisma as unknown as PrismaService);
killswitchService = {
killAgent: vi.fn().mockResolvedValue(undefined),
};
service = new AgentControlService(
prisma as unknown as PrismaService,
killswitchService as unknown as KillswitchService
);
});
afterEach(() => {
@@ -137,4 +148,25 @@ describe("AgentControlService", () => {
});
});
});
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

@@ -1,10 +1,14 @@
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) {}
constructor(
private readonly prisma: PrismaService,
private readonly killswitchService: KillswitchService
) {}
private toJsonValue(value: Record<string, unknown>): Prisma.InputJsonValue {
return value as Prisma.InputJsonValue;
@@ -13,7 +17,7 @@ export class AgentControlService {
private async createOperatorAuditLog(
agentId: string,
operatorId: string,
action: "inject" | "pause" | "resume",
action: "inject" | "pause" | "resume" | "kill",
payload: Record<string, unknown>
): Promise<void> {
await this.prisma.operatorAuditLog.create({
@@ -65,4 +69,9 @@ export class AgentControlService {
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,137 @@
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>;
};
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")
);
});
});

View File

@@ -0,0 +1,57 @@
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 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

@@ -7,6 +7,8 @@ describe("AgentTreeService", () => {
let prisma: {
agentSessionTree: {
findMany: ReturnType<typeof vi.fn>;
count: ReturnType<typeof vi.fn>;
findUnique: ReturnType<typeof vi.fn>;
};
};
@@ -14,6 +16,8 @@ describe("AgentTreeService", () => {
prisma = {
agentSessionTree: {
findMany: vi.fn(),
count: vi.fn(),
findUnique: vi.fn(),
},
};
@@ -24,6 +28,141 @@ describe("AgentTreeService", () => {
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([

View File

@@ -1,11 +1,78 @@
import { Injectable } from "@nestjs/common";
import { PrismaService } from "../../prisma/prisma.service";
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" },
@@ -27,4 +94,53 @@ export class AgentTreeService {
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

@@ -10,6 +10,8 @@ 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, PrismaModule],
@@ -20,6 +22,9 @@ import { AgentTreeService } from "./agent-tree.service";
AgentMessagesService,
AgentControlService,
AgentTreeService,
InternalAgentProvider,
AgentProviderRegistry,
],
exports: [InternalAgentProvider, AgentProviderRegistry],
})
export class AgentsModule {}

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

@@ -5,6 +5,7 @@ import { ThrottlerModule } from "@nestjs/throttler";
import { HealthModule } from "./api/health/health.module";
import { AgentsModule } from "./api/agents/agents.module";
import { QueueApiModule } from "./api/queue/queue-api.module";
import { AgentProvidersModule } from "./api/agent-providers/agent-providers.module";
import { CoordinatorModule } from "./coordinator/coordinator.module";
import { BudgetModule } from "./budget/budget.module";
import { CIModule } from "./ci";
@@ -51,6 +52,7 @@ import { orchestratorConfig } from "./config/orchestrator.config";
]),
HealthModule,
AgentsModule,
AgentProvidersModule,
QueueApiModule,
CoordinatorModule,
BudgetModule,

View File

@@ -126,7 +126,7 @@ Target version: `v0.0.23`
| 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 | in-progress | 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 | | 20K | — | Phase 0 gate: SSE stream verified via curl |
| 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";