Enforce PY-1 import contracts with import-linter

.github/workflows/ci.yml

@@ -157,6 +157,45 @@ jobs:
       - name: Check Alembic metadata does not drift (db/schema.py)
         run: uv run python scripts/check_db_schema_drift.py
 
+  python_lint_import_dependency_boundaries:
+    runs-on: ubuntu-latest
+    timeout-minutes: 15
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v4
+
+      - name: Set up Python 3.12
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+
+      - name: Cache uv packages and lockfile
+        uses: actions/cache@v4
+        with:
+          path: |
+            ~/.cache/uv
+            uv.lock
+          key: ${{ runner.os }}-uv-python-lint-import-dependency-boundaries-3.12-${{ hashFiles('pyproject.toml', 'uv.lock') }}
+          restore-keys: |
+            ${{ runner.os }}-uv-python-lint-import-dependency-boundaries-3.12-
+
+      - name: Install uv
+        uses: astral-sh/setup-uv@v4
+        with:
+          version: "latest"
+
+      - name: Resolve deps (create/refresh uv.lock)
+        run: uv lock
+
+      - name: Install dependencies
+        run: uv sync --extra test
+
+      - name: Add project directory to PYTHONPATH
+        run: echo "PYTHONPATH=${{ github.workspace }}" >> $GITHUB_ENV
+
+      - name: python_lint_import_dependency_boundaries
+        run: uv run lint-imports --config pyproject.toml
+
   test:
     runs-on: ubuntu-latest
     timeout-minutes: 30

.pre-commit-config.yaml

@@ -40,6 +40,11 @@ repos:
         entry: bash -lc "root=$(git rev-parse --show-toplevel) && cd \"$root\" && PYTHONPATH=\"$root\" uv run --extra test pyright ."
         language: system
         pass_filenames: false
+      - id: python_lint_import_dependency_boundaries
+        name: python_lint_import_dependency_boundaries
+        entry: bash -lc "root=$(git rev-parse --show-toplevel) && cd \"$root\" && PYTHONPATH=\"$root\" uv run --extra test lint-imports --config pyproject.toml"
+        language: system
+        pass_filenames: false
       - id: oxlint
         name: oxlint + React Doctor (ui)
         entry: bash -lc "cd ui && (test -d node_modules || npm install) && npx oxlint . --react-plugin && node scripts/check-react-doctor.mjs"

docs/SUGGESTED_CUSTOM_LINTERS.md

@@ -60,8 +60,9 @@ These are good candidates for linting because they are:
 
 #### How (implementation) (PY-1)
 
-- Add `import-linter` (or `grimp`-based checks) and a `contracts` file (e.g. `lint/import_contracts.ini`).
-- Wire into pre-commit + CI (`uv run import-linter`).
+- Add `import-linter` and configure contracts in `pyproject.toml` under `[tool.importlinter]`.
+- Run locally: `uv run lint-imports --config pyproject.toml`.
+- Wire into pre-commit + CI (pre-commit hook + CI job run `uv run lint-imports --config pyproject.toml`).
 
 #### Known failing examples in this repo (PY-1)
 

pyproject.toml

@@ -35,6 +35,7 @@ test = [
     "pyright>=1.1.0",
     "complexipy>=5.0.0",
     "pre-commit>=4.0.0",
+    "import-linter>=2.0.0",
 ]
 
 [build-system]
@@ -108,6 +109,56 @@ pythonVersion = "3.12"
 typeCheckingMode = "basic"
 reportMissingTypeStubs = false
 
+[tool.importlinter]
+root_packages = ["simulation", "db", "lib", "feeds", "ai", "jobs", "ml_tooling"]
+include_external_packages = false
+exclude_type_checking_imports = true
+
+[[tool.importlinter.contracts]]
+name = "Domain purity: simulation.core.models"
+type = "forbidden"
+source_modules = ["simulation.core.models"]
+forbidden_modules = [
+    "simulation.api",
+    "db",
+    "feeds",
+    "ai",
+    "jobs",
+    "ml_tooling",
+    # simulation.core.* excluding simulation.core.models
+    "simulation.core.action_generators",
+    "simulation.core.action_history",
+    "simulation.core.action_policy",
+    "simulation.core.command_service",
+    "simulation.core.engine",
+    "simulation.core.exceptions",
+    "simulation.core.factories",
+    "simulation.core.handle_utils",
+    "simulation.core.metrics",
+    "simulation.core.query_service",
+    "simulation.core.utils",
+    "simulation.core.validators",
+]
+
+[[tool.importlinter.contracts]]
+name = "lib is leaf"
+type = "forbidden"
+source_modules = ["lib"]
+forbidden_modules = ["simulation", "db", "feeds", "ai", "jobs", "ml_tooling"]
+
+[[tool.importlinter.contracts]]
+name = "ml_tooling constraints"
+type = "forbidden"
+source_modules = ["ml_tooling"]
+forbidden_modules = ["simulation", "db", "feeds", "ai", "jobs"]
+
+[[tool.importlinter.contracts]]
+name = "API routes boundary: simulation.api.routes"
+type = "forbidden"
+source_modules = ["simulation.api.routes"]
+forbidden_modules = ["simulation.core", "db", "feeds", "ai", "jobs", "ml_tooling"]
+allow_indirect_imports = true
+
 
 [tool.alembic]
 

