Don't let a corrupt JSON state file crash server startup (#204)

coworker/connections.py

@@ -22,11 +22,12 @@
 
 from __future__ import annotations
 
-import json
 import threading
 from pathlib import Path
 from typing import Optional
 
+from .jsonstore import read_json, write_json_atomic
+
 
 class PersonaConnectionStore:
     """``{persona_id: {connector: bool}}`` — the per-persona default on/off for each connector."""
@@ -38,21 +39,14 @@ def __init__(self, path: Optional[str | Path] = None) -> None:
         self._load()
 
     def _load(self) -> None:
-        if self.path and self.path.is_file():
-            data = json.loads(self.path.read_text(encoding="utf-8"))
-            self._rows = {
-                pid: {str(c): bool(v) for c, v in (row or {}).items()}
-                for pid, row in data.get("personas", {}).items()
-            }
+        data = read_json(self.path, {}) or {}
+        self._rows = {
+            pid: {str(c): bool(v) for c, v in (row or {}).items()}
+            for pid, row in data.get("personas", {}).items()
+        }
 
     def _save(self) -> None:
-        if not self.path:
-            return
-        self.path.parent.mkdir(parents=True, exist_ok=True)
-        self.path.write_text(
-            json.dumps({"personas": self._rows}, indent=2),
-            encoding="utf-8",
-        )
+        write_json_atomic(self.path, {"personas": self._rows})
 
     # -- queries ----------------------------------------------------------------
     def get(self, persona_id: str) -> dict[str, bool]:
@@ -111,21 +105,14 @@ def __init__(self, path: Optional[str | Path] = None) -> None:
         self._load()
 
     def _load(self) -> None:
-        if self.path and self.path.is_file():
-            data = json.loads(self.path.read_text(encoding="utf-8"))
-            self._rows = {
-                sid: {str(c): bool(v) for c, v in (row or {}).items()}
-                for sid, row in data.get("sessions", {}).items()
-            }
+        data = read_json(self.path, {}) or {}
+        self._rows = {
+            sid: {str(c): bool(v) for c, v in (row or {}).items()}
+            for sid, row in data.get("sessions", {}).items()
+        }
 
     def _save(self) -> None:
-        if not self.path:
-            return
-        self.path.parent.mkdir(parents=True, exist_ok=True)
-        self.path.write_text(
-            json.dumps({"sessions": self._rows}, indent=2),
-            encoding="utf-8",
-        )
+        write_json_atomic(self.path, {"sessions": self._rows})
 
     # -- queries ----------------------------------------------------------------
     def get(self, session_id: str) -> dict[str, bool]:

coworker/inbox.py

@@ -22,6 +22,8 @@
 from pathlib import Path
 from typing import Any, Optional
 
+from .jsonstore import read_json, write_json_atomic
+
 KIND_APPROVAL = "approval"
 KIND_QUESTION = "question"
 KIND_NOTIFICATION = "notification"
@@ -97,19 +99,14 @@ def __init__(self, path: Optional[str | Path] = None) -> None:
 
     # -- persistence ------------------------------------------------------------
     def _load(self) -> None:
-        if self.path and self.path.is_file():
-            data = json.loads(self.path.read_text(encoding="utf-8"))
-            for raw in data.get("items", []):
-                item = InboxItem(**raw)
-                self._items[item.id] = item
+        data = read_json(self.path, {}) or {}
+        for raw in data.get("items", []):
+            item = InboxItem(**raw)
+            self._items[item.id] = item
 
     def _save(self) -> None:
