diff --git a/docker/Dockerfile.transformed-efd b/docker/Dockerfile.transformed-efd
index fffde83b..bea2a66b 100644
--- a/docker/Dockerfile.transformed-efd
+++ b/docker/Dockerfile.transformed-efd
@@ -5,9 +5,13 @@ FROM ghcr.io/lsst/scipipe:al9-${OBS_LSST_VERSION}
 # Set user
 USER lsst
 
-# Install dependencies
-RUN source loadLSST.bash && mamba install -y aiokafka httpx && \
-    pip install \
+# Install dependencies (clean conda pkgs cache first — base image may ship
+# corrupt/empty tarballs that make mamba abort on checksum mismatch)
+RUN source loadLSST.bash \
+    && mamba clean --all --yes \
+    && mamba install --yes aiokafka httpx \
+    && mamba clean --all --yes \
+    && pip install \
     kafkit==0.2.1 \
     lsst_efd_client==0.12.0
 
@@ -18,7 +22,7 @@ COPY --chown=lsst:lsst python /opt/lsst/software/stack/python
 RUN mkdir -p /opt/lsst/software/stack/data
 
 # Environment variables (overridden in docker run)
-ENV CONFIG_FILE="/opt/lsst/software/stack/python/lsst/consdb/efd_transform/config_LATISS.yml"
+ENV CONFIG_FILE="/opt/lsst/software/stack/python/lsst/consdb/efd_transform/config_lsstcam.yml"
 ENV INSTRUMENT=""
 ENV BUTLER_REPO=""
 ENV S3_ENDPOINT_URL=""
diff --git a/python/lsst/consdb/transformed_efd/__init__.py b/python/lsst/consdb/transformed_efd/__init__.py
index 4ab08f7b..359350c3 100644
--- a/python/lsst/consdb/transformed_efd/__init__.py
+++ b/python/lsst/consdb/transformed_efd/__init__.py
@@ -57,6 +57,7 @@
 --------------------
 The module uses YAML files for instrument-specific configurations:
 - `config_latiss.yaml`: Configuration for LATISS.
+- `config_lsstcam.yaml`: Configuration for LSSTCam.
 - `config_lsstcomcam.yaml`: Configuration for LSSTComCam.
 - `config_lsstcomcamsim.yaml`: Configuration for LSSTComCamSim.
 
diff --git a/python/lsst/consdb/transformed_efd/config_model.py b/python/lsst/consdb/transformed_efd/config_model.py
index 8b2c047a..655b5843 100644
--- a/python/lsst/consdb/transformed_efd/config_model.py
+++ b/python/lsst/consdb/transformed_efd/config_model.py
@@ -73,6 +73,9 @@ class Column(BaseModel):
         function (str): The transformation function to apply to the column.
         function_args (Optional[Dict]): A dictionary of arguments for the
             transformation function.
+        start_offset (Optional[float]): Hours to shift the exposure/visit
+            window start when querying EFD and computing values (e.g. -0.5
+            looks back 30 minutes before the timespan begin).
         datatype (str): The data type of the column. Must match the
             Felis type.
         ivoa (Optional[Dict]): Dictionary containing IVOA metadata for TAP,
@@ -94,6 +97,7 @@ class Column(BaseModel):
     store_unpivoted: Optional[bool] = False
     function: str
     function_args: Optional[Dict] = None
+    start_offset: Optional[float] = None
     pre_aggregate_interval: Optional[str] = None
     datatype: str
     ivoa: Optional[Dict] = None
diff --git a/python/lsst/consdb/transformed_efd/dao/__init__.py b/python/lsst/consdb/transformed_efd/dao/__init__.py
index 8ffcc1a4..01ce20b0 100644
--- a/python/lsst/consdb/transformed_efd/dao/__init__.py
+++ b/python/lsst/consdb/transformed_efd/dao/__init__.py
@@ -26,14 +26,12 @@
         management and query execution.
     - butler: Provides data access methods for querying dimensions using a
         Butler object.
-    - exposure_efd: Handles operations related to the "exposure_efd" table.
-    - exposure_efd_unpivoted: Manages data for the
-        "exposure_efd_unpivoted" table.
+    - exposure_efd: Handles operations related to the "exposure_efd" and
+        "exposure_efd_unpivoted" tables.
     - influxdb: Interfaces with the InfluxDB API for time-series data queries.
     - transformd: Manages transformed EFD scheduler data.
-    - visit_efd: Handles data operations for the "visit1_efd" table.
-    - visit_efd_unpivoted: Accesses data for the "visit1_efd_unpivoted"
-        table.
+    - visit_efd: Handles data operations for the "visit1_efd" and
+        "visit1_efd_unpivoted" tables.
 
 These modules collectively facilitate database operations, data transformation,
 and efficient retrieval of time-series and structured data.
