import type { BackgroundJob, JobPublisher } from "@rakazo/adapter-kit"; import { calculateRoutineBackoffMs, evaluateOutputDelta, evaluateRoutineNotificationPolicy, isTransientRoutineError, } from "@rakazo/core"; import type { PrismaClient } from "@rakazo/db"; import { describe, expect, it, vi } from "vitest"; import { finalizeRoutineFailureNotification, finalizeRoutineNotification, handleRoutineSelfHealing, MAX_ROUTINE_ATTEMPTS, } from "./executor-routine.js"; import type { ExecutorDeps } from "./executor-types.js"; import { createJobReconciler } from "./job-reconciler.js"; function createMockPublisher() { const enqueue = vi.fn(async (_job: BackgroundJob) => undefined); const jobs: JobPublisher = { enqueue, cancel: async () => undefined, close: async () => undefined, }; return { jobs, enqueue }; } describe("Empirical Challenge: Routine Self-Healing, Jitter Distribution & Quiet Policy Concurrency", () => { describe("Dimension 1: Transient Error Classification Stress", () => { it("correctly identifies transient vs terminal errors across varied thrown types", () => { // 1. Standard Error instances expect(isTransientRoutineError(new Error("ECONNRESET: socket hang up"))).toBe(true); expect(isTransientRoutineError(new Error("fetch failed"))).toBe(true); expect(isTransientRoutineError(new Error("network timeout"))).toBe(true); expect(isTransientRoutineError(new Error("database deadlock detected"))).toBe(true); expect(isTransientRoutineError(new Error("container booting, please wait"))).toBe(true); expect(isTransientRoutineError(new Error("sandbox unavailable"))).toBe(true); // Terminal standard errors expect(isTransientRoutineError(new Error("Validation failed: input missing"))).toBe(false); expect(isTransientRoutineError(new Error("Unauthorized: invalid token"))).toBe(false); expect(isTransientRoutineError(new Error("Syntax error in routine script"))).toBe(false); expect(isTransientRoutineError(new Error("Loop guard: max steps reached"))).toBe(false); expect(isTransientRoutineError(new Error("budget exceeded"))).toBe(false); // 2. HTTP status codes (Axios, Fetch, REST-style error shapes) expect(isTransientRoutineError({ status: 429, message: "Too many requests" })).toBe(true); expect(isTransientRoutineError({ status: 503, message: "Service Unavailable" })).toBe(true); expect(isTransientRoutineError({ statusCode: 502, message: "Bad Gateway" })).toBe(true); expect(isTransientRoutineError({ status: 500, message: "Internal Server Error" })).toBe(true); expect(isTransientRoutineError({ status: 504, message: "Gateway Timeout" })).toBe(true); expect(isTransientRoutineError({ status: 408, message: "Request Timeout" })).toBe(true); // Terminal HTTP codes expect(isTransientRoutineError({ status: 401, message: "Unauthorized" })).toBe(false); expect(isTransientRoutineError({ status: 403, message: "Forbidden" })).toBe(false); expect(isTransientRoutineError({ status: 404, message: "Not Found" })).toBe(false); expect(isTransientRoutineError({ status: 400, message: "Bad Request" })).toBe(false); expect(isTransientRoutineError({ status: 422, message: "Unprocessable Entity" })).toBe(false); // 3. Node/Undici/Socket codes in object properties expect(isTransientRoutineError({ code: "ECONNRESET" })).toBe(true); expect(isTransientRoutineError({ code: "ETIMEDOUT" })).toBe(true); expect(isTransientRoutineError({ code: "ECONNREFUSED" })).toBe(true); expect(isTransientRoutineError({ code: "EAI_AGAIN" })).toBe(true); expect(isTransientRoutineError({ code: "ENOTFOUND" })).toBe(true); expect(isTransientRoutineError({ code: "UND_ERR_CONNECT_TIMEOUT" })).toBe(true); expect(isTransientRoutineError({ code: "UND_ERR_SOCKET" })).toBe(true); expect(isTransientRoutineError({ code: "UND_ERR_HEADERS_TIMEOUT" })).toBe(true); expect(isTransientRoutineError({ code: "53300" })).toBe(true); // Postgres too many connections expect(isTransientRoutineError({ code: "P2034" })).toBe(true); // Prisma deadlock // Terminal error codes expect(isTransientRoutineError({ code: "ENOENT" })).toBe(false); expect(isTransientRoutineError({ code: "EACCES" })).toBe(false); expect(isTransientRoutineError({ code: "ERR_INVALID_ARG_TYPE" })).toBe(false); // 4. Raw string throws expect(isTransientRoutineError("fetch failed due to network disruption")).toBe(true); expect(isTransientRoutineError("socket hang up")).toBe(true); expect(isTransientRoutineError("connection reset by peer")).toBe(true); expect(isTransientRoutineError("network timeout on upstream API")).toBe(true); expect(isTransientRoutineError("computer busy")).toBe(true); expect(isTransientRoutineError("unauthorized: missing bearer token")).toBe(false); expect(isTransientRoutineError("invalid arguments provided")).toBe(false); // 5. Degenerate and falsy edge cases expect(isTransientRoutineError(null)).toBe(false); expect(isTransientRoutineError(undefined)).toBe(false); expect(isTransientRoutineError("")).toBe(false); expect(isTransientRoutineError(0)).toBe(false); expect(isTransientRoutineError(false)).toBe(false); expect(isTransientRoutineError({})).toBe(false); expect(isTransientRoutineError([])).toBe(false); expect(isTransientRoutineError(12345)).toBe(false); }); }); describe("Dimension 1: Exponential Backoff & Full Jitter Distribution Stress", () => { it("bounds 1,000 randomized backoff samples strictly within min and maxDelay", () => { const trials = 1000; const baseDelay = 1000; const maxDelay = 60000; // Attempt 0 should return 0 expect(calculateRoutineBackoffMs(0)).toBe(0); expect(calculateRoutineBackoffMs(-1)).toBe(0); // Test across attempts 1 through 10 for (let attempt = 1; attempt <= 10; attempt += 1) { const factor = 2 ** Math.min(attempt - 1, 6); const expectedBoundedMax = Math.min(maxDelay, baseDelay * factor); const samples: number[] = []; for (let i = 0; i < trials; i += 1) { const delay = calculateRoutineBackoffMs(attempt, { baseBackoffMs: baseDelay, maxBackoffMs: maxDelay, }); expect(delay).toBeGreaterThanOrEqual(0); expect(delay).toBeLessThanOrEqual(expectedBoundedMax); expect(delay).toBeLessThanOrEqual(maxDelay); expect(Number.isInteger(delay)).toBe(true); samples.push(delay); } // Verify decorrelated distribution (non-constant samples and non-zero variance) const uniqueValues = new Set(samples); expect(uniqueValues.size).toBeGreaterThan(50); // High entropy const mean = samples.reduce((acc, v) => acc + v, 0) / trials; // Mean of uniform distribution [0, boundedMax] is boundedMax / 2 const expectedMean = expectedBoundedMax / 2; // Allow statistical tolerance within 15% of expected mean expect(mean).toBeGreaterThan(expectedMean * 0.85); expect(mean).toBeLessThan(expectedMean * 1.15); // Verify spread across lower and upper quartiles const lowQuartile = samples.filter((v) => v < expectedBoundedMax * 0.25).length; const highQuartile = samples.filter((v) => v > expectedBoundedMax * 0.75).length; expect(lowQuartile).toBeGreaterThan(trials * 0.15); expect(highQuartile).toBeGreaterThan(trials * 0.15); } }); it("respects custom retry configs and enforces attempt ceiling", async () => { const customConfig = { baseBackoffMs: 250, maxBackoffMs: 5000, }; for (let i = 0; i < 500; i += 1) { const delay = calculateRoutineBackoffMs(10, customConfig); expect(delay).toBeGreaterThanOrEqual(0); expect(delay).toBeLessThanOrEqual(5000); } }); it("strictly halts self-healing when attempt count reaches MAX_ROUTINE_ATTEMPTS (ceiling = 3)", async () => { const updateRun = vi.fn().mockResolvedValue({}); const enqueueJob = vi.fn().mockResolvedValue({}); const mockDeps = { prisma: { run: { update: updateRun } }, jobs: { enqueue: enqueueJob }, } as unknown as ExecutorDeps; const transientError = Object.assign(new Error("ETIMEDOUT: connect timeout"), { code: "ETIMEDOUT", }); // Attempt 1: should self-heal and requeue const r1 = await handleRoutineSelfHealing(mockDeps, "run-x", 1, transientError); expect(r1).toBe(true); expect(updateRun).toHaveBeenCalledTimes(1); expect(enqueueJob).toHaveBeenCalledTimes(1); updateRun.mockClear(); enqueueJob.mockClear(); // Attempt 2: should self-heal and requeue const r2 = await handleRoutineSelfHealing(mockDeps, "run-x", 2, transientError); expect(r2).toBe(true); expect(updateRun).toHaveBeenCalledTimes(1); expect(enqueueJob).toHaveBeenCalledTimes(1); updateRun.mockClear(); enqueueJob.mockClear(); // Attempt 3 (MAX_ROUTINE_ATTEMPTS): must refuse to requeue and allow permanent failure const r3 = await handleRoutineSelfHealing( mockDeps, "run-x", MAX_ROUTINE_ATTEMPTS, transientError, ); expect(r3).toBe(false); expect(updateRun).not.toHaveBeenCalled(); expect(enqueueJob).not.toHaveBeenCalled(); // Attempt 4+: must also refuse const r4 = await handleRoutineSelfHealing( mockDeps, "run-x", MAX_ROUTINE_ATTEMPTS + 1, transientError, ); expect(r4).toBe(false); expect(updateRun).not.toHaveBeenCalled(); expect(enqueueJob).not.toHaveBeenCalled(); }); }); describe("Dimension 2: Reconciler Zombie Run Watchdog & Cron Stampede Guard", () => { it("evicts zombie routine runs exceeding MAX_ROUTINE_ATTEMPTS and prevents requeue", async () => { const now = new Date(); // Runs in database: // - run-zombie-1: routine run, expired lease, 3 attempts (zombie ceiling reached) // - run-zombie-2: routine run, expired lease, 4 attempts (zombie ceiling exceeded) // - run-healthy-1: routine run, expired lease, 1 attempt (can continue) // - run-manual-1: manual run, expired lease, 5 attempts (non-routine, should continue) const mockRuns = [ { id: "run-zombie-1", updatedAt: new Date(now.getTime() - 120_000), trigger: "routine", attempts: [{ id: "a1" }, { id: "a2" }, { id: "a3" }], }, { id: "run-zombie-2", updatedAt: new Date(now.getTime() - 110_000), trigger: "routine", attempts: [{ id: "a1" }, { id: "a2" }, { id: "a3" }, { id: "a4" }], }, { id: "run-healthy-1", updatedAt: new Date(now.getTime() - 100_000), trigger: "routine", attempts: [{ id: "a1" }], }, { id: "run-manual-1", updatedAt: new Date(now.getTime() - 90_000), trigger: "manual", attempts: [{ id: "a1" }, { id: "a2" }, { id: "a3" }, { id: "a4" }, { id: "a5" }], }, ]; const updateMany = vi.fn().mockResolvedValue({ count: 1 }); const mockPrisma = { run: { findMany: vi.fn(async (args?: { where?: { trigger?: string } }) => { if (args?.where?.trigger === "messaging") { return []; } return mockRuns; }), updateMany, }, routine: { findMany: vi.fn(async () => []) }, computer: { findMany: vi.fn(async () => []) }, messagingOutbound: { findFirst: vi.fn(async () => null) }, } as unknown as PrismaClient; const { jobs, enqueue } = createMockPublisher(); const reconciler = createJobReconciler({ prisma: mockPrisma, jobs }); await reconciler.reconcileOnce(); // Exactly the 2 zombie runs should be updated to failed in database expect(updateMany).toHaveBeenCalledTimes(2); expect(updateMany).toHaveBeenCalledWith({ where: { id: "run-zombie-1", status: { in: ["queued", "leased", "running"] } }, data: expect.objectContaining({ status: "failed", error: "Routine exceeded maximum retry attempts during execution", leaseOwner: null, leaseExpiresAt: null, }), }); expect(updateMany).toHaveBeenCalledWith({ where: { id: "run-zombie-2", status: { in: ["queued", "leased", "running"] } }, data: expect.objectContaining({ status: "failed", error: "Routine exceeded maximum retry attempts during execution", }), }); // Exactly the 2 eligible runs (run-healthy-1 and run-manual-1) should be requeued expect(enqueue).toHaveBeenCalledTimes(2); expect(enqueue).toHaveBeenCalledWith( expect.objectContaining({ name: "run.continue", payload: { runId: "run-healthy-1" }, }), ); expect(enqueue).toHaveBeenCalledWith( expect.objectContaining({ name: "run.continue", payload: { runId: "run-manual-1" }, }), ); // Verify zombie runs were NOT enqueued const enqueuedIds = enqueue.mock.calls.map( (call) => (call[0]?.payload as { runId?: string })?.runId, ); expect(enqueuedIds).not.toContain("run-zombie-1"); expect(enqueuedIds).not.toContain("run-zombie-2"); }); it("clamps 14-day overdue nextRunAt to next future slot preventing thundering herd cron stampedes", async () => { const now = new Date("2026-10-12T12:00:00.000Z"); // System was asleep / down for 14 days const fourteenDaysAgo = new Date(now.getTime() - 14 * 24 * 60 * 60 * 1000); // Daily routine at 09:00 UTC const staleRoutine = { id: "routine-daily-9am", nextRunAt: fourteenDaysAgo, crons: ["0 9 * * *"], timezone: "UTC", }; const routineUpdate = vi.fn().mockResolvedValue({}); const mockPrisma = { run: { findMany: vi.fn(async () => []), updateMany: vi.fn().mockResolvedValue({ count: 0 }), }, routine: { findMany: vi.fn(async () => [staleRoutine]), update: routineUpdate, }, computer: { findMany: vi.fn(async () => []) }, messagingOutbound: { findFirst: vi.fn(async () => null) }, } as unknown as PrismaClient; const { jobs, enqueue } = createMockPublisher(); const reconciler = createJobReconciler({ prisma: mockPrisma, jobs }); await reconciler.reconcileOnce(); // Check that routine.update was called with nextRunAt in the future (> now) expect(routineUpdate).toHaveBeenCalledTimes(1); const updatedArgs = routineUpdate.mock.calls[0]![0] as { where: { id: string }; data: { nextRunAt: Date }; }; expect(updatedArgs.where.id).toBe("routine-daily-9am"); const nextRunDate = updatedArgs.data.nextRunAt; expect(nextRunDate.getTime()).toBeGreaterThan(Date.now() - 60_000); // Check that exactly ONE wakeup job was enqueued with future date and stable replaceKey expect(enqueue).toHaveBeenCalledTimes(1); const enqueuedJob = enqueue.mock.calls[0]![0] as BackgroundJob; expect(enqueuedJob.name).toBe("routine.wakeup"); expect(enqueuedJob.payload).toMatchObject({ routineId: "routine-daily-9am", }); expect(enqueuedJob.replaceKey).toBe("routine:routine-daily-9am"); expect(enqueuedJob.availableAt?.getTime()).toBeGreaterThan(Date.now() - 60_000); }); it("preserves immediate scheduling for mildly late routines (<15 minutes overdue)", async () => { const now = new Date(); // Only 5 minutes late (transient worker delay, not outage) const fiveMinutesAgo = new Date(now.getTime() - 5 * 60 * 1000); const routine = { id: "routine-5m-late", nextRunAt: fiveMinutesAgo, crons: ["0 9 * * *"], timezone: "UTC", }; const routineUpdate = vi.fn().mockResolvedValue({}); const mockPrisma = { run: { findMany: vi.fn(async () => []), updateMany: vi.fn().mockResolvedValue({ count: 0 }), }, routine: { findMany: vi.fn(async () => [routine]), update: routineUpdate, }, computer: { findMany: vi.fn(async () => []) }, messagingOutbound: { findFirst: vi.fn(async () => null) }, } as unknown as PrismaClient; const { jobs, enqueue } = createMockPublisher(); const reconciler = createJobReconciler({ prisma: mockPrisma, jobs }); await reconciler.reconcileOnce(); // Mildly late routines should NOT be jumped into the future; they should execute now expect(routineUpdate).not.toHaveBeenCalled(); expect(enqueue).toHaveBeenCalledTimes(1); const enqueuedJob = enqueue.mock.calls[0]![0] as BackgroundJob; expect(enqueuedJob.availableAt).toEqual(fiveMinutesAgo); }); }); describe("Dimension 3: Quiet Notification Delivery & Concurrency Stress", () => { it("concurrently suppresses 100% of notifications and unread marks when outputs are unchanged under on_change", async () => { const notifySend = vi.fn().mockResolvedValue(true); const mockDeps = { notifications: { send: notifySend }, prisma: { run: { findFirst: vi.fn().mockResolvedValue({ bot: { notifyOnFinish: true }, thread: { groupId: null }, }), }, }, } as unknown as ExecutorDeps; const concurrentCalls = 50; const tasks = Array.from({ length: concurrentCalls }, (_, index) => finalizeRoutineNotification({ deps: mockDeps, run: { id: `run-${index}`, spaceId: "space-1", botId: "bot-1", threadId: "thread-1", userId: "user-1", routineId: "routine-1", }, bot: { id: "bot-1", name: "Cluster Monitor" }, routine: { id: "routine-1", name: "Cluster Health", notify: true }, notifyMode: "on_change", outputText: "All 12 cluster nodes operating at 24% load", hasChanged: false, }), ); const results = await Promise.all(tasks); // Verify every single call was quiet and did not mark thread unread for (const res of results) { expect(res.markUnread).toBe(false); expect(res.isQuiet).toBe(true); } // 100% of notifications suppressed expect(notifySend).toHaveBeenCalledTimes(0); }); it("dispatches exactly 1 notification with concise delta summary (<120 chars) when output changes", async () => { const notifySend = vi.fn().mockResolvedValue(true); const mockDeps = { notifications: { send: notifySend }, prisma: { run: { findFirst: vi.fn().mockResolvedValue({ bot: { notifyOnFinish: true }, thread: { groupId: null }, }), }, }, } as unknown as ExecutorDeps; const deltaSummary = "Node 4 memory threshold exceeded (94% utilized)"; expect(deltaSummary.length).toBeLessThan(120); const result = await finalizeRoutineNotification({ deps: mockDeps, run: { id: "run-delta-1", spaceId: "space-1", botId: "bot-1", threadId: "thread-1", userId: "user-1", routineId: "routine-1", }, bot: { id: "bot-1", name: "Cluster Monitor" }, routine: { id: "routine-1", name: "Cluster Health", notify: true }, notifyMode: "on_change", outputText: "Cluster status alert: Node 4 memory threshold exceeded", hasChanged: true, deltaSummary, }); expect(result.markUnread).toBe(true); expect(result.isQuiet).toBe(false); expect(notifySend).toHaveBeenCalledTimes(1); const payload = notifySend.mock.calls[0]![0]; expect(payload).toMatchObject({ kind: "completion", title: "Cluster Monitor", body: deltaSummary, }); // Verifies AGENTS.md minimal copy: title must not be bloated with "Routine executed successfully: " expect(payload.title).not.toMatch(/Routine executed successfully/i); }); it("evaluates notification policies comprehensively across failure and modes", async () => { const notifySend = vi.fn().mockResolvedValue(true); const mockDeps = { notifications: { send: notifySend }, prisma: { run: { findFirst: vi.fn().mockResolvedValue({ bot: { notifyOnFinish: true }, thread: { groupId: null }, }), }, }, } as unknown as ExecutorDeps; const baseRun = { spaceId: "space-1", userId: "user-1", botId: "bot-1", threadId: "thread-1", routineId: "routine-1", }; const bot = { id: "bot-1", name: "Monitor" }; const routine = { id: "routine-1", name: "Watcher", notify: true }; // 1. Failure under on_change: MUST notify notifySend.mockClear(); const nOnChange = await finalizeRoutineFailureNotification( mockDeps, baseRun, bot, routine, "on_change", "Gateway timeout 504", ); expect(nOnChange).toBe(true); expect(notifySend).toHaveBeenCalledTimes(1); expect(notifySend.mock.calls[0]![0]).toMatchObject({ kind: "failure", title: "Monitor failed", body: "Gateway timeout 504", }); // 2. Failure under on_failure: MUST notify notifySend.mockClear(); const nOnFailure = await finalizeRoutineFailureNotification( mockDeps, baseRun, bot, routine, "on_failure", "Connection refused", ); expect(nOnFailure).toBe(true); expect(notifySend).toHaveBeenCalledTimes(1); // 3. Failure under always: MUST notify notifySend.mockClear(); const nAlways = await finalizeRoutineFailureNotification( mockDeps, baseRun, bot, routine, "always", "Downstream error", ); expect(nAlways).toBe(true); expect(notifySend).toHaveBeenCalledTimes(1); // 4. Failure under never: MUST BE SUPPRESSED notifySend.mockClear(); const nNever = await finalizeRoutineFailureNotification( mockDeps, baseRun, bot, routine, "never", "Fatal error", ); expect(nNever).toBe(false); expect(notifySend).not.toHaveBeenCalled(); // 5. Failure when routine.notify is false: MUST BE SUPPRESSED regardless of mode notifySend.mockClear(); const nDisabled = await finalizeRoutineFailureNotification( mockDeps, baseRun, bot, { ...routine, notify: false }, "always", "Fatal error", ); expect(nDisabled).toBe(false); expect(notifySend).not.toHaveBeenCalled(); }); it("verifies volatile field masking ignores timestamps, durationMs, and ephemeral IDs in delta evaluation", () => { const outputRun1 = { status: "healthy", uptimeSeconds: 86400, timestamp: 1728750000000, durationMs: 145, runId: "run-001", attemptId: "att-001", taskId: "tsk-001", leaseOwner: "worker-prod-1", }; const outputRun2 = { status: "healthy", uptimeSeconds: 86400, timestamp: 1728836400000, // 24 hours later durationMs: 298, // different execution time runId: "run-002", attemptId: "att-002", taskId: "tsk-002", leaseOwner: "worker-prod-2", }; // Both runs carry identical business state, differing only in volatile telemetry keys const delta = evaluateOutputDelta(outputRun1, outputRun2); expect(delta.hasChanged).toBe(false); const decision = evaluateRoutineNotificationPolicy({ routineNotifyEnabled: true, notifyMode: "on_change", hasChanged: delta.hasChanged, outcome: "completed", }); expect(decision.shouldNotify).toBe(false); expect(decision.isQuiet).toBe(true); expect(decision.markUnread).toBe(false); // Mutate a substantive business property const outputRun3 = { ...outputRun2, status: "degraded", uptimeSeconds: 86450, }; const substantiveDelta = evaluateOutputDelta(outputRun2, outputRun3); expect(substantiveDelta.hasChanged).toBe(true); expect(substantiveDelta.deltaSummary).toBeDefined(); const changedDecision = evaluateRoutineNotificationPolicy({ routineNotifyEnabled: true, notifyMode: "on_change", hasChanged: substantiveDelta.hasChanged, outcome: "completed", }); expect(changedDecision.shouldNotify).toBe(true); expect(changedDecision.isQuiet).toBe(false); expect(changedDecision.markUnread).toBe(true); }); }); describe("Dimension 4: Advanced Adversarial Edge Cases & Stress Invariants", () => { it("gracefully survives circular references without stack overflow", () => { interface CircularObj { name: string; child?: CircularObj; self?: CircularObj; } const circular: CircularObj = { name: "parent" }; circular.self = circular; circular.child = { name: "child", self: circular }; const delta = evaluateOutputDelta(circular, circular); expect(delta.hasChanged).toBe(false); expect(delta.changeHash).toBeDefined(); expect(typeof delta.changeHash).toBe("string"); }); it("handles extreme attempt numbers without numeric overflow or NaN in backoff calculation", () => { const extremeAttempts = [-1000, -1, 0, 100, 10_000, 1_000_000, Number.MAX_SAFE_INTEGER]; for (const att of extremeAttempts) { const delay = calculateRoutineBackoffMs(att); expect(delay).toBeGreaterThanOrEqual(0); expect(delay).toBeLessThanOrEqual(60_000); expect(Number.isFinite(delay)).toBe(true); expect(Number.isNaN(delay)).toBe(false); } }); it("exhibits zero hash collisions across 5,000 distinct structured objects", () => { const hashes = new Set(); const count = 5000; for (let i = 0; i < count; i += 1) { const obj = { sensor: `sensor-${i % 25}`, reading: i * 1.0001, status: i % 2 === 0 ? "active" : "standby", batch: i, }; const hash = evaluateOutputDelta(null, obj).changeHash; hashes.add(hash); } expect(hashes.size).toBe(count); }); it("follower reconciler with failed leadership acquisition executes zero queries and enqueues zero jobs", async () => { const mockPrisma = { run: { findMany: vi.fn() }, routine: { findMany: vi.fn() }, computer: { findMany: vi.fn() }, messagingOutbound: { findFirst: vi.fn() }, } as unknown as PrismaClient; const { jobs, enqueue } = createMockPublisher(); const leadership = { tryAcquire: vi.fn().mockResolvedValue(false), // Follower node! release: vi.fn().mockResolvedValue(undefined), }; const reconciler = createJobReconciler({ prisma: mockPrisma, jobs, leadership, }); await reconciler.reconcileOnce(); expect(leadership.tryAcquire).toHaveBeenCalledTimes(1); expect(mockPrisma.run.findMany).not.toHaveBeenCalled(); expect(mockPrisma.routine.findMany).not.toHaveBeenCalled(); expect(enqueue).not.toHaveBeenCalled(); }); it("verifies sequential routine lifecycle: initial notifies, unchanged suppresses, update notifies", async () => { const notifySend = vi.fn().mockResolvedValue(true); const mockDeps = { notifications: { send: notifySend }, prisma: { run: { findFirst: vi.fn().mockResolvedValue({ bot: { notifyOnFinish: true }, thread: { groupId: null }, }), }, }, } as unknown as ExecutorDeps; const runTemplate = { spaceId: "space-1", botId: "bot-1", threadId: "thread-1", userId: "user-1", routineId: "routine-seq", }; const bot = { id: "bot-1", name: "DB Pulse" }; const routine = { id: "routine-seq", name: "Pulse", notify: true }; // Step 1: Initial run output const run1 = await finalizeRoutineNotification({ deps: mockDeps, run: { ...runTemplate, id: "seq-1" }, bot, routine, notifyMode: "on_change", outputText: "Active connections: 42", hasChanged: true, // first run always changed deltaSummary: "Active connections: 42", }); expect(run1.markUnread).toBe(true); expect(notifySend).toHaveBeenCalledTimes(1); notifySend.mockClear(); // Step 2: Second run (identical output) -> must suppress const run2 = await finalizeRoutineNotification({ deps: mockDeps, run: { ...runTemplate, id: "seq-2" }, bot, routine, notifyMode: "on_change", outputText: "Active connections: 42", hasChanged: false, }); expect(run2.markUnread).toBe(false); expect(run2.isQuiet).toBe(true); expect(notifySend).not.toHaveBeenCalled(); // Step 3: Third run (still identical) -> must suppress const run3 = await finalizeRoutineNotification({ deps: mockDeps, run: { ...runTemplate, id: "seq-3" }, bot, routine, notifyMode: "on_change", outputText: "Active connections: 42", hasChanged: false, }); expect(run3.markUnread).toBe(false); expect(run3.isQuiet).toBe(true); expect(notifySend).not.toHaveBeenCalled(); // Step 4: Fourth run (state changes: spike in connections) -> must notify const run4 = await finalizeRoutineNotification({ deps: mockDeps, run: { ...runTemplate, id: "seq-4" }, bot, routine, notifyMode: "on_change", outputText: "Active connections: 98", hasChanged: true, deltaSummary: "Active connections: 98", }); expect(run4.markUnread).toBe(true); expect(run4.isQuiet).toBe(false); expect(notifySend).toHaveBeenCalledTimes(1); notifySend.mockClear(); // Step 5: Fifth run (remains at 98) -> must suppress again const run5 = await finalizeRoutineNotification({ deps: mockDeps, run: { ...runTemplate, id: "seq-5" }, bot, routine, notifyMode: "on_change", outputText: "Active connections: 98", hasChanged: false, }); expect(run5.markUnread).toBe(false); expect(run5.isQuiet).toBe(true); expect(notifySend).not.toHaveBeenCalled(); }); }); });