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)