chore: prune dead ledger functions + pre-commit import hygiene

autopilot/config_loader.py

@@ -10,7 +10,7 @@
 
 import copy
 import os
-from typing import Any, Optional
+from typing import Any
 
 DEFAULT_CONFIG_PATH = os.environ.get("AUTOPILOT_CONFIG_PATH", "autopilot/config.yaml")
 
@@ -42,7 +42,7 @@
     },
 }
 
-_cache: Optional[dict[str, Any]] = None
+_cache: dict[str, Any] | None = None
 
 
 def _deep_merge(base: dict[str, Any], over: dict[str, Any]) -> dict[str, Any]:
@@ -54,7 +54,7 @@ def _deep_merge(base: dict[str, Any], over: dict[str, Any]) -> dict[str, Any]:
     return base
 
 
-def load_config(path: Optional[str] = None) -> dict[str, Any]:
+def load_config(path: str | None = None) -> dict[str, Any]:
     """Load and cache the merged config. Falls back to DEFAULTS if PyYAML
     is unavailable or the file is missing.
 
@@ -73,15 +73,15 @@ def load_config(path: Optional[str] = None) -> dict[str, Any]:
         pass
     else:
         if os.path.exists(path):
-            with open(path, "r", encoding="utf-8") as f:
+            with open(path, encoding="utf-8") as f:
                 loaded = yaml.safe_load(f) or {}
             if isinstance(loaded, dict):
                 _deep_merge(cfg, loaded)
     _cache = cfg
     return cfg
 
 
-def get_debounce_hours(status: str) -> Optional[int]:
+def get_debounce_hours(status: str) -> int | None:
     """Hours to suppress a re-raise, or None for never-repost."""
     policy = load_config()["recommendation_debounce"]["repost_policy"].get(status, {})
     if policy.get("never_repost"):

autopilot/decisions/ledger.py

@@ -58,13 +58,6 @@ def _load(path: str = DEFAULT_LEDGER_PATH) -> list[dict]:
                 continue
     return out
 
-
-def _next_seq(path: str = DEFAULT_LEDGER_PATH) -> int:
-    """Monotonic append counter — breaks sub-second timestamp ties so the
-    latest-written entry wins in latest_match()."""
-    return len(_load(path))
-
-
 def _lock(fd):
     """Exclusive lock the ledger file handle (POSIX). No-op elsewhere."""
     if _HAS_FCNTL:
@@ -76,24 +69,6 @@ def _unlock(fd):
         fcntl.flock(fd.fileno(), fcntl.LOCK_UN)
 
 
-def _append(entry: dict, path: str = DEFAULT_LEDGER_PATH) -> None:
-    """Append a JSON line under an exclusive file lock.
-
-    Holds the lock across _next_seq + write when called via _append_locked
-    so concurrent writers can't collide on sequence numbers or interleave
-    large entries.
-    """
-    os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
-    with open(path, "a", encoding="utf-8") as f:
-        _lock(f)
-        try:
-            f.write(json.dumps(entry, ensure_ascii=False) + "\n")
-            f.flush()
-            os.fsync(f.fileno())
-        finally:
-            _unlock(f)
-
-
 def _append_with_seq(entry: dict, path: str = DEFAULT_LEDGER_PATH) -> dict:
     """Atomic: assign seq under lock, then append. Prevents _next_seq races."""
     os.makedirs(os.path.dirname(path) or ".", exist_ok=True)

autopilot/tests/test_recommendation_contract.py

@@ -228,6 +228,7 @@ def test_transition_rejects_unknown_status():
 def test_concurrent_writers_do_not_lose_entries():
     """Parallel record() calls must all land — file locking prevents races."""
     import threading
+
     from decisions.ledger import record
 
     ledger = tempfile.NamedTemporaryFile(suffix=".jsonl", delete=False)