fix(stores): a corrupt state file must not stop server startup

coworker/inbox.py

@@ -22,13 +22,16 @@
 
 import asyncio
 import json
+import logging
 import threading
 import uuid
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any, Optional
 
+logger = logging.getLogger(__name__)
+
 KIND_APPROVAL = "approval"
 KIND_QUESTION = "question"
 KIND_NOTIFICATION = "notification"
@@ -119,10 +122,15 @@ 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
+            try:
+                data = json.loads(self.path.read_text(encoding="utf-8"))
+                for raw in data.get("items", []):
+                    item = InboxItem(**raw)
+                    self._items[item.id] = item
+            except (OSError, ValueError, TypeError, AttributeError):
+                # A corrupt file must never block startup (#204).
+                logger.warning("inbox file at %s is unreadable; starting empty", self.path)
+                self._items = {}
 
     def _save(self) -> None:
         if not self.path:

coworker/inbox_routing.py

@@ -12,12 +12,15 @@
 from __future__ import annotations
 
 import json
+import logging
 import re
 import threading
 from dataclasses import asdict, dataclass
 from pathlib import Path
 from typing import Callable, Optional
 
+logger = logging.getLogger(__name__)
+
 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
@@ -45,12 +48,21 @@ def __init__(self, path: Optional[str | Path] = None) -> None:
 
     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)
+            try:
+                data = json.loads(self.path.read_text(encoding="utf-8"))
+                bindings = [InboxBinding(**raw) for raw in data.get("bindings", [])]
+                persona_default = dict(data.get("persona_default", {}))
+                session_override = dict(data.get("session_override", {}))
+            except (OSError, ValueError, TypeError, AttributeError):
+                # A corrupt file must never block startup (#204).
+                logger.warning(
+                    "inbox routing file at %s is unreadable; using the defaults", self.path
+                )
+                return
+            for b in bindings:
                 self._bindings[b.name] = b
-            self._persona_default = dict(data.get("persona_default", {}))
-            self._session_override = dict(data.get("session_override", {}))
+            self._persona_default = persona_default
+            self._session_override = session_override
 
     def _save(self) -> None:
         if not self.path:

coworker/selfwake.py

@@ -12,13 +12,16 @@
 from __future__ import annotations
 
 import json
+import logging
 import threading
 import uuid
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
 from typing import Optional
 
+logger = logging.getLogger(__name__)
+
 KIND_TIMER = "timer"
 KIND_COMPLETION = "completion"
 KIND_EVENT = "event"  # wake when a named connector/webhook event fires (Phase 3)
@@ -55,11 +58,17 @@ def __init__(self, path: Optional[str | Path] = None) -> None:
         self._wakes: dict[str, Wake] = {}
         self.stopped_sessions: set[str] = set()
         if self.path and self.path.is_file():
-            saved = json.loads(self.path.read_text(encoding="utf-8"))
-            self.stopped_sessions = set(saved.get("stopped_sessions", []))
-            for raw in saved.get("wakes", []):
-                w = Wake(**raw)
-                self._wakes[w.id] = w
+            try:
+                saved = json.loads(self.path.read_text(encoding="utf-8"))
+                self.stopped_sessions = set(saved.get("stopped_sessions", []))
+                for raw in saved.get("wakes", []):
+                    w = Wake(**raw)
+                    self._wakes[w.id] = w
+            except (OSError, ValueError, TypeError, AttributeError):
+                # A corrupt file must never block startup (#204).
+                logger.warning("wake file at %s is unreadable; starting empty", self.path)
+                self.stopped_sessions = set()
+                self._wakes = {}
             # Older versions accumulated timers. Keep only the newest pending
             # sleep per session while retaining the replaced records for audit.
             newest = {}

coworker/subscriptions.py

@@ -18,13 +18,16 @@
 from __future__ import annotations
 
 import json
+import logging
 import re
 import threading
 from collections import deque
 from dataclasses import asdict, dataclass, field
 from pathlib import Path
 from typing import Callable, Optional
 
+logger = logging.getLogger(__name__)
+
 
 @dataclass
 class Subscription:
