diff --git a/src/murfey/server/feedback.py b/src/murfey/server/feedback.py
index 832524816..f4c4ddd98 100644
--- a/src/murfey/server/feedback.py
+++ b/src/murfey/server/feedback.py
@@ -454,14 +454,17 @@ def _get_spa_params(
 
 
 def _release_2d_hold(message: dict, _db):
-    relion_params, feedback_params = _get_spa_params(message["program_id"], _db)
-    pj_id = _pj_id(message["program_id"], _db, recipe="em-spa-class2d")
+    # Extract relevant IDs from message and ensure they are ints
+    session_id = int(message["session_id"])
+    program_id = int(message["program_id"])
+    relion_params, feedback_params = _get_spa_params(program_id, _db)
+    pj_id = _pj_id(program_id, _db, recipe="em-spa-class2d")
     if feedback_params.rerun_class2d:
         first_class2d = _db.exec(
             select(db.Class2DParameters).where(db.Class2DParameters.pj_id == pj_id)
         ).first()
         instrument_name = (
-            _db.exec(select(db.Session).where(db.Session.id == message["session_id"]))
+            _db.exec(select(db.Session).where(db.Session.id == session_id))
             .one()
             .instrument_name
         )
@@ -474,7 +477,7 @@ def _release_2d_hold(message: dict, _db):
             # its existing (already reserved) directory (message["job_dir"]), so
             # reserve only the trailing jobs: combine goes at the end and the
             # autoselect job is combine - 1 (see select_classes).
-            visit_name = _visit_name_for_session(message["session_id"], _db)
+            visit_name = _visit_name_for_session(session_id, _db)
             trailing = 3 if default_spa_parameters.do_icebreaker_jobs else 2
             feedback_params.next_job = _reserve_pipeline_job_numbers(
                 visit_name, trailing, feedback_params.next_job
@@ -490,14 +493,14 @@ def _release_2d_hold(message: dict, _db):
                 "mask_diameter": relion_params.mask_diameter or 0,
                 "combine_star_job_number": feedback_params.star_combination_job,
                 "autoselect_min_score": feedback_params.class_selection_score or 0,
-                "autoproc_program_id": message["program_id"],
+                "autoproc_program_id": program_id,
                 "nr_iter": default_spa_parameters.nr_iter_2d,
                 "nr_classes": default_spa_parameters.nr_classes_2d,
                 "do_icebreaker_jobs": default_spa_parameters.do_icebreaker_jobs,
                 "class2d_fraction_of_classes_to_remove": default_spa_parameters.fraction_of_classes_to_remove_2d,
                 "seed": int(np.random.randint(1, 100)),
                 "class_uuids": _2d_class_murfey_ids(
-                    first_class2d.particles_file, message["program_id"], _db
+                    first_class2d.particles_file, program_id, _db
                 ),
                 "class2d_grp_uuid": _db.exec(
                     select(db.Class2DParameters)
@@ -509,7 +512,7 @@ def _release_2d_hold(message: dict, _db):
                 )
                 .one()
                 .murfey_id,
-                "session_id": message["session_id"],
+                "session_id": session_id,
                 "node_creator_queue": machine_config.node_creator_queue,
             },
             "recipes": [machine_config.recipes.get("em-spa-class2d", "em-spa-class2d")],
@@ -535,8 +538,11 @@ def _release_2d_hold(message: dict, _db):
 
 
 def _release_3d_hold(message: dict, _db):
-    pj_id_params = _pj_id(message["program_id"], _db, recipe="em-spa-preprocess")
-    pj_id = _pj_id(message["program_id"], _db, recipe="em-spa-class3d")
+    # Extract IDs and ensure they are ints
+    session_id = int(message["session_id"])
+    program_id = int(message["program_id"])
+    pj_id_params = _pj_id(program_id, _db, recipe="em-spa-preprocess")
+    pj_id = _pj_id(program_id, _db, recipe="em-spa-class3d")
     relion_params = _db.exec(
         select(db.SPARelionParameters).where(
             db.SPARelionParameters.pj_id == pj_id_params
@@ -552,7 +558,7 @@ def _release_3d_hold(message: dict, _db):
     ).one()
     if class3d_params.run:
         instrument_name = (
-            _db.exec(select(db.Session).where(db.Session.id == message["session_id"]))
+            _db.exec(select(db.Session).where(db.Session.id == session_id))
             .one()
             .instrument_name
         )
@@ -588,9 +594,9 @@ def _release_3d_hold(message: dict, _db):
                 "nr_classes": default_spa_parameters.nr_classes_3d,
                 "do_icebreaker_jobs": default_spa_parameters.do_icebreaker_jobs,
                 "class2d_fraction_of_classes_to_remove": default_spa_parameters.fraction_of_classes_to_remove_2d,
-                "session_id": message["session_id"],
+                "session_id": session_id,
                 "autoproc_program_id": _app_id(
-                    _pj_id(message["program_id"], _db, recipe="em-spa-class3d"), _db
+                    _pj_id(program_id, _db, recipe="em-spa-class3d"), _db
                 ),
                 "node_creator_queue": machine_config.node_creator_queue,
             },
@@ -613,8 +619,11 @@ def _release_3d_hold(message: dict, _db):
 
 
 def _release_refine_hold(message: dict, _db):
-    pj_id_params = _pj_id(message["program_id"], _db, recipe="em-spa-preprocess")
-    pj_id = _pj_id(message["program_id"], _db, recipe="em-spa-refine")
+    # Extract IDs and ensure they are ints
+    session_id = int(message["session_id"])
+    program_id = int(message["program_id"])
+    pj_id_params = _pj_id(program_id, _db, recipe="em-spa-preprocess")
+    pj_id = _pj_id(program_id, _db, recipe="em-spa-refine")
     relion_params = _db.exec(
         select(db.SPARelionParameters).where(
             db.SPARelionParameters.pj_id == pj_id_params
@@ -637,7 +646,7 @@ def _release_refine_hold(message: dict, _db):
     ).one()
     if refine_params.run:
         instrument_name = (
-            _db.exec(select(db.Session).where(db.Session.id == message["session_id"]))
+            _db.exec(select(db.Session).where(db.Session.id == session_id))
             .one()
             .instrument_name
         )
@@ -670,9 +679,9 @@ def _release_refine_hold(message: dict, _db):
                     _db=_db,
                 ),
                 "symmetry_refined_grp_uuid": symmetry_refine_params.murfey_id,
-                "session_id": message["session_id"],
+                "session_id": session_id,
                 "autoproc_program_id": _app_id(
-                    _pj_id(message["program_id"], _db, recipe="em-spa-refine"), _db
+                    _pj_id(program_id, _db, recipe="em-spa-refine"), _db
                 ),
             },
             "recipes": [machine_config.recipes.get("em-spa-refine", "em-spa-refine")],
@@ -696,17 +705,19 @@ def _release_refine_hold(message: dict, _db):
 def _register_incomplete_2d_batch(message: dict, _db):
     """Received first batch from particle selection service"""
     # the general parameters are stored using the preprocessing auto proc program ID
