From 95ec63a868e14a93ebd1b08d4ea5074aececbdbb Mon Sep 17 00:00:00 2001 From: Jason Woltje Date: Sat, 7 Mar 2026 16:44:57 -0600 Subject: [PATCH] =?UTF-8?q?test(orchestrator):=20MS23-P3-004=20OpenClaw=20?= =?UTF-8?q?provider=20E2E=20=E2=80=94=20Phase=203=20gate?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../openclaw.provider.integration.spec.ts | 183 ++++++++++ .../tests/integration/ms23-p3-gate.spec.ts | 315 ++++++++++++++++++ .../tests/integration/vitest.config.ts | 2 +- 3 files changed, 499 insertions(+), 1 deletion(-) create mode 100644 apps/orchestrator/src/api/providers/openclaw/openclaw.provider.integration.spec.ts create mode 100644 apps/orchestrator/tests/integration/ms23-p3-gate.spec.ts diff --git a/apps/orchestrator/src/api/providers/openclaw/openclaw.provider.integration.spec.ts b/apps/orchestrator/src/api/providers/openclaw/openclaw.provider.integration.spec.ts new file mode 100644 index 0000000..ee56c89 --- /dev/null +++ b/apps/orchestrator/src/api/providers/openclaw/openclaw.provider.integration.spec.ts @@ -0,0 +1,183 @@ +import type { HttpService } from "@nestjs/axios"; +import { ServiceUnavailableException } from "@nestjs/common"; +import type { AgentMessage } from "@mosaic/shared"; +import type { AgentProviderConfig } from "@prisma/client"; +import { Readable } from "node:stream"; +import { afterEach, 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("Phase 3 gate: OpenClaw provider config registered in DB → provider loaded on boot → sessions returned from /api/mission-control/sessions → inject/pause/kill proxied to gateway", () => { + let provider: OpenClawProvider; + let httpService: { + axiosRef: { + get: ReturnType; + post: ReturnType; + }; + }; + let encryptionService: { + decryptIfNeeded: ReturnType; + }; + + const config: AgentProviderConfig = { + id: "cfg-openclaw-1", + workspaceId: "workspace-1", + name: "openclaw-home", + provider: "openclaw", + gatewayUrl: "https://gateway.example.com", + credentials: { + apiToken: "enc:token", + }, + 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"), + }; + + provider = new OpenClawProvider( + config, + encryptionService as unknown as EncryptionService, + httpService as unknown as HttpService, + new OpenClawSseBridge(httpService as unknown as HttpService) + ); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it("maps listSessions from mocked OpenClaw gateway HTTP responses", 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, + }, + }); + + await expect(provider.listSessions()).resolves.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, + }); + + expect(httpService.axiosRef.get).toHaveBeenCalledWith( + "https://gateway.example.com/api/sessions", + { + headers: { + Authorization: "Bearer plain-token", + }, + params: { + limit: 50, + }, + } + ); + }); + + it("maps streamMessages from mock SSE events into AgentMessage output", async () => { + httpService.axiosRef.get.mockResolvedValue({ + data: Readable.from([ + 'event: message\ndata: {"id":"msg-1","role":"assistant","content":"hello from stream","timestamp":"2026-03-07T15:03:00.000Z"}\n\n', + 'event: status\ndata: {"status":"paused","timestamp":"2026-03-07T15:04:00.000Z"}\n\n', + "data: [DONE]\n\n", + ]), + }); + + const messages = await collectMessages(provider.streamMessages("session-1")); + + expect(messages).toEqual([ + { + id: "msg-1", + sessionId: "session-1", + role: "assistant", + content: "hello from stream", + timestamp: new Date("2026-03-07T15:03:00.000Z"), + }, + { + id: expect.any(String), + sessionId: "session-1", + role: "system", + content: "Session status changed to paused", + timestamp: new Date("2026-03-07T15:04:00.000Z"), + metadata: { + status: "paused", + timestamp: "2026-03-07T15:04:00.000Z", + }, + }, + ]); + }); + + it("handles unavailable gateway errors", async () => { + httpService.axiosRef.get.mockRejectedValue(new Error("gateway unavailable")); + + await expect(provider.listSessions()).rejects.toBeInstanceOf(ServiceUnavailableException); + await expect(provider.listSessions()).rejects.toThrow("gateway unavailable"); + }); + + it("handles bad token decryption errors", async () => { + encryptionService.decryptIfNeeded.mockImplementation(() => { + throw new Error("bad token"); + }); + + await expect(provider.listSessions()).rejects.toBeInstanceOf(ServiceUnavailableException); + await expect(provider.listSessions()).rejects.toThrow("Failed to decrypt API token"); + }); + + it("handles malformed SSE stream responses", async () => { + vi.useFakeTimers(); + + httpService.axiosRef.get.mockResolvedValue({ + data: { + malformed: true, + }, + }); + + const streamPromise = collectMessages(provider.streamMessages("session-malformed")); + const rejection = expect(streamPromise).rejects.toThrow( + "OpenClaw provider openclaw-home failed to stream messages for session session-malformed" + ); + + 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): Promise { + const messages: AgentMessage[] = []; + + for await (const message of stream) { + messages.push(message); + } + + return messages; +} diff --git a/apps/orchestrator/tests/integration/ms23-p3-gate.spec.ts b/apps/orchestrator/tests/integration/ms23-p3-gate.spec.ts new file mode 100644 index 0000000..f823fc5 --- /dev/null +++ b/apps/orchestrator/tests/integration/ms23-p3-gate.spec.ts @@ -0,0 +1,315 @@ +import type { HttpService } from "@nestjs/axios"; +import type { + AgentMessage, + AgentSession, + AgentSessionList, + IAgentProvider, + InjectResult, +} from "@mosaic/shared"; +import type { AgentProviderConfig } from "@prisma/client"; +import { beforeEach, describe, expect, it, vi } from "vitest"; +import type { InternalAgentProvider } from "../../src/api/agents/internal-agent.provider"; +import { AgentProviderRegistry } from "../../src/api/agents/agent-provider.registry"; +import { MissionControlController } from "../../src/api/mission-control/mission-control.controller"; +import { MissionControlService } from "../../src/api/mission-control/mission-control.service"; +import { OpenClawProviderFactory } from "../../src/api/providers/openclaw/openclaw.provider-factory"; +import { OpenClawSseBridge } from "../../src/api/providers/openclaw/openclaw-sse.bridge"; +import { ProvidersModule } from "../../src/api/providers/providers.module"; +import type { PrismaService } from "../../src/prisma/prisma.service"; +import type { EncryptionService } from "../../src/security/encryption.service"; + +type MockProvider = IAgentProvider & { + listSessions: ReturnType; + getSession: ReturnType; + injectMessage: ReturnType; + pauseSession: ReturnType; + killSession: ReturnType; +}; + +type MockPrisma = { + agentProviderConfig: { + create: ReturnType; + findMany: ReturnType; + }; + operatorAuditLog: { + create: ReturnType; + }; +}; + +const emptyMessageStream = async function* (): AsyncIterable { + return; +}; + +describe("MS23-P3-004 API integration", () => { + let controller: MissionControlController; + let providersModule: ProvidersModule; + let registry: AgentProviderRegistry; + let prisma: MockPrisma; + let httpService: { + axiosRef: { + get: ReturnType; + post: ReturnType; + }; + }; + + const gatewayUrl = "https://openclaw-gateway.example.com"; + const internalSession: AgentSession = { + id: "session-internal-1", + providerId: "internal", + providerType: "internal", + status: "active", + createdAt: new Date("2026-03-07T16:00:00.000Z"), + updatedAt: new Date("2026-03-07T16:02:00.000Z"), + }; + + const openClawGatewaySession = { + id: "session-openclaw-1", + status: "running", + createdAt: "2026-03-07T16:01:00.000Z", + updatedAt: "2026-03-07T16:03:00.000Z", + }; + + const createInternalProvider = (session: AgentSession): MockProvider => ({ + providerId: "internal", + providerType: "internal", + displayName: "Internal", + listSessions: vi.fn().mockResolvedValue({ sessions: [session], total: 1 } as AgentSessionList), + getSession: vi.fn().mockImplementation(async (sessionId: string) => { + return sessionId === session.id ? session : 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), + }); + + beforeEach(() => { + const providerConfigs: AgentProviderConfig[] = []; + + prisma = { + agentProviderConfig: { + create: vi.fn().mockImplementation(async (args: { data: Record }) => { + const now = new Date("2026-03-07T15:00:00.000Z"); + const record: AgentProviderConfig = { + id: `cfg-${String(providerConfigs.length + 1)}`, + workspaceId: String(args.data.workspaceId), + name: String(args.data.name), + provider: String(args.data.provider), + gatewayUrl: String(args.data.gatewayUrl), + credentials: (args.data.credentials ?? {}) as AgentProviderConfig["credentials"], + isActive: args.data.isActive !== false, + createdAt: now, + updatedAt: now, + }; + + providerConfigs.push(record); + return record; + }), + findMany: vi + .fn() + .mockImplementation( + async (args: { where?: { provider?: string; isActive?: boolean } }) => { + const where = args.where ?? {}; + return providerConfigs.filter((config) => { + if (where.provider !== undefined && config.provider !== where.provider) { + return false; + } + + if (where.isActive !== undefined && config.isActive !== where.isActive) { + return false; + } + + return true; + }); + } + ), + }, + operatorAuditLog: { + create: vi.fn().mockResolvedValue(undefined), + }, + }; + + httpService = { + axiosRef: { + get: vi.fn().mockImplementation(async (url: string) => { + if (url === `${gatewayUrl}/api/sessions`) { + return { + data: { + sessions: [openClawGatewaySession], + total: 1, + }, + }; + } + + if (url === `${gatewayUrl}/api/sessions/${openClawGatewaySession.id}`) { + return { + data: { + session: openClawGatewaySession, + }, + }; + } + + throw new Error(`Unexpected GET ${url}`); + }), + post: vi.fn().mockImplementation(async (url: string) => { + if (url.endsWith("/inject")) { + return { data: { accepted: true, messageId: "msg-inject-1" } }; + } + + if (url.endsWith("/pause") || url.endsWith("/kill")) { + return { data: {} }; + } + + throw new Error(`Unexpected POST ${url}`); + }), + }, + }; + + const internalProvider = createInternalProvider(internalSession); + registry = new AgentProviderRegistry(internalProvider as unknown as InternalAgentProvider); + registry.onModuleInit(); + + const encryptionService = { + decryptIfNeeded: vi.fn().mockReturnValue("plain-openclaw-token"), + }; + + const sseBridge = new OpenClawSseBridge(httpService as unknown as HttpService); + const openClawProviderFactory = new OpenClawProviderFactory( + encryptionService as unknown as EncryptionService, + httpService as unknown as HttpService, + sseBridge + ); + + providersModule = new ProvidersModule( + prisma as unknown as PrismaService, + registry, + openClawProviderFactory + ); + + const missionControlService = new MissionControlService( + registry, + prisma as unknown as PrismaService + ); + + controller = new MissionControlController(missionControlService); + }); + + it("Phase 3 gate: OpenClaw provider config registered in DB → provider loaded on boot → sessions returned from /api/mission-control/sessions → inject/pause/kill proxied to gateway", async () => { + await prisma.agentProviderConfig.create({ + data: { + workspaceId: "workspace-ms23", + name: "openclaw-home", + provider: "openclaw", + gatewayUrl, + credentials: { + apiToken: "enc:test-openclaw-token", + }, + isActive: true, + }, + }); + + await providersModule.onModuleInit(); + + // Equivalent to GET /api/mission-control/sessions + const sessionsResponse = await controller.listSessions(); + + expect(sessionsResponse.sessions.map((session) => session.id)).toEqual([ + "session-openclaw-1", + "session-internal-1", + ]); + expect(sessionsResponse.sessions).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + id: "session-internal-1", + providerId: "internal", + }), + expect.objectContaining({ + id: "session-openclaw-1", + providerId: "openclaw-home", + providerType: "openclaw", + }), + ]) + ); + + const operatorRequest = { + user: { + id: "operator-ms23", + }, + }; + + await expect( + controller.injectMessage( + "session-openclaw-1", + { + message: "Ship it", + }, + operatorRequest + ) + ).resolves.toEqual({ accepted: true, messageId: "msg-inject-1" }); + + await expect(controller.pauseSession("session-openclaw-1", operatorRequest)).resolves.toEqual({ + message: "Session session-openclaw-1 paused", + }); + + await expect( + controller.killSession( + "session-openclaw-1", + { + force: false, + }, + operatorRequest + ) + ).resolves.toEqual({ message: "Session session-openclaw-1 killed" }); + + expect(httpService.axiosRef.post).toHaveBeenNthCalledWith( + 1, + `${gatewayUrl}/api/sessions/session-openclaw-1/inject`, + { content: "Ship it" }, + { + headers: { + Authorization: "Bearer plain-openclaw-token", + }, + } + ); + + expect(httpService.axiosRef.post).toHaveBeenNthCalledWith( + 2, + `${gatewayUrl}/api/sessions/session-openclaw-1/pause`, + {}, + { + headers: { + Authorization: "Bearer plain-openclaw-token", + }, + } + ); + + expect(httpService.axiosRef.post).toHaveBeenNthCalledWith( + 3, + `${gatewayUrl}/api/sessions/session-openclaw-1/kill`, + { force: false }, + { + headers: { + Authorization: "Bearer plain-openclaw-token", + }, + } + ); + + expect(prisma.operatorAuditLog.create).toHaveBeenNthCalledWith(1, { + data: { + sessionId: "session-openclaw-1", + userId: "operator-ms23", + provider: "openclaw-home", + action: "inject", + content: "Ship it", + metadata: { + payload: { + message: "Ship it", + }, + }, + }, + }); + }); +}); diff --git a/apps/orchestrator/tests/integration/vitest.config.ts b/apps/orchestrator/tests/integration/vitest.config.ts index 45a1a0b..a1ff7d1 100644 --- a/apps/orchestrator/tests/integration/vitest.config.ts +++ b/apps/orchestrator/tests/integration/vitest.config.ts @@ -4,7 +4,7 @@ export default defineConfig({ test: { globals: true, environment: "node", - include: ["**/*.e2e-spec.ts"], + include: ["tests/integration/**/*.e2e-spec.ts", "tests/integration/**/*.spec.ts"], testTimeout: 30000, }, });