chore: enforce import placement via pre-commit, pre-push, and CI

AGENTS.md has required all imports at the top of the file for a long time, but
nothing checked it, so function-level imports accumulated. check_import_placement.py
fails a nested import unless a comment explains why (>=4 words; noqa-style
directives do not count), and treats a try/except ImportError guard as
self-explanatory.

Also fixes every existing violation, so the repository starts clean and any
future failure is something the change introduced.

Author: AnkushMalaker

.github/workflows/code-style.yml

@@ -0,0 +1,29 @@
+name: Code Style
+
+# Enforces the Chronicle-specific style rules that pre-commit runs locally, so
+# a `git push --no-verify` cannot land them.
+
+on:
+  pull_request:
+    paths:
+      - "**/*.py"
+      - ".github/workflows/code-style.yml"
+  push:
+    branches: [dev, main]
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+jobs:
+  import-placement:
+    name: Imports at top of file
+    runs-on: ubuntu-latest
+    timeout-minutes: 5
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+      - name: Check nested imports are explained
+        run: python3 scripts/check_import_placement.py

.pre-commit-config.yaml

@@ -1,4 +1,17 @@
+default_install_hook_types: [pre-commit, pre-push]
+
 repos:
+  # Code style rules that are Chronicle-specific (AGENTS.md → Code Style)
+  - repo: local
+    hooks:
+      - id: import-placement
+        name: imports at top of file (or explained)
+        entry: python3 scripts/check_import_placement.py
+        language: system
+        types: [python]
+        exclude: \.venv/
+        stages: [pre-commit, pre-push]
+
   # Code formatting
   - repo: https://github.com/psf/black
     rev: 26.5.1

AGENTS.md

@@ -607,6 +607,19 @@ tailscale ip -4
   - ALL imports must be at the top of the file after the docstring
   - Use lazy imports sparingly and only when absolutely necessary for circular import issues
   - Group imports: standard library, third-party, local imports
