refactor(agent): move volume lifecycle cleanup to agent

app/drizzle/20260505165117_early_purple_man/migration.sql

@@ -13,4 +13,4 @@ CREATE TABLE `agents_table` (
 );
 --> statement-breakpoint
 CREATE INDEX `agents_table_organization_id_idx` ON `agents_table` (`organization_id`);--> statement-breakpoint
-CREATE INDEX `agents_table_status_idx` ON `agents_table` (`status`);
+CREATE INDEX `agents_table_status_idx` ON `agents_table` (`status`);

app/server/jobs/__tests__/cleanup-dangling.test.ts

@@ -0,0 +1,18 @@
+import { afterEach, expect, test, vi } from "vitest";
+import { config } from "../../core/config";
+import { CleanupDanglingMountsJob } from "../cleanup-dangling";
+import * as mountinfo from "../../utils/mountinfo";
+
+afterEach(() => {
+	config.flags.enableLocalAgent = true;
+	vi.restoreAllMocks();
+});
+
+test("skips controller-local mount inspection when local volume execution is agent-owned", async () => {
+	config.flags.enableLocalAgent = true;
+	const readMountInfo = vi.spyOn(mountinfo, "readMountInfo");
+
+	await new CleanupDanglingMountsJob().run();
+
+	expect(readMountInfo).not.toHaveBeenCalled();
+});

app/server/jobs/cleanup-dangling.ts

@@ -10,9 +10,16 @@ import { toMessage } from "../utils/errors";
 import { VOLUME_MOUNT_BASE } from "../core/constants";
 import { db } from "../db/db";
 import { withContext } from "../core/request-context";
