diff --git a/activitysim/abm/models/disaggregate_accessibility.py b/activitysim/abm/models/disaggregate_accessibility.py
index df229c6ad7..aa2703e19c 100644
--- a/activitysim/abm/models/disaggregate_accessibility.py
+++ b/activitysim/abm/models/disaggregate_accessibility.py
@@ -622,9 +622,11 @@ def create_proto_pop(self):
 
         # Create ID columns, defaults to "%tablename%_id"
         hhid, perid, tourid = (
-            self.params[x]["index_col"]
-            if len(self.params[x]["index_col"]) > 0
-            else x + "_id"
+            (
+                self.params[x]["index_col"]
+                if len(self.params[x]["index_col"]) > 0
+                else x + "_id"
+            )
             for x in klist
         )
 
diff --git a/activitysim/abm/models/initialize_tours.py b/activitysim/abm/models/initialize_tours.py
index da69e8d227..79b0263ced 100644
--- a/activitysim/abm/models/initialize_tours.py
+++ b/activitysim/abm/models/initialize_tours.py
@@ -10,6 +10,7 @@
 from activitysim.core import expressions, tracing, workflow
 from activitysim.core.configuration import PydanticReadable
 from activitysim.core.configuration.base import PreprocessorSettings
+from activitysim.core.exceptions import InputTableError
 from activitysim.core.input import read_input_table
 
 logger = logging.getLogger(__name__)
@@ -140,7 +141,7 @@ def initialize_tours(
             f"{tours_without_persons.sum()} tours out of {len(persons)} without persons\n"
             f"{pd.Series({'person_id': tours_without_persons.index.values})}"
         )
-        raise RuntimeError(f"{tours_without_persons.sum()} tours with bad person_id")
+        raise InputTableError(f"{tours_without_persons.sum()} tours with bad person_id")
 
     if trace_hh_id:
         state.tracing.trace_df(tours, label="initialize_tours", warn_if_empty=True)
diff --git a/activitysim/abm/models/input_checker.py b/activitysim/abm/models/input_checker.py
index 568da851e1..68274ba694 100644
--- a/activitysim/abm/models/input_checker.py
+++ b/activitysim/abm/models/input_checker.py
@@ -13,6 +13,7 @@
 
 from activitysim.core import workflow
 from activitysim.core.input import read_input_table
+from activitysim.core.exceptions import ModelConfigurationError
 
 logger = logging.getLogger(__name__)
 file_logger = logger.getChild("logfile")
@@ -468,6 +469,6 @@ def input_checker(state: workflow.State):
 
     if input_check_failure:
         logger.error("Run is killed due to input checker failure!!")