@@ -43,8 +46,14 @@ def __init__(self, path: Optional[str | Path] = None) -> None:
 
     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", [])]
+            try:
+                data = json.loads(self.path.read_text(encoding="utf-8"))
+                self._subs = [Subscription(**raw) for raw in data.get("subscriptions", [])]
+            except (OSError, ValueError, TypeError, AttributeError):
+                # A corrupt file must never block startup (#204).
+                logger.warning(
+                    "subscriptions file at %s is unreadable; starting empty", self.path
+                )
 
     def _save(self) -> None:
         if not self.path:

coworker/unattended.py

@@ -19,10 +19,13 @@
 from __future__ import annotations
 
 import json
+import logging
 import threading
 from pathlib import Path
 from typing import Optional
 
+logger = logging.getLogger(__name__)
+
 ATTENDED = "attended"
 INBOX = "inbox"
 AUTO = "auto"
@@ -66,11 +69,19 @@ def __init__(self, path: Optional[str | Path] = None) -> None:
         self._lock = threading.Lock()
         self._values: dict[str, str] = {}
         if self.path and self.path.is_file():
-            raw = json.loads(self.path.read_text(encoding="utf-8"))
-            for sid, value in dict(raw).items():
-                normalized = normalize_attendance(value)
-                if normalized != ATTENDED:
-                    self._values[str(sid)] = normalized
+            try:
+                raw = json.loads(self.path.read_text(encoding="utf-8"))
+                for sid, value in dict(raw).items():
+                    normalized = normalize_attendance(value)
+                    if normalized != ATTENDED:
+                        self._values[str(sid)] = normalized
+            except (OSError, ValueError, TypeError, AttributeError):
+                # A corrupt file must never block startup (#204). Every session reads as
+                # attended, the default that still asks before acting.
+                logger.warning(
+                    "attendance file at %s is unreadable; starting empty", self.path
+                )
+                self._values = {}
 
     def _save(self) -> None:
         if not self.path:

coworker/unrouted.py

@@ -12,12 +12,15 @@
 from __future__ import annotations
 
 import json
+import logging
 import threading
 import time
 from dataclasses import asdict, dataclass, field
 from pathlib import Path
 from typing import Optional
 
+logger = logging.getLogger(__name__)
+
 
 @dataclass
 class UnroutedItem:
@@ -38,8 +41,12 @@ def __init__(self, path: Optional[str | Path] = None, *, cap: int = 200) -> None
 
     def _load(self) -> None:
         if self.path and self.path.is_file():
-            data = json.loads(self.path.read_text(encoding="utf-8"))
-            self._items = [UnroutedItem(**raw) for raw in data.get("items", [])]
+            try:
+                data = json.loads(self.path.read_text(encoding="utf-8"))
+                self._items = [UnroutedItem(**raw) for raw in data.get("items", [])]
+            except (OSError, ValueError, TypeError, AttributeError):
+                # A corrupt file must never block startup (#204).
+                logger.warning("unrouted file at %s is unreadable; starting empty", self.path)
 
     def _save(self) -> None:
         if not self.path:

tests/test_store_robustness.py

