diff --git a/src/dlstbx/services/archiver.py b/src/dlstbx/services/archiver.py
index 2f93e9c2a..cc989cfbc 100644
--- a/src/dlstbx/services/archiver.py
+++ b/src/dlstbx/services/archiver.py
@@ -6,10 +6,13 @@
 import os.path
 import xml.etree.ElementTree as ET
 from datetime import datetime
+from pathlib import Path
 
 import workflows.recipe
 from workflows.services.common_service import CommonService
 
+from dlstbx.util import INDUSTRIAL_CODES
+
 
 class Dropfile:
     """A class encapsulating the XML dropfile tree as it is built up."""
@@ -132,6 +135,16 @@ def rangifier(numbers):
             group = list(group)
             yield group[0][1], group[-1][1]
 
+    def visit_is_archivable(
+        self, visit: str, allowed_industrial_visit_codes: tuple[str, ...]
+    ) -> bool:
+        """Return false if visit has an industrial code that is not in the list of allowed industrial visit codes."""
+        if visit.startswith(tuple(INDUSTRIAL_CODES)) and not visit.startswith(
+            allowed_industrial_visit_codes
+        ):
+            return False
+        return True
+
     def archive_dcid(self, rw, header, message):
         """Archive collected datafiles connected to a data collection."""
 
@@ -141,6 +154,21 @@ def archive_dcid(self, rw, header, message):
 
         # Extract parameters from the recipe
         params = rw.recipe_step["parameters"]
+
+        allowed_industrial_visit_codes = tuple(
+            self.config.storage.get(
+                "zocalo.archiver.allowed_industrial_visit_codes", []
+            )
+        )
+        if not self.visit_is_archivable(
+            params["visit"], allowed_industrial_visit_codes
+        ):
+            self.log.info(
+                f"Skipping archiving of {params['pattern']} because it is from a forbidden visit"
+            )
+            self._transport.transaction_commit(txn)
+            return
+
         self.log.info("Attempting to archive %s", params["pattern"])
 
         settings = params.copy()
@@ -151,10 +179,12 @@ def archive_dcid(self, rw, header, message):
 
         file_range_limit = int(settings.get("limit-files", 0))
 
-        filepaths = params["pattern"].split("/")
-        _, _, beamline, _, _, visit_id = filepaths[0:6]
+        filepath = Path(params["pattern"])
+        dataset_name = Path(*filepath.parts[6:-1]).as_posix() or "topdir"
+        beamline = params["beamline"]
+        visit_id = params["visit"]
 
-        df = Dropfile(visit_id.upper(), beamline, "/".join(filepaths[6:-1]) or "topdir")
+        df = Dropfile(visit_id.upper(), beamline, dataset_name)
 
         message_out = {"success": 0, "failed": 0}
         files_not_found = []
@@ -287,26 +317,28 @@ def archive_filelist(self, rw, header, message):
         )
         file_range_limit = int(params.get("limit-files", 0))
 
-        filepaths = filelist[0].split("/")
-        beamline = "unknown"
-        visit_id = "unknown"
-        try:
-            if filepaths[1:3] == ["dls", "mx"]:
-                beamline = "i02-2"  # VMXi currently only beamline with new visit path structure
-            else:
-                beamline = filepaths[2]
-            visit_id = filepaths[5]
-        except IndexError:
-            pass
-        visit_id = params.get("visit", visit_id)
-        beamline = params.get("beamline", beamline)
-
         # Conditionally acknowledge receipt of the message
         txn = self._transport.transaction_begin(subscription_id=header["subscription"])
         self._transport.ack(header, transaction=txn)
 
+        visit_id = params["visit"]
+        beamline = params["beamline"]
+        allowed_industrial_visit_codes = tuple(
+            self.config.storage.get(
+                "zocalo.archiver.allowed_industrial_visit_codes", []
+            )
+        )
+        if not self.visit_is_archivable(visit_id, allowed_industrial_visit_codes):
+            self.log.info(
+                f"Skipping archiving of {filelist} because it is from a forbidden visit"
+            )
+            self._transport.transaction_commit(txn)
+            return
+
+        filepath = Path(filelist[0])
+        dataset_name = Path(*filepath.parts[6:-1]).as_posix() or "topdir"
         # Archive files
-        df = Dropfile(visit_id.upper(), beamline, "/".join(filepaths[6:-1]) or "topdir")
+        df = Dropfile(visit_id.upper(), beamline, dataset_name)
 
         message_out = {"success": 0, "failed": 0}
         files_not_found = []