-        raise RuntimeError(
+        raise ModelConfigurationError(
             "Encountered error in input checker, see input_checker.log for details"
         )
diff --git a/activitysim/abm/models/joint_tour_participation.py b/activitysim/abm/models/joint_tour_participation.py
index 47bc2b8ff9..e0d4d37fb3 100644
--- a/activitysim/abm/models/joint_tour_participation.py
+++ b/activitysim/abm/models/joint_tour_participation.py
@@ -21,6 +21,7 @@
 from activitysim.core.configuration.base import ComputeSettings, PreprocessorSettings
 from activitysim.core.configuration.logit import LogitComponentSettings
 from activitysim.core.util import assign_in_place, reindex
+from activitysim.core.exceptions import InvalidTravelError
 
 logger = logging.getLogger(__name__)
 
@@ -218,11 +219,11 @@ def participants_chooser(
                 non_choice_col = [col for col in probs.columns if col != choice_col][0]
                 probs[non_choice_col] = 1 - probs[choice_col]
                 if iter > MAX_ITERATIONS + 1:
-                    raise RuntimeError(
+                    raise InvalidTravelError(
                         f"{num_tours_remaining} tours could not be satisfied even with forcing participation"
                     )
             else:
-                raise RuntimeError(
+                raise InvalidTravelError(
                     f"{num_tours_remaining} tours could not be satisfied after {iter} iterations"
                 )
 
diff --git a/activitysim/abm/models/location_choice.py b/activitysim/abm/models/location_choice.py
index e21653cf92..7f032a8ae6 100644
--- a/activitysim/abm/models/location_choice.py
+++ b/activitysim/abm/models/location_choice.py
@@ -18,6 +18,7 @@
 from activitysim.core.interaction_sample import interaction_sample
 from activitysim.core.interaction_sample_simulate import interaction_sample_simulate
 from activitysim.core.util import reindex
+from activitysim.core.exceptions import DuplicateWorkflowTableError
 
 """
 The school/workplace location model predicts the zones in which various people will
@@ -1125,7 +1126,7 @@ def iterate_location_choice(
         assert len(save_sample_df.index.get_level_values(0).unique()) == len(choices_df)
         # lest they try to put school and workplace samples into the same table
         if state.is_table(sample_table_name):
-            raise RuntimeError(
+            raise DuplicateWorkflowTableError(
                 "dest choice sample table %s already exists" % sample_table_name
             )
         state.extend_table(sample_table_name, save_sample_df)
diff --git a/activitysim/abm/models/parking_location_choice.py b/activitysim/abm/models/parking_location_choice.py
index 075bf6174d..32f3aabee2 100644
--- a/activitysim/abm/models/parking_location_choice.py
+++ b/activitysim/abm/models/parking_location_choice.py
@@ -23,6 +23,7 @@
 from activitysim.core.interaction_sample_simulate import interaction_sample_simulate
 from activitysim.core.tracing import print_elapsed_time
 from activitysim.core.util import assign_in_place, drop_unused_columns
+from activitysim.core.exceptions import DuplicateWorkflowTableError
 
 logger = logging.getLogger(__name__)
 
@@ -500,7 +501,9 @@ def parking_location(
 
         # lest they try to put tour samples into the same table
         if state.is_table(sample_table_name):
-            raise RuntimeError("sample table %s already exists" % sample_table_name)
+            raise DuplicateWorkflowTableError(
+                "sample table %s already exists" % sample_table_name
+            )
         state.extend_table(sample_table_name, save_sample_df)
 
     expressions.annotate_tables(
diff --git a/activitysim/abm/models/settings_checker.py b/activitysim/abm/models/settings_checker.py
index 907e7e9995..77be268b13 100644
--- a/activitysim/abm/models/settings_checker.py
+++ b/activitysim/abm/models/settings_checker.py
@@ -21,6 +21,7 @@
     eval_nest_coefficients,
     read_model_coefficient_template,
 )
+from activitysim.core.exceptions import ModelConfigurationError
 
 # import model settings
 from activitysim.abm.models.accessibility import AccessibilitySettings
@@ -760,7 +761,7 @@ def check_model_settings(
         for e in all_errors:
             logger.error(f"\t{str(e)}")
             file_logger.error(f"\t{str(e)}")
-        raise RuntimeError(
+        raise ModelConfigurationError(
             f"Encountered one or more errors in settings checker. See f{log_file} for details."
         )
     msg = f"Setting Checker Complete. No runtime errors were raised. Check f{log_file} for warnings. These *may* prevent model from successfully running."
diff --git a/activitysim/abm/models/trip_departure_choice.py b/activitysim/abm/models/trip_departure_choice.py
index 0e4dd05d9d..7b34f8e742 100644
--- a/activitysim/abm/models/trip_departure_choice.py
+++ b/activitysim/abm/models/trip_departure_choice.py
@@ -27,6 +27,7 @@
 from activitysim.core.skim_dataset import SkimDataset
 from activitysim.core.skim_dictionary import SkimDict
 from activitysim.core.util import reindex
+from activitysim.core.exceptions import SegmentedSpecificationError
 
 logger = logging.getLogger(__name__)
 
@@ -219,7 +220,7 @@ def choose_tour_leg_pattern(
         )
 
     if len(spec.columns) > 1:
-        raise RuntimeError("spec must have only one column")
+        raise SegmentedSpecificationError("spec must have only one column")
 
     # - join choosers and alts
     # in vanilla interaction_simulate interaction_df is cross join of choosers and alternatives
diff --git a/activitysim/abm/models/trip_destination.py b/activitysim/abm/models/trip_destination.py
index 6584134efb..8cdf8c3692 100644
--- a/activitysim/abm/models/trip_destination.py
+++ b/activitysim/abm/models/trip_destination.py
@@ -35,6 +35,7 @@
 from activitysim.core.skim_dictionary import DataFrameMatrix
 from activitysim.core.tracing import print_elapsed_time
 from activitysim.core.util import assign_in_place, reindex
+from activitysim.core.exceptions import InvalidTravelError, DuplicateWorkflowTableError
 
 logger = logging.getLogger(__name__)
 
@@ -1664,7 +1665,7 @@ def trip_destination(
     # testing feature t0 make sure at least one trip fails so trip_purpose_and_destination model is run
     if state.settings.testing_fail_trip_destination and not trips_df.failed.any():
         if (trips_df.trip_num < trips_df.trip_count).sum() == 0:
-            raise RuntimeError(
+            raise InvalidTravelError(
                 "can't honor 'testing_fail_trip_destination' setting because no intermediate trips"
             )
 
@@ -1745,7 +1746,9 @@ def trip_destination(
 
         # lest they try to put tour samples into the same table
         if state.is_table(sample_table_name):
-            raise RuntimeError("sample table %s already exists" % sample_table_name)
+            raise DuplicateWorkflowTableError(
+                "sample table %s already exists" % sample_table_name
+            )
         state.extend_table(sample_table_name, save_sample_df)
 
     expressions.annotate_tables(
diff --git a/activitysim/abm/models/trip_purpose.py b/activitysim/abm/models/trip_purpose.py
index 695882938d..73ccaaf8c0 100644
--- a/activitysim/abm/models/trip_purpose.py
+++ b/activitysim/abm/models/trip_purpose.py
@@ -22,6 +22,7 @@
 )
 from activitysim.core.configuration.base import PreprocessorSettings, PydanticReadable
 from activitysim.core.util import reindex
+from activitysim.core.exceptions import InvalidTravelError
 
 logger = logging.getLogger(__name__)
 
@@ -134,7 +135,7 @@ def choose_intermediate_trip_purpose(
             state.tracing.write_csv(
                 unmatched_choosers, file_name=file_name, transpose=False
             )
-            raise RuntimeError(
+            raise InvalidTravelError(
                 "Some trips could not be matched to probs based on join columns %s."
                 % probs_join_cols
             )
diff --git a/activitysim/abm/models/trip_scheduling.py b/activitysim/abm/models/trip_scheduling.py
index 0e45d463dc..0ce4f2ffff 100644
--- a/activitysim/abm/models/trip_scheduling.py
+++ b/activitysim/abm/models/trip_scheduling.py
@@ -18,6 +18,7 @@
 from activitysim.core import chunk, config, estimation, expressions, tracing, workflow
 from activitysim.core.configuration.base import PreprocessorSettings, PydanticReadable
 from activitysim.core.util import reindex
+from activitysim.core.exceptions import InvalidTravelError, PipelineError
 
 logger = logging.getLogger(__name__)
 
@@ -584,9 +585,11 @@ def trip_scheduling(
         i = 0
         while (i < max_iterations) and not trips_chunk.empty:
             # only chunk log first iteration since memory use declines with each iteration
-            with chunk.chunk_log(
-                state, trace_label
-            ) if i == 0 else chunk.chunk_log_skip():
+            with (
+                chunk.chunk_log(state, trace_label)
+                if i == 0
+                else chunk.chunk_log_skip()
+            ):
                 i += 1
                 is_last_iteration = i == max_iterations
 
@@ -615,7 +618,9 @@ def trip_scheduling(
                 logger.info("%s %s failed", trace_label_i, failed.sum())
 
                 if (failed.sum() > 0) & (model_settings.scheduling_mode == "relative"):
-                    raise RuntimeError("failed trips with relative scheduling mode")
+                    raise InvalidTravelError(
+                        "failed trips with relative scheduling mode"
+                    )
 
                 if not is_last_iteration:
                     # boolean series of trips whose leg scheduling failed
@@ -653,7 +658,7 @@ def trip_scheduling(
         )
 
         if failfix != FAILFIX_DROP_AND_CLEANUP:
-            raise RuntimeError(
+            raise PipelineError(
                 "%s setting '%s' not enabled in settings"
                 % (FAILFIX, FAILFIX_DROP_AND_CLEANUP)
             )
diff --git a/activitysim/abm/models/util/cdap.py b/activitysim/abm/models/util/cdap.py
index 3b4c41466e..21f42de827 100644
--- a/activitysim/abm/models/util/cdap.py
+++ b/activitysim/abm/models/util/cdap.py
@@ -10,6 +10,7 @@
 
 from activitysim.core import chunk, logit, simulate, tracing, workflow
 from activitysim.core.configuration.base import ComputeSettings
+from activitysim.core.exceptions import ModelConfigurationError
 
 logger = logging.getLogger(__name__)
 
@@ -48,7 +49,7 @@ def add_pn(col, pnum):
     elif isinstance(col, (list, tuple)):
         return [c if c == _hh_id_ else "%s_p%s" % (c, pnum) for c in col]
     else:
-        raise RuntimeError("add_pn col not list or str")
+        raise TypeError("add_pn col not list or str")
 
 
 def assign_cdap_rank(
@@ -270,7 +271,7 @@ def preprocess_interaction_coefficients(interaction_coefficients):
             "Error in cdap_interaction_coefficients at row %s. Expect only M, N, or H!"
             % coefficients[~coefficients["activity"].isin(["M", "N", "H"])].index.values
         )
-        raise RuntimeError(msg)
+        raise ModelConfigurationError(msg)
 
     coefficients["cardinality"] = (
         coefficients["interaction_ptypes"].astype(str).str.len()
@@ -470,8 +471,9 @@ def build_cdap_spec(
             continue
 
         if not (0 <= row.cardinality <= MAX_INTERACTION_CARDINALITY):
-            raise RuntimeError(
-                "Bad row cardinality %d for %s" % (row.cardinality, row.slug)
+            raise ModelConfigurationError(
+                "Bad row cardinality %d for %s. Try checking that all interaction terms include 3 or fewer person types."
+                % (row.cardinality, row.slug)
             )
 
         # for all other interaction rules, we need to generate a row in the spec for each
diff --git a/activitysim/abm/models/util/probabilistic_scheduling.py b/activitysim/abm/models/util/probabilistic_scheduling.py
index cdaf64da5a..193a1b703d 100644
--- a/activitysim/abm/models/util/probabilistic_scheduling.py
+++ b/activitysim/abm/models/util/probabilistic_scheduling.py
@@ -8,6 +8,7 @@
 import pandas as pd
 
 from activitysim.core import chunk, logit, tracing, workflow
+from activitysim.core.exceptions import InvalidTravelError
 
 logger = logging.getLogger(__name__)
 
@@ -210,7 +211,7 @@ def _postprocess_scheduling_choices(
 
     if scheduling_mode == "relative":
         if failed.any():
-            RuntimeError(
+            InvalidTravelError(
                 f"Failed trips in realtive mode for {failed.sum()} trips: {choosers[failed]}"
             )
 
diff --git a/activitysim/abm/models/util/vectorize_tour_scheduling.py b/activitysim/abm/models/util/vectorize_tour_scheduling.py
index d4593c21fa..0ec73c3662 100644
--- a/activitysim/abm/models/util/vectorize_tour_scheduling.py
+++ b/activitysim/abm/models/util/vectorize_tour_scheduling.py
@@ -82,7 +82,7 @@ def skims_for_logsums(
     elif isinstance(destination_for_tour_purpose, dict):
         dest_col_name = destination_for_tour_purpose.get(tour_purpose)
     else:
-        raise RuntimeError(
+        raise TypeError(
             f"expected string or dict DESTINATION_FOR_TOUR_PURPOSE model_setting for {tour_purpose}"
         )
 
diff --git a/activitysim/abm/tables/households.py b/activitysim/abm/tables/households.py
index 9f121e7082..c0c33dcbcf 100644
--- a/activitysim/abm/tables/households.py
+++ b/activitysim/abm/tables/households.py
@@ -11,6 +11,7 @@
 from activitysim.abm.tables.util import simple_table_join
 from activitysim.core import tracing, workflow
 from activitysim.core.input import read_input_table
+from activitysim.core.exceptions import MissingInputTableDefinition
 
 logger = logging.getLogger(__name__)
 
@@ -45,7 +46,7 @@ def households(state: workflow.State) -> pd.DataFrame:
             )
 
         if df.shape[0] == 0:
-            raise RuntimeError("No override households found in store")
+            raise MissingInputTableDefinition("No override households found in store")
 
     # if we are tracing hh exclusively
     elif _trace_hh_id and households_sample_size == 1:
diff --git a/activitysim/abm/tables/persons.py b/activitysim/abm/tables/persons.py
index d5ab67fb57..2c7716ddcc 100644
--- a/activitysim/abm/tables/persons.py
+++ b/activitysim/abm/tables/persons.py
@@ -9,6 +9,7 @@
 
 from activitysim.abm.tables.util import simple_table_join
 from activitysim.core import workflow
+from activitysim.core.exceptions import InputTableError
 from activitysim.core.input import read_input_table
 
 logger = logging.getLogger(__name__)
@@ -55,7 +56,7 @@ def persons(state: workflow.State) -> pd.DataFrame:
             f"{persons_without_households.sum()} persons out of {len(df)} without households\n"
             f"{pd.Series({'person_id': persons_without_households.index.values})}"
         )
-        raise RuntimeError(
+        raise InputTableError(
             f"{persons_without_households.sum()} persons with bad household_id"
         )
 
@@ -67,7 +68,7 @@ def persons(state: workflow.State) -> pd.DataFrame:
             f"{households_without_persons.sum()} households out of {len(households.index)} without  persons\n"
             f"{pd.Series({'household_id': households_without_persons.index.values})}"
         )
-        raise RuntimeError(
+        raise InputTableError(
             f"{households_without_persons.sum()} households with no persons"
         )
 
@@ -107,5 +108,5 @@ def persons_merged(
             left_on="person_id",
         )
     if n_persons != len(persons):
-        raise RuntimeError("number of persons changed")
+        raise InputTableError("number of persons changed")
     return persons
diff --git a/activitysim/abm/tables/shadow_pricing.py b/activitysim/abm/tables/shadow_pricing.py
index 586924efc9..fa2832181c 100644
--- a/activitysim/abm/tables/shadow_pricing.py
+++ b/activitysim/abm/tables/shadow_pricing.py
@@ -18,6 +18,7 @@
 from activitysim.core.configuration import PydanticReadable
 from activitysim.core.configuration.logit import TourLocationComponentSettings
 from activitysim.core.input import read_input_table
+from activitysim.core.exceptions import SystemConfigurationError, MissingNameError
 
 logger = logging.getLogger(__name__)
 
@@ -181,7 +182,7 @@ def __init__(
             logger.warning(
                 "deprecated combination of multiprocessing and not fail_fast"
             )
-            raise RuntimeError(
+            raise SystemConfigurationError(
                 "Shadow pricing requires fail_fast setting in multiprocessing mode"
             )
 
@@ -904,7 +905,10 @@ def update_shadow_prices(self, state):
             self.sampled_persons = sampled_persons
 
         else:
-            raise RuntimeError("unknown SHADOW_PRICE_METHOD %s" % shadow_price_method)
+            raise SystemConfigurationError(
+                "unknown SHADOW_PRICE_METHOD %s, method must be one of 'ctramp', 'daysim', or 'simulation'"
+                % shadow_price_method
+            )
 
     def dest_size_terms(self, segment):
         assert segment in self.segment_ids
@@ -922,8 +926,9 @@ def dest_size_terms(self, segment):
             elif shadow_price_method == "simulation":
                 utility_adjustment = self.shadow_prices[segment]
             else:
-                raise RuntimeError(
-                    "unknown SHADOW_PRICE_METHOD %s" % shadow_price_method
+                raise SystemConfigurationError(
+                    "unknown SHADOW_PRICE_METHOD %s, method must be one of 'ctramp', 'daysim', or 'simulation'"
+                    % shadow_price_method
                 )
 
         size_terms = pd.DataFrame(
@@ -1036,9 +1041,7 @@ def buffers_for_shadow_pricing(shadow_pricing_info):
         if np.issubdtype(dtype, np.int64):
             typecode = ctypes.c_int64
         else:
-            raise RuntimeError(
-                "buffer_for_shadow_pricing unrecognized dtype %s" % dtype
-            )
+            raise TypeError("buffer_for_shadow_pricing unrecognized dtype %s" % dtype)
 
         shared_data_buffer = multiprocessing.Array(typecode, buffer_size)
 
@@ -1085,9 +1088,7 @@ def buffers_for_shadow_pricing_choice(state, shadow_pricing_choice_info):
         if np.issubdtype(dtype, np.int64):
             typecode = ctypes.c_int64
         else:
-            raise RuntimeError(
-                "buffer_for_shadow_pricing unrecognized dtype %s" % dtype
-            )
+            raise TypeError("buffer_for_shadow_pricing unrecognized dtype %s" % dtype)
 
         shared_data_buffer = multiprocessing.Array(typecode, buffer_size)
 
@@ -1145,12 +1146,12 @@ def shadow_price_data_from_buffers_choice(
     block_shapes = shadow_pricing_info["block_shapes"]
 
     if model_selector not in block_shapes:
-        raise RuntimeError(
+        raise MissingNameError(
             "Model selector %s not in shadow_pricing_info" % model_selector
         )
 
     if block_name(model_selector + "_choice") not in data_buffers:
-        raise RuntimeError(
+        raise MissingNameError(
             "Block %s not in data_buffers" % block_name(model_selector + "_choice")
         )
 
@@ -1195,12 +1196,14 @@ def shadow_price_data_from_buffers(data_buffers, shadow_pricing_info, model_sele
     block_shapes = shadow_pricing_info["block_shapes"]
 
     if model_selector not in block_shapes:
-        raise RuntimeError(
+        raise MissingNameError(
             "Model selector %s not in shadow_pricing_info" % model_selector
         )
 
     if block_name(model_selector) not in data_buffers:
-        raise RuntimeError("Block %s not in data_buffers" % block_name(model_selector))
+        raise MissingNameError(
+            "Block %s not in data_buffers" % block_name(model_selector)
+        )
 
     shape = block_shapes[model_selector]
     data = data_buffers[block_name(model_selector)]
diff --git a/activitysim/abm/test/test_pipeline/test_pipeline.py b/activitysim/abm/test/test_pipeline/test_pipeline.py
index 70bc26f4b1..b245e97f5e 100644
--- a/activitysim/abm/test/test_pipeline/test_pipeline.py
+++ b/activitysim/abm/test/test_pipeline/test_pipeline.py
@@ -13,6 +13,10 @@
 import pytest
 
 from activitysim.core import random, tracing, workflow
+from activitysim.core.exceptions import (
+    CheckpointNameNotFoundError,
+    DuplicateWorkflowNameError,
+)
 
 # set the max households for all tests (this is to limit memory use on travis)
 HOUSEHOLDS_SAMPLE_SIZE = 50
@@ -190,12 +194,12 @@ def test_mini_pipeline_run():
     regress_mini_location_choice_logsums(state)
 
     # try to get a non-existant table
-    with pytest.raises(RuntimeError) as excinfo:
+    with pytest.raises(CheckpointNameNotFoundError) as excinfo:
         state.checkpoint.load_dataframe("bogus")
     assert "never checkpointed" in str(excinfo.value)
 
     # try to get an existing table from a non-existant checkpoint
-    with pytest.raises(RuntimeError) as excinfo:
+    with pytest.raises(CheckpointNameNotFoundError) as excinfo:
         state.checkpoint.load_dataframe("households", checkpoint_name="bogus")
     assert "not in checkpoints" in str(excinfo.value)
 
@@ -235,7 +239,7 @@ def test_mini_pipeline_run2():
     regress_mini_auto(state)
 
     # try to run a model already in pipeline
-    with pytest.raises(RuntimeError) as excinfo:
+    with pytest.raises(DuplicateWorkflowNameError) as excinfo:
         state.run.by_name("auto_ownership_simulate")
     assert "run model 'auto_ownership_simulate' more than once" in str(excinfo.value)
 
diff --git a/activitysim/core/chunk.py b/activitysim/core/chunk.py
index 7f09187f1f..32d00fd15c 100644
--- a/activitysim/core/chunk.py
+++ b/activitysim/core/chunk.py
@@ -1078,7 +1078,7 @@ def ledger(self):
             if mem_monitor is not None:
                 if not mem_monitor.is_alive():
                     logger.error(f"mem_monitor for {self.trace_label} died!")
-                    raise RuntimeError("bug")
+                    raise RuntimeError("mem_monitor for {self.trace_label} died!")
 
                 if stop_snooping is not None:
                     stop_snooping.set()
diff --git a/activitysim/core/config.py b/activitysim/core/config.py
index 730c383157..e395769160 100644
--- a/activitysim/core/config.py
+++ b/activitysim/core/config.py
@@ -7,6 +7,7 @@
 from activitysim.core import workflow
 from activitysim.core.configuration.base import PydanticBase
 from activitysim.core.configuration.logit import LogitComponentSettings
+from activitysim.core.exceptions import ModelConfigurationError
 
 # ActivitySim
 # See full license in LICENSE.txt.
@@ -100,7 +101,7 @@ def get_model_constants(model_settings):
 
 
 def get_logit_model_settings(
-    model_settings: LogitComponentSettings | dict[str, Any] | None
+    model_settings: LogitComponentSettings | dict[str, Any] | None,
 ):
     """
     Read nest spec (for nested logit) from model settings file
@@ -123,13 +124,18 @@ def get_logit_model_settings(
 
         if logit_type not in ["NL", "MNL"]:
             logger.error("Unrecognized logit type '%s'" % logit_type)
-            raise RuntimeError("Unrecognized logit type '%s'" % logit_type)
+            raise ModelConfigurationError(
+                "Unrecognized logit type '%s'. Logit type must be 'NL' for nested logit or 'MNL' for multinomial logit"
+                % logit_type
+            )
 
         if logit_type == "NL":
             nests = model_settings.get("NESTS", None)
             if nests is None:
                 logger.error("No NEST found in model spec for NL model type")
-                raise RuntimeError("No NEST found in model spec for NL model type")
+                raise ModelConfigurationError(
+                    "No NEST found in model spec for NL model type"
+                )
 
     return nests
 
diff --git a/activitysim/core/configuration/filesystem.py b/activitysim/core/configuration/filesystem.py
index ba2049e31c..75bd761365 100644
--- a/activitysim/core/configuration/filesystem.py
+++ b/activitysim/core/configuration/filesystem.py
@@ -16,7 +16,10 @@
 
 from activitysim.core.configuration.base import PydanticBase
 from activitysim.core.configuration.logit import LogitComponentSettings
-from activitysim.core.exceptions import SettingsFileNotFoundError
+from activitysim.core.exceptions import (
+    SettingsFileNotFoundError,
+    SystemConfigurationError,
+)
 from activitysim.core.util import parse_suffix_args, suffix_tables_in_settings
 
 logger = logging.getLogger(__name__)
@@ -768,7 +771,7 @@ def backfill_settings(settings, backfill):
                         logger.error(
                             f"Unexpected additional settings: {additional_settings}"
                         )
-                        raise RuntimeError(
+                        raise SystemConfigurationError(
                             "'include_settings' must appear alone in settings file."
                         )
 
diff --git a/activitysim/core/estimation.py b/activitysim/core/estimation.py
index 5b29de6960..bbbe376ee4 100644
--- a/activitysim/core/estimation.py
+++ b/activitysim/core/estimation.py
@@ -16,6 +16,11 @@
 from activitysim.core.configuration.base import PydanticBase
 from activitysim.core.util import reindex
 from activitysim.core.yaml_tools import safe_dump
+from activitysim.core.exceptions import (
+    DuplicateWorkflowTableError,
+    DuplicateLoadableObjectError,
+    EstimationDataError,
+)
 
 logger = logging.getLogger("estimation")
 
@@ -433,7 +438,7 @@ def write_table(
 
         def cache_table(df, table_name, append):
             if table_name in self.tables and not append:
-                raise RuntimeError(
+                raise DuplicateWorkflowTableError(
                     "cache_table %s append=False and table exists" % (table_name,)
                 )
             if table_name in self.tables:
@@ -450,7 +455,7 @@ def write_table(df, table_name, index, append, bundle_directory, filetype):
             # check if file exists
             file_exists = os.path.isfile(file_path)
             if file_exists and not append:
-                raise RuntimeError(
+                raise DuplicateLoadableObjectError(
                     "write_table %s append=False and file exists: %s"
                     % (table_name, file_path)
                 )
@@ -470,7 +475,7 @@ def write_table(df, table_name, index, append, bundle_directory, filetype):
             elif filetype == "pkl":
                 self.write_pickle(df, file_path, index, append)
             else:
-                raise RuntimeError(
+                raise IOError(
                     f"Unsupported filetype: {filetype}, allowed options are csv, parquet, pkl"
                 )
 
@@ -536,7 +541,7 @@ def write_omnibus_table(self):
                 elif "household_id" in df.columns:
                     df.set_index("household_id", inplace=True)
                 else:
-                    RuntimeError(
+                    EstimationDataError(
                         f"No index column found in omnibus table {omnibus_table}: {df}"
                     )
 
@@ -564,7 +569,7 @@ def write_omnibus_table(self):
                 self.write_pickle(df, file_path, index=True, append=False)
 
             else:
-                raise RuntimeError(f"Unsupported filetype: {filetype}")
+                raise IOError(f"Unsupported filetype: {filetype}")
 
             self.debug("wrote_omnibus_choosers: %s" % file_path)
 
@@ -945,7 +950,7 @@ def get_survey_values(self, model_values, table_name, column_names):
                 % (missing_columns, table_name)
             )
             print("survey table columns: %s" % (survey_df.columns,))
-            raise RuntimeError(
+            raise EstimationDataError(
                 "missing columns (%s) in survey table %s"
                 % (missing_columns, table_name)
             )
@@ -998,7 +1003,7 @@ def get_survey_values(self, model_values, table_name, column_names):
                 logger.error(
                     "couldn't get_survey_values for %s in %s\n" % (c, table_name)
                 )
-                raise RuntimeError(
+                raise EstimationDataError(
                     "couldn't get_survey_values for %s in %s\n" % (c, table_name)
                 )
 
diff --git a/activitysim/core/exceptions.py b/activitysim/core/exceptions.py
index 29d8f03a1a..878ed6cfc9 100644
--- a/activitysim/core/exceptions.py
+++ b/activitysim/core/exceptions.py
@@ -56,3 +56,39 @@ class ReadOnlyError(IOError):
 
 class MissingInputTableDefinition(RuntimeError):
     """An input table definition was expected but not found."""
+
+
+class SystemConfigurationError(RuntimeError):
+    """An error in the system configuration (possibly in settings.yaml) was found."""
+
+
+class ModelConfigurationError(RuntimeError):
+    """An error in the model configuration was found."""
+
+
+class InvalidTravelError(RuntimeError):
+    """Travel behavior could not be completed in a valid way."""
+
+
+class TableSlicingError(RuntimeError):
+    """An error occurred trying to slice a table."""
+
+
+class InputTableError(RuntimeError):
+    """An issue with the input population was found."""
+
+
+class SubprocessError(RuntimeError):
+    """An error occurred in a subprocess."""
+
+
+class SegmentedSpecificationError(RuntimeError):
+    """An error was caused by creating an invalid spec table for a segmented model component."""
+
+
+class TableIndexError(RuntimeError):
+    """An error related to the index of a table in the pipeline."""
+
+
+class EstimationDataError(RuntimeError):
+    """An error related to estimation data."""
diff --git a/activitysim/core/input.py b/activitysim/core/input.py
index 734679996a..ecac76b07b 100644
--- a/activitysim/core/input.py
+++ b/activitysim/core/input.py
@@ -10,7 +10,10 @@
 
 from activitysim.core import util, workflow
 from activitysim.core.configuration import InputTable
-from activitysim.core.exceptions import MissingInputTableDefinition
+from activitysim.core.exceptions import (
+    MissingInputTableDefinition,
+    ModelConfigurationError,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -204,7 +207,9 @@ def read_from_table_info(table_info: InputTable, state):
                 f"index_col '{index_col}' specified in configs but not in {tablename} table!"
             )
             logger.error(f"{tablename} columns are: {list(df.columns)}")
-            raise RuntimeError(f"index_col '{index_col}' not in {tablename} table!")
+            raise ModelConfigurationError(
+                f"index_col '{index_col}' not in {tablename} table!"
+            )
 
     if keep_columns:
         logger.debug("keeping columns: %s" % keep_columns)
@@ -214,7 +219,9 @@ def read_from_table_info(table_info: InputTable, state):
                 f"{list(set(keep_columns).difference(set(df.columns)))}"
             )
             logger.error(f"{tablename} table has columns: {list(df.columns)}")
-            raise RuntimeError(f"Required columns missing from {tablename} table")
+            raise ModelConfigurationError(
+                f"Required columns missing from {tablename} table"
+            )
 
         df = df[keep_columns]
 
diff --git a/activitysim/core/interaction_sample.py b/activitysim/core/interaction_sample.py
index 8db19a416d..52116638ba 100644
--- a/activitysim/core/interaction_sample.py
+++ b/activitysim/core/interaction_sample.py
@@ -20,6 +20,7 @@
 from activitysim.core.configuration.base import ComputeSettings
 from activitysim.core.skim_dataset import DatasetWrapper
 from activitysim.core.skim_dictionary import SkimWrapper
+from activitysim.core.exceptions import SegmentedSpecificationError
 
 logger = logging.getLogger(__name__)
 
@@ -219,7 +220,7 @@ def _interaction_sample(
         )
 
     if len(spec.columns) > 1:
-        raise RuntimeError("spec must have only one column")
+        raise SegmentedSpecificationError("spec must have only one column")
 
     # if using skims, copy index into the dataframe, so it will be
     # available as the "destination" for set_skim_wrapper_targets
diff --git a/activitysim/core/interaction_sample_simulate.py b/activitysim/core/interaction_sample_simulate.py
index 9cdea3292d..7ba9e66ea2 100644
--- a/activitysim/core/interaction_sample_simulate.py
+++ b/activitysim/core/interaction_sample_simulate.py
@@ -10,6 +10,7 @@
 from activitysim.core import chunk, interaction_simulate, logit, tracing, util, workflow
 from activitysim.core.configuration.base import ComputeSettings
 from activitysim.core.simulate import set_skim_wrapper_targets
+from activitysim.core.exceptions import SegmentedSpecificationError
 
 logger = logging.getLogger(__name__)
 
@@ -115,7 +116,7 @@ def _interaction_sample_simulate(
         )
 
     if len(spec.columns) > 1:
-        raise RuntimeError("spec must have only one column")
+        raise SegmentedSpecificationError("spec must have only one column")
 
     # if using skims, copy index into the dataframe, so it will be
     # available as the "destination" for the skims dereference below
diff --git a/activitysim/core/interaction_simulate.py b/activitysim/core/interaction_simulate.py
index d0af58e77c..6b776c6d94 100644
--- a/activitysim/core/interaction_simulate.py
+++ b/activitysim/core/interaction_simulate.py
@@ -16,6 +16,7 @@
 from activitysim.core import chunk, logit, simulate, timing, tracing, util, workflow
 from activitysim.core.configuration.base import ComputeSettings
 from activitysim.core.fast_eval import fast_eval
+from activitysim.core.exceptions import SegmentedSpecificationError
 
 logger = logging.getLogger(__name__)
 
@@ -722,7 +723,7 @@ def _interaction_simulate(
         )
 
     if len(spec.columns) > 1:
-        raise RuntimeError("spec must have only one column")
+        raise SegmentedSpecificationError("spec must have only one column")
 
     sample_size = sample_size or len(alternatives)
 
diff --git a/activitysim/core/logit.py b/activitysim/core/logit.py
index 034d970ca6..105e18fecc 100644
--- a/activitysim/core/logit.py
+++ b/activitysim/core/logit.py
@@ -11,6 +11,11 @@
 from activitysim.core import tracing, workflow
 from activitysim.core.choosing import choice_maker
 from activitysim.core.configuration.logit import LogitNestSpec
+from activitysim.core.exceptions import (
+    InvalidTravelError,
+    ModelConfigurationError,
+    TableIndexError,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -83,7 +88,7 @@ def report_bad_choices(
         logger.warning(row_msg)
 
     if raise_error:
-        raise RuntimeError(msg_with_count)
+        raise InvalidTravelError(msg_with_count)
 
 
 def utils_to_logsums(utils, exponentiated=False, allow_zero_probs=False):
@@ -360,11 +365,11 @@ def interaction_dataset(
 
     """
     if not choosers.index.is_unique:
-        raise RuntimeError(
+        raise TableIndexError(
             "ERROR: choosers index is not unique, " "sample will not work correctly"
         )
     if not alternatives.index.is_unique:
-        raise RuntimeError(
+        raise TableIndexError(
             "ERROR: alternatives index is not unique, " "sample will not work correctly"
         )
 
@@ -469,7 +474,7 @@ def validate_nest_spec(nest_spec: dict | LogitNestSpec, trace_label: str):
         # nest.print()
 
     if duplicates:
-        raise RuntimeError(
+        raise ModelConfigurationError(
             f"validate_nest_spec:duplicate nest key/s '{duplicates}' in nest spec - {trace_label}"
         )
 
@@ -568,7 +573,9 @@ def each_nest(nest_spec: dict | LogitNestSpec, type=None, post_order=False):
             Nest object with info about the current node (nest or leaf)
     """
     if type is not None and type not in Nest.nest_types():
-        raise RuntimeError("Unknown nest type '%s' in call to each_nest" % type)
+        raise ModelConfigurationError(
+            "Unknown nest type '%s' in call to each_nest" % type
+        )
 
     if isinstance(nest_spec, dict):
         nest_spec = LogitNestSpec.model_validate(nest_spec)
diff --git a/activitysim/core/mp_tasks.py b/activitysim/core/mp_tasks.py
index 7d31134bc0..42b1ab444c 100644
--- a/activitysim/core/mp_tasks.py
+++ b/activitysim/core/mp_tasks.py
@@ -26,6 +26,7 @@
     NON_TABLE_COLUMNS,
     ParquetStore,
 )
+from activitysim.core.exceptions import *
 
 logger = logging.getLogger(__name__)
 
@@ -440,7 +441,9 @@ def build_slice_rules(state: workflow.State, slice_info, pipeline_tables):
         tables[table_name] = pipeline_tables[table_name]
 
     if primary_slicer not in tables:
-        raise RuntimeError("primary slice table '%s' not in pipeline" % primary_slicer)
+        raise SystemConfigurationError(
+            "primary slice table '%s' not in pipeline" % primary_slicer
+        )
 
     # allow wildcard 'True' to avoid slicing (or coalescing) any tables no explicitly listed in slice_info.tables
     # populationsim uses slice.except wildcards to avoid listing control tables (etc) that should not be sliced,
@@ -532,7 +535,7 @@ def apportion_pipeline(state: workflow.State, sub_proc_names, step_info):
     """
     slice_info = step_info.get("slice", None)
     if slice_info is None:
-        raise RuntimeError("missing slice_info.slice")
+        raise SystemConfigurationError("missing slice_info.slice")
     multiprocess_step_name = step_info.get("name", None)
 
     pipeline_file_name = state.get_injectable("pipeline_file_name")
@@ -542,14 +545,16 @@ def apportion_pipeline(state: workflow.State, sub_proc_names, step_info):
         "last_checkpoint_in_previous_multiprocess_step", None
     )
     if last_checkpoint_in_previous_multiprocess_step is None:
-        raise RuntimeError("missing last_checkpoint_in_previous_multiprocess_step")
+        raise CheckpointNameNotFoundError(
+            "missing last_checkpoint_in_previous_multiprocess_step"
+        )
     state.checkpoint.restore(resume_after=last_checkpoint_in_previous_multiprocess_step)
 
     # ensure all tables are in the pipeline
     checkpointed_tables = state.checkpoint.list_tables()
     for table_name in slice_info["tables"]:
         if table_name not in checkpointed_tables:
-            raise RuntimeError(f"slicer table {table_name} not found in pipeline")
+            raise StateAccessError(f"slicer table {table_name} not found in pipeline")
 
     checkpoints_df = state.checkpoint.get_inventory()
 
@@ -601,7 +606,7 @@ def apportion_pipeline(state: workflow.State, sub_proc_names, step_info):
 
                     if rule["slice_by"] is not None and num_sub_procs > len(df):
                         # almost certainly a configuration error
-                        raise RuntimeError(
+                        raise SystemConfigurationError(
                             f"apportion_pipeline: multiprocess step {multiprocess_step_name} "
                             f"slice table {table_name} has fewer rows {df.shape} "
                             f"than num_processes ({num_sub_procs})."
@@ -634,7 +639,7 @@ def apportion_pipeline(state: workflow.State, sub_proc_names, step_info):
                         # don't slice mirrored tables
                         sliced_tables[table_name] = df
                     else:
-                        raise RuntimeError(
+                        raise TableSlicingError(
                             "Unrecognized slice rule '%s' for table %s"
                             % (rule["slice_by"], table_name)
                         )
@@ -678,7 +683,7 @@ def apportion_pipeline(state: workflow.State, sub_proc_names, step_info):
 
                 if rule["slice_by"] is not None and num_sub_procs > len(df):
                     # almost certainly a configuration error
-                    raise RuntimeError(
+                    raise SystemConfigurationError(
                         f"apportion_pipeline: multiprocess step {multiprocess_step_name} "
                         f"slice table {table_name} has fewer rows {df.shape} "
                         f"than num_processes ({num_sub_procs})."
@@ -711,7 +716,7 @@ def apportion_pipeline(state: workflow.State, sub_proc_names, step_info):
                     # don't slice mirrored tables
                     sliced_tables[table_name] = df
                 else:
-                    raise RuntimeError(
+                    raise TableSlicingError(
                         "Unrecognized slice rule '%s' for table %s"
                         % (rule["slice_by"], table_name)
                     )
@@ -970,7 +975,7 @@ def adjust_chunk_size_for_shared_memory(chunk_size, data_buffers, num_processes)
     )
 
     if adjusted_chunk_size <= 0:
-        raise RuntimeError(
+        raise SystemConfigurationError(
             f"adjust_chunk_size_for_shared_memory: chunk_size too small for shared memory.  "
             f"adjusted_chunk_size: {adjusted_chunk_size}"
         )
@@ -1366,7 +1371,7 @@ def check_proc_status(state: workflow.State):
                                         state,
                                         f"error terminating process {op.name}: {e}",
                                     )
-                        raise RuntimeError("Process %s failed" % (p.name,))
+                        raise SubprocessError("Process %s failed" % (p.name,))
 
     step_name = step_info["name"]
 
@@ -1531,7 +1536,7 @@ def run_sub_task(state: workflow.State, p):
 
     if p.exitcode:
         error(state, f"Process {p.name} returned exitcode {p.exitcode}")
-        raise RuntimeError("Process %s returned exitcode %s" % (p.name, p.exitcode))
+        raise SubprocessError("Process %s returned exitcode %s" % (p.name, p.exitcode))
 
 
 def drop_breadcrumb(state: workflow.State, step_name, crumb, value=True):
@@ -1596,7 +1601,7 @@ def run_multiprocess(state: workflow.State, injectables):
     run_list = get_run_list(state)
 
     if not run_list["multiprocess"]:
-        raise RuntimeError(
+        raise SubprocessError(
             "run_multiprocess called but multiprocess flag is %s"
             % run_list["multiprocess"]
         )
@@ -1719,7 +1724,7 @@ def find_breadcrumb(crumb, default=None):
             )
 
             if len(completed) != num_processes:
-                raise RuntimeError(
+                raise SubprocessError(
                     "%s processes failed in step %s"
                     % (num_processes - len(completed), step_name)
                 )
@@ -1787,7 +1792,9 @@ def get_breadcrumbs(state: workflow.State, run_list):
     # - can't resume multiprocess without breadcrumbs file
     if not breadcrumbs:
         error(state, f"empty breadcrumbs for resume_after '{resume_after}'")
-        raise RuntimeError("empty breadcrumbs for resume_after '%s'" % resume_after)
+        raise CheckpointNameNotFoundError(
+            "empty breadcrumbs for resume_after '%s'" % resume_after
+        )
 
     # if resume_after is specified by name
     if resume_after != LAST_CHECKPOINT:
@@ -1808,7 +1815,7 @@ def get_breadcrumbs(state: workflow.State, run_list):
 
         if resume_step_name not in previous_steps:
             error(state, f"resume_after model '{resume_after}' not in breadcrumbs")
-            raise RuntimeError(
+            raise CheckpointNameNotFoundError(
                 "resume_after model '%s' not in breadcrumbs" % resume_after
             )
 
@@ -1826,7 +1833,7 @@ def get_breadcrumbs(state: workflow.State, run_list):
 
     multiprocess_step_names = [step["name"] for step in run_list["multiprocess_steps"]]
     if list(breadcrumbs.keys()) != multiprocess_step_names[: len(breadcrumbs)]:
-        raise RuntimeError(
+        raise CheckpointNameNotFoundError(
             "last run steps don't match run list: %s" % list(breadcrumbs.keys())
         )
 
@@ -1911,15 +1918,15 @@ def get_run_list(state: workflow.State):
     }
 
     if not models or not isinstance(models, list):
-        raise RuntimeError("No models list in settings file")
+        raise SystemConfigurationError("No models list in settings file")
     if resume_after == models[-1]:
-        raise RuntimeError(
+        raise SystemConfigurationError(
             "resume_after '%s' is last model in models list" % resume_after
         )
 
     if multiprocess:
         if not multiprocess_steps:
-            raise RuntimeError(
+            raise SystemConfigurationError(
                 "multiprocess setting is %s but no multiprocess_steps setting"
                 % multiprocess
             )
@@ -1935,15 +1942,15 @@ def get_run_list(state: workflow.State):
             # - validate step name
             name = step.get("name", None)
             if not name:
-                raise RuntimeError(
+                raise SystemConfigurationError(
                     "missing name for step %s" " in multiprocess_steps" % istep
                 )
             if name in step_names:
-                raise RuntimeError(
+                raise SystemConfigurationError(
                     "duplicate step name %s" " in multiprocess_steps" % name
                 )
             if name in models:
-                raise RuntimeError(
+                raise SystemConfigurationError(
                     f"multiprocess_steps step name '{name}' cannot also be a model name"
                 )
 
@@ -1953,7 +1960,7 @@ def get_run_list(state: workflow.State):
             num_processes = step.get("num_processes", 0) or 0
 
             if not isinstance(num_processes, int) or num_processes < 0:
-                raise RuntimeError(
+                raise SystemConfigurationError(
                     "bad value (%s) for num_processes for step %s"
                     " in multiprocess_steps" % (num_processes, name)
                 )
@@ -1975,7 +1982,7 @@ def get_run_list(state: workflow.State):
                 if num_processes == 0:
                     num_processes = 1
                 if num_processes > 1:
-                    raise RuntimeError(
+                    raise SystemConfigurationError(
                         "num_processes > 1 but no slice info for step %s"
                         " in multiprocess_steps" % name
                     )
@@ -2004,19 +2011,19 @@ def get_run_list(state: workflow.State):
             slice = step.get("slice", None)
             if slice:
                 if "tables" not in slice:
-                    raise RuntimeError(
+                    raise SystemConfigurationError(
                         "missing tables list for step %s"
                         " in multiprocess_steps" % istep
                     )
 
             start = step.get(start_tag, None)
             if not name:
-                raise RuntimeError(
+                raise SystemConfigurationError(
                     "missing %s tag for step '%s' (%s)"
                     " in multiprocess_steps" % (start_tag, name, istep)
                 )
             if start not in models:
-                raise RuntimeError(
+                raise SystemConfigurationError(
                     "%s tag '%s' for step '%s' (%s) not in models list"
                     % (start_tag, start, name, istep)
                 )
@@ -2024,14 +2031,14 @@ def get_run_list(state: workflow.State):
             starts[istep] = models.index(start)
 
             if istep == 0 and starts[istep] != 0:
-                raise RuntimeError(
+                raise SystemConfigurationError(
                     "%s tag '%s' for first step '%s' (%s)"
                     " is not first model in models list"
                     % (start_tag, start, name, istep)
                 )
 
             if istep > 0 and starts[istep] <= starts[istep - 1]:
-                raise RuntimeError(
+                raise SystemConfigurationError(
                     "%s tag '%s' for step '%s' (%s)"
                     " falls before that of prior step in models list"
                     % (start_tag, start, name, istep)
@@ -2048,7 +2055,7 @@ def get_run_list(state: workflow.State):
             step_models = models[starts[istep] : starts[istep + 1]]
 
             if step_models[-1][0] == LAST_CHECKPOINT:
-                raise RuntimeError(
+                raise CheckpointNameNotFoundError(
                     "Final model '%s' in step %s models list not checkpointed"
                     % (step_models[-1], name)
                 )
diff --git a/activitysim/core/pathbuilder_cache.py b/activitysim/core/pathbuilder_cache.py
index a59ca17ec4..6ed3a7061e 100644
--- a/activitysim/core/pathbuilder_cache.py
+++ b/activitysim/core/pathbuilder_cache.py
@@ -14,6 +14,7 @@
 import psutil
 
 from activitysim.core import config, los, util
+from activitysim.core.exceptions import StateAccessError, TableTypeError
 
 logger = logging.getLogger(__name__)
 
@@ -177,7 +178,7 @@ def open(self):
                 f"TVPBCache.open {self.cache_tag} read fully_populated data array from mmap file"
             )
         else:
-            raise RuntimeError(
+            raise StateAccessError(
                 f"Pathbuilder cache not found. Did you forget to run initialize tvpb?"
                 f"Expected cache file: {self.cache_path}"
             )
@@ -253,7 +254,7 @@ def allocate_data_buffer(self, shared=False):
             elif dtype_name == "float32":
                 typecode = "f"
             else:
-                raise RuntimeError(
+                raise TableTypeError(
                     "allocate_data_buffer unrecognized dtype %s" % dtype_name
                 )
 
diff --git a/activitysim/core/random.py b/activitysim/core/random.py
index 9960e98e27..37b1976403 100644
--- a/activitysim/core/random.py
+++ b/activitysim/core/random.py
@@ -10,6 +10,7 @@
 import pandas as pd
 
 from activitysim.core.util import reindex
+from activitysim.core.exceptions import DuplicateLoadableObjectError, TableIndexError
 
 from .tracing import print_elapsed_time
 
@@ -395,7 +396,7 @@ def get_channel_for_df(self, df):
 
         channel_name = self.index_to_channel.get(df.index.name, None)
         if channel_name is None:
-            raise RuntimeError("No channel with index name '%s'" % df.index.name)
+            raise TableIndexError("No channel with index name '%s'" % df.index.name)
         return self.channels[channel_name]
 
     # step handling
@@ -528,7 +529,9 @@ def set_base_seed(self, seed=None):
         """
 
         if self.step_name is not None or self.channels:
-            raise RuntimeError("Can only call set_base_seed before the first step.")
+            raise DuplicateLoadableObjectError(
+                "Can only call set_base_seed before the first step."
+            )
 
         assert len(list(self.channels.keys())) == 0
 
diff --git a/activitysim/core/simulate.py b/activitysim/core/simulate.py
index 6cd4a51f62..24d472ea72 100644
--- a/activitysim/core/simulate.py
+++ b/activitysim/core/simulate.py
@@ -40,6 +40,7 @@
     SPEC_EXPRESSION_NAME,
     SPEC_LABEL_NAME,
 )
+from activitysim.core.exceptions import ModelConfigurationError
 
 logger = logging.getLogger(__name__)
 
@@ -187,13 +188,13 @@ def read_model_coefficients(
             f"duplicate coefficients in {file_path}\n"
             f"{coefficients[coefficients.index.duplicated(keep=False)]}"
         )
-        raise RuntimeError(f"duplicate coefficients in {file_path}")
+        raise ModelConfigurationError(f"duplicate coefficients in {file_path}")
 
     if coefficients.value.isnull().any():
         logger.warning(
             f"null coefficients in {file_path}\n{coefficients[coefficients.value.isnull()]}"
         )
-        raise RuntimeError(f"null coefficients in {file_path}")
+        raise ModelConfigurationError(f"null coefficients in {file_path}")
 
     return coefficients
 
@@ -250,7 +251,7 @@ def spec_for_segment(
         try:
             assert (spec.astype(float) == spec).all(axis=None)
         except (ValueError, AssertionError):
-            raise RuntimeError(
+            raise ModelConfigurationError(
                 f"No coefficient file specified for {spec_file_name} "
                 f"but not all spec column values are numeric"
             ) from None
@@ -395,7 +396,7 @@ def get_segment_coefficients(
             FutureWarning,
         )
     else:
-        raise RuntimeError("No COEFFICIENTS setting in model_settings")
+        raise ModelConfigurationError("No COEFFICIENTS setting in model_settings")
 
     if legacy:
         constants = config.get_model_constants(model_settings)
@@ -1638,11 +1639,13 @@ def list_of_skims(skims):
         return (
             skims
             if isinstance(skims, list)
-            else skims.values()
-            if isinstance(skims, dict)
-            else [skims]
-            if skims is not None
-            else []
+            else (
+                skims.values()
+                if isinstance(skims, dict)
+                else [skims]
+                if skims is not None
+                else []
+            )
         )
 
     return [
diff --git a/activitysim/core/skim_dict_factory.py b/activitysim/core/skim_dict_factory.py
index e0bd23ef3e..2ca90c4c66 100644
--- a/activitysim/core/skim_dict_factory.py
+++ b/activitysim/core/skim_dict_factory.py
@@ -14,6 +14,7 @@
 import openmatrix as omx
 
 from activitysim.core import skim_dictionary, util
+from activitysim.core.exceptions import TableTypeError
 
 logger = logging.getLogger(__name__)
 
@@ -462,7 +463,7 @@ def allocate_skim_buffer(self, skim_info, shared=False):
             elif dtype_name == "float32":
                 typecode = "f"
             else:
-                raise RuntimeError(
+                raise TableTypeError(
                     "allocate_skim_buffer unrecognized dtype %s" % dtype_name
                 )
 
diff --git a/activitysim/core/skim_dictionary.py b/activitysim/core/skim_dictionary.py
index 020002e252..59692e3d37 100644
--- a/activitysim/core/skim_dictionary.py
+++ b/activitysim/core/skim_dictionary.py
@@ -10,7 +10,7 @@
 import pandas as pd
 
 from activitysim.core import workflow
-from activitysim.core.exceptions import StateAccessError
+from activitysim.core.exceptions import StateAccessError, TableIndexError
 
 logger = logging.getLogger(__name__)
 
@@ -400,7 +400,6 @@ def wrap_3d(self, orig_key, dest_key, dim3_key):
 
 
 class SkimWrapper(object):
-
     """
     A SkimWrapper object is an access wrapper around a SkimDict of multiple skim objects,
     where each object is identified by a key.
@@ -907,7 +906,7 @@ def get(self, row_ids, col_ids):
             not_in_skim = not_in_skim.values
             logger.warning(f"row_ids: {row_ids[not_in_skim]}")
             logger.warning(f"col_ids: {col_ids[not_in_skim]}")
-            raise RuntimeError(
+            raise TableIndexError(
                 f"DataFrameMatrix: {not_in_skim.sum()} row_ids of {len(row_ids)} not in skim."
             )
 
diff --git a/activitysim/core/test/_tools.py b/activitysim/core/test/_tools.py
index 618b467bb9..3169b771c2 100644
--- a/activitysim/core/test/_tools.py
+++ b/activitysim/core/test/_tools.py
@@ -7,6 +7,8 @@
 
 import pandas as pd
 
+from activitysim.core.exceptions import PipelineError
+
 
 def run_if_exists(filename):
     import pytest
@@ -176,6 +178,6 @@ def progressive_checkpoint_test(
     # generate the reference pipeline if it did not exist
     if not ref_target.exists():
         state.checkpoint.store.make_zip_archive(ref_target)
-        raise RuntimeError(
+        raise PipelineError(
             f"Reference pipeline {ref_target} did not exist, so it was created."
         )
diff --git a/activitysim/core/test/test_pipeline.py b/activitysim/core/test/test_pipeline.py
index dfa6d770a8..12f31dbc66 100644
--- a/activitysim/core/test/test_pipeline.py
+++ b/activitysim/core/test/test_pipeline.py
@@ -9,6 +9,7 @@
 import tables
 
 from activitysim.core import workflow
+from activitysim.core.exceptions import CheckpointNameNotFoundError
 from activitysim.core.test.extensions import steps
 
 # set the max households for all tests (this is to limit memory use on travis)
@@ -70,17 +71,17 @@ def test_pipeline_run(state):
     state.checkpoint.load_dataframe("table1", checkpoint_name="step3")
 
     # try to get a table from a step before it was checkpointed
-    with pytest.raises(RuntimeError) as excinfo:
+    with pytest.raises(CheckpointNameNotFoundError) as excinfo:
         state.checkpoint.load_dataframe("table2", checkpoint_name="step1")
     assert "not in checkpoint 'step1'" in str(excinfo.value)
 
     # try to get a non-existant table
-    with pytest.raises(RuntimeError) as excinfo:
+    with pytest.raises(CheckpointNameNotFoundError) as excinfo:
         state.checkpoint.load_dataframe("bogus")
     assert "never checkpointed" in str(excinfo.value)
 
     # try to get an existing table from a non-existant checkpoint
-    with pytest.raises(RuntimeError) as excinfo:
+    with pytest.raises(CheckpointNameNotFoundError) as excinfo:
         state.checkpoint.load_dataframe("table1", checkpoint_name="bogus")
     assert "not in checkpoints" in str(excinfo.value)
 
@@ -111,12 +112,12 @@ def test_pipeline_checkpoint_drop(state):
 
     state.checkpoint.load_dataframe("table1")
 
-    with pytest.raises(RuntimeError) as excinfo:
+    with pytest.raises(CheckpointNameNotFoundError) as excinfo:
         state.checkpoint.load_dataframe("table2")
     # assert "never checkpointed" in str(excinfo.value)
 
     # can't get a dropped table from current checkpoint
-    with pytest.raises(RuntimeError) as excinfo:
+    with pytest.raises(CheckpointNameNotFoundError) as excinfo:
         state.checkpoint.load_dataframe("table3")
     # assert "was dropped" in str(excinfo.value)
 
diff --git a/activitysim/core/test/test_random.py b/activitysim/core/test/test_random.py
index 63809278c1..bcbc602685 100644
--- a/activitysim/core/test/test_random.py
+++ b/activitysim/core/test/test_random.py
@@ -8,6 +8,7 @@
 import pytest
 
 from activitysim.core import random
+from activitysim.core.exceptions import DuplicateLoadableObjectError
 
 
 def test_basic():
@@ -27,7 +28,7 @@ def test_basic():
     assert "Arrays are not almost equal" in str(excinfo.value)
 
     # second call should return something different
-    with pytest.raises(RuntimeError) as excinfo:
+    with pytest.raises(DuplicateLoadableObjectError) as excinfo:
         rng.set_base_seed(1)
     assert "call set_base_seed before the first step" in str(excinfo.value)
 
diff --git a/activitysim/core/timetable.py b/activitysim/core/timetable.py
index 5743aeef0a..cedb3d2673 100644
--- a/activitysim/core/timetable.py
+++ b/activitysim/core/timetable.py
@@ -10,6 +10,7 @@
 import pandas as pd
 
 from activitysim.core import chunk, configuration, workflow
+from activitysim.core.exceptions import DuplicateWorkflowTableError
 
 logger = logging.getLogger(__name__)
 
@@ -463,7 +464,9 @@ def replace_table(self, state: workflow.State):
                     % self.windows_table_name,
                     level=logging.ERROR,
                 )
-            raise RuntimeError("Attempt to replace_table while in transaction")
+            raise DuplicateWorkflowTableError(
+                "Attempt to replace_table while in transaction"
+            )
 
         # get windows_df from bottleneck function in case updates to self.person_window
         # do not write through to pandas dataframe
diff --git a/activitysim/core/tracing.py b/activitysim/core/tracing.py
index 88bf0fc167..83c9f876aa 100644
--- a/activitysim/core/tracing.py
+++ b/activitysim/core/tracing.py
@@ -11,6 +11,8 @@
 import numpy as np
 import pandas as pd
 
+from activitysim.core.exceptions import TableSlicingError
+
 # Configurations
 ASIM_LOGGER = "activitysim"
 CSV_FILE_TYPE = "csv"
@@ -247,7 +249,9 @@ def slice_ids(df, ids, column=None):
     except KeyError:
         # this happens if specified slicer column is not in df
         # df = df[0:0]
-        raise RuntimeError("slice_ids slicer column '%s' not in dataframe" % column)
+        raise TableSlicingError(
+            "slice_ids slicer column '%s' not in dataframe" % column
+        )
 
     return df
 
diff --git a/activitysim/core/util.py b/activitysim/core/util.py
index 429d3eee01..c3f0385152 100644
--- a/activitysim/core/util.py
+++ b/activitysim/core/util.py
@@ -298,7 +298,7 @@ def quick_loc_series(loc_list, target_series):
     elif isinstance(loc_list, np.ndarray) or isinstance(loc_list, list):
         left_df = pd.DataFrame({left_on: loc_list})
     else:
-        raise RuntimeError(
+        raise TypeError(
             "quick_loc_series loc_list of unexpected type %s" % type(loc_list)
         )
 
diff --git a/activitysim/core/workflow/checkpoint.py b/activitysim/core/workflow/checkpoint.py
index d5a7286359..7391e1c9b9 100644
--- a/activitysim/core/workflow/checkpoint.py
+++ b/activitysim/core/workflow/checkpoint.py
@@ -18,6 +18,7 @@
     TableNameNotFound,
 )
 from activitysim.core.workflow.accessor import FromState, StateAccessor
+from activitysim.core.exceptions import CheckpointNameNotFoundError, PipelineError
 
 logger = logging.getLogger(__name__)
 
@@ -565,7 +566,7 @@ def open_store(
         """
 
         if self._checkpoint_store is not None:
-            raise RuntimeError("Pipeline store is already open!")
+            raise PipelineError("Pipeline store is already open!")
 
         if pipeline_file_name is None:
             pipeline_file_path = self.default_pipeline_file_path()
@@ -775,7 +776,7 @@ def load(self, checkpoint_name: str, store=None):
             msg = f"Couldn't find checkpoint '{checkpoint_name}' in checkpoints"
             print(checkpoints[CHECKPOINT_NAME])
             logger.error(msg)
-            raise RuntimeError(msg) from None
+            raise CheckpointNameNotFoundError(msg) from None
 
         # convert pandas dataframe back to array of checkpoint dicts
         checkpoints = checkpoints.to_dict(orient="records")
@@ -1201,7 +1202,7 @@ def load_dataframe(self, table_name, checkpoint_name=None):
 
         if table_name not in self.last_checkpoint and self._obj.is_table(table_name):
             if checkpoint_name is not None:
-                raise RuntimeError(
+                raise CheckpointNameNotFoundError(
                     f"checkpoint.dataframe: checkpoint_name ({checkpoint_name!r}) not "
                     f"supported for non-checkpointed table {table_name!r}"
                 )
@@ -1211,10 +1212,14 @@ def load_dataframe(self, table_name, checkpoint_name=None):
         # if there is no checkpoint name given, do not attempt to read from store
         if checkpoint_name is None:
             if table_name not in self.last_checkpoint:
-                raise RuntimeError("table '%s' never checkpointed." % table_name)
+                raise CheckpointNameNotFoundError(
+                    "table '%s' never checkpointed." % table_name
+                )
 
             if not self.last_checkpoint[table_name]:
-                raise RuntimeError("table '%s' was dropped." % table_name)
+                raise CheckpointNameNotFoundError(
+                    "table '%s' was dropped." % table_name
+                )
 
             return self._obj.get_dataframe(table_name)
 
@@ -1224,13 +1229,15 @@ def load_dataframe(self, table_name, checkpoint_name=None):
             None,
         )
         if checkpoint is None:
-            raise RuntimeError("checkpoint '%s' not in checkpoints." % checkpoint_name)
+            raise CheckpointNameNotFoundError(
+                "checkpoint '%s' not in checkpoints." % checkpoint_name
+            )
 
         # find the checkpoint that table was written to store
         last_checkpoint_name = checkpoint.get(table_name, None)
 
         if not last_checkpoint_name:
-            raise RuntimeError(
+            raise CheckpointNameNotFoundError(
                 "table '%s' not in checkpoint '%s'." % (table_name, checkpoint_name)
             )
 
diff --git a/activitysim/core/workflow/runner.py b/activitysim/core/workflow/runner.py
index 8eba53cbef..79ecd0ed4f 100644
--- a/activitysim/core/workflow/runner.py
+++ b/activitysim/core/workflow/runner.py
@@ -265,7 +265,7 @@ def _pre_run_step(self, model_name: str) -> bool | None:
         if model_name in checkpointed_models:
             if self._obj.settings.duplicate_step_execution == "error":
                 checkpointed_model_bullets = "\n - ".join(checkpointed_models)
-                raise RuntimeError(
+                raise DuplicateWorkflowNameError(
                     f"Checkpointed Models:\n - {checkpointed_model_bullets}\n"
                     f"Cannot run model '{model_name}' more than once"
                 )
diff --git a/activitysim/core/workflow/state.py b/activitysim/core/workflow/state.py
index 6178f24884..9f7dcd4d6f 100644
--- a/activitysim/core/workflow/state.py
+++ b/activitysim/core/workflow/state.py
@@ -20,7 +20,7 @@
 
 import activitysim.core.random
 from activitysim.core.configuration import FileSystem, NetworkSettings, Settings
-from activitysim.core.exceptions import StateAccessError
+from activitysim.core.exceptions import StateAccessError, CheckpointNameNotFoundError
 from activitysim.core.workflow.checkpoint import LAST_CHECKPOINT, Checkpoints
 from activitysim.core.workflow.chunking import Chunking
 from activitysim.core.workflow.dataset import Datasets
@@ -1017,7 +1017,7 @@ def get_table(self, table_name, checkpoint_name=None):
             table_name
         ):
             if checkpoint_name is not None:
-                raise RuntimeError(
+                raise CheckpointNameNotFoundError(
                     f"get_table: checkpoint_name ({checkpoint_name!r}) not "
                     f"supported for non-checkpointed table {table_name!r}"
                 )
@@ -1027,10 +1027,14 @@ def get_table(self, table_name, checkpoint_name=None):
         # if they want current version of table, no need to read from pipeline store
         if checkpoint_name is None:
             if table_name not in self.checkpoint.last_checkpoint:
-                raise RuntimeError("table '%s' never checkpointed." % table_name)
+                raise CheckpointNameNotFoundError(
+                    "table '%s' never checkpointed." % table_name
+                )
 
             if not self.checkpoint.last_checkpoint[table_name]:
-                raise RuntimeError("table '%s' was dropped." % table_name)
+                raise CheckpointNameNotFoundError(
+                    "table '%s' was dropped." % table_name
+                )
 
             return self._context.get(table_name)
 
@@ -1044,13 +1048,15 @@ def get_table(self, table_name, checkpoint_name=None):
             None,
         )
         if checkpoint is None:
-            raise RuntimeError("checkpoint '%s' not in checkpoints." % checkpoint_name)
+            raise CheckpointNameNotFoundError(
+                "checkpoint '%s' not in checkpoints." % checkpoint_name
+            )
 
         # find the checkpoint that table was written to store
         last_checkpoint_name = checkpoint.get(table_name, None)
 
         if not last_checkpoint_name:
-            raise RuntimeError(
+            raise CheckpointNameNotFoundError(
                 "table '%s' not in checkpoint '%s'." % (table_name, checkpoint_name)
             )
 
diff --git a/activitysim/core/workflow/tracing.py b/activitysim/core/workflow/tracing.py
index 580c6fad9b..2669e5b9b8 100644
--- a/activitysim/core/workflow/tracing.py
+++ b/activitysim/core/workflow/tracing.py
@@ -21,6 +21,7 @@
 from activitysim.core import tracing
 from activitysim.core.test import assert_equal, assert_frame_substantively_equal
 from activitysim.core.workflow.accessor import FromState, StateAccessor
+from activitysim.core.exceptions import TableSlicingError
 
 logger = logging.getLogger(__name__)
 
@@ -548,7 +549,7 @@ def interaction_trace_rows(self, interaction_df, choosers, sample_size=None):
             targets = traceable_table_ids["proto_tours"]
         else:
             print(choosers.columns)
-            raise RuntimeError(
+            raise TableSlicingError(
                 "interaction_trace_rows don't know how to slice index '%s'"
                 % choosers.index.name
             )
@@ -629,7 +630,7 @@ def get_trace_target(self, df: pd.DataFrame, slicer: str, column: Any = None):
                 column = slicer
 
         if column is None and df.index.name != slicer:
-            raise RuntimeError(
+            raise TableSlicingError(
                 "bad slicer '%s' for df with index '%s'" % (slicer, df.index.name)
             )
 
diff --git a/activitysim/estimation/test/test_larch_estimation.py b/activitysim/estimation/test/test_larch_estimation.py
index 2d2427c41b..e688b1098a 100644
--- a/activitysim/estimation/test/test_larch_estimation.py
+++ b/activitysim/estimation/test/test_larch_estimation.py
@@ -44,7 +44,7 @@ def _regression_check(dataframe_regression, df, basename=None, rtol=None):
         # pandas 1.3 handles int8 dtypes as actual numbers, so holdfast needs to be dropped manually
         # we're dropping it not adding to the regression check so older pandas will also work.
         basename=basename,
-        default_tolerance=dict(atol=1e-6, rtol=rtol)
+        default_tolerance=dict(atol=1e-6, rtol=rtol),
         # can set a little loose, as there is sometimes a little variance in these
         # results when switching backend implementations. We're checking all
         # the parameters and the log likelihood, so modest variance in individual
@@ -129,7 +129,7 @@ def test_location_model(
     dataframe_regression.check(
         size_spec,
         basename=f"test_loc_{name}_{method}_size_spec",
-        default_tolerance=dict(atol=1e-6, rtol=5e-2)
+        default_tolerance=dict(atol=1e-6, rtol=5e-2),
         # set a little loose, as there is sometimes a little variance in these
         # results when switching backend implementations.
     )
diff --git a/activitysim/examples/placeholder_multiple_zone/scripts/three_zone_example_data.py b/activitysim/examples/placeholder_multiple_zone/scripts/three_zone_example_data.py
index 7291c65337..1ce3861681 100644
--- a/activitysim/examples/placeholder_multiple_zone/scripts/three_zone_example_data.py
+++ b/activitysim/examples/placeholder_multiple_zone/scripts/three_zone_example_data.py
@@ -149,13 +149,15 @@
 tap_df.to_csv(os.path.join(output_data, "tap.csv"), index=False)
 
 # create taz_z3 and tap skims
-with omx.open_file(
-    os.path.join(input_data, "skims.omx"), "r"
-) as ur_skims, omx.open_file(
-    os.path.join(output_data, "taz_skims.omx"), "w"
-) as output_taz_skims_file, omx.open_file(
-    os.path.join(output_data, "tap_skims.omx"), "w"
-) as output_tap_skims_file:
+with (
+    omx.open_file(os.path.join(input_data, "skims.omx"), "r") as ur_skims,
+    omx.open_file(
+        os.path.join(output_data, "taz_skims.omx"), "w"
+    ) as output_taz_skims_file,
+    omx.open_file(
+        os.path.join(output_data, "tap_skims.omx"), "w"
+    ) as output_tap_skims_file,
+):
     for skim_name in ur_skims.list_matrices():
         ur_skim = ur_skims[skim_name][:]
         new_skim = ur_skim[taz_zone_indexes, :][:, taz_zone_indexes]
diff --git a/activitysim/examples/placeholder_multiple_zone/scripts/two_zone_example_data.py b/activitysim/examples/placeholder_multiple_zone/scripts/two_zone_example_data.py
index a4c6c46fcb..20c7bccf53 100644
--- a/activitysim/examples/placeholder_multiple_zone/scripts/two_zone_example_data.py
+++ b/activitysim/examples/placeholder_multiple_zone/scripts/two_zone_example_data.py
@@ -104,11 +104,10 @@
 
 # ### Create taz skims
 
-with omx.open_file(
-    os.path.join(input_data, "skims.omx"), "r"
-) as skims_file, omx.open_file(
-    os.path.join(output_data, "taz_skims.omx"), "w"
-) as output_skims_file:
+with (
+    omx.open_file(os.path.join(input_data, "skims.omx"), "r") as skims_file,
+    omx.open_file(os.path.join(output_data, "taz_skims.omx"), "w") as output_skims_file,
+):
     skims = skims_file.list_matrices()
     num_zones = skims_file.shape()[0]
 
diff --git a/activitysim/examples/placeholder_multiple_zone/three_zone_example_data.py b/activitysim/examples/placeholder_multiple_zone/three_zone_example_data.py
index 0a43cce220..54224087d6 100644
--- a/activitysim/examples/placeholder_multiple_zone/three_zone_example_data.py
+++ b/activitysim/examples/placeholder_multiple_zone/three_zone_example_data.py
@@ -143,13 +143,15 @@
 tap_df.to_csv(os.path.join(output_data, "tap.csv"), index=False)
 
 # create taz_z3 and tap skims
-with omx.open_file(
-    os.path.join(input_data, "skims.omx"), "r"
-) as ur_skims, omx.open_file(
-    os.path.join(output_data, "taz_skims.omx"), "w"
-) as output_taz_skims_file, omx.open_file(
-    os.path.join(output_data, "tap_skims.omx"), "w"
-) as output_tap_skims_file:
+with (
+    omx.open_file(os.path.join(input_data, "skims.omx"), "r") as ur_skims,
+    omx.open_file(
+        os.path.join(output_data, "taz_skims.omx"), "w"
+    ) as output_taz_skims_file,
+    omx.open_file(
+        os.path.join(output_data, "tap_skims.omx"), "w"
+    ) as output_tap_skims_file,
+):
     for skim_name in ur_skims.list_matrices():
         ur_skim = ur_skims[skim_name][:]
         new_skim = ur_skim[taz_zone_indexes, :][:, taz_zone_indexes]
diff --git a/activitysim/examples/placeholder_multiple_zone/two_zone_example_data.py b/activitysim/examples/placeholder_multiple_zone/two_zone_example_data.py
index fbb26e2aaa..a6f5750a0f 100644
--- a/activitysim/examples/placeholder_multiple_zone/two_zone_example_data.py
+++ b/activitysim/examples/placeholder_multiple_zone/two_zone_example_data.py
@@ -101,11 +101,10 @@
 
 # ### Create taz skims
 
-with omx.open_file(
-    os.path.join(input_data, "skims.omx"), "r"
-) as skims_file, omx.open_file(
-    os.path.join(output_data, "taz_skims.omx"), "w"
-) as output_skims_file:
+with (
+    omx.open_file(os.path.join(input_data, "skims.omx"), "r") as skims_file,
+    omx.open_file(os.path.join(output_data, "taz_skims.omx"), "w") as output_skims_file,
+):
     skims = skims_file.list_matrices()
     num_zones = skims_file.shape()[0]
 
diff --git a/activitysim/examples/production_semcog/data_model/enums.py b/activitysim/examples/production_semcog/data_model/enums.py
index 71c4369248..8ae8f9eb4c 100644
--- a/activitysim/examples/production_semcog/data_model/enums.py
+++ b/activitysim/examples/production_semcog/data_model/enums.py
@@ -3,6 +3,7 @@
 
 Instructions: modify these enumerated variables as needed for your ActivitySim implementation.
 """
+
 from enum import IntEnum
 
 
diff --git a/activitysim/examples/production_semcog/data_model/input_checks.py b/activitysim/examples/production_semcog/data_model/input_checks.py
index b9b1f338b1..ce93a86e0c 100644
--- a/activitysim/examples/production_semcog/data_model/input_checks.py
+++ b/activitysim/examples/production_semcog/data_model/input_checks.py
@@ -3,6 +3,7 @@
 
 Instructions: customize these example values for your own ActivitySim implementation
 """
+
 from typing import List, Optional
 import os, sys, logging
 
diff --git a/activitysim/examples/prototype_mtc_extended/data_model/enums.py b/activitysim/examples/prototype_mtc_extended/data_model/enums.py
index 124647e5db..b28960501a 100644
--- a/activitysim/examples/prototype_mtc_extended/data_model/enums.py
+++ b/activitysim/examples/prototype_mtc_extended/data_model/enums.py
@@ -3,6 +3,7 @@
 
 Instructions: modify these enumerated variables as needed for your ActivitySim implementation.
 """
+
 from enum import IntEnum
 
 
diff --git a/activitysim/examples/prototype_mtc_extended/data_model/input_checks.py b/activitysim/examples/prototype_mtc_extended/data_model/input_checks.py
index 1cc63e5836..f87767ab69 100644
--- a/activitysim/examples/prototype_mtc_extended/data_model/input_checks.py
+++ b/activitysim/examples/prototype_mtc_extended/data_model/input_checks.py
@@ -3,6 +3,7 @@
 
 Instructions: customize these example values for your own ActivitySim implementation
 """
+
 from __future__ import annotations
 
 import csv
diff --git a/activitysim/examples/prototype_mtc_extended/data_model/input_checks_pydantic_dev.py b/activitysim/examples/prototype_mtc_extended/data_model/input_checks_pydantic_dev.py
index b171119066..f2c3ddb997 100644
--- a/activitysim/examples/prototype_mtc_extended/data_model/input_checks_pydantic_dev.py
+++ b/activitysim/examples/prototype_mtc_extended/data_model/input_checks_pydantic_dev.py
@@ -3,6 +3,7 @@
 
 Instructions: customize these example values for your own ActivitySim implementation
 """
+
 from typing import List, Optional
 import os, sys, logging
 
diff --git a/activitysim/workflows/steps/cmd/__init__.py b/activitysim/workflows/steps/cmd/__init__.py
index e62b19e851..1463609e30 100644
--- a/activitysim/workflows/steps/cmd/__init__.py
+++ b/activitysim/workflows/steps/cmd/__init__.py
@@ -4,6 +4,7 @@
 environment,variable expansion, and expansion of ~ to a user’s home
 directory.
 """
+
 import logging
 
 from .dsl import CmdStep
diff --git a/activitysim/workflows/steps/cmd/dsl.py b/activitysim/workflows/steps/cmd/dsl.py
index 1d05a7d3b6..7337eab3ca 100644
--- a/activitysim/workflows/steps/cmd/dsl.py
+++ b/activitysim/workflows/steps/cmd/dsl.py
@@ -1,4 +1,5 @@
 """pypyr step yaml definition for commands - domain specific language."""
+
 import logging
 import os
 import shlex
diff --git a/activitysim/workflows/steps/contrast/data_inventory.py b/activitysim/workflows/steps/contrast/data_inventory.py
index a3ff172a31..70b5f4a104 100644
--- a/activitysim/workflows/steps/contrast/data_inventory.py
+++ b/activitysim/workflows/steps/contrast/data_inventory.py
@@ -63,7 +63,7 @@ def run_step(context: Context) -> None:
                     1
                 ),
                 index=dtypes_table.index,
-            ).apply(lambda x: "" if x else "\u2B05")
+            ).apply(lambda x: "" if x else "\u2b05")
             report << dtypes_table
 
     with report:
diff --git a/activitysim/workflows/steps/main.py b/activitysim/workflows/steps/main.py
index ccefaa825b..9859254d2b 100644
--- a/activitysim/workflows/steps/main.py
+++ b/activitysim/workflows/steps/main.py
@@ -1,4 +1,5 @@
 """Naive custom loader without any error handling."""
+
 from __future__ import annotations
 
 import os
diff --git a/activitysim/workflows/steps/progression.py b/activitysim/workflows/steps/progression.py
index c0832dc46c..d77513bfa8 100644
--- a/activitysim/workflows/steps/progression.py
+++ b/activitysim/workflows/steps/progression.py
@@ -96,7 +96,7 @@ def update_progress_overall(description, formatting=""):
 
 def reset_progress_step(*args, description="", prefix="", **kwargs):
     if not os.environ.get("NO_RICH", False):
-        print(f"\u23F1  {time.strftime('%I:%M:%S %p')} - {description}")
+        print(f"\u23f1  {time.strftime('%I:%M:%S %p')} - {description}")
         progress.reset(progress_step, *args, description=prefix + description, **kwargs)
     else:
         print("╭" + "─" * (len(description) + 2) + "╮")
diff --git a/test/cdap/test_cdap.py b/test/cdap/test_cdap.py
index 38e2657769..b26078c904 100644
--- a/test/cdap/test_cdap.py
+++ b/test/cdap/test_cdap.py
@@ -162,9 +162,9 @@ def test_cdap_from_pipeline(reconnect_pipeline: workflow.State, caplog):
         lambda x: x[:-1].upper() if x.endswith("0") else x.upper()
     )
     household_df["cdap_activity"] = household_df.apply(
-        lambda x: x["cdap_activity"] + "J"
-        if x["has_joint_tour"] == 1
-        else x["cdap_activity"],
+        lambda x: (
+            x["cdap_activity"] + "J" if x["has_joint_tour"] == 1 else x["cdap_activity"]
+        ),
         axis=1,
     )