+    session_id = int(message["session_id"])
     logger.info("Registering incomplete particle batch for 2D classification")
     instrument_name = (
-        _db.exec(select(db.Session).where(db.Session.id == message["session_id"]))
+        _db.exec(select(db.Session).where(db.Session.id == session_id))
         .one()
         .instrument_name
     )
     machine_config = get_machine_config(instrument_name=instrument_name)[
         instrument_name
     ]
-    pj_id_params = _pj_id(message["program_id"], _db, recipe="em-spa-preprocess")
-    pj_id = _pj_id(message["program_id"], _db, recipe="em-spa-class2d")
+    program_id = int(message["program_id"])
+    pj_id_params = _pj_id(program_id, _db, recipe="em-spa-preprocess")
+    pj_id = _pj_id(program_id, _db, recipe="em-spa-class2d")
     relion_params = _db.exec(
         select(db.SPARelionParameters).where(
             db.SPARelionParameters.pj_id == pj_id_params
@@ -727,7 +738,7 @@ def _register_incomplete_2d_batch(message: dict, _db):
     # incomplete batch runs Class2D only (no autoselect/combine), so one job is
     # enough; reserving advances the Pipeliner counter now so the next batch
     # cannot be handed the same number before this job is registered.
-    visit_name = _visit_name_for_session(message["session_id"], _db)
+    visit_name = _visit_name_for_session(session_id, _db)
     class2d_job = _current_pipeline_job_counter(visit_name, FIRST_FEEDBACK_JOB)
     feedback_params.next_job = _reserve_pipeline_job_numbers(
         visit_name, 1, FIRST_FEEDBACK_JOB
@@ -746,7 +757,7 @@ def _register_incomplete_2d_batch(message: dict, _db):
     ).one():
         class2d_params = db.Class2DParameters(
             pj_id=pj_id,
-            murfey_id=_murfey_id(message["program_id"], _db)[0],
+            murfey_id=_murfey_id(program_id, _db)[0],
             particles_file=class2d_message["particles_file"],
             class2d_dir=class2d_message["class2d_dir"],
             batch_size=class2d_message["batch_size"],
@@ -755,11 +766,9 @@ def _register_incomplete_2d_batch(message: dict, _db):
         _db.add(class2d_params)
         _db.commit()
         murfey_ids = _murfey_id(
-            message["program_id"], _db, number=default_spa_parameters.nr_classes_2d
-        )
-        _murfey_class2ds(
-            murfey_ids, class2d_message["particles_file"], message["program_id"], _db
+            program_id, _db, number=default_spa_parameters.nr_classes_2d
         )
+        _murfey_class2ds(murfey_ids, class2d_message["particles_file"], program_id, _db)
     zocalo_message: dict = {
         "parameters": {
             "particles_file": class2d_message["particles_file"],
@@ -786,9 +795,9 @@ def _register_incomplete_2d_batch(message: dict, _db):
             )
             .one()
             .murfey_id,
-            "session_id": message["session_id"],
+            "session_id": session_id,
             "autoproc_program_id": _app_id(
-                _pj_id(message["program_id"], _db, recipe="em-spa-class2d"), _db
+                _pj_id(program_id, _db, recipe="em-spa-class2d"), _db
             ),
             "node_creator_queue": machine_config.node_creator_queue,
         },
@@ -807,8 +816,9 @@ def _register_incomplete_2d_batch(message: dict, _db):
 
 def _register_complete_2d_batch(message: dict, _db):
     """Received full batch from particle selection service"""
+    session_id = int(message["session_id"])
     instrument_name = (
-        _db.exec(select(db.Session).where(db.Session.id == message["session_id"]))
+        _db.exec(select(db.Session).where(db.Session.id == session_id))
         .one()
         .instrument_name
     )
@@ -817,8 +827,10 @@ def _register_complete_2d_batch(message: dict, _db):
     ]
     class2d_message = message.get("class2d_message")
     assert isinstance(class2d_message, dict)
-    pj_id_params = _pj_id(message["program_id"], _db, recipe="em-spa-preprocess")
-    pj_id = _pj_id(message["program_id"], _db, recipe="em-spa-class2d")
+
+    program_id = int(message["program_id"])
+    pj_id_params = _pj_id(program_id, _db, recipe="em-spa-preprocess")
+    pj_id = _pj_id(program_id, _db, recipe="em-spa-class2d")
     relion_params = _db.exec(
         select(db.SPARelionParameters).where(
             db.SPARelionParameters.pj_id == pj_id_params
@@ -858,7 +870,7 @@ def _register_complete_2d_batch(message: dict, _db):
         else:
             class2d_params = db.Class2DParameters(
                 pj_id=pj_id,
-                murfey_id=_murfey_id(message["program_id"], _db)[0],
+                murfey_id=_murfey_id(program_id, _db)[0],
                 particles_file=class2d_message["particles_file"],
                 class2d_dir=class2d_message["class2d_dir"],
                 batch_size=class2d_message["batch_size"],
@@ -875,7 +887,7 @@ def _register_complete_2d_batch(message: dict, _db):
     elif not feedback_params.class_selection_score:
         # Reserve Class2D + autoselect (+ combine on the first batch) up front so
         # the numbers cannot be reused before the jobs are registered.
-        visit_name = _visit_name_for_session(message["session_id"], _db)
+        visit_name = _visit_name_for_session(session_id, _db)
         if _db.exec(
             select(func.count(db.Class2DParameters.particles_file))
             .where(db.Class2DParameters.pj_id == pj_id)
@@ -942,9 +954,9 @@ def _register_complete_2d_batch(message: dict, _db):
                 "nr_classes": default_spa_parameters.nr_classes_2d,
                 "do_icebreaker_jobs": default_spa_parameters.do_icebreaker_jobs,
                 "class2d_fraction_of_classes_to_remove": default_spa_parameters.fraction_of_classes_to_remove_2d,
-                "session_id": message["session_id"],
+                "session_id": session_id,
                 "autoproc_program_id": _app_id(
-                    _pj_id(message["program_id"], _db, recipe="em-spa-class2d"), _db
+                    _pj_id(program_id, _db, recipe="em-spa-class2d"), _db
                 ),
                 "node_creator_queue": machine_config.node_creator_queue,
             },
@@ -964,7 +976,7 @@ def _register_complete_2d_batch(message: dict, _db):
     else:
         # star_combination_job is already set by now, so this reserves just the
         # Class2D + autoselect jobs for this batch.
-        visit_name = _visit_name_for_session(message["session_id"], _db)
+        visit_name = _visit_name_for_session(session_id, _db)
         class2d_job = _reserve_2d_classification_jobs(visit_name, feedback_params)
         if _db.exec(
             select(func.count(db.Class2DParameters.particles_file))
@@ -1017,9 +1029,9 @@ def _register_complete_2d_batch(message: dict, _db):
                 "nr_classes": default_spa_parameters.nr_classes_2d,
                 "do_icebreaker_jobs": default_spa_parameters.do_icebreaker_jobs,
                 "class2d_fraction_of_classes_to_remove": default_spa_parameters.fraction_of_classes_to_remove_2d,
-                "session_id": message["session_id"],
+                "session_id": session_id,
                 "autoproc_program_id": _app_id(
-                    _pj_id(message["program_id"], _db, recipe="em-spa-class2d"), _db
+                    _pj_id(program_id, _db, recipe="em-spa-class2d"), _db
                 ),
                 "node_creator_queue": machine_config.node_creator_queue,
             },
@@ -1124,8 +1136,10 @@ def _flush_class2d(
 
 def _register_class_selection(message: dict, _db):
     """Received selection score from class selection service"""
-    pj_id_params = _pj_id(message["program_id"], _db, recipe="em-spa-preprocess")
-    pj_id = _pj_id(message["program_id"], _db, recipe="em-spa-class2d")
+    session_id = int(message["session_id"])
+    program_id = int(message["program_id"])
+    pj_id_params = _pj_id(program_id, _db, recipe="em-spa-preprocess")
+    pj_id = _pj_id(program_id, _db, recipe="em-spa-class2d")
     relion_params = _db.exec(
         select(db.SPARelionParameters).where(
             db.SPARelionParameters.pj_id == pj_id_params
@@ -1145,8 +1159,8 @@ def _register_class_selection(message: dict, _db):
     feedback_params.class_selection_score = message.get("class_selection_score") or 0
     feedback_params.hold_class2d = False
     _flush_class2d(
-        message["session_id"],
-        message["program_id"],
+        session_id,
+        program_id,
         _db,
         relion_params=relion_params,
         feedback_params=feedback_params,
@@ -1313,16 +1327,20 @@ def _register_3d_batch(message: dict, _db):
     """Received 3d batch from class selection service"""
     class3d_message = message.get("class3d_message")
     assert isinstance(class3d_message, dict)
+
+    session_id = int(message["session_id"])
     instrument_name = (
-        _db.exec(select(db.Session).where(db.Session.id == message["session_id"]))
+        _db.exec(select(db.Session).where(db.Session.id == session_id))
         .one()
         .instrument_name
     )
     machine_config = get_machine_config(instrument_name=instrument_name)[
         instrument_name
     ]
-    pj_id_params = _pj_id(message["program_id"], _db, recipe="em-spa-preprocess")
-    pj_id = _pj_id(message["program_id"], _db, recipe="em-spa-class3d")
+
+    program_id = int(message["program_id"])
+    pj_id_params = _pj_id(program_id, _db, recipe="em-spa-preprocess")
+    pj_id = _pj_id(program_id, _db, recipe="em-spa-class3d")
     relion_params = _db.exec(
         select(db.SPARelionParameters).where(
             db.SPARelionParameters.pj_id == pj_id_params
@@ -1336,9 +1354,7 @@ def _register_3d_batch(message: dict, _db):
     ).one()
 
     visit_name = (
-        _db.exec(select(db.Session).where(db.Session.id == message["session_id"]))
-        .one()
-        .visit
+        _db.exec(select(db.Session).where(db.Session.id == session_id)).one().visit
     )
 
     provided_initial_model = _find_initial_model(visit_name, machine_config)
@@ -1372,8 +1388,8 @@ def _register_3d_batch(message: dict, _db):
         _db.add(feedback_params)
         _db.commit()
 
-        class3d_grp_uuid = _murfey_id(message["program_id"], _db)[0]
-        class_uuids = _murfey_id(message["program_id"], _db, number=4)
+        class3d_grp_uuid = _murfey_id(program_id, _db)[0]
+        class_uuids = _murfey_id(program_id, _db, number=4)
         class3d_params = db.Class3DParameters(
             pj_id=pj_id,
             murfey_id=class3d_grp_uuid,
@@ -1386,7 +1402,7 @@ def _register_3d_batch(message: dict, _db):
         _murfey_class3ds(
             class_uuids,
             class3d_message["particles_file"],
-            message["program_id"],
+            program_id,
             _db,
         )
 
@@ -1411,8 +1427,8 @@ def _register_3d_batch(message: dict, _db):
             visit_name, 2, feedback_params.next_job
         )
         class3d_dir = f"{class3d_message['class3d_dir']}{(class3d_job):03}"
-        class3d_grp_uuid = _murfey_id(message["program_id"], _db)[0]
-        class_uuids = _murfey_id(message["program_id"], _db, number=4)
+        class3d_grp_uuid = _murfey_id(program_id, _db)[0]
+        class_uuids = _murfey_id(program_id, _db, number=4)
         class3d_params = db.Class3DParameters(
             pj_id=pj_id,
             murfey_id=class3d_grp_uuid,
@@ -1423,7 +1439,7 @@ def _register_3d_batch(message: dict, _db):
         _db.add(class3d_params)
         _db.commit()
         _murfey_class3ds(
-            class_uuids, class3d_message["particles_file"], message["program_id"], _db
+            class_uuids, class3d_message["particles_file"], program_id, _db
         )
 
         feedback_params.hold_class3d = True
@@ -1444,9 +1460,9 @@ def _register_3d_batch(message: dict, _db):
                 "nr_classes": default_spa_parameters.nr_classes_3d,
                 "do_icebreaker_jobs": default_spa_parameters.do_icebreaker_jobs,
                 "class2d_fraction_of_classes_to_remove": default_spa_parameters.fraction_of_classes_to_remove_2d,
-                "session_id": message["session_id"],
+                "session_id": session_id,
                 "autoproc_program_id": _app_id(
-                    _pj_id(message["program_id"], _db, recipe="em-spa-class3d"), _db
+                    _pj_id(program_id, _db, recipe="em-spa-class3d"), _db
                 ),
                 "node_creator_queue": machine_config.node_creator_queue,
             },
@@ -1487,9 +1503,9 @@ def _register_3d_batch(message: dict, _db):
                 "nr_classes": default_spa_parameters.nr_classes_3d,
                 "do_icebreaker_jobs": default_spa_parameters.do_icebreaker_jobs,
                 "class2d_fraction_of_classes_to_remove": default_spa_parameters.fraction_of_classes_to_remove_2d,
-                "session_id": message["session_id"],
+                "session_id": session_id,
                 "autoproc_program_id": _app_id(
-                    _pj_id(message["program_id"], _db, recipe="em-spa-class3d"), _db
+                    _pj_id(program_id, _db, recipe="em-spa-class3d"), _db
                 ),
                 "node_creator_queue": machine_config.node_creator_queue,
             },
@@ -1510,7 +1526,8 @@ def _register_3d_batch(message: dict, _db):
 
 def _register_initial_model(message: dict, _db):
     """Received initial model from 3d classification service"""
-    pj_id_params = _pj_id(message["program_id"], _db, recipe="em-spa-preprocess")
+    program_id = int(message["program_id"])
+    pj_id_params = _pj_id(program_id, _db, recipe="em-spa-preprocess")
     # Add the initial model file to the database
     feedback_params = _db.exec(
         select(db.ClassificationFeedbackParameters).where(
@@ -1524,7 +1541,7 @@ def _register_initial_model(message: dict, _db):
 
 
 def _flush_tomography_preprocessing(message: dict, _db):
-    session_id = message["session_id"]
+    session_id = int(message["session_id"])
     instrument_name = (
         _db.exec(select(db.Session).where(db.Session.id == session_id))
         .one()
@@ -1630,7 +1647,7 @@ def _flush_tomography_preprocessing(message: dict, _db):
 
 def _flush_grid_square_records(message: dict, _db):
     tag = message["tag"]
-    session_id = message["session_id"]
+    session_id = int(message["session_id"])
     gs_ids = []
     for gs in _db.exec(
         select(db.GridSquare)
@@ -1652,16 +1669,18 @@ def _flush_foil_hole_records(grid_square_id: int, _db):
 
 def _register_refinement(message: dict, _db):
     """Received class to refine from 3D classification"""
+    session_id = int(message["session_id"])
     instrument_name = (
-        _db.exec(select(db.Session).where(db.Session.id == message["session_id"]))
+        _db.exec(select(db.Session).where(db.Session.id == session_id))
         .one()
         .instrument_name
     )
     machine_config = get_machine_config(instrument_name=instrument_name)[
         instrument_name
     ]
-    pj_id_params = _pj_id(message["program_id"], _db, recipe="em-spa-preprocess")
-    pj_id = _pj_id(message["program_id"], _db, recipe="em-spa-refine")
+    program_id = int(message["program_id"])
+    pj_id_params = _pj_id(program_id, _db, recipe="em-spa-preprocess")
+    pj_id = _pj_id(program_id, _db, recipe="em-spa-refine")
     relion_params = _db.exec(
         select(db.SPARelionParameters).where(
             db.SPARelionParameters.pj_id == pj_id_params
@@ -1705,7 +1724,7 @@ def _register_refinement(message: dict, _db):
             # Reserve the contiguous refinement block: re-extraction
             # Select (base) + Extract (base + 1), Refine3D (base + 2),
             # MaskCreate (base + 3) and PostProcess (base + 4).
-            visit_name = _visit_name_for_session(message["session_id"], _db)
+            visit_name = _visit_name_for_session(session_id, _db)
             refine_job = (
                 _current_pipeline_job_counter(visit_name, feedback_params.next_job) + 2
             )
@@ -1719,10 +1738,10 @@ def _register_refinement(message: dict, _db):
                     visit_name, 5, feedback_params.next_job
                 )
             refine_dir = f"{message['refine_dir']}{refine_job:03}"
-            refined_grp_uuid = _murfey_id(message["program_id"], _db)[0]
-            refined_class_uuid = _murfey_id(message["program_id"], _db)[0]
-            symmetry_refined_grp_uuid = _murfey_id(message["program_id"], _db)[0]
-            symmetry_refined_class_uuid = _murfey_id(message["program_id"], _db)[0]
+            refined_grp_uuid = _murfey_id(program_id, _db)[0]
+            refined_class_uuid = _murfey_id(program_id, _db)[0]
+            symmetry_refined_grp_uuid = _murfey_id(program_id, _db)[0]
+            symmetry_refined_class_uuid = _murfey_id(program_id, _db)[0]
 
             refine_params = db.RefineParameters(
                 tag="first",
@@ -1747,14 +1766,14 @@ def _register_refinement(message: dict, _db):
                 murfey_id=refined_class_uuid,
                 refine_dir=refine_dir,
                 tag="first",
-                app_id=message["program_id"],
+                app_id=program_id,
                 _db=_db,
             )
             _murfey_refine(
                 murfey_id=symmetry_refined_class_uuid,
                 refine_dir=refine_dir,
                 tag="symmetry",
-                app_id=message["program_id"],
+                app_id=program_id,
                 _db=_db,
             )
 
@@ -1786,9 +1805,9 @@ def _register_refinement(message: dict, _db):
                     _db=_db,
                 ),
                 "symmetry_refined_grp_uuid": symmetry_refine_params.murfey_id,
-                "session_id": message["session_id"],
+                "session_id": session_id,
                 "autoproc_program_id": _app_id(
-                    _pj_id(message["program_id"], _db, recipe="em-spa-refine"), _db
+                    _pj_id(program_id, _db, recipe="em-spa-refine"), _db
                 ),
             },
             "recipes": [machine_config.recipes.get("em-spa-refine", "em-spa-refine")],
@@ -1808,16 +1827,19 @@ def _register_refinement(message: dict, _db):
 
 def _register_bfactors(message: dict, _db):
     """Received refined class to calculate b-factor"""
+    session_id = int(message["session_id"])
     instrument_name = (
-        _db.exec(select(db.Session).where(db.Session.id == message["session_id"]))
+        _db.exec(select(db.Session).where(db.Session.id == session_id))
         .one()
         .instrument_name
     )
     machine_config = get_machine_config(instrument_name=instrument_name)[
         instrument_name
     ]
-    pj_id_params = _pj_id(message["program_id"], _db, recipe="em-spa-preprocess")
-    pj_id = _pj_id(message["program_id"], _db, recipe="em-spa-refine")
+
+    program_id = int(message["program_id"])
+    pj_id_params = _pj_id(program_id, _db, recipe="em-spa-preprocess")
+    pj_id = _pj_id(program_id, _db, recipe="em-spa-refine")
     relion_params = _db.exec(
         select(db.SPARelionParameters).where(
             db.SPARelionParameters.pj_id == pj_id_params
@@ -1908,9 +1930,9 @@ def _register_bfactors(message: dict, _db):
                 "node_creator_queue": machine_config.node_creator_queue,
                 "refined_grp_uuid": bfactor_params.refined_grp_uuid,
                 "refined_class_uuid": bfactor_params.refined_class_uuid,
-                "session_id": message["session_id"],
+                "session_id": session_id,
                 "autoproc_program_id": _app_id(
-                    _pj_id(message["program_id"], _db, recipe="em-spa-refine"), _db
+                    _pj_id(program_id, _db, recipe="em-spa-refine"), _db
                 ),
             },
             "recipes": ["em-spa-bfactor"],
@@ -1928,7 +1950,8 @@ def _register_bfactors(message: dict, _db):
 
 def _save_bfactor(message: dict, _db):
     """Received b-factor from refinement run"""
-    pj_id = _pj_id(message["program_id"], _db, recipe="em-spa-refine")
+    program_id = int(message["program_id"])
+    pj_id = _pj_id(program_id, _db, recipe="em-spa-refine")
     bfactor_run = _db.exec(
         select(db.BFactors)
         .where(db.BFactors.pj_id == pj_id)
@@ -1962,7 +1985,7 @@ def _save_bfactor(message: dict, _db):
                     "buffer_command": {
                         "ispyb_command": "insert_particle_classification"
                     },
-                    "program_id": message["program_id"],
+                    "program_id": program_id,
                     "bfactor_fit_intercept": str(float(bfactor_fitting[1])),
                     "bfactor_fit_linear": str(float(bfactor_fitting[0])),
                 },
@@ -1992,6 +2015,8 @@ def feedback_callback(
         # The context manager will handle session cleanup on error
         with Session(engine, expire_on_commit=False) as _db:
             if message["register"] == "motion_corrected":
+                program_id = int(message["program_id"])
+
                 collected_ids = _db.exec(
                     select(
                         db.DataCollectionGroup,
@@ -2002,10 +2027,17 @@ def feedback_callback(
                     .where(db.DataCollection.dcg_id == db.DataCollectionGroup.id)
                     .where(db.ProcessingJob.dc_id == db.DataCollection.id)
                     .where(db.AutoProcProgram.pj_id == db.ProcessingJob.id)
-                    .where(db.AutoProcProgram.id == message["program_id"])
+                    .where(db.AutoProcProgram.id == program_id)
                 ).one()
                 session_id = collected_ids[0].session_id
 
+                # Update the 'last_active' column for the associated Murfey session
+                murfey_session = _db.exec(
+                    select(db.Session).where(db.Session.id == session_id)
+                ).one()
+                murfey_session.last_active = datetime.now()
+                _db.add(murfey_session)
+
                 # Find the autoprocprogram id for the alignment recipe
                 alignment_ids = _db.exec(
                     select(
@@ -2035,11 +2067,7 @@ def feedback_callback(
                     and not relevant_tilt_series.processing_requested
                     and relevant_tilt_series.tilt_series_length > 2
                 ):
-                    instrument_name = (
-                        _db.exec(select(db.Session).where(db.Session.id == session_id))
-                        .one()
-                        .instrument_name
-                    )
+                    instrument_name = murfey_session.instrument_name
                     machine_config = get_machine_config(
                         instrument_name=instrument_name
                     )[instrument_name]
@@ -2104,7 +2132,7 @@ def feedback_callback(
                     murfey.server._transport_object.transport.ack(header)
                 return None
             elif message["register"] == "spa_processing_parameters":
-                session_id = message["session_id"]
+                session_id = int(message["session_id"])
                 collected_ids = _db.exec(
                     select(
                         db.DataCollectionGroup,
@@ -2176,7 +2204,7 @@ def feedback_callback(
                     murfey.server._transport_object.transport.ack(header)
                 return None
             elif message["register"] == "tomography_processing_parameters":
-                session_id = message["session_id"]
+                session_id = int(message["session_id"])
                 collected_ids = _db.exec(
                     select(
                         db.DataCollectionGroup,
@@ -2253,10 +2281,10 @@ def feedback_callback(
                     murfey.server._transport_object.transport.ack(header)
                 return None
             elif message["register"] == "run_class3d":
+                session_id = int(message["session_id"])
                 session_processing_parameters = _db.exec(
                     select(db.SessionProcessingParameters).where(
-                        db.SessionProcessingParameters.session_id
-                        == message["session_id"]
+                        db.SessionProcessingParameters.session_id == session_id
                     )
                 ).all()
                 if (
diff --git a/src/murfey/util/db.py b/src/murfey/util/db.py
index 3b3b77757..fcc49f6d5 100644
--- a/src/murfey/util/db.py
+++ b/src/murfey/util/db.py
@@ -62,6 +62,12 @@ class Session(SQLModel, table=True):  # type: ignore
     visit_end_time: Optional[datetime] = Field(
         default=None, sa_type=DateTime(timezone=False)
     )
+    last_active: Optional[datetime] = Field(
+        default=None, sa_type=DateTime(timezone=False)
+    )
+
+    # Relationships
+    # -------------
     smartem_acquisition_uuid: str | None = Field(default=None)
 
     # Image sites associated with this session
diff --git a/src/murfey/workflows/clem/register_align_and_merge_results.py b/src/murfey/workflows/clem/register_align_and_merge_results.py
index d1c039b6d..f9c448463 100644
--- a/src/murfey/workflows/clem/register_align_and_merge_results.py
+++ b/src/murfey/workflows/clem/register_align_and_merge_results.py
@@ -72,11 +72,7 @@ def run(message: dict, murfey_db: Session) -> dict[str, bool]:
         ...
     """
 
-    session_id: int = (
-        int(message["session_id"])
-        if not isinstance(message["session_id"], int)
-        else message["session_id"]
-    )
+    session_id = int(message["session_id"])
 
     # Validate message and try and load results
     try:
diff --git a/src/murfey/workflows/register_atlas_update.py b/src/murfey/workflows/register_atlas_update.py
index 151746c2d..7f0bb2220 100644
--- a/src/murfey/workflows/register_atlas_update.py
+++ b/src/murfey/workflows/register_atlas_update.py
@@ -20,7 +20,7 @@ def run(
         return {"success": False, "requeue": False}
 
     logger.info(f"Registering updated atlas: \n{message}")
-
+    murfey_session_id = int(message["session_id"])
     murfey.server._transport_object.do_update_atlas(
         atlas_id=message["atlas_id"],
         atlas_image=message["atlas"],
@@ -39,7 +39,7 @@ def run(
     ):
         dcgs_atlas = murfey_db.exec(
             select(DataCollectionGroup)
-            .where(DataCollectionGroup.session_id == message["session_id"])
+            .where(DataCollectionGroup.session_id == murfey_session_id)
             .where(DataCollectionGroup.atlas == message["atlas"])
             .where(DataCollectionGroup.sample == message["sample"])
         ).all()
@@ -54,7 +54,7 @@ def run(
     ):
         try:
             for hook in dcg_hooks:
-                hook.load()(message["dcgid"], session_id=message["session_id"])
+                hook.load()(message["dcgid"], session_id=murfey_session_id)
         except Exception:
             logger.error("Call to data collection group hook failed", exc_info=True)
     return {"success": True}
diff --git a/src/murfey/workflows/register_data_collection.py b/src/murfey/workflows/register_data_collection.py
index 87de8a3f6..f06aeb18b 100644
--- a/src/murfey/workflows/register_data_collection.py
+++ b/src/murfey/workflows/register_data_collection.py
@@ -5,8 +5,8 @@
 from sqlmodel import select
 from sqlmodel.orm.session import Session as SQLModelSession
 
-import murfey.util.db as MurfeyDB
 import murfey.server
+import murfey.util.db as MurfeyDB
 from murfey.server.ispyb import ISPyBSession, get_session_id
 from murfey.util import sanitise
 
@@ -21,7 +21,7 @@ def run(message: dict, murfey_db: SQLModelSession) -> dict[str, bool]:
 
     logger.info(f"Registering the following data collection: \n{message}")
 
-    murfey_session_id = message["session_id"]
+    murfey_session_id = int(message["session_id"])
     ispyb_session_id = get_session_id(
         microscope=message["microscope"],
         proposal_code=message["proposal_code"],
diff --git a/src/murfey/workflows/register_data_collection_group.py b/src/murfey/workflows/register_data_collection_group.py
index cce049df7..4bcae10d6 100644
--- a/src/murfey/workflows/register_data_collection_group.py
+++ b/src/murfey/workflows/register_data_collection_group.py
@@ -22,6 +22,7 @@ def register_dcg(
         logger.error("Unable to find transport manager")
         return None
 
+    murfey_session_id = int(message["session_id"])
     ispyb_session_id = get_session_id(
         microscope=message["microscope"],
         proposal_code=message["proposal_code"],
@@ -31,7 +32,7 @@ def register_dcg(
     )
     if ispyb_session_id is None:
         murfey_dcg = DataCollectionGroup(
-            session_id=message["session_id"],
+            session_id=murfey_session_id,
             tag=message.get("tag"),
             smartem_grid_uuid=message.get("smartem_grid_uuid"),
         )
@@ -70,7 +71,7 @@ def register_dcg(
             atlas=message.get("atlas", ""),
             atlas_pixel_size=message.get("atlas_pixel_size"),
             sample=message.get("sample"),
-            session_id=message["session_id"],
+            session_id=murfey_session_id,
             tag=message.get("tag"),
             smartem_grid_uuid=message.get("smartem_grid_uuid"),
         )
@@ -78,7 +79,7 @@ def register_dcg(
     if dcgid is not None and message.get("atlas_x_stage_position"):
         atlas_site = ImagingSite(
             dcg_id=dcgid,
-            session_id=message["session_id"],
+            session_id=murfey_session_id,
             site_name=message.get("tag"),
             data_type="atlas",
             pos_x=message.get("atlas_x_stage_position"),
@@ -99,10 +100,10 @@ def run(message: dict, murfey_db: SQLModelSession) -> dict[str, bool]:
         return {"success": False, "requeue": False}
 
     logger.info(f"Registering the following data collection group: \n{message}")
-
+    murfey_session_id = int(message["session_id"])
     if dcg_murfey := murfey_db.exec(
         select(DataCollectionGroup)
-        .where(DataCollectionGroup.session_id == message["session_id"])
+        .where(DataCollectionGroup.session_id == murfey_session_id)
         .where(DataCollectionGroup.tag == message.get("tag"))
     ).all():
         dcgid = dcg_murfey[0].id
@@ -128,7 +129,7 @@ def run(message: dict, murfey_db: SQLModelSession) -> dict[str, bool]:
     ):
         dcgs_atlas = murfey_db.exec(
             select(DataCollectionGroup)
-            .where(DataCollectionGroup.session_id == message["session_id"])
+            .where(DataCollectionGroup.session_id == murfey_session_id)
             .where(DataCollectionGroup.atlas == message["atlas"])
             .where(DataCollectionGroup.sample == message["sample"])
         ).all()
@@ -140,7 +141,7 @@ def run(message: dict, murfey_db: SQLModelSession) -> dict[str, bool]:
     if dcg_hooks := entry_points(group="murfey.hooks", name="data_collection_group"):
         try:
             for hook in dcg_hooks:
-                hook.load()(dcgid, session_id=message["session_id"])
+                hook.load()(dcgid, session_id=murfey_session_id)
         except Exception:
             logger.error("Call to data collection group hook failed", exc_info=True)
 
diff --git a/src/murfey/workflows/register_processing_job.py b/src/murfey/workflows/register_processing_job.py
index b6ab7b5a5..e05ae6299 100644
--- a/src/murfey/workflows/register_processing_job.py
+++ b/src/murfey/workflows/register_processing_job.py
@@ -5,9 +5,9 @@
 from sqlmodel import select
 from sqlmodel.orm.session import Session as SQLModelSession
 
+import murfey.server
 import murfey.server.prometheus as prom
 import murfey.util.db as MurfeyDB
-import murfey.server
 from murfey.server.ispyb import ISPyBSession
 from murfey.util import sanitise
 
@@ -22,7 +22,7 @@ def run(message: dict, murfey_db: SQLModelSession):
 
     logger.info(f"Registering the following processing job: \n{message}")
 
-    murfey_session_id = message["session_id"]
+    murfey_session_id = int(message["session_id"])
     dc = murfey_db.exec(
         select(MurfeyDB.DataCollection, MurfeyDB.DataCollectionGroup)
         .where(MurfeyDB.DataCollection.dcg_id == MurfeyDB.DataCollectionGroup.id)
@@ -86,9 +86,9 @@ def run(message: dict, murfey_db: SQLModelSession):
             record = ISPyBDB.AutoProcProgram(
                 processingJobId=pid, processingStartTime=datetime.now()
             )
-            appid = murfey.server._transport_object.do_update_processing_status(record).get(
-                "return_value", None
-            )
+            appid = murfey.server._transport_object.do_update_processing_status(
+                record
+            ).get("return_value", None)
             if appid is None:
                 return {"success": False, "requeue": True}
             murfey_app = MurfeyDB.AutoProcProgram(id=appid, pj_id=pid)
diff --git a/src/murfey/workflows/spa/ctf_estimation.py b/src/murfey/workflows/spa/ctf_estimation.py
index 605442859..3dc3693fc 100644
--- a/src/murfey/workflows/spa/ctf_estimation.py
+++ b/src/murfey/workflows/spa/ctf_estimation.py
@@ -42,13 +42,14 @@
 def ctf_estimated(message: dict, murfey_db: Session) -> dict[str, bool]:
     if not SMARTEM_ACTIVE:
         return {"success": True}
+    session_id = int(message["session_id"])
     movie = murfey_db.exec(
         select(Movie).where(Movie.murfey_id == message["mc_uuid"])
     ).one()
     if movie.smartem_uuid:
         try:
             session = murfey_db.exec(
-                select(MurfeySession).where(MurfeySession.id == message["session_id"])
+                select(MurfeySession).where(MurfeySession.id == session_id)
             ).one()
             machine_config = get_machine_config(
                 instrument_name=session.instrument_name
diff --git a/src/murfey/workflows/spa/flush_spa_preprocess.py b/src/murfey/workflows/spa/flush_spa_preprocess.py
index a8849a163..17cdf97dd 100644
--- a/src/murfey/workflows/spa/flush_spa_preprocess.py
+++ b/src/murfey/workflows/spa/flush_spa_preprocess.py
@@ -302,7 +302,7 @@ def _flush_position_analysis(
 
 
 def flush_spa_preprocess(message: dict, murfey_db: Session) -> dict[str, bool]:
-    session_id = message["session_id"]
+    session_id = int(message["session_id"])
     stashed_files = murfey_db.exec(
         select(PreprocessStash)
         .where(PreprocessStash.session_id == session_id)
@@ -312,7 +312,7 @@ def flush_spa_preprocess(message: dict, murfey_db: Session) -> dict[str, bool]:
         return {"success": True}
 
     murfey_session = murfey_db.exec(
-        select(MurfeySession).where(MurfeySession.id == message["session_id"])
+        select(MurfeySession).where(MurfeySession.id == session_id)
     ).one()
     machine_config = get_machine_config(instrument_name=murfey_session.instrument_name)[
         murfey_session.instrument_name
diff --git a/src/murfey/workflows/spa/motion_correction.py b/src/murfey/workflows/spa/motion_correction.py
index fa77be2be..891967804 100644
--- a/src/murfey/workflows/spa/motion_correction.py
+++ b/src/murfey/workflows/spa/motion_correction.py
@@ -42,13 +42,14 @@
 def motion_corrected(message: dict, murfey_db: Session) -> dict[str, bool]:
     if not SMARTEM_ACTIVE:
         return {"success": True}
+    session_id = int(message["session_id"])
     movie = murfey_db.exec(
         select(Movie).where(Movie.murfey_id == message["mc_uuid"])
     ).one()
     if movie.smartem_uuid:
         try:
             session = murfey_db.exec(
-                select(MurfeySession).where(MurfeySession.id == message["session_id"])
+                select(MurfeySession).where(MurfeySession.id == session_id)
             ).one()
             machine_config = get_machine_config(
                 instrument_name=session.instrument_name
diff --git a/src/murfey/workflows/spa/picking.py b/src/murfey/workflows/spa/picking.py
index 1f5610282..efc3fac8d 100644
--- a/src/murfey/workflows/spa/picking.py
+++ b/src/murfey/workflows/spa/picking.py
@@ -1,4 +1,5 @@
 import asyncio
+from datetime import datetime
 from logging import getLogger
 from typing import List
 
@@ -57,7 +58,10 @@ def _register_picked_particles_use_diameter(message: dict, _db: Session):
     # Add this message to the table of seen messages
     params_to_forward = message.get("extraction_parameters")
     assert isinstance(params_to_forward, dict)
-    pj_id = _pj_id(message["program_id"], _db)
+
+    session_id = int(message["session_id"])
+    program_id = int(message["program_id"])
+    pj_id = _pj_id(program_id, _db)
     ctf_params = CtfParameters(
         pj_id=pj_id,
         micrographs_file=params_to_forward["micrographs_file"],
@@ -80,9 +84,7 @@ def _register_picked_particles_use_diameter(message: dict, _db: Session):
     if picking_db_len > default_spa_parameters.nr_picks_before_diameter:
         # If there are enough particles to get a diameter
         instrument_name = (
-            _db.exec(
-                select(MurfeySession).where(MurfeySession.id == message["session_id"])
-            )
+            _db.exec(select(MurfeySession).where(MurfeySession.id == session_id))
             .one()
             .instrument_name
         )
@@ -130,9 +132,9 @@ def _register_picked_particles_use_diameter(message: dict, _db: Session):
                         "downscale": relion_options["downscale"],
                         "kv": relion_options["voltage"],
                         "node_creator_queue": machine_config.node_creator_queue,
-                        "session_id": message["session_id"],
+                        "session_id": session_id,
                         "autoproc_program_id": _app_id(
-                            _pj_id(message["program_id"], _db, recipe="em-spa-extract"),
+                            _pj_id(program_id, _db, recipe="em-spa-extract"),
                             _db,
                         ),
                         "batch_size": default_spa_parameters.batch_size_2d,
@@ -171,9 +173,9 @@ def _register_picked_particles_use_diameter(message: dict, _db: Session):
                     "downscale": relion_options["downscale"],
                     "kv": relion_options["voltage"],
                     "node_creator_queue": machine_config.node_creator_queue,
-                    "session_id": message["session_id"],
+                    "session_id": session_id,
                     "autoproc_program_id": _app_id(
-                        _pj_id(message["program_id"], _db, recipe="em-spa-extract"), _db
+                        _pj_id(program_id, _db, recipe="em-spa-extract"), _db
                     ),
                     "batch_size": default_spa_parameters.batch_size_2d,
                 },
@@ -204,15 +206,17 @@ def _register_picked_particles_use_boxsize(message: dict, _db: Session):
     params_to_forward = message.get("extraction_parameters")
     assert isinstance(params_to_forward, dict)
 
+    session_id = int(message["session_id"])
     instrument_name = (
-        _db.exec(select(MurfeySession).where(MurfeySession.id == message["session_id"]))
+        _db.exec(select(MurfeySession).where(MurfeySession.id == session_id))
         .one()
         .instrument_name
     )
     machine_config = get_machine_config(instrument_name=instrument_name)[
         instrument_name
     ]
-    pj_id = _pj_id(message["program_id"], _db)
+    program_id = int(message["program_id"])
+    pj_id = _pj_id(program_id, _db)
     ctf_params = CtfParameters(
         pj_id=pj_id,
         micrographs_file=params_to_forward["micrographs_file"],
@@ -253,9 +257,9 @@ def _register_picked_particles_use_boxsize(message: dict, _db: Session):
             "downscale": relion_params.downscale,
             "kv": relion_params.voltage,
             "node_creator_queue": machine_config.node_creator_queue,
-            "session_id": message["session_id"],
+            "session_id": session_id,
             "autoproc_program_id": _app_id(
-                _pj_id(message["program_id"], _db, recipe="em-spa-extract"), _db
+                _pj_id(program_id, _db, recipe="em-spa-extract"), _db
             ),
             "batch_size": default_spa_parameters.batch_size_2d,
         },
@@ -298,11 +302,13 @@ def _request_email(
 
 
 def _check_notifications(message: dict, murfey_db: Session) -> None:
+    session_id = int(message["session_id"])
+    program_id = int(message["program_id"])
     data_collection_hierarchy = murfey_db.exec(
         select(DataCollection, ProcessingJob, AutoProcProgram)
         .where(ProcessingJob.dc_id == DataCollection.id)
         .where(AutoProcProgram.pj_id == ProcessingJob.id)
-        .where(AutoProcProgram.id == message["program_id"])
+        .where(AutoProcProgram.id == program_id)
     ).all()
     dcgid = data_collection_hierarchy[0][0].dcg_id
     notification_parameters = murfey_db.exec(
@@ -383,11 +389,19 @@ def _check_notifications(message: dict, murfey_db: Session) -> None:
             "Requested email notification for the following abnormal parameters: \n"
             f"{', '.join([f'{p}' for p in failures])}"
         )
-        _request_email(failures, dcgid, message["session_id"], murfey_db)
+        _request_email(failures, dcgid, session_id, murfey_db)
     return None
 
 
 def particles_picked(message: dict, murfey_db: Session) -> dict[str, bool]:
+    # Update the last_active' column for the session
+    session_id = int(message["session_id"])
+    session = murfey_db.exec(
+        select(MurfeySession).where(MurfeySession.id == session_id)
+    ).one()
+    session.last_active = datetime.now()
+    murfey_db.add(session)
+
     movie = murfey_db.exec(
         select(Movie).where(Movie.murfey_id == message["motion_correction_id"])
     ).one()
@@ -396,9 +410,6 @@ def particles_picked(message: dict, murfey_db: Session) -> dict[str, bool]:
     murfey_db.commit()
     if SMARTEM_ACTIVE and movie.smartem_uuid:
         try:
-            session = murfey_db.exec(
-                select(MurfeySession).where(MurfeySession.id == message["session_id"])
-            ).one()
             machine_config = get_machine_config(
                 instrument_name=session.instrument_name
             )[session.instrument_name]
@@ -461,18 +472,16 @@ async def _publish_particle_picking_completed(
                 "Failed to emit particle picking complete event to smartem",
                 exc_info=True,
             )
+    program_id = int(message["program_id"])
     feedback_params = murfey_db.exec(
         select(ClassificationFeedbackParameters).where(
-            ClassificationFeedbackParameters.pj_id
-            == _pj_id(message["program_id"], murfey_db)
+            ClassificationFeedbackParameters.pj_id == _pj_id(program_id, murfey_db)
         )
     ).one()
     if feedback_params.estimate_particle_diameter:
         _register_picked_particles_use_diameter(message, murfey_db)
     else:
         _register_picked_particles_use_boxsize(message, murfey_db)
-    prom.preprocessed_movies.labels(
-        processing_job=_pj_id(message["program_id"], murfey_db)
-    ).inc()
+    prom.preprocessed_movies.labels(processing_job=_pj_id(program_id, murfey_db)).inc()
     _check_notifications(message, murfey_db)
     return {"success": True}
diff --git a/src/murfey/workflows/spa/register_grid_square.py b/src/murfey/workflows/spa/register_grid_square.py
index c6893f77e..1106f35ee 100644
--- a/src/murfey/workflows/spa/register_grid_square.py
+++ b/src/murfey/workflows/spa/register_grid_square.py
@@ -218,7 +218,7 @@ def register_grid_square(
 
 def run(message: dict, murfey_db: SQLModelSession) -> dict[str, bool]:
     return register_grid_square(
-        message["session_id"],
+        int(message["session_id"]),
         message["gsid"],
         GridSquareParameters(**message["grid_square_params"]),
         murfey_db,
diff --git a/src/murfey/workflows/sxt/process_sxt_tilt_series.py b/src/murfey/workflows/sxt/process_sxt_tilt_series.py
index 7269e57b9..2b0110e24 100644
--- a/src/murfey/workflows/sxt/process_sxt_tilt_series.py
+++ b/src/murfey/workflows/sxt/process_sxt_tilt_series.py
@@ -205,7 +205,7 @@ def process_sxt_tilt_series(
 def run(message: dict, murfey_db: SQLModelSession) -> dict[str, bool]:
     return process_sxt_tilt_series(
         message["visit_name"],
-        message["session_id"],
+        int(message["session_id"]),
         SXTTiltSeriesInfo(**message["tilt_series_info"]),
         murfey_db,
     )
diff --git a/src/murfey/workflows/sxt/sxt_metadata.py b/src/murfey/workflows/sxt/sxt_metadata.py
index 149cb38a1..69e19913f 100644
--- a/src/murfey/workflows/sxt/sxt_metadata.py
+++ b/src/murfey/workflows/sxt/sxt_metadata.py
@@ -130,7 +130,7 @@ def register_sxt_roi(
 
 def run(message: dict, murfey_db: SQLModelSession) -> dict[str, bool]:
     return register_sxt_roi(
-        message["session_id"],
+        int(message["session_id"]),
         message["roi_name"],
         SearchMapParameters(**message["roi_info"]),
         murfey_db,
diff --git a/src/murfey/workflows/tomo/picking.py b/src/murfey/workflows/tomo/picking.py
index bb5c749d5..cee36e7e6 100644
--- a/src/murfey/workflows/tomo/picking.py
+++ b/src/murfey/workflows/tomo/picking.py
@@ -52,9 +52,9 @@ def _ids_tomo_classification(
 def _register_picked_tomogram_use_diameter(message: dict, murfey_db: Session):
     """Received picked particles from the tomogram autopick service"""
     # Add this message to the table of seen messages
-    dcg_id, pj_id = _ids_tomo_classification(
-        message["program_id"], "em-tomo-class2d", murfey_db
-    )
+    session_id = int(message["session_id"])
+    program_id = int(message["program_id"])
+    dcg_id, pj_id = _ids_tomo_classification(program_id, "em-tomo-class2d", murfey_db)
 
     pick_params = TomogramPicks(
         pj_id=pj_id,
@@ -72,9 +72,7 @@ def _register_picked_tomogram_use_diameter(message: dict, murfey_db: Session):
     if picking_db_len > default_tomo_parameters.batch_size_2d:
         # If there are enough particles to get a diameter
         instrument_name = (
-            murfey_db.exec(
-                select(MurfeySession).where(MurfeySession.id == message["session_id"])
-            )
+            murfey_db.exec(select(MurfeySession).where(MurfeySession.id == session_id))
             .one()
             .instrument_name
         )
@@ -131,7 +129,7 @@ def _register_picked_tomogram_use_diameter(message: dict, murfey_db: Session):
                         "particle_diameter": particle_diameter,
                         "kv": tomo_params.voltage,
                         "node_creator_queue": machine_config.node_creator_queue,
-                        "session_id": message["session_id"],
+                        "session_id": session_id,
                         "autoproc_program_id": _app_id(pj_id, murfey_db),
                         "batch_size": default_tomo_parameters.batch_size_2d,
                         "nr_classes": default_tomo_parameters.nr_classes_2d,
@@ -172,7 +170,7 @@ def _register_picked_tomogram_use_diameter(message: dict, murfey_db: Session):
                     "particle_diameter": particle_diameter,
                     "kv": tomo_params.voltage,
                     "node_creator_queue": machine_config.node_creator_queue,
-                    "session_id": message["session_id"],
+                    "session_id": session_id,
                     "autoproc_program_id": _app_id(pj_id, murfey_db),
                     "batch_size": default_tomo_parameters.batch_size_2d,
                     "nr_classes": default_tomo_parameters.nr_classes_2d,
diff --git a/tests/workflows/test_register_atlas_update.py b/tests/workflows/test_register_atlas_update.py
index f5cd16295..671cf4cbb 100644
--- a/tests/workflows/test_register_atlas_update.py
+++ b/tests/workflows/test_register_atlas_update.py
@@ -14,6 +14,7 @@ def test_run(
     mock_murfey_db = MagicMock()
     message = {
         "register": "atlas_update",
+        "session_id": 1,
         "atlas_id": mock.sentinel,
         "atlas": mock.sentinel,
         "atlas_pixel_size": mock.sentinel,
diff --git a/tests/workflows/tomo/test_tomo_picking.py b/tests/workflows/tomo/test_tomo_picking.py
index 5e3dbd434..f4d1d899d 100644
--- a/tests/workflows/tomo/test_tomo_picking.py
+++ b/tests/workflows/tomo/test_tomo_picking.py
@@ -120,6 +120,7 @@ def test_picked_tomogram_not_run_class2d(
     set_up_picking_db(murfey_db_session)
 
     message = {
+        "session_id": 1,
         "program_id": 0,
         "cbox_3d": f"{tmp_path}/AutoPick/job007/CBOX_3d/sample.cbox",
         "particle_count": 2,