diff --git a/README.md b/README.md
index bf03ccd..0c60c3b 100644
--- a/README.md
+++ b/README.md
@@ -52,8 +52,42 @@ cpu          Detect and optionally apply a migration-safe CPU model
 maintenance  Toggle local HA maintenance mode
 keepalived   Deploy Keepalived across the cluster
 iso-nfs      Create an LXC NFS server and cluster-wide shared ISO storage
+orphan-disks Report cluster-wide orphan candidates and unused disks (read-only)
 ```
 
+## Orphaned and unused disks
+
+Open **Cluster & Maintenance → Find orphaned / unused disks (read-only)**.
+The menu offers CSV export after displaying the report. Direct commands:
+
+```bash
+tapm orphan-disks
+tapm orphan-disks --csv /root/orphan-disks.csv
+```
+
+Run as root on a Proxmox node. The report uses `pvesh` and Python 3, inventories
+online nodes' storage-managed VM and container volumes, and compares them with
+cluster guest configurations in `/etc/pve/nodes`, including offline guests,
+snapshots, pending changes, EFI/TPM disks, and detached `unusedN` entries.
+Shared storage IDs are scanned once; local storage remains scoped to its node.
+Results are grouped by node and storage, with volume IDs and provisioned sizes.
+Backups, ISOs, unmanaged files, and physical/passthrough disks are excluded.
+
+An **Orphan candidate** has no configuration reference. **Unused disk** means
+only current `unusedN` entries reference it. Remote-node references and possible
+template/clone backing relationships are marked **Needs review**. This is an
+inventory, not proof that a volume can be deleted: replication copies, storage
+aliases, and backend-specific backing dependencies require review. No guest or
+storage changes are made.
+
+Offline nodes, inactive storage, and failed reads are reported as incomplete
+coverage; missing guest configurations prevent orphan classification. Changes
+to guest configurations during the scan require a rerun. Scan when migration,
+replication, and disk operations are idle because the inventory is not atomic.
+CSV export includes coverage warnings and refuses to overwrite an existing file.
+Exit status is 0 for a completed scan, 2 for incomplete coverage or changed
+configurations, and 1 for a fatal scan/export error.
+
 ## Companion files
 
 Large installer artifacts used by this project are stored in the private
diff --git a/inc/orphan-disks.py b/inc/orphan-disks.py
new file mode 100644
index 0000000..3b0eaab
--- /dev/null
+++ b/inc/orphan-disks.py
@@ -0,0 +1,198 @@
+#!/usr/bin/env python3
+"""Read-only Proxmox storage/reference inventory; no third-party dependencies."""
+import argparse
+import csv
+import json
+from pathlib import Path
+import re
+import subprocess
+import sys
+
+VOLUME = re.compile(r"(?<![\w.-])([\w.-]+:[^\s,;\"']+)")
+FIELDS = ['node', 'storage', 'volume', 'size_bytes', 'status', 'detail']
+
+
+def api(path, *options):
+    result = subprocess.run(['pvesh', 'get', path, *options, '--output-format', 'json'],
+                            capture_output=True, text=True, timeout=90)
+    if result.returncode:
+        raise RuntimeError(result.stderr.strip() or 'pvesh failed')
+    data = json.loads(result.stdout)
+    if not isinstance(data, list):
+        raise ValueError('expected an API array')
+    return data
+
+
+def enabled(value):
+    return value in (1, True, '1')
+
+
+def references(root, nodes, guests):
+    """Read pmxcfs on this host, including offline guests and all config sections."""
+    refs, errors, seen, contents = {}, [], set(), {}
+    for node in nodes:
+        for kind, directory in [('qemu', 'qemu-server'), ('lxc', 'lxc')]:
+            folder = root / node['node'] / directory
+            try:
+                paths = list(folder.iterdir())
+            except OSError as exc:
+                errors.append(f'Cannot read {folder}: {exc}')
+                continue
+            for path in paths:
+                if path.suffix != '.conf':
+                    continue
+                try:
+                    content = path.read_text()
+                except OSError as exc:
+                    errors.append(f'Cannot read {path}: {exc}')
+                    continue
+                contents[str(path)] = content
+                seen.add((node['node'], kind, path.stem))
+                section = 'current'
+                for line in content.splitlines():
+                    line = line.strip()
+                    if not line or line.startswith('#'):
+                        continue
+                    if line.startswith('['):
+                        section = line
+                        continue
+                    key, sep, value = line.partition(':')
+                    if not sep or key in ('description', 'name'):
+                        continue
+                    for volume in VOLUME.findall(value):
+                        refs.setdefault(volume, []).append({
+                            'node': node['node'], 'guest': f'{kind} {path.stem}',
+                            'unused': bool(re.fullmatch(r'unused\d+', key)) and section == 'current',
+                            'section': section,
+                        })
+    for guest in guests:
+        identity = (guest['node'], guest['type'], str(guest['vmid']))
+        if identity not in seen:
+            errors.append(f'Missing guest configuration: {identity}')
+    return refs, errors, contents
+
+
+def classify(volume, node, shared, refs, incomplete):
+    all_refs = refs.get(volume['volid'], [])
+    local_refs = [r for r in all_refs if shared or r['node'] == node]
+    if any(not r['unused'] for r in local_refs):
+        return None
+    if incomplete:
+        return 'Needs review', 'Guest reference inventory incomplete'
+    if local_refs:
+        return 'Unused disk', ', '.join(sorted({r['guest'] for r in local_refs}))
+    if all_refs:
+        return 'Needs review', 'Referenced on another node; possible replication copy: ' + ', '.join(
+            sorted({r['node'] + '/' + r['guest'] for r in all_refs}))
+    if re.search(r'(^|[/ :])base[-_]', volume['volid']) or volume.get('parent'):
+        return 'Needs review', 'Possible template/clone backing relationship'
+    return 'Orphan candidate', 'No guest configuration reference; verify replication and backing dependencies'
+
+
+def scan(root=Path('/etc/pve/nodes'), get=api):
+    warnings, rows, coverage = [], [], []
+    nodes = get('/nodes')
+    if not nodes:
+        raise RuntimeError('No nodes returned; cannot inventory cluster')
+    guests = get('/cluster/resources', '--type', 'vm')
+    refs, errors, before = references(root, nodes, guests)
+    warnings.extend(errors)
+    shared_done = set()
+    for node in sorted(nodes, key=lambda item: item['node']):
+        name = node['node']
+        if node.get('status') != 'online':
+            warnings.append(f'{name}: offline; storage not scanned')
+            continue
+        print(f'Scanning {name} storage...', file=sys.stderr, flush=True)
+        try:
+            storages = get(f'/nodes/{name}/storage')
+        except (RuntimeError, ValueError, OSError, subprocess.TimeoutExpired) as exc:
+            warnings.append(f'{name}: storage discovery failed: {exc}')
+            continue
+        for storage in sorted(storages, key=lambda item: item['storage']):
+            sid = storage['storage']
+            if not {'images', 'rootdir'} & set(storage.get('content', '').split(',')):
+                continue
+            shared = enabled(storage.get('shared', 0))
+            if shared and sid in shared_done:
+                continue
+            if not enabled(storage.get('active', 0)) or not enabled(storage.get('enabled', 1)):
+                warnings.append(f'{name}/{sid}: inactive or disabled; not scanned')
+                continue
+            try:
+                volumes = get(f'/nodes/{name}/storage/{sid}/content')
+            except (RuntimeError, ValueError, OSError, subprocess.TimeoutExpired) as exc:
+                warnings.append(f'{name}/{sid}: content listing failed: {exc}')
+                continue
+            if shared:
+                shared_done.add(sid)
+            label = 'shared (via ' + name + ')' if shared else name
+            disks = [v for v in volumes if v.get('content') in ('images', 'rootdir')]
+            coverage.append(f'{label}/{sid}: {len(disks)} disk volume(s) scanned')
+            for volume in disks:
+                result = classify(volume, name, shared, refs, bool(errors))
+                if result:
+                    rows.append(dict(zip(FIELDS, [label, sid, volume['volid'], volume.get('size', 0), *result])))
+    # Detect migrations/config changes during this non-atomic inventory.
+    _, after_errors, after = references(root, nodes, guests)
+    if after != before or after_errors:
+        warnings.append('Guest configurations changed or became unreadable during scan; rerun report')
+        for row in rows:
+            row.update(status='Needs review', detail='Guest configurations changed during scan')
+    return rows, warnings, coverage
+
+
+def render(rows, warnings, coverage):
+    print('\nCluster orphaned / unused disk report (read-only)')
+    print('Coverage: ' + ('INCOMPLETE / NEEDS REVIEW' if warnings else 'all discovered disk storage scanned'))
+    print('Candidates are not proof of safe deletion. Storage-managed volumes only; excludes backups, ISOs and passthrough disks.')
+    for entry in coverage:
+        print('  ' + entry)
+    for warning in warnings:
+        print('WARNING: ' + warning)
+    group = None
+    for row in sorted(rows, key=lambda r: (r['node'], r['storage'], r['volume'])):
+        if group != (row['node'], row['storage']):
+            group = (row['node'], row['storage'])
+            print(f'\n{group[0]} / {group[1]}')
+        size = float(row['size_bytes'] or 0) / (1024 ** 3)
+        print(f"  {row['volume']}  {size:.2f} GiB  {row['status']}\n    {row['detail']}")
+    print(f'\n{len(rows)} volume(s) reported.')
+    if not rows:
+        print('No unreferenced or unused volumes found in the successfully scanned storage.')
+
+
+def export_csv(path, rows, warnings):
+    # Exclusive creation avoids overwriting existing files, including symlink targets.
+    with open(path, 'x', newline='') as handle:
+        writer = csv.DictWriter(handle, fieldnames=FIELDS + ['coverage_warnings'])
+        writer.writeheader()
+        for row in rows:
+            writer.writerow(dict(row, coverage_warnings='; '.join(warnings)))
+        if warnings and not rows:
+            writer.writerow({'status': 'Incomplete scan', 'coverage_warnings': '; '.join(warnings)})
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--csv', metavar='PATH', help='export to a new CSV file')
+    parser.add_argument('--interactive', action='store_true', help='offer CSV export after the report')
+    # The TAPM preloader pads its forwarded arguments with empty strings.
+    args = parser.parse_args([arg for arg in sys.argv[1:] if arg])
+    try:
+        rows, warnings, coverage = scan()
+        render(rows, warnings, coverage)
+        destination = args.csv
+        if args.interactive and sys.stdin.isatty() and not destination:
+            destination = input('\nCSV export path (Enter to skip): ').strip()
+        if destination:
+            export_csv(destination, rows, warnings)
+            print('CSV saved: ' + destination)
+        return 2 if warnings else 0
+    except (OSError, RuntimeError, ValueError, subprocess.TimeoutExpired) as exc:
+        print(f'Disk report failed: {exc}', file=sys.stderr)
+        return 1
+
+
+if __name__ == '__main__':
+    sys.exit(main())
diff --git a/proxmenu-scripts.sh b/proxmenu-scripts.sh
index d204454..39e01f8 100755
--- a/proxmenu-scripts.sh
+++ b/proxmenu-scripts.sh
@@ -2443,9 +2443,25 @@ MONITORING_MENU() {
 	done
 }
 
+TAPM_ORPHAN_DISK_REPORT() {
+	local result
+	if ! command -v python3 >/dev/null || ! command -v pvesh >/dev/null; then
+		printf 'Disk reporting requires python3 and pvesh on a Proxmox host.\n'
+		FINISH_FAILED_ACTION
+		return 1
+	fi
+	python3 /opt/idssys/ta-proxmenu/inc/orphan-disks.py "$@"
+	result=$?
+	if (( ACTION_REQUESTED == 1 )); then
+		exit "$result"
+	fi
+	ENTER2CONTINUE
+	return "$result"
+}
+
 CLUSTER_MENU() {
 	local -a labels
-	local -a values=("maintenance" "services" "keepalived" "iso_nfs")
+	local -a values=("maintenance" "services" "keepalived" "iso_nfs" "orphan_disks")
 	local maintenance_status
 	local node
 
@@ -2470,12 +2486,14 @@ CLUSTER_MENU() {
 			labels+=("Deploy/reconfigure Keepalived (installed locally)") ||
 			labels+=("Deploy Keepalived on all cluster hosts")
 		labels+=("Create shared ISO storage using an LXC NFS server")
+		labels+=("Find orphaned / unused disks (read-only)")
 
 		SELECT_MENU "Cluster & Maintenance" labels values
 		case "$MENU_SELECTION" in
 			maintenance) MAINTENANCE_MODE;;
 			services) SERVICE_RECOVERY_MENU;;
 			keepalived) INSTALL_KEEPALIVE;;
+			orphan_disks) TAPM_ORPHAN_DISK_REPORT --interactive;;
 			iso_nfs)
 				TAPM_DEPLOY_ISO_NFS_LXC
 				FINISH_ACTION
@@ -2763,6 +2781,10 @@ if (( ACTION_REQUESTED == 1 )); then
 		maintenance|mm) MAINTENANCE_MODE;;
 		keepalived) INSTALL_KEEPALIVE;;
 		iso-nfs|iso_nfs) TAPM_DEPLOY_ISO_NFS_LXC;;
+		orphan-disks|orphan_disks)
+			shift
+			TAPM_ORPHAN_DISK_REPORT "${@}"
+			;;
     *) MAIN_MENU;;
 	esac
 else
diff --git a/tests/test-orphan-disks.py b/tests/test-orphan-disks.py
new file mode 100644
index 0000000..ff4d72c
--- /dev/null
+++ b/tests/test-orphan-disks.py
@@ -0,0 +1,121 @@
+#!/usr/bin/env python3
+import contextlib
+import importlib.util
+import io
+from pathlib import Path
+import tempfile
+import unittest
+
+spec = importlib.util.spec_from_file_location('disks', Path(__file__).resolve().parents[1] / 'inc/orphan-disks.py')
+disks = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(disks)
+
+
+class ReportTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.root = Path(self.temp.name)
+        for node in ('a', 'b'):
+            for kind in ('qemu-server', 'lxc'):
+                (self.root / node / kind).mkdir(parents=True)
+        (self.root / 'a/qemu-server/100.conf').write_text(
+            'scsi0: local:vm-100-disk-0,size=10G\n'
+            'unused0: local:vm-100-disk-1\n'
+            'efidisk0: shared:vm-100-efi\n'
+            'tpmstate0: shared:vm-100-tpm\n'
+            '[snap]\nscsi0: local:vm-100-old\n'
+            '[PENDING]\nscsi1: local:vm-100-pending\n')
+        (self.root / 'b/lxc/200.conf').write_text('rootfs: local:subvol-200-disk-0\nmp0: shared:subvol-200-disk-1,mp=/data\n')
+        self.nodes = [{'node': n, 'status': 'online'} for n in ('a', 'b')]
+        self.guests = [{'node': 'a', 'type': 'qemu', 'vmid': 100}, {'node': 'b', 'type': 'lxc', 'vmid': 200}]
+        self.refs, self.errors, _ = disks.references(self.root, self.nodes, self.guests)
+
+    def result(self, vol, node='a', shared=False, incomplete=False):
+        return disks.classify({'volid': vol}, node, shared, self.refs, incomplete)
+
+    def test_current_snapshots_pending_and_containers_are_protected(self):
+        self.assertEqual(self.errors, [])
+        for vol in ('vm-100-disk-0', 'vm-100-old', 'vm-100-pending'):
+            self.assertIsNone(self.result('local:' + vol))
+        for vol in ('vm-100-efi', 'vm-100-tpm', 'subvol-200-disk-1'):
+            self.assertIsNone(self.result('shared:' + vol, shared=True))
+        self.assertIsNone(self.result('local:subvol-200-disk-0', node='b'))
+
+    def test_candidates_unused_replication_and_backing(self):
+        self.assertEqual(self.result('local:vm-999-disk-0')[0], 'Orphan candidate')
+        self.assertEqual(self.result('local:vm-100-disk-1')[0], 'Unused disk')
+        self.assertEqual(self.result('local:vm-100-disk-0', node='b')[0], 'Needs review')
+        self.assertEqual(self.result('local:base-900-disk-0')[0], 'Needs review')
+        self.assertEqual(self.result('local:999/base-999-disk-0.qcow2')[0], 'Needs review')
+        self.assertEqual(self.result('local:vm-999-disk-0', incomplete=True)[0], 'Needs review')
+
+    def test_missing_config_is_incomplete(self):
+        (self.root / 'a/qemu-server/100.conf').unlink()
+        _, errors, _ = disks.references(self.root, self.nodes, self.guests)
+        self.assertTrue(errors)
+
+    def test_scan_deduplicates_shared_but_not_local_and_reports_failures(self):
+        calls = []
+        def get(path, *options):
+            calls.append(path)
+            if path == '/nodes':
+                return self.nodes
+            if path == '/cluster/resources':
+                return self.guests
+            if path.endswith('/storage'):
+                return [{'storage': sid, 'shared': shared, 'active': 1, 'content': 'images,rootdir'}
+                        for sid, shared in [('shared', 1), ('local', 0), ('broken', 0)]]
+            if '/broken/' in path:
+                raise RuntimeError('unavailable')
+            sid = path.split('/')[-2]
+            return [{'volid': sid + ':vm-999-disk-0', 'content': 'images', 'size': 1024},
+                    {'volid': sid + ':iso/test.iso', 'content': 'iso'}]
+        with contextlib.redirect_stderr(io.StringIO()):
+            rows, warnings, coverage = disks.scan(self.root, get)
+        self.assertEqual(len(rows), 3)
+        self.assertEqual(len(warnings), 2)
+        self.assertEqual(len(coverage), 3)
+        self.assertNotIn('/nodes/b/storage/shared/content', calls)
+
+    def test_csv_preserves_warnings_and_refuses_overwrite(self):
+        path = self.root / 'report.csv'
+        disks.export_csv(path, [], ['node offline'])
+        self.assertIn('node offline', path.read_text())
+        with self.assertRaises(FileExistsError):
+            disks.export_csv(path, [], [])
+
+    def test_offline_node_and_config_change(self):
+        def get(path, *options):
+            if path == '/nodes':
+                return [self.nodes[0], {'node': 'b', 'status': 'offline'}]
+            if path == '/cluster/resources':
+                return self.guests
+            if path.endswith('/storage'):
+                return [{'storage': 'local', 'content': 'images', 'active': 1}]
+            (self.root / 'a/qemu-server/100.conf').write_text('scsi0: local:vm-999-disk-0\n')
+            return [{'volid': 'local:vm-999-disk-0', 'content': 'images'}]
+        with contextlib.redirect_stderr(io.StringIO()):
+            rows, warnings, _ = disks.scan(self.root, get)
+        self.assertEqual(rows[0]['status'], 'Needs review')
+        self.assertTrue(any('offline' in warning for warning in warnings))
+        self.assertTrue(any('changed' in warning for warning in warnings))
+
+    def test_launcher_padded_arguments(self):
+        from unittest.mock import patch
+        with patch.object(disks.sys, 'argv', ['orphan-disks.py', '', '', '']), \
+             patch.object(disks, 'scan', return_value=([], [], [])), \
+             contextlib.redirect_stdout(io.StringIO()):
+            self.assertEqual(disks.main(), 0)
+
+    def test_api_is_read_only(self):
+        from unittest.mock import patch
+        with patch.object(disks.subprocess, 'run') as run:
+            run.return_value.returncode = 0
+            run.return_value.stdout = '[]'
+            disks.api('/nodes')
+            self.assertEqual(run.call_args.args[0], ['pvesh', 'get', '/nodes', '--output-format', 'json'])
+
+
+if __name__ == '__main__':
+    unittest.main()