diff --git a/coworker/inbox.py b/coworker/inbox.py index 39a4cb0b3..47fdef810 100644 --- a/coworker/inbox.py +++ b/coworker/inbox.py @@ -22,6 +22,7 @@ import asyncio import json +import logging import threading import uuid from dataclasses import asdict, dataclass, field @@ -29,6 +30,8 @@ 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: diff --git a/coworker/inbox_routing.py b/coworker/inbox_routing.py index bcbf33f8a..7a27d1720 100644 --- a/coworker/inbox_routing.py +++ b/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: diff --git a/coworker/selfwake.py b/coworker/selfwake.py index d9d55595a..9b4bf0e36 100644 --- a/coworker/selfwake.py +++ b/coworker/selfwake.py @@ -12,6 +12,7 @@ from __future__ import annotations import json +import logging import threading import uuid from dataclasses import asdict, dataclass, field @@ -19,6 +20,8 @@ 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 = {} diff --git a/coworker/subscriptions.py b/coworker/subscriptions.py index db800bc2d..84ecb9f43 100644 --- a/coworker/subscriptions.py +++ b/coworker/subscriptions.py @@ -18,6 +18,7 @@ from __future__ import annotations import json +import logging import re import threading from collections import deque @@ -25,6 +26,8 @@ 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: diff --git a/coworker/unattended.py b/coworker/unattended.py index b5259aad9..39aaf6b05 100644 --- a/coworker/unattended.py +++ b/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: diff --git a/coworker/unrouted.py b/coworker/unrouted.py index e17328ac5..cd633f7a9 100644 --- a/coworker/unrouted.py +++ b/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: diff --git a/tests/test_store_robustness.py b/tests/test_store_robustness.py new file mode 100644 index 000000000..7ece464f7 --- /dev/null +++ b/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