round-12: ruff hygiene on migration code (import sort, narrow store except)

- test_chv_muse_round12.py: drop unused socket/time imports, drop
  unnecessary sys.path.insert (P3 pattern needs none), sort method-local
  imports (activity/capabilities, approval_card/channel/sentinel)
- audit_trail.py: single-line method-local imports (I001)
- swarm.py / presentation.py / browser_tasks.py: store-path fail-safe
  except narrowed from bare Exception to (sqlite3.Error, OSError)
  (DB/file failures only); transport spawn/ping/close handlers in
  browser_tasks.py deliberately stay bare Exception (arbitrary
  transport errors must map to failed/False/close_error)
- NOT fixed (pre-existing style, left as-is): UP031 % formatting
  (used throughout the codebase incl. the copied lines), BLE001/
  TRY004 in untouched code, F401 atomic_write + RUF059 swarm
  (P7/P8 pre-existing)

cloud/chv/audit_trail.py

@@ -338,8 +338,7 @@ class StoreAuditTrail:
 
     def _open_store(self):
         """打开宿主级状态库（每次调用新建连接，用完即关）。"""
-        from cloud.chv.app_state import AppStateStore, \
-            host_state_db_path
+        from cloud.chv.app_state import AppStateStore, host_state_db_path
         store = AppStateStore(host_state_db_path(self.state_dir))
         store.open()
         return store
@@ -453,8 +452,7 @@ def migrate_jsonl_to_store(path: str, state_dir: str) -> int:
         raise ValueError("JSONL 审计链校验失败，拒绝迁移：行 %d"
                          % bad)
     rows = trail._read_all()
-    from cloud.chv.app_state import AppStateStore, \
-        host_state_db_path
+    from cloud.chv.app_state import AppStateStore, host_state_db_path
     store = AppStateStore(host_state_db_path(state_dir))
     store.open()
     try:

cloud/chv/browser_tasks.py

@@ -23,6 +23,7 @@ browser-tasks.json 首次加载时自动导入并改名 ``.migrated``；
 from __future__ import annotations
 
 import json
+import sqlite3
 import logging
 import os
 import time
@@ -85,7 +86,7 @@ def _load_registry(state_dir: str) -> dict:
     try:
         with _open_store(state_dir) as store:
             raw = store.kv_get(_KV_NAMESPACE, _KV_KEY)
-    except Exception:
+    except (sqlite3.Error, OSError):
         _LOGGER.warning("chv_browser_tasks_store_unreadable "
                         "state_dir=%s", state_dir)
         return {"instances": {}}
@@ -124,7 +125,7 @@ def _import_legacy(state_dir: str) -> dict:
     try:
         with _open_store(state_dir) as store:
             store.kv_put(_KV_NAMESPACE, _KV_KEY, json.dumps(data))
-    except Exception:
+    except (sqlite3.Error, OSError):
         _LOGGER.warning("chv_browser_tasks_import_failed "
                         "state_dir=%s", state_dir)
         return data
@@ -140,7 +141,7 @@ def _save_registry(state_dir: str, data: dict) -> None:
     try:
         with _open_store(state_dir) as store:
             store.kv_put(_KV_NAMESPACE, _KV_KEY, json.dumps(data))
-    except Exception:
+    except (sqlite3.Error, OSError):
         _LOGGER.warning("chv_browser_tasks_save_failed "
                         "state_dir=%s", state_dir)
 

cloud/chv/presentation.py

@@ -24,6 +24,7 @@ notification-ledger.json 首次加载时自动导入并改名
 from __future__ import annotations
 
 import json
+import sqlite3
 import logging
 import os
 import time
@@ -102,7 +103,7 @@ def _load_ledger(state_dir: str) -> dict:
     try:
         with _open_store(state_dir) as store:
             raw = store.kv_get(_KV_NAMESPACE, _KV_KEY)
-    except Exception:
+    except (sqlite3.Error, OSError):
         _LOGGER.warning("chv_ledger_store_unreadable state_dir=%s",
                         state_dir)
         return {"notified": {}}
@@ -140,7 +141,7 @@ def _import_legacy(state_dir: str) -> dict:
     try:
         with _open_store(state_dir) as store:
             store.kv_put(_KV_NAMESPACE, _KV_KEY, json.dumps(data))
-    except Exception:
+    except (sqlite3.Error, OSError):
         _LOGGER.warning("chv_ledger_import_failed state_dir=%s",
                         state_dir)
         return data