+  - **Enforced** by `scripts/check_import_placement.py`, which runs as a pre-commit
+    and pre-push hook and in the `Code Style` CI workflow. An import nested inside a
+    function or class fails the check unless a comment on the same line, or directly
+    above it, explains why (≥4 words; `# noqa`-style directives don't count). One
+    comment covers a contiguous run of imports:
+    ```python
+    def build_router():
+        # Imported here to break the circular import with plugins.router.
+        from advanced_omi_backend.plugins.router import PluginRouter
+    ```
+    An import guarded by `try/except ImportError` needs no comment — the structure
+    already says "optional dependency". The repository is currently clean, so any
+    failure is something the change introduced.
 - **Error Handling Guidelines**:
   - **Always raise errors, never silently ignore**: Use explicit error handling with proper exceptions rather than silent failures
   - **Understand data structures**: Research and understand input/response or class structure instead of adding defensive `hasattr()` checks

backends/advanced/scripts/evaluate_memory_executor.py

@@ -442,6 +442,7 @@ def _frontmatter(text: str) -> tuple[dict[str, Any] | None, str | None]:
     if len(boundaries) < 2:
         return None, "missing YAML frontmatter"
     try:
+        # Soft dependency: the except below covers hosts without it.
         import yaml
 
         value = yaml.safe_load(text[boundaries[0].end() : boundaries[1].start()]) or {}
@@ -668,6 +669,7 @@ def _safe_runtime_metadata(executor: str, modules: Mapping[str, Any]) -> dict[st
     metadata: dict[str, Any] = {"operation": "memory_write"}
     if executor == "direct":
         try:
+            # Soft dependency: the except below covers hosts without it.
             from advanced_omi_backend.model_registry import get_models_registry
 
             registry = get_models_registry()
@@ -693,6 +695,7 @@ def _safe_runtime_metadata(executor: str, modules: Mapping[str, Any]) -> dict[st
             {"executor_available": bool(available), "executor_detail": str(detail)}
         )
         try:
+            # Soft dependency: the except below covers hosts without it.
             from advanced_omi_backend.model_registry import get_models_registry
 
             registry = get_models_registry()
@@ -747,6 +750,7 @@ def _load_agent(executor: str, args: argparse.Namespace) -> tuple[type, dict[str
     backend = Path(__file__).resolve().parents[1]
     sys.path.insert(0, str(backend / "src"))
 
+    # Imported here: backends/advanced/src only joins sys.path at runtime (above).
     from advanced_omi_backend.services.memory import vault_lock
     from advanced_omi_backend.services.memory.agent import vault_tools
     from advanced_omi_backend.services.memory.agent.memory_agent import MemoryAgent
@@ -759,12 +763,14 @@ def _load_agent(executor: str, args: argparse.Namespace) -> tuple[type, dict[str
     if executor == "direct":
         agent_type = MemoryAgent
     elif executor == "pi":
+        # Imported here: backends/advanced/src only joins sys.path at runtime (above).
         from advanced_omi_backend.services.memory.agent import pi_agent
 
         pi_agent.vault_run_lock = _isolated_vault_lock
         modules["pi_agent"] = pi_agent
         agent_type = pi_agent.PiMemoryAgent
     else:
+        # Imported here: backends/advanced/src only joins sys.path at runtime (above).
         from advanced_omi_backend.services.memory.agent import codex_agent
 
         configured = codex_agent._validated_codex_settings()
@@ -796,6 +802,7 @@ async def _run_case(
     expected: Mapping[str, BenchmarkCase],
     initial_scaffold: Mapping[str, Any],
 ) -> dict[str, Any]:
+    # Imported here: backends/advanced/src only joins sys.path at runtime (above).
     from advanced_omi_backend.services.memory.conversation_note import (
         ConversationNoteError,
         canonicalize_conversation_note,
@@ -947,6 +954,7 @@ async def _run(args: argparse.Namespace) -> int:
 
     backend = Path(__file__).resolve().parents[1]
     sys.path.insert(0, str(backend / "src"))
+    # Imported here: backends/advanced/src only joins sys.path at runtime (above).
     from advanced_omi_backend.services.memory.vault_scaffold import seed_vault_scaffold
 
     seed_vault_scaffold(vault)

backends/advanced/scripts/evaluate_memory_retrieval.py

@@ -317,6 +317,7 @@ def _runtime_metadata(executor: str, pi_module: Any = None) -> dict[str, Any]:
     metadata: dict[str, Any] = {"operation": "memory_search"}
     if executor == "direct":
         try:
+            # Soft dependency: the except below covers hosts without it.
             from advanced_omi_backend.model_registry import get_models_registry
 
             registry = get_models_registry()
@@ -342,6 +343,7 @@ def _runtime_metadata(executor: str, pi_module: Any = None) -> dict[str, Any]:
             {"executor_available": bool(available), "executor_detail": str(detail)}
         )
         try:
+            # Soft dependency: the except below covers hosts without it.
             from advanced_omi_backend.model_registry import get_models_registry
 
             registry = get_models_registry()
@@ -375,9 +377,11 @@ def load_search_executor(
     backend = Path(__file__).resolve().parents[1]
     sys.path.insert(0, str(backend / "src"))
     if executor == "direct":
+        # Imported here: backends/advanced/src only joins sys.path at runtime (above).
         from advanced_omi_backend.services.memory.agent.memory_agent import search_vault
 
         return search_vault, _runtime_metadata(executor)
+    # Imported here: backends/advanced/src only joins sys.path at runtime (above).
     from advanced_omi_backend.services.memory.agent import pi_agent
 
     return pi_agent.search_vault_with_pi, _runtime_metadata(executor, pi_agent)

backends/advanced/scripts/tests/test_memory_executor_benchmark.py

@@ -11,6 +11,8 @@
 
 import pytest
 
+from advanced_omi_backend import model_registry
+
 SCRIPTS = Path(__file__).resolve().parents[1]
 sys.path.insert(0, str(SCRIPTS))
 sys.path.insert(0, str(SCRIPTS.parent / "src"))
@@ -392,8 +394,6 @@ async def run(self, _transcript, source_id, **_kwargs):
 
 
 def test_pi_runtime_metadata_uses_resolved_override_and_effective_limits(monkeypatch):
-    from advanced_omi_backend import model_registry
-
     ambient_operation = SimpleNamespace(
         model_name="ambient-memory-write-model",
         model_provider="ambient-provider",
@@ -432,8 +432,6 @@ def test_pi_runtime_metadata_uses_resolved_override_and_effective_limits(monkeyp
 
 
 def test_codex_runtime_metadata_uses_actual_codex_model(monkeypatch):
-    from advanced_omi_backend import model_registry
-
     ambient_operation = SimpleNamespace(
         model_name="ambient-memory-write-model",
         model_provider="ambient-provider",

backends/advanced/scripts/tests/test_memory_retrieval_benchmark.py

@@ -11,6 +11,8 @@
 
 import pytest
 
+from advanced_omi_backend import model_registry
+
 SCRIPTS = Path(__file__).resolve().parents[1]
 sys.path.insert(0, str(SCRIPTS))
 
@@ -62,8 +64,6 @@ def test_load_questions_requires_unique_ids(tmp_path):
 def test_direct_runtime_metadata_records_effective_operation_without_secrets(
     monkeypatch,
 ):
-    from advanced_omi_backend import model_registry
-
     operation = SimpleNamespace(
         model_name="qwen-upstream",
         model_def=SimpleNamespace(
@@ -102,8 +102,6 @@ def test_direct_runtime_metadata_records_effective_operation_without_secrets(
 
 
 def test_pi_runtime_metadata_records_resolved_override_without_secrets(monkeypatch):
-    from advanced_omi_backend import model_registry
-
     registry = object()
     monkeypatch.setattr(model_registry, "get_models_registry", lambda: registry)
     resolved = SimpleNamespace(

backends/advanced/worker_healthcheck.py

@@ -40,6 +40,7 @@ def main() -> int:
     # 1. Fresh RQ worker registrations. Redis can retain registrations from dead
     #    containers, so Worker.all() alone is not a liveness signal.
     try:
+        # Soft dependency: this healthcheck must still run without rq installed.
         from rq import Worker
 
         rq_count = sum(

clients.py

@@ -20,6 +20,7 @@
 import json
 import os
 import plistlib
+import shlex
 import shutil
 import socket
 import subprocess
@@ -226,7 +227,6 @@ def _install_linux(name: str, extras=()) -> None:
         raise RuntimeError(
             "no systemd user instance — on WSL set systemd=true in /etc/wsl.conf"
         )
-    import shlex
 
     uv = _find_uv()
     project = _component_project(name)

extras/asr-services/tests/test_batching.py

@@ -15,6 +15,7 @@
 """
 
 import difflib
+import json
 import os
 import sys
 import tempfile
@@ -28,6 +29,8 @@
 sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
 
 from common.batching import (
+    _clip_segments,
+    _clip_words,
     extract_context_tail,
     split_audio_file,
     stitch_transcription_results,
@@ -465,7 +468,6 @@ def test_words_clipped_at_boundary(self):
 
     def test_clip_helpers_drop_degenerate(self):
         """Segments/words that become start >= end after clipping are dropped."""
-        from common.batching import _clip_segments, _clip_words
 
         segs = [
             Segment(text="survives", start=10.0, end=20.0),
@@ -550,6 +552,8 @@ class TestBatchedTranscriptionQuality:
     @pytest.fixture(scope="class")
     def transcriber(self):
         """Load VibeVoice model once for all tests in this class."""
+        # Imported here so the module does not require torch: only this GPU-only
+        # class needs the VibeVoice provider.
         from providers.vibevoice.transcriber import VibeVoiceTranscriber
 
         t = VibeVoiceTranscriber()
@@ -693,8 +697,6 @@ def ground_truth(self):
                 f"Ground truth fixture not found: {self.FIXTURE_PATH}\n"
                 f"Run: uv run python tests/capture_vibevoice_ground_truth.py"
             )
-        import json
-
         with open(self.FIXTURE_PATH) as f:
             return json.load(f)