From 5c79be2e6d914b384d4689d895669582b65f5414 Mon Sep 17 00:00:00 2001 From: GadgetAI <232155002+labgadget015-dotcom@users.noreply.github.com> Date: Fri, 17 Jul 2026 07:48:39 +0000 Subject: [PATCH] chore: prune dead ledger functions + apply pre-commit import hygiene - Remove unused `_append()` and `_next_seq()` from decisions/ledger.py (record/transition both go through `_append_with_seq`; no live callers). - config_loader.py: pyupgrade Optional[X] -> X | None, drop redundant open() mode arg (pre-commit auto-fix). - test_recommendation_contract.py: blank-line import hygiene. 1001 unit tests pass; ruff clean on ledger.py. Co-Authored-By: Hermes Agent --- autopilot/config_loader.py | 10 ++++---- autopilot/decisions/ledger.py | 25 ------------------- .../tests/test_recommendation_contract.py | 1 + 3 files changed, 6 insertions(+), 30 deletions(-) diff --git a/autopilot/config_loader.py b/autopilot/config_loader.py index d00bae4..d8b0a4a 100644 --- a/autopilot/config_loader.py +++ b/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,7 +73,7 @@ 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) @@ -81,7 +81,7 @@ def load_config(path: Optional[str] = None) -> dict[str, Any]: 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"): diff --git a/autopilot/decisions/ledger.py b/autopilot/decisions/ledger.py index 10a2f35..2778d44 100644 --- a/autopilot/decisions/ledger.py +++ b/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) diff --git a/autopilot/tests/test_recommendation_contract.py b/autopilot/tests/test_recommendation_contract.py index d1724a4..b4f12c9 100644 --- a/autopilot/tests/test_recommendation_contract.py +++ b/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)