@@ -160,7 +161,7 @@ def _save_ledger(state_dir: str, data: dict) -> None:
     try:
         with _open_store(state_dir) as store:
             store.kv_put(_KV_NAMESPACE, _KV_KEY, json.dumps(data))
-    except Exception:
+    except (sqlite3.Error, OSError):
         _LOGGER.warning("chv_ledger_save_failed state_dir=%s", state_dir)
 
 

cloud/chv/swarm.py

@@ -21,6 +21,7 @@ orphans 同款口径），下次成功保存即覆盖自愈。
 from __future__ import annotations
 
 import json
+import sqlite3
 import logging
 import os
 import time
@@ -79,7 +80,7 @@ def _load_registry(state_dir: str) -> dict:
     try:
         with _open_store(state_dir) as store:
             raw = store.kv_get(_KV_NAMESPACE, _KV_KEY)
-    except Exception:
+    except (sqlite3.Error, OSError):
         _LOGGER.warning("chv_swarm_registry_store_unreadable "
                         "state_dir=%s", state_dir)
         return _empty_registry()
@@ -128,7 +129,7 @@ def _import_legacy(state_dir: str) -> dict:
     try:
         with _open_store(state_dir) as store:
             store.kv_put(_KV_NAMESPACE, _KV_KEY, json.dumps(data))
-    except Exception:
+    except (sqlite3.Error, OSError):
         _LOGGER.warning("chv_swarm_registry_import_failed "
                         "state_dir=%s", state_dir)
         return data
@@ -148,7 +149,7 @@ def _save_registry(state_dir: str, data: dict) -> None:
     try:
         with _open_store(state_dir) as store:
             store.kv_put(_KV_NAMESPACE, _KV_KEY, json.dumps(data))
-    except Exception:
+    except (sqlite3.Error, OSError):
         _LOGGER.warning("chv_swarm_registry_save_failed "
                         "state_dir=%s", state_dir)
 

cloud/tests/test_chv_muse_round12.py

@@ -15,18 +15,14 @@
 import importlib.util
 import json
 import os
-import socket
 import sqlite3
 import subprocess
-import sys
 import tempfile
 import threading
-import time
 import unittest
 from pathlib import Path
 
 REPO_ROOT = Path(__file__).resolve().parents[2]
-sys.path.insert(0, str(REPO_ROOT))
 
 from cloud.chv.app_state import AppStateStore, host_state_db_path
 from cloud.chv.audit_trail import (
@@ -34,16 +30,16 @@ from cloud.chv.audit_trail import (
     StoreAuditTrail,
     migrate_jsonl_to_store,
 )
-from cloud.chv.swarm import (
-    create_swarm,
-    get_swarm,
-)
-from cloud.chv.presentation import NotificationLedger
 from cloud.chv.browser_tasks import (
     BrowserTransport,
     create_instance,
     get_instance,
 )
+from cloud.chv.presentation import NotificationLedger
+from cloud.chv.swarm import (
+    create_swarm,
+    get_swarm,
+)
 
 TOKEN = "fake-token-123"
 T0 = 1700000000.0
@@ -471,12 +467,12 @@ class P12DictContractTest(unittest.TestCase):
     """P12 接线用真实 P3/P5 代码，不用替身。"""
 
     def test_approval_card_from_real_p3_queue(self):
-        from cloud.chv.sentinel import ApprovalQueue, EgressRequest
         from cloud.chv.approval_card import (
             build_approval_card,
             card_matches_channel,
         )
         from cloud.chv.approval_channel import request_approval
+        from cloud.chv.sentinel import ApprovalQueue, EgressRequest
         queue = ApprovalQueue(clock=lambda: T0)
         req = EgressRequest("gitea", "POST", "https",
                             "gitea.tose.top", 443,
@@ -496,8 +492,8 @@ class P12DictContractTest(unittest.TestCase):
             card_matches_channel(card, channel.payload()))
 
     def test_permission_registry_from_real_p5_describe(self):
-        from cloud.chv.capabilities import CapabilityIssuer
         from cloud.chv.activity import PermissionRegistry
+        from cloud.chv.capabilities import CapabilityIssuer
         issuer = CapabilityIssuer(clock=lambda: T0)
         cap = issuer.issue(
             "one-shot", "gitea", "gitea.tose.top", ["GET"],