fix: rebase conflicts

app/server/modules/agents/__tests__/agents-manager.backups.test.ts

@@ -0,0 +1,56 @@
+import { afterEach, expect, test, vi } from "vitest";
+import waitForExpect from "wait-for-expect";
+import { fromPartial } from "@total-typescript/shoehorn";
+import { agentManager } from "../agents-manager";
+import type { AgentManagerRuntime } from "../controller/server";
+import type { BackupRunPayload } from "@zerobyte/contracts/agent-protocol";
+
+type ProcessWithAgentRuntime = NodeJS.Process & {
+	__zerobyteAgentRuntime?: {
+		agentManager: AgentManagerRuntime | null;
+		localAgent: null;
+		isStoppingLocalAgent: boolean;
+		localAgentRestartTimeout: null;
+	};
+};
+
+const setAgentRuntime = (agentManagerRuntime: Partial<AgentManagerRuntime> | null) => {
+	(process as ProcessWithAgentRuntime).__zerobyteAgentRuntime = {
+		agentManager: agentManagerRuntime ? fromPartial<AgentManagerRuntime>(agentManagerRuntime) : null,
+		localAgent: null,
+		isStoppingLocalAgent: false,
+		localAgentRestartTimeout: null,
+	};
+};
+
+afterEach(() => {
+	delete (process as ProcessWithAgentRuntime).__zerobyteAgentRuntime;
+	vi.restoreAllMocks();
+});
+
+test("cancelBackup resolves a running backup when the cancel command cannot be delivered", async () => {
+	const sendBackup = vi.fn().mockResolvedValue(true);
+	const cancelBackup = vi.fn().mockResolvedValue(false);
+	setAgentRuntime({ sendBackup, cancelBackup });
+
+	const resultPromise = agentManager.runBackup("local", {
+		scheduleId: 42,
+		payload: fromPartial<BackupRunPayload>({
+			jobId: "job-1",
+			scheduleId: "schedule-1",
+		}),
+		signal: new AbortController().signal,
+		onProgress: vi.fn(),
+	});
+
+	await waitForExpect(() => {
+		expect(sendBackup).toHaveBeenCalledTimes(1);
+	});
+
+	await expect(agentManager.cancelBackup("local", 42)).resolves.toBe(true);
+	await expect(resultPromise).resolves.toEqual({ status: "cancelled" });
+	expect(cancelBackup).toHaveBeenCalledWith("local", {
+		jobId: "job-1",
+		scheduleId: "schedule-1",
+	});
+});

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

