feat(backups): execute backups through the agent

app/server/modules/agents/agents-manager.ts

@@ -1,5 +1,6 @@
 import type { ChildProcess } from "node:child_process";
-import type { AgentManagerRuntime } from "./controller/server";
+import type { BackupCancelPayload, BackupRunPayload } from "@zerobyte/contracts/agent-protocol";
+import type { AgentBackupEventHandlers, AgentManagerRuntime } from "./controller/server";
 import { spawnLocalAgentProcess, stopLocalAgentProcess } from "./local/process";
 
 export type { AgentBackupEventHandlers } from "./controller/server";
@@ -32,6 +33,8 @@ const getAgentRuntimeState = () => {
 
 const getAgentManagerRuntime = () => getAgentRuntimeState().agentManager;
 
+let backupEventHandlers: AgentBackupEventHandlers = {};
+
 export const startAgentRuntime = async () => {
 	const runtime = getAgentRuntimeState();
 
@@ -41,11 +44,38 @@ export const startAgentRuntime = async () => {
 
 	const { createAgentManagerRuntime } = await import("./controller/server");
 	const agentManager = createAgentManagerRuntime();
+	agentManager.setBackupEventHandlers(backupEventHandlers);
 
 	await agentManager.start();
 	runtime.agentManager = agentManager;
 };
 
+export const agentManager = {
+	sendBackup: (agentId: string, payload: BackupRunPayload) => {
+		const runtime = getAgentManagerRuntime();
+		if (!runtime) {
+			return false;
+		}
+
+		return runtime.sendBackup(agentId, payload);
+	},
+	cancelBackup: (agentId: string, payload: BackupCancelPayload) => {
+		const runtime = getAgentManagerRuntime();
+		if (!runtime) {
+			return false;
+		}
+
+		return runtime.cancelBackup(agentId, payload);
+	},
+	setBackupEventHandlers: (handlers: AgentBackupEventHandlers) => {
+		backupEventHandlers = handlers;
+		getAgentManagerRuntime()?.setBackupEventHandlers(handlers);
+	},
+	getBackupEventHandlers: () => {
+		return getAgentManagerRuntime()?.getBackupEventHandlers() ?? backupEventHandlers;
+	},
+};
+
 export const spawnLocalAgent = async () => {
 	await spawnLocalAgentProcess(getAgentRuntimeState());
 };

app/server/modules/agents/controller/server.ts

@@ -231,7 +231,7 @@ export function createAgentManagerRuntime() {
 
 	return {
 		start,
-		sendBackup: async (agentId: string, payload: BackupRunPayload) => {
+		sendBackup: (agentId: string, payload: BackupRunPayload) => {
 			const session = getSession(agentId);
 
 			if (!session) {
@@ -244,7 +244,7 @@ export function createAgentManagerRuntime() {
 				return false;
 			}
 
-			if (!(await Effect.runPromise(session.sendBackup(payload)))) {
+			if (!Effect.runSync(session.sendBackup(payload))) {
 				logger.warn(`Cannot send backup command. Agent ${agentId} is no longer accepting commands.`);
 				return false;
 			}
@@ -252,7 +252,7 @@ export function createAgentManagerRuntime() {
 			logger.info(`Sent backup command ${payload.jobId} to agent ${agentId} for schedule ${payload.scheduleId}`);
 			return true;
 		},
-		cancelBackup: async (agentId: string, payload: BackupCancelPayload) => {
+		cancelBackup: (agentId: string, payload: BackupCancelPayload) => {
 			const session = getSession(agentId);
 
 			if (!session) {
@@ -260,7 +260,7 @@ export function createAgentManagerRuntime() {
 				return false;
 			}
 
-			if (!(await Effect.runPromise(session.sendBackupCancel(payload)))) {
+			if (!Effect.runSync(session.sendBackupCancel(payload))) {
 				logger.warn(`Cannot cancel backup command. Agent ${agentId} is no longer accepting commands.`);
 				return false;
 			}

app/server/modules/backups/__tests__/backups.service.execution.test.ts

@@ -1,7 +1,6 @@
 import waitForExpect from "wait-for-expect";
 import { afterEach, describe, expect, test, vi } from "vitest";
 import { backupsService } from "../backups.service";
-import { backupsExecutionService } from "../backups.execution";
 import { createTestVolume } from "~/test/helpers/volume";
 import { createTestBackupSchedule } from "~/test/helpers/backup";
 import { createTestRepository } from "~/test/helpers/repository";
@@ -14,9 +13,14 @@ import type { SafeSpawnParams } from "@zerobyte/core/node";
 import { logger } from "@zerobyte/core/node";
 import { restic } from "~/server/core/restic";
 import { NotFoundError, BadRequestError } from "http-errors-enhanced";
+import { fromAny } from "@total-typescript/shoehorn";
+import { scheduleQueries } from "../backups.queries";
 import { repositoriesService } from "~/server/modules/repositories/repositories.service";
 import { repoMutex } from "~/server/core/repository-mutex";
 import { notificationsService } from "~/server/modules/notifications/notifications.service";
+import { agentManager } from "~/server/modules/agents/agents-manager";
+import { createAgentBackupMocks } from "~/test/helpers/agent-mock";
+import { getScheduleByIdOrShortId } from "../helpers/backup-schedule-lookups";
 
 const setup = () => {
 	const resticBackupMock = vi.fn((_: SafeSpawnParams) =>
@@ -24,6 +28,7 @@ const setup = () => {
 	);
 	const resticForgetMock = vi.fn(() => Promise.resolve({ success: true, data: null }));
 	const resticCopyMock = vi.fn(() => Promise.resolve({ success: true, output: "" }));
+	const { sendBackupMock, cancelBackupMock } = createAgentBackupMocks(resticBackupMock);
 	const refreshStatsMock = vi.fn(() =>
 		Promise.resolve({
 			total_size: 0,
@@ -39,12 +44,16 @@ const setup = () => {
 	vi.spyOn(restic, "forget").mockImplementation(resticForgetMock);
 	vi.spyOn(restic, "copy").mockImplementation(resticCopyMock);
 	vi.spyOn(repositoriesService, "refreshRepositoryStats").mockImplementation(refreshStatsMock);
+	vi.spyOn(agentManager, "sendBackup").mockImplementation(sendBackupMock);
+	vi.spyOn(agentManager, "cancelBackup").mockImplementation(cancelBackupMock);
 	vi.spyOn(context, "getOrganizationId").mockReturnValue(TEST_ORG_ID);
 
 	return {
 		resticBackupMock,
 		resticForgetMock,
 		resticCopyMock,
+		sendBackupMock,
+		cancelBackupMock,
 		refreshStatsMock,
 	};
 };
@@ -65,7 +74,7 @@ describe("backup execution - validation failures", () => {
 		});
 
 		// act
-		const result = await backupsExecutionService.validateBackupExecution(schedule.id);
+		const result = await backupsService.validateBackupExecution(schedule.id);
 
 		// assert
 		expect(result.type).toBe("failure");
@@ -76,10 +85,70 @@ describe("backup execution - validation failures", () => {
 		expect(resticBackupMock).not.toHaveBeenCalled();
 	});
 
+	test("should fail backup when volume does not exist", async () => {
+		// arrange
+		setup();
+		const volume = await createTestVolume();
+		const repository = await createTestRepository();
+		const schedule = await createTestBackupSchedule({
+			volumeId: volume.id,
+			repositoryId: repository.id,
+		});
+
+		const hydratedSchedule = await scheduleQueries.findById(schedule.id, TEST_ORG_ID);
+		expect(hydratedSchedule).toBeDefined();
+		const scheduleWithoutVolume = {
+			...hydratedSchedule,
+			volume: null,
+		};
+		vi.spyOn(scheduleQueries, "findById").mockResolvedValueOnce(fromAny(scheduleWithoutVolume));
+
+		// act
+		const result = await backupsService.validateBackupExecution(schedule.id);
+
+		// assert
+		expect(result.type).toBe("failure");
+		if (result.type === "failure") {
+			expect(result.error).toBeInstanceOf(NotFoundError);
+			expect(result.error.message).toBe("Volume not found");
+			expect(result.partialContext?.schedule).toBeDefined();
+		}
+	});
+
+	test("should fail backup when repository does not exist", async () => {
+		// arrange
+		setup();
+		const volume = await createTestVolume();
+		const repository = await createTestRepository();
+		const schedule = await createTestBackupSchedule({
+			volumeId: volume.id,
+			repositoryId: repository.id,
+		});
+
+		const hydratedSchedule = await scheduleQueries.findById(schedule.id, TEST_ORG_ID);
+		expect(hydratedSchedule).toBeDefined();
+		const scheduleWithoutRepository = {
+			...hydratedSchedule,
+			repository: null,
+		};
+		vi.spyOn(scheduleQueries, "findById").mockResolvedValueOnce(fromAny(scheduleWithoutRepository));
+
+		// act
+		const result = await backupsService.validateBackupExecution(schedule.id);
+
+		// assert
+		expect(result.type).toBe("failure");
+		if (result.type === "failure") {
+			expect(result.error).toBeInstanceOf(NotFoundError);
+			expect(result.error.message).toBe("Repository not found");
+			expect(result.partialContext?.schedule).toBeDefined();
+			expect(result.partialContext?.volume).toBeDefined();
+		}
+	});
 	test("should fail backup when schedule does not exist", async () => {
 		setup();
 		// act
-		const result = await backupsExecutionService.validateBackupExecution(99999);
+		const result = await backupsService.validateBackupExecution(99999);
 
 		// assert
 		expect(result.type).toBe("failure");
@@ -106,7 +175,7 @@ describe("backup execution - validation failures", () => {
 			Promise.resolve({ exitCode: 1, summary: generateBackupOutput(), error: "failed" }),
 		);
 
-		await backupsExecutionService.executeBackup(schedule.id);
+		await backupsService.executeBackup(schedule.id);
 
 		expect(notificationSpy).toHaveBeenCalled();
 		expect(notificationSpy.mock.calls.at(-1)?.[2]?.error).toBe("failed");
@@ -128,7 +197,7 @@ describe("backup execution - validation failures", () => {
 			Promise.resolve({ exitCode: 1, summary: generateBackupOutput(), error: "manual failure" }),
 		);
 
-		await backupsExecutionService.executeBackup(schedule.id, true);
+		await backupsService.executeBackup(schedule.id, true);
 
 		expect(
 			errorSpy.mock.calls.some(([message]) => String(message).includes('Failed to parse cron expression ""')),
@@ -156,9 +225,9 @@ describe("stop backup", () => {
 			});
 		});
 
-		await backupsExecutionService.executeBackup(schedule.id);
+		await backupsService.executeBackup(schedule.id);
 
-		const updatedSchedule = await backupsService.getScheduleById(schedule.id);
+		const updatedSchedule = await getScheduleByIdOrShortId(schedule.id);
 		expect(updatedSchedule.lastBackupStatus).toBe("warning");
 		expect(updatedSchedule.lastBackupError).toBe("error: open /mnt/data/private.db: permission denied");
 	});
@@ -184,15 +253,63 @@ describe("stop backup", () => {
 			});
 		});
 
-		await backupsExecutionService.executeBackup(schedule.id);
+		await backupsService.executeBackup(schedule.id);
 
-		const updatedSchedule = await backupsService.getScheduleById(schedule.id);
+		const updatedSchedule = await getScheduleByIdOrShortId(schedule.id);
 		expect(updatedSchedule.lastBackupStatus).toBe("error");
 		expect(updatedSchedule.lastBackupError).toBe(
 			"Permissions 0755 for '/tmp/zerobyte-ssh-key' are too open.\nThis private key will be ignored.",
 		);
 	});
 
+	test("should block forget on the same repository until the active backup completes", async () => {
+		const { resticBackupMock, resticForgetMock, sendBackupMock } = setup();
+		const volume = await createTestVolume();
+		const repository = await createTestRepository();
+		const schedule = await createTestBackupSchedule({
+			volumeId: volume.id,
+			repositoryId: repository.id,
+			retentionPolicy: { keepHourly: 24 },
+		});
+
+		let completeBackup: (() => void) | undefined;
+		resticBackupMock.mockImplementationOnce(
+			() =>
+				new Promise((resolve) => {
+					completeBackup = () => resolve({ exitCode: 0, summary: generateBackupOutput(), error: "" });
+				}),
+		);
+
+		const backupPromise = backupsService.executeBackup(schedule.id);
+
+		await waitForExpect(() => {
+			expect(sendBackupMock).toHaveBeenCalledTimes(1);
+		});
+
+		let forgetFinished = false;
+		const forgetPromise = backupsService.runForget(schedule.id).finally(() => {
+			forgetFinished = true;
+		});
+
+		await new Promise((resolve) => setTimeout(resolve, 50));
+
+		expect(resticForgetMock).not.toHaveBeenCalled();
+		expect(forgetFinished).toBe(false);
+
+		expect(completeBackup).toBeDefined();
+		completeBackup?.();
+
+		await backupPromise;
+		await forgetPromise;
+
+		expect(resticForgetMock).toHaveBeenCalled();
+		expect(resticForgetMock).toHaveBeenCalledWith(
+			repository.config,
+			expect.objectContaining({ keepHourly: 24 }),
+			expect.objectContaining({ tag: schedule.shortId, organizationId: TEST_ORG_ID }),
+		);
+	});
+
 	test("should stop a running backup", async () => {
 		// arrange
 		const { resticBackupMock } = setup();
@@ -220,19 +337,19 @@ describe("stop backup", () => {
 			});
 		});
 
-		const executePromise = backupsExecutionService.executeBackup(schedule.id);
+		const executePromise = backupsService.executeBackup(schedule.id);
 
 		await waitForExpect(async () => {
-			const runningSchedule = await backupsService.getScheduleById(schedule.id);
+			const runningSchedule = await getScheduleByIdOrShortId(schedule.id);
 			expect(runningSchedule.lastBackupStatus).toBe("in_progress");
 		});
 
 		// act
-		await backupsExecutionService.stopBackup(schedule.id);
+		await backupsService.stopBackup(schedule.id);
 		await executePromise;
 
 		// assert
-		const updatedSchedule = await backupsService.getScheduleById(schedule.id);
+		const updatedSchedule = await getScheduleByIdOrShortId(schedule.id);
 		expect(updatedSchedule.lastBackupStatus).toBe("warning");
 		expect(updatedSchedule.lastBackupError).toBe("Backup was stopped by the user");
 	});
@@ -246,36 +363,25 @@ describe("stop backup", () => {
 			repositoryId: repository.id,
 		});
 
-		vi.spyOn(repoMutex, "acquireShared").mockImplementation((_repositoryId, _operation, signal) => {
+		const releaseLock = await repoMutex.acquireExclusive(repository.id, "test");
-			return new Promise((_, reject) => {
+		const executePromise = backupsService.executeBackup(schedule.id);
-				if (signal?.aborted) {
-					reject(signal.reason instanceof Error ? signal.reason : new Error("Operation aborted"));
-					return;
-				}
 
-				signal?.addEventListener(
+		try {
-					"abort",
+			await waitForExpect(async () => {
-					() => {
+				const queuedSchedule = await getScheduleByIdOrShortId(schedule.id);
-						reject(signal.reason instanceof Error ? signal.reason : new Error("Operation aborted"));
+				expect(queuedSchedule.lastBackupStatus).toBe("in_progress");
-					},
-					{ once: true },
-				);
 			});
-		});
 
-		const executePromise = backupsExecutionService.executeBackup(schedule.id);
+			expect(resticBackupMock).not.toHaveBeenCalled();
 
-		await waitForExpect(async () => {
+			await backupsService.stopBackup(schedule.id);
-			const queuedSchedule = await backupsService.getScheduleById(schedule.id);
+		} finally {
-			expect(queuedSchedule.lastBackupStatus).toBe("in_progress");
+			releaseLock();
-		});
+		}
 
-		expect(resticBackupMock).not.toHaveBeenCalled();
-
-		await backupsExecutionService.stopBackup(schedule.id);
 		await executePromise;
 
-		const updatedSchedule = await backupsService.getScheduleById(schedule.id);
+		const updatedSchedule = await getScheduleByIdOrShortId(schedule.id);
 		expect(updatedSchedule.lastBackupStatus).toBe("warning");
 		expect(updatedSchedule.lastBackupError).toBe("Backup was stopped by the user");
 		expect(resticBackupMock).not.toHaveBeenCalled();
@@ -297,7 +403,7 @@ describe("stop backup", () => {
 			Promise.resolve({ exitCode: 1, summary: generateBackupOutput(), error: "retry me" }),
 		);
 
-		await backupsExecutionService.executeBackup(schedule.id);
+		await backupsService.executeBackup(schedule.id);
 
 		const failedSchedule = await backupsService.getScheduleById(schedule.id);
 		expect(failedSchedule.failureRetryCount).toBe(1);
@@ -319,14 +425,14 @@ describe("stop backup", () => {
 			});
 		});
 
-		const executePromise = backupsExecutionService.executeBackup(schedule.id);
+		const executePromise = backupsService.executeBackup(schedule.id);
 
 		await waitForExpect(async () => {
 			const retryingSchedule = await backupsService.getScheduleById(schedule.id);
 			expect(retryingSchedule.lastBackupStatus).toBe("in_progress");
 		});
 
-		await backupsExecutionService.stopBackup(schedule.id);
+		await backupsService.stopBackup(schedule.id);
 		await executePromise;
 
 		const cancelledSchedule = await backupsService.getScheduleById(schedule.id);
@@ -348,20 +454,40 @@ describe("stop backup", () => {
 		});
 
 		// act & assert
-		await expect(backupsExecutionService.stopBackup(schedule.id)).rejects.toThrow(
+		await expect(backupsService.stopBackup(schedule.id)).rejects.toThrow(
 			"No backup is currently running for this schedule",
 		);
 
-		const updatedSchedule = await backupsService.getScheduleById(schedule.id);
+		const updatedSchedule = await getScheduleByIdOrShortId(schedule.id);
 		expect(updatedSchedule.lastBackupAt).toBe(previousLastBackupAt);
 		expect(updatedSchedule.lastBackupStatus).toBe("warning");
 		expect(updatedSchedule.lastBackupError).toBe("Backup was stopped by the user");
 	});
 
+	test("should reset a stuck in_progress status even when no backup is running", async () => {
+		// arrange
+		setup();
+		const volume = await createTestVolume();
+		const repository = await createTestRepository();
+		const schedule = await createTestBackupSchedule({
+			volumeId: volume.id,
+			repositoryId: repository.id,
+			lastBackupStatus: "in_progress",
+		});
+
+		// act
+		await backupsService.stopBackup(schedule.id).catch(() => {});
+
+		// assert
+		const updatedSchedule = await getScheduleByIdOrShortId(schedule.id);
+		expect(updatedSchedule.lastBackupStatus).toBe("warning");
+		expect(updatedSchedule.lastBackupError).toBe("Backup was stopped by the user");
+	});
+
 	test("should throw NotFoundError when schedule does not exist", async () => {
 		setup();
 		// act & assert
-		await expect(backupsExecutionService.stopBackup(99999)).rejects.toThrow("Backup schedule not found");
+		await expect(backupsService.stopBackup(99999)).rejects.toThrow("Backup schedule not found");
 	});
 });
 
@@ -382,7 +508,7 @@ describe("retention policy - runForget", () => {
 		});
 
 		// act
-		await backupsExecutionService.runForget(schedule.id);
+		await backupsService.runForget(schedule.id);
 
 		// assert
 		expect(resticForgetMock).toHaveBeenCalledWith(
@@ -411,7 +537,7 @@ describe("retention policy - runForget", () => {
 		});
 
 		// act & assert
-		await expect(backupsExecutionService.runForget(schedule.id)).rejects.toThrow(
+		await expect(backupsService.runForget(schedule.id)).rejects.toThrow(
 			"No retention policy configured for this schedule",
 		);
 	});
@@ -419,7 +545,7 @@ describe("retention policy - runForget", () => {
 	test("should throw NotFoundError when schedule does not exist", async () => {
 		setup();
 		// act & assert
-		await expect(backupsExecutionService.runForget(99999)).rejects.toThrow("Backup schedule not found");
+		await expect(backupsService.runForget(99999)).rejects.toThrow("Backup schedule not found");
 	});
 
 	test("should throw NotFoundError when repository does not exist", async () => {
@@ -432,9 +558,7 @@ describe("retention policy - runForget", () => {
 		});
 
 		// act & assert
-		await expect(backupsExecutionService.runForget(schedule.id, "non-existent-repo")).rejects.toThrow(
+		await expect(backupsService.runForget(schedule.id, "non-existent-repo")).rejects.toThrow("Repository not found");
-			"Repository not found",
-		);
 	});
 });
 
@@ -453,7 +577,7 @@ describe("mirror operations", () => {
 		await createTestBackupScheduleMirror(schedule.id, mirrorRepository.id);
 
 		// act
-		await backupsExecutionService.copyToMirrors(schedule.id, sourceRepository, null);
+		await backupsService.copyToMirrors(schedule.id, sourceRepository, null);
 
 		// assert
 		expect(resticCopyMock).toHaveBeenCalledWith(
@@ -480,7 +604,7 @@ describe("mirror operations", () => {
 		await createTestBackupScheduleMirror(schedule.id, mirrorRepository.id, { enabled: false });
 
 		// act
-		await backupsExecutionService.copyToMirrors(schedule.id, sourceRepository, null);
+		await backupsService.copyToMirrors(schedule.id, sourceRepository, null);
 
 		// assert
 		expect(resticCopyMock).not.toHaveBeenCalled();
@@ -500,7 +624,7 @@ describe("mirror operations", () => {
 		const mirror = await createTestBackupScheduleMirror(schedule.id, mirrorRepository.id);
 
 		// act
-		await backupsExecutionService.copyToMirrors(schedule.id, sourceRepository, null);
+		await backupsService.copyToMirrors(schedule.id, sourceRepository, null);
 
 		// assert
 		const mirrors = await backupsService.getMirrors(schedule.id);
@@ -531,7 +655,7 @@ describe("mirror operations", () => {
 		});
 
 		// act
-		await backupsExecutionService.copyToMirrors(schedule.id, sourceRepository, null);
+		await backupsService.copyToMirrors(schedule.id, sourceRepository, null);
 
 		// assert
 		const mirrors = await backupsService.getMirrors(schedule.id);
@@ -558,7 +682,7 @@ describe("mirror operations", () => {
 		resticCopyMock.mockImplementationOnce(() => Promise.reject(new Error("Copy failed")));
 
 		// act
-		await backupsExecutionService.copyToMirrors(schedule.id, sourceRepository, null);
+		await backupsService.copyToMirrors(schedule.id, sourceRepository, null);
 
 		// assert
 		const mirrors = await backupsService.getMirrors(schedule.id);
@@ -586,7 +710,7 @@ describe("mirror operations", () => {
 		resticCopyMock.mockImplementation(() => Promise.resolve({ success: true, output: "" }));
 
 		// act
-		await backupsExecutionService.copyToMirrors(schedule.id, sourceRepository, schedule.retentionPolicy);
+		await backupsService.copyToMirrors(schedule.id, sourceRepository, schedule.retentionPolicy);
 
 		await waitForExpect(() => {
 			expect(resticCopyMock).toHaveBeenCalled();
@@ -617,7 +741,7 @@ describe("mirror operations", () => {
 		resticForgetMock.mockClear();
 
 		// act
-		await backupsExecutionService.copyToMirrors(schedule.id, sourceRepository, schedule.retentionPolicy);
+		await backupsService.copyToMirrors(schedule.id, sourceRepository, schedule.retentionPolicy);
 
 		await waitForExpect(() => {
 			expect(resticCopyMock).toHaveBeenCalled();
@@ -660,10 +784,10 @@ describe("mirror operations", () => {
 		);
 		resticCopyMock.mockImplementation(() => Promise.resolve({ success: true, output: "" }));
 
-		const firstCopyPromise = backupsExecutionService.copyToMirrors(firstSchedule.id, sourceRepository, null);
+		const firstCopyPromise = backupsService.copyToMirrors(firstSchedule.id, sourceRepository, null);
 		await firstCopyStarted;
 
-		const secondCopyPromise = backupsExecutionService.copyToMirrors(secondSchedule.id, sourceRepository, null);
+		const secondCopyPromise = backupsService.copyToMirrors(secondSchedule.id, sourceRepository, null);
 
 		try {
 			const secondCopyState = await Promise.race<"resolved" | "timeout">([

app/server/modules/backups/__tests__/backups.service.test.ts

@@ -12,11 +12,14 @@ import { db } from "~/server/db/db";
 import { backupScheduleMirrorsTable, repositoriesTable, volumesTable } from "~/server/db/schema";
 import { TEST_ORG_ID } from "~/test/helpers/organization";
 import * as context from "~/server/core/request-context";
-import { backupsExecutionService } from "../backups.execution";
 import { repositoriesService } from "~/server/modules/repositories/repositories.service";
+import { agentManager } from "~/server/modules/agents/agents-manager";
+import { createAgentBackupMocks } from "~/test/helpers/agent-mock";
+import { getScheduleByIdOrShortId } from "../helpers/backup-schedule-lookups";
 
 const setup = () => {
-	const resticBackupMock = vi.fn(() => Promise.resolve({ exitCode: 0, summary: "", error: "" }));
+	const resticBackupMock = vi.fn((_: unknown) => Promise.resolve({ exitCode: 0, summary: "", error: "" }));
+	const { sendBackupMock, cancelBackupMock } = createAgentBackupMocks(resticBackupMock);
 	const refreshStatsMock = vi.fn(() =>
 		Promise.resolve({
 			total_size: 0,
@@ -29,10 +32,14 @@ const setup = () => {
 	);
 	vi.spyOn(spawnModule, "safeSpawn").mockImplementation(resticBackupMock);
 	vi.spyOn(repositoriesService, "refreshRepositoryStats").mockImplementation(refreshStatsMock);
+	vi.spyOn(agentManager, "sendBackup").mockImplementation(sendBackupMock);
+	vi.spyOn(agentManager, "cancelBackup").mockImplementation(cancelBackupMock);
 	vi.spyOn(context, "getOrganizationId").mockReturnValue(TEST_ORG_ID);
 
 	return {
 		resticBackupMock,
+		sendBackupMock,
+		cancelBackupMock,
 		refreshStatsMock,
 	};
 };
@@ -59,10 +66,10 @@ describe("execute backup", () => {
 		);
 
 		// act
-		await backupsExecutionService.executeBackup(schedule.id);
+		await backupsService.executeBackup(schedule.id);
 
 		// assert
-		const updatedSchedule = await backupsService.getScheduleById(schedule.id);
+		const updatedSchedule = await getScheduleByIdOrShortId(schedule.id);
 		expect(updatedSchedule.nextBackupAt).not.toBeNull();
 
 		const nextBackupAt = new Date(updatedSchedule.nextBackupAt ?? 0);
@@ -84,7 +91,7 @@ describe("execute backup", () => {
 		});
 
 		// act
-		await backupsExecutionService.executeBackup(schedule.id);
+		await backupsService.executeBackup(schedule.id);
 
 		// assert
 		expect(resticBackupMock).not.toHaveBeenCalled();
@@ -106,11 +113,11 @@ describe("execute backup", () => {
 		);
 
 		// act
-		await backupsExecutionService.executeBackup(schedule.id, true);
+		await backupsService.executeBackup(schedule.id, true);
 
 		// assert
 		expect(resticBackupMock).toHaveBeenCalled();
-		const updatedSchedule = await backupsService.getScheduleById(schedule.id);
+		const updatedSchedule = await getScheduleByIdOrShortId(schedule.id);
 		expect(updatedSchedule.lastBackupStatus).toBe("success");
 		expect(updatedSchedule.lastBackupAt).not.toBeNull();
 	});
@@ -132,10 +139,10 @@ describe("execute backup", () => {
 		);
 
 		// act
-		await backupsExecutionService.executeBackup(schedule.id, true);
+		await backupsService.executeBackup(schedule.id, true);
 
 		// assert
-		const updatedSchedule = await backupsService.getScheduleById(schedule.id);
+		const updatedSchedule = await getScheduleByIdOrShortId(schedule.id);
 		expect(updatedSchedule.nextBackupAt).toBeNull();
 	});
 
@@ -155,13 +162,13 @@ describe("execute backup", () => {
 		});
 
 		// act
-		void backupsExecutionService.executeBackup(schedule.id);
+		void backupsService.executeBackup(schedule.id);
 
 		await waitForExpect(() => {
 			expect(resticBackupMock).toHaveBeenCalledTimes(1);
 		});
 
-		await backupsExecutionService.executeBackup(schedule.id);
+		await backupsService.executeBackup(schedule.id);
 
 		// assert
 		expect(resticBackupMock).toHaveBeenCalledTimes(1);
@@ -182,10 +189,10 @@ describe("execute backup", () => {
 		);
 
 		// act
-		await backupsExecutionService.executeBackup(schedule.id);
+		await backupsService.executeBackup(schedule.id);
 
 		// assert
-		const updatedSchedule = await backupsService.getScheduleById(schedule.id);
+		const updatedSchedule = await getScheduleByIdOrShortId(schedule.id);
 		expect(updatedSchedule.lastBackupStatus).toBe("warning");
 	});
 
@@ -204,10 +211,10 @@ describe("execute backup", () => {
 		);
 
 		// act
-		await backupsExecutionService.executeBackup(schedule.id);
+		await backupsService.executeBackup(schedule.id);
 
 		// assert
-		const updatedSchedule = await backupsService.getScheduleById(schedule.id);
+		const updatedSchedule = await getScheduleByIdOrShortId(schedule.id);
 		expect(updatedSchedule.lastBackupStatus).toBe("error");
 	});
 });
@@ -229,7 +236,7 @@ describe("getSchedulesToExecute", () => {
 		});
 
 		// act
-		const schedulesToExecute = await backupsExecutionService.getSchedulesToExecute();
+		const schedulesToExecute = await backupsService.getSchedulesToExecute();
 
 		// assert
 		expect(schedulesToExecute).toContain(schedule.id);
@@ -246,7 +253,7 @@ describe("getScheduleByIdOrShortId", () => {
 			repositoryId: repository.id,
 		});
 
-		const found = await backupsService.getScheduleByIdOrShortId(String(schedule.id));
+		const found = await getScheduleByIdOrShortId(String(schedule.id));
 
 		expect(found.id).toBe(schedule.id);
 		expect(found.shortId).toBe(schedule.shortId);
@@ -261,7 +268,7 @@ describe("getScheduleByIdOrShortId", () => {
 			repositoryId: repository.id,
 		});
 
-		const found = await backupsService.getScheduleByIdOrShortId(schedule.shortId);
+		const found = await getScheduleByIdOrShortId(schedule.shortId);
 
 		expect(found.id).toBe(schedule.id);
 		expect(found.shortId).toBe(schedule.shortId);
@@ -274,10 +281,8 @@ describe("getScheduleByIdOrShortId", () => {
 			organizationId: otherOrgId,
 		});
 
-		await expect(backupsService.getScheduleByIdOrShortId(schedule.shortId)).rejects.toThrow(
+		await expect(getScheduleByIdOrShortId(schedule.shortId)).rejects.toThrow("Backup schedule not found");
-			"Backup schedule not found",
+		await expect(getScheduleByIdOrShortId(schedule.id)).rejects.toThrow("Backup schedule not found");
-		);
-		await expect(backupsService.getScheduleByIdOrShortId(schedule.id)).rejects.toThrow("Backup schedule not found");
 	});
 });
 

app/server/modules/backups/backup-executor.ts

@@ -1,12 +1,18 @@
-import { Effect } from "effect";
+import type { BackupSchedule, Volume, Repository } from "../../db/schema";
-import { restic } from "../../core/restic";
+import { logger } from "@zerobyte/core/node";
-import type { BackupSchedule, Repository, Volume } from "../../db/schema";
+import { resticDeps } from "../../core/restic";
-import type { ResticBackupOutputDto, ResticBackupProgressDto } from "@zerobyte/core/restic";
+import type { ResticBackupOutputDto } from "@zerobyte/core/restic";
-import { createBackupOptions } from "./backup.helpers";
+import type { BackupProgressPayload, BackupRunPayload } from "@zerobyte/contracts/agent-protocol";
+import { agentManager } from "../agents/agents-manager";
 import { getVolumePath } from "../volumes/helpers";
+import { decryptRepositoryConfig } from "../repositories/repository-config-secrets";
+import { createBackupOptions } from "./backup.helpers";
+
+const LOCAL_AGENT_ID = "local";
 
 type BackupExecutionRequest = {
 	scheduleId: number;
+	jobId: string;
 	schedule: BackupSchedule;
 	volume: Volume;
 	repository: Repository;
@@ -15,7 +21,14 @@ type BackupExecutionRequest = {
 	onProgress: (progress: BackupExecutionProgress) => void;
 };
 
-export type BackupExecutionProgress = ResticBackupProgressDto;
+type ActiveBackupExecution = {
+	scheduleId: number;
+	scheduleShortId: string;
+	onProgress: (progress: BackupExecutionProgress) => void;
+	resolve: (result: BackupExecutionResult) => void;
+};
+
+export type BackupExecutionProgress = BackupProgressPayload["progress"];
 
 export type BackupExecutionResult =
 	| {
@@ -30,15 +43,138 @@ export type BackupExecutionResult =
 	  }
 	| {
 			status: "failed";
-			error: unknown;
+			error: string;
 	  }
 	| {
 			status: "cancelled";
 			message?: string;
 	  };
 
+const activeExecutionsByJobId = new Map<string, ActiveBackupExecution>();
+const activeExecutionJobIdsByScheduleId = new Map<number, string>();
+const requestedCancellationsByScheduleId = new Set<number>();
 const activeControllersByScheduleId = new Map<number, AbortController>();
 
+const createBackupRunPayload = async ({
+	jobId,
+	schedule,
+	volume,
+	repository,
+	organizationId,
+}: BackupExecutionRequest): Promise<BackupRunPayload> => {
+	const sourcePath = getVolumePath(volume);
+	const { signal: _, ...options } = createBackupOptions(schedule, sourcePath);
+	const repositoryConfig = await decryptRepositoryConfig(repository.config);
+	const encryptedResticPassword = await resticDeps.getOrganizationResticPassword(organizationId);
+	const resticPassword = await resticDeps.resolveSecret(encryptedResticPassword);
+
+	return {
+		jobId,
+		scheduleId: schedule.shortId,
+		organizationId,
+		sourcePath,
+		repositoryConfig,
+		options: {
+			...options,
+			compressionMode: repository.compressionMode ?? "auto",
+		},
+		runtime: {
+			password: resticPassword,
+			cacheDir: resticDeps.resticCacheDir,
+			passFile: resticDeps.resticPassFile,
+			defaultExcludes: resticDeps.defaultExcludes,
+			rcloneConfigFile: resticDeps.rcloneConfigFile,
+			hostname: resticDeps.hostname,
+		},
+	};
+};
+
+const clearActiveExecution = (jobId: string) => {
+	const activeExecution = activeExecutionsByJobId.get(jobId);
+	if (!activeExecution) {
+		return null;
+	}
+
+	activeExecutionsByJobId.delete(jobId);
+	activeExecutionJobIdsByScheduleId.delete(activeExecution.scheduleId);
+	return activeExecution;
+};
+
+const clearExecutionState = (jobId: string, scheduleId: number) => {
+	requestedCancellationsByScheduleId.delete(scheduleId);
+	clearActiveExecution(jobId);
+};
+
+const resolveExecution = (jobId: string, activeExecution: ActiveBackupExecution, result: BackupExecutionResult) => {
+	clearExecutionState(jobId, activeExecution.scheduleId);
+	activeExecution.resolve(result);
+};
+
+const getActiveExecution = (jobId: string, scheduleId: string, eventName: string, agentId: string) => {
+	const activeExecution = activeExecutionsByJobId.get(jobId);
+	if (!activeExecution) {
+		logger.warn(`Received ${eventName} for unknown job ${jobId} from agent ${agentId}`);
+		return null;
+	}
+
+	if (activeExecution.scheduleShortId !== scheduleId) {
+		logger.warn(`Ignoring ${eventName} for job ${jobId} due to schedule mismatch ${scheduleId} from agent ${agentId}`);
+		return null;
+	}
+
+	return activeExecution;
+};
+
+agentManager.setBackupEventHandlers({
+	onBackupStarted: ({ agentId, payload }) => {
+		getActiveExecution(payload.jobId, payload.scheduleId, "backup.started", agentId);
+	},
+	onBackupProgress: ({ agentId, payload }) => {
+		const activeExecution = getActiveExecution(payload.jobId, payload.scheduleId, "backup.progress", agentId);
+		if (!activeExecution) {
+			return;
+		}
+
+		activeExecution.onProgress(payload.progress);
+	},
+	onBackupCompleted: ({ agentId, payload }) => {
+		const activeExecution = getActiveExecution(payload.jobId, payload.scheduleId, "backup.completed", agentId);
+		if (!activeExecution) {
+			return;
+		}
+
+		resolveExecution(payload.jobId, activeExecution, {
+			status: "completed",
+			exitCode: payload.exitCode,
+			result: payload.result,
+			warningDetails: payload.warningDetails ?? null,
+		});
+	},
+	onBackupFailed: ({ agentId, payload }) => {
+		const activeExecution = getActiveExecution(payload.jobId, payload.scheduleId, "backup.failed", agentId);
+		if (!activeExecution) {
+			return;
+		}
+
+		resolveExecution(payload.jobId, activeExecution, {
+			status: "failed",
+			error: payload.errorDetails ?? payload.error,
+		});
+	},
+	onBackupCancelled: ({ agentId, payload }) => {
+		const activeExecution = getActiveExecution(payload.jobId, payload.scheduleId, "backup.cancelled", agentId);
+		if (!activeExecution) {
+			return;
+		}
+
+		const wasRequested = requestedCancellationsByScheduleId.has(activeExecution.scheduleId);
+		resolveExecution(payload.jobId, activeExecution, {
+			status: "cancelled",
+			message: wasRequested ? undefined : payload.message,
+		});
+	},
+});
+
 export const backupExecutor = {
 	track: (scheduleId: number) => {
 		const abortController = new AbortController();
@@ -50,43 +186,67 @@ export const backupExecutor = {
 			activeControllersByScheduleId.delete(scheduleId);
 		}
 	},
-	execute: async (params: BackupExecutionRequest): Promise<BackupExecutionResult> => {
+	execute: async (request: Omit<BackupExecutionRequest, "jobId">) => {
-		const { schedule, volume, repository, organizationId, signal, onProgress } = params;
+		const jobId = Bun.randomUUIDv7();
+		const completion = new Promise<BackupExecutionResult>((resolve) => {
+			activeExecutionsByJobId.set(jobId, {
+				scheduleId: request.scheduleId,
+				scheduleShortId: request.schedule.shortId,
+				onProgress: request.onProgress,
+				resolve,
+			});
+			activeExecutionJobIdsByScheduleId.set(request.scheduleId, jobId);
+		});
+
 		try {
-			const volumePath = getVolumePath(volume);
+			if (request.signal.aborted) {
-			const backupOptions = createBackupOptions(schedule, volumePath, signal);
+				throw request.signal.reason || new Error("Operation aborted");
-
-			const execution = await Effect.runPromise(
-				restic
-					.backup(repository.config, volumePath, {
-						...backupOptions,
-						compressionMode: repository.compressionMode ?? "auto",
-						organizationId,
-						onProgress,
-					})
-					.pipe(
-						Effect.map((result) => ({ success: true as const, result })),
-						Effect.catchAll((error) => Effect.succeed({ success: false as const, error })),
-					),
-			);
-
-			if (!execution.success) {
-				throw execution.error;
 			}
 
-			const { exitCode, result, warningDetails } = execution.result;
+			const payload = await createBackupRunPayload({ ...request, jobId });
-			return { status: "completed", exitCode, result, warningDetails };
+
+			if (request.signal.aborted) {
+				throw request.signal.reason || new Error("Operation aborted");
+			}
+
+			if (!agentManager.sendBackup(LOCAL_AGENT_ID, payload)) {
+				clearExecutionState(jobId, request.scheduleId);
+				return {
+					status: "unavailable",
+					error: new Error("Local backup agent is not connected"),
+				} satisfies BackupExecutionResult;
+			}
+
+			return completion;
 		} catch (error) {
-			return { status: "failed", error };
+			clearExecutionState(jobId, request.scheduleId);
+			throw error;
 		}
 	},
 	cancel: (scheduleId: number) => {
 		const abortController = activeControllersByScheduleId.get(scheduleId);
-		if (!abortController) {
+		if (abortController) {
+			abortController.abort();
+		}
+
+		const jobId = activeExecutionJobIdsByScheduleId.get(scheduleId);
+		if (!jobId) {
+			return abortController !== undefined;
+		}
+
+		const activeExecution = activeExecutionsByJobId.get(jobId);
+		if (!activeExecution) {
+			activeExecutionJobIdsByScheduleId.delete(scheduleId);
+			requestedCancellationsByScheduleId.delete(scheduleId);
 			return false;
 		}
 
-		abortController.abort();
+		requestedCancellationsByScheduleId.add(scheduleId);
+		agentManager.cancelBackup(LOCAL_AGENT_ID, {
+			jobId,
+			scheduleId: activeExecution.scheduleShortId,
+		});
+
 		return true;
 	},
 };

app/server/modules/backups/backups.execution.ts

@@ -1,11 +0,0 @@
-import { backupsService } from "./backups.service";
-
-export const backupsExecutionService = {
-	executeBackup: backupsService.executeBackup,
-	validateBackupExecution: backupsService.validateBackupExecution,
-	getSchedulesToExecute: backupsService.getSchedulesToExecute,
-	stopBackup: backupsService.stopBackup,
-	runForget: backupsService.runForget,
-	copyToMirrors: backupsService.copyToMirrors,
-	getBackupProgress: backupsService.getBackupProgress,
-};

app/server/modules/backups/backups.service.ts

@@ -37,14 +37,6 @@ const listSchedules = async () => {
 	return schedules.filter((schedule) => schedule.volume && schedule.repository);
 };
 
-const getScheduleById = async (scheduleId: number) => {
-	return getScheduleByIdOrShortId(scheduleId);
-};
-
-const getScheduleByShortId = async (shortId: ShortId) => {
-	return getScheduleByIdOrShortId(shortId);
-};
-
 const createSchedule = async (data: CreateBackupScheduleBody) => {
 	const organizationId = getOrganizationId();
 	if (data.cronExpression && !isValidCron(data.cronExpression)) {
@@ -479,9 +471,6 @@ const stopBackup = async (scheduleId: number) => {
 
 export const backupsService = {
 	listSchedules,
-	getScheduleById,
-	getScheduleByShortId,
-	getScheduleByIdOrShortId,
 	createSchedule,
 	updateSchedule,
 	deleteSchedule,

app/test/helpers/agent-mock.ts

@@ -0,0 +1,105 @@
+import { vi } from "vitest";
+import { fromAny } from "@total-typescript/shoehorn";
+import { agentManager } from "~/server/modules/agents/agents-manager";
+
+export const createAgentBackupMocks = (
+	resticBackupMock: (params: never) => Promise<{
+		exitCode: number;
+		summary: string;
+		error: string;
+		stderr?: string;
+	}>,
+) => {
+	const runningJobs = new Map<string, { scheduleId: string; cancelled: boolean }>();
+
+	const sendBackupMock = vi.fn((_agentId: string, payload: { jobId: string; scheduleId: string }) => {
+		const handlers = agentManager.getBackupEventHandlers();
+
+		runningJobs.set(payload.jobId, { scheduleId: payload.scheduleId, cancelled: false });
+
+		handlers.onBackupStarted?.({
+			agentId: "local",
+			agentName: "local",
+			payload: { jobId: payload.jobId, scheduleId: payload.scheduleId },
+		});
+
+		void (async () => {
+			const stderrLines: string[] = [];
+			const result = await resticBackupMock(
+				fromAny({
+					onStderr: (line: string) => {
+						stderrLines.push(line);
+					},
+				}),
+			);
+			const running = runningJobs.get(payload.jobId);
+			if (!running || running.cancelled) {
+				return;
+			}
+
+			if (result.exitCode === 0 || result.exitCode === 3) {
+				let parsedResult: Record<string, unknown> | null = null;
+				if (result.summary) {
+					try {
+						parsedResult = JSON.parse(result.summary) as Record<string, unknown>;
+					} catch {
+						parsedResult = null;
+					}
+				}
+
+				handlers.onBackupCompleted?.({
+					agentId: "local",
+					agentName: "local",
+					payload: {
+						jobId: payload.jobId,
+						scheduleId: payload.scheduleId,
+						exitCode: result.exitCode,
+						result: fromAny(parsedResult),
+						warningDetails: stderrLines.join("\n") || undefined,
+					},
+				});
+			} else {
+				const resultWithStderr = result as typeof result & { stderr?: string };
+				const errorDetails = stderrLines.join("\n") || resultWithStderr.stderr || result.error;
+
+				handlers.onBackupFailed?.({
+					agentId: "local",
+					agentName: "local",
+					payload: {
+						jobId: payload.jobId,
+						scheduleId: payload.scheduleId,
+						error: result.error || `Backup failed with code ${result.exitCode}`,
+						errorDetails,
+					},
+				});
+			}
+
+			runningJobs.delete(payload.jobId);
+		})().catch(() => {});
+
+		return true;
+	});
+
+	const cancelBackupMock = vi.fn((_agentId: string, payload: { jobId: string; scheduleId: string }) => {
+		const running = runningJobs.get(payload.jobId);
+		if (!running) {
+			return false;
+		}
+
+		running.cancelled = true;
+		const handlers = agentManager.getBackupEventHandlers();
+		handlers.onBackupCancelled?.({
+			agentId: "local",
+			agentName: "local",
+			payload: {
+				jobId: payload.jobId,
+				scheduleId: payload.scheduleId,
+				message: "Backup was stopped by user",
+			},
+		});
+		runningJobs.delete(payload.jobId);
+		return true;
+	});
+
+	return { sendBackupMock, cancelBackupMock };
+};