-        if not self.path:
-            return
-        self.path.parent.mkdir(parents=True, exist_ok=True)
-        self.path.write_text(
-            json.dumps({"items": [asdict(i) for i in self._items.values()]}, indent=2),
-            encoding="utf-8",
+        write_json_atomic(
+            self.path, {"items": [asdict(i) for i in self._items.values()]}
         )
 
     # -- adding -----------------------------------------------------------------

coworker/inbox_routing.py

@@ -11,13 +11,14 @@
 
 from __future__ import annotations
 
-import json
 import re
 import threading
 from dataclasses import asdict, dataclass
 from pathlib import Path
 from typing import Callable, Optional
 
+from .jsonstore import read_json, write_json_atomic
+
 DEFAULT_INBOX = "default"
 # Embeds the item id in a delivered message. Emitted as [ow:…] since the bot's rebrand
 # to OpenWorker (2026-07-22); the legacy [ocw:…] spelling stays parseable so replies to
@@ -44,28 +45,21 @@ def __init__(self, path: Optional[str | Path] = None) -> None:
         self._load()
 
     def _load(self) -> None:
-        if self.path and self.path.is_file():
-            data = json.loads(self.path.read_text(encoding="utf-8"))
-            for raw in data.get("bindings", []):
-                b = InboxBinding(**raw)
-                self._bindings[b.name] = b
-            self._persona_default = dict(data.get("persona_default", {}))
-            self._session_override = dict(data.get("session_override", {}))
+        data = read_json(self.path, {}) or {}
+        for raw in data.get("bindings", []):
+            b = InboxBinding(**raw)
+            self._bindings[b.name] = b
+        self._persona_default = dict(data.get("persona_default", {}))
+        self._session_override = dict(data.get("session_override", {}))
 
     def _save(self) -> None:
-        if not self.path:
-            return
-        self.path.parent.mkdir(parents=True, exist_ok=True)
-        self.path.write_text(
-            json.dumps(
-                {
-                    "bindings": [asdict(b) for b in self._bindings.values()],
-                    "persona_default": self._persona_default,
-                    "session_override": self._session_override,
-                },
-                indent=2,
-            ),
-            encoding="utf-8",
+        write_json_atomic(
+            self.path,
+            {
+                "bindings": [asdict(b) for b in self._bindings.values()],
+                "persona_default": self._persona_default,
+                "session_override": self._session_override,
+            },
         )
 
     # -- config -----------------------------------------------------------------

coworker/jsonstore.py

@@ -0,0 +1,79 @@
+"""Durable JSON persistence helpers shared by the file-backed stores.
+
+The tiny JSON-backed stores (Inbox, subscriptions, mention threads, connection
+overrides, …) are all constructed in ``SessionManager.__init__``, and each one loads its
+file eagerly there. That makes their read/write robustness a *startup* concern, and two
+failure modes bite:
+
+- **A corrupt/truncated file must never block startup.** An unguarded ``json.loads`` in a
+  store's ``_load`` turns one unreadable file into a server that won't boot at all —
+  recoverable only by finding and deleting the file by hand. ``read_json`` returns the
+  caller's default instead, moving the bad file aside (``<name>.corrupt``) so it is neither
+  lost nor silently re-hit on the next load.
+- **A save must be atomic.** Writing in place leaves a half-written file if the process dies
+  mid-write — which is exactly what produces the corrupt file above. ``write_json_atomic``
+  writes a sibling temp file and ``os.replace``s it into place (atomic on POSIX and Windows).
+
+This just generalizes the pattern already applied inline to ``ChannelBuffer`` and
+``WorkspaceTrustStore``, so every store gets the same treatment instead of a few.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+from pathlib import Path
+from typing import Any
+
+logger = logging.getLogger("coworker.jsonstore")
+
+
+def read_json(path: str | Path | None, default: Any = None) -> Any:
+    """Load JSON from ``path``, tolerating a missing or unreadable file.
+
+    Returns ``default`` when the path is ``None``, absent, or unreadable/corrupt. A corrupt
+    file is moved aside to ``<name>.corrupt`` (best effort) so the bad data is preserved for
+    inspection and the next ``write_json_atomic`` starts from a clean slate rather than the
+    store re-hitting the same parse error on every boot.
+    """
+    if path is None:
+        return default
+    p = Path(path)
+    if not p.is_file():
+        return default
+    try:
+        return json.loads(p.read_text(encoding="utf-8"))
+    except (OSError, ValueError) as exc:
+        logger.warning("ignoring unreadable JSON at %s (%s); moving it aside", p, exc)
+        _quarantine(p)
+        return default
+
+
+def _quarantine(path: Path) -> None:
+    try:
+        path.replace(path.with_name(path.name + ".corrupt"))
+    except OSError:
+        pass
+
+
+def write_json_atomic(
+    path: str | Path | None, data: Any, *, indent: int | None = 2
+) -> None:
+    """Serialize ``data`` and write it to ``path`` atomically, creating parent dirs.
+
+    A no-op when ``path`` is ``None`` (the in-memory-only store case). Writes to a sibling
+    temp file first, then ``os.replace``s it over the target, so a crash mid-write can never
+    leave a truncated file that would brick the next startup.
+    """
+    if path is None:
+        return
+    p = Path(path)
+    p.parent.mkdir(parents=True, exist_ok=True)
+    tmp = p.with_name(f".{p.name}.{os.getpid()}.tmp")
+    try:
+        tmp.write_text(json.dumps(data, indent=indent), encoding="utf-8")
+        os.replace(tmp, p)
+    except BaseException:
+        tmp.unlink(missing_ok=True)
+        raise

coworker/mentions.py

@@ -15,12 +15,13 @@
 
 from __future__ import annotations
 
-import json
 import threading
 from dataclasses import asdict, dataclass
 from pathlib import Path
 from typing import Optional
 
+from .jsonstore import read_json, write_json_atomic
+
 
 @dataclass
 class MentionThread:
@@ -37,18 +38,11 @@ def __init__(self, path: Optional[str | Path] = None) -> None:
         self._load()
 
     def _load(self) -> None:
-        if self.path and self.path.is_file():
-            data = json.loads(self.path.read_text(encoding="utf-8"))
-            self._threads = [MentionThread(**raw) for raw in data.get("threads", [])]
+        data = read_json(self.path, {}) or {}
+        self._threads = [MentionThread(**raw) for raw in data.get("threads", [])]
 
     def _save(self) -> None:
-        if not self.path:
-            return
-        self.path.parent.mkdir(parents=True, exist_ok=True)
-        self.path.write_text(
-            json.dumps({"threads": [asdict(t) for t in self._threads]}, indent=2),
-            encoding="utf-8",
-        )
+        write_json_atomic(self.path, {"threads": [asdict(t) for t in self._threads]})
 
     # -- mutations --------------------------------------------------------------
     def set(self, thread_target: str, session_id: str, channel: str) -> MentionThread:

coworker/overrides.py

@@ -11,12 +11,12 @@
 
 from __future__ import annotations
 
-import json
 from dataclasses import dataclass
 from fnmatch import fnmatchcase
 from pathlib import Path
 from typing import Callable, Optional
 
+from .jsonstore import read_json, write_json_atomic
 from .risk import RiskClass
 
 
@@ -39,32 +39,19 @@ def __init__(self, path: Optional[str | Path] = None) -> None:
         self._rules: list[_Rule] = self._load()
 
     def _load(self) -> list[_Rule]:
-        if not (self.path and self.path.is_file()):
-            return []
-        data = json.loads(self.path.read_text(encoding="utf-8"))
+        data = read_json(self.path, {}) or {}
         rules = []
         for r in data.get("rules", []):
             try:
                 rules.append(_Rule(str(r["pattern"]), RiskClass(str(r["risk"]))))
-            except (KeyError, ValueError):
+            except (KeyError, ValueError, TypeError):
                 continue  # skip malformed rules rather than failing the whole store
         return rules
 
     def save(self) -> None:
-        if not self.path:
-            return
-        self.path.parent.mkdir(parents=True, exist_ok=True)
-        self.path.write_text(
-            json.dumps(
-                {
-                    "rules": [
-                        {"pattern": r.pattern, "risk": r.risk.value}
-                        for r in self._rules
-                    ]
-                },
-                indent=2,
-            ),
-            encoding="utf-8",
+        write_json_atomic(
+            self.path,
+            {"rules": [{"pattern": r.pattern, "risk": r.risk.value} for r in self._rules]},
         )
 
     def set_rule(self, pattern: str, risk: RiskClass | str) -> None:

coworker/subscriptions.py

@@ -25,6 +25,8 @@
 from pathlib import Path
 from typing import Optional
 
+from .jsonstore import read_json, write_json_atomic
+
 
 @dataclass
 class Subscription:
@@ -42,17 +44,12 @@ def __init__(self, path: Optional[str | Path] = None) -> None:
         self._load()
 
     def _load(self) -> None:
-        if self.path and self.path.is_file():
-            data = json.loads(self.path.read_text(encoding="utf-8"))
-            self._subs = [Subscription(**raw) for raw in data.get("subscriptions", [])]
+        data = read_json(self.path, {}) or {}
+        self._subs = [Subscription(**raw) for raw in data.get("subscriptions", [])]
 
     def _save(self) -> None:
-        if not self.path:
-            return
-        self.path.parent.mkdir(parents=True, exist_ok=True)
-        self.path.write_text(
-            json.dumps({"subscriptions": [asdict(s) for s in self._subs]}, indent=2),
-            encoding="utf-8",
+        write_json_atomic(
+            self.path, {"subscriptions": [asdict(s) for s in self._subs]}
         )
 
     # -- mutations --------------------------------------------------------------

coworker/unattended.py

@@ -8,25 +8,21 @@
 
 from __future__ import annotations
 
-import json
 import threading
 from pathlib import Path
 from typing import Optional
 
+from .jsonstore import read_json, write_json_atomic
+
 
 class UnattendedRegistry:
     def __init__(self, path: Optional[str | Path] = None) -> None:
         self.path = Path(path) if path else None
         self._lock = threading.Lock()
-        self._flags: dict[str, bool] = {}
-        if self.path and self.path.is_file():
-            self._flags = dict(json.loads(self.path.read_text(encoding="utf-8")))
+        self._flags: dict[str, bool] = dict(read_json(self.path, {}) or {})
 
     def _save(self) -> None:
-        if not self.path:
-            return
-        self.path.parent.mkdir(parents=True, exist_ok=True)
-        self.path.write_text(json.dumps(self._flags, indent=2), encoding="utf-8")
+        write_json_atomic(self.path, self._flags)
 
     def is_unattended(self, session_id: str) -> bool:
         return bool(self._flags.get(session_id, False))