@@ -15,10 +15,6 @@ vi.mock("node:child_process", async () => {
 
 const { spawnLocalAgent, stopLocalAgent } = await import("../agents-manager");
 
-const flushMicrotasks = async () => {
-	await Promise.resolve();
-};
-
 type FakeChildProcess = EventEmitter & {
 	stdout: PassThrough;
 	stderr: PassThrough;
@@ -64,8 +60,7 @@ test("respawns the local agent after an unexpected exit", async () => {
 	firstChild.exitCode = 1;
 	firstChild.emit("exit", 1, null);
 
-	vi.advanceTimersByTime(1_000);
+	await vi.advanceTimersByTimeAsync(1_000);
-	await flushMicrotasks();
 
 	expect(spawnMock).toHaveBeenCalledTimes(2);
 });
@@ -79,8 +74,7 @@ test("does not respawn the local agent after an intentional stop", async () => {
 	await spawnLocalAgent();
 	await stopLocalAgent();
 
-	vi.advanceTimersByTime(1_000);
+	await vi.advanceTimersByTimeAsync(1_000);
-	await flushMicrotasks();
 
 	expect(spawnMock).toHaveBeenCalledTimes(1);
 	expect(child.kill).toHaveBeenCalledTimes(1);

app/server/modules/agents/__tests__/session.test.ts

@@ -5,9 +5,7 @@ import { fromPartial } from "@total-typescript/shoehorn";
 import { createAgentMessage } from "@zerobyte/contracts/agent-protocol";
 import { createControllerAgentSession } from "../controller/session";
 
-const createSocket = (
+const createSocket = (overrides: Partial<Parameters<typeof createControllerAgentSession>[0]> = {}) => {
-	overrides: Partial<Parameters<typeof createControllerAgentSession>[0]> = {},
-) => {
 	return fromPartial<Parameters<typeof createControllerAgentSession>[0]>({
 		data: { id: "connection-1", agentId: "local", organizationId: null, agentName: "Local Agent" },
 		send: vi.fn(() => 1),
@@ -16,10 +14,7 @@ const createSocket = (
 	});
 };
 
-const createSession = (
+const createSession = (handlers: Parameters<typeof createControllerAgentSession>[1] = {}, socket = createSocket()) => {
-	handlers: Parameters<typeof createControllerAgentSession>[1] = {},
-	socket = createSocket(),
-) => {
 	const scope = Effect.runSync(Scope.make());
 
 	try {
@@ -28,12 +23,15 @@ const createSession = (
 		return {
 			session,
 			run: () => {
-				Effect.runSync(Scope.extend(Effect.forkScoped(session.run), scope));
+				Effect.runFork(Scope.extend(session.run, scope));
 			},
 			socket,
 			close: () => {
 				Effect.runSync(Scope.close(scope, Exit.succeed(undefined)));
 			},
+			closeAsync: () => {
+				return Effect.runPromise(Scope.close(scope, Exit.succeed(undefined)));
+			},
 		};
 	} catch (error) {
 		Effect.runSync(Scope.close(scope, Exit.fail(error)));
@@ -62,7 +60,8 @@ test("close emits a synthetic backup.cancelled for a started backup", () => {
 	expect(onBackupCancelled).toHaveBeenCalledWith({
 		jobId: "job-1",
 		scheduleId: "schedule-1",
-		message: "The connection to the backup agent was lost while this backup was running. Restart the backup to ensure it completes.",
+		message:
+			"The connection to the backup agent was lost while this backup was running. Restart the backup to ensure it completes.",
 	});
 });
 
@@ -154,7 +153,8 @@ test("close emits a synthetic backup.cancelled for a queued backup", () => {
 	expect(onBackupCancelled).toHaveBeenCalledWith({
 		jobId: "job-queued",
 		scheduleId: "schedule-queued",
-		message: "The connection to the backup agent was lost before this backup started. Restart the backup to ensure it completes.",
+		message:
+			"The connection to the backup agent was lost before this backup started. Restart the backup to ensure it completes.",
 	});
 });
 
@@ -162,7 +162,7 @@ test("a dropped backup.cancel closes the session and emits a synthetic backup.ca
 	const send = vi.fn(() => 0);
 	const socket = createSocket({ send, close: vi.fn() });
 	const onBackupCancelled = vi.fn();
-	const { session, run, close: closeSession } = createSession({ onBackupCancelled }, socket);
+	const { session, run, closeAsync } = createSession({ onBackupCancelled }, socket);
 
 	try {
 		run();
@@ -193,6 +193,6 @@ test("a dropped backup.cancel closes the session and emits a synthetic backup.ca
 			});
 		});
 	} finally {
-		closeSession();
+		await closeAsync();
 	}
 });

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

@@ -1,9 +1,23 @@
 import type { ChildProcess } from "node:child_process";
-import type { BackupCancelPayload, BackupRunPayload } from "@zerobyte/contracts/agent-protocol";
+import { logger } from "@zerobyte/core/node";
+import type { ResticBackupOutputDto } from "@zerobyte/core/restic";
+import type { BackupProgressPayload, 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";
+export type BackupExecutionProgress = BackupProgressPayload["progress"];
+export type BackupExecutionResult =
+	| { status: "unavailable"; error: Error }
+	| { status: "completed"; exitCode: number; result: ResticBackupOutputDto | null; warningDetails: string | null }
+	| { status: "failed"; error: string }
+	| { status: "cancelled"; message?: string };
+
+export type AgentRunBackupRequest = {
+	scheduleId: number;
+	payload: BackupRunPayload;
+	signal: AbortSignal;
+	onProgress: (progress: BackupExecutionProgress) => void;
+};
 
 type AgentRuntimeState = {
 	agentManager: AgentManagerRuntime | null;
@@ -37,7 +51,147 @@ const getAgentRuntimeState = () => {
 
 const getAgentManagerRuntime = () => getAgentRuntimeState().agentManager;
 
-let backupEventHandlers: AgentBackupEventHandlers = {};
+type ActiveBackupRun = {
+	scheduleId: number;
+	jobId: string;
+	scheduleShortId: string;
+	onProgress: (progress: BackupExecutionProgress) => void;
+	resolve: (result: BackupExecutionResult) => void;
+	cancellationRequested: boolean;
+};
+
+const activeBackupsByScheduleId = new Map<number, ActiveBackupRun>();
+const activeBackupScheduleIdsByJobId = new Map<string, number>();
+
+const getUnavailableError = (agentId: string) => {
+	if (agentId === "local") {
+		return new Error("Local backup agent is not connected");
+	}
+
+	return new Error(`Backup agent ${agentId} is not connected`);
+};
+
+const clearActiveBackupRun = (scheduleId: number) => {
+	const activeBackupRun = activeBackupsByScheduleId.get(scheduleId);
+	if (!activeBackupRun) {
+		return null;
+	}
+
+	activeBackupsByScheduleId.delete(scheduleId);
+	activeBackupScheduleIdsByJobId.delete(activeBackupRun.jobId);
+	return activeBackupRun;
+};
+
+const resolveActiveBackupRun = (scheduleId: number, result: BackupExecutionResult) => {
+	const activeBackupRun = clearActiveBackupRun(scheduleId);
+	if (!activeBackupRun) {
+		return false;
+	}
+
+	activeBackupRun.resolve(result);
+	return true;
+};
+
+const getActiveBackupRun = (jobId: string, scheduleId: string, eventName: string, agentId: string) => {
+	const trackedScheduleId = activeBackupScheduleIdsByJobId.get(jobId);
+	if (trackedScheduleId === undefined) {
+		logger.warn(`Received ${eventName} for unknown job ${jobId} from agent ${agentId}`);
+		return null;
+	}
+
+	const activeBackupRun = activeBackupsByScheduleId.get(trackedScheduleId);
+	if (!activeBackupRun) {
+		logger.warn(`Received ${eventName} for inactive job ${jobId} from agent ${agentId}`);
+		return null;
+	}
+
+	if (activeBackupRun.scheduleShortId !== scheduleId) {
+		logger.warn(`Ignoring ${eventName} for job ${jobId} due to schedule mismatch ${scheduleId} from agent ${agentId}`);
+		return null;
+	}
+
+	return activeBackupRun;
+};
+
+const requestBackupCancellation = async (agentId: string, scheduleId: number) => {
+	const activeBackupRun = activeBackupsByScheduleId.get(scheduleId);
+	if (!activeBackupRun) {
+		return false;
+	}
+
+	if (activeBackupRun.cancellationRequested) {
+		return true;
+	}
+
+	activeBackupRun.cancellationRequested = true;
+
+	const runtime = getAgentManagerRuntime();
+	if (!runtime) {
+		resolveActiveBackupRun(scheduleId, { status: "cancelled" });
+		return true;
+	}
+
+	if (
+		await runtime.cancelBackup(agentId, {
+			jobId: activeBackupRun.jobId,
+			scheduleId: activeBackupRun.scheduleShortId,
+		})
+	) {
+		return true;
+	}
+
+	resolveActiveBackupRun(scheduleId, { status: "cancelled" });
+	return true;
+};
+
+const backupEventHandlers: AgentBackupEventHandlers = {
+	onBackupStarted: ({ agentId, payload }) => {
+		getActiveBackupRun(payload.jobId, payload.scheduleId, "backup.started", agentId);
+	},
+	onBackupProgress: ({ agentId, payload }) => {
+		const activeBackupRun = getActiveBackupRun(payload.jobId, payload.scheduleId, "backup.progress", agentId);
+		if (!activeBackupRun) {
+			return;
+		}
+
+		activeBackupRun.onProgress(payload.progress);
+	},
+	onBackupCompleted: ({ agentId, payload }) => {
+		const activeBackupRun = getActiveBackupRun(payload.jobId, payload.scheduleId, "backup.completed", agentId);
+		if (!activeBackupRun) {
+			return;
+		}
+
+		resolveActiveBackupRun(activeBackupRun.scheduleId, {
+			status: "completed",
+			exitCode: payload.exitCode,
+			result: payload.result,
+			warningDetails: payload.warningDetails ?? null,
+		});
+	},
+	onBackupFailed: ({ agentId, payload }) => {
+		const activeBackupRun = getActiveBackupRun(payload.jobId, payload.scheduleId, "backup.failed", agentId);
+		if (!activeBackupRun) {
+			return;
+		}
+
+		resolveActiveBackupRun(activeBackupRun.scheduleId, {
+			status: "failed",
+			error: payload.errorDetails ?? payload.error,
+		});
+	},
+	onBackupCancelled: ({ agentId, payload }) => {
+		const activeBackupRun = getActiveBackupRun(payload.jobId, payload.scheduleId, "backup.cancelled", agentId);
+		if (!activeBackupRun) {
+			return;
+		}
+
+		resolveActiveBackupRun(activeBackupRun.scheduleId, {
+			status: "cancelled",
+			message: activeBackupRun.cancellationRequested ? undefined : payload.message,
+		});
+	},
+};
 
 export const startAgentRuntime = async () => {
 	const runtime = getAgentRuntimeState();
@@ -55,28 +209,52 @@ export const startAgentRuntime = async () => {
 };
 
 export const agentManager = {
-	sendBackup: (agentId: string, payload: BackupRunPayload) => {
+	runBackup: async (agentId: string, request: AgentRunBackupRequest) => {
 		const runtime = getAgentManagerRuntime();
 		if (!runtime) {
-			return false;
+			return {
+				status: "unavailable",
+				error: getUnavailableError(agentId),
+			} satisfies BackupExecutionResult;
 		}
 
-		return runtime.sendBackup(agentId, payload);
+		if (request.signal.aborted) {
-	},
+			throw request.signal.reason || new Error("Operation aborted");
-	cancelBackup: (agentId: string, payload: BackupCancelPayload) => {
-		const runtime = getAgentManagerRuntime();
-		if (!runtime) {
-			return false;
 		}
 
-		return runtime.cancelBackup(agentId, payload);
+		const completion = new Promise<BackupExecutionResult>((resolve) => {
+			activeBackupsByScheduleId.set(request.scheduleId, {
+				scheduleId: request.scheduleId,
+				jobId: request.payload.jobId,
+				scheduleShortId: request.payload.scheduleId,
+				onProgress: request.onProgress,
+				resolve,
+				cancellationRequested: false,
+			});
+			activeBackupScheduleIdsByJobId.set(request.payload.jobId, request.scheduleId);
+		});
+
+		try {
+			if (!(await runtime.sendBackup(agentId, request.payload))) {
+				clearActiveBackupRun(request.scheduleId);
+				return {
+					status: "unavailable",
+					error: getUnavailableError(agentId),
+				} satisfies BackupExecutionResult;
+			}
+
+			if (request.signal.aborted) {
+				await requestBackupCancellation(agentId, request.scheduleId);
+			}
+
+			return completion;
+		} catch (error) {
+			clearActiveBackupRun(request.scheduleId);
+			throw error;
+		}
 	},
-	setBackupEventHandlers: (handlers: AgentBackupEventHandlers) => {
+	cancelBackup: async (agentId: string, scheduleId: number) => {
-		backupEventHandlers = handlers;
+		return requestBackupCancellation(agentId, scheduleId);
-		getAgentManagerRuntime()?.setBackupEventHandlers(handlers);
-	},
-	getBackupEventHandlers: () => {
-		return getAgentManagerRuntime()?.getBackupEventHandlers() ?? backupEventHandlers;
 	},
 };
 

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

@@ -231,7 +231,7 @@ export function createAgentManagerRuntime() {
 
 	return {
 		start,
-		sendBackup: (agentId: string, payload: BackupRunPayload) => {
+		sendBackup: async (agentId: string, payload: BackupRunPayload) => {
 			const session = getSession(agentId);
 
 			if (!session) {
@@ -244,7 +244,7 @@ export function createAgentManagerRuntime() {
 				return false;
 			}
 
-			if (!Effect.runSync(session.sendBackup(payload))) {
+			if (!(await Effect.runPromise(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: (agentId: string, payload: BackupCancelPayload) => {
+		cancelBackup: async (agentId: string, payload: BackupCancelPayload) => {
 			const session = getSession(agentId);
 
 			if (!session) {
@@ -260,7 +260,7 @@ export function createAgentManagerRuntime() {
 				return false;
 			}
 
-			if (!Effect.runSync(session.sendBackupCancel(payload))) {
+			if (!(await Effect.runPromise(session.sendBackupCancel(payload)))) {
 				logger.warn(`Cannot cancel backup command. Agent ${agentId} is no longer accepting commands.`);
 				return false;
 			}

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

@@ -57,6 +57,7 @@ export const createControllerAgentSession = (
 	handlers: ControllerAgentSessionHandlers = {},
 ): Effect.Effect<ControllerAgentSession, never, Scope.Scope> =>
 	Effect.gen(function* () {
+		let isClosed = false;
 		const outboundQueue = yield* Queue.bounded<ControllerWireMessage>(64);
 		const trackedBackupJobs = yield* Ref.make<Map<string, TrackedBackupJob>>(new Map());
 		const state = yield* Ref.make<SessionState>({
@@ -102,10 +103,7 @@ export const createControllerAgentSession = (
 			yield* updateState((current) => ({ ...current, isReady: false }));
 			const trackedJobs = yield* takeTrackedBackupJobs;
 			for (const [jobId, trackedJob] of trackedJobs) {
-				const message =
+				const message = "The connection to the backup agent was lost. Restart the backup to ensure it completes.";
-					trackedJob.state === "pending"
-						? "The connection to the backup agent was lost before this backup started. Restart the backup to ensure it completes."
-						: "The connection to the backup agent was lost while this backup was running. Restart the backup to ensure it completes.";
 
 				yield* Effect.sync(() => {
 					handlers.onBackupCancelled?.({ jobId, scheduleId: trackedJob.scheduleId, message });
@@ -115,36 +113,48 @@ export const createControllerAgentSession = (
 			yield* Queue.shutdown(outboundQueue);
 		});
 
-		yield* Effect.addFinalizer(() => releaseSession);
+		const closeSession = () =>
+			Effect.suspend(() => {
+				if (isClosed) {
+					return Effect.sync(() => undefined);
+				}
+
+				isClosed = true;
+				return releaseSession;
+			});
+
+		yield* Effect.addFinalizer(() => closeSession());
 
 		const handleSendFailure = (reason: string) => {
 			logger.error(
 				`Closing session for agent ${socket.data.agentId} on ${socket.data.id} after an outbound websocket send failed: ${reason}`,
 			);
 
-			try {
+			socket.close();
-				socket.close();
+
-			} catch (error) {
+			void Effect.runPromise(closeSession()).catch((error) => {
-				logger.error(`Failed to close socket for agent ${socket.data.agentId} on ${socket.data.id}: ${toMessage(error)}`);
+				logger.error(
-			}
+					`Failed to close session for agent ${socket.data.agentId} on ${socket.data.id}: ${toMessage(error)}`,
+				);
+			});
 		};
 
 		const run = Effect.gen(function* () {
 			yield* Effect.forkScoped(
 				Effect.forever(
-				Effect.gen(function* () {
+					Effect.gen(function* () {
-					const message = yield* Queue.take(outboundQueue);
+						const message = yield* Queue.take(outboundQueue);
-					yield* Effect.sync(() => {
+						yield* Effect.sync(() => {
-						try {
+							try {
-							const sendResult = socket.send(message);
+								const sendResult = socket.send(message);
-							if (sendResult <= 0) {
+								if (sendResult <= 0) {
-								handleSendFailure(sendResult === 0 ? "connection issue" : "backpressure");
+									handleSendFailure(sendResult === 0 ? "connection issue" : "backpressure");
+								}
+							} catch (error) {
+								handleSendFailure(toMessage(error));
 							}
-						} catch (error) {
+						});
-							handleSendFailure(toMessage(error));
+					}),
-						}
-					});
-				}),
 				),
 			);
 

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

@@ -28,7 +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 { runBackupMock, cancelBackupMock } = createAgentBackupMocks(resticBackupMock);
 	const refreshStatsMock = vi.fn(() =>
 		Promise.resolve({
 			total_size: 0,
@@ -44,7 +44,7 @@ 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, "runBackup").mockImplementation(runBackupMock);
 	vi.spyOn(agentManager, "cancelBackup").mockImplementation(cancelBackupMock);
 	vi.spyOn(context, "getOrganizationId").mockReturnValue(TEST_ORG_ID);
 
@@ -52,7 +52,7 @@ const setup = () => {
 		resticBackupMock,
 		resticForgetMock,
 		resticCopyMock,
-		sendBackupMock,
+		runBackupMock,
 		cancelBackupMock,
 		refreshStatsMock,
 	};
@@ -203,6 +203,27 @@ describe("backup execution - validation failures", () => {
 			errorSpy.mock.calls.some(([message]) => String(message).includes('Failed to parse cron expression ""')),
 		).toBe(false);
 	});
+
+	test("should fail backup when the local agent is unavailable", async () => {
+		const { runBackupMock } = setup();
+		const volume = await createTestVolume();
+		const repository = await createTestRepository();
+		const schedule = await createTestBackupSchedule({
+			volumeId: volume.id,
+			repositoryId: repository.id,
+		});
+
+		runBackupMock.mockResolvedValueOnce({
+			status: "unavailable",
+			error: new Error("Local backup agent is not connected"),
+		});
+
+		await backupsService.executeBackup(schedule.id);
+
+		const updatedSchedule = await getScheduleByIdOrShortId(schedule.id);
+		expect(updatedSchedule.lastBackupStatus).toBe("error");
+		expect(updatedSchedule.lastBackupError).toBe("Local backup agent is not connected");
+	});
 });
 
 describe("stop backup", () => {
@@ -263,7 +284,7 @@ describe("stop backup", () => {
 	});
 
 	test("should block forget on the same repository until the active backup completes", async () => {
-		const { resticBackupMock, resticForgetMock, sendBackupMock } = setup();
+		const { resticBackupMock, resticForgetMock, runBackupMock } = setup();
 		const volume = await createTestVolume();
 		const repository = await createTestRepository();
 		const schedule = await createTestBackupSchedule({
@@ -283,7 +304,7 @@ describe("stop backup", () => {
 		const backupPromise = backupsService.executeBackup(schedule.id);
 
 		await waitForExpect(() => {
-			expect(sendBackupMock).toHaveBeenCalledTimes(1);
+			expect(runBackupMock).toHaveBeenCalledTimes(1);
 		});
 
 		let forgetFinished = false;
@@ -354,6 +375,33 @@ describe("stop backup", () => {
 		expect(updatedSchedule.lastBackupError).toBe("Backup was stopped by the user");
 	});
 
+	test("should stop a running backup when the cancel command cannot be delivered", async () => {
+		const { resticBackupMock, cancelBackupMock } = setup();
+		const volume = await createTestVolume();
+		const repository = await createTestRepository();
+		const schedule = await createTestBackupSchedule({
+			volumeId: volume.id,
+			repositoryId: repository.id,
+		});
+
+		resticBackupMock.mockImplementation(() => new Promise(() => {}));
+		cancelBackupMock.mockResolvedValueOnce(false);
+
+		const executePromise = backupsService.executeBackup(schedule.id);
+
+		await waitForExpect(async () => {
+			const runningSchedule = await getScheduleByIdOrShortId(schedule.id);
+			expect(runningSchedule.lastBackupStatus).toBe("in_progress");
+		});
+
+		await backupsService.stopBackup(schedule.id);
+		await executePromise;
+
+		const updatedSchedule = await getScheduleByIdOrShortId(schedule.id);
+		expect(updatedSchedule.lastBackupStatus).toBe("warning");
+		expect(updatedSchedule.lastBackupError).toBe("Backup was stopped by the user");
+	});
+
 	test("should stop a queued backup before it acquires the repository lock", async () => {
 		const { resticBackupMock } = setup();
 		const volume = await createTestVolume();
@@ -405,7 +453,7 @@ describe("stop backup", () => {
 
 		await backupsService.executeBackup(schedule.id);
 
-		const failedSchedule = await backupsService.getScheduleById(schedule.id);
+		const failedSchedule = await getScheduleByIdOrShortId(schedule.id);
 		expect(failedSchedule.failureRetryCount).toBe(1);
 
 		resticBackupMock.mockImplementationOnce(({ signal }: SafeSpawnParams) => {
@@ -428,14 +476,14 @@ describe("stop backup", () => {
 		const executePromise = backupsService.executeBackup(schedule.id);
 
 		await waitForExpect(async () => {
-			const retryingSchedule = await backupsService.getScheduleById(schedule.id);
+			const retryingSchedule = await getScheduleByIdOrShortId(schedule.id);
 			expect(retryingSchedule.lastBackupStatus).toBe("in_progress");
 		});
 
 		await backupsService.stopBackup(schedule.id);
 		await executePromise;
 
-		const cancelledSchedule = await backupsService.getScheduleById(schedule.id);
+		const cancelledSchedule = await getScheduleByIdOrShortId(schedule.id);
 		expect(cancelledSchedule.lastBackupStatus).toBe("warning");
 		expect(cancelledSchedule.failureRetryCount).toBe(0);
 	});

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

@@ -8,6 +8,7 @@ import { createTestRepository } from "~/test/helpers/repository";
 import { generateBackupOutput } from "~/test/helpers/restic";
 import { faker } from "@faker-js/faker";
 import * as spawnModule from "@zerobyte/core/node";
+import type { SafeSpawnParams } from "@zerobyte/core/node";
 import { db } from "~/server/db/db";
 import { backupScheduleMirrorsTable, repositoriesTable, volumesTable } from "~/server/db/schema";
 import { TEST_ORG_ID } from "~/test/helpers/organization";
@@ -18,8 +19,8 @@ import { createAgentBackupMocks } from "~/test/helpers/agent-mock";
 import { getScheduleByIdOrShortId } from "../helpers/backup-schedule-lookups";
 
 const setup = () => {
-	const resticBackupMock = vi.fn((_: unknown) => Promise.resolve({ exitCode: 0, summary: "", error: "" }));
+	const resticBackupMock = vi.fn((_: SafeSpawnParams) => Promise.resolve({ exitCode: 0, summary: "", error: "" }));
-	const { sendBackupMock, cancelBackupMock } = createAgentBackupMocks(resticBackupMock);
+	const { runBackupMock, cancelBackupMock } = createAgentBackupMocks(resticBackupMock);
 	const refreshStatsMock = vi.fn(() =>
 		Promise.resolve({
 			total_size: 0,
@@ -32,13 +33,13 @@ const setup = () => {
 	);
 	vi.spyOn(spawnModule, "safeSpawn").mockImplementation(resticBackupMock);
 	vi.spyOn(repositoriesService, "refreshRepositoryStats").mockImplementation(refreshStatsMock);
-	vi.spyOn(agentManager, "sendBackup").mockImplementation(sendBackupMock);
+	vi.spyOn(agentManager, "runBackup").mockImplementation(runBackupMock);
 	vi.spyOn(agentManager, "cancelBackup").mockImplementation(cancelBackupMock);
 	vi.spyOn(context, "getOrganizationId").mockReturnValue(TEST_ORG_ID);
 
 	return {
 		resticBackupMock,
-		sendBackupMock,
+		runBackupMock,
 		cancelBackupMock,
 		refreshStatsMock,
 	};

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

@@ -1,9 +1,7 @@
 import type { BackupSchedule, Volume, Repository } from "../../db/schema";
-import { logger } from "@zerobyte/core/node";
 import { resticDeps } from "../../core/restic";
-import type { ResticBackupOutputDto } from "@zerobyte/core/restic";
+import type { BackupRunPayload } from "@zerobyte/contracts/agent-protocol";
-import type { BackupProgressPayload, BackupRunPayload } from "@zerobyte/contracts/agent-protocol";
+import { agentManager, type BackupExecutionProgress } from "../agents/agents-manager";
-import { agentManager } from "../agents/agents-manager";
 import { getVolumePath } from "../volumes/helpers";
 import { decryptRepositoryConfig } from "../repositories/repository-config-secrets";
 import { createBackupOptions } from "./backup.helpers";
@@ -12,7 +10,6 @@ const LOCAL_AGENT_ID = "local";
 
 type BackupExecutionRequest = {
 	scheduleId: number;
-	jobId: string;
 	schedule: BackupSchedule;
 	volume: Volume;
 	repository: Repository;
@@ -21,38 +18,8 @@ type BackupExecutionRequest = {
 	onProgress: (progress: BackupExecutionProgress) => void;
 };
 
-type ActiveBackupExecution = {
+export type { BackupExecutionResult } from "../agents/agents-manager";
-	scheduleId: number;
-	scheduleShortId: string;
-	onProgress: (progress: BackupExecutionProgress) => void;
-	resolve: (result: BackupExecutionResult) => void;
-};
 
-export type BackupExecutionProgress = BackupProgressPayload["progress"];
-
-export type BackupExecutionResult =
-	| {
-			status: "unavailable";
-			error: Error;
-	  }
-	| {
-			status: "completed";
-			exitCode: number;
-			result: ResticBackupOutputDto | null;
-			warningDetails: string | null;
-	  }
-	| {
-			status: "failed";
-			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 ({
@@ -61,7 +28,7 @@ const createBackupRunPayload = async ({
 	volume,
 	repository,
 	organizationId,
-}: BackupExecutionRequest): Promise<BackupRunPayload> => {
+}: BackupExecutionRequest & { jobId: string }): Promise<BackupRunPayload> => {
 	const sourcePath = getVolumePath(volume);
 	const { signal: _, ...options } = createBackupOptions(schedule, sourcePath);
 	const repositoryConfig = await decryptRepositoryConfig(repository.config);
@@ -89,92 +56,6 @@ const createBackupRunPayload = async ({
 	};
 };
 
-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();
@@ -187,66 +68,37 @@ export const backupExecutor = {
 		}
 	},
 	execute: async (request: Omit<BackupExecutionRequest, "jobId">) => {
+		const trackedAbortController = activeControllersByScheduleId.get(request.scheduleId);
+		if (!trackedAbortController || trackedAbortController.signal !== request.signal) {
+			throw new Error(`Backup execution for schedule ${request.scheduleId} was not tracked`);
+		}
+
 		const jobId = Bun.randomUUIDv7();
-		const completion = new Promise<BackupExecutionResult>((resolve) => {
+		if (request.signal.aborted) {
-			activeExecutionsByJobId.set(jobId, {
+			throw request.signal.reason || new Error("Operation aborted");
-				scheduleId: request.scheduleId,
+		}
-				scheduleShortId: request.schedule.shortId,
+
-				onProgress: request.onProgress,
+		const payload = await createBackupRunPayload({ ...request, jobId });
-				resolve,
+
-			});
+		if (request.signal.aborted) {
-			activeExecutionJobIdsByScheduleId.set(request.scheduleId, jobId);
+			throw request.signal.reason || new Error("Operation aborted");
+		}
+
+		return agentManager.runBackup(LOCAL_AGENT_ID, {
+			scheduleId: request.scheduleId,
+			payload,
+			signal: request.signal,
+			onProgress: request.onProgress,
 		});
-
-		try {
-			if (request.signal.aborted) {
-				throw request.signal.reason || new Error("Operation aborted");
-			}
-
-			const payload = await createBackupRunPayload({ ...request, jobId });
-
-			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) {
-			clearExecutionState(jobId, request.scheduleId);
-			throw error;
-		}
 	},
-	cancel: (scheduleId: number) => {
+	cancel: async (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;
 		}
 
-		requestedCancellationsByScheduleId.add(scheduleId);
+		abortController.abort();
-		agentManager.cancelBackup(LOCAL_AGENT_ID, {
+		await agentManager.cancelBackup(LOCAL_AGENT_ID, scheduleId);
-			jobId,
-			scheduleId: activeExecution.scheduleShortId,
-		});
-
 		return true;
 	},
 };

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

@@ -459,7 +459,7 @@ const stopBackup = async (scheduleId: number) => {
 	}
 
 	try {
-		if (!backupExecutor.cancel(scheduleId)) {
+		if (!(await backupExecutor.cancel(scheduleId))) {
 			throw new ConflictError("No backup is currently running for this schedule");
 		}
 

app/server/modules/backups/helpers/backup-lifecycle.ts

@@ -10,7 +10,7 @@ import { getOrganizationId } from "~/server/core/request-context";
 import type { BackupProgressEventDto } from "~/schemas/events-dto";
 import { calculateNextRun } from "../backup.helpers";
 import { scheduleQueries } from "../backups.queries";
-import type { BackupExecutionProgress } from "../backup-executor";
+import type { BackupExecutionProgress } from "../../agents/agents-manager";
 import { repositoriesService } from "../../repositories/repositories.service";
 import { copyToMirrors, runForget } from "./backup-maintenance";
 

app/test/helpers/agent-mock.ts

@@ -1,105 +1,95 @@
 import { vi } from "vitest";
-import { fromAny } from "@total-typescript/shoehorn";
+import { fromAny, fromPartial } from "@total-typescript/shoehorn";
-import { agentManager } from "~/server/modules/agents/agents-manager";
+import type { SafeSpawnParams } from "@zerobyte/core/node";
+import type { BackupExecutionResult } from "~/server/modules/agents/agents-manager";
 
 export const createAgentBackupMocks = (
-	resticBackupMock: (params: never) => Promise<{
+	resticBackupMock: (params: SafeSpawnParams) => Promise<{
 		exitCode: number;
 		summary: string;
 		error: string;
 		stderr?: string;
 	}>,
 ) => {
-	const runningJobs = new Map<string, { scheduleId: string; cancelled: boolean }>();
+	const runningBackups = new Map<number, { resolve: (result: BackupExecutionResult) => void; cancelled: boolean }>();
 
-	const sendBackupMock = vi.fn((_agentId: string, payload: { jobId: string; scheduleId: string }) => {
+	const runBackupMock = vi.fn(
-		const handlers = agentManager.getBackupEventHandlers();
+		async (_agentId: string, request: { scheduleId: number; payload: { jobId: string }; signal: AbortSignal }) => {
+			return new Promise<BackupExecutionResult>((resolve) => {
+				runningBackups.set(request.scheduleId, { resolve, cancelled: false });
 
-		runningJobs.set(payload.jobId, { scheduleId: payload.scheduleId, cancelled: false });
+				request.signal.addEventListener(
+					"abort",
+					() => {
+						const running = runningBackups.get(request.scheduleId);
+						if (!running || running.cancelled) {
+							return;
+						}
 
-		handlers.onBackupStarted?.({
+						running.cancelled = true;
-			agentId: "local",
+						runningBackups.delete(request.scheduleId);
-			agentName: "local",
+						resolve({ status: "cancelled" });
-			payload: { jobId: payload.jobId, scheduleId: payload.scheduleId },
-		});
-
-		void (async () => {
-			const stderrLines: string[] = [];
-			const result = await resticBackupMock(
-				fromAny({
-					onStderr: (line: string) => {
-						stderrLines.push(line);
 					},
-				}),
+					{ once: true },
-			);
+				);
-			const running = runningJobs.get(payload.jobId);
-			if (!running || running.cancelled) {
-				return;
-			}
 
-			if (result.exitCode === 0 || result.exitCode === 3) {
+				void (async () => {
-				let parsedResult: Record<string, unknown> | null = null;
+					const stderrLines: string[] = [];
-				if (result.summary) {
+					const result = await resticBackupMock(
-					try {
+						fromPartial<SafeSpawnParams>({
-						parsedResult = JSON.parse(result.summary) as Record<string, unknown>;
+							signal: request.signal,
-					} catch {
+							onStderr: (line: string) => {
-						parsedResult = null;
+								stderrLines.push(line);
+							},
+						}),
+					);
+					const running = runningBackups.get(request.scheduleId);
+					if (!running || running.cancelled) {
+						return;
 					}
-				}
 
-				handlers.onBackupCompleted?.({
+					runningBackups.delete(request.scheduleId);
-					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?.({
+					if (result.exitCode === 0 || result.exitCode === 3) {
-					agentId: "local",
+						let parsedResult: Record<string, unknown> | null = null;
-					agentName: "local",
+						if (result.summary) {
-					payload: {
+							try {
-						jobId: payload.jobId,
+								parsedResult = JSON.parse(result.summary) as Record<string, unknown>;
-						scheduleId: payload.scheduleId,
+							} catch {
-						error: result.error || `Backup failed with code ${result.exitCode}`,
+								parsedResult = null;
-						errorDetails,
+							}
-					},
+						}
-				});
-			}
 
-			runningJobs.delete(payload.jobId);
+						resolve({
-		})().catch(() => {});
+							status: "completed",
+							exitCode: result.exitCode,
+							result: fromAny(parsedResult),
+							warningDetails: stderrLines.join("\n") || null,
+						});
+						return;
+					}
 
-		return true;
+					const resultWithStderr = result as typeof result & { stderr?: string };
-	});
+					resolve({
+						status: "failed",
+						error: stderrLines.join("\n") || resultWithStderr.stderr || result.error,
+					});
+				})().catch(() => {});
+			});
+		},
+	);
 
-	const cancelBackupMock = vi.fn((_agentId: string, payload: { jobId: string; scheduleId: string }) => {
+	const cancelBackupMock = vi.fn(async (_agentId: string, scheduleId: number) => {
-		const running = runningJobs.get(payload.jobId);
+		const running = runningBackups.get(scheduleId);
 		if (!running) {
 			return false;
 		}
 
 		running.cancelled = true;
-		const handlers = agentManager.getBackupEventHandlers();
+		runningBackups.delete(scheduleId);
-		handlers.onBackupCancelled?.({
+		running.resolve({ status: "cancelled" });
-			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 };
+	return { runBackupMock, cancelBackupMock };
 };

apps/agent/tsconfig.json

@@ -2,12 +2,12 @@
 	"compilerOptions": {
 		"plugins": [{ "name": "@effect/language-service" }],
 		// Environment setup & latest features
-		"lib": ["DOM", "ESNext"],
+		"lib": ["ESNext"],
 		"target": "ESNext",
 		"module": "Preserve",
 		"moduleDetection": "force",
 		"jsx": "react-jsx",
-		"types": ["node", "bun"],
+		"types": ["bun", "node"],
 		"allowJs": true,
 
 		// Bundler mode