+import { config } from "../core/config";
+import { LOCAL_AGENT_ID } from "../modules/agents/constants";
 
 export class CleanupDanglingMountsJob extends Job {
 	async run() {
+		if (config.flags.enableLocalAgent) {
+			logger.debug("Skipping controller-local dangling mount cleanup because local volume execution is agent-owned.");
+			return { done: true, timestamp: new Date() };
+		}
+
 		const organizations = await db.query.organization.findMany({});
 		if (organizations.length === 0) {
 			logger.warn("No organizations found; skipping dangling mount cleanup to avoid false positives.");
@@ -22,7 +29,7 @@ export class CleanupDanglingMountsJob extends Job {
 		const allVolumes = [];
 		for (const org of organizations) {
 			const volumes = await withContext({ organizationId: org.id }, async () => volumeService.listVolumes());
-			allVolumes.push(...volumes);
+			allVolumes.push(...volumes.filter((volume) => volume.agentId === LOCAL_AGENT_ID));
 		}
 
 		const allSystemMounts = await readMountInfo();

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

@@ -48,11 +48,13 @@ test("cancelBackup resolves a running backup when the cancel command cannot be d
 });
 
 test("runVolumeCommand sends the command to the selected agent", async () => {
-	const runVolumeCommand = vi.fn().mockResolvedValue({
-		commandId: "command-1",
-		status: "success",
-		command: { name: "volume.mount", result: { status: "mounted" } },
-	});
+	const runVolumeCommand = vi.fn(() =>
+		Effect.succeed({
+			commandId: "command-1",
+			status: "success" as const,
+			command: { name: "volume.mount" as const, result: { status: "mounted" as const } },
+		}),
+	);
 	setAgentRuntime({ runVolumeCommand });
 
 	const command = fromPartial<VolumeCommand>({ name: "volume.mount", volume: { agentId: "agent-1" } });

app/server/modules/lifecycle/__tests__/shutdown.test.ts

@@ -1,9 +1,10 @@
 import { afterEach, describe, expect, test, vi } from "vitest";
 import { Scheduler } from "../../../core/scheduler";
-import * as volumeHostModule from "../../../../../apps/agent/src/volume-host";
-import type { VolumeBackend } from "../../../../../apps/agent/src/volume-host";
 import * as bootstrapModule from "../bootstrap";
+import { agentManager } from "../../agents/agents-manager";
 import { createTestVolume } from "~/test/helpers/volume";
+import { config } from "~/server/core/config";
+import { db } from "~/server/db/db";
 
 const loadShutdownModule = async () => {
 	const moduleUrl = new URL("../shutdown.ts", import.meta.url);
@@ -12,11 +13,12 @@ const loadShutdownModule = async () => {
 };
 
 afterEach(() => {
+	config.flags.enableLocalAgent = true;
 	vi.restoreAllMocks();
 });
 
 describe("shutdown", () => {
-	test("stops the agent runtime before unmounting mounted volumes", async () => {
+	test("does not unmount agent-owned volumes during controller shutdown", async () => {
 		const events: string[] = [];
 		const stopScheduler = vi.fn(async () => {
 			events.push("scheduler.stop");
@@ -24,35 +26,51 @@ describe("shutdown", () => {
 		const stopApplicationRuntime = vi.fn(async () => {
 			events.push("agents.stop");
 		});
-		const unmountVolume = vi.fn(async () => {
-			events.push("backend.unmount");
-			return { status: "unmounted" as const };
-		});
+		const runVolumeCommand = vi.spyOn(agentManager, "runVolumeCommand");
 
-		await createTestVolume({
+		const volume = await createTestVolume({
 			name: "Shutdown test volume",
 			config: {
 				backend: "directory",
 				path: "/Applications",
 			},
 			status: "mounted",
+			agentId: "agent-1",
 		});
 
 		vi.spyOn(Scheduler, "stop").mockImplementation(stopScheduler);
 		vi.spyOn(bootstrapModule, "stopApplicationRuntime").mockImplementation(stopApplicationRuntime);
-		vi.spyOn(volumeHostModule, "createVolumeBackend").mockImplementation(
-			() =>
-				({
-					mount: async () => ({ status: "mounted" as const }),
-					unmount: unmountVolume,
-					checkHealth: async () => ({ status: "mounted" as const }),
-				}) satisfies VolumeBackend,
-		);
 
 		const { shutdown } = await loadShutdownModule();
 
 		await shutdown();
 
-		expect(events).toEqual(["scheduler.stop", "agents.stop", "backend.unmount"]);
+		expect(events).toEqual(["scheduler.stop", "agents.stop"]);
+		expect(runVolumeCommand).not.toHaveBeenCalled();
+	});
+
+	test("keeps legacy controller-local fallback unmount on shutdown", async () => {
+		config.flags.enableLocalAgent = false;
+		const events: string[] = [];
+		vi.spyOn(Scheduler, "stop").mockImplementation(async () => {
+			events.push("scheduler.stop");
+		});
+		vi.spyOn(bootstrapModule, "stopApplicationRuntime").mockImplementation(async () => {
+			events.push("agents.stop");
+		});
+
+		const volume = await createTestVolume({
+			name: "Fallback shutdown test volume",
+			config: { backend: "directory", path: "/Applications" },
+			status: "mounted",
+		});
+
+		const { shutdown } = await loadShutdownModule();
+
+		await shutdown();
+
+		const updated = await db.query.volumesTable.findFirst({ where: { id: volume.id } });
+		expect(events).toEqual(["scheduler.stop", "agents.stop"]);
+		expect(updated?.status).toBe("unmounted");
 	});
 });

app/server/modules/lifecycle/shutdown.ts

@@ -2,29 +2,28 @@ import { Scheduler } from "../../core/scheduler";
 import { db } from "../../db/db";
 import { logger } from "@zerobyte/core/node";
 import { stopApplicationRuntime } from "./bootstrap";
-import { decryptVolumeConfig } from "../volumes/volume-config-secrets";
-import {
-	createVolumeBackend,
-	type AgentVolume,
-	type BackendConfig as HostBackendConfig,
-} from "../../../../apps/agent/src/volume-host";
+import { withContext } from "../../core/request-context";
+import { volumeService } from "../volumes/volume.service";
+import { toMessage } from "../../utils/errors";
+import { config } from "../../core/config";
+import { LOCAL_AGENT_ID } from "../agents/constants";
 
 export const shutdown = async () => {
 	await Scheduler.stop();
-	await stopApplicationRuntime();
 
-	const volumes = await db.query.volumesTable.findMany({
-		where: { status: "mounted" },
-	});
+	if (!config.flags.enableLocalAgent) {
+		const volumes = await db.query.volumesTable.findMany({
+			where: { AND: [{ status: "mounted" }, { agentId: LOCAL_AGENT_ID }] },
+		});
 
-	for (const volume of volumes) {
-		const backend = createVolumeBackend({
-			...volume,
-			config: (await decryptVolumeConfig(volume.config)) as HostBackendConfig,
-			provisioningId: volume.provisioningId ?? null,
-		} satisfies AgentVolume);
-		const { status, error } = await backend.unmount();
+		for (const volume of volumes) {
+			const { status, error } = await withContext({ organizationId: volume.organizationId }, () =>
+				volumeService.unmountVolume(volume.shortId),
+			).catch((error) => ({ status: "error" as const, error: toMessage(error) }));
 
-		logger.info(`Volume ${volume.name} unmount status: ${status}${error ? `, error: ${error}` : ""}`);
+			logger.info(`Volume ${volume.name} unmount status: ${status}${error ? `, error: ${error}` : ""}`);
+		}
 	}
+
+	await stopApplicationRuntime();
 };

app/server/modules/lifecycle/startup.ts

@@ -17,6 +17,7 @@ import { backupsService } from "../backups/backups.service";
 import { config } from "~/server/core/config";
 import { syncProvisionedResources } from "../provisioning/provisioning";
 import { toMessage } from "~/server/utils/errors";
+import { LOCAL_AGENT_ID } from "../agents/constants";
 
 const ensureLatestConfigurationSchema = async () => {
 	const volumes = await db.query.volumesTable.findMany({});
@@ -71,23 +72,30 @@ export const startup = async () => {
 		logger.warn(`Removed ${deletedSchedules} orphaned backup schedule(s) during startup`);
 	}
 
-	const volumes = await db.query.volumesTable.findMany({
-		where: {
-			OR: [
-				{ status: "mounted" },
-				{
-					AND: [{ autoRemount: true }, { status: "error" }],
-				},
-			],
-		},
-	});
-
-	for (const volume of volumes) {
-		await withContext({ organizationId: volume.organizationId }, async () => {
-			await volumeService.mountVolume(volume.shortId).catch((err) => {
-				logger.error(`Error auto-remounting volume ${volume.name} on startup: ${err.message}`);
-			});
+	if (!config.flags.enableLocalAgent) {
+		const volumes = await db.query.volumesTable.findMany({
+			where: {
+				AND: [
+					{ agentId: LOCAL_AGENT_ID },
+					{
+						OR: [
+							{ status: "mounted" },
+							{
+								AND: [{ autoRemount: true }, { status: "error" }],
+							},
+						],
+					},
+				],
+			},
 		});
+
+		for (const volume of volumes) {
+			await withContext({ organizationId: volume.organizationId }, async () => {
+				await volumeService.mountVolume(volume.shortId).catch((err) => {
+					logger.error(`Error auto-remounting volume ${volume.name} on startup: ${err.message}`);
+				});
+			});
+		}
 	}
 
 	await db
@@ -102,7 +110,9 @@ export const startup = async () => {
 			logger.error(`Failed to update stuck backup schedules on startup: ${err.message}`);
 		});
 
-	Scheduler.build(CleanupDanglingMountsJob).schedule("0 * * * *");
+	if (!config.flags.enableLocalAgent) {
+		Scheduler.build(CleanupDanglingMountsJob).schedule("0 * * * *");
+	}
 	Scheduler.build(VolumeHealthCheckJob).schedule("*/30 * * * *");
 	Scheduler.build(RepositoryHealthCheckJob).schedule("50 12 * * *");
 	Scheduler.build(BackupExecutionJob).schedule("* * * * *");

app/server/modules/volumes/volume.service.ts

@@ -32,19 +32,8 @@ import {
 } from "../../../../apps/agent/src/volume-host/operations";
 
 type EnsureHealthyVolumeResult =
-	| {
-			ready: true;
-			volume: Volume;
-			remounted: boolean;
-	  }
-	| {
-			ready: false;
-			volume: Volume;
-			reason: string;
-	  };
-
-type VolumeBackendCommandName = "volume.mount" | "volume.unmount" | "volume.checkHealth";
-type VolumeBackendCommandResult = Extract<VolumeCommandResult, { name: VolumeBackendCommandName }>["result"];
+	| { ready: true; volume: Volume; remounted: boolean }
+	| { ready: false; volume: Volume; reason: string };
 
 const listVolumes = async () => {
 	const organizationId = getOrganizationId();
@@ -86,14 +75,18 @@ const volumeForHost = async (volume: Volume): Promise<AgentVolume> => ({
 	provisioningId: volume.provisioningId ?? null,
 });
 
-// TODO(agent-rollout): Remove the local host execution branch once all installs run volume operations through agents.
+// Transitional fallback: older controller-only installs do not run the supervised local agent.
+// Keep all controller-local host execution behind this predicate so the fallback is easy to delete
+// once volume operations always go through the local agent.
 const shouldRunViaAgent = (volume: Volume) => volume.agentId !== LOCAL_AGENT_ID || config.flags.enableLocalAgent;
 
+const shouldUseControllerLocalVolumeFallback = (volume: Volume) => !shouldRunViaAgent(volume);
+
 const runVolumeBackendCommand = async (
 	volume: Volume,
 	name: "volume.mount" | "volume.unmount" | "volume.checkHealth",
 ) => {
-	if (!shouldRunViaAgent(volume)) {
+	if (shouldUseControllerLocalVolumeFallback(volume)) {
 		const backend = createVolumeBackend(await volumeForHost(volume));
 		switch (name) {
 			case "volume.mount":
@@ -108,8 +101,8 @@ const runVolumeBackendCommand = async (
 	const command = await runVolumeCommand(volume.agentId, {
 		name,
 		volume: await volumeForAgent(volume),
-	} as VolumeCommand);
-	return command.result as VolumeBackendCommandResult;
+	});
+	return command.result;
 };
 
 const mapAgentFileError = (error: unknown) => {
@@ -229,7 +222,7 @@ const getVolume = async (shortId: ShortId) => {
 	let statfs: Partial<StatFs> = {};
 	if (volume.status === "mounted") {
 		statfs = await withTimeout(
-			shouldRunViaAgent(volume)
+			!shouldUseControllerLocalVolumeFallback(volume)
 				? runVolumeCommand(volume.agentId, { name: "volume.statfs", volume: await volumeForAgent(volume) }).then(
 						(command) => command.result,
 					)
@@ -402,7 +395,7 @@ const listFiles = async (shortId: ShortId, subPath?: string, offset: number = 0,
 	}
 
 	try {
-		if (!shouldRunViaAgent(volume)) {
+		if (shouldUseControllerLocalVolumeFallback(volume)) {
 			return await listVolumeFiles(await volumeForHost(volume), subPath, offset, limit);
 		}
 

apps/agent/src/commands/volume.test.ts

@@ -0,0 +1,127 @@
+import { afterEach, expect, test, vi } from "vitest";
+import { Effect } from "effect";
+import { fromPartial } from "@total-typescript/shoehorn";
+import {
+	parseAgentMessage,
+	type AgentWireMessage,
+	type VolumeCommandPayload,
+} from "@zerobyte/contracts/agent-protocol";
+import type { ControllerCommandContext } from "../context";
+
+const volumeHostMock = vi.hoisted(() => ({
+	createVolumeBackend: vi.fn(),
+	getStatFs: vi.fn(),
+	getVolumePath: vi.fn(),
+}));
+
+const operationsMock = vi.hoisted(() => ({
+	browseFilesystem: vi.fn(),
+	listVolumeFiles: vi.fn(),
+	testVolumeConnection: vi.fn(),
+}));
+
+vi.mock("../volume-host", () => volumeHostMock);
+vi.mock("../volume-host/operations", () => operationsMock);
+
+import { handleVolumeCommand } from "./volume";
+
+afterEach(() => {
+	vi.restoreAllMocks();
+	volumeHostMock.createVolumeBackend.mockReset();
+	volumeHostMock.getStatFs.mockReset();
+	volumeHostMock.getVolumePath.mockReset();
+	operationsMock.browseFilesystem.mockReset();
+	operationsMock.listVolumeFiles.mockReset();
+	operationsMock.testVolumeConnection.mockReset();
+});
+
+const runVolumeCommand = async (payload: VolumeCommandPayload) => {
+	const outboundMessages: string[] = [];
+	const context = fromPartial<ControllerCommandContext>({
+		offerOutbound: (message: AgentWireMessage) =>
+			Effect.sync(() => {
+				outboundMessages.push(message);
+				return true;
+			}),
+	});
+
+	await Effect.runPromise(handleVolumeCommand(context, payload));
+	return outboundMessages.map((message) => parseAgentMessage(message));
+};
+
+test("runs backend-backed volume commands on the agent host", async () => {
+	const mount = vi.fn(async () => ({ status: "mounted" as const }));
+	volumeHostMock.createVolumeBackend.mockReturnValue({ mount });
+
+	const messages = await runVolumeCommand(
+		fromPartial<VolumeCommandPayload>({
+			commandId: "command-1",
+			command: {
+				name: "volume.mount",
+				volume: { id: 1, config: { backend: "directory", path: "/tmp/source" }, provisioningId: undefined },
+			},
+		}),
+	);
+
+	expect(volumeHostMock.createVolumeBackend).toHaveBeenCalledWith(
+		expect.objectContaining({ id: 1, config: { backend: "directory", path: "/tmp/source" }, provisioningId: null }),
+	);
+	expect(mount).toHaveBeenCalledOnce();
+	expect(messages[0]?.success).toBe(true);
+	if (messages[0]?.success && messages[0].data.type === "volume.commandResult") {
+		expect(messages[0].data.payload).toEqual({
+			commandId: "command-1",
+			status: "success",
+			command: { name: "volume.mount", result: { status: "mounted" } },
+		});
+	}
+});
+
+test("returns command errors without throwing", async () => {
+	operationsMock.browseFilesystem.mockRejectedValue(new Error("permission denied"));
+
+	const messages = await runVolumeCommand({
+		commandId: "command-2",
+		command: { name: "filesystem.browse", path: "/root" },
+	});
+
+	expect(messages[0]?.success).toBe(true);
+	if (messages[0]?.success && messages[0].data.type === "volume.commandResult") {
+		expect(messages[0].data.payload).toEqual({
+			commandId: "command-2",
+			status: "error",
+			error: "permission denied",
+		});
+	}
+});
+
+test("routes file listing commands to host operations", async () => {
+	operationsMock.listVolumeFiles.mockResolvedValue({
+		files: [],
+		path: "/logs",
+		offset: 0,
+		limit: 10,
+		total: 0,
+		hasMore: false,
+	});
+
+	await runVolumeCommand(
+		fromPartial<VolumeCommandPayload>({
+			commandId: "command-3",
+			command: {
+				name: "volume.listFiles",
+				volume: { id: 1, config: { backend: "directory", path: "/tmp/source" } },
+				subPath: "/logs",
+				offset: 0,
+				limit: 10,
+			},
+		}),
+	);
+
+	expect(operationsMock.listVolumeFiles).toHaveBeenCalledWith(
+		expect.objectContaining({ id: 1, config: { backend: "directory", path: "/tmp/source" } }),
+		"/logs",
+		0,
+		10,
+	);
+});

apps/agent/src/index.ts

@@ -1,14 +1,32 @@
 import { logger } from "@zerobyte/core/node";
 import { createControllerSession, type ControllerSession } from "./controller-session";
+import { cleanupDanglingVolumeMountDirectories } from "./volume-host/cleanup";
 
 const controllerUrl = process.env.ZEROBYTE_CONTROLLER_URL;
 const agentToken = process.env.ZEROBYTE_AGENT_TOKEN;
 const RECONNECT_DELAY_MS = 1000;
+const VOLUME_CLEANUP_INTERVAL_MS = 60 * 60 * 1000;
 
 export class Agent {
 	private ws: WebSocket | null = null;
 	private controllerSession: ControllerSession | null = null;
 	private reconnectTimeout: ReturnType<typeof setTimeout> | null = null;
+	private volumeCleanupInterval: ReturnType<typeof setInterval> | null = null;
+
+	private runVolumeCleanup() {
+		void cleanupDanglingVolumeMountDirectories().catch((error) => {
+			logger.warn(`Agent volume cleanup failed: ${error instanceof Error ? error.message : String(error)}`);
+		});
+	}
+
+	private startVolumeCleanup() {
+		if (this.volumeCleanupInterval) {
+			return;
+		}
+
+		this.runVolumeCleanup();
+		this.volumeCleanupInterval = setInterval(() => this.runVolumeCleanup(), VOLUME_CLEANUP_INTERVAL_MS);
+	}
 
 	private scheduleReconnect() {
 		if (this.reconnectTimeout) {
@@ -22,6 +40,8 @@ export class Agent {
 	}
 
 	connect() {
+		this.startVolumeCleanup();
+
 		if (this.reconnectTimeout) {
 			clearTimeout(this.reconnectTimeout);
 			this.reconnectTimeout = null;

apps/agent/src/volume-host/__tests__/cleanup.test.ts

@@ -0,0 +1,49 @@
+import * as fs from "node:fs/promises";
+import * as os from "node:os";
+import * as path from "node:path";
+import { afterEach, expect, test, vi } from "vitest";
+
+let tempRoot: string | undefined;
+let mockMountPoints: string[] = [];
+
+afterEach(async () => {
+	vi.doUnmock("./constants");
+	vi.doUnmock("./fs");
+	vi.resetModules();
+	mockMountPoints = [];
+	if (tempRoot) {
+		await fs.rm(tempRoot, { recursive: true, force: true });
+		tempRoot = undefined;
+	}
+});
+
+const loadCleanup = async () => {
+	vi.resetModules();
+	tempRoot = await fs.mkdtemp(path.join(os.tmpdir(), "zerobyte-agent-cleanup-"));
+	vi.doMock("./constants", () => ({ VOLUME_MOUNT_BASE: tempRoot }));
+	vi.doMock("./fs", () => ({
+		readMountInfo: async () => mockMountPoints.map((mountPoint) => ({ mountPoint, fstype: "fuse.sshfs" })),
+	}));
+
+	return import("../cleanup");
+};
+
+test("removes stale volume directories that are not mounted on the agent host", async () => {
+	const { cleanupDanglingVolumeMountDirectories } = await loadCleanup();
+	await fs.mkdir(path.join(tempRoot!, "stale-volume", "_data"), { recursive: true });
+
+	await cleanupDanglingVolumeMountDirectories();
+
+	await expect(fs.access(path.join(tempRoot!, "stale-volume"))).rejects.toThrow();
+});
+
+test("keeps volume directories that are still mounted on the agent host", async () => {
+	const { cleanupDanglingVolumeMountDirectories } = await loadCleanup();
+	const localVolumeDir = path.join(tempRoot!, "mounted-volume");
+	mockMountPoints = [path.join(localVolumeDir, "_data")];
+	await fs.mkdir(path.join(localVolumeDir, "_data"), { recursive: true });
+
+	await cleanupDanglingVolumeMountDirectories();
+
+	await expect(fs.access(localVolumeDir)).resolves.toBeNull();
+});

apps/agent/src/volume-host/__tests__/operations.test.ts

@@ -0,0 +1,66 @@
+import * as fs from "node:fs/promises";
+import * as os from "node:os";
+import * as path from "node:path";
+import { afterEach, expect, test } from "vitest";
+import { listVolumeFiles } from "../operations";
+import type { AgentVolume } from "../types";
+
+let tempRoot: string | undefined;
+
+afterEach(async () => {
+	if (tempRoot) {
+		await fs.rm(tempRoot, { recursive: true, force: true });
+		tempRoot = undefined;
+	}
+});
+
+const createDirectoryVolume = async (): Promise<AgentVolume> => {
+	tempRoot = await fs.mkdtemp(path.join(os.tmpdir(), "zerobyte-volume-ops-"));
+	return {
+		id: 1,
+		shortId: "volume-1",
+		name: "Test volume",
+		config: { backend: "directory", path: tempRoot },
+		createdAt: Date.now(),
+		updatedAt: Date.now(),
+		lastHealthCheck: Date.now(),
+		type: "directory",
+		status: "mounted",
+		lastError: null,
+		provisioningId: null,
+		autoRemount: true,
+		agentId: "local",
+		organizationId: "org-1",
+	};
+};
+
+test("listVolumeFiles returns sorted paginated entries inside the volume", async () => {
+	const volume = await createDirectoryVolume();
+	await fs.mkdir(path.join(tempRoot!, "z-dir"));
+	await fs.mkdir(path.join(tempRoot!, "a-dir"));
+	await fs.writeFile(path.join(tempRoot!, "b-file.txt"), "hello");
+
+	const result = await listVolumeFiles(volume, undefined, 1, 2);
+
+	expect(result).toMatchObject({
+		path: "/",
+		offset: 1,
+		limit: 2,
+		total: 3,
+		hasMore: false,
+	});
+	expect(result.files.map((entry) => entry.name)).toEqual(["z-dir", "b-file.txt"]);
+	expect(result.files[1]).toMatchObject({ path: "/b-file.txt", type: "file", size: 5 });
+});
+
+test("listVolumeFiles rejects traversal outside the volume", async () => {
+	const volume = await createDirectoryVolume();
+
+	await expect(listVolumeFiles(volume, "../outside", 0, 10)).rejects.toThrow("Invalid path");
+});
+
+test("listVolumeFiles reports missing directories consistently", async () => {
+	const volume = await createDirectoryVolume();
+
+	await expect(listVolumeFiles(volume, "missing", 0, 10)).rejects.toThrow("Directory not found");
+});

apps/agent/src/volume-host/cleanup.ts

@@ -0,0 +1,28 @@
+import * as fs from "node:fs/promises";
+import * as path from "node:path";
+import { logger } from "@zerobyte/core/node";
+import { toMessage } from "@zerobyte/core/utils";
+import { VOLUME_MOUNT_BASE } from "./constants";
+import { readMountInfo } from "./fs";
+
+export const cleanupDanglingVolumeMountDirectories = async () => {
+	const mounts = await readMountInfo().catch((error) => {
+		logger.warn(`Failed to read mount info for volume cleanup: ${toMessage(error)}`);
+		return [];
+	});
+	const mountedPaths = new Set(mounts.map((mount) => mount.mountPoint));
+	const volumeDirs = await fs.readdir(VOLUME_MOUNT_BASE).catch(() => []);
+
+	for (const dir of volumeDirs) {
+		const mountPath = path.join(VOLUME_MOUNT_BASE, dir, "_data");
+		if (mountedPaths.has(mountPath)) {
+			continue;
+		}
+
+		const fullPath = path.join(VOLUME_MOUNT_BASE, dir);
+		logger.info(`Removing stale volume mount directory at ${fullPath}`);
+		await fs.rm(fullPath, { recursive: true, force: true }).catch((error) => {
+			logger.warn(`Failed to remove stale volume mount directory ${fullPath}: ${toMessage(error)}`);
+		});
+	}
+};