diff --git a/tests/services/test_archiver.py b/tests/services/test_archiver.py
new file mode 100644
index 000000000..e04d53dad
--- /dev/null
+++ b/tests/services/test_archiver.py
@@ -0,0 +1,519 @@
+from __future__ import annotations
+
+from unittest import mock
+
+import pytest
+from workflows.recipe.wrapper import RecipeWrapper
+from workflows.transport.offline_transport import OfflineTransport
+
+from dlstbx.services.archiver import DLSArchiver, Dropfile
+
+
+@pytest.fixture
+def mock_transport():
+    """Create a mock transport object."""
+    transport = mock.Mock()
+    txn = mock.Mock()
+    transport.transaction_begin.return_value = txn
+    return transport, txn
+
+
+@pytest.fixture
+def archiver_service(mock_transport):
+    """Create a DLSArchiver service instance with mocked transport and config."""
+    transport, _ = mock_transport
+    service = DLSArchiver()
+    service._transport = transport
+    service._environment = {
+        "config": mock.Mock(
+            storage={"zocalo.archiver.allowed_industrial_visit_codes": ["sw"]},
+        )
+    }
+    return service
+
+
+def create_test_files(tmp_path, file_list):
+    """Helper to create test files in tmp_path."""
+    files = []
+    for filename in file_list:
+        filepath = tmp_path / filename
+        filepath.parent.mkdir(parents=True, exist_ok=True)
+        filepath.write_text(f"Content of {filename}")
+        files.append(str(filepath))
+    return files
+
+
+def create_recipe_message_filelist(parameters, files):
+    """Create a recipe message for filelist archiving."""
+    message = {
+        "recipe": {
+            "1": {
+                "service": "DLS Archiver",
+                "queue": "archive.filelist",
+                "parameters": parameters,
+            },
+        },
+        "recipe-pointer": "1",
+        "recipe-path": [],
+        "environment": {
+            "ID": mock.sentinel.GUID,
+            "source": mock.sentinel.source,
+            "timestamp": mock.sentinel.timestamp,
+        },
+        "payload": {"filelist": files},
+    }
+    return message
+
+
+def create_recipe_message_dcid(parameters):
+    """Create a recipe message for dcid archiving."""
+    message = {
+        "recipe": {
+            "1": {
+                "service": "DLS Archiver",
+                "queue": "archive.pattern",
+                "parameters": parameters,
+            },
+        },
+        "recipe-pointer": "1",
+        "recipe-path": [],
+        "environment": {
+            "ID": mock.sentinel.GUID,
+            "source": mock.sentinel.source,
+            "timestamp": mock.sentinel.timestamp,
+        },
+        "payload": {},
+    }
+    return message
+
+
+class TestArchiverFilelist:
+    """Tests for the archive_filelist method."""
+
+    def test_archive_filelist_creates_dropfile(
+        self, archiver_service, mock_transport, tmp_path
+    ):
+        """Test that a dropfile is created when valid files are provided."""
+        transport, txn = mock_transport
+        archiver_service._transport = transport
+
+        # Create test files
+        files = create_test_files(tmp_path, ["file1.txt", "file2.txt"])
+
+        # Create recipe parameters
+        dropfile_path = tmp_path / "dropfile.xml"
+        parameters = {
+            "visit": "cm00001-1",
+            "beamline": "i03",
+            "dropfile": str(dropfile_path),
+            "filelist": files,
+        }
+
+        # Create message
+        message = create_recipe_message_filelist(parameters, files)
+        header = {
+            "message-id": "test-message-id",
+            "subscription": "test-subscription",
+        }
+
+        # Create wrapper and call method
+        rw = RecipeWrapper(message=message, transport=OfflineTransport())
+        rw._transport = OfflineTransport()
+        archiver_service.archive_filelist(rw, header, message.get("payload"))
+
+        # Verify dropfile was created
+        assert dropfile_path.exists(), "Dropfile should be created"
+
+        # Verify content is XML
+        content = dropfile_path.read_text()
+        assert "<?xml version" in content
+        assert "CM00001" in content  # Visit codes are converted to uppercase
+        assert "i03" in content
+
+    def test_archive_filelist_with_forbidden_visit_code(
+        self, archiver_service, mock_transport, tmp_path
+    ):
+        """Test that no dropfile is created for forbidden visit codes."""
+        transport, txn = mock_transport
+        archiver_service._transport = transport
+
+        # Create test files
+        files = create_test_files(tmp_path, ["file1.txt"])
+
+        # Create recipe parameters with forbidden industrial visit code
+        dropfile_path = tmp_path / "dropfile.xml"
+        parameters = {
+            "visit": "in12345-1",
+            "beamline": "i03",
+            "dropfile": str(dropfile_path),
+            "filelist": files,
+        }
+
+        # Create message
+        message = create_recipe_message_filelist(parameters, files)
+        header = {
+            "message-id": "test-message-id",
+            "subscription": "test-subscription",
+        }
+
+        # Create wrapper and call method
+        rw = RecipeWrapper(message=message, transport=OfflineTransport())
+        rw._transport = OfflineTransport()
+        archiver_service.archive_filelist(rw, header, message.get("payload"))
+
+        # Verify no dropfile was created
+        assert not dropfile_path.exists(), (
+            "Dropfile should not be created for forbidden visit"
+        )
+
+        # Verify transaction was committed
+        transport.transaction_commit.assert_called_with(txn)
+
+    def test_archive_filelist_with_allowed_industrial_visit_code(
+        self, archiver_service, mock_transport, tmp_path
+    ):
+        """Test that a dropfile is created when an industrial visit is explicitly allowed."""
+        transport, txn = mock_transport
+        archiver_service._transport = transport
+
+        # Create test files
+        files = create_test_files(tmp_path, ["file1.txt", "file2.txt"])
+
+        # Create recipe parameters with allowed industrial visit code
+        dropfile_path = tmp_path / "dropfile.xml"
+        parameters = {
+            "visit": "sw12345-1",
+            "beamline": "i03",
+            "dropfile": str(dropfile_path),
+            "filelist": files,
+        }
+
+        # Create message
+        message = create_recipe_message_filelist(parameters, files)
+        header = {
+            "message-id": "test-message-id",
+            "subscription": "test-subscription",
+        }
+
+        # Create wrapper and call method
+        rw = RecipeWrapper(message=message, transport=OfflineTransport())
+        rw._transport = OfflineTransport()
+        archiver_service.archive_filelist(rw, header, message.get("payload"))
+
+        # Verify dropfile was created
+        assert dropfile_path.exists(), (
+            "Dropfile should be created for allowed industrial visit"
+        )
+
+        # Verify transaction was committed
+        transport.transaction_commit.assert_called_with(txn)
+
+    def test_archive_filelist_empty_files_list(self, archiver_service, mock_transport):
+        """Test that empty file list is rejected."""
+        transport, txn = mock_transport
+        archiver_service._transport = transport
+
+        # Create recipe parameters with empty file list
+        parameters = {
+            "visit": "cm00001-1",
+            "beamline": "i03",
+            "filelist": [],
+        }
+
+        # Create message with empty filelist
+        message = create_recipe_message_filelist(parameters, [])
+        header = {
+            "message-id": "test-message-id",
+            "subscription": "test-subscription",
+        }
+
+        # Create wrapper and call method
+        rw = RecipeWrapper(message=message, transport=OfflineTransport())
+        rw._transport = OfflineTransport()
+        archiver_service.archive_filelist(rw, header, message.get("payload"))
+
+        # Verify NACK was called
+        transport.nack.assert_called_once_with(header)
+
+    def test_archive_filelist_missing_files(
+        self, archiver_service, mock_transport, tmp_path
+    ):
+        """Test that missing files are handled correctly."""
+        transport, txn = mock_transport
+        archiver_service._transport = transport
+
+        # Create only one test file
+        files = create_test_files(tmp_path, ["file1.txt"])
+        # Add a missing file to the list
+        missing_file = str(tmp_path / "missing.txt")
+        files.append(missing_file)
+
+        # Create recipe parameters
+        dropfile_path = tmp_path / "dropfile.xml"
+        parameters = {
+            "visit": "cm00001-1",
+            "beamline": "i03",
+            "dropfile": str(dropfile_path),
+            "filelist": files,
+        }
+
+        # Create message
+        message = create_recipe_message_filelist(parameters, files)
+        header = {
+            "message-id": "test-message-id",
+            "subscription": "test-subscription",
+        }
+
+        # Create wrapper and call method
+        rw = RecipeWrapper(message=message, transport=OfflineTransport())
+        rw._transport = OfflineTransport()
+        archiver_service.archive_filelist(rw, header, message.get("payload"))
+
+        # Verify dropfile was still created
+        assert dropfile_path.exists(), (
+            "Dropfile should be created even with missing files"
+        )
+
+
+class TestArchiverDCID:
+    """Tests for the archive_dcid method."""
+
+    def test_archive_dcid_creates_dropfile(
+        self, archiver_service, mock_transport, tmp_path
+    ):
+        """Test that a dropfile is created for valid dcid archiving."""
+        transport, txn = mock_transport
+        archiver_service._transport = transport
+
+        # Create test files matching a pattern
+        image_dir = (
+            tmp_path / "data" / "cm" / "cm00001" / "i03" / "2024" / "Jan" / "cm00001-1"
+        )
+        image_dir.mkdir(parents=True, exist_ok=True)
+
+        # Create a few test image files
+        pattern = str(image_dir / "image_%06d.h5")
+        for i in range(1, 4):
+            filepath = image_dir / f"image_{i:06d}.h5"
+            filepath.write_text(f"Image data {i}")
+
+        # Create recipe parameters
+        dropfile_path = tmp_path / "dropfile.xml"
+        parameters = {
+            "visit": "cm00001-1",
+            "beamline": "i23",
+            "pattern": pattern,
+            "pattern-start": "1",
+            "pattern-end": "3",
+            "dropfile": str(dropfile_path),
+        }
+
+        # Create message
+        message = create_recipe_message_dcid(parameters)
+        header = {
+            "message-id": "test-message-id",
+            "subscription": "test-subscription",
+        }
+
+        # Create wrapper and call method
+        rw = RecipeWrapper(message=message, transport=OfflineTransport())
+        rw._transport = OfflineTransport()
+        archiver_service.archive_dcid(rw, header, message.get("payload"))
+
+        # Verify dropfile was created
+        assert dropfile_path.exists(), "Dropfile should be created"
+
+        # Verify content
+        content = dropfile_path.read_text()
+        assert "<?xml version" in content
+        assert "CM00001" in content
+        assert "i23" in content
+
+    def test_archive_dcid_with_forbidden_visit_code(
+        self, archiver_service, mock_transport, tmp_path
+    ):
+        """Test that no dropfile is created for forbidden visit codes in dcid."""
+        transport, txn = mock_transport
+        archiver_service._transport = transport
+
+        # Create test files
+        image_dir = (
+            tmp_path / "data" / "cm" / "TEST" / "i03" / "2024" / "Jan" / "TEST-1"
+        )
+        image_dir.mkdir(parents=True, exist_ok=True)
+        pattern = str(image_dir / "image_%06d.h5")
+        for i in range(1, 2):
+            filepath = image_dir / f"image_{i:06d}.h5"
+            filepath.write_text(f"Image data {i}")
+
+        # Create recipe parameters with forbidden industrial visit code
+        dropfile_path = tmp_path / "dropfile.xml"
+        parameters = {
+            "visit": "in12345-1",
+            "pattern": pattern,
+            "pattern-start": "1",
+            "pattern-end": "1",
+            "dropfile": str(dropfile_path),
+        }
+
+        # Create message
+        message = create_recipe_message_dcid(parameters)
+        header = {
+            "message-id": "test-message-id",
+            "subscription": "test-subscription",
+        }
+
+        # Create wrapper and call method
+        rw = RecipeWrapper(message=message, transport=OfflineTransport())
+        rw._transport = OfflineTransport()
+        archiver_service.archive_dcid(rw, header, message.get("payload"))
+
+        # Verify no dropfile was created
+        assert not dropfile_path.exists(), (
+            "Dropfile should not be created for forbidden visit"
+        )
+
+        # Verify transaction was committed
+        transport.transaction_commit.assert_called_with(txn)
+
+    def test_archive_dcid_with_allowed_industrial_visit_code(
+        self, archiver_service, mock_transport, tmp_path
+    ):
+        """Test that a dropfile is created when an industrial visit is explicitly allowed."""
+        transport, txn = mock_transport
+        archiver_service._transport = transport
+
+        # Create test files
+        image_dir = (
+            tmp_path / "data" / "cm" / "TEST" / "i03" / "2024" / "Jan" / "TEST-1"
+        )
+        image_dir.mkdir(parents=True, exist_ok=True)
+        pattern = str(image_dir / "image_%06d.h5")
+        for i in range(1, 2):
+            filepath = image_dir / f"image_{i:06d}.h5"
+            filepath.write_text(f"Image data {i}")
+
+        # Create recipe parameters with allowed industrial visit code
+        dropfile_path = tmp_path / "dropfile.xml"
+        parameters = {
+            "visit": "sw12345-1",
+            "beamline": "i03",
+            "pattern": pattern,
+            "pattern-start": "1",
+            "pattern-end": "1",
+            "dropfile": str(dropfile_path),
+        }
+
+        # Create message
+        message = create_recipe_message_dcid(parameters)
+        header = {
+            "message-id": "test-message-id",
+            "subscription": "test-subscription",
+        }
+
+        # Create wrapper and call method
+        rw = RecipeWrapper(message=message, transport=OfflineTransport())
+        rw._transport = OfflineTransport()
+        archiver_service.archive_dcid(rw, header, message.get("payload"))
+
+        # Verify dropfile was created
+        assert dropfile_path.exists(), (
+            "Dropfile should be created for allowed industrial visit"
+        )
+
+        # Verify transaction was committed
+        transport.transaction_commit.assert_called_with(txn)
+
+    def test_archive_dcid_no_files_found(
+        self, archiver_service, mock_transport, tmp_path
+    ):
+        """Test behavior when no files match the pattern."""
+        transport, txn = mock_transport
+        archiver_service._transport = transport
+
+        # Create a pattern that doesn't match any files
+        pattern = str(tmp_path / "nonexistent" / "image_%06d.h5")
+
+        # Create recipe parameters
+        dropfile_path = tmp_path / "dropfile.xml"
+        parameters = {
+            "visit": "cm00001-1",
+            "beamline": "i23",
+            "pattern": pattern,
+            "pattern-start": "1",
+            "pattern-end": "3",
+            "dropfile": str(dropfile_path),
+        }
+
+        # Create message
+        message = create_recipe_message_dcid(parameters)
+        header = {
+            "message-id": "test-message-id",
+            "subscription": "test-subscription",
+        }
+
+        # Create wrapper and call method
+        rw = RecipeWrapper(message=message, transport=OfflineTransport())
+        rw._transport = OfflineTransport()
+        archiver_service.archive_dcid(rw, header, message.get("payload"))
+
+        # Verify no dropfile was created (empty dropfile is skipped)
+        assert not dropfile_path.exists(), (
+            "Dropfile should not be created when no files found"
+        )
+
+
+class TestDropfileGeneration:
+    """Tests for dropfile XML generation."""
+
+    def test_dropfile_basic_structure(self, tmp_path):
+        """Test that dropfile creates valid XML structure."""
+        # Create test file
+        testfile = tmp_path / "testdata.h5"
+        testfile.write_text("test data")
+
+        # Create dropfile
+        df = Dropfile("cm00001-1", "i03", "testdir")
+        df.add(str(testfile))
+        df.close()
+
+        # Get XML string
+        xml_str = df.to_string().decode("latin-1")
+
+        # Verify structure
+        assert '<?xml version="1.0" ?>' in xml_str
+        assert "<icat" in xml_str
+        assert "<visit_id>CM00001-1</visit_id>" in xml_str
+        assert "<instrument>i03</instrument>" in xml_str
+        assert "<name>testdata.h5</name>" in xml_str
+        assert f"<location>{str(testfile)}</location>" in xml_str
+
+    def test_dropfile_multiple_files(self, tmp_path):
+        """Test dropfile with multiple files."""
+        # Create test files
+        testfiles = []
+        for i in range(3):
+            testfile = tmp_path / f"data_{i}.h5"
+            testfile.write_text(f"test data {i}")
+            testfiles.append(str(testfile))
+
+        # Create dropfile
+        df = Dropfile("cm00001-1", "i03", "testdir")
+        for testfile in testfiles:
+            df.add(testfile)
+        df.close()
+
+        # Get XML string
+        xml_str = df.to_string().decode("latin-1")
+
+        # Verify all files are in the XML
+        for testfile in testfiles:
+            assert f"<name>data_{testfiles.index(testfile)}.h5</name>" in xml_str
+
+    def test_dropfile_visit_uppercase_conversion(self):
+        """Test that visit codes are converted to uppercase in dropfile."""
+        df = Dropfile("cm00001-1", "i03", "testdir")
+        df.close()
+
+        xml_str = df.to_string().decode("latin-1")
+        assert "<visit_id>CM00001-1</visit_id>" in xml_str