SEP-1655: Move the remaining Nomad-specific vocabulary out of app/tasks/models.py

app/sep/apps/framework/spec.py

@@ -69,8 +69,8 @@
 from app.sep.deps import get_created_entity, InventoryAPI
 from app.sep.inventory import CreatedEntity, CreatedService
 from app.sep.models import SyncInventoryEntityTypeEnum
+from app.tasks.execution.executors.nomad.constants import RUN_SCRIPT_OUTPUT_FILES_PATH
 from app.tasks.models import (
-    RUN_SCRIPT_OUTPUT_FILES_PATH,
     TaskBackendEnum,
     TaskWrite,
 )

app/tasks/crud.py

@@ -305,24 +305,24 @@ class TaskHistoryManager(BaseSQLModelManager):
     Model = TaskHistory
 
     @classmethod
-    async def get_log_allocation_epoch(
+    async def get_log_producer_epoch(
         cls,
         session: AsyncSession,
         task_history_id: int,
         *,
         for_update: bool = False,
     ) -> int:
-        """Return the task-level log allocation-epoch high-water mark.
+        """Return the task-level log producer-epoch high-water mark.
 
         The log writer consults this on the first-insert path — before any
         per-stream ``TaskHistoryLogState`` row exists — to discard writes from a
-        superseded allocation. A missing row yields the ``0`` sentinel so the
+        superseded producer. A missing row yields the ``0`` sentinel so the
         caller trusts the write.
 
         With ``for_update`` the ``TaskHistory`` row is locked (``SELECT ... FOR
         UPDATE``) and the lock is held until the caller's transaction ends. This
         serialises the first-insert discard decision against
-        :meth:`bump_log_allocation_epoch` (stamped during a frontier reset) so a
+        :meth:`bump_log_producer_epoch` (stamped during a frontier reset) so a
         reset cannot commit a newer epoch in the window between the guard read
         and the row insert. Both the first-insert writer and the frontier reset
         acquire this same row first, giving a consistent lock order. On SQLite
@@ -333,10 +333,10 @@ async def get_log_allocation_epoch(
         :param task_history_id: The ``TaskHistory`` identifier.
         :param for_update: Whether to lock the row for the duration of the
             transaction.
-        :return: The stored ``log_allocation_epoch``, or ``0`` when the row is
+        :return: The stored ``log_producer_epoch``, or ``0`` when the row is
             absent.
         """
-        query = select(TaskHistory.log_allocation_epoch).where(
+        query = select(TaskHistory.log_producer_epoch).where(
             col(TaskHistory.id) == task_history_id
         )
         if for_update:
@@ -346,14 +346,14 @@ async def get_log_allocation_epoch(
         return epoch if epoch is not None else 0
 
     @classmethod
-    async def bump_log_allocation_epoch(
+    async def bump_log_producer_epoch(
         cls,
         session: AsyncSession,
         task_history_id: int,
         *,
-        new_allocation_epoch: int,
+        new_producer_epoch: int,
     ) -> None:
-        """Advance the task-level allocation-epoch high-water mark monotonically.
+        """Advance the task-level producer-epoch high-water mark monotonically.
 
         Stamped wherever the log frontier is reset. The ``< new`` guard makes the
         update monotonic: an out-of-order or stale reset carrying a smaller epoch
@@ -364,16 +364,16 @@ async def bump_log_allocation_epoch(
         :param session: The SQLAlchemy asynchronous session to use for query
             execution.
         :param task_history_id: The ``TaskHistory`` identifier.
-        :param new_allocation_epoch: The ``CreateIndex`` of the allocation the
-            frontier is being reset onto.
+        :param new_producer_epoch: The producer epoch the frontier is being
+            reset onto (for example a Nomad allocation ``CreateIndex``).
         """
         stmt = (
             update(TaskHistory)
             .where(
                 col(TaskHistory.id) == task_history_id,
-                col(TaskHistory.log_allocation_epoch) < new_allocation_epoch,
+                col(TaskHistory.log_producer_epoch) < new_producer_epoch,
             )
-            .values(log_allocation_epoch=new_allocation_epoch)
+            .values(log_producer_epoch=new_producer_epoch)
         )
         await session.exec(stmt)
 
@@ -1172,8 +1172,8 @@ def build_default(
             stream=stream,
             persisted_offset=0,
             producer_offset=0,
-            nomad_offset=0,
-            allocation_epoch=0,
+            producer_fetch_offset=0,
+            producer_epoch=0,
             staging=b"",
             staging_updated_at=utc_now(),
             version=0,
@@ -1205,33 +1205,33 @@ async def reset_allocation_frontier(
         session: AsyncSession,
         task_history_id: int,
         *,
-        new_allocation_epoch: int,
+        new_producer_epoch: int,
     ) -> None:
         """Reset both cursors to zero and stamp the new epoch for every stream.
 
-        Called when Nomad reschedules a task to a follow-up allocation: the
-        new allocation's log file starts at byte 0, so both allocation-relative
-        cursors (``producer_offset`` and ``nomad_offset``) must be cleared in
-        the database before the writer dedups or fetches against them, and
-        ``allocation_epoch`` must be advanced to the new allocation's
-        ``CreateIndex`` so stale-allocation writes are discarded by the write
-        guard. Bumps ``version`` so concurrent writers re-read the row.
+        Called when the executor reschedules onto a follow-up producer (for
+        example a Nomad allocation): the new producer's log file starts at byte
+        0, so both producer-relative cursors (``producer_offset`` and
+        ``producer_fetch_offset``) must be cleared in the database before the
+        writer dedups or fetches against them, and ``producer_epoch`` must be
+        advanced so stale-producer writes are discarded by the write guard.
+        Bumps ``version`` so concurrent writers re-read the row.
 
         :param session: The SQLAlchemy asynchronous session.
         :type session: AsyncSession
         :param task_history_id: The ``TaskHistory`` identifier whose state
             rows should be reset.
         :type task_history_id: int
-        :param new_allocation_epoch: The ``CreateIndex`` of the allocation the
-            frontier is being reset onto.
+        :param new_producer_epoch: The producer epoch the frontier is being
+            reset onto (for example a Nomad allocation ``CreateIndex``).
         """
         stmt = (
             update(TaskHistoryLogState)
             .where(col(TaskHistoryLogState.task_history_id) == task_history_id)
             .values(
                 producer_offset=0,
-                nomad_offset=0,
-                allocation_epoch=new_allocation_epoch,
+                producer_fetch_offset=0,
+                producer_epoch=new_producer_epoch,
                 version=col(TaskHistoryLogState.version) + 1,
                 updated_at=utc_now(),
             )
@@ -1248,8 +1248,8 @@ async def insert_row_idempotent(
         stream: TaskLogType,
         persisted_offset: int,
         producer_offset: int,
-        nomad_offset: int,
-        allocation_epoch: int,
+        producer_fetch_offset: int,
+        producer_epoch: int,
         staging: bytes,
         version: int,
         now: datetime,
@@ -1272,10 +1272,12 @@ async def insert_row_idempotent(
         :param persisted_offset: The user-facing byte offset already persisted.
         :type persisted_offset: int
         :param producer_offset: The producer-relative byte offset already
-            consumed from the current allocation.
+            consumed from the current producer epoch.
         :type producer_offset: int
-        :param nomad_offset: The raw Nomad-space fetch offset for the next read.
-        :param allocation_epoch: The Nomad ``CreateIndex`` the cursors belong to.
+        :param producer_fetch_offset: The raw producer-space fetch offset for
+            the next read.
+        :param producer_epoch: The producer-generation stamp the cursors belong
+            to (for example a Nomad allocation ``CreateIndex``).
         :param staging: Bytes pending flush to the chunk store.
         :type staging: bytes
         :param version: The initial optimistic-locking version counter.
@@ -1292,8 +1294,8 @@ async def insert_row_idempotent(
             stream=stream,
             persisted_offset=persisted_offset,
             producer_offset=producer_offset,
-            nomad_offset=nomad_offset,
-            allocation_epoch=allocation_epoch,
+            producer_fetch_offset=producer_fetch_offset,
+            producer_epoch=producer_epoch,
             staging=staging,
             staging_updated_at=now,
             version=version,
@@ -1314,8 +1316,8 @@ async def update_row_if_version(
         new_version: int,
         persisted_offset: int,
         producer_offset: int,
-        nomad_offset: int,
-        allocation_epoch: int,
+        producer_fetch_offset: int,
+        producer_epoch: int,
         staging: bytes,
         now: datetime,
     ) -> bool:
@@ -1343,9 +1345,10 @@ async def update_row_if_version(
         :type persisted_offset: int
         :param producer_offset: The updated producer-relative offset.
         :type producer_offset: int
-        :param nomad_offset: The updated raw Nomad-space fetch offset.
-        :param allocation_epoch: The updated Nomad ``CreateIndex`` the cursors
-            belong to.
+        :param producer_fetch_offset: The updated raw producer-space fetch
+            offset.
+        :param producer_epoch: The updated producer-generation stamp the
+            cursors belong to (for example a Nomad allocation ``CreateIndex``).
         :param staging: The updated staging bytes buffer.
         :type staging: bytes
         :param now: The update timestamp used for the audit columns.
@@ -1365,8 +1368,8 @@ async def update_row_if_version(
             .values(
                 persisted_offset=persisted_offset,
                 producer_offset=producer_offset,
-                nomad_offset=nomad_offset,
-                allocation_epoch=allocation_epoch,
+                producer_fetch_offset=producer_fetch_offset,
+                producer_epoch=producer_epoch,
                 staging=staging,
                 staging_updated_at=now,
                 version=new_version,

app/tasks/db/seed.py

@@ -35,10 +35,12 @@
 from app.tasks.crud import TaskManager
 from app.tasks.db import get_async_session_maker
 from app.tasks.db.engine import engine
-from app.tasks.models import (
+from app.tasks.execution.executors.nomad.constants import (
     CHECK_NOMAD_CERT_EXPIRY_TASK_NAME,
-    INVENTORY_SYNC_TASK_NAME,
     RUN_SCRIPT_OUTPUT_FILES_PATH,
+)
+from app.tasks.models import (
+    INVENTORY_SYNC_TASK_NAME,
     SYNC_RUNNING_TASKS_TASK_NAME,
     SYSTEM_USER,
     Task,

app/tasks/execution/executors/nomad/__init__.py

@@ -13,4 +13,31 @@
 # You should have received a copy of the GNU Affero General Public License
 # along with this program. If not, see <https://www.gnu.org/licenses/>.
 
-from app.tasks.execution.executors.nomad.models import NomadExecutor
+"""Nomad task executor package.
+
+``NomadExecutor`` is resolved lazily via ``__getattr__`` so submodule imports
+do not deadlock against ``app.tasks.config`` (which imports ``NomadExecutor``
+back from this package).
+"""
+
+__all__ = ["NomadExecutor"]
+
+
+def __getattr__(name: str) -> object:
+    """Resolve ``NomadExecutor`` on first attribute access.
+
+    :param name: The attribute being read.
+    :return: The resolved attribute.
+    :raises AttributeError: If ``name`` is not exported by this package.
+    """
+    if name == "NomadExecutor":
+        from app.tasks.execution.executors.nomad.models import NomadExecutor
+
+        globals()[name] = NomadExecutor
+        return NomadExecutor
+    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+
+
+def __dir__() -> list[str]:
+    """Return attribute names for ``dir()``, including the lazy export."""
+    return sorted({*globals(), *__all__})

app/tasks/execution/executors/nomad/constants.py

@@ -0,0 +1,33 @@
+# Copyright (C) 2026 Percona LLC
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program. If not, see <https://www.gnu.org/licenses/>.
+
+"""Define Nomad-owned constants for allocation layout and Nomad-only system tasks.
+
+Imports nothing from the rest of the tasks or sep packages so seed and framework
+specs can use these values without the Nomad executor import graph.
+"""
+
+#: Allocation-relative output-files directory of every job spec that pins its
+#: ``run-script`` task's ``work_dir`` to ``${NOMAD_TASK_DIR}/output_files``
+#: (``run-python``, ``exec-artifact``, ``exec-python-artifact``). It is the
+#: :attr:`~app.tasks.models.TaskBase.output_files_path` those specs run under,
+#: so a payload's working directory and the path SEP reads its files back from
+#: are the same place. ``run-command`` pins no ``work_dir`` and so has no
+#: output-files path. The ``run-script/local/`` prefix is the Nomad allocation
+#: layout (``run-script`` task name + ``${NOMAD_TASK_DIR}``).
+RUN_SCRIPT_OUTPUT_FILES_PATH = "run-script/local/output_files"
+
+#: Seeded name of the Nomad-only periodic task that checks TLS cert expiry.
+CHECK_NOMAD_CERT_EXPIRY_TASK_NAME = "tasks__check_nomad_cert_expiry"