diff --git a/python/lsst/consdb/transformed_efd/dao/base.py b/python/lsst/consdb/transformed_efd/dao/base.py
index 5793f933..1e486447 100644
--- a/python/lsst/consdb/transformed_efd/dao/base.py
+++ b/python/lsst/consdb/transformed_efd/dao/base.py
@@ -24,7 +24,7 @@
 import logging
 import warnings
 from datetime import datetime, timezone
-from typing import Any, Dict, List
+from typing import Any, Dict, List, Optional
 
 import numpy
 import pandas
@@ -32,7 +32,8 @@
 from sqlalchemy import exc as sa_exc
 from sqlalchemy import func
 from sqlalchemy.dialects import postgresql, sqlite
-from sqlalchemy.pool import NullPool
+from sqlalchemy.engine import make_url
+from sqlalchemy.pool import QueuePool
 
 
 class DBBase:
@@ -46,7 +47,6 @@ class DBBase:
     ----------
         db_uris (list[str]): All database URIs.
         db_uri (str): The primary database URI (first in list).
-        connexion: The database connection (primary).
         dialect: The database dialect.
         logger (logging.Logger): The logger for the class.
 
@@ -66,7 +66,7 @@ def __init__(self, db_uri: str | list[str], schema: str = None, logger: logging.
 
         Raises:
         ------
-            Exception: If the dialect has not been implemented.
+            NotImplementedError: If the dialect has not been implemented.
 
         """
         if isinstance(db_uri, str):
@@ -75,24 +75,55 @@ def __init__(self, db_uri: str | list[str], schema: str = None, logger: logging.
             self.db_uris = list(db_uri)
 
         self.db_uri = self.db_uris[0]
-        self._engines: list[Engine | None] = [None] * len(self.db_uris)
-        self.connexion = None
-        self.log = logger
+        self.log = logger or logging.getLogger(__name__)
+        self._setup_warning_redirect()
+
+        dialect_name = make_url(self.db_uri).get_backend_name()
+        supported_dialects = {"postgresql": postgresql, "sqlite": sqlite}
+        if dialect_name not in supported_dialects:
+            raise NotImplementedError(f"The dialect for {dialect_name} has not yet been implemented.")
+        self.dialect = supported_dialects[dialect_name]
+
+        # Only construct engines after dialect validation.
+        self._engines: list[Engine] = [
+            create_engine(
+                uri,
+                poolclass=QueuePool,
+                pool_size=1,
+                max_overflow=0,
+                pool_pre_ping=True,
+                connect_args={"connect_timeout": 5},
+            )
+            for uri in self.db_uris
+        ]
 
-        sgbd = self.db_uri.split(":")[0]
+    def _setup_warning_redirect(self) -> None:
+        """Redirect SQLAlchemy warnings to the structured logger.
 
-        if sgbd == "sqlite":
-            self.dialect = sqlite
-        elif sgbd == "postgresql":
-            self.dialect = postgresql
-        else:
-            raise Exception(f"The dialect for {sgbd} has not yet been implemented.")
+        SAWarning is intercepted and logged with ``event=db_sa_warning``
+        so it is discoverable in Loki / log-based monitoring. All other
+        warnings pass through to the default handler unchanged.
+        """
+        _original = warnings.showwarning
+
+        def _handler(message, category, filename, lineno, file=None, line=None):
+            if issubclass(category, sa_exc.SAWarning):
+                self.log.warning(
+                    "event=db_sa_warning message=%s filename=%s lineno=%s",
+                    str(message),
+                    filename,
+                    lineno,
+                )
+            else:
+                _original(message, category, filename, lineno, file, line)
+
+        warnings.showwarning = _handler
 
     def get_db_engine(self, index: int = 0) -> Engine:
         """Return the database engine for the given index.
 
-        If the engine is not already created, it creates a new engine
-        using the corresponding database URI and returns it.
+        Engines are created eagerly in ``__init__``, so this method
+        is thread-safe and guaranteed to return immediately.
 
         Parameters
         ----------
@@ -104,35 +135,13 @@ def get_db_engine(self, index: int = 0) -> Engine:
             The database engine.
 
         """
-        if self._engines[index] is None:
-            self._engines[index] = create_engine(self.db_uris[index], poolclass=NullPool)
-
         return self._engines[index]
 
-    def get_con(self):
-        """Return the database connection.
-
-        If the connection is not already established, it creates a new
-        connection using the database engine obtained
-        from `get_db_engine` method.
-
-        Returns
-        -------
-            The database connection.
-
-        """
-        if self.connexion is None:
-            engine = self.get_db_engine()
-            self.connexion = engine.connect()
-
-        return self.connexion
-
     def _write_to_all_engines(self, write_fn):
         """Execute write_fn(engine) on every configured engine.
 
         Primary (index 0) failure is fatal and re-raised. Secondary
-        failures are logged as partial success and do not interrupt
-        processing.
+        failures are logged and do not interrupt processing.
 
         Parameters
         ----------
@@ -158,7 +167,7 @@ def _write_to_all_engines(self, write_fn):
             except Exception as e:
                 if i == 0:
                     raise
-                self.log.error("event=db_write_failed db=%s uri=...@%s error=%s", db_label, safe_uri, e)
+                self.log.warning("event=db_write_failed db=%s uri=...@%s error=%s", db_label, safe_uri, e)
 
         return primary_result
 
@@ -176,18 +185,14 @@ def get_table(self, tablename, schema=None) -> Table:
             Table: The table object representing the requested table.
 
         """
-        with warnings.catch_warnings():
-            warnings.simplefilter("ignore", category=sa_exc.SAWarning)
-
-            engine = self.get_db_engine()
-            metadata = MetaData()
-            metadata.reflect(engine)
-
-            if schema is not None and self.dialect == postgresql:
-                tbl = Table(tablename, metadata, autoload_with=self.get_con(), schema=schema)
-            else:
-                tbl = Table(tablename, metadata, autoload_with=self.get_con())
-            return tbl
+        engine = self.get_db_engine()
+        metadata = MetaData()
+        metadata.reflect(engine)
+        if schema is not None and self.dialect == postgresql:
+            with engine.connect() as con:
+                return Table(tablename, metadata, autoload_with=con, schema=schema)
+        with engine.connect() as con:
+            return Table(tablename, metadata, autoload_with=con)
 
     def execute(self, stm):
         """Execute SQL statement on the database.
@@ -201,11 +206,9 @@ def execute(self, stm):
             ResultProxy: The result of the executed statement.
 
         """
-        with warnings.catch_warnings():
-            warnings.simplefilter("ignore", category=sa_exc.SAWarning)
-            engine = self.get_db_engine()
-            with engine.connect() as con:
-                return con.execute(stm)
+        engine = self.get_db_engine()
+        with engine.connect() as con:
+            return con.execute(stm)
 
     def fetch_all_dict(self, stm) -> List[Dict]:
         """Fetch all rows from the database using the provided SQL statement.
@@ -219,22 +222,19 @@ def fetch_all_dict(self, stm) -> List[Dict]:
             List[Dict]: A list of dictionaries representing the fetched rows.
 
         """
-        with warnings.catch_warnings():
-            warnings.simplefilter("ignore", category=sa_exc.SAWarning)
-
-            engine = self.get_db_engine()
-            with engine.connect() as con:
+        engine = self.get_db_engine()
+        with engine.connect() as con:
 
-                queryset = con.execute(stm)
+            queryset = con.execute(stm)
 
-                rows = []
-                for row in queryset:
-                    d = row._asdict()
-                    rows.append(d)
+            rows = []
+            for row in queryset:
+                d = row._asdict()
+                rows.append(d)
 
-                return rows
+            return rows
 
-    def fetch_one_dict(self, stm) -> Dict:
+    def fetch_one_dict(self, stm) -> Optional[Dict]:
         """Fetch single row from the database and returns it as a dictionary.
 
         Args:
@@ -247,19 +247,16 @@ def fetch_one_dict(self, stm) -> Dict:
                 row is found.
 
         """
-        with warnings.catch_warnings():
-            warnings.simplefilter("ignore", category=sa_exc.SAWarning)
-
-            engine = self.get_db_engine()
-            with engine.connect() as con:
+        engine = self.get_db_engine()
+        with engine.connect() as con:
 
-                queryset = con.execute(stm).fetchone()
+            queryset = con.execute(stm).fetchone()
 
-                if queryset is not None:
-                    d = queryset._asdict()
-                    return d
-                else:
-                    return None
+            if queryset is not None:
+                d = queryset._asdict()
+                return d
+            else:
+                return None
 
     def fetch_scalar(self, stm) -> Any:
         """Execute SQL statement and returns the first column of the first row.
@@ -364,7 +361,10 @@ def execute_upsert(self, tbl: Table, df: pandas.DataFrame, commit_every: int = 1
             if update_cols:
                 upsert_stm = insert_stm.on_conflict_do_update(
                     index_elements=tbl.primary_key.columns,
-                    set_={k: getattr(insert_stm.excluded, k) for k in update_cols},
+                    set_={
+                        k: func.coalesce(getattr(insert_stm.excluded, k), getattr(tbl.c, k))
+                        for k in update_cols
+                    },
                 )
             else:
                 upsert_stm = insert_stm.on_conflict_do_nothing()
@@ -437,49 +437,6 @@ def _do_insert(engine, _stm=insert_stm):
 
         return affected_rows
 
-    def debug_query(self, stm, with_parameters=False) -> str:
-        """Returns SQL representation of the statement for debugging.
-
-        Args:
-        ----
-            stm: The statement to be converted to SQL.
-            with_parameters: Whether to include parameter values in the SQL.
-
-        Returns:
-        -------
-            str: The SQL representation of the statement.
-
-        """
-        sql = self.stm_to_str(stm, with_parameters)
-        return sql
-
-    def stm_to_str(self, stm, with_parameters=False):
-        """Convert SQLAlchemy statement object to a string representation.
-
-        Args:
-        ----
-            stm (sqlalchemy.sql.expression.Selectable): The SQLAlchemy
-                statement object to convert.
-            with_parameters (bool, optional): Whether to include parameter
-                values in the resulting SQL string. Defaults to False.
-
-        Returns:
-        -------
-            str: The string representation of the SQL query.
-
-        """
-        sql = str(
-            stm.compile(
-                dialect=self.dialect.dialect(),
-                compile_kwargs={"literal_binds": with_parameters},
-            )
-        )
-
-        # Remove new lines
-        sql = sql.replace("\n", " ").replace("\r", "")
-
-        return sql
-
     @staticmethod
     def _ensure_utc(dt: datetime) -> datetime:
         """Ensure datetime is in UTC (naive or aware).
diff --git a/python/lsst/consdb/transformed_efd/dao/exposure_efd.py b/python/lsst/consdb/transformed_efd/dao/exposure_efd.py
index eecf687c..4e5c963e 100644
--- a/python/lsst/consdb/transformed_efd/dao/exposure_efd.py
+++ b/python/lsst/consdb/transformed_efd/dao/exposure_efd.py
@@ -41,17 +41,17 @@ class ExposureEfdDao(DBBase):
 
     """
 
-    def __init__(self, db_uri: str, schema: str, logger: logging.Logger = None):
+    def __init__(self, db_uri: str | list[str], schema: str, logger: logging.Logger = None):
         """Initialize the `ExposureEfdDao` class.
 
         Args:
         ----
-            db_uri (str): The URI of the database.
+            db_uri (str | list[str]): Database URI(s). First is primary.
             schema (str): The schema name in the database.
             logger (logging.Logger, optional): Logger instance for logging.
 
         """
-        super(ExposureEfdDao, self).__init__(db_uri, schema, logger)
+        super().__init__(db_uri, schema, logger)
 
         self.tbl = self.get_table("exposure_efd", schema=schema)
 
@@ -67,7 +67,7 @@ def get_by_exposure_id(self, exposure_id: int):
             list: List of dictionaries for rows retrieved from the table.
 
         """
-        stm = select(self.tbl.c).where(and_(self.tbl.c.exposure_id == exposure_id))
+        stm = select(self.tbl.c).where(self.tbl.c.exposure_id == exposure_id)
 
         rows = self.fetch_all_dict(stm)
 
@@ -122,17 +122,17 @@ class ExposureEfdUnpivotedDao(DBBase):
 
     """
 
-    def __init__(self, db_uri: str, schema: str, logger: logging.Logger = None):
+    def __init__(self, db_uri: str | list[str], schema: str, logger: logging.Logger = None):
         """Initialize the `ExposureEfdUnpivotedDao` class.
 
         Args:
         ----
-            db_uri (str): The URI of the database.
+            db_uri (str | list[str]): Database URI(s). First is primary.
             schema (str): The schema name in the database.
             logger (logging.Logger, optional): Logger instance for logging.
 
         """
-        super(ExposureEfdUnpivotedDao, self).__init__(db_uri, schema, logger)
+        super().__init__(db_uri, schema, logger)
 
         self.tbl = self.get_table("exposure_efd_unpivoted", schema=schema)
 
diff --git a/python/lsst/consdb/transformed_efd/dao/influxdb.py b/python/lsst/consdb/transformed_efd/dao/influxdb.py
index 684215b0..1575db24 100644
--- a/python/lsst/consdb/transformed_efd/dao/influxdb.py
+++ b/python/lsst/consdb/transformed_efd/dao/influxdb.py
@@ -198,7 +198,7 @@ def sorter(prefix, val):
             ret[bfield].sort(key=part)
         return ret, n
 
-    def make_fields(self, fields: str, base_fields: [str, bytes]):
+    def make_fields(self, fields: str, base_fields: str | bytes | list[str]):
         """Construct a list of fields based on provided base field names.
 
         This function was adapted from the original implementation found at
@@ -212,7 +212,7 @@ def make_fields(self, fields: str, base_fields: [str, bytes]):
         ----------
         fields : str
             A string representing all fields.
-        base_fields : str or bytes
+        base_fields : str or bytes or list[str]
             The base field name(s) to expand.
 
         Returns
@@ -376,7 +376,12 @@ def merge_packed_time_series(
             # vals.update({"times": df["times"]})
             return pd.DataFrame(vals, index=df.index)
         except Exception as e:
-            self.log.error("event=influx_merge_packed_failed field=%s error=%s", f, e)
+            self.log.error(
+                "event=influx_merge_packed_failed base_fields=%s error=%s",
+                base_fields,
+                e,
+                exc_info=True,
+            )
             raise
 
     def _convert_index_format(self, x):
diff --git a/python/lsst/consdb/transformed_efd/dao/transformd.py b/python/lsst/consdb/transformed_efd/dao/transformd.py
index 31cb170c..317fc831 100644
--- a/python/lsst/consdb/transformed_efd/dao/transformd.py
+++ b/python/lsst/consdb/transformed_efd/dao/transformd.py
@@ -87,22 +87,6 @@ def _update_task_status(self, id: int, status: str, **kwargs) -> None:
             self.log.error("event=task_update_failed id=%s status=%s error=%s", id, status, e, exc_info=True)
             raise Exception(f"Error updating task: error={e}") from e
 
-    @staticmethod
-    def _ensure_utc_naive(dt: Optional[datetime]) -> Optional[datetime]:
-        """Convert datetime to UTC-naive format.
-
-        Args:
-            dt: Input datetime (naive or aware)
-
-        Returns:
-            UTC-naive datetime, or None if input was None
-        """
-        if dt is None:
-            return None
-        if dt.tzinfo is not None:
-            return dt.astimezone(timezone.utc).replace(tzinfo=None)
-        return dt  # Assume naive datetimes are UTC
-
     def select_by_id(self, id: int) -> Task:
         """Get task by ID.
 
@@ -302,7 +286,7 @@ def task_update_counts(self, id: int, exposures: int, visits1: int) -> None:
         visits1 : int
             Visit count
         """
-        self._update_task_status(id, "failed", exposures=exposures, visits1=visits1)
+        self._update_task_status(id, "running", exposures=exposures, visits1=visits1)
 
     def task_completed(self, id: int) -> None:
         """Mark task as completed.
@@ -313,6 +297,8 @@ def task_completed(self, id: int) -> None:
             Task ID to update
         """
         row = self.select_by_id(id)
+        if row is None:
+            raise ValueError(f"Task not found: id={id}")
         end_time = self._ensure_utc(datetime.now(timezone.utc))
         exec_time = (end_time - row["process_start_time"]).total_seconds()
         self._update_task_status(
@@ -334,6 +320,8 @@ def task_failed(self, id: int, error: str) -> None:
             Error message
         """
         row = self.select_by_id(id)
+        if row is None:
+            raise ValueError(f"Task not found: id={id}")
         end_time = self._ensure_utc(datetime.now(timezone.utc))
         exec_time = (end_time - row["process_start_time"]).total_seconds()
         self._update_task_status(
@@ -464,11 +452,11 @@ def select_failed(self, butler_repo: str, max_retries: Optional[int] = None) ->
         List[Task]
             List of task records
         """
-        query = select(self.tbl.c).where(self.tbl.c.status == "failed")
-        if max_retries:
-            query = query.where(
-                and_(self.tbl.c.retries <= max_retries, self.tbl.c.butler_repo == butler_repo)
-            )
+        query = select(self.tbl.c).where(
+            and_(self.tbl.c.status == "failed", self.tbl.c.butler_repo == butler_repo)
+        )
+        if max_retries is not None:
+            query = query.where(self.tbl.c.retries <= max_retries)
         query = query.order_by(self.tbl.c.created_at.asc(), self.tbl.c.id.asc())
         return self.fetch_all_dict(query)
 
diff --git a/python/lsst/consdb/transformed_efd/dao/visit_efd.py b/python/lsst/consdb/transformed_efd/dao/visit_efd.py
index 962f2e66..c0c4bbe5 100644
--- a/python/lsst/consdb/transformed_efd/dao/visit_efd.py
+++ b/python/lsst/consdb/transformed_efd/dao/visit_efd.py
@@ -41,17 +41,17 @@ class VisitEfdDao(DBBase):
 
     """
 
-    def __init__(self, db_uri: str, schema: str, logger: logging.Logger = None):
+    def __init__(self, db_uri: str | list[str], schema: str, logger: logging.Logger = None):
         """Initialize the `VisitEfdDao` class.
 
         Args:
         ----
-            db_uri (str): The URI of the database.
+            db_uri (str | list[str]): Database URI(s). First is primary.
             schema (str): The schema name in the database.
             logger (logging.Logger, optional): Logger instance for logging.
 
         """
-        super(VisitEfdDao, self).__init__(db_uri, schema, logger)
+        super().__init__(db_uri, schema, logger)
 
         self.tbl = self.get_table("visit1_efd", schema=schema)
 
@@ -67,7 +67,7 @@ def get_by_visit_id(self, visit_id: int):
             list: A list of rows matching the visit_id.
 
         """
-        stm = select(self.tbl.c).where(and_(self.tbl.c.visit_id == visit_id))
+        stm = select(self.tbl.c).where(self.tbl.c.visit_id == visit_id)
 
         rows = self.fetch_all_dict(stm)
 
@@ -121,17 +121,17 @@ class VisitEfdUnpivotedDao(DBBase):
 
     """
 
-    def __init__(self, db_uri: str, schema: str, logger: logging.Logger = None):
+    def __init__(self, db_uri: str | list[str], schema: str, logger: logging.Logger = None):
         """Initialize the `VisitEfdUnpivotedDao` class.
 
         Args:
         ----
-            db_uri (str): The URI of the database.
+            db_uri (str | list[str]): Database URI(s). First is primary.
             schema (str): The schema name in the database.
             logger (logging.Logger, optional): Logger instance for logging.
 
         """
-        super(VisitEfdUnpivotedDao, self).__init__(db_uri, schema, logger)
+        super().__init__(db_uri, schema, logger)
 
         self.tbl = self.get_table("visit1_efd_unpivoted", schema=schema)
 
diff --git a/python/lsst/consdb/transformed_efd/failure_monitor.py b/python/lsst/consdb/transformed_efd/failure_monitor.py
index d53024ef..0ca45001 100644
--- a/python/lsst/consdb/transformed_efd/failure_monitor.py
+++ b/python/lsst/consdb/transformed_efd/failure_monitor.py
@@ -212,6 +212,8 @@ def __init__(
         self.log = log
         self.schema = _schema_by_instrument(instrument)
         self.butler_instrument = _butler_instrument(instrument)
+        self.exp_dao = ExposureEfdDao(self.db_uri, self.schema, self.log)
+        self.vis_dao = VisitEfdDao(self.db_uri, self.schema, self.log)
 
     def run(self, args: Any) -> List[Dict[str, Any]]:
         window_days = max(1, int(getattr(args, "monitor_window_days", 7)))
@@ -223,11 +225,8 @@ def run(self, args: Any) -> List[Dict[str, Any]]:
         exp_records = self.butler_dao.exposures_by_day_obs(self.butler_instrument, day_start, day_end)
         vis_records = self.butler_dao.visits_by_day_obs(self.butler_instrument, day_start, day_end)
 
-        exp_dao = ExposureEfdDao(self.db_uri, self.schema, self.log)
-        vis_dao = VisitEfdDao(self.db_uri, self.schema, self.log)
-
-        existing_exp_ids = exp_dao.select_ids_by_day_obs(day_start, day_end)
-        existing_vis_ids = vis_dao.select_ids_by_day_obs(day_start, day_end)
+        existing_exp_ids = self.exp_dao.select_ids_by_day_obs(day_start, day_end)
+        existing_vis_ids = self.vis_dao.select_ids_by_day_obs(day_start, day_end)
 
         missing_exp = [record for record in exp_records if record["id"] not in existing_exp_ids]
         missing_vis = [record for record in vis_records if record["id"] not in existing_vis_ids]
diff --git a/python/lsst/consdb/transformed_efd/queue_manager.py b/python/lsst/consdb/transformed_efd/queue_manager.py
index 08107cd4..053e181b 100644
--- a/python/lsst/consdb/transformed_efd/queue_manager.py
+++ b/python/lsst/consdb/transformed_efd/queue_manager.py
@@ -109,10 +109,11 @@ def create_tasks(
         # Convert process_interval to TimeDelta once
         process_interval_td = TimeDelta(process_interval * 60, format="sec")
 
+        current_utc = Time.now().utc
+
         # Handle start_time logic
         if start_time is None:
             last_task = self.dao.select_last()
-            current_utc = Time.now().utc  # Get current time in UTC using Astropy's built-in method
 
             if last_task is None:
                 # First task: [current - interval, current]
@@ -123,7 +124,7 @@ def create_tasks(
 
         # Handle end_time logic
         if end_time is None:
-            end_time = current_utc  # Current UTC time
+            end_time = current_utc
             # Ensure at least one interval exists / deal with floating point
             time_delta = TimeDelta(round((end_time - start_time).to_value("sec"), 6), format="sec")
             if time_delta < process_interval_td:
@@ -312,7 +313,7 @@ def waiting_tasks(
 
         return task
 
-    def failed_tasks(self, butler_repo: str, max_retries: int = 3) -> Optional[dict]:
+    def failed_tasks(self, butler_repo: str, max_retries: int = 3) -> List[dict]:
         """Retrieves failed tasks.
 
         Args:
@@ -322,17 +323,17 @@ def failed_tasks(self, butler_repo: str, max_retries: int = 3) -> Optional[dict]
 
         Returns:
         -------
-        Optional[dict]: A dictionary representing the next task with the
-            specified status. If no task is found, returns `None`.
+        List[dict]: A list of dictionaries representing the failed tasks.
+            Empty list if no tasks found.
         """
 
-        task = self.dao.select_failed(butler_repo, max_retries=max_retries)
+        tasks = self.dao.select_failed(butler_repo, max_retries=max_retries)
 
         # add a retry key to act as a verifier to increment retries
-        if task:
-            task = [d.update({"retry": True}) or d for d in task]
+        if tasks:
+            tasks = [d.update({"retry": True}) or d for d in tasks]
 
-        return task
+        return tasks
 
     def get_task_by_interval(
         self, start_time: Time, end_time: Time, butler_repo: str, status: str
diff --git a/python/lsst/consdb/transformed_efd/summary.py b/python/lsst/consdb/transformed_efd/summary.py
index 3e26d045..d1f0b8d0 100644
--- a/python/lsst/consdb/transformed_efd/summary.py
+++ b/python/lsst/consdb/transformed_efd/summary.py
@@ -18,7 +18,8 @@
 #
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
-"""Provides the `Summary` class to perform the EDF transformations."""
+"""Provides the `Summary` class to perform the EFD transformations."""
+
 from typing import Optional, Union
 
 import numpy as np
diff --git a/python/lsst/consdb/transformed_efd/transform.py b/python/lsst/consdb/transformed_efd/transform.py
index 5f4bf623..f66d64d8 100644
--- a/python/lsst/consdb/transformed_efd/transform.py
+++ b/python/lsst/consdb/transformed_efd/transform.py
@@ -69,6 +69,11 @@ def __init__(
         self.efd = efd
         self.config = config
         self.commit_every = commit_every
+        # DAO cache — created lazily on first _store_results call.
+        self._exp_dao: ExposureEfdDao | None = None
+        self._vis_dao: VisitEfdDao | None = None
+        self._exp_unpivoted_dao: ExposureEfdUnpivotedDao | None = None
+        self._vis_unpivoted_dao: VisitEfdUnpivotedDao | None = None
 
     def get_schema_by_instrument(self, instrument: str) -> str:
         """Get the schema name for the given instrument."""
@@ -213,6 +218,17 @@ def _process_interval(
         log_context: Dict[str, Any] | None = None,
     ) -> Dict[str, Any]:
         """Process the given time interval for a specific instrument."""
+        # Only process exposures / visits completely within the time window.
+        # Partial overlaps are handled by the adjacent task.
+        full_exposures = [
+            e
+            for e in exposures
+            if e["timespan"].begin.utc >= start_time and e["timespan"].end.utc <= end_time
+        ]
+        full_visits = [
+            v for v in visits if v["timespan"].begin.utc >= start_time and v["timespan"].end.utc <= end_time
+        ]
+
         results = {
             "exposures": {
                 exp["id"]: {
@@ -220,7 +236,7 @@ def _process_interval(
                     "seq_num": exp["seq_num"],
                     "exposure_id": exp["id"],
                 }
-                for exp in exposures
+                for exp in full_exposures
             },
             "visits": {
                 vis["id"]: {
@@ -228,13 +244,13 @@ def _process_interval(
                     "seq_num": vis["seq_num"],
                     "visit_id": vis["id"],
                 }
-                for vis in visits
+                for vis in full_visits
             },
             "exposures_unpivoted": [],
             "visits_unpivoted": [],
         }
 
-        topic_interval = self._get_topic_interval(start_time, end_time, exposures, visits)
+        topic_interval = self._get_topic_interval(start_time, end_time, full_exposures, full_visits)
         self.log.debug("event=topic_interval start=%s end=%s", topic_interval[0], topic_interval[1])
         topics_map = self._map_topics()
         for key, topic in topics_map.items():
@@ -246,22 +262,32 @@ def _process_interval(
             if pre_aggregate_interval is not None and function in ["mean", "max", "min"]:
                 processing_topic["pre_aggregate_interval"] = pre_aggregate_interval
 
-            # apply offset to the topic interval if specified
+            # Apply start_offset to topic and propagate to all columns in this
+            # group so _compute_column_value can shift the per-exposure window.
             if start_offset is not None:
-                processing_topic["function_args"]["start_offset"] = start_offset
+                processing_topic.setdefault("function_args", {})["start_offset"] = start_offset
+                for col in processing_topic["columns"]:
+                    if col.get("function_args") is None:
+                        col["function_args"] = {}
+                    col["function_args"]["start_offset"] = start_offset
                 offset_topic_interval = topic_interval.copy()
                 offset_topic_interval[0] += astropy.time.TimeDelta(float(start_offset) * 3600, format="sec")
                 self._process_topic(
                     processing_topic,
                     offset_topic_interval,
-                    exposures,
-                    visits,
+                    full_exposures,
+                    full_visits,
                     results,
                     log_context=log_context,
                 )
             else:
                 self._process_topic(
-                    processing_topic, topic_interval, exposures, visits, results, log_context=log_context
+                    processing_topic,
+                    topic_interval,
+                    full_exposures,
+                    full_visits,
+                    results,
+                    log_context=log_context,
                 )
 
         return results
@@ -536,13 +562,26 @@ def _store_results(
 
         schema = self.get_schema_by_instrument(instrument)
 
+        # Lazy-init DAOs once per Transform instance.
+        if self._exp_dao is None:
+            self._exp_dao = ExposureEfdDao(db_uri=self.db_uri, schema=schema, logger=self.log)
+            self._vis_dao = VisitEfdDao(db_uri=self.db_uri, schema=schema, logger=self.log)
+            self._exp_unpivoted_dao = ExposureEfdUnpivotedDao(
+                db_uri=self.db_uri, schema=schema, logger=self.log
+            )
+            self._vis_unpivoted_dao = VisitEfdUnpivotedDao(db_uri=self.db_uri, schema=schema, logger=self.log)
+
+        assert self._exp_dao is not None  # lazy-init above guarantees this
+        assert self._vis_dao is not None
+        assert self._exp_unpivoted_dao is not None
+        assert self._vis_unpivoted_dao is not None
+
         # Store exposures
         if results["exposures"]:
             exposures_list = list(results["exposures"].values())
             df_exposures = pandas.DataFrame(exposures_list)
             if not df_exposures.empty:
-                exp_dao = ExposureEfdDao(db_uri=self.db_uri, schema=schema, logger=self.log)
-                affected_rows = exp_dao.upsert(df=df_exposures, commit_every=self.commit_every)
+                affected_rows = self._exp_dao.upsert(df=df_exposures, commit_every=self.commit_every)
                 count["exposures"] = affected_rows
                 self.log.debug("event=store_exposure_records affected_rows=%s", affected_rows)
 
@@ -551,8 +590,7 @@ def _store_results(
             visits_list = list(results["visits"].values())
             df_visits = pandas.DataFrame(visits_list)
             if not df_visits.empty:
-                vis_dao = VisitEfdDao(db_uri=self.db_uri, schema=schema, logger=self.log)
-                affected_rows = vis_dao.upsert(df=df_visits, commit_every=self.commit_every)
+                affected_rows = self._vis_dao.upsert(df=df_visits, commit_every=self.commit_every)
                 count["visits1"] = affected_rows
                 self.log.debug("event=store_visit_records affected_rows=%s", affected_rows)
 
@@ -560,10 +598,7 @@ def _store_results(
         if results["exposures_unpivoted"]:
             df_exposures_unpivoted = pandas.DataFrame(results["exposures_unpivoted"])
             if not df_exposures_unpivoted.empty:
-                exp_unpivoted_dao = ExposureEfdUnpivotedDao(
-                    db_uri=self.db_uri, schema=schema, logger=self.log
-                )
-                affected_rows = exp_unpivoted_dao.upsert(
+                affected_rows = self._exp_unpivoted_dao.upsert(
                     df=df_exposures_unpivoted, commit_every=self.commit_every
                 )
                 count["exposures_unpivoted"] = affected_rows
@@ -575,8 +610,7 @@ def _store_results(
         if results["visits_unpivoted"]:
             df_visits_unpivoted = pandas.DataFrame(results["visits_unpivoted"])
             if not df_visits_unpivoted.empty:
-                vis_unpivoted_dao = VisitEfdUnpivotedDao(db_uri=self.db_uri, schema=schema, logger=self.log)
-                affected_rows = vis_unpivoted_dao.upsert(
+                affected_rows = self._vis_unpivoted_dao.upsert(
                     df=df_visits_unpivoted, commit_every=self.commit_every
                 )
                 count["visits1_unpivoted"] = affected_rows
diff --git a/python/lsst/consdb/transformed_efd/transform_efd.py b/python/lsst/consdb/transformed_efd/transform_efd.py
index 2e4f0bdd..493f5c70 100644
--- a/python/lsst/consdb/transformed_efd/transform_efd.py
+++ b/python/lsst/consdb/transformed_efd/transform_efd.py
@@ -423,7 +423,7 @@ async def process_tasks(
                 log.warning("event=graceful_shutdown_between_tasks")
                 break  # Exit task loop
 
-        if shutdown_event.is_set():
+        if shutdown_event and shutdown_event.is_set():
             break  # Exit batch loop
 
     log.info(
diff --git a/tests/transformed_efd/test_config_model.py b/tests/transformed_efd/test_config_model.py
index 82abe491..9e887138 100644
--- a/tests/transformed_efd/test_config_model.py
+++ b/tests/transformed_efd/test_config_model.py
@@ -83,6 +83,22 @@ def test_column_with_empty_tables():
     assert column.tables == []
 
 
+def test_column_start_offset():
+    column = Column(
+        name="hexapod_u",
+        function="most_recent_value",
+        start_offset=-0.5,
+        datatype="float",
+        description="Test start_offset",
+        packed_series=False,
+        topics=[Topic(name="TestTopic", fields=[Field(name="u")])],
+    )
+    assert column.start_offset == -0.5
+
+    dumped = column.model_dump()
+    assert dumped["start_offset"] == -0.5
+
+
 def test_config_model():
     config = ConfigModel(
         version="1.0.0",
diff --git a/tests/transformed_efd/test_multi_db.py b/tests/transformed_efd/test_multi_db.py
index 971f91de..aad00417 100644
--- a/tests/transformed_efd/test_multi_db.py
+++ b/tests/transformed_efd/test_multi_db.py
@@ -72,7 +72,7 @@ def test_dialect_from_primary(self, logger):
         assert db.dialect is sqlite
 
     def test_unsupported_dialect_raises(self, logger):
-        with pytest.raises(Exception, match="not yet been implemented"):
+        with pytest.raises(NotImplementedError, match="not yet been implemented"):
             DBBase("mysql://localhost/db", logger=logger)