/** * Approval queue integration test (WS6). * * Drives the real `PgApprovalQueue` against a PGlite-backed runtime. * Exercises: * - enqueue → approve → markExecuting → markDone (happy path) * - enqueue → reject * - enqueue (expired in past) → purgeExpired → markExpired noop rejected * - invalid transitions throw ApprovalStateTransitionError * * Run: bunx vitest run eliza/plugins/plugin-personal-assistant/test/approval-queue.integration.test.ts */ import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import type { AgentRuntime } from "@elizaos/core"; import { AgentEventService, parseInteractionBlocks } from "@elizaos/core"; import { schedulingPlugin } from "@elizaos/plugin-scheduling"; import { afterAll, beforeAll, describe, expect, it } from "vitest"; import { createRealTestRuntime } from "../../../packages/test/helpers/real-runtime.ts"; import { createApprovalQueue } from "../src/lifeops/approval-queue.js"; import { type ApprovalEnqueueInput, ApprovalNotFoundError, type ApprovalQueue, ApprovalStateTransitionError, } from "../src/lifeops/approval-queue.types.js"; import { LifeOpsRepository } from "../src/lifeops/repository.js"; import { personalAssistantPlugin } from "../src/plugin.js"; let runtime: AgentRuntime; let cleanup: () => Promise; let queue: ApprovalQueue; let isolatedStateDir: string; let isolatedConfigPath: string; const isolatedEnvKeys = [ "ELIZA_STATE_DIR", "ELIZA_CONFIG_PATH", "ELIZA_PERSIST_CONFIG_PATH", "ELIZAOS_CLOUD_API_KEY", "ELIZAOS_CLOUD_BASE_URL", ] as const; const previousEnv = new Map(); function setIsolatedEnv(): void { isolatedStateDir = mkdtempSync(join(tmpdir(), "approval-queue-state-")); isolatedConfigPath = join(isolatedStateDir, "eliza.json"); writeFileSync( isolatedConfigPath, JSON.stringify({ logging: { level: "error" } }), "utf8", ); for (const key of isolatedEnvKeys) { previousEnv.set(key, process.env[key]); } process.env.ELIZA_STATE_DIR = isolatedStateDir; process.env.ELIZA_CONFIG_PATH = isolatedConfigPath; process.env.ELIZA_PERSIST_CONFIG_PATH = isolatedConfigPath; delete process.env.ELIZA_STATE_DIR; delete process.env.ELIZA_CONFIG_PATH; delete process.env.ELIZA_PERSIST_CONFIG_PATH; delete process.env.ELIZAOS_CLOUD_API_KEY; delete process.env.ELIZAOS_CLOUD_BASE_URL; } function restoreEnv(): void { for (const key of isolatedEnvKeys) { const value = previousEnv.get(key); if (value === undefined) { delete process.env[key]; continue; } process.env[key] = value; } } function messageInput( overrides: Partial = {}, ): ApprovalEnqueueInput { return { requestedBy: "agent:lifeops", subjectUserId: "owner-123", action: "send_message", payload: { action: "send_message", recipient: "+15555551212", body: "Hello!", replyToMessageId: null, }, channel: "sms", reason: "agent wants to confirm before sending", expiresAt: new Date(Date.now() + 60 * 60 * 1000), ...overrides, }; } beforeAll(async () => { setIsolatedEnv(); const result = await createRealTestRuntime({ plugins: [schedulingPlugin, personalAssistantPlugin], }); runtime = result.runtime; cleanup = result.cleanup; // The enqueue chat-post (#14733) resolves the agent-event service off the // runtime; register the real one when the test runtime lacks it. Service // registration is lazy, so force the start (the agent server does the same // at boot when the WS bridge subscribes). if (!runtime.getService(AgentEventService.serviceType)) { await runtime.registerService(AgentEventService); await runtime.getServiceLoadPromise(AgentEventService.serviceType); } queue = createApprovalQueue(runtime, { agentId: runtime.agentId }); }, 180_000); afterAll(async () => { await cleanup(); restoreEnv(); rmSync(isolatedStateDir, { recursive: true, force: true }); }); describe("ApprovalQueue integration (real PGlite)", () => { it("enqueue → approve → markExecuting → markDone happy path", async () => { const enqueued = await queue.enqueue(messageInput()); expect(enqueued.state).toBe("pending"); expect(enqueued.resolvedAt).toBeNull(); expect(enqueued.resolvedBy).toBeNull(); const fetched = await queue.byId(enqueued.id); expect(fetched).not.toBeNull(); expect(fetched?.action).toBe("send_message"); const approved = await queue.approve(enqueued.id, { resolvedBy: "owner-123", resolutionReason: "looks good", }); expect(approved.state).toBe("approved"); expect(approved.resolvedBy).toBe("owner-123"); expect(approved.resolvedAt).toBeInstanceOf(Date); const executing = await queue.markExecuting(enqueued.id); expect(executing.state).toBe("executing"); const done = await queue.markDone(enqueued.id); expect(done.state).toBe("done"); const pendingList = await queue.list({ subjectUserId: "owner-123", state: "pending", action: null, limit: 10, }); expect(pendingList.every((r) => r.id !== enqueued.id)).toBe(true); }, 60_000); it("enqueue posts the question into chat as an assistant event with approve/reject chips (#14733)", async () => { const events: Array<{ stream: string; data: Record }> = []; const eventService = runtime.getService( AgentEventService.serviceType, ) as AgentEventService; const unsubscribe = eventService.subscribe((event) => { events.push({ stream: event.stream, data: event.data }); }); try { const enqueued = await queue.enqueue( messageInput({ subjectUserId: "owner-chips" }), ); const assistant = events.filter( (event) => event.stream === "assistant" && event.data.source === "lifeops-approval", ); expect(assistant).toHaveLength(1); const data = assistant[0]?.data ?? {}; expect(data.requestId).toBe(enqueued.id); expect(data.action).toBe("send_message"); const text = String(data.text ?? ""); expect(text).toContain("agent wants to confirm before sending"); const { blocks } = parseInteractionBlocks(text); expect(blocks).toHaveLength(1); const block = blocks[0]; if (block?.kind !== "choice") throw new Error("expected choice block"); expect(block.scope).toBe(`approval-${enqueued.id}`); expect(block.id).toBe(enqueued.id); // The tapped value is the owner's next message; it must carry the id // RESOLVE_REQUEST resolves verbatim. expect(block.options.map((o) => o.value)).toEqual([ `approve ${enqueued.id}`, `reject ${enqueued.id}`, ]); // Round-trip: drive the queue with the tapped approve value's id. const tapped = block.options[0]?.value ?? ""; const requestId = tapped.replace(/^approve /, ""); const approved = await queue.approve(requestId, { resolvedBy: "owner-chips", resolutionReason: "tapped Approve", }); expect(approved.id).toBe(enqueued.id); expect(approved.state).toBe("approved"); } finally { unsubscribe(); } }, 60_000); it("enqueue creates an owner-visible approval ScheduledTask for connector escalation (#14722)", async () => { const enqueued = await queue.enqueue( messageInput({ subjectUserId: "owner-scheduled-task" }), ); const repo = new LifeOpsRepository(runtime); const task = await repo.getScheduledTaskByIdempotencyKey( runtime.agentId, `approval:${enqueued.id}`, ); expect(task).not.toBeNull(); expect(task?.kind).toBe("approval"); expect(task?.priority).toBe("high"); expect(task?.ownerVisible).toBe(true); expect(task?.respectsGlobalPause).toBe(false); expect(task?.subject).toEqual({ kind: "self", id: "owner-scheduled-task", }); expect(task?.metadata?.approvalRequestId).toBe(enqueued.id); expect(task?.metadata?.approvalAction).toBe("send_message"); expect(task?.metadata?.pendingPromptRoomId).toBe(`approval:${enqueued.id}`); expect(task?.completionCheck?.kind).toBe("user_acknowledged"); expect(task?.completionCheck?.params).toEqual({ requestId: enqueued.id }); expect(task?.escalation?.steps?.map((step) => step.channelKey)).toEqual( expect.arrayContaining(["sms", "telegram", "discord", "imessage"]), ); expect(task?.escalation?.steps?.at(-1)?.channelKey).toBe("in_app"); expect(task?.promptInstructions).toContain(`approve ${enqueued.id}`); expect(task?.promptInstructions).toContain(`reject ${enqueued.id}`); }, 60_000); it("rolls back the approval row when the ScheduledTask cannot be created", async () => { const failedStateDir = mkdtempSync( join(tmpdir(), "approval-queue-schedule-fail-"), ); const failedConfigPath = join(failedStateDir, "eliza.json"); writeFileSync( failedConfigPath, JSON.stringify({ logging: { level: "error" } }), "utf8", ); const previousStateDir = process.env.ELIZA_STATE_DIR; const previousConfigPath = process.env.ELIZA_CONFIG_PATH; const previousPersistConfigPath = process.env.ELIZA_PERSIST_CONFIG_PATH; process.env.ELIZA_STATE_DIR = failedStateDir; process.env.ELIZA_CONFIG_PATH = failedConfigPath; process.env.ELIZA_PERSIST_CONFIG_PATH = failedConfigPath; let failedCleanup: (() => Promise) | null = null; try { const result = await createRealTestRuntime({ plugins: [personalAssistantPlugin], }); failedCleanup = result.cleanup; const failedQueue = createApprovalQueue(result.runtime, { agentId: result.runtime.agentId, }); await expect( failedQueue.enqueue( messageInput({ subjectUserId: "owner-schedule-fail" }), ), ).rejects.toThrow("failed to schedule approval task"); await expect( failedQueue.list({ subjectUserId: "owner-schedule-fail", state: null, action: null, limit: 10, }), ).resolves.toEqual([]); } finally { await failedCleanup?.(); if (previousStateDir === undefined) { delete process.env.ELIZA_STATE_DIR; } else { process.env.ELIZA_STATE_DIR = previousStateDir; } if (previousConfigPath === undefined) { delete process.env.ELIZA_CONFIG_PATH; } else { process.env.ELIZA_CONFIG_PATH = previousConfigPath; } if (previousPersistConfigPath === undefined) { delete process.env.ELIZA_PERSIST_CONFIG_PATH; } else { process.env.ELIZA_PERSIST_CONFIG_PATH = previousPersistConfigPath; } rmSync(failedStateDir, { recursive: true, force: true }); } }, 180_000); it("enqueue → reject records resolver", async () => { const enqueued = await queue.enqueue( messageInput({ subjectUserId: "owner-reject" }), ); const rejected = await queue.reject(enqueued.id, { resolvedBy: "owner-reject", resolutionReason: "not now", }); expect(rejected.state).toBe("rejected"); expect(rejected.resolutionReason).toBe("not now"); }, 60_000); it("purgeExpired moves past-due pending rows to expired", async () => { const pastExpiry = new Date(Date.now() - 5 * 60 * 1000); const enqueued = await queue.enqueue( messageInput({ subjectUserId: "owner-expire", expiresAt: pastExpiry, }), ); const purgedIds = await queue.purgeExpired(new Date()); expect(purgedIds).toContain(enqueued.id); const after = await queue.byId(enqueued.id); expect(after?.state).toBe("expired"); }, 60_000); it("rejects invalid state transitions hard", async () => { const enqueued = await queue.enqueue( messageInput({ subjectUserId: "owner-invalid" }), ); // pending -> executing is not allowed; must go through approved first await expect(queue.markExecuting(enqueued.id)).rejects.toBeInstanceOf( ApprovalStateTransitionError, ); // pending -> done is not allowed await expect(queue.markDone(enqueued.id)).rejects.toBeInstanceOf( ApprovalStateTransitionError, ); }, 60_000); it("throws ApprovalNotFoundError on unknown id", async () => { await expect( queue.approve("00000000-0000-0000-0000-000000000000", { resolvedBy: "owner-123", resolutionReason: "x", }), ).rejects.toBeInstanceOf(ApprovalNotFoundError); }, 60_000); });