@@ -0,0 +1,111 @@
+"""A corrupt JSON state file must not stop the server from starting (#204).
+
+These stores are built eagerly by ``SessionManager.__init__``, so an exception from any
+of their loaders aborts startup until the user finds and deletes the file by hand. A
+save interrupted mid-write leaves a truncated document, and a file can also be valid
+JSON of the wrong shape. Each store should treat all of these as "start empty".
+"""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from coworker.inbox import InboxStore
+from coworker.inbox_routing import DEFAULT_INBOX, InboxRouting
+from coworker.selfwake import WakeStore
+from coworker.subscriptions import SubscriptionStore
+from coworker.unattended import ATTENDED, UnattendedRegistry
+from coworker.unrouted import UnroutedStore
+
+_CORRUPT_DOCUMENTS = [
+    pytest.param('{"items": [{"id": "abc", "ses', id="truncated"),
+    pytest.param("", id="empty-file"),
+    pytest.param("null", id="json-null"),
+    pytest.param('"text"', id="json-string"),
+]
+
+# Valid JSON with the right top-level keys, but records the dataclass can't accept.
+_BAD_RECORD = {"not_a_field": 1}
+_BAD_RECORDS = {
+    "inbox": {"items": [_BAD_RECORD]},
+    "routing": {"bindings": [_BAD_RECORD]},
+    "wakes": {"wakes": [_BAD_RECORD]},
+    "subscriptions": {"subscriptions": [_BAD_RECORD]},
+    "unrouted": {"items": [_BAD_RECORD]},
+}
+
+
+def _write(tmp_path, text):
+    path = tmp_path / "state.json"
+    path.write_text(text, encoding="utf-8")
+    return path
+
+
+@pytest.mark.parametrize("document", _CORRUPT_DOCUMENTS)
+def test_inbox_starts_empty(tmp_path, document):
+    assert InboxStore(_write(tmp_path, document)).pending() == []
+
+
+@pytest.mark.parametrize("document", _CORRUPT_DOCUMENTS)
+def test_inbox_routing_keeps_defaults(tmp_path, document):
+    routing = InboxRouting(_write(tmp_path, document))
+    assert routing.route_for("session", "persona") == DEFAULT_INBOX
+    assert [b["name"] for b in routing.bindings()] == [DEFAULT_INBOX]
+
+
+@pytest.mark.parametrize("document", _CORRUPT_DOCUMENTS)
+def test_attendance_reads_as_attended(tmp_path, document):
+    assert UnattendedRegistry(_write(tmp_path, document)).attendance("session") == ATTENDED
+
+
+@pytest.mark.parametrize("document", _CORRUPT_DOCUMENTS)
+def test_wakes_start_empty(tmp_path, document):
+    store = WakeStore(_write(tmp_path, document))
+    assert store.due() == []
+    assert store.stopped_sessions == set()
+
+
+@pytest.mark.parametrize("document", _CORRUPT_DOCUMENTS)
+def test_subscriptions_start_empty(tmp_path, document):
+    assert SubscriptionStore(_write(tmp_path, document)).all() == []
+
+
+@pytest.mark.parametrize("document", _CORRUPT_DOCUMENTS)
+def test_unrouted_starts_empty(tmp_path, document):
+    assert UnroutedStore(_write(tmp_path, document)).list() == []
+
+
+def test_a_record_with_unknown_fields_does_not_raise(tmp_path):
+    assert InboxStore(_write(tmp_path, json.dumps(_BAD_RECORDS["inbox"]))).pending() == []
+    assert [
+        b["name"]
+        for b in InboxRouting(_write(tmp_path, json.dumps(_BAD_RECORDS["routing"]))).bindings()
+    ] == [DEFAULT_INBOX]
+    assert WakeStore(_write(tmp_path, json.dumps(_BAD_RECORDS["wakes"]))).due() == []
+    assert SubscriptionStore(_write(tmp_path, json.dumps(_BAD_RECORDS["subscriptions"]))).all() == []
+    assert UnroutedStore(_write(tmp_path, json.dumps(_BAD_RECORDS["unrouted"]))).list() == []
+
+
+def test_a_good_record_before_a_bad_one_is_not_kept(tmp_path):
+    """A half-read file must not leave a store with only some of its records."""
+    wake = {"id": "w1", "session_id": "s1", "kind": "timer", "state": "pending"}
+    document = {"stopped_sessions": ["s9"], "wakes": [wake, _BAD_RECORD]}
+    store = WakeStore(_write(tmp_path, json.dumps(document)))
+    assert store.stopped_sessions == set()
+    assert store.due() == []
+
+
+def test_a_save_replaces_the_corrupt_file(tmp_path):
+    path = _write(tmp_path, "{not json")
+    store = UnroutedStore(path)
+    store.record("slack", "someone", "hello", "no route")
+    assert [item["text"] for item in UnroutedStore(path).list()] == ["hello"]
+
+
+def test_valid_files_still_load(tmp_path):
+    path = _write(tmp_path, "{}")
+    store = UnroutedStore(path)
+    store.record("slack", "someone", "hello", "no route")
+    assert len(UnroutedStore(path).list()) == 1