simulation/api/errors.py

@@ -0,0 +1,26 @@
+"""API-layer exceptions for route error mapping.
+
+Routes should not import core/db modules directly; they should translate failures from
+services into consistent HTTP responses. These exceptions are intentionally simple and
+carry only the data the route needs to build an error payload.
+"""
+
+from __future__ import annotations
+
+
+class ApiHandleAlreadyExistsError(Exception):
+    def __init__(self, handle: str):
+        super().__init__(handle)
+        self.handle = handle
+
+
+class ApiRunNotFoundError(Exception):
+    def __init__(self, run_id: str):
+        super().__init__(run_id)
+        self.run_id = run_id
+
+
+class ApiRunCreationFailedError(Exception):
+    def __init__(self, message: str):
+        super().__init__(message)
+        self.message = message

simulation/api/routes/simulation.py

@@ -16,6 +16,11 @@
     MAX_AGENT_LIST_LIMIT,
 )
 from simulation.api.dependencies.auth import require_auth
+from simulation.api.errors import (
+    ApiHandleAlreadyExistsError,
+    ApiRunCreationFailedError,
+    ApiRunNotFoundError,
+)
 from simulation.api.schemas.simulation import (
     AgentSchema,
     CreateAgentRequest,
@@ -31,18 +36,14 @@
 )
 from simulation.api.services.agent_command_service import create_agent
 from simulation.api.services.agent_query_service import list_agents
+from simulation.api.services.metadata_service import list_feed_algorithms, list_metrics
 from simulation.api.services.run_execution_service import execute
 from simulation.api.services.run_query_service import (
     get_posts_by_uris,
     get_run_details,
     get_turns_for_run,
     list_runs,
 )
-from simulation.core.exceptions import (
-    HandleAlreadyExistsError,
-    RunNotFoundError,
-    SimulationRunFailure,
-)
 
 logger = logging.getLogger(__name__)
 
@@ -238,39 +239,13 @@ async def get_simulation_run_turns(
     return await _execute_get_simulation_run_turns(request, run_id=run_id)
 
 
-def _get_feed_algorithms_list() -> list[FeedAlgorithmSchema]:
-    """Return feed algorithms with metadata for the API."""
-    from feeds.algorithms import get_registered_algorithms
-
-    return [
-        FeedAlgorithmSchema(id=alg_id, **meta.model_dump())
-        for alg_id, meta in get_registered_algorithms()
-    ]
-
-
-def _get_metrics_list() -> list[MetricSchema]:
-    """Return metrics with metadata for the API."""
-    from simulation.core.metrics.defaults import get_registered_metrics_metadata
-
-    return [
-        MetricSchema(
-            key=key,
-            display_name=display_name,
-            description=description,
-            scope=scope,
-            author=author,
-        )
-        for key, display_name, description, scope, author in get_registered_metrics_metadata()
-    ]
-
-
 @timed(attach_attr="duration_ms", log_level=None)
 async def _execute_get_metrics(
     request: Request,
 ) -> list[MetricSchema] | Response:
     """Fetch metrics and convert unexpected failures to HTTP responses."""
     try:
-        return await asyncio.to_thread(_get_metrics_list)
+        return await asyncio.to_thread(list_metrics)
     except Exception:
         logger.exception("Unexpected error while listing metrics")
         return _error_response(
@@ -287,7 +262,7 @@ async def _execute_get_feed_algorithms(
 ) -> list[FeedAlgorithmSchema] | Response:
     """Fetch feed algorithms and convert unexpected failures to HTTP responses."""
     try:
-        return await asyncio.to_thread(_get_feed_algorithms_list)
+        return await asyncio.to_thread(list_feed_algorithms)
     except Exception:
         logger.exception("Unexpected error while listing feed algorithms")
         return _error_response(
@@ -361,7 +336,7 @@ async def _execute_post_simulation_agents(
     """Create agent and convert known failures to HTTP responses."""
     try:
         return await asyncio.to_thread(create_agent, body)
-    except HandleAlreadyExistsError as e:
+    except ApiHandleAlreadyExistsError as e:
         return _error_response(
             status_code=409,
             code="HANDLE_ALREADY_EXISTS",
@@ -417,12 +392,12 @@ async def _execute_simulation_run(
             request=body,
             engine=engine,
         )
-    except SimulationRunFailure as e:
+    except ApiRunCreationFailedError as e:
         logger.exception("Simulation run failed before run creation")
         return _error_response(
             status_code=500,
             code="RUN_CREATION_FAILED",
-            message=e.args[0] if e.args else "Run creation or status update failed",
+            message=e.message,
             detail=None,
         )
     except Exception:
@@ -444,7 +419,7 @@ async def _execute_get_simulation_run_turns(
     try:
         engine = request.app.state.engine
         return await asyncio.to_thread(get_turns_for_run, run_id=run_id, engine=engine)
-    except RunNotFoundError as e:
+    except ApiRunNotFoundError as e:
         return _error_response(
             status_code=404,
             code="RUN_NOT_FOUND",
@@ -480,7 +455,7 @@ async def _execute_get_simulation_run(
             run_id=run_id,
             engine=engine,
         )
-    except RunNotFoundError as e:
+    except ApiRunNotFoundError as e:
         return _error_response(
             status_code=404,
             code="RUN_NOT_FOUND",

simulation/api/services/agent_command_service.py

@@ -15,8 +15,8 @@
     create_sqlite_user_agent_profile_metadata_repository,
 )
 from lib.timestamp_utils import get_current_timestamp
+from simulation.api.errors import ApiHandleAlreadyExistsError
 from simulation.api.schemas.simulation import AgentSchema, CreateAgentRequest
-from simulation.core.exceptions import HandleAlreadyExistsError
 from simulation.core.handle_utils import normalize_handle
 from simulation.core.models.agent import Agent, PersonaSource
 from simulation.core.models.agent_bio import AgentBio, PersonaBioSource
@@ -59,7 +59,7 @@ def create_agent(
     # before the below context manager for writing the agent to the database.
     # This is a known issue, and we'll revisit this in the future.
     if agent_repo.get_agent_by_handle(handle) is not None:
-        raise HandleAlreadyExistsError(handle)
+        raise ApiHandleAlreadyExistsError(handle)
 
     agent_id = _generate_agent_id()
     now = get_current_timestamp()

simulation/api/services/metadata_service.py

@@ -0,0 +1,31 @@
+"""Service helpers for listing simulation metadata (algorithms, metrics)."""
+
+from __future__ import annotations
+
+from simulation.api.schemas.simulation import FeedAlgorithmSchema, MetricSchema
+
+
+def list_feed_algorithms() -> list[FeedAlgorithmSchema]:
+    """Return registered feed algorithms with metadata for the API."""
+    from feeds.algorithms import get_registered_algorithms
+
+    return [
+        FeedAlgorithmSchema(id=alg_id, **meta.model_dump())
+        for alg_id, meta in get_registered_algorithms()
+    ]
+
+
+def list_metrics() -> list[MetricSchema]:
+    """Return registered metrics with metadata for the API."""
+    from simulation.core.metrics.defaults import get_registered_metrics_metadata
+
+    return [
+        MetricSchema(
+            key=key,
+            display_name=display_name,
+            description=description,
+            scope=scope,
+            author=author,
+        )
+        for key, display_name, description, scope, author in get_registered_metrics_metadata()
+    ]

simulation/api/services/run_execution_service.py

@@ -3,6 +3,7 @@
 from collections.abc import Iterable
 
 from lib.timestamp_utils import get_current_timestamp
+from simulation.api.errors import ApiRunCreationFailedError
 from simulation.api.schemas.simulation import (
     ErrorDetail,
     RunRequest,
@@ -38,7 +39,8 @@ def execute(
         run: Run = engine.execute_run(run_config=run_config)
     except SimulationRunFailure as e:
         if e.run_id is None:
-            raise
+            message = e.args[0] if e.args else "Run creation or status update failed"
+            raise ApiRunCreationFailedError(message) from e
         metadata_list = engine.list_turn_metadata(e.run_id)
         turn_metrics_list = engine.list_turn_metrics(e.run_id)
         _validate_turn_data_consistency(

simulation/api/services/run_query_service.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 from lib.validation_decorators import validate_inputs
+from simulation.api.errors import ApiRunNotFoundError
 from simulation.api.schemas.simulation import (
     FeedSchema,
     PostSchema,
@@ -13,12 +14,11 @@
     TurnSchema,
 )
 from simulation.core.engine import SimulationEngine
-from simulation.core.exceptions import RunNotFoundError
 from simulation.core.models.metrics import RunMetrics, TurnMetrics
 from simulation.core.models.posts import BlueskyFeedPost
 from simulation.core.models.runs import Run
 from simulation.core.models.turns import TurnMetadata
-from simulation.core.validators import validate_run_exists, validate_run_id
+from simulation.core.validators import validate_run_id
 
 MAX_UNFILTERED_POSTS: int = 500
 
@@ -50,7 +50,7 @@ def get_turns_for_run(
     validated_run_id = validate_run_id(run_id)
     run = engine.get_run(validated_run_id)
     if run is None:
-        raise RunNotFoundError(validated_run_id)
+        raise ApiRunNotFoundError(validated_run_id)
 
     metadata_list = engine.list_turn_metadata(validated_run_id)
     metadata_sorted = sorted(metadata_list, key=lambda m: m.turn_number)
@@ -124,7 +124,8 @@ def get_run_details(*, run_id: str, engine: SimulationEngine) -> RunDetailsRespo
         RunNotFoundError: If the run does not exist.
     """
     run = engine.get_run(run_id)
-    run = validate_run_exists(run=run, run_id=run_id)
+    if run is None:
+        raise ApiRunNotFoundError(run_id)
 
     metadata_list: list[TurnMetadata] = engine.list_turn_metadata(run_id)
     turn_metrics_list: list[TurnMetrics] = engine.list_turn